import logging from collections import namedtuple from dataclasses import dataclass from typing import Any, Dict, List, Optional, Set, Tuple, Union, cast from samtranslator.feature_toggle.feature_toggle import FeatureToggle from samtranslator.metrics.method_decorator import cw_timer from samtranslator.model import Resource from samtranslator.model.apigateway import ( ApiGatewayApiKey, ApiGatewayAuthorizer, ApiGatewayBasePathMapping, ApiGatewayBasePathMappingV2, ApiGatewayDeployment, ApiGatewayDomainName, ApiGatewayDomainNameAccessAssociation, ApiGatewayDomainNameV2, ApiGatewayResponse, ApiGatewayRestApi, ApiGatewayStage, ApiGatewayUsagePlan, ApiGatewayUsagePlanKey, ) from samtranslator.model.exceptions import ( ExpectedType, InvalidDocumentException, InvalidResourceException, InvalidTemplateException, ) from samtranslator.model.intrinsics import fnGetAtt, fnSub, is_intrinsic, make_or_condition, ref from samtranslator.model.lambda_ import LambdaPermission from samtranslator.model.route53 import Route53RecordSetGroup from samtranslator.model.s3_utils.uri_parser import parse_s3_uri from samtranslator.model.tags.resource_tagging import get_tag_list from samtranslator.model.types import PassThrough from samtranslator.region_configuration import RegionConfiguration from samtranslator.swagger.swagger import SwaggerEditor from samtranslator.translator.arn_generator import ArnGenerator from samtranslator.translator.logical_id_generator import LogicalIdGenerator from samtranslator.utils.py27hash_fix import Py27Dict, Py27UniStr from samtranslator.utils.types import Intrinsicable from samtranslator.utils.utils import InvalidValueType, dict_deep_get from samtranslator.validator.value_validator import sam_expect LOG = logging.getLogger(__name__) FEATURE_FLAG_NORMALIZED_OPENAPI_VERSION = "normalized_open_api_version" _CORS_WILDCARD = "'*'" CorsProperties = namedtuple( "CorsProperties", ["AllowMethods", "AllowHeaders", "AllowOrigin", "MaxAge", "AllowCredentials"] ) # Default the Cors Properties to '*' wildcard and False AllowCredentials. Other properties are actually Optional CorsProperties.__new__.__defaults__ = (None, None, _CORS_WILDCARD, None, False) AuthProperties = namedtuple( "AuthProperties", [ "Authorizers", "DefaultAuthorizer", "InvokeRole", "AddDefaultAuthorizerToCorsPreflight", "AddApiKeyRequiredToCorsPreflight", "ApiKeyRequired", "ResourcePolicy", "UsagePlan", ], ) AuthProperties.__new__.__defaults__ = (None, None, None, True, True, None, None, None) UsagePlanProperties = namedtuple( "UsagePlanProperties", ["CreateUsagePlan", "Description", "Quota", "Tags", "Throttle", "UsagePlanName"] ) UsagePlanProperties.__new__.__defaults__ = (None, None, None, None, None, None) GatewayResponseProperties = ["ResponseParameters", "ResponseTemplates", "StatusCode"] @dataclass class ApiDomainResponse: domain: Optional[ApiGatewayDomainName] apigw_basepath_mapping_list: Optional[List[ApiGatewayBasePathMapping]] recordset_group: Any @dataclass class ApiDomainResponseV2: domain: Optional[ApiGatewayDomainNameV2] apigw_basepath_mapping_list: Optional[List[ApiGatewayBasePathMappingV2]] recordset_group: Any domain_access_association: Any class SharedApiUsagePlan: """ Collects API information from different API resources in the same template, so that these information can be used in the shared usage plan """ SHARED_USAGE_PLAN_CONDITION_NAME = "SharedUsagePlanCondition" def __init__(self) -> None: self.usage_plan_shared = False self.stage_keys_shared: List[str] = [] self.api_stages_shared: List[str] = [] self.depends_on_shared: List[str] = [] # shared resource level attributes self.conditions: Set[str] = set() self.any_api_without_condition = False self.deletion_policy: Optional[str] = None self.update_replace_policy: Optional[str] = None def get_combined_resource_attributes(self, resource_attributes, conditions): # type: ignore[no-untyped-def] """ This method returns a dictionary which combines 'DeletionPolicy', 'UpdateReplacePolicy' and 'Condition' values of API definitions that could be used in Shared Usage Plan resources. Parameters ---------- resource_attributes: Dict[str] A dictionary of resource level attributes of the API resource conditions: Dict[str] Conditions section of the template """ self._set_deletion_policy(resource_attributes.get("DeletionPolicy")) # type: ignore[no-untyped-call] self._set_update_replace_policy(resource_attributes.get("UpdateReplacePolicy")) # type: ignore[no-untyped-call] self._set_condition(resource_attributes.get("Condition"), conditions) # type: ignore[no-untyped-call] combined_resource_attributes = {} if self.deletion_policy: combined_resource_attributes["DeletionPolicy"] = self.deletion_policy if self.update_replace_policy: combined_resource_attributes["UpdateReplacePolicy"] = self.update_replace_policy # do not set Condition if any of the API resource does not have Condition in it if self.conditions and not self.any_api_without_condition: combined_resource_attributes["Condition"] = SharedApiUsagePlan.SHARED_USAGE_PLAN_CONDITION_NAME return combined_resource_attributes def _set_deletion_policy(self, deletion_policy): # type: ignore[no-untyped-def] if deletion_policy: if self.deletion_policy: # update only if new deletion policy is Retain if deletion_policy == "Retain": self.deletion_policy = deletion_policy else: self.deletion_policy = deletion_policy def _set_update_replace_policy(self, update_replace_policy): # type: ignore[no-untyped-def] if update_replace_policy: if self.update_replace_policy: # if new value is Retain or # new value is retain and current value is Delete then update its value if (update_replace_policy == "Retain") or ( update_replace_policy == "Snapshot" and self.update_replace_policy == "Delete" ): self.update_replace_policy = update_replace_policy else: self.update_replace_policy = update_replace_policy def _set_condition(self, condition, template_conditions): # type: ignore[no-untyped-def] # if there are any API without condition, then skip if self.any_api_without_condition: return if condition and condition not in self.conditions: if template_conditions is None: raise InvalidTemplateException( "Can't have condition without having 'Conditions' section in the template" ) if self.conditions: self.conditions.add(condition) or_condition = make_or_condition(self.conditions) template_conditions[SharedApiUsagePlan.SHARED_USAGE_PLAN_CONDITION_NAME] = or_condition else: self.conditions.add(condition) template_conditions[SharedApiUsagePlan.SHARED_USAGE_PLAN_CONDITION_NAME] = condition elif condition is None: self.any_api_without_condition = True if template_conditions and SharedApiUsagePlan.SHARED_USAGE_PLAN_CONDITION_NAME in template_conditions: del template_conditions[SharedApiUsagePlan.SHARED_USAGE_PLAN_CONDITION_NAME] class ApiGenerator: def __init__( # noqa: PLR0913 self, logical_id: str, cache_cluster_enabled: Optional[Intrinsicable[bool]], cache_cluster_size: Optional[Intrinsicable[str]], variables: Optional[Dict[str, Any]], depends_on: Optional[List[str]], definition_body: Optional[Dict[str, Any]], definition_uri: Optional[Intrinsicable[str]], name: Optional[Intrinsicable[str]], stage_name: Optional[Intrinsicable[str]], shared_api_usage_plan: Any, template_conditions: Any, merge_definitions: Optional[bool] = None, tags: Optional[Dict[str, Any]] = None, endpoint_configuration: Optional[Dict[str, Any]] = None, method_settings: Optional[List[Any]] = None, binary_media: Optional[List[Any]] = None, minimum_compression_size: Optional[Intrinsicable[int]] = None, disable_execute_api_endpoint: Optional[Intrinsicable[bool]] = None, cors: Optional[Intrinsicable[str]] = None, auth: Optional[Dict[str, Any]] = None, gateway_responses: Optional[Dict[str, Any]] = None, access_log_setting: Optional[Dict[str, Any]] = None, canary_setting: Optional[Dict[str, Any]] = None, tracing_enabled: Optional[Intrinsicable[bool]] = None, resource_attributes: Optional[Dict[str, Any]] = None, passthrough_resource_attributes: Optional[Dict[str, Any]] = None, open_api_version: Optional[Intrinsicable[str]] = None, models: Optional[Dict[str, Any]] = None, domain: Optional[Dict[str, Any]] = None, fail_on_warnings: Optional[Intrinsicable[bool]] = None, description: Optional[Intrinsicable[str]] = None, mode: Optional[Intrinsicable[str]] = None, api_key_source_type: Optional[Intrinsicable[str]] = None, always_deploy: Optional[bool] = False, feature_toggle: Optional[FeatureToggle] = None, policy: Optional[Union[Dict[str, Any], Intrinsicable[str]]] = None, ): """Constructs an API Generator class that generates API Gateway resources :param logical_id: Logical id of the SAM API Resource :param cache_cluster_enabled: Whether cache cluster is enabled :param cache_cluster_size: Size of the cache cluster :param variables: API Gateway Variables :param depends_on: Any resources that need to be depended on :param definition_body: API definition :param definition_uri: URI to API definition :param name: Name of the API Gateway resource :param stage_name: Name of the Stage :param tags: Stage Tags :param access_log_setting: Whether to send access logs and where for Stage :param canary_setting: Canary Setting for Stage :param tracing_enabled: Whether active tracing with X-ray is enabled :param resource_attributes: Resource attributes to add to API resources :param passthrough_resource_attributes: Attributes such as `Condition` that are added to derived resources :param models: Model definitions to be used by API methods :param description: Description of the API Gateway resource """ self.logical_id = logical_id self.cache_cluster_enabled = cache_cluster_enabled self.cache_cluster_size = cache_cluster_size self.variables = variables self.depends_on = depends_on self.definition_body = definition_body self.definition_uri = definition_uri self.merge_definitions = merge_definitions self.name = name self.stage_name = stage_name self.tags = tags self.endpoint_configuration = endpoint_configuration self.method_settings = method_settings self.binary_media = binary_media self.minimum_compression_size = minimum_compression_size self.disable_execute_api_endpoint = disable_execute_api_endpoint self.cors = cors self.auth = auth self.gateway_responses = gateway_responses self.access_log_setting = access_log_setting self.canary_setting = canary_setting self.tracing_enabled = tracing_enabled self.resource_attributes = resource_attributes self.passthrough_resource_attributes = passthrough_resource_attributes self.open_api_version = open_api_version self.remove_extra_stage = open_api_version self.models = models self.domain = domain self.fail_on_warnings = fail_on_warnings self.description = description self.shared_api_usage_plan = shared_api_usage_plan self.template_conditions = template_conditions self.mode = mode self.api_key_source_type = api_key_source_type self.always_deploy = always_deploy self.feature_toggle = feature_toggle self.policy = policy def _construct_rest_api(self) -> ApiGatewayRestApi: """Constructs and returns the ApiGateway RestApi. :returns: the RestApi to which this SAM Api corresponds :rtype: model.apigateway.ApiGatewayRestApi """ self._validate_properties() rest_api = ApiGatewayRestApi(self.logical_id, depends_on=self.depends_on, attributes=self.resource_attributes) # NOTE: For backwards compatibility we need to retain BinaryMediaTypes on the CloudFormation Property # Removing this and only setting x-amazon-apigateway-binary-media-types results in other issues. rest_api.BinaryMediaTypes = self.binary_media rest_api.MinimumCompressionSize = self.minimum_compression_size if self.endpoint_configuration: self._set_endpoint_configuration(rest_api, self.endpoint_configuration) elif not RegionConfiguration.is_apigw_edge_configuration_supported(): # Since this region does not support EDGE configuration, we explicitly set the endpoint type # to Regional which is the only supported config. self._set_endpoint_configuration(rest_api, "REGIONAL") self._add_cors() self._add_auth() self._add_gateway_responses() self._add_binary_media_types() self._add_models() if self.fail_on_warnings: rest_api.FailOnWarnings = self.fail_on_warnings if self.disable_execute_api_endpoint is not None: self._add_endpoint_extension() if self.definition_uri: rest_api.BodyS3Location = self._construct_body_s3_dict() elif self.definition_body: # # Post Process OpenApi Auth Settings self.definition_body = self._openapi_postprocess(self.definition_body) rest_api.Body = self.definition_body if self.name: rest_api.Name = self.name if self.description: rest_api.Description = self.description if self.mode: rest_api.Mode = self.mode if self.api_key_source_type: rest_api.ApiKeySourceType = self.api_key_source_type if self.policy: rest_api.Policy = self.policy return rest_api def _validate_properties(self) -> None: if self.definition_uri and self.definition_body: raise InvalidResourceException( self.logical_id, "Specify either 'DefinitionUri' or 'DefinitionBody' property and not both." ) if self.definition_uri and self.merge_definitions: raise InvalidResourceException( self.logical_id, "Cannot set 'MergeDefinitions' to True when using `DefinitionUri`." ) if self.open_api_version and not SwaggerEditor.safe_compare_regex_with_string( SwaggerEditor.get_openapi_versions_supported_regex(), self.open_api_version ): raise InvalidResourceException(self.logical_id, "The OpenApiVersion value must be of the format '3.0.0'.") def _add_endpoint_extension(self) -> None: """Add disableExecuteApiEndpoint if it is set in SAM Note: If neither DefinitionUri nor DefinitionBody are specified, SAM will generate a openapi definition body based on template configuration. https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/sam-resource-api.html#sam-api-definitionbody For this reason, we always put DisableExecuteApiEndpoint into openapi object irrespective of origin of DefinitionBody. """ if self.disable_execute_api_endpoint is not None and not self.definition_body: raise InvalidResourceException( self.logical_id, "DisableExecuteApiEndpoint works only within 'DefinitionBody' property." ) editor = SwaggerEditor(self.definition_body) editor.add_disable_execute_api_endpoint_extension(self.disable_execute_api_endpoint) self.definition_body = editor.swagger def _construct_body_s3_dict(self) -> Dict[str, Any]: """Constructs the RestApi's `BodyS3Location property`_, from the SAM Api's DefinitionUri property. :returns: a BodyS3Location dict, containing the S3 Bucket, Key, and Version of the Swagger definition :rtype: dict """ if isinstance(self.definition_uri, dict): if not self.definition_uri.get("Bucket", None) or not self.definition_uri.get("Key", None): # DefinitionUri is a dictionary but does not contain Bucket or Key property raise InvalidResourceException( self.logical_id, "'DefinitionUri' requires Bucket and Key properties to be specified." ) s3_pointer = self.definition_uri else: # DefinitionUri is a string _parsed_s3_pointer = parse_s3_uri(self.definition_uri) if _parsed_s3_pointer is None: raise InvalidResourceException( self.logical_id, "'DefinitionUri' is not a valid S3 Uri of the form " "'s3://bucket/key' with optional versionId query parameter.", ) s3_pointer = _parsed_s3_pointer if isinstance(self.definition_uri, Py27UniStr): # self.defintion_uri is a Py27UniStr instance if it is defined in the template # we need to preserve the Py27UniStr type s3_pointer["Bucket"] = Py27UniStr(s3_pointer["Bucket"]) s3_pointer["Key"] = Py27UniStr(s3_pointer["Key"]) if "Version" in s3_pointer: s3_pointer["Version"] = Py27UniStr(s3_pointer["Version"]) # Construct body_s3 as py27 dict body_s3 = Py27Dict() body_s3["Bucket"] = s3_pointer["Bucket"] body_s3["Key"] = s3_pointer["Key"] if "Version" in s3_pointer: body_s3["Version"] = s3_pointer["Version"] return body_s3 def _construct_deployment(self, rest_api: ApiGatewayRestApi) -> ApiGatewayDeployment: """Constructs and returns the ApiGateway Deployment. :param model.apigateway.ApiGatewayRestApi rest_api: the RestApi for this Deployment :returns: the Deployment to which this SAM Api corresponds :rtype: model.apigateway.ApiGatewayDeployment """ deployment = ApiGatewayDeployment( self.logical_id + "Deployment", attributes=self.passthrough_resource_attributes ) deployment.RestApiId = rest_api.get_runtime_attr("rest_api_id") if not self.remove_extra_stage: deployment.StageName = "Stage" return deployment def _construct_stage( self, deployment: ApiGatewayDeployment, swagger: Optional[Dict[str, Any]], redeploy_restapi_parameters: Any ) -> ApiGatewayStage: """Constructs and returns the ApiGateway Stage. :param model.apigateway.ApiGatewayDeployment deployment: the Deployment for this Stage :returns: the Stage to which this SAM Api corresponds :rtype: model.apigateway.ApiGatewayStage """ # If StageName is some intrinsic function, then don't prefix the Stage's logical ID # This will NOT create duplicates because we allow only ONE stage per API resource stage_name_prefix = self.stage_name if isinstance(self.stage_name, str) else "" if stage_name_prefix.isalnum(): stage_logical_id = self.logical_id + stage_name_prefix + "Stage" else: generator = LogicalIdGenerator(self.logical_id + "Stage", stage_name_prefix) stage_logical_id = generator.gen() stage = ApiGatewayStage(stage_logical_id, attributes=self.passthrough_resource_attributes) stage.RestApiId = ref(self.logical_id) stage.update_deployment_ref(deployment.logical_id) stage.StageName = self.stage_name stage.CacheClusterEnabled = self.cache_cluster_enabled stage.CacheClusterSize = self.cache_cluster_size stage.Variables = self.variables stage.MethodSettings = self.method_settings stage.AccessLogSetting = self.access_log_setting stage.CanarySetting = self.canary_setting stage.TracingEnabled = self.tracing_enabled if swagger is not None: deployment.make_auto_deployable( stage, self.remove_extra_stage, swagger, self.domain, redeploy_restapi_parameters, self.always_deploy, ) if self.tags is not None: stage.Tags = get_tag_list(self.tags) return stage def _construct_api_domain( # noqa: PLR0912, PLR0915 self, rest_api: ApiGatewayRestApi, route53_record_set_groups: Any ) -> ApiDomainResponse: """ Constructs and returns the ApiGateway Domain and BasepathMapping """ if self.domain is None: return ApiDomainResponse(None, None, None) sam_expect(self.domain, self.logical_id, "Domain").to_be_a_map() domain_name: PassThrough = sam_expect( self.domain.get("DomainName"), self.logical_id, "Domain.DomainName" ).to_not_be_none() certificate_arn: PassThrough = sam_expect( self.domain.get("CertificateArn"), self.logical_id, "Domain.CertificateArn" ).to_not_be_none() api_domain_name = "{}{}".format("ApiGatewayDomainName", LogicalIdGenerator("", domain_name).gen()) self.domain["ApiDomainName"] = api_domain_name domain = ApiGatewayDomainName(api_domain_name, attributes=self.passthrough_resource_attributes) domain.DomainName = domain_name endpoint = self.domain.get("EndpointConfiguration") if endpoint is None: endpoint = "REGIONAL" self.domain["EndpointConfiguration"] = "REGIONAL" elif endpoint not in ["EDGE", "REGIONAL", "PRIVATE"]: raise InvalidResourceException( self.logical_id, "EndpointConfiguration for Custom Domains must be" " one of {}.".format(["EDGE", "REGIONAL", "PRIVATE"]), ) if endpoint == "REGIONAL": domain.RegionalCertificateArn = certificate_arn else: domain.CertificateArn = certificate_arn domain.EndpointConfiguration = {"Types": [endpoint]} mutual_tls_auth = self.domain.get("MutualTlsAuthentication", None) if mutual_tls_auth: sam_expect(mutual_tls_auth, self.logical_id, "Domain.MutualTlsAuthentication").to_be_a_map() if not set(mutual_tls_auth.keys()).issubset({"TruststoreUri", "TruststoreVersion"}): invalid_keys = [] for key in mutual_tls_auth: if key not in {"TruststoreUri", "TruststoreVersion"}: invalid_keys.append(key) invalid_keys.sort() raise InvalidResourceException( self.logical_id, "Available Domain.MutualTlsAuthentication fields are {}.".format( ["TruststoreUri", "TruststoreVersion"] ), ) domain.MutualTlsAuthentication = {} if mutual_tls_auth.get("TruststoreUri", None): domain.MutualTlsAuthentication["TruststoreUri"] = mutual_tls_auth["TruststoreUri"] if mutual_tls_auth.get("TruststoreVersion", None): domain.MutualTlsAuthentication["TruststoreVersion"] = mutual_tls_auth["TruststoreVersion"] self._set_optional_domain_properties(domain) basepaths: Optional[List[str]] basepath_value = self.domain.get("BasePath") # Create BasepathMappings if self.domain.get("BasePath") and isinstance(basepath_value, str): basepaths = [basepath_value] elif self.domain.get("BasePath") and isinstance(basepath_value, list): basepaths = cast(Optional[List[Any]], basepath_value) else: basepaths = None # Boolean to allow/disallow symbols in BasePath property normalize_basepath = self.domain.get("NormalizeBasePath", True) basepath_resource_list: List[ApiGatewayBasePathMapping] = [] if basepaths is None: basepath_mapping = self._create_basepath_mapping(api_domain_name, rest_api, None, None) basepath_resource_list.extend([basepath_mapping]) else: sam_expect(basepaths, self.logical_id, "Domain.BasePath").to_be_a_list_of(ExpectedType.STRING) for basepath in basepaths: # Remove possible leading and trailing '/' because a base path may only # contain letters, numbers, and one of "$-_.+!*'()" path = "".join(e for e in basepath if e.isalnum()) mapping_basepath = path if normalize_basepath else basepath logical_id = "{}{}{}".format(self.logical_id, path, "BasePathMapping") basepath_mapping = self._create_basepath_mapping( api_domain_name, rest_api, logical_id, mapping_basepath ) basepath_resource_list.extend([basepath_mapping]) # Create the Route53 RecordSetGroup resource record_set_group = None route53 = self.domain.get("Route53") if route53 is not None: sam_expect(route53, self.logical_id, "Domain.Route53").to_be_a_map() if route53.get("HostedZoneId") is None and route53.get("HostedZoneName") is None: raise InvalidResourceException( self.logical_id, "HostedZoneId or HostedZoneName is required to enable Route53 support on Custom Domains.", ) logical_id_suffix = LogicalIdGenerator( "", route53.get("HostedZoneId") or route53.get("HostedZoneName") ).gen() logical_id = "RecordSetGroup" + logical_id_suffix record_set_group = route53_record_set_groups.get(logical_id) if route53.get("SeparateRecordSetGroup"): sam_expect( route53.get("SeparateRecordSetGroup"), self.logical_id, "Domain.Route53.SeparateRecordSetGroup" ).to_be_a_bool() return ApiDomainResponse( domain, basepath_resource_list, self._construct_single_record_set_group(self.domain, api_domain_name, route53), ) if not record_set_group: record_set_group = self._get_record_set_group(logical_id, route53) route53_record_set_groups[logical_id] = record_set_group record_set_group.RecordSets += self._construct_record_sets_for_domain(self.domain, api_domain_name, route53) return ApiDomainResponse(domain, basepath_resource_list, record_set_group) def _construct_api_domain_v2( self, rest_api: ApiGatewayRestApi, route53_record_set_groups: Any ) -> ApiDomainResponseV2: """ Constructs and returns the ApiGateway Domain V2 and BasepathMapping V2 """ if self.domain is None: return ApiDomainResponseV2(None, None, None, None) sam_expect(self.domain, self.logical_id, "Domain").to_be_a_map() domain_name: PassThrough = sam_expect( self.domain.get("DomainName"), self.logical_id, "Domain.DomainName" ).to_not_be_none() certificate_arn: PassThrough = sam_expect( self.domain.get("CertificateArn"), self.logical_id, "Domain.CertificateArn" ).to_not_be_none() api_domain_name = "{}{}".format("ApiGatewayDomainNameV2", LogicalIdGenerator("", domain_name).gen()) domain_name_arn = ref(api_domain_name) domain = ApiGatewayDomainNameV2(api_domain_name, attributes=self.passthrough_resource_attributes) domain.DomainName = domain_name endpoint = self.domain.get("EndpointConfiguration") if endpoint not in ["EDGE", "REGIONAL", "PRIVATE"]: raise InvalidResourceException( self.logical_id, "EndpointConfiguration for Custom Domains must be" " one of {}.".format(["EDGE", "REGIONAL", "PRIVATE"]), ) domain.CertificateArn = certificate_arn domain.EndpointConfiguration = {"Types": [endpoint]} self._set_optional_domain_properties(domain) basepaths: Optional[List[str]] = self._get_basepaths() # Boolean to allow/disallow symbols in BasePath property normalize_basepath = self.domain.get("NormalizeBasePath", True) basepath_resource_list: List[ApiGatewayBasePathMappingV2] = [] if basepaths is None: basepath_mapping = self._create_basepath_mapping_v2(domain_name_arn, rest_api) basepath_resource_list.extend([basepath_mapping]) else: sam_expect(basepaths, self.logical_id, "Domain.BasePath").to_be_a_list_of(ExpectedType.STRING) for basepath in basepaths: # Remove possible leading and trailing '/' because a base path may only # contain letters, numbers, and one of "$-_.+!*'()" path = "".join(e for e in basepath if e.isalnum()) logical_id = "{}{}{}".format(self.logical_id, path, "BasePathMapping") basepath_mapping = ApiGatewayBasePathMappingV2( logical_id, attributes=self.passthrough_resource_attributes ) basepath_mapping.DomainNameArn = domain_name_arn basepath_mapping.RestApiId = ref(rest_api.logical_id) basepath_mapping.Stage = ref(rest_api.logical_id + ".Stage") basepath_mapping.BasePath = path if normalize_basepath else basepath basepath_resource_list.extend([basepath_mapping]) # Create the DomainNameAccessAssociation domain_access_association = self.domain.get("AccessAssociation") domain_access_association_resource = None if domain_access_association is not None: domain_access_association_resource = self._generate_domain_access_association( domain_access_association, domain_name_arn, api_domain_name ) # Create the Route53 RecordSetGroup resource record_set_group = None route53 = self.domain.get("Route53") if route53 is not None: sam_expect(route53, self.logical_id, "Domain.Route53").to_be_a_map() if route53.get("HostedZoneId") is None and route53.get("HostedZoneName") is None: raise InvalidResourceException( self.logical_id, "HostedZoneId or HostedZoneName is required to enable Route53 support on Custom Domains.", ) logical_id_suffix = LogicalIdGenerator( "", route53.get("HostedZoneId") or route53.get("HostedZoneName") ).gen() logical_id = "RecordSetGroup" + logical_id_suffix record_set_group = route53_record_set_groups.get(logical_id) if route53.get("SeparateRecordSetGroup"): sam_expect( route53.get("SeparateRecordSetGroup"), self.logical_id, "Domain.Route53.SeparateRecordSetGroup" ).to_be_a_bool() return ApiDomainResponseV2( domain, basepath_resource_list, self._construct_single_record_set_group(self.domain, domain_name, route53), domain_access_association_resource, ) if not record_set_group: record_set_group = self._get_record_set_group(logical_id, route53) route53_record_set_groups[logical_id] = record_set_group record_set_group.RecordSets += self._construct_record_sets_for_domain(self.domain, domain_name, route53) return ApiDomainResponseV2(domain, basepath_resource_list, record_set_group, domain_access_association_resource) def _get_basepaths(self) -> Optional[List[str]]: if self.domain is None: return None basepath_value = self.domain.get("BasePath") if self.domain.get("BasePath") and isinstance(basepath_value, str): return [basepath_value] if self.domain.get("BasePath") and isinstance(basepath_value, list): return cast(Optional[List[Any]], basepath_value) return None def _set_optional_domain_properties(self, domain: Union[ApiGatewayDomainName, ApiGatewayDomainNameV2]) -> None: if self.domain is None: return if self.domain.get("SecurityPolicy", None): domain.SecurityPolicy = self.domain["SecurityPolicy"] if self.domain.get("Policy", None): domain.Policy = self.domain["Policy"] if self.domain.get("OwnershipVerificationCertificateArn", None): domain.OwnershipVerificationCertificateArn = self.domain["OwnershipVerificationCertificateArn"] def _get_record_set_group(self, logical_id: str, route53: Dict[str, Any]) -> Route53RecordSetGroup: record_set_group = Route53RecordSetGroup(logical_id, attributes=self.passthrough_resource_attributes) if "HostedZoneId" in route53: record_set_group.HostedZoneId = route53.get("HostedZoneId") if "HostedZoneName" in route53: record_set_group.HostedZoneName = route53.get("HostedZoneName") record_set_group.RecordSets = [] return record_set_group def _construct_single_record_set_group( self, domain: Dict[str, Any], api_domain_name: str, route53: Any ) -> Route53RecordSetGroup: hostedZoneId = route53.get("HostedZoneId") hostedZoneName = route53.get("HostedZoneName") domainName = domain.get("DomainName") logical_id = LogicalIdGenerator("RecordSetGroup", [hostedZoneId or hostedZoneName, domainName]).gen() record_set_group = Route53RecordSetGroup(logical_id, attributes=self.passthrough_resource_attributes) if hostedZoneId: record_set_group.HostedZoneId = hostedZoneId if hostedZoneName: record_set_group.HostedZoneName = hostedZoneName record_set_group.RecordSets = [] record_set_group.RecordSets += self._construct_record_sets_for_domain(domain, api_domain_name, route53) return record_set_group def _construct_record_sets_for_domain( self, custom_domain_config: Dict[str, Any], api_domain_name: str, route53_config: Dict[str, Any] ) -> List[Dict[str, Any]]: recordset_list = [] alias_target = self._construct_alias_target(custom_domain_config, api_domain_name, route53_config) recordset = {} recordset["Name"] = custom_domain_config.get("DomainName") recordset["Type"] = "A" recordset["AliasTarget"] = alias_target self._update_route53_routing_policy_properties(route53_config, recordset) recordset_list.append(recordset) if route53_config.get("IpV6") is not None and route53_config.get("IpV6") is True: recordset_ipv6 = {} recordset_ipv6["Name"] = custom_domain_config.get("DomainName") recordset_ipv6["Type"] = "AAAA" recordset_ipv6["AliasTarget"] = alias_target self._update_route53_routing_policy_properties(route53_config, recordset_ipv6) recordset_list.append(recordset_ipv6) return recordset_list @staticmethod def _update_route53_routing_policy_properties(route53_config: Dict[str, Any], recordset: Dict[str, Any]) -> None: if route53_config.get("Region") is not None: recordset["Region"] = route53_config.get("Region") if route53_config.get("SetIdentifier") is not None: recordset["SetIdentifier"] = route53_config.get("SetIdentifier") def _construct_alias_target(self, domain: Dict[str, Any], api_domain_name: str, route53: Any) -> Dict[str, Any]: alias_target = {} target_health = route53.get("EvaluateTargetHealth") if target_health is not None: alias_target["EvaluateTargetHealth"] = target_health if domain.get("EndpointConfiguration") == "REGIONAL": alias_target["HostedZoneId"] = fnGetAtt(api_domain_name, "RegionalHostedZoneId") alias_target["DNSName"] = fnGetAtt(api_domain_name, "RegionalDomainName") elif domain.get("EndpointConfiguration") == "EDGE": if route53.get("DistributionDomainName") is None: route53["DistributionDomainName"] = fnGetAtt(api_domain_name, "DistributionDomainName") alias_target["HostedZoneId"] = "Z2FDTNDATAQYW2" alias_target["DNSName"] = route53.get("DistributionDomainName") else: alias_target["HostedZoneId"] = route53.get("VpcEndpointHostedZoneId") alias_target["DNSName"] = route53.get("VpcEndpointDomainName") return alias_target def _create_basepath_mapping( self, api_domain_name: PassThrough, rest_api: ApiGatewayRestApi, logical_id: Optional[str], basepath: Optional[str], ) -> ApiGatewayBasePathMapping: basepath_mapping: ApiGatewayBasePathMapping basepath_mapping = ( ApiGatewayBasePathMapping(logical_id, attributes=self.passthrough_resource_attributes) if logical_id else ApiGatewayBasePathMapping( self.logical_id + "BasePathMapping", attributes=self.passthrough_resource_attributes ) ) basepath_mapping.DomainName = ref(api_domain_name) basepath_mapping.RestApiId = ref(rest_api.logical_id) basepath_mapping.Stage = ref(rest_api.logical_id + ".Stage") if basepath is not None: basepath_mapping.BasePath = basepath return basepath_mapping def _create_basepath_mapping_v2( self, domain_name_arn: PassThrough, rest_api: ApiGatewayRestApi ) -> ApiGatewayBasePathMappingV2: basepath_mapping = ApiGatewayBasePathMappingV2( self.logical_id + "BasePathMapping", attributes=self.passthrough_resource_attributes ) basepath_mapping.DomainNameArn = domain_name_arn basepath_mapping.RestApiId = ref(rest_api.logical_id) basepath_mapping.Stage = ref(rest_api.logical_id + ".Stage") return basepath_mapping @cw_timer(prefix="Generator", name="Api") def to_cloudformation( self, redeploy_restapi_parameters: Optional[Any], route53_record_set_groups: Dict[str, Route53RecordSetGroup] ) -> List[Resource]: """Generates CloudFormation resources from a SAM API resource :returns: a tuple containing the RestApi, Deployment, and Stage for an empty Api. :rtype: tuple """ api_domain_response: Union[ApiDomainResponseV2, ApiDomainResponse] domain: Union[Resource, None] basepath_mapping: Union[List[ApiGatewayBasePathMapping], List[ApiGatewayBasePathMappingV2], None] rest_api = self._construct_rest_api() is_private_domain = isinstance(self.domain, dict) and self.domain.get("EndpointConfiguration") == "PRIVATE" api_domain_response = ( self._construct_api_domain_v2(rest_api, route53_record_set_groups) if is_private_domain else self._construct_api_domain(rest_api, route53_record_set_groups) ) domain_access_association = None if is_private_domain: domain_access_association = cast(ApiDomainResponseV2, api_domain_response).domain_access_association domain = api_domain_response.domain basepath_mapping = api_domain_response.apigw_basepath_mapping_list route53_recordsetGroup = api_domain_response.recordset_group deployment = self._construct_deployment(rest_api) swagger = None if rest_api.Body is not None: swagger = rest_api.Body elif rest_api.BodyS3Location is not None: swagger = rest_api.BodyS3Location stage = self._construct_stage(deployment, swagger, redeploy_restapi_parameters) permissions = self._construct_authorizer_lambda_permission() usage_plan = self._construct_usage_plan(rest_api_stage=stage) # mypy complains if the type in List doesn't match exactly # TODO: refactor to have a list of single resource generated_resources: List[ Union[ Optional[Resource], List[Resource], Tuple[Resource], List[LambdaPermission], List[ApiGatewayBasePathMapping], List[ApiGatewayBasePathMappingV2], ], ] = [] generated_resources.extend( [ rest_api, deployment, stage, permissions, domain, basepath_mapping, route53_recordsetGroup, usage_plan, ] ) if domain_access_association is not None: generated_resources.append(domain_access_association) # Make a list of single resources generated_resources_list: List[Resource] = [] for resource in generated_resources: if resource: if isinstance(resource, (list, tuple)): generated_resources_list.extend(resource) else: generated_resources_list.extend([resource]) return generated_resources_list def _add_cors(self) -> None: """ Add CORS configuration to the Swagger file, if necessary """ INVALID_ERROR = "Invalid value for 'Cors' property" if not self.cors: return if self.cors and not self.definition_body: raise InvalidResourceException( self.logical_id, "Cors works only with inline Swagger specified in 'DefinitionBody' property." ) if isinstance(self.cors, str) or is_intrinsic(self.cors): # Just set Origin property. Others will be defaults properties = CorsProperties(AllowOrigin=self.cors) # type: ignore[call-arg] elif isinstance(self.cors, dict): # Make sure keys in the dict are recognized for key in self.cors: if key not in CorsProperties._fields: raise InvalidResourceException(self.logical_id, f"Invalid key '{key}' for 'Cors' property.") properties = CorsProperties(**self.cors) else: raise InvalidResourceException(self.logical_id, INVALID_ERROR) if not SwaggerEditor.is_valid(self.definition_body): raise InvalidResourceException( self.logical_id, "Unable to add Cors configuration because " "'DefinitionBody' does not contain a valid Swagger definition.", ) if properties.AllowCredentials is True and properties.AllowOrigin == _CORS_WILDCARD: raise InvalidResourceException( self.logical_id, "Unable to add Cors configuration because " "'AllowCredentials' can not be true when " "'AllowOrigin' is \"'*'\" or not set", ) editor = SwaggerEditor(self.definition_body) for path in editor.iter_on_path(): try: editor.add_cors( # type: ignore[no-untyped-call] path, properties.AllowOrigin, properties.AllowHeaders, properties.AllowMethods, max_age=properties.MaxAge, allow_credentials=properties.AllowCredentials, ) except InvalidTemplateException as ex: raise InvalidResourceException(self.logical_id, ex.message) from ex # Assign the Swagger back to template self.definition_body = editor.swagger def _add_binary_media_types(self) -> None: """ Add binary media types to Swagger """ if not self.binary_media: return # We don't raise an error here like we do for similar cases because that would be backwards incompatible if self.binary_media and not self.definition_body: return editor = SwaggerEditor(self.definition_body) editor.add_binary_media_types(self.binary_media) # type: ignore[no-untyped-call] # Assign the Swagger back to template self.definition_body = editor.swagger def _add_auth(self) -> None: """ Add Auth configuration to the Swagger file, if necessary """ if not self.auth: return if self.auth and not self.definition_body: raise InvalidResourceException( self.logical_id, "Auth works only with inline Swagger specified in 'DefinitionBody' property." ) # Make sure keys in the dict are recognized if not all(key in AuthProperties._fields for key in self.auth): raise InvalidResourceException(self.logical_id, "Invalid value for 'Auth' property") if not SwaggerEditor.is_valid(self.definition_body): raise InvalidResourceException( self.logical_id, "Unable to add Auth configuration because " "'DefinitionBody' does not contain a valid Swagger definition.", ) swagger_editor = SwaggerEditor(self.definition_body) auth_properties = AuthProperties(**self.auth) authorizers = self._get_authorizers(auth_properties.Authorizers, auth_properties.DefaultAuthorizer) # type: ignore[no-untyped-call] if authorizers: swagger_editor.add_authorizers_security_definitions(authorizers) # type: ignore[no-untyped-call] self._set_default_authorizer( swagger_editor, authorizers, auth_properties.DefaultAuthorizer, auth_properties.AddDefaultAuthorizerToCorsPreflight, ) if auth_properties.ApiKeyRequired: swagger_editor.add_apikey_security_definition() self._set_default_apikey_required(swagger_editor, auth_properties.AddApiKeyRequiredToCorsPreflight) if auth_properties.ResourcePolicy: SwaggerEditor.validate_is_dict( auth_properties.ResourcePolicy, "ResourcePolicy must be a map (ResourcePolicyStatement)." ) for path in swagger_editor.iter_on_path(): swagger_editor.add_resource_policy(auth_properties.ResourcePolicy, path, self.stage_name) if auth_properties.ResourcePolicy.get("CustomStatements"): swagger_editor.add_custom_statements(auth_properties.ResourcePolicy.get("CustomStatements")) # type: ignore[no-untyped-call] self.definition_body = self._openapi_postprocess(swagger_editor.swagger) def _construct_usage_plan(self, rest_api_stage: Optional[ApiGatewayStage] = None) -> Any: # noqa: PLR0912 """Constructs and returns the ApiGateway UsagePlan, ApiGateway UsagePlanKey, ApiGateway ApiKey for Auth. :param model.apigateway.ApiGatewayStage stage: the stage of rest api :returns: UsagePlan, UsagePlanKey, ApiKey for this rest Api :rtype: model.apigateway.ApiGatewayUsagePlan, model.apigateway.ApiGatewayUsagePlanKey, model.apigateway.ApiGatewayApiKey """ create_usage_plans_accepted_values = ["SHARED", "PER_API", "NONE"] if not self.auth: return [] auth_properties = AuthProperties(**self.auth) if auth_properties.UsagePlan is None: return [] usage_plan_properties = auth_properties.UsagePlan # throws error if UsagePlan is not a dict if not isinstance(usage_plan_properties, dict): raise InvalidResourceException(self.logical_id, "'UsagePlan' must be a dictionary") # throws error if the property invalid/ unsupported for UsagePlan if not all(key in UsagePlanProperties._fields for key in usage_plan_properties): raise InvalidResourceException(self.logical_id, "Invalid property for 'UsagePlan'") create_usage_plan = usage_plan_properties.get("CreateUsagePlan") usage_plan: Optional[ApiGatewayUsagePlan] = None api_key = None usage_plan_key = None if create_usage_plan is None: raise InvalidResourceException(self.logical_id, "'CreateUsagePlan' is a required field for UsagePlan.") if create_usage_plan not in create_usage_plans_accepted_values: raise InvalidResourceException( self.logical_id, f"'CreateUsagePlan' accepts one of {create_usage_plans_accepted_values}." ) if create_usage_plan == "NONE": return [] if not rest_api_stage: return [] # create usage plan for this api only if usage_plan_properties.get("CreateUsagePlan") == "PER_API": usage_plan_logical_id = self.logical_id + "UsagePlan" usage_plan = ApiGatewayUsagePlan( logical_id=usage_plan_logical_id, depends_on=[self.logical_id], attributes=self.passthrough_resource_attributes, ) api_stages = [] api_stage = {} api_stage["ApiId"] = ref(self.logical_id) api_stage["Stage"] = ref(rest_api_stage.logical_id) api_stages.append(api_stage) usage_plan.ApiStages = api_stages api_key = self._construct_api_key(usage_plan_logical_id, create_usage_plan, rest_api_stage) usage_plan_key = self._construct_usage_plan_key(usage_plan_logical_id, create_usage_plan, api_key) # create a usage plan for all the Apis elif create_usage_plan == "SHARED": LOG.info("Creating SHARED usage plan for all the Apis") usage_plan_logical_id = "ServerlessUsagePlan" if self.logical_id not in self.shared_api_usage_plan.depends_on_shared: self.shared_api_usage_plan.depends_on_shared.append(self.logical_id) usage_plan = ApiGatewayUsagePlan( logical_id=usage_plan_logical_id, depends_on=self.shared_api_usage_plan.depends_on_shared, attributes=self.shared_api_usage_plan.get_combined_resource_attributes( self.passthrough_resource_attributes, self.template_conditions ), ) api_stage = {} api_stage["ApiId"] = ref(self.logical_id) api_stage["Stage"] = ref(rest_api_stage.logical_id) if api_stage not in self.shared_api_usage_plan.api_stages_shared: self.shared_api_usage_plan.api_stages_shared.append(api_stage) usage_plan.ApiStages = self.shared_api_usage_plan.api_stages_shared api_key = self._construct_api_key(usage_plan_logical_id, create_usage_plan, rest_api_stage) usage_plan_key = self._construct_usage_plan_key(usage_plan_logical_id, create_usage_plan, api_key) for name in ["UsagePlanName", "Description", "Quota", "Tags", "Throttle"]: if usage_plan and usage_plan_properties.get(name): setattr(usage_plan, name, usage_plan_properties.get(name)) return usage_plan, api_key, usage_plan_key def _construct_api_key( self, usage_plan_logical_id: str, create_usage_plan: Any, rest_api_stage: ApiGatewayStage ) -> ApiGatewayApiKey: """ :param usage_plan_logical_id: String :param create_usage_plan: String :param rest_api_stage: model.apigateway.ApiGatewayStage stage: the stage of rest api :return: api_key model.apigateway.ApiGatewayApiKey resource which is created for the given usage plan """ if create_usage_plan == "SHARED": # create an api key resource for all the apis LOG.info("Creating api key resource for all the Apis from SHARED usage plan") api_key_logical_id = "ServerlessApiKey" api_key = ApiGatewayApiKey( logical_id=api_key_logical_id, depends_on=[usage_plan_logical_id], attributes=self.shared_api_usage_plan.get_combined_resource_attributes( self.passthrough_resource_attributes, self.template_conditions ), ) api_key.Enabled = True stage_key = {} stage_key["RestApiId"] = ref(self.logical_id) stage_key["StageName"] = ref(rest_api_stage.logical_id) if stage_key not in self.shared_api_usage_plan.stage_keys_shared: self.shared_api_usage_plan.stage_keys_shared.append(stage_key) api_key.StageKeys = self.shared_api_usage_plan.stage_keys_shared # for create_usage_plan = "PER_API" else: # create an api key resource for this api api_key_logical_id = self.logical_id + "ApiKey" api_key = ApiGatewayApiKey( logical_id=api_key_logical_id, depends_on=[usage_plan_logical_id], attributes=self.passthrough_resource_attributes, ) api_key.Enabled = True stage_keys = [] stage_key = {} stage_key["RestApiId"] = ref(self.logical_id) stage_key["StageName"] = ref(rest_api_stage.logical_id) stage_keys.append(stage_key) api_key.StageKeys = stage_keys return api_key def _construct_usage_plan_key( self, usage_plan_logical_id: str, create_usage_plan: Any, api_key: ApiGatewayApiKey ) -> ApiGatewayUsagePlanKey: """ :param usage_plan_logical_id: String :param create_usage_plan: String :param api_key: model.apigateway.ApiGatewayApiKey resource :return: model.apigateway.ApiGatewayUsagePlanKey resource that contains the mapping between usage plan and api key """ if create_usage_plan == "SHARED": # create a mapping between api key and the usage plan usage_plan_key_logical_id = "ServerlessUsagePlanKey" resource_attributes = self.shared_api_usage_plan.get_combined_resource_attributes( self.passthrough_resource_attributes, self.template_conditions ) # for create_usage_plan = "PER_API" else: # create a mapping between api key and the usage plan usage_plan_key_logical_id = self.logical_id + "UsagePlanKey" resource_attributes = self.passthrough_resource_attributes usage_plan_key = ApiGatewayUsagePlanKey( logical_id=usage_plan_key_logical_id, depends_on=[api_key.logical_id], attributes=resource_attributes, ) usage_plan_key.KeyId = ref(api_key.logical_id) usage_plan_key.KeyType = "API_KEY" usage_plan_key.UsagePlanId = ref(usage_plan_logical_id) return usage_plan_key def _add_gateway_responses(self) -> None: """ Add Gateway Response configuration to the Swagger file, if necessary """ if not self.gateway_responses: return if self.gateway_responses and not self.definition_body: raise InvalidResourceException( self.logical_id, "GatewayResponses works only with inline Swagger specified in 'DefinitionBody' property.", ) # Make sure keys in the dict are recognized for responses_key, responses_value in self.gateway_responses.items(): if is_intrinsic(responses_value): # TODO: Add intrinsic support for this field. raise InvalidResourceException( self.logical_id, "Unable to set GatewayResponses attribute because " "intrinsic functions are not supported for this field.", ) if not isinstance(responses_value, dict): raise InvalidResourceException( self.logical_id, f"Invalid property type '{type(responses_value).__name__}' for GatewayResponses. " "Expected an object of type 'GatewayResponse'.", ) for response_key in responses_value: if response_key not in GatewayResponseProperties: raise InvalidResourceException( self.logical_id, f"Invalid property '{response_key}' in 'GatewayResponses' property '{responses_key}'.", ) if not SwaggerEditor.is_valid(self.definition_body): raise InvalidResourceException( self.logical_id, "Unable to add Auth configuration because " "'DefinitionBody' does not contain a valid Swagger definition.", ) swagger_editor = SwaggerEditor(self.definition_body) # The dicts below will eventually become part of swagger/openapi definition, thus requires using Py27Dict() gateway_responses = Py27Dict() for response_type, response in self.gateway_responses.items(): sam_expect(response, self.logical_id, f"GatewayResponses.{response_type}").to_be_a_map() response_parameters = response.get("ResponseParameters", Py27Dict()) response_templates = response.get("ResponseTemplates", Py27Dict()) if response_parameters: sam_expect( response_parameters, self.logical_id, f"GatewayResponses.{response_type}.ResponseParameters" ).to_be_a_map() gateway_responses[response_type] = ApiGatewayResponse( api_logical_id=self.logical_id, response_parameters=response_parameters, response_templates=response_templates, status_code=response.get("StatusCode", None), ) if gateway_responses: swagger_editor.add_gateway_responses(gateway_responses) # type: ignore[no-untyped-call] # Assign the Swagger back to template self.definition_body = swagger_editor.swagger def _add_models(self) -> None: """ Add Model definitions to the Swagger file, if necessary :return: """ if not self.models: return if self.models and not self.definition_body: raise InvalidResourceException( self.logical_id, "Models works only with inline Swagger specified in 'DefinitionBody' property." ) if not SwaggerEditor.is_valid(self.definition_body): raise InvalidResourceException( self.logical_id, "Unable to add Models definitions because " "'DefinitionBody' does not contain a valid Swagger definition.", ) if not all(isinstance(model, dict) for model in self.models.values()): raise InvalidResourceException(self.logical_id, "Invalid value for 'Models' property") swagger_editor = SwaggerEditor(self.definition_body) swagger_editor.add_models(self.models) # type: ignore[no-untyped-call] # Assign the Swagger back to template self.definition_body = self._openapi_postprocess(swagger_editor.swagger) def _openapi_postprocess(self, definition_body: Dict[str, Any]) -> Dict[str, Any]: # noqa: PLR0912 """ Convert definitions to openapi 3 in definition body if OpenApiVersion flag is specified. If the is swagger defined in the definition body, we treat it as a swagger spec and do not make any openapi 3 changes to it """ if definition_body.get("swagger") is not None: return definition_body if self.feature_toggle and self.feature_toggle.is_enabled(FEATURE_FLAG_NORMALIZED_OPENAPI_VERSION): normalized_open_api_version = definition_body.get("openapi", self.open_api_version) elif definition_body.get("openapi") is not None and self.open_api_version is None: normalized_open_api_version = definition_body.get("openapi") else: normalized_open_api_version = self.open_api_version if normalized_open_api_version and SwaggerEditor.safe_compare_regex_with_string( SwaggerEditor._OPENAPI_VERSION_3_REGEX, normalized_open_api_version ): if definition_body.get("securityDefinitions"): components = definition_body.get("components", Py27Dict()) # In the previous line, the default value `Py27Dict()` will be only returned only if `components` # property is not in definition_body dict, but if it exist, and its value is None, so None will be # returned and not the default value. That is why the below line is required. components = components if components else Py27Dict() components["securitySchemes"] = definition_body["securityDefinitions"] definition_body["components"] = components del definition_body["securityDefinitions"] if definition_body.get("definitions"): components = definition_body.get("components", Py27Dict()) # the following line to check if components is None # is copied from the previous if... # In the previous line, the default value `Py27Dict()` will be only returned only if `components` # property is not in definition_body dict, but if it exist, and its value is None, so None will be # returned and not the default value. That is why the below line is required. components = components if components else Py27Dict() components["schemas"] = definition_body["definitions"] definition_body["components"] = components del definition_body["definitions"] # removes `consumes` and `produces` options for CORS in openapi3 and # adds `schema` for the headers in responses for openapi3 paths = definition_body.get("paths") if paths: SwaggerEditor.validate_is_dict( paths, "Value of paths must be a dictionary according to Swagger spec.", ) for path, path_item in paths.items(): SwaggerEditor.validate_path_item_is_dict(path_item, path) if path_item.get("options"): SwaggerEditor.validate_is_dict( path_item.get("options"), f"Value of options method for path {path} must be a " "dictionary according to Swagger spec.", ) options = path_item.get("options").copy() for field, field_val in options.items(): # remove unsupported produces and consumes in options for openapi3 if field in ["produces", "consumes"]: del definition_body["paths"][path]["options"][field] # add schema for the headers in options section for openapi3 if field in ["responses"]: try: response_200_headers = dict_deep_get(field_val, "200.headers") except InvalidValueType as ex: raise InvalidDocumentException( [ InvalidTemplateException( f"Invalid responses in options method for path {path}: {ex!s}.", ) ] ) from ex if not response_200_headers: continue SwaggerEditor.validate_is_dict( response_200_headers, f"Value of response's headers in options method for path {path} must be a " "dictionary according to Swagger spec.", ) for header, header_val in response_200_headers.items(): new_header_val_with_schema = Py27Dict() new_header_val_with_schema["schema"] = header_val definition_body["paths"][path]["options"][field]["200"]["headers"][ header ] = new_header_val_with_schema return definition_body def _get_authorizers(self, authorizers_config, default_authorizer=None): # type: ignore[no-untyped-def] # The dict below will eventually become part of swagger/openapi definition, thus requires using Py27Dict() authorizers = Py27Dict() if default_authorizer == "AWS_IAM": authorizers[default_authorizer] = ApiGatewayAuthorizer( api_logical_id=self.logical_id, name=default_authorizer, is_aws_iam_authorizer=True ) if not authorizers_config: if "AWS_IAM" in authorizers: return authorizers return None sam_expect(authorizers_config, self.logical_id, "Auth.Authorizers").to_be_a_map() for authorizer_name, authorizer in authorizers_config.items(): sam_expect(authorizer, self.logical_id, f"Auth.Authorizers.{authorizer_name}").to_be_a_map() authorizers[authorizer_name] = ApiGatewayAuthorizer( api_logical_id=self.logical_id, name=authorizer_name, user_pool_arn=authorizer.get("UserPoolArn"), function_arn=authorizer.get("FunctionArn"), identity=authorizer.get("Identity"), function_payload_type=authorizer.get("FunctionPayloadType"), function_invoke_role=authorizer.get("FunctionInvokeRole"), authorization_scopes=authorizer.get("AuthorizationScopes"), disable_function_default_permissions=authorizer.get("DisableFunctionDefaultPermissions"), ) return authorizers def _get_permission(self, authorizer_name, authorizer_lambda_function_arn): # type: ignore[no-untyped-def] """Constructs and returns the Lambda Permission resource allowing the Authorizer to invoke the function. :returns: the permission resource :rtype: model.lambda_.LambdaPermission """ rest_api = ApiGatewayRestApi(self.logical_id, depends_on=self.depends_on, attributes=self.resource_attributes) api_id = rest_api.get_runtime_attr("rest_api_id") partition = ArnGenerator.get_partition_name() resource = "${__ApiId__}/authorizers/*" source_arn = fnSub( ArnGenerator.generate_arn(partition=partition, service="execute-api", resource=resource), {"__ApiId__": api_id}, ) lambda_permission = LambdaPermission( self.logical_id + authorizer_name + "AuthorizerPermission", attributes=self.passthrough_resource_attributes ) lambda_permission.Action = "lambda:InvokeFunction" lambda_permission.FunctionName = authorizer_lambda_function_arn lambda_permission.Principal = "apigateway.amazonaws.com" lambda_permission.SourceArn = source_arn return lambda_permission def _construct_authorizer_lambda_permission(self) -> List[LambdaPermission]: if not self.auth: return [] auth_properties = AuthProperties(**self.auth) authorizers = self._get_authorizers(auth_properties.Authorizers) # type: ignore[no-untyped-call] if not authorizers: return [] permissions = [] for authorizer_name, authorizer in authorizers.items(): # Construct permissions for Lambda Authorizers only if not authorizer.function_arn or authorizer.disable_function_default_permissions: continue permission = self._get_permission(authorizer_name, authorizer.function_arn) # type: ignore[no-untyped-call] permissions.append(permission) return permissions def _set_default_authorizer( self, swagger_editor: SwaggerEditor, authorizers: Dict[str, ApiGatewayAuthorizer], default_authorizer: str, add_default_auth_to_preflight: bool = True, ) -> None: if not default_authorizer: return if not isinstance(default_authorizer, str): raise InvalidResourceException( self.logical_id, "DefaultAuthorizer is not a string.", ) if not authorizers.get(default_authorizer) and default_authorizer != "AWS_IAM": raise InvalidResourceException( self.logical_id, "Unable to set DefaultAuthorizer because '" + default_authorizer + "' was not defined in 'Authorizers'.", ) for path in swagger_editor.iter_on_path(): swagger_editor.set_path_default_authorizer( path, default_authorizer, authorizers=authorizers, add_default_auth_to_preflight=add_default_auth_to_preflight, ) def _set_default_apikey_required(self, swagger_editor: SwaggerEditor, required_options_api_key: bool) -> None: for path in swagger_editor.iter_on_path(): swagger_editor.set_path_default_apikey_required(path, required_options_api_key) def _set_endpoint_configuration(self, rest_api: ApiGatewayRestApi, value: Union[str, Dict[str, Any]]) -> None: """ Sets endpoint configuration property of AWS::ApiGateway::RestApi resource :param rest_api: RestApi resource :param string/dict value: Value to be set """ if isinstance(value, dict) and value.get("Type"): rest_api.Parameters = {"endpointConfigurationTypes": value.get("Type")} rest_api.EndpointConfiguration = {"Types": [value.get("Type")]} # SAM API `EndpointConfiguration` uses `VPCEndpointIds` but APIGateway RestApi uses `VpcEndpointIds`. # Deny when both properties are defined at the same time. if "VPCEndpointIds" in value and "VpcEndpointIds" in value: raise InvalidResourceException( rest_api.logical_id, "'VPCEndpointIds' and 'VpcEndpointIds' cannot be used together in EndpointConfiguration.", ) # Accept when either `VPCEndpointIds` or `VpcEndpointIds` is defined by users if "VPCEndpointIds" in value or "VpcEndpointIds" in value: rest_api.EndpointConfiguration["VpcEndpointIds"] = value.get("VPCEndpointIds") or value.get( "VpcEndpointIds" ) else: rest_api.EndpointConfiguration = {"Types": [value]} rest_api.Parameters = {"endpointConfigurationTypes": value} def _generate_domain_access_association( self, domain_access_association: Dict[str, Any], domain_name_arn: Dict[str, str], domain_logical_id: str, ) -> ApiGatewayDomainNameAccessAssociation: """ Generate domain access association resource """ vpcEndpointId = domain_access_association.get("VpcEndpointId") logical_id = LogicalIdGenerator("DomainNameAccessAssociation", [vpcEndpointId, domain_logical_id]).gen() domain_access_association_resource = ApiGatewayDomainNameAccessAssociation( logical_id, attributes=self.passthrough_resource_attributes ) domain_access_association_resource.DomainNameArn = domain_name_arn domain_access_association_resource.AccessAssociationSourceType = "VPCE" domain_access_association_resource.AccessAssociationSource = vpcEndpointId return domain_access_association_resource