この記事について

CloudTrailにはAWSアカウントに対する操作履歴をログとして取得する機能があります。
AWSアカウントが外部から不正アクセスを受けた際に、このログを確認することで不正アクセスの詳細について把握することができます。

CloudTrailだけでも操作履歴の閲覧はできますが、Athenaを利用すると自由に閲覧することができます。

この記事ではAthenaを利用してCloudTrailの操作履歴ログを分析する手順について記載します。

CloudTrailについてのおさらい

まずClouTrailについておさらいです。

デフォルトで取得される操作履歴 (= イベント履歴)

AWS(アカウント)において、CloudTrailはデフォルトで有効になっています。
ただしデフォルトでは以下の制限があります。

保存期間は90日間
記録対象は管理イベントのみ

(管理イベントはAWS Management Console、AWS Command Line Interface、AWSのSDKおよびAPIを通じて行われたイベントを指します)

デフォルトで取得された操作履歴はCloudTrailイベント履歴機能から閲覧することが可能です。

証跡機能

デフォルトではデータイベントとInsightsイベントは取得されません。また90日間以上の長期保存を行うこともできません。

長期保存や全イベントを取得するには、証跡を作成してS3バケットへログファイルとしてイベント継続的に保存します。

やること

あらかじめCloudTrailの証跡を作成して、S3バケットに全イベントの操作履歴ログを保存するように設定しておきます。

このS3バケット内のログに対して、Athenaからクエリを発行することで操作履歴の分析を行います。

Athena ワークグループの作成

Athena ワークグループはクエリ実行や履歴をユーザ/グループの単位で分離するための機能です。

ワークグループを作成することで以下のことが可能になります。

ワークグループへのアクセス制御ができる
クエリ毎のスキャンされるデータ量を制限できる

CloudTrailの操作履歴ログの量が膨大な場合、Athneaのクエリを実行することでとんでもない量のデータに対してスキャンが走り、想定外の利用料金が発生してしまいます。ワークグループを作成して、スキャンされるデータ量を制限することで、想定外の利用料金が発生することを防げます。

以下のようにワークグループを作成しましょう。

設定	値
エンジンタイプ	Athena SQL
認証	IAM
データ制限	10GB

テーブルの作成

クエリエディタの画面を開きます。
右上のワークグループに先ほど作成したものを選択します。

データソースに「AwsDataCatalog」、データベースに「default」を指定します。

クエリ画面に以下のクエリ文を貼り付けます。
なお、貼り付ける前に以下を環境に合わせて置換してください。

置換前	置換後	補足
{テーブル名}	任意	作成されるテーブル名になります
{AWSアカウントID}	分析対象のAWSアカウントID
{リージョン}	分析対象のリージョン	複数ある場合はカンマ区切りで指定します例)us-east-1,ap-northeast-1
{S3バケット名}	CloudTrail証跡の保存先であるS3バケットの名前
{分析期間}	分析対象の期間	分析対象の機関をカンマ区切りで指定します例)2023/11/01から2023/11/30を対象にする。「2023/11/01,2023/11/30」

CREATE EXTERNAL TABLE {テーブル名}(
    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>,
            ec2RoleDelivery:string,
            webIdFederationData:map<string,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,
    serviceEventDetails STRING,
    sharedEventID STRING,
    vpcendpointid STRING,
    tlsDetails struct<
        tlsVersion:string,
        cipherSuite:string,
        clientProvidedHostHeader:string>
  )
PARTITIONED BY (
   `region` string,
   `timestamp` string)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://{S3バケット名}/AWSLogs/{AWSアカウントID}/CloudTrail'
TBLPROPERTIES (
  'projection.enabled'='true', 
  'projection.timestamp.format'='yyyy/MM/dd', 
  'projection.timestamp.interval'='1', 
  'projection.timestamp.interval.unit'='DAYS', 
  'projection.timestamp.range'='{分析期間}', 
  'projection.timestamp.type'='date', 
  'projection.region.type' = 'enum',
  'projection.region.values'='{リージョン}',
  'storage.location.template'='s3://{S3バケット名}/AWSLogs/{AWSアカウントID}/CloudTrail/${region}/${timestamp}')

そしてクエリを実行すると、テーブルが作成されます。

新しいクエリタブを開いて、以下を実行します。

select * from <作成したテーブル名> LIMIT 10;

これでクエリ結果が表示されたら、成功です。

レコードのフィールドは以下リンクになります。
CloudTrail コンテンツの記録

eventSourceはイベントの実行元であるAWSサービス名です。

以下のようにフィルタリングすることで、DynamoDBに関する操作履歴のみを表示させることが可能です。

select * from cloudtrail_logs_test WHERE eventSource = 'dynamodb.amazonaws.com' LIMIT 10;

他にもSQLで記述可能な操作内容であれば、実現することができます。

参考

AWS CloudTrail ログのクエリ