如何使用 CloudWatch Logs Insights 分析我的 CloudTrail 日志?

上次更新日期:2022 年 4 月 18 日

我想使用 Amazon CloudWatch Logs Insights 分析我的 Amazon CloudTrail 日志。我该如何操作?

简短描述

您可以将 CloudTrail 配置为将日志发送到 CloudWatch Logs。这样,您就可以使用 CloudWatch Logs Insights 分析 CloudTrail 日志,以监控特定的账户活动。

为了展示 Logs Insights 分析 CloudTrail 日志的能力,以下解决方法提供了一些示例查询。这些查询涵盖了最常见的使用案例:

  • 隔离特定的日志字段。
  • 根据不同的条件进行筛选。
  • 聚合事件。
  • 构建时间序列。

解决方法

以下查询将探讨 Amazon Simple Storage Service(Amazon S3)存储桶和对象活动。默认情况下,CloudTrail 不会捕获 S3 数据事件。您可以在 CloudTrail 中启用事件日志记录。有关更多信息,请参阅为 S3 存储桶和对象启用 CloudTrail 事件日志记录

在这些示例查询的基础上,您可以根据您的使用场景创建其他更复杂的 Logs Insights 查询。您还可以将查询与 CloudWatch 控制面板集成,将查询可视化为图表和图形以及相关指标。

查询 1:最新事件

目标

使用默认的 @timestamp@message 字段检索最近的 CloudTrail 日志事件。

查询

#Retrieve the most recent CloudTrail events
fields @timestamp, @message
| sort @timestamp desc
| limit 2

结果

@timestamp @message
2022-02-18 17:52:31.118

{"eventVersion":"1.08","userIdentity":{"type":"AssumedRole","principalId":"AROAWZKRRJU47ARZN7ECC:620d7d78144334d6933c27195cae2a98", "arn":"arn:aws:sts::123456789012:assumed- role/Amazon_EventBridge_Invoke_Run_Command_371790151/620d7d78144334d6933c27195cae2a98","accountId":"123456789012", "accessKeyId":"ASIAWZKRRJU4Y45M4SC6","sessionContext":{"sessionIssuer": {"type":"Role","principalId":"AROAWZKRRJU47ARZN7ECC","arn":"arn:aws:iam::123456789012:role/service- role/Amazon_EventBridge_Invoke_Run_Command_371790151","accountId":"123456789012","userName": "Amazon_EventBridge_Invoke_Run_Command_371790151" (output truncated)

2022-02-18 17:51:52.137 {"eventVersion":"1.08","userIdentity":{"type":"AssumedRole","principalId":"AROAWZKRRJU43YP4FHR2N:StateManagerService","arn":"arn:aws:sts::123456789012:assumed-role/AWSServiceRoleForAmazonSSM/StateManagerService","accountId":"123456789012","sessionContext":{"sessionIssuer":{"type":"Role","principalId":"AROAWZKRRJU43YP4FHR2N","arn":"arn:aws:iam::123456789012:role/aws-service-role/ssm.amazonaws.com/AWSServiceRoleForAmazonSSM","accountId":"123456789012","userName":"AWSServiceRoleForAmazonSSM"}, "webIdFederationData":{},"attributes":{"creationDate":"2022-02-18T17:50:06Z","mfaAuthenticated":"false"}},"invokedBy":"ssm.amazonaws.com"},"eventTime":"2022-02-18T17:50:06Z","eventSource":"ec2.amazonaws.com","eventName":"DescribeInstances","awsRegion":"eu-west-1","sourceIPAddress":"ssm.amazonaws.com","userAgent":"ssm.amazonaws.com","requestParameters":{"maxResults":50,"instancesSet": (output truncated)

查询 2:分解各个字段

目标

  • 隔离 @message 中的各个字段。
  • CloudTrail event(CloudTrail 事件)中显示选择字段。

查询

#Breakout Individual Fields
fields @timestamp, awsRegion, eventCategory, eventSource, eventName, eventType, sourceIPAddress, userIdentity.type
| sort @timestamp desc
| limit 2

结果

@timestamp awsRegion eventCategory eventSource eventName eventType sourceIPAddress userIdentity.type
2022-02-18 18:00:09.647 ca-central-1 管理 sts.amazonaws.com AssumeRole AwsApiCall cloudtrail.amazonaws.com AWSService
2022-02-18 18:00:09.647 ca-central-1 管理 sts.amazonaws.com AssumeRole AwsApiCall cloudtrail.amazonaws.com AWSService

查询 3:按 Amazon Elastic Compute Cloud(Amazon EC2)RunInstances 筛选

目标

  • 检索 Cloudtrail 事件中的特定字段。
  • 使用更有意义的标签重命名字段。
  • 根据 API 调用筛选在此账户中启动的 latest EC2 instances(最新 EC2 实例)。

查询

#EC2: Recently Launched Instances
fields eventTime, eventName as API, responseElements.instancesSet.items.0.instanceId as InstanceID, userIdentity.sessionContext.sessionIssuer.type as IssuerType, userIdentity.type as IdentityType, userIdentity.sessionContext.sessionIssuer.userName as userName
| filter eventName = 'RunInstances'
| sort eventTime desc
| limit 2

结果

eventTime API InstanceID IssuerType IdentityType userName
2022-02-18T17:36:38Z RunInstances i-0325b4d6ae4e93c75 角色 AssumedRole AWSServiceRoleForAutoScaling
2022-02-18T13:45:18Z RunInstances i-04d17a8425b7cb59a 角色 AssumedRole AWSServiceRoleForAutoScaling

查询 4:按控制台登录筛选:最近

目标

  • 检索 Cloudtrail 事件中的特定字段。
  • 使用更有意义的标签重命名字段。
  • 根据 API 调用筛选 latest logins via the AWS Console(通过 AWS 控制台的最新登录)。

查询

#Console Login: Most Recent API Calls
fields eventTime, eventName, responseElements.ConsoleLogin as Response, userIdentity.arn as ARN, userIdentity.type as User_Type
| filter eventName = 'ConsoleLogin'
| sort eventTime desc
| limit 10

结果

eventTime eventName 响应 ARN User_Type
2022-02-18T17:35:44Z ConsoleLogin 成功 arn:aws:iam::123456789012:user/test_user IAMUser
2022-02-17T13:53:58Z ConsoleLogin 成功 arn:aws:sts::123456789012:assumed-role/Admin/test_user AssumedRole

查询 5:按控制台登录进行筛选:身份验证失败

目标

  • 检索 Cloudtrail 事件中的特定字段(例如,用户名、用户类型、源 IP)。
  • 使用更有意义的标签重命名字段。
  • 筛选 latest unsuccessful logins via the AWS Console(通过 AWS 控制台的最新失败登录)。

查询

#ConsoleLogin: Filter on Failed Logins
fields eventTime, eventName, responseElements.ConsoleLogin as Response, userIdentity.userName as User, userIdentity.type as User_Type, sourceIPAddress, errorMessage
| filter eventName = 'ConsoleLogin' and responseElements.ConsoleLogin = 'Failure'
| sort eventTime desc
| limit 10

结果

eventTime eventName 响应 用户 User_Type sourceIPAddress errorMessage
2022-02-18T20:10:55Z ConsoleLogin 失败 echo IAMUser 12.34.56.89 身份验证失败
2022-02-18T20:10:43Z ConsoleLogin 失败 echo IAMUser 12.34.56.89 身份验证失败

查询 6:按 Simple Storage Service(Amazon S3)对象上载筛选

目标

  • 检索 Cloudtrail 事件中的特定字段。
  • 使用更有意义的标签重命名字段。
  • API 调用筛选并定位 S3 存储桶

查询

#Filter PutObject API Calls on a specific S3 Bucket
fields @timestamp, eventName as API, requestParameters.bucketName as BucketName, requestParameters.key as Key, userIdentity.sessionContext.sessionIssuer.userName as UserName
| filter eventName = 'PutObject' and BucketName = 'target-s3-bucket'
| sort @timestamp desc
| limit 2

结果

@timestamp API BucketName 密钥 UserName
2022-02-12 17:16:07.415 PutObject test_bucket1 w4r9Hg4V7g.jpg  
2022-02-12 16:29:43.470 PutObject test_bucket2 6wyBy0hBoB.jpg  

查询 7:汇总 S3 活动

目标

  • 根据 S3 服务进行筛选。
  • 根据计数统计数据聚合所有匹配的事件。
  • 根据 API、S3 存储桶密钥拼接结果。
  • 使用 stats 命令重命名字段。
  • 降序排序。

查询

#S3 Activity: Bucket & Key Details
filter eventSource = 's3.amazonaws.com'
| stats count(*) as Hits by eventName as API, requestParameters.bucketName as BucketName, requestParameters.key as Key
| sort Hits desc
| limit 5

结果

API BucketName 密钥 命中结果
ListAccessPoints     44
GetBucketAcl team1-ctrail-multi-region   27
GetBucketAcl team2-dub-cloudtrail   27
GetBucketAcl aws-cloudtrail-logs-123456789012-ba940dd7   26
GetObject devsupport-prod rdscr/individual/123456789012 18

查询 8:汇总 AWS Key Management Service(AWS KMS)解密活动

目标

  • 根据 KMS 服务Decrypt API 进行筛选。
  • 使用 fields 命令重命名字段并聚合用户友好名称。
  • 根据计数统计数据聚合所有匹配的事件。
  • 根据 KMS 密钥用户拼接结果。
  • 按降序排序。

查询

#KMS Decrypt Activity: Key & User Details
fields resources.0.ARN as KMS_Key, userIdentity.sessionContext.sessionIssuer.userName as User
| filter eventSource='kms.amazonaws.com' and eventName='Decrypt'
| stats count(*) as Hits by KMS_Key, User
| sort Hits desc
| limit 2

结果

KMS_Key 用户 命中结果
arn:aws:kms:us-east-1:123456789012:key/03f2923d-e213-439d-92cf-cbb444bd85bd AWSServiceRoleForConfig 12
arn:aws:kms:us-east-1:123456789012:key/03f2923d-e213-439d-92cf-cbb444bd85bd FoxTrot-1UQJBODTWZYZ6 8

查询 9:汇总包含错误的 API 调用

目标

  • 根据是否存在 errorCode 字段进行筛选。
  • 根据计数统计数据聚合所有匹配的事件。
  • 根据 AWS 服务、APIerrorCode 拼接结果。
  • 使用 stats 命令重命名字段。
  • 按最大匹配数进行排序。

查询

#Summarize API Calls with Errors
filter ispresent(errorCode)
| stats count(*) as Num_of_Events by eventSource as AWS_Service, eventName as API, errorCode
| sort Num_of_Events desc
| limit 5

结果

AWS_Service API errorCode Num_of_Events
s3.amazonaws.com GetBucketPublicAccessBlock NoSuchPublicAccessBlockConfiguration 79
lambda.amazonaws.com GetLayerVersionPolicy20181031 ResourceNotFoundException 66
s3.amazonaws.com GetBucketPolicyStatus NoSuchBucketPolicy 60
s3.amazonaws.com HeadBucket AccessDenied 47
logs.amazonaws.com CreateLogStream ResourceNotFoundException 21

查询 10:汇总包含错误代码的 S3 API 调用

目标

  • 根据 S3 服务和是否存在 errorCode 字段进行筛选。
  • 根据计数统计数据聚合所有匹配的事件。
  • 根据 errorCodeerrorMessage 拼接结果。
  • 按最大匹配数进行排序。

查询

#S3: Summarize Error Codes
filter eventSource = 's3.amazonaws.com' and ispresent(errorCode)
| stats count(*) as Hits by errorCode, errorMessage
| sort Hits desc
| limit 5

结果

errorCode errorMessage 命中结果
AccessDenied 访问遭拒 86
NoSuchBucketPolicy 存储桶策略不存在 80
NoSuchPublicAccessBlockConfiguration 找不到公有访问阻止配置 79
ObjectLockConfigurationNotFoundError 此存储桶的对象锁定配置不存在 3
ServerSideEncryptionConfigurationNotFoundError 找不到服务器端加密配置 3

查询 11:汇总 AWS 服务、API 和 AWS Identity and Access Management(IAM)用户的 AccessDenied/UnauthorizedOperation API 调用

目标

  • 根据 AccessDeniedUnauthorizedOperation CloudTrail 事件筛选。
  • 根据计数统计数据聚合所有匹配的事件。
  • 根据 errorCode、AWS 服务、APIIAM 用户/角色拼接结果。
  • 使用 stats 命令重命名字段。
  • 按降序排序。

查询

#Summarize AccessDenied/UnauthorizedOperation API Calls by AWS Service, API, IAM User
filter (errorCode='AccessDenied' or errorCode='UnauthorizedOperation')
| stats count(*) as NumberOfEvents by errorCode, eventSource as AWS_Service, eventName as API, userIdentity.type as IdentityType, userIdentity.invokedBy as InvokedBy
| sort NumberOfEvents desc
| limit 10

结果

errorCode AWS_Service API IdentityType InvokedBy NumberOfEvents
AccessDenied s3.amazonaws.com HeadBucket AWSService delivery.logs.amazonaws.com 83
AccessDenied s3.amazonaws.com GetObject AssumedRole   9

查询 12:时间序列:KMS 每小时调用量

目标

  • 根据 KMS 服务Decrypt API 进行筛选。
  • 将所有匹配的事件聚合到 1 小时 bin 中。
  • 在折线图上显示可视化结果。

查询

#KMS: Hourly Decrypt Call Volume
filter eventSource='kms.amazonaws.com' and eventName='Decrypt'
| stats count(*) as Hits by bin(1h)

结果

bin(1h) 命中结果
2022-02-18 19:00:00.000 16
2022-02-18 18:00:00.000 25
2022-02-18 17:00:00.000 28
2022-02-18 16:00:00.000 14
2022-02-18 15:00:00.000 16