Python lambda 函数检查我的 S3 存储桶是否公开并将它们设为私有(private)

Question

我编写了一个 Python lambda 函数来检查我的 S3 存储桶是否在我的帐户中是公开的并将它们设为私有(private)。但我不断收到错误消息:

({ "errorMessage": "'S3' object has no attribute 'put_bucket_access_block'", "errorType": "AttributeError", "stackTrace": [ " File \"/var/task/lambda_function.py\", line 66, in lambda_handler\n validate_instance(event)\n", " File \"/var/task/lambda_function.py\", line 32, in validate_instance\n addacl = s3_client.put_bucket_access_block(Bucket = s3_bucketName,\n", " File \"/var/runtime/botocore/client.py\", line 563, in getattr\n self.class.name, item)\n" ] })

下面是用 Python Boto3 编写的 Lambda 函数:

import boto3
from botocore.exceptions import ClientError
import json
import logging
import sys
log = logging.getLogger()
log.setLevel(logging.DEBUG)

log = logging.getLogger()
log.setLevel(logging.DEBUG)
print('Loading function')
sts_client = boto3.client('sts')
def validate_instance(rec_event):
    sns_msg = json.loads(rec_event['Records'][0]['Sns']['Message'])
    account_id = sns_msg['account']
    event_region = sns_msg['region']
    assumedRoleObject = sts_client.assume_role(
        RoleArn="arn:aws:iam::{}:role/{}".format(account_id, 'VSC-Admin-Account-Lambda-Execution-Role'),
        RoleSessionName="AssumeRoleSession1"
    )
    credentials = assumedRoleObject['Credentials']
    print(credentials)
    s3_client = boto3.client('s3', event_region, aws_access_key_id=credentials['AccessKeyId'],
                              aws_secret_access_key=credentials['SecretAccessKey'],
                              aws_session_token=credentials['SessionToken'],
                              )
    s3_bucketName = sns_msg['detail']['requestParameters']['bucketName']

    public_block = s3_client.put_public_access_block(Bucket = s3_bucketName,
            PublicAccessBlockConfiguration={
                'BlockPublicAcls': true,
                'IgnorePublicAcls':false,
                'BlockPublicPolicy':true,
                'RestrictPublicBuckets':true
                })        
    enableacl = s3_client.put_bucket_acl(Bucket = s3_bucketName,
                 ACL='private'
                 )


    put_public_access_block
    try:
        checkencryption=s3_client.get_bucket_encryption(Bucket=s3_bucketName)
        print("checking the encrytption")
        rules = checkencryption['ServerSideEncryptionConfiguration']['Rules']
        print('Bucket: %s, Encryption: %s' % (s3_bucketName, rules))
    except ClientError as e:
        if e.response['Error']['Code'] == 'ServerSideEncryptionConfigurationNotFoundError':
            response = s3_client.put_bucket_encryption(Bucket = s3_bucketName,
            ServerSideEncryptionConfiguration={
                'Rules': [
                    {
                        'ApplyServerSideEncryptionByDefault':{
                            'SSEAlgorithm': 'AES256'
                        }
                    },]
            })

        else:
            print("Bucket: %s, unexpected error: %s" % (s3_bucketName, e))


def lambda_handler(event, context):
    log.info("Here is the Received Event")
    log.info(json.dumps(event))
    validate_instance(event)

Answer 1

在撰写本文时，问题是与 Lambda Python 运行时环境捆绑在一起的 boto3 版本还不包含此功能，因此 put_public_access_block 不是 S3 客户端上的方法。

一种解决方法是部署您自己的 Lambda 层，其中包含更高版本的 boto3(确实支持 put_public_access_block)。参见 related answer有关在 Lambda 层中包含 boto3 的选项。