---

name: implementing-cloud-trail-log-analysis description: > 实施 AWS CloudTrail 日志分析，利用 Athena、CloudWatch Logs Insights 和 SIEM 集成进行安全监控、威胁检测和取证调查，识别未授权访问、权限提升和可疑 API 活动。 domain: cybersecurity subdomain: cloud-security tags: [cloud-security, aws, cloudtrail, log-analysis, threat-detection, forensics] version: "1.0" author: mahipal license: Apache-2.0

实施 CloudTrail 日志分析

适用场景

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

不适用于：实时威胁检测（使用已分析 CloudTrail 的 GuardDuty）、应用级日志记录（使用 CloudWatch 应用程序日志），或网络流量分析（使用 VPC 流日志）。

前置条件

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

工作流程

步骤 1：配置 CloudTrail 进行全面日志记录

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

# 创建组织跟踪（捕获所有账户）
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

# 开始记录
aws cloudtrail start-logging --name org-security-trail

# 为 S3 和 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"]}
      ]
    }
  ]'

# 验证跟踪配置
aws cloudtrail describe-trails --trail-name-list org-security-trail

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

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

-- 创建 CloudTrail Athena 表
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/';

-- 为近期数据添加分区
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 查询

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

-- 检测未使用 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;

-- 查找 IAM 权限提升尝试
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;

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

-- 查找来自 Tor 出口节点或异常 IP 的 API 调用
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;

-- 检测未授权 API 调用（AccessDenied 模式）
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 构建实时检测

创建用于主动安全监控的实时查询。

# 检测 Root 账户使用
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
  '

# 检测安全组变更
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
  '

# 检测新创建的 IAM 用户或访问密钥
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 基准建议为关键安全事件设置自动告警。

# CIS 3.1：未授权 API 调用告警
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 账户使用告警
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 策略变更告警
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 配置变更告警
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 服务，记录对 AWS 服务发起的 API 调用，提供用户、角色和服务所采取操作的审计跟踪
管理事件（Management Events）	控制平面操作的 CloudTrail 事件，如创建资源、修改 IAM 和配置服务
数据事件（Data Events）	数据平面操作的 CloudTrail 事件，如 S3 对象访问和 Lambda 函数调用，提供细粒度活动日志
日志文件验证（Log File Validation）	CloudTrail 功能，创建摘要文件用于验证日志文件传输后未被篡改
CloudTrail Lake	用于 CloudTrail 事件的托管数据湖，无需管理 Athena 表或 S3 数据即可进行基于 SQL 的查询
组织跟踪（Organization Trail）	将 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 事件在 S3 和 CloudWatch Logs 中最多可能延迟 15 分钟。在活跃事件期间进行实时可见性时，应使用 CloudTrail Lake 或 CloudWatch Logs Insights，而非针对 S3 的 Athena 查询。跨区域攻击需要在 Athena 中查询多个区域分区。

输出格式

CloudTrail 安全分析报告
======================================
账户: 123456789012
分析周期: 2026-02-16 至 2026-02-23
跟踪: org-security-trail（组织范围）

检测到的安全事件:
  Root 账户登录:                  2
  未使用 MFA 的控制台登录:        7
  权限提升尝试:                  12
  CloudTrail 配置变更:            0
  安全组修改:                    34
  未授权 API 调用:               156

高优先级发现结果:
[CT-001] 未使用 MFA 的控制台登录
  用户: admin-user
  时间: 2026-02-22T14:30:00Z
  IP: 203.0.113.50
  所需操作: 通过 IAM 策略强制执行 MFA

[CT-002] IAM 权限提升
  用户: dev-user
  时间: 2026-02-23T03:15:00Z
  事件: CreatePolicyVersion -> AttachRolePolicy
  IP: 185.x.x.x（可疑）
  所需操作: 调查凭据泄露

告警状态:
  已配置 CIS 指标过滤器: 14 / 14
  活跃 CloudWatch 告警: 14 / 14
  已触发告警（过去 7 天）: 8