如何在AWS Lambda中判断CloudTrail事件ARN账号是否在白名单内

Got it, let's walk through building this Lambda function step by step. I'll break down each part so you understand exactly what's happening, and provide a full working code example tailored to your use case.

Step 1: Set Up Imports & Core Configuration #

First, we'll bring in the necessary AWS SDK tools and define your account whitelist plus the target SNS topic.

import boto3
import json

# Your predefined account whitelist (adjust as needed)
account_whitelist = ["999900000000", "1234567891011"]
# Replace this with your actual SNS topic ARN
SNS_TOPIC_ARN = "arn:aws:sns:us-east-1:YOUR_ACCOUNT_ID:your-security-alert-topic"

# Initialize the SNS client once (best practice for Lambda)
sns_client = boto3.client('sns')

Step 2: Parse CloudTrail Events & Extract Account IDs #

CloudTrail events have a nested structure—we'll dig into the event detail to pull out the policyDocument, then extract account IDs from the ARNs in the AWS field.

def lambda_handler(event, context):
    try:
        # Pull the core event details from the CloudTrail payload
        cloudtrail_detail = event['detail']
        
        # Get the policy document (handle both dict and string formats—CloudTrail sometimes sends it as JSON string)
        policy_doc = cloudtrail_detail.get('requestParameters', {}).get('policyDocument', {})
        if isinstance(policy_doc, str):
            policy_doc = json.loads(policy_doc)
        
        # Collect all unique account IDs from the policy's AWS principals
        detected_accounts = set()
        statements = policy_doc.get('Statement', [])
        
        for stmt in statements:
            # Only check for sts:AssumeRole actions as per your requirement
            if stmt.get('Action') == 'sts:AssumeRole':
                aws_principals = stmt.get('AWS', [])
                for arn in aws_principals:
                    # Extract account ID from ARN (format: arn:aws:iam::ACCOUNT_ID:root)
                    arn_parts = arn.split(':')
                    if len(arn_parts) >= 5:  # Guard against malformed ARNs
                        account_id = arn_parts[4]
                        detected_accounts.add(account_id)

Step 3: Compare to Whitelist & Send SNS Alert #

Now we'll check which detected accounts aren't in your whitelist, and trigger an SNS notification if any are found.

# Filter out accounts that are in the whitelist
        non_whitelisted = [acc for acc in detected_accounts if acc not in account_whitelist]
        
        if non_whitelisted:
            # Compose a clear, actionable alert message
            alert_msg = f"🚨 ALERT: Non-whitelisted accounts detected using sts:AssumeRole!\n"
            alert_msg += f"Non-whitelisted Account IDs: {', '.join(non_whitelisted)}\n"
            alert_msg += f"CloudTrail Event ID: {cloudtrail_detail.get('eventID', 'Unknown')}\n"
            alert_msg += f"Event Timestamp: {cloudtrail_detail.get('eventTime', 'Unknown')}\n"
            alert_msg += f"Source IP: {cloudtrail_detail.get('sourceIPAddress', 'Unknown')}"
            
            # Send the notification to your SNS topic
            sns_client.publish(
                TopicArn=SNS_TOPIC_ARN,
                Subject="IAM Security Alert: Non-Whitelisted Account Access",
                Message=alert_msg
            )
            
            print(f"Alert sent for non-whitelisted accounts: {non_whitelisted}")
            return {
                'statusCode': 200,
                'body': json.dumps(f"Alert triggered for non-whitelisted accounts: {non_whitelisted}")
            }
        else:
            print("All detected accounts are in the whitelist—no action needed.")
            return {
                'statusCode': 200,
                'body': json.dumps("All accounts are whitelisted.")
            }
            
    except Exception as e:
        print(f"Error processing event: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps(f"Error processing event: {str(e)}")
        }

Key Edge Cases & Best Practices #

• Policy Document Format: CloudTrail sometimes sends policyDocument as a JSON string instead of a dictionary—our code handles both cases.
• Malformed ARNs: We added a check to ensure ARNs have enough parts before extracting the account ID, preventing index errors.
• Lambda Permissions: Make sure your Lambda execution role has the sns:Publish permission for your target SNS topic, plus permissions to receive events from CloudWatch Events/EventBridge (which triggers Lambda from CloudTrail).
• Duplicate Accounts: Using a set for detected accounts ensures we don't process the same account ID multiple times.

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