killvxk 实施 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 告警。安全团队需要追踪受泄露凭据执行的所有操作。
方法：
- 跨所有区域查询 CloudTrail，获取受泄露 AccessKeyId 的所有事件
- 构建 API 调用时间线，了解攻击序列
- 识别初始访问点（密钥首次从恶意 IP 出现的时间）
- 映射攻击者创建、修改或访问的所有资源
- 检查持久化机制（新用户、访问密钥、Lambda 函数、EC2 实例）
- 验证 CloudTrail 未被篡改（检查 StopLogging 或 UpdateTrail 事件）
- 记录完整攻击链和影响范围，用于事件响应报告
常见陷阱：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