S3に溜めたCloudTrailログを出力してみる – OPT PLANNING ブログ

１．目的

証跡ログとしてS3に出力したCloudTrailログ（gzで圧縮）の中身を確認したく，AWSコンソールからだと，いちいちファイルを選択して開く必要があります。いい方法は無いかと，ネット検索しましたら，athenaを利用すると良いようです。しかし，こちらの方法では，料金がかかるので，pythonで出力するプログラムを作成してみました。

２．システム構成図

３．S3バケットの構造

保存先のs3バケット名は，以下の名前。
target_bucket = 'trail-test'
パスは以下のようになります。（東京リージョンのログの場合）。
target_path = 'AWSLogs/アカウントid/CloudTrail/ap-northeast-1/'
この下に yyyy/mm/dd/ファイル名.json.gzのように保存されています。

４．pythonプログラム

#!/usr/bin/env python3.8
import boto3
import gzip
import io
import sys
import os
import traceback
import json

target_bucket = 'trail-test'
target_path = 'AWSLogs/123456789012/CloudTrail/ap-northeast-1/2024/07/22'

def main():
  try:
    s3 = boto3.resource('s3')
    s3client = boto3.client('s3')
    my_bucket = s3.Bucket(target_bucket)
    f = open("22.json","w+")
    decoder = json.JSONDecoder()
    for object in my_bucket.objects.all():
      if target_path in object.key:
        obj = s3client.get_object(
          Bucket=target_bucket,
          Key=object.key)['Body'].read()
        file = gzip.open(io.BytesIO(obj), 'rt')
        for row in file.readlines():
          str_t = decoder.raw_decode(row)
          str_s = json.dumps(str_t,indent=2,ensure_ascii=False)
          str_s = str_s[1:-1]
          f.write(str_s)
    f.close()

  except json.decoder.JSONDecodeError as ex:
    err_message = ex.__class__.__name__
    t = traceback.format_exception_only(type(ex), ex)
  except Exception as ex:
    err_message = ex.__class__.__name__
    t = traceback.format_exception_only(type(ex), ex)
    print(t,err_message)
    sys.exit(1)

if __name__ == '__main__':
  main()

説明

target_path = 'AWSLogs/123456789012/CloudTrail/ap-northeast-1/2024/07/22'
ログの量が大量なので日付を指定して取得するようにしました。
月単位であれば、　2024/07　で大丈夫ですが，時間がかかると思います。

f = open("22.json","w+")        ログを出力するファイル名
for object in my_bucket.objects.all(): 
　　　バケット配下のすべてのファイルを取得

if target_path in object.key:
　取得したファイルパス(object.key)が出力したい対象のパス(target_path)を含んでいたら取得対象のファイルとする。それ以外は，取得しない（何もしない）。

obj = s3client.get_object(
         Bucket=target_bucket,
         Key=object.key)['Body'].read()
オブジェクト（ファイル）を取得

file = gzip.open(io.BytesIO(obj), 'rt')
取得したファイル（json.gz)を開く

５．見直し後のpythonプログラム

このままでは、ログを取得するのに時間がかかるので、処理を見直しました。

保存先のs3バケット名は，以下の名前。
target_bucket = 'trail-test'
target_path = 'AWSLogs/アカウントid/CloudTrail/ap-northeast-1/'
この下に yyyy/mm/dd/ファイル名.json.gzのように保存されています。

見直した箇所

for object in my_bucket.objects.all():　すべて取得から
objects.filter(prefix=target_path)で、上記パスの下のオブジェクトを取得
for object in my_bucket.objects.filter(Prefix=target_path):

#!/usr/bin/env python3.8
import boto3
import gzip
import io
import sys
import os
import traceback
import json

target_bucket = 'trail-test'
target_path = 'AWSLogs/123456789012/CloudTrail/ap-northeast-1/2024/07/22'

def main():
  try:
    s3 = boto3.resource('s3')
    s3client = boto3.client('s3')
    my_bucket = s3.Bucket(target_bucket)
    f = open("22.json","w+")
    decoder = json.JSONDecoder()
    for object in my_bucket.objects.filter(Prefix=target_path):
      obj = s3client.get_object(
        Bucket=target_bucket,
        Key=object.key)['Body'].read()
      file = gzip.open(io.BytesIO(obj), 'rt')
      for row in file.readlines():
        str_t = decoder.raw_decode(row)
        str_s = json.dumps(str_t,indent=2,ensure_ascii=False)
        str_s = str_s[1:-1]
        f.write(str_s)
    f.close()

  except json.decoder.JSONDecodeError as ex:
    err_message = ex.__class__.__name__
    t = traceback.format_exception_only(type(ex), ex)
  except Exception as ex:
    err_message = ex.__class__.__name__
    t = traceback.format_exception_only(type(ex), ex)
    print(t,err_message)
    sys.exit(1)

if __name__ == '__main__':
  main()

出力したファイル(22.json)の中身（一部分）早くなりました。
{
    "Records": [
      {
        "eventVersion": "1.09",
        "userIdentity": {
          "type": "AWSService",
          "invokedBy": "cloudtrail.amazonaws.com"
        },
        "eventTime": "2024-07-22T00:01:06Z",
        "eventSource": "s3.amazonaws.com",
        "eventName": "PutObject",
        "awsRegion": "ap-northeast-1",
        "sourceIPAddress": "cloudtrail.amazonaws.com",
        "userAgent": "cloudtrail.amazonaws.com",
        "requestParameters": {
          "bucketName": "cloudtrail-log-123456789012",
          "x-amz-acl": "bucket-owner-full-control",

eventNameと eventTime　eventSourceのみを抽出
grep -e eventName -e eventTime -e eventSource 22.json
        "eventTime": "2024-07-22T00:01:06Z",
        "eventSource": "s3.amazonaws.com",
        "eventName": "PutObject",
        "eventTime": "2024-07-22T00:01:05Z",
        "eventSource": "dynamodb.amazonaws.com",
        "eventName": "DescribeStream",
        "eventTime": "2024-07-22T00:01:23Z",
        "eventSource": "s3.amazonaws.com",
        "eventName": "PutObject",
　　　　　"eventTime": "2024-07-22T00:00:26Z",
        "eventSource": "sts.amazonaws.com",
        "eventName": "AssumeRole",
        "eventTime": "2024-07-22T00:00:41Z",
        "eventSource": "sts.amazonaws.com",
        "eventName": "AssumeRole",
        "eventTime": "2024-07-22T00:01:01Z",
        "eventSource": "s3.amazonaws.com",
        "eventName": "GetBucketAcl",

１．目的

２．システム構成図

３．S3バケットの構造

４．pythonプログラム

５．見直し後のpythonプログラム

関連