如何在Lambda函数中读取S3中存储的CloudTrail日志文件

• IAM执行角色权限：Lambda的执行角色需要添加对应S3桶的读权限，若桶开启KMS加密还需添加KMS解密权限，参考策略如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::替换为你的CloudTrail存储桶名/*"
        },
        {
            "Effect": "Allow",
            "Action": "kms:Decrypt",
            "Resource": "arn:aws:kms:替换为区域:替换为账号ID:key/替换为你的KMS密钥ID",
            "Condition": {
                "StringEquals": {
                    "kms:EncryptionContext:aws:s3:arn": "arn:aws:s3:::替换为你的CloudTrail存储桶名"
                }
            }
        }
    ]
}

如果存储桶使用SSE-S3默认加密，不需要添加KMS相关权限。

• Lambda触发配置（推荐）：为Lambda添加S3触发规则，触发事件选择所有对象创建事件，前缀可配置为CloudTrail日志的存储前缀，避免触发不必要的执行
• 运行环境配置：Lambda Python运行环境默认已预装boto3、gzip、json依赖，无需额外上传自定义层，可根据日志文件大小调整Lambda内存和超时时间，避免执行报错

import boto3
import gzip
import json

# 初始化s3客户端
s3_client = boto3.client('s3')

def lambda_handler(event, context):
    # 从S3触发事件中解析桶名和文件Key，若手动读取历史日志可直接在此处写死bucket和key的值
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    # 过滤非CloudTrail日志文件
    if not key.endswith('.json.gz'):
        return {
            'statusCode': 200,
            'body': json.dumps('非CloudTrail日志文件，跳过处理')
        }
    
    # 读取S3文件内容
    s3_resp = s3_client.get_object(Bucket=bucket, Key=key)
    # 解压gzip格式日志
    with gzip.GzipFile(fileobj=s3_resp['Body']) as f:
        log_data = json.loads(f.read().decode('utf-8'))
    
    # 自定义日志处理逻辑，此处示例打印核心操作信息
    for record in log_data['Records']:
        print(f"操作事件名: {record['eventName']}, 操作源IP: {record['sourceIPAddress']}, 操作账号ID: {record['userIdentity']['accountId']}")
    
    return {
        'statusCode': 200,
        'body': json.dumps('日志解析完成')
    }

• CloudTrail生成的日志默认是.json.gz压缩格式，必须先解压再解析内容
• 若不需要自动触发，也可手动指定桶名和文件Key批量读取历史日志，只需替换代码中bucket和key的赋值逻辑即可
• 日志解析后可根据业务需求过滤指定事件（如控制台登录、资源删除等），进行告警或审计存储

内容的提问来源于stack exchange，提问作者SWASTIK