def _add_vpc_resource_policy_for_method()

in samtranslator/swagger/swagger.py [0:0]

• 50 lines of code
• 18 McCabe index (conditional complexity)

    def _add_vpc_resource_policy_for_method(self, endpoint_dict, conditional, resource_list):
        """
        This method generates a policy statement to grant/deny specific VPC/VPCE access to the API method and
        appends it to the swagger under `x-amazon-apigateway-policy`
        :raises ValueError: If the conditional passed in does not match the allowed values.
        """

        if conditional not in ["StringNotEquals", "StringEquals"]:
            raise ValueError("Conditional must be one of {}".format(["StringNotEquals", "StringEquals"]))

        condition = Py27Dict()
        string_endpoint_list = endpoint_dict.get("StringEndpointList")
        intrinsic_vpc_endpoint_list = endpoint_dict.get("IntrinsicVpcList")
        intrinsic_vpce_endpoint_list = endpoint_dict.get("IntrinsicVpceList")

        if string_endpoint_list is not None:
            vpce_regex = r"^vpce-"
            vpc_regex = r"^vpc-"
            vpc_list = []
            vpce_list = []
            for endpoint in string_endpoint_list:
                if re.match(vpce_regex, endpoint):
                    vpce_list.append(endpoint)
                if re.match(vpc_regex, endpoint):
                    vpc_list.append(endpoint)
            if vpc_list:
                condition.setdefault("aws:SourceVpc", []).extend(vpc_list)
            if vpce_list:
                condition.setdefault("aws:SourceVpce", []).extend(vpce_list)
        if intrinsic_vpc_endpoint_list is not None:
            condition.setdefault("aws:SourceVpc", []).extend(intrinsic_vpc_endpoint_list)
        if intrinsic_vpce_endpoint_list is not None:
            condition.setdefault("aws:SourceVpce", []).extend(intrinsic_vpce_endpoint_list)

        # Skip writing to transformed template if both vpc and vpce endpoint lists are empty
        if (not condition.get("aws:SourceVpc", [])) and (not condition.get("aws:SourceVpce", [])):
            return

        self.resource_policy["Version"] = "2012-10-17"
        allow_statement = Py27Dict()
        allow_statement["Effect"] = "Allow"
        allow_statement["Action"] = "execute-api:Invoke"
        allow_statement["Resource"] = resource_list
        allow_statement["Principal"] = "*"

        deny_statement = Py27Dict()
        deny_statement["Effect"] = "Deny"
        deny_statement["Action"] = "execute-api:Invoke"
        deny_statement["Resource"] = resource_list
        deny_statement["Principal"] = "*"
        deny_statement["Condition"] = {conditional: condition}

        if self.resource_policy.get("Statement") is None:
            self.resource_policy["Statement"] = [allow_statement, deny_statement]
        else:
            statement = self.resource_policy["Statement"]
            if not isinstance(statement, list):
                statement = [statement]
            if allow_statement not in statement:
                statement.extend([allow_statement])
            if deny_statement not in statement:
                statement.extend([deny_statement])
            self.resource_policy["Statement"] = statement