Source code for ecs_composex.dns.dns_records

#  -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <john@compose-x.io>

"""
Module to handle creation of the DNS records for specific targets
"""

from copy import deepcopy
from re import compile

from compose_x_common.compose_x_common import keyisset
from troposphere import AWS_REGION, AWS_STACK_NAME, GetAtt, Ref
from troposphere.route53 import AliasTarget, BaseRecordSet, RecordSetType

from ecs_composex.common import LOG, NONALPHANUM, add_outputs
from ecs_composex.common.ecs_composex import X_KEY
from ecs_composex.dns.dns_params import RES_KEY
from ecs_composex.elbv2.elbv2_params import LB_DNS_NAME, LB_DNS_ZONE_ID
from ecs_composex.resources_import import import_record_properties


[docs]def define_external_record_set(properties): """ Function to define a record set without any targets in ComposeX :param properties: :return: """ props = import_record_properties(properties, BaseRecordSet) record = RecordSetType( f"ExternalR53Record{props['Type']}{NONALPHANUM.sub('', props['Name'])}"[:254], **props, ) return record
[docs]def handle_elbv2_target(record, elbv2, settings, root_stack, dns_settings): """ Function to define the TargetAlias properties for the record from an ELBv2 :param Record record: :param ecs_composex.elbv2.elbv2_stack Elbv2 elbv2: :param ecs_composex.common.settings.ComposeXSettings settings: :param ecs_composex.common.stacks.ComposeXStack root_stack: :param ecs_composex.dns.DnsSettings dns_settings: :return: """ if elbv2.lookup and not elbv2.cfn_resource: LOG.error("Cannot associate Lookedup ELBv2 at this time.") return elbv2_root_stack = root_stack.stack_template.resources[record.target_mod] elbv2.init_outputs() elbv2.generate_outputs() add_outputs(elbv2_root_stack.stack_template, elbv2.outputs) alias_tgt = AliasTarget( HostedZoneId=GetAtt( elbv2_root_stack.title, f"Outputs.{elbv2.logical_name}{LB_DNS_ZONE_ID.title}", ), DNSName=GetAtt( elbv2_root_stack.title, f"Outputs.{elbv2.logical_name}{LB_DNS_NAME.title}", ), ) record_props = import_record_properties(record.properties, BaseRecordSet) record_props["AliasTarget"] = alias_tgt record_props["Region"] = Ref(AWS_REGION) if ( dns_settings.public_zone.id_value or dns_settings.public_zone.cfn_resource ) and not keyisset("HostedZoneId", record_props): record_props["HostedZoneId"] = dns_settings.public_zone.id_value elif ( not dns_settings.public_zone or not dns_settings.public_zone.id_value or not dns_settings.public_zone.cfn_resource and not keyisset("HostedZoneId", record_props) ): raise ValueError( "No Public zone was defined, neither in x-dns or in the properties" ) if not keyisset("SetIdentifier", record_props): record_props["SetIdentifier"] = Ref(AWS_STACK_NAME) record.cfn_resource = RecordSetType( f"Elv2R53Record{record_props['Type']}{NONALPHANUM.sub('', record_props['Name'])}"[ :254 ], **record_props, ) root_stack.stack_template.add_resource(record.cfn_resource)
[docs]class Record(object): """ Class to represent a DNS record """ allowed_keys = [("Properties", dict), ("Target", str), ("Names", list)] alias_targets = [("elbv2", handle_elbv2_target), ("s3", None)] alias_target_types = [f"{X_KEY}{key[0]}" for key in alias_targets] def __init__(self, definition): self.definition = deepcopy(definition) self.target = None self.cfn_resource = None for prop in self.definition.keys(): if prop not in [key[0] for key in self.allowed_keys]: raise KeyError( f"Property {prop} is not valid. Only properties are", [key[0] for key in self.allowed_keys], ) for key in self.allowed_keys: if keyisset(key[0], self.definition) and not isinstance( self.definition[key[0]], key[1] ): raise TypeError( f"Property {key[0]} must be of type", key[1], "Got", type(self.definition[key[0]]), ) if keyisset("Target", self.definition): target_definition = self.definition["Target"].split("::") if not target_definition[0] in self.alias_target_types: raise ValueError( f"ComposeX resource type {target_definition[0]} is not valid. It must be one of", self.alias_target_types, ) self.target_type = target_definition[0] self.target_name = target_definition[1] self.target_mod = compile(X_KEY).sub("", self.target_type) else: self.target_type = None self.target_name = None self.properties = ( self.definition["Properties"] if keyisset("Properties", self.definition) else {} ) self.names = ( self.definition["Names"] if keyisset("Names", self.definition) else {} ) self.validate_names()
[docs] def validate_names(self): domain_name_re = compile( r"(?:(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9][a-z0-9-]{0,61}[a-z0-9])$" ) if self.names: for name in self.names: if not isinstance(name, str): raise TypeError( "the Names in the records definition must be a list of strings. Got", type(name), ) if not domain_name_re.match(name): raise NameError( "The value", name, "is invalid. domain name must comply with", domain_name_re.pattern, )
[docs] def map_record_to_target(self, settings, root_stack, dns_settings): """ Method to go and identify the target resource :param ecs_composex.common.settings.ComposeXSettings settings: :param ecs_composex.common.stacks.ComposeXStack root_stack: :param ecs_composex.dns.DnsSettings dns_settings: :return: """ if keyisset(self.target_type, settings.compose_content) and keyisset( self.target_name, settings.compose_content[self.target_type] ): target = settings.compose_content[self.target_type][self.target_name] for target_type in self.alias_targets: if target_type[0] == self.target_mod and target_type[1]: target_type[1](self, target, settings, root_stack, dns_settings) break
[docs]class DnsRecords(object): main_key = "Records" def __init__(self, settings): """ Function to update the DNS Records in compose content with the class :param ecs_composex.common.settings.ComposeXSettings settings: :return: """ self.records = [] if keyisset(RES_KEY, settings.compose_content) and keyisset( self.main_key, settings.compose_content[RES_KEY] ): for record in settings.compose_content[RES_KEY][self.main_key]: dns_record = Record(record) self.records.append(dns_record)
[docs] def associate_records_to_resources(self, settings, root_stack, dns_settings): for dns_record in self.records: if dns_record.target_name and dns_record.target_type: dns_record.map_record_to_target(settings, root_stack, dns_settings)