implementing-cloud-trail-log-analysis

实现 CloudTrail 日志分析

何时使用

• 在为 AWS API 活动构建安全监控管道时
• 在调查安全事件以追踪攻击者跨 AWS 服务的操作时
• 当合规要求对所有管理和数据访问操作进行审计日志记录时
• 在为 AWS 环境中的已知攻击模式创建检测规则时
• 在建立 API 行为基线以进行异常检测时

不要使用于实时威胁检测（使用 GuardDuty，它已经分析 CloudTrail）、应用级日志记录（使用 CloudWatch Application Logs）或网络流量分析（使用 VPC Flow Logs）。

先决条件

• 在所有账户中启用 CloudTrail 并包含管理事件以及可选的数据事件
• 将 S3 存储桶配置为 CloudTrail 交付通道并设置适当的保留策略
• 配置 Amazon Athena 并使用 CloudTrail 日志表进行即席查询
• 配置 CloudWatch Logs 订阅并使用 Logs Insights 进行实时分析
• 用于生产监控的 SIEM 集成（Splunk、Elastic 或 Security Lake）

工作流

步骤 1：配置 CloudTrail 以实现全面日志记录

确保 CloudTrail 捕获组织内所有相关事件类型。

# Create an organization trail (captures all accounts)
aws cloudtrail create-trail \
  --name org-security-trail \
  --s3-bucket-name cloudtrail-logs-org-ACCOUNT \
  --is-organization-trail \
  --is-multi-region-trail \
  --include-global-service-events \
  --enable-log-file-validation \
  --kms-key-id alias/cloudtrail-key \
  --cloud-watch-logs-log-group-arn arn:aws:logs:us-east-1:ACCOUNT:log-group:cloudtrail-org:* \
  --cloud-watch-logs-role-arn arn:aws:iam::ACCOUNT:role/CloudTrailCloudWatchRole

# Start logging
aws cloudtrail start-logging --name org-security-trail

# Enable data events for S3 and Lambda
aws cloudtrail put-event-selectors \
  --trail-name org-security-trail \
  --advanced-event-selectors '[
    {
      "Name": "S3DataEvents",
      "FieldSelectors": [
        {"Field": "eventCategory", "Equals": ["Data"]},
        {"Field": "resources.type", "Equals": ["AWS::S3::Object"]}
      ]
    },
    {
      "Name": "LambdaDataEvents",
      "FieldSelectors": [
        {"Field": "eventCategory", "Equals": ["Data"]},
        {"Field": "resources.type", "Equals": ["AWS::Lambda::Function"]}
      ]
    }
  ]'

# Verify trail configuration
aws cloudtrail describe-trails --trail-name-list org-security-trail

步骤 2：设置 Athena 进行 CloudTrail 查询分析

创建 Athena 表以使用 SQL 查询 CloudTrail 日志。

-- Create CloudTrail Athena table
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,
  readOnly STRING,
  resources ARRAY<STRUCT<arn:STRING, accountId:STRING, type:STRING>>,
  eventType STRING,
  apiVersion STRING,
  recipientAccountId STRING,
  sharedEventId STRING,
  vpcEndpointId STRING
)
PARTITIONED BY (region STRING, year STRING, month STRING, day STRING)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION 's3://cloudtrail-logs-org-ACCOUNT/AWSLogs/ORG_ID/';

-- Add partitions for recent data
ALTER TABLE cloudtrail_logs ADD
  PARTITION (region='us-east-1', year='2026', month='02', day='23')
  LOCATION 's3://cloudtrail-logs-org-ACCOUNT/AWSLogs/ORG_ID/ACCOUNT/CloudTrail/us-east-1/2026/02/23/';

步骤 3：运行以安全为重点的 Athena 查询

执行查询以检测常见攻击模式和可疑活动。

-- Detect console logins without MFA
SELECT eventtime, useridentity.username, sourceipaddress, useridentity.arn
FROM cloudtrail_logs
WHERE eventname = 'ConsoleLogin'
  AND additionalEventData LIKE '%"MFAUsed":"No"%'
  AND errorcode IS NULL
ORDER BY eventtime DESC;

-- Find IAM privilege escalation attempts
SELECT eventtime, useridentity.arn, eventname, errorcode, sourceipaddress
FROM cloudtrail_logs
WHERE eventname IN (
  'CreatePolicyVersion', 'SetDefaultPolicyVersion', 'AttachUserPolicy',
  'AttachRolePolicy', 'PutUserPolicy', 'PutRolePolicy',
  'CreateAccessKey', 'CreateLoginProfile', 'UpdateLoginProfile',
  'PassRole', 'AssumeRole'
)
ORDER BY eventtime DESC
LIMIT 100;

-- Detect CloudTrail tampering
SELECT eventtime, useridentity.arn, eventname, requestparameters, sourceipaddress
FROM cloudtrail_logs
WHERE eventname IN ('StopLogging', 'DeleteTrail', 'UpdateTrail', 'PutEventSelectors')
ORDER BY eventtime DESC;

-- Find API calls from Tor exit nodes or unusual IPs
SELECT eventtime, useridentity.arn, eventname, sourceipaddress, awsregion
FROM cloudtrail_logs
WHERE sourceipaddress NOT LIKE '10.%'
  AND sourceipaddress NOT LIKE '172.%'
  AND sourceipaddress NOT LIKE '192.168.%'
  AND useridentity.type = 'IAMUser'
  AND errorcode IS NULL
GROUP BY eventtime, useridentity.arn, eventname, sourceipaddress, awsregion
ORDER BY eventtime DESC
LIMIT 200;

-- Detect unauthorized API calls (AccessDenied patterns)
SELECT useridentity.arn, eventname, COUNT(*) as denied_count
FROM cloudtrail_logs
WHERE errorcode IN ('AccessDenied', 'UnauthorizedAccess', 'Client.UnauthorizedAccess')
  AND eventtime > date_format(date_add('day', -7, now()), '%Y-%m-%dT%H:%i:%sZ')
GROUP BY useridentity.arn, eventname
HAVING COUNT(*) > 10
ORDER BY denied_count DESC;

步骤 4：使用 CloudWatch Logs Insights 构建实时检测

创建实时查询以进行主动安全监控。

# Detect root account usage
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, eventName, sourceIPAddress, userAgent
    | filter userIdentity.type = "Root"
    | sort @timestamp desc
  '

# Detect security group changes
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, userIdentity.arn, eventName, requestParameters.groupId, sourceIPAddress
    | filter eventName in ["AuthorizeSecurityGroupIngress", "AuthorizeSecurityGroupEgress", "RevokeSecurityGroupIngress", "CreateSecurityGroup"]
    | sort @timestamp desc
  '

# Detect new IAM users or access keys created
aws logs start-query \
  --log-group-name cloudtrail-org \
  --start-time $(date -d "24 hours ago" +%s) \
  --end-time $(date +%s) \
  --query-string '
    fields @timestamp, userIdentity.arn, eventName, requestParameters.userName, sourceIPAddress
    | filter eventName in ["CreateUser", "CreateAccessKey", "CreateLoginProfile"]
    | sort @timestamp desc
  '

步骤 5：创建 CloudWatch 指标筛选器和警报

基于 CIS Benchmark 建议为关键安全事件设置自动警报。

# CIS 3.1: Unauthorized API calls alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name unauthorized-api-calls \
  --filter-pattern '{($.errorCode = "*UnauthorizedAccess") || ($.errorCode = "AccessDenied*")}' \
  --metric-transformations '[{"metricName":"UnauthorizedAPICalls","metricNamespace":"CISBenchmark","metricValue":"1"}]'

aws cloudwatch put-metric-alarm \
  --alarm-name cis-unauthorized-api-calls \
  --metric-name UnauthorizedAPICalls --namespace CISBenchmark \
  --statistic Sum --period 300 --threshold 10 \
  --comparison-operator GreaterThanThreshold --evaluation-periods 1 \
  --alarm-actions arn:aws:sns:us-east-1:ACCOUNT:security-alerts

# CIS 3.3: Root account usage alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name root-account-usage \
  --filter-pattern '{$.userIdentity.type = "Root" && $.userIdentity.invokedBy NOT EXISTS && $.eventType != "AwsServiceEvent"}' \
  --metric-transformations '[{"metricName":"RootAccountUsage","metricNamespace":"CISBenchmark","metricValue":"1"}]'

# CIS 3.4: IAM policy changes alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name iam-policy-changes \
  --filter-pattern '{($.eventName=CreatePolicy) || ($.eventName=DeletePolicy) || ($.eventName=AttachRolePolicy) || ($.eventName=DetachRolePolicy) || ($.eventName=AttachUserPolicy) || ($.eventName=DetachUserPolicy)}' \
  --metric-transformations '[{"metricName":"IAMPolicyChanges","metricNamespace":"CISBenchmark","metricValue":"1"}]'

# CIS 3.5: CloudTrail configuration changes alarm
aws logs put-metric-filter \
  --log-group-name cloudtrail-org \
  --filter-name cloudtrail-changes \
  --filter-pattern '{($.eventName = StopLogging) || ($.eventName = DeleteTrail) || ($.eventName = UpdateTrail)}' \
  --metric-transformations '[{"metricName":"CloudTrailChanges","metricNamespace":"CISBenchmark","metricValue":"1"}]'

关键概念

术语	定义
CloudTrail	记录对 AWS 服务所做 API 调用的 AWS 服务，提供用户、角色和服务所采取操作的审计追踪
管理事件	用于控制平面操作的 CloudTrail 事件，例如创建资源、修改 IAM 和配置服务
数据事件	用于数据平面操作的 CloudTrail 事件，例如 S3 对象访问和 Lambda 函数调用，提供细粒度活动日志记录
日志文件验证	CloudTrail 功能，创建摘要文件以验证日志文件在交付后未被篡改
CloudTrail Lake	用于 CloudTrail 事件的托管数据湖，支持基于 SQL 的查询，而无需管理 Athena 表或 S3 数据
组织跟踪	单一跟踪，将 AWS Organization 中所有账户的 API 活动捕获到中央 S3 存储桶

工具与系统

• Amazon Athena：无服务器 SQL 查询引擎，用于大规模分析存储在 S3 中的 CloudTrail 日志
• CloudWatch Logs Insights：实时日志查询服务，用于在过去 30 天内进行交互式 CloudTrail 分析
• CloudTrail Lake：托管事件数据湖，内置 SQL 查询功能和 7 年保留期
• Amazon Security Lake：集中式安全数据湖，将 CloudTrail 数据规范化为 OCSF 格式以供 SIEM 使用
• AWS CloudTrail：核心审计日志记录服务，捕获跨 AWS 账户和服务的所有 API 活动

常见场景

场景：通过 CloudTrail 调查 IAM 凭证泄露

背景：GuardDuty 针对某开发人员的访问密钥触发了 UnauthorizedAccess:IAMUser/MaliciousIPCaller 告警。安全团队需要追踪该泄露凭证所执行的所有操作。

方法：

1. 在 CloudTrail 中查询所有区域中被泄露的 AccessKeyId 的所有事件
2. 构建 API 调用时间线，以了解攻击序列
3. 确定初始访问点（该密钥首次从恶意 IP 出现的时间）
4. 梳理攻击者创建、修改或访问的所有资源
5. 检查持久化机制（新用户、访问密钥、Lambda 函数、EC2 实例）
6. 验证 CloudTrail 未被篡改（检查是否有 StopLogging 或 UpdateTrail 事件）
7. 为事件响应报告记录完整的攻击链和影响范围

注意事项：CloudTrail 事件最多需要 15 分钟才会出现在 S3 和 CloudWatch Logs 中。在活跃事件期间需要实时可见性时，请使用 CloudTrail Lake 或 CloudWatch Logs Insights，而不是对 S3 执行 Athena 查询。跨区域攻击需要在 Athena 中查询多个区域分区。

输出格式

CloudTrail Security Analysis Report
======================================
Account: 123456789012
Analysis Period: 2026-02-16 to 2026-02-23
Trail: org-security-trail (organization-wide)

SECURITY EVENTS DETECTED:
  Root account logins:                  2
  Console logins without MFA:           7
  Privilege escalation attempts:       12
  CloudTrail configuration changes:     0
  Security group modifications:        34
  Unauthorized API calls:             156

HIGH-PRIORITY FINDINGS:
[CT-001] Console Login Without MFA
  User: admin-user
  Time: 2026-02-22T14:30:00Z
  IP: 203.0.113.50
  Action Required: Enforce MFA via IAM policy

[CT-002] IAM Privilege Escalation
  User: dev-user
  Time: 2026-02-23T03:15:00Z
  Events: CreatePolicyVersion -> AttachRolePolicy
  IP: 185.x.x.x (suspicious)
  Action Required: Investigate credential compromise

ALERTING STATUS:
  CIS metric filters configured: 14 / 14
  CloudWatch alarms active: 14 / 14
  Alerts fired (last 7 days): 8