def _add_ssm_param_injector()

in cli/aws_orbit/models/manifest.py [0:0]

• 39 lines of code
• 11 McCabe index (conditional complexity)

def _add_ssm_param_injector(tag: str = "!SSM") -> Set[str]:
    """
    Load a yaml configuration file and resolve any SSM parameters
    The SSM parameters must have !SSM before them and be in this format
    to be parsed: ${SSM_PARAMETER_PATH::JSONPATH}.
    E.g.:
    database:
        host: !SSM ${/orbit-f/dev-env/resources::/UserAccessPolicy}
        port: !SSM ${/orbit-f/dev-env/resources::/PublicSubnet/*}
    """
    # pattern for global vars: look for ${word}
    pattern = re.compile(".*?\${([^}]+::[^}]*)}.*?")  # noqa: W605
    loader = yaml.SafeLoader

    # the tag will be used to mark where to start searching for the pattern
    # e.g. somekey: !SSM somestring${MYENVVAR}blah blah blah
    loader.add_implicit_resolver(tag, pattern, None)  # type: ignore

    ssm_parameters = set()

    def constructor_ssm_parameter(loader, node) -> Any:  # type: ignore
        """
        Extracts the environment variable from the node's value
        :param yaml.Loader loader: the yaml loader
        :param node: the current node in the yaml
        :return: the parsed string that contains the value of the environment
        variable
        """
        value = loader.construct_scalar(node)
        match = pattern.findall(value)  # to find all env variables in line
        if match:
            full_value = value
            for g in match:
                _logger.debug(f"match: {g}")
                (ssm_param_name, jsonpath) = g.split("::")
                if "${" in ssm_param_name:
                    ssm_param_name = ssm_param_name.replace("$", "").format(os.environ)
                _logger.debug(f"found injected parameter {(ssm_param_name, jsonpath)}")
                if ssm_param_name not in SSM_CONTEXT:
                    ssm = boto3_client("ssm")
                    try:
                        SSM_CONTEXT[ssm_param_name] = json.loads(
                            ssm.get_parameter(Name=ssm_param_name)["Parameter"]["Value"]
                        )
                        ssm_parameters.add(ssm_param_name)
                    except Exception as e:
                        _logger.error(f"Error resolving injected parameter {g}: {e}")

                json_expr = jsonpath_ng.parse(jsonpath)
                json_data = SSM_CONTEXT[ssm_param_name]
                json_match = json_expr.find(json_data)

                if len(json_match) > 1:
                    raise Exception(f"Injected parameter {g} is ambiguous")
                elif len(json_match) == 0:
                    raise Exception(f"Injected parameter {jsonpath} not found in SSM {ssm_param_name}")

                param_value: str = json_match[0].value
                _logger.debug(f"injected SSM parameter {g} resolved to {param_value}")
                return param_value
            return full_value
        return value

    loader.add_constructor(tag, constructor_ssm_parameter)  # type: ignore
    return ssm_parameters