パソコン日和: Python boto3 を使って、CloudTrail からデータを取得する。

こんばんわ

今日は、Python boto3 を使って、AWSのCloudTrailから値を取得するという話です。詳しい方にとってはそんなに難しい話ではないようなのですが、私はいつもかなり四苦八苦しながら書いています。

大まかな流れとしては、以下のようになると思います。結構めんどくさいかもしれません。ちなみにmacOSでやる場合です。Python動くまではWindowsと異なるところもありますが、それ以外はWindowsでも大丈夫だと思います。

・Xcodeコマンドラインツールをインストールする
・（オプション）MacPortをインストールする
・（オプション）MacPortから任意のバージョンのPythonをインストールする
・AWSにIAMユーザを作成してシークレットキーを作成する
・AWSコマンドラインツールをインストールしてシークレットキーを設定する
・Pythonでプログラムを書く

Xcodeコマンドラインツールをインストールする

私も、先日Macの再インストールをしてしまって、綺麗に消してしまったので入れ直しです。AppleがリリースしているXcodeのコマンドラインツールというのをインストールするとPythonはインストールされるので、特にこだわりがなければそれを使って良いと思います。以下のようにコマンドを実行します。

xcode-select --install

MacPortをインストールする

MacPortのインストールは、「MacPort再び。」という記事に書かれていると思いますのでそちらを参照してみてください。

MacPortから任意のバージョンのPythonをインストールする

MacPortをインストールしたら、Pythonをインストールします。ものすごくたくさんのバージョンがあってどれがいいんだかよくわからないのですが、安定版で新そうなものをインストールします。いくつか依存するパッケージもインストールされます。boto3が2.7または3.4以上ということらしいので3.4以上にしておけば良いのではないかと思います。

sudo port install python312

# python3.12
Python 3.12.11 (main, Jun 6 2025, 23:18:08) [Clang 16.0.0 (clang-1600.0.26.6)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>

毎回python3.12と入力するのも面倒ですので、お好みで以下のいずれかのコマンドを実行します。私は、python3 に設定しました。

sudo port select --set python python312
sudo port select --set python3 python312

ターミナルを一度再起動して、python3 と実行すると、python3.12 が起動するようになると思います。rehashコマンドでコマンドの一覧を更新しても大丈夫だったかも。

# python3
Python 3.12.11 (main, Jun 6 2025, 23:18:08) [Clang 16.0.0 (clang-1600.0.26.6)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>>

あと、AWSにアクセスするために boto3 というライブラリを使いますので、そちらもインストールします。以下のようにコマンドを入力します。

sudo port install py312-boto3

ついでにpythonのライブラリの追加削除するツールも入れちゃいます。

sudo port inistall py312-pip
sudo port select --set pip3 pip312
Password:
Selecting 'pip312' for 'pip3' succeeded. 'pip312' is now active.

だんだんと面倒になってきましたね。。。。私どうせ大したことしませんので、XcodeのPython3とpip3で良かったような気がしてきました。みなさんはそうしても良いかと思います。

AWSにIAMユーザを作成してシークレットキーを作成する

「aws cli を使ってEC2のインスタンスの起動や停止を行う」にIAMユーザを作成してシークレットキーを作成する手順が書いてありますのでそちらを参照しながら設定してください。記事ではEC2へのアクセスを許可する設定をしていると思うのですが、同様にして、CloudTrailへのアクセスも設定してください。

AWSコマンドラインツールをインストールしてシークレットキーを設定する

Pythonでboto3を使うだけならばコマンドラインツールはなくてもいいようなのですが、ちょっとした確認などで使いたいのでインストールします。

「AWSコマンドラインインターフェイスを使ってみる」にコマンドラインツールのインストール方法が書かれていますのでそちらを参照するか、特に難しいことはなかったと思いますので、以下のURLから説明を読んでインストールしてください。こちらは私のブログではなくAWSのドキュメントです。

AWS CLI の最新バージョンのインストールまたは更新

説明がどんどん手抜きになっていきますね。。。すみません。

AWSコマンドラインツールが動作しているか確認する

以下のコマンドを入力して、何か応答があるか確認します。

aws ec2 describe-availability-zones

以下のように結果が表示されれば動作していると思います。

aws ec2 describe-availability-zones

{
"AvailabilityZones": [
{
"OptInStatus": "opt-in-not-required",
"Messages": [],
"RegionName": "ap-northeast-1",
"ZoneName": "ap-northeast-1a",
"ZoneId": "apne1-az4",
　　・
　　・
　　・

認証情報の設定はpythonからも参照されますので、ちゃんと動作するように設定しておく必要があります。

Pythonでプログラムを書く

動作したスクリプトを以下に。。。すみません、どんどん手抜きに。。。

CloudTrailからStartInstancesイベントで、現在から過去２４時間の物を取得してきて、イベントの日時と起動時のエラーがあればエラーメッセージをとインスタンスIDを表示するだけのものを書きました。プログラムの中に、ポイントと思われる箇所をコメントします。

#!/opt/local/bin/python3

import warnings
import boto3
import json
import pprint

from datetime import datetime
from datetime import timedelta
from dateutil import tz
JST = tz.gettz('Asiz/Tokyo')
UTC = tz.gettz('UTC')

cloudtrail = boto3.client('cloudtrail', region_name = 'ap-northeast-1')

#AWS内では時刻はUTCで扱われているので、UTCで現在時刻と、２４時間前の時刻を求めています。

End = datetime.now().astimezone(UTC).replace(second=0,microsecond=0)
Start = End - timedelta(hours=24)

# アトリビュートのキーと値、検索開始・終了時刻と、取得する結果の最大数を指定します。たしか５０が最大だったと思います。取りこぼさないようにするには、StartTime/EndTimeを工夫する必要がありそうです。

response = cloudtrail.lookup_events(
LookupAttributes = [
{
'AttributeKey': 'EventName',
'AttributeValue': 'StartInstances'
},
],
StartTime = Start,
EndTime = End,
MaxResults = 50
)

events = response.get('Events', [] )

# 結果はjson形式で返ってきますので、配列に読み込みます。どういう形式で値が送られてくるかわかりにくいのですが、pprint.pprint(response)のようにして内容を確認するか、cloudtrailのイベント履歴の中をじっくり見るのが良いと思います。

for event in events:
cloudtrail_event = json.loads(event['CloudTrailEvent'])
eventtime = cloudtrail_event['eventTime']

# エラーがなかったときは、アトリビュートそのものがなくなってしまうので、値が定義されていなかったら新たに定義するようにしています。例外処理でやるのが正しいのかどうかはわかりません。

try:
cloudtrail_event['errorMessage']
except:
errormessage = "NoError."
else:
errormessage = cloudtrail_event['errorMessage']

# いつも奥まった値のアクセスに「あれ？」となります。結局値やドキュメントと睨めっこが始まってしまうのですが、何かいい方法がないのでしょうか？

instance_id = cloudtrail_event['requestParameters']['instancesSet']['items'][0]['instanceId']

# 時刻はUTCですので、表示する際にJSTに変換しています。

temp_datetime = datetime.strptime(eventtime, '%Y-%m-%dT%H:%M:%SZ')
temp_datetime = temp_datetime + timedelta(hours=9)

msg = "%s,%s,%s" % ( temp_datetime,errormessage,instance_id)

print(msg)

どんどん手抜きになって行って、最後はなんだかよくわからない感じになってしまったかもしれませんが、自分の覚書というつもりなのでご容赦ください。

パソコン日和

2025年7月11日金曜日

Python boto3 を使って、CloudTrail からデータを取得する。