def _get_bucket_policy_for_cloudwatch_logs()

in cli/src/pcluster/models/s3_bucket.py [0:0]

• 52 lines of code
• 1 McCabe index (conditional complexity)

    def _get_bucket_policy_for_cloudwatch_logs(self):
        """Generate the required bucket policy statements for CloudWatch Logs export."""
        partition = self.partition
        region = self.region
        account_id = self.account_id
        bucket_arn = format_arn(partition, "s3", "", "", self.name)
        bucket_arn_with_wildcard = f"{bucket_arn}/*"
        logs_service_principal = get_service_principal(
            service_name="logs", partition=partition, region=region, regional=True
        )

        # Define allowed log group ARNs
        log_group_arns = [
            format_arn(partition, "logs", region, account_id, "log-group:/aws/parallelcluster/*"),
            format_arn(partition, "logs", region, account_id, "log-group:/aws/imagebuilder/*"),
        ]

        policy_statements = [
            {
                "Sid": "AllowReadBucketAclForExportLogs",
                "Action": "s3:GetBucketAcl",
                "Effect": "Allow",
                "Resource": bucket_arn,
                "Principal": {"Service": logs_service_principal},
                "Condition": {
                    "StringEquals": {"aws:SourceAccount": account_id},
                    "ArnLike": {
                        "aws:SourceArn": log_group_arns,
                    },
                },
            },
            {
                "Sid": "AllowPutObjectForExportLogs",
                "Action": "s3:PutObject",
                "Effect": "Allow",
                "Resource": bucket_arn_with_wildcard,
                "Principal": {"Service": logs_service_principal},
                "Condition": {
                    "StringEquals": {
                        "s3:x-amz-acl": "bucket-owner-full-control",
                        "aws:SourceAccount": account_id,
                    },
                    "ArnLike": {
                        "aws:SourceArn": log_group_arns,
                    },
                },
            },
            {
                "Sid": "DenyPutObjectOnReservedPath",
                "Action": "s3:PutObject",
                "Effect": "Deny",
                "Resource": f"{bucket_arn}/parallelcluster/*",
                "Principal": {"Service": logs_service_principal},
            },
        ]
        return policy_statements