implementing-cloud-trail-log-analysis

# 实施 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 捕获组织中所有相关事件类型。

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

```sql
-- 创建 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 查询

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

```sql
-- 检测未使用 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 构建实时检测

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

```bash
# 检测 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 基准建议为关键安全事件设置自动告警。

```bash
# 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
```