# -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <john@compose-x.io>
"""
Module for the XStack SSM
"""
import json
from os import path
import yaml
from compose_x_common.compose_x_common import keyisset
from troposphere import AWS_ACCOUNT_ID, AWS_PARTITION, AWS_REGION, Base64, Ref, Sub
from troposphere.ssm import Parameter as CfnSsmParameter
from yaml import Loader
from ecs_composex.common import LOG, add_outputs, build_template
from ecs_composex.common.compose_resources import (
XResource,
set_lookup_resources,
set_new_resources,
set_resources,
set_use_resources,
)
from ecs_composex.common.stacks import ComposeXStack
from ecs_composex.resources_import import import_record_properties
from ecs_composex.ssm_parameter.ssm_parameter_ecs import create_ssm_param_mappings
from ecs_composex.ssm_parameter.ssm_params import (
MAPPINGS_KEY,
MOD_KEY,
RES_KEY,
SSM_PARAM_ARN,
SSM_PARAM_NAME,
)
from ecs_composex.ssm_parameter.ssm_perms import get_access_types
[docs]def handle_yaml_validation(resource, value, file_path):
"""
Function to evaluate the JSON content
:param SsmParamter resource:
:param str value: Value read from file
:param str file_path:
:return:
"""
try:
payload = yaml.load(value, Loader=Loader)
if keyisset("RenderToJson", resource.parameters):
return json.dumps(payload, separators=(",", ":"))
return value
except yaml.YAMLError:
if keyisset("IgnoreInvalidYaml", resource.parameters):
LOG.warn(
f"{resource.name} - The content of {file_path} "
"did not pass YAML validation. Skipping due to IgnoreInvalidYaml"
)
return value
else:
LOG.error(
f"{resource.name} - The content of {file_path} "
"did not pass YAML validation."
)
raise
[docs]def handle_json_validation(resource, value, file_path):
"""
Function to evaluate the JSON content
:param SsmParamter resource:
:param str value: Value read from file
:param str file_path:
:return:
"""
try:
payload = json.loads(value)
if keyisset("MinimizeJson", resource.parameters):
return json.dumps(payload, separators=(",", ":"))
return value
except json.decoder.JSONDecodeError:
if keyisset("IgnoreInvalidJson", resource.parameters):
LOG.warn(
f"{resource.name} - The content of {file_path} "
"did not pass JSON validation. Skipping due to IgnoreInvalidJson"
)
return value
else:
LOG.error(
f"{resource.name} - The content of {file_path} "
"did not pass JSON validation."
)
raise
[docs]def import_value_from_file(resource):
"""
Function to import file into the SSM Parameter value
:param SsmParameter resource:
:return: The value
:rtype: str
"""
file_path = path.abspath(resource.parameters["FromFile"])
with open(file_path, "r") as file_fd:
value = file_fd.read()
if keyisset("EncodeToBase64", resource.parameters):
return Base64(value)
if keyisset("ValidateJson", resource.parameters):
return handle_json_validation(resource, value, file_path)
elif keyisset("ValidateYaml", resource.parameters):
return handle_yaml_validation(resource, value, file_path)
return value
[docs]def render_new_parameters(new_resources, root_stack):
"""
:param list[SsmParameter] new_resources:
:param ecs_composex.common.stacks.ComposeXStack root_stack:
:return:
"""
for new_res in new_resources:
value = None
if (
keyisset("Type", new_res.definition)
and new_res.definition["Type"] == "SecureString"
):
raise ValueError(f"{new_res.name} AWS CFN does not support SecureString.")
if new_res.parameters and keyisset("FromFile", new_res.parameters):
value = import_value_from_file(new_res)
if keyisset("Value", new_res.properties):
if value:
LOG.warn(
"Both Value and FromFile properties were set. Using Value from Properties"
)
value = new_res.properties["Value"]
if not value:
raise ValueError(f"{new_res.name} - Failed to determine the value")
new_res.properties.update({"Value": value})
param_props = import_record_properties(
new_res.properties, CfnSsmParameter, ignore_missing_required=False
)
new_res.cfn_resource = CfnSsmParameter(new_res.logical_name, **param_props)
root_stack.stack_template.add_resource(new_res.cfn_resource)
new_res.init_outputs()
new_res.generate_outputs()
add_outputs(root_stack.stack_template, new_res.outputs)
[docs]class SsmParameter(XResource):
"""
Class to represent a SQS Queue
"""
policies_scaffolds = get_access_types()
[docs] def init_outputs(self):
spacer = ""
if (
self.properties
and keyisset("Name", self.properties)
and not self.properties["Name"].startswith(r"/")
):
spacer = "/"
self.output_properties = {
SSM_PARAM_NAME: (self.logical_name, self.cfn_resource, Ref, None),
SSM_PARAM_ARN: (
self.logical_name,
self.cfn_resource,
Sub,
f"arn:${{{AWS_PARTITION}}}:ssm:${{{AWS_REGION}}}:${{{AWS_ACCOUNT_ID}}}:parameter{spacer}"
f"${{{self.cfn_resource.title}}}",
),
}
[docs]class XStack(ComposeXStack):
"""
Class to handle SQS Root stack related actions
"""
def __init__(self, title, settings, **kwargs):
set_resources(
settings, SsmParameter, RES_KEY, MOD_KEY, mapping_key=MAPPINGS_KEY
)
x_resources = settings.compose_content[RES_KEY].values()
new_resources = set_new_resources(x_resources, RES_KEY, False)
lookup_resources = set_lookup_resources(x_resources, RES_KEY)
use_resources = set_use_resources(x_resources, RES_KEY, False)
if new_resources:
template = build_template("Parent template for SSM in ECS Compose-X")
super().__init__(title, stack_template=template, **kwargs)
render_new_parameters(new_resources, self)
else:
self.is_void = True
for resource in settings.compose_content[RES_KEY].values():
resource.stack = self
if lookup_resources or use_resources:
if not keyisset(RES_KEY, settings.mappings):
settings.mappings[RES_KEY] = {}
create_ssm_param_mappings(
settings.mappings[RES_KEY], lookup_resources, settings
)