如何通过REST API拉取AWS CloudTrail日志？API获取及Python适配咨询

Hey there! Let's tackle your questions about pulling AWS CloudTrail logs using the REST API and implementing this in Python—this is a common use case, so I'll break it down step by step.

AWS CloudTrail offers two primary ways to retrieve logs through its REST API, depending on your needs:

Option 1: Query Recent/Real-Time Events with LookupEvents #

This API lets you search for specific CloudTrail events directly (perfect for ad-hoc queries or checking recent activity). Here's the breakdown:

• Endpoint: https://cloudtrail.{your-region}.amazonaws.com/ (replace {your-region} with your target AWS region like us-east-1)
• HTTP Method: POST
• Authentication: All AWS REST APIs require AWS Signature Version 4—unauthenticated requests won’t work. You’ll need to sign your request using your AWS access key, secret key, and region details.
• Sample Request Body:

  {
    "StartTime": "2024-05-01T00:00:00Z",
    "EndTime": "2024-05-02T00:00:00Z",
    "LookupAttributes": [
      {
        "AttributeKey": "EventName",
        "AttributeValue": "RunInstances"
      }
    ]
  }
  

  This request would return all EC2 instance launch events between the specified dates.

Option 2: Retrieve Full Log Files from S3 #

By default, CloudTrail delivers compressed log files to an S3 bucket you configure. To access these files, use S3's REST API:

1. First, list objects in your CloudTrail bucket using the ListObjectsV2 endpoint: https://{bucket-name}.s3.{region}.amazonaws.com/
2. Then, download individual log files using the GetObject API. Log file keys follow this pattern: AWSLogs/{account-id}/CloudTrail/{region}/YYYY/MM/DD/{account-id}_CloudTrail_{region}_YYYYMMDDTHHmmZ_hash.json.gz

You have two approaches here: using the official AWS SDK (boto3, the easiest route) or directly calling the REST API (more manual, useful for edge cases).

Using Boto3 (Recommended) #

Boto3 handles authentication, request signing, and API complexity for you. Here's how to get started:

Step 1: Install Boto3 #

pip install boto3

Step 2: Configure AWS Credentials #

Set up your credentials via one of these methods:

• Environment variables (AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY)
• ~/.aws/credentials file
• IAM role (if running on AWS infrastructure like EC2/EKS)

Step 3: Example 1: Query Events with LookupEvents #

import boto3
from datetime import datetime, timedelta

# Initialize CloudTrail client
cloudtrail = boto3.client('cloudtrail', region_name='us-east-1')

# Define time range (last 24 hours)
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=24)

# Query EC2-related events
response = cloudtrail.lookup_events(
    StartTime=start_time,
    EndTime=end_time,
    LookupAttributes=[
        {
            'AttributeKey': 'EventSource',
            'AttributeValue': 'ec2.amazonaws.com'
        }
    ]
)

# Print formatted results
for event in response['Events']:
    print(f"Event Name: {event['EventName']}")
    print(f"Event Time: {event['EventTime']}")
    print(f"Initiated By: {event['Username']}\n")

Step 4: Example 2: Download CloudTrail Logs from S3 #

import boto3
import gzip
import json

# Initialize S3 client
s3 = boto3.client('s3', region_name='us-east-1')
bucket_name = 'your-cloudtrail-bucket-name'
# Replace with your account ID and region
prefix = 'AWSLogs/123456789012/CloudTrail/us-east-1/'

# List all log objects in the prefix
response = s3.list_objects_v2(Bucket=bucket_name, Prefix=prefix)

# Download and process each log file
for obj in response.get('Contents', []):
    obj_key = obj['Key']
    # Skip directory entries
    if obj_key.endswith('/'):
        continue
    
    # Download the gzipped log file
    s3_object = s3.get_object(Bucket=bucket_name, Key=obj_key)
    # Decompress and parse the JSON data
    with gzip.GzipFile(fileobj=s3_object['Body']) as decompressed_file:
        log_data = json.load(decompressed_file)
    
    # Print key details from each log record
    for record in log_data['Records']:
        print(f"Action: {record['eventName']} | User: {record['userIdentity'].get('userName', 'N/A')}")

Direct REST API Call (Advanced) #

If you need to call the REST API directly, use botocore to handle AWS Signature V4 signing (manually signing requests is error-prone):

import requests
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.session import Session

region = 'us-east-1'
service = 'cloudtrail'
session = Session()
credentials = session.get_credentials()

# Build the API request
request = AWSRequest(
    method='POST',
    url=f'https://cloudtrail.{region}.amazonaws.com/',
    headers={
        'Content-Type': 'application/x-amz-json-1.1',
        'X-Amz-Target': 'CloudTrail_20131101.LookupEvents'
    },
    body='{"StartTime": "2024-05-01T00:00:00Z", "EndTime": "2024-05-02T00:00:00Z"}'
)

# Sign the request with AWS SigV4
SigV4Auth(credentials, service, region).add_auth(request)

# Send the request and print results
response = requests.post(request.url, headers=request.headers, data=request.body)
print(response.json())

Key Reminders #

• Permissions: Ensure your AWS identity has these IAM permissions: cloudtrail:LookupEvents for event queries, and s3:ListBucket + s3:GetObject for accessing S3 logs.
• Pagination: Both LookupEvents and S3 ListObjectsV2 return paginated results—use the NextToken from responses to fetch all data.

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