"""
Interface module for generating CSP configuration.
Handles parsing and validation of inputs and passes them on to the
internal configuration functions in :py:mod:`.config`.
"""
import copy
import json
from typing import Union
from . import config, version
[docs]def make_csp_config(
csp_interface_version: str,
sdp_interface_version: str,
scan_type: str,
csp_config_str: Union[str, dict],
sdp_receive_addrs_map_str: Union[str, dict],
) -> str:
"""Generate CSP scan configuration for a scan using SDP receive addresses.
This should be used right before CSP is configured so that data
streams are sent to the right ingest nodes.
:param csp_interface_version: Version of CSP interface (URI)
:param sdp_interface_version: Version of SDP interface (URI)
:param scan_type: Type of scan to configure
:param csp_config_in: General CSP configuration
:param sdp_receive_addrs: Receive addresses map for scan types,
generated by SDP
:return: A validated JSON string with CSP configuration.
:raise: `ValueError` when the input JSON configuration fails validation.
"""
from .. import schema
# Parse JSON
if isinstance(csp_config_str, str):
csp_config = json.loads(csp_config_str)
else:
csp_config = copy.deepcopy(dict(csp_config_str))
if isinstance(sdp_receive_addrs_map_str, str):
sdp_receive_addrs = json.loads(sdp_receive_addrs_map_str)
else:
sdp_receive_addrs = dict(sdp_receive_addrs_map_str)
# Convert version to standard format
csp_interface_version = version.normalize_csp_config_version(
csp_interface_version, csp_config
)
# Valid?
version.check_csp_interface_version(
csp_interface_version, version.CSP_CONFIG_PREFIX
)
schema.validate(csp_interface_version, csp_config)
# Get receive addresses for scan type
if scan_type not in sdp_receive_addrs:
raise ValueError(
f"No receive addresses found for scan type '{scan_type}'!"
)
scan_receive_addrs = sdp_receive_addrs[scan_type]
# Add receive addresses into CSP configuration
csp_config = config.add_receive_addresses(
scan_type,
csp_config,
scan_receive_addrs,
csp_interface_version,
sdp_interface_version,
)
# Do schema check
schema.validate(csp_interface_version, csp_config)
# Translate back into JSON
return json.dumps(csp_config, indent=2)