diff --git a/docs/source/reference/running/data-pipeline-plugins.rst b/docs/source/reference/running/data-pipeline-plugins.rst index 08d604ac..3376957f 100644 --- a/docs/source/reference/running/data-pipeline-plugins.rst +++ b/docs/source/reference/running/data-pipeline-plugins.rst @@ -119,7 +119,7 @@ A typical BUFR4 plugin workflow definition would be defined as follows: This plugin takes the incoming XML file, then validates it against the `CAP v1.2 schema `_ -and verifies the digital signature before publishing. +and optionally verifies the digital signature before publishing. The validation is performed using the `capvalidator `_ package. @@ -135,6 +135,12 @@ A typical CAP message plugin workflow definition would be defined as follows: - ${WIS2BOX_STORAGE_INCOMING} file-pattern: '^.*\.xml$' +By default the XML signature validation is set to ``False``. To enable the validation add the following environment variable to your ``wis2box.env`` file: + +.. code-block:: bash + + CHECK_CAP_SIGNATURE=True + ``wis2box.data.universal.UniversalData`` ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ diff --git a/wis2box-management/wis2box/data/cap_message.py b/wis2box-management/wis2box/data/cap_message.py index a64526d1..2dce70a5 100644 --- a/wis2box-management/wis2box/data/cap_message.py +++ b/wis2box-management/wis2box/data/cap_message.py @@ -20,13 +20,16 @@ ############################################################################### from datetime import datetime +import os import logging + from pathlib import Path from typing import Union from capvalidator import validate_cap_message, get_dates from wis2box.data.base import BaseAbstractData +from wis2box.util import str2bool LOGGER = logging.getLogger(__name__) @@ -74,8 +77,12 @@ def transform(self, input_data: Union[Path, bytes], # add relative filepath to _meta _meta['relative_filepath'] = self.get_local_filepath(_meta['data_date']) # noqa + # check CAP signature based on ENV variable, default is False + check_cap_signature = str2bool(os.getenv('CHECK_CAP_SIGNATURE', False)) + + LOGGER.info(f'Checking CAP signature: {check_cap_signature}') # validate the CAP XML string content using the capvalidator package - result = validate_cap_message(input_bytes, strict=False) + result = validate_cap_message(input_bytes, strict=check_cap_signature) if not result.passed: LOGGER.error( f'Invalid CAP XML, not publishing. Reason: {result.message}') diff --git a/wis2box-management/wis2box/util.py b/wis2box-management/wis2box/util.py index 75618107..60386394 100644 --- a/wis2box-management/wis2box/util.py +++ b/wis2box-management/wis2box/util.py @@ -34,6 +34,26 @@ LOGGER = logging.getLogger(__name__) +def str2bool(value: Union[bool, str]) -> bool: + """ + helper function to return Python boolean + type (source: https://stackoverflow.com/a/715468) + + :param value: value to be evaluated + + :returns: `bool` of whether the value is boolean-ish + """ + + value2 = False + + if isinstance(value, bool): + value2 = value + else: + value2 = value.lower() in ('yes', 'true', 't', '1', 'on') + + return value2 + + def get_typed_value(value) -> Union[float, int, str]: """ Derive true type from data value