def add_resource_policy()

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

• 56 lines of code
• 9 McCabe index (conditional complexity)

    def add_resource_policy(self, resource_policy, path, stage):
        """
        Add resource policy definition to Swagger.

        :param dict resource_policy: Dictionary of resource_policy statements which gets translated
        :return:
        """
        if resource_policy is None:
            return
        SwaggerEditor.validate_is_dict(resource_policy, "Resource Policy is not a valid dictionary.")

        aws_account_whitelist = resource_policy.get("AwsAccountWhitelist")
        aws_account_blacklist = resource_policy.get("AwsAccountBlacklist")
        ip_range_whitelist = resource_policy.get("IpRangeWhitelist")
        ip_range_blacklist = resource_policy.get("IpRangeBlacklist")
        source_vpc_whitelist = resource_policy.get("SourceVpcWhitelist")
        source_vpc_blacklist = resource_policy.get("SourceVpcBlacklist")

        # Intrinsic's supported in these properties
        source_vpc_intrinsic_whitelist = resource_policy.get("IntrinsicVpcWhitelist")
        source_vpce_intrinsic_whitelist = resource_policy.get("IntrinsicVpceWhitelist")
        source_vpc_intrinsic_blacklist = resource_policy.get("IntrinsicVpcBlacklist")
        source_vpce_intrinsic_blacklist = resource_policy.get("IntrinsicVpceBlacklist")

        if aws_account_whitelist is not None:
            resource_list = self._get_method_path_uri_list(path, stage)
            self._add_iam_resource_policy_for_method(aws_account_whitelist, "Allow", resource_list)

        if aws_account_blacklist is not None:
            resource_list = self._get_method_path_uri_list(path, stage)
            self._add_iam_resource_policy_for_method(aws_account_blacklist, "Deny", resource_list)

        if ip_range_whitelist is not None:
            resource_list = self._get_method_path_uri_list(path, stage)
            self._add_ip_resource_policy_for_method(ip_range_whitelist, "NotIpAddress", resource_list)

        if ip_range_blacklist is not None:
            resource_list = self._get_method_path_uri_list(path, stage)
            self._add_ip_resource_policy_for_method(ip_range_blacklist, "IpAddress", resource_list)

        if not SwaggerEditor._validate_list_property_is_resolved(source_vpc_blacklist):
            raise InvalidDocumentException(
                [
                    InvalidTemplateException(
                        "SourceVpcBlacklist must be a list of strings. Use IntrinsicVpcBlacklist instead for values that use Intrinsic Functions"
                    )
                ]
            )

        # FIXME: check if this requires py27 dict?
        blacklist_dict = {
            "StringEndpointList": source_vpc_blacklist,
            "IntrinsicVpcList": source_vpc_intrinsic_blacklist,
            "IntrinsicVpceList": source_vpce_intrinsic_blacklist,
        }
        resource_list = self._get_method_path_uri_list(path, stage)
        self._add_vpc_resource_policy_for_method(blacklist_dict, "StringEquals", resource_list)

        if not SwaggerEditor._validate_list_property_is_resolved(source_vpc_whitelist):
            raise InvalidDocumentException(
                [
                    InvalidTemplateException(
                        "SourceVpcWhitelist must be a list of strings. Use IntrinsicVpcWhitelist instead for values that use Intrinsic Functions"
                    )
                ]
            )

        whitelist_dict = {
            "StringEndpointList": source_vpc_whitelist,
            "IntrinsicVpcList": source_vpc_intrinsic_whitelist,
            "IntrinsicVpceList": source_vpce_intrinsic_whitelist,
        }
        self._add_vpc_resource_policy_for_method(whitelist_dict, "StringNotEquals", resource_list)

        self._doc[self._X_APIGW_POLICY] = self.resource_policy