AWS CloudTrail Forensics for Security Incident Investigation

Introduction

When a security incident occurs in AWS, CloudTrail is the primary forensic evidence source. Every API call, console login, and resource modification is recorded with timestamps, source IPs, user identities, and request parameters. However, effective forensic investigation requires knowing what to look for, how to correlate events across services, and how to preserve evidence before it ages out.

In practice, the median time to detect a cloud breach is 197 days. When organizations finally detect an incident, they need to reconstruct attacker activity across potentially millions of CloudTrail events. This guide provides the investigation methodology and query patterns that reduce investigation time from days to hours.

CloudTrail Event Anatomy

Every CloudTrail event contains critical forensic data:

Field	Forensic Value	Example
eventTime	Timeline reconstruction	2025-12-15T14:32:17Z
eventSource	Service targeted	iam.amazonaws.com
eventName	Action performed	CreateAccessKey
sourceIPAddress	Attacker location	198.51.100.45
userIdentity	Who performed action	Role/User ARN
userAgent	Tool identification	aws-cli/2.x, Boto3
requestParameters	What was changed	AccessKeyId, PolicyArn
responseElements	Result of action	Created key details
errorCode	Failed attempts	AccessDenied

Phase 1: Evidence Preservation

Before beginning investigation, preserve evidence to prevent data loss:

# Create a forensic S3 bucket with legal hold
aws s3api create-bucket \
  --bucket forensic-evidence-$(date +%Y%m%d) \
  --region us-east-1

# Enable object lock (write-once-read-many)
aws s3api put-object-lock-configuration \
  --bucket forensic-evidence-$(date +%Y%m%d) \
  --object-lock-configuration '{
    "ObjectLockEnabled": "Enabled",
    "Rule": {
      "DefaultRetention": {
        "Mode": "COMPLIANCE",
        "Days": 365
      }
    }
  }'

# Copy CloudTrail logs to forensic bucket
aws s3 sync \
  s3://organization-trail-bucket/AWSLogs/ \
  s3://forensic-evidence-$(date +%Y%m%d)/cloudtrail/ \
  --source-region us-east-1

Phase 2: Initial Triage with Athena

Create Athena Table for CloudTrail

CREATE EXTERNAL TABLE cloudtrail_logs (
  eventVersion STRING,
  userIdentity STRUCT<
    type: STRING,
    principalId: STRING,
    arn: STRING,
    accountId: STRING,
    invokedBy: STRING,
    accessKeyId: STRING,
    userName: STRING,
    sessionContext: STRUCT<
      attributes: STRUCT<
        mfaAuthenticated: STRING,
        creationDate: STRING>,
      sessionIssuer: STRUCT<
        type: STRING,
        principalId: STRING,
        arn: STRING,
        accountId: STRING,
        userName: STRING>>>,
  eventTime STRING,
  eventSource STRING,
  eventName STRING,
  awsRegion STRING,
  sourceIPAddress STRING,
  userAgent STRING,
  errorCode STRING,
  errorMessage STRING,
  requestParameters STRING,
  responseElements STRING,
  additionalEventData STRING,
  requestId STRING,
  eventId STRING,
  resources ARRAY<STRUCT<
    arn: STRING,
    accountId: STRING,
    type: STRING>>,
  eventType STRING,
  recipientAccountId STRING
)
PARTITIONED BY (region STRING, year STRING, month STRING, day STRING)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION 's3://forensic-evidence-20251228/cloudtrail/'

Key Investigation Queries

-- Find all actions by a compromised access key
SELECT eventTime, eventSource, eventName, sourceIPAddress,
       awsRegion, errorCode, requestParameters
FROM cloudtrail_logs
WHERE userIdentity.accessKeyId = 'AKIAIOSFODNN7EXAMPLE'
ORDER BY eventTime ASC;

-- Identify unusual source IPs for a specific user
SELECT sourceIPAddress, COUNT(*) as event_count,
       MIN(eventTime) as first_seen, MAX(eventTime) as last_seen
FROM cloudtrail_logs
WHERE userIdentity.arn LIKE '%compromised-role%'
GROUP BY sourceIPAddress
ORDER BY event_count DESC;

-- Find reconnaissance activity (describe/list/get calls)
SELECT eventSource, eventName, COUNT(*) as call_count
FROM cloudtrail_logs
WHERE sourceIPAddress = '198.51.100.45'
  AND eventName LIKE 'Describe%' OR eventName LIKE 'List%' OR eventName LIKE 'Get%'
GROUP BY eventSource, eventName
ORDER BY call_count DESC
LIMIT 50;

Phase 3: Attack Timeline Reconstruction

Common Attack Patterns

Phase	CloudTrail Indicators	Timeframe
Initial Access	ConsoleLogin, GetCallerIdentity	T+0
Reconnaissance	Describe*, List*, Get* calls	T+0 to T+1h
Persistence	CreateAccessKey, CreateRole, CreateUser	T+1h to T+4h
Privilege Escalation	AttachRolePolicy, PutRolePolicy	T+2h to T+6h
Defense Evasion	DeleteTrail, StopLogging, DeleteFlowLogs	T+3h to T+12h
Data Exfiltration	GetObject, CopyObject, SnapshotExport	T+4h to T+24h

Persistence Mechanism Detection

-- Detect backdoor access keys created
SELECT eventTime, userIdentity.arn as creator,
       JSON_EXTRACT_SCALAR(responseElements, '$.accessKey.accessKeyId') as new_key,
       JSON_EXTRACT_SCALAR(requestParameters, '$.userName') as target_user
FROM cloudtrail_logs
WHERE eventName = 'CreateAccessKey'
  AND eventTime BETWEEN '2025-12-01' AND '2025-12-28'
ORDER BY eventTime;

-- Detect role trust policy modifications (backdoor roles)
SELECT eventTime, userIdentity.arn,
       JSON_EXTRACT_SCALAR(requestParameters, '$.roleName') as role_name,
       requestParameters
FROM cloudtrail_logs
WHERE eventName = 'UpdateAssumeRolePolicy'
  AND eventTime BETWEEN '2025-12-01' AND '2025-12-28'
ORDER BY eventTime;

-- Detect defense evasion attempts
SELECT eventTime, eventName, userIdentity.arn, sourceIPAddress
FROM cloudtrail_logs
WHERE eventName IN (
  'DeleteTrail', 'StopLogging', 'UpdateTrail',
  'DeleteFlowLogs', 'DeleteDetector',
  'DisableSecurityHub', 'DeleteLogGroup',
  'PutBucketLifecycle'  -- Used to auto-delete logs
)
ORDER BY eventTime;

Phase 4: Scope Assessment

Determine the blast radius of the incident:

-- All accounts accessed by the attacker IP
SELECT DISTINCT recipientAccountId, COUNT(*) as events
FROM cloudtrail_logs
WHERE sourceIPAddress = '198.51.100.45'
GROUP BY recipientAccountId;

-- All regions with attacker activity
SELECT awsRegion, COUNT(*) as events,
       MIN(eventTime) as first_activity,
       MAX(eventTime) as last_activity
FROM cloudtrail_logs
WHERE sourceIPAddress = '198.51.100.45'
GROUP BY awsRegion
ORDER BY events DESC;

-- Resources modified by attacker
SELECT resources, eventName, eventTime
FROM cloudtrail_logs
WHERE sourceIPAddress = '198.51.100.45'
  AND errorCode IS NULL
  AND eventName NOT LIKE 'Describe%'
  AND eventName NOT LIKE 'List%'
  AND eventName NOT LIKE 'Get%'
ORDER BY eventTime;

Automated Investigation with Python

import boto3
import json
from datetime import datetime, timedelta

class CloudTrailForensics:
    def __init__(self, region='us-east-1'):
        self.client = boto3.client('cloudtrail', region_name=region)
        self.athena = boto3.client('athena', region_name=region)

    def lookup_by_access_key(self, access_key_id, days_back=90):
        """Find all events for a specific access key."""
        events = []
        paginator = self.client.get_paginator('lookup_events')

        for page in paginator.paginate(
            LookupAttributes=[{
                'AttributeKey': 'AccessKeyId',
                'AttributeValue': access_key_id
            }],
            StartTime=datetime.utcnow() - timedelta(days=days_back),
            EndTime=datetime.utcnow()
        ):
            for event in page['Events']:
                events.append({
                    'time': event['EventTime'].isoformat(),
                    'name': event['EventName'],
                    'source': event['EventSource'],
                    'user': event.get('Username', 'N/A'),
                    'resources': event.get('Resources', [])
                })

        return sorted(events, key=lambda x: x['time'])

    def detect_persistence(self, start_time, end_time):
        """Detect persistence mechanisms created in timeframe."""
        persistence_events = [
            'CreateAccessKey', 'CreateUser', 'CreateRole',
            'CreateLoginProfile', 'UpdateAssumeRolePolicy',
            'AttachUserPolicy', 'PutUserPolicy',
            'CreateFunction'  # Lambda as backdoor
        ]

        results = []
        paginator = self.client.get_paginator('lookup_events')

        for event_name in persistence_events:
            for page in paginator.paginate(
                LookupAttributes=[{
                    'AttributeKey': 'EventName',
                    'AttributeValue': event_name
                }],
                StartTime=start_time,
                EndTime=end_time
            ):
                results.extend(page['Events'])

        return results

Key Takeaways

• Preserve CloudTrail evidence immediately by copying logs to a bucket with compliance-mode object lock before beginning investigation.
• Use Athena for efficient querying over large volumes of CloudTrail data as the LookupEvents API only supports single-attribute lookups and 90-day retention.
• Follow the attack timeline pattern (Initial Access, Reconnaissance, Persistence, Escalation, Evasion, Exfiltration) to systematically identify all attacker actions.
• Focus on persistence mechanisms first since backdoor access keys, modified trust policies, and new IAM users indicate ongoing compromise.
• Correlate across multiple indicators including source IP, user agent, access key ID, and time windows to build a complete picture.
• Set CloudTrail retention to at least 365 days given the 197-day median detection time, as default 90-day retention is insufficient for most investigations.
• Automate initial triage queries to reduce investigation start time from hours to minutes when incidents are detected.