in samtranslator/model/eventsources/pull.py [0:0]
def get_policy_statements(self) -> List[Dict[str, Any]]:
basic_auth_uri = self._validate_source_access_configurations(["BASIC_AUTH"], "BASIC_AUTH")
statements = [
{
"Action": [
"secretsmanager:GetSecretValue",
],
"Effect": "Allow",
"Resource": basic_auth_uri,
},
{
"Action": [
"rds:DescribeDBClusterParameters",
],
"Effect": "Allow",
"Resource": {"Fn::Sub": "arn:${AWS::Partition}:rds:${AWS::Region}:${AWS::AccountId}:cluster-pg:*"},
},
{
"Action": [
"rds:DescribeDBSubnetGroups",
],
"Effect": "Allow",
"Resource": {"Fn::Sub": "arn:${AWS::Partition}:rds:${AWS::Region}:${AWS::AccountId}:subgrp:*"},
},
{
"Action": [
"rds:DescribeDBClusters",
],
"Effect": "Allow",
"Resource": self.Cluster,
},
{
"Action": [
"ec2:CreateNetworkInterface",
"ec2:DescribeNetworkInterfaces",
"ec2:DeleteNetworkInterface",
"ec2:DescribeVpcs",
"ec2:DescribeSubnets",
"ec2:DescribeSecurityGroups",
],
"Effect": "Allow",
"Resource": "*",
},
]
if self.SecretsManagerKmsKeyId:
self.validate_secrets_manager_kms_key_id()
kms_policy = self._get_kms_decrypt_policy(self.SecretsManagerKmsKeyId)
statements.append(kms_policy)
document = {
"PolicyName": "SamAutoGeneratedDocumentDBPolicy",
"PolicyDocument": {"Statement": statements},
}
return [document]