def get_policy_statements()

in samtranslator/model/eventsources/pull.py [0:0]
68 lines of code
10 McCabe index (conditional complexity)

    def get_policy_statements(self):
        if not self.SourceAccessConfigurations:
            raise InvalidEventException(
                self.relative_id,
                "No SourceAccessConfigurations for Amazon MQ event provided.",
            )
        if not type(self.SourceAccessConfigurations) is list:
            raise InvalidEventException(
                self.relative_id,
                "Provided SourceAccessConfigurations cannot be parsed into a list.",
            )
        basic_auth_uri = None
        for conf in self.SourceAccessConfigurations:
            event_type = conf.get("Type")
            if event_type not in ("BASIC_AUTH", "VIRTUAL_HOST"):
                raise InvalidEventException(
                    self.relative_id,
                    "Invalid property specified in SourceAccessConfigurations for Amazon MQ event.",
                )
            if event_type == "BASIC_AUTH":
                if basic_auth_uri:
                    raise InvalidEventException(
                        self.relative_id,
                        "Multiple BASIC_AUTH properties specified in SourceAccessConfigurations for Amazon MQ event.",
                    )
                basic_auth_uri = conf.get("URI")
                if not basic_auth_uri:
                    raise InvalidEventException(
                        self.relative_id,
                        "No BASIC_AUTH URI property specified in SourceAccessConfigurations for Amazon MQ event.",
                    )

        if not basic_auth_uri:
            raise InvalidEventException(
                self.relative_id,
                "No BASIC_AUTH property specified in SourceAccessConfigurations for Amazon MQ event.",
            )
        document = {
            "PolicyName": "SamAutoGeneratedAMQPolicy",
            "PolicyDocument": {
                "Statement": [
                    {
                        "Action": [
                            "secretsmanager:GetSecretValue",
                        ],
                        "Effect": "Allow",
                        "Resource": basic_auth_uri,
                    },
                    {
                        "Action": [
                            "mq:DescribeBroker",
                        ],
                        "Effect": "Allow",
                        "Resource": self.Broker,
                    },
                ]
            },
        }
        if self.SecretsManagerKmsKeyId:
            kms_policy = {
                "Action": "kms:Decrypt",
                "Effect": "Allow",
                "Resource": {
                    "Fn::Sub": "arn:${AWS::Partition}:kms:${AWS::Region}:${AWS::AccountId}:key/"
                    + self.SecretsManagerKmsKeyId
                },
            }
            document["PolicyDocument"]["Statement"].append(kms_policy)
        return [document]