おい、ちょっと手を止めてくれ。
先週、某クライアントのAWS環境で起きたインシデントのフォレンジック調査が終わったんだがね。攻撃者はWebアプリケーションの脆弱性を突いて一時的な足がかりを得たあと、静かに、そして素早くIAMの権限昇格を試みていた。幸い、我々が仕込んでおいたSIEM（Security Information and Event Management）の相関分析ルールが火を吹き、データが持ち出される数分前にアラートを検知、事なきを得た。
「なぜ攻撃者の足音に気づけたのか？」
その答えが、今回お前たちに共有する「CloudTrailとVPC Flow Logsの完全集約、そしてSIEMによるリアルタイム相関分析」だ。
教科書には「ログを収集しましょう」としか書いてないが、現場の現実はそんな甘くない。数千万行のログの海から、本物の攻撃者を見つけ出すための泥臭い設計と、実務でそのまま使えるコードを叩き込んでいく。ついて来いよ。
—
1. 攻撃者が狙う盲点：なぜ「ログを貯めているだけ」では防げないのか？
多くの開発チームは、「AWS CloudTrailを有効にしているから大丈夫」「S3にログが溜まっているから監査もバッチリ」と安心しきっている。だが、それはセキュリティ屋から言わせれば「防犯カメラの映像をただ録画しているだけで、強盗が来ても誰もモニターを見ていない状態」と同じだ。
攻撃者は、我々が想像するよりもはるかに巧妙に「通常のオペレーション」を模倣する。
例えば、次のようなシナリオを考えてみてほしい。
1. Webサーバーの脆弱性（RCE等）を突かれ、インスタンスプロファイル（IAMロール）のクレデンシャルが窃取される。
2. 攻撃者はその一時クレデンシャルを使い、外部のIPアドレスから AssumeRole や不審な GetObject を実行する。
3. 個別のログを単体で見れば、アクセスキーは正しく、エラーも出ていないため「ただの正常なAPIコール」に見える。
これを防ぐには、「どのIPから」「どのユーザーが」「どのリソースに対して」「普段とは異なる挙動をしたか」を時間軸（タイムスタンプ）で結びつける相関分析が不可欠なのだ。
—
2. ログ集約のアーキテクチャとSIEM連携の要諦
実務では、AWS Organizationsを活用して全メンバーアカウントの CloudTrail と VPC Flow Logs を、セキュリティ専用の集約アカウント（ログアーカイブアカウント）のS3バケットへリアルタイムに集約する。
そして、そのログをAmazon Kinesis Data Firehoseなどを経由して、ELK Stack（Elasticsearch / OpenSearch）やDatadog、SplunkといったSIEM基盤へ流し込むのが王道だ。
ここで絶対に外してはいけないポイントが2つある。
- VPC Flow Logsのプレフィックスとフォーマット：デフォルトのログフォーマットでは、セキュリティ解析に必要な情報（特にパケットの方向やTCPフラグ）が不足することがあるため、カスタムフォーマットを定義してメタデータをリッチにしておくこと。
- タイムスタンプの同期（NTP）：クラウドインフラ全体で時刻がズレていれば、相関分析の精度は一瞬で崩壊する。Amazon Time Sync Serviceを確実に全インスタンスで有効化しておけ。
—
3. 【実務で使える】不審なAPIコールを検知するPython相関分析スクリプト
ここでは、SIEMに集約されたCloudTrailのJSONログから、「深夜帯の異常な権限昇格（AttachRolePolicy や CreateUser 等）」と「通常と異なるIPからのアクセス」を検知し、Slack等へ即座に通知するPythonスクリプトの実装例を共有する。
実務の現場でそのままCronやAWS Lambdaのバックエンドとして動かせるよう、例外処理と構造化ログ出力を組み込んでいる。
import json
import logging
from datetime import datetime, timezone
import boto3
from botocore.exceptions import ClientError
# ログ出力の設定（SIEMやCloudWatch Logsへ流す前提のフォーマット）
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)
# 検知対象とする高リスクなIAMアクションのホワイトリスト（これ以外は要警戒）
HIGH_RISK_ACTIONS = {
"CreateUser",
"CreateAccessKey",
"AttachRolePolicy",
"PutUserPolicy",
"UpdateAssumeRolePolicy"
}
# 許可された社内網のCIDR（実環境ではIPAMやVPCの設定値と連動させること）
TRUSTED_CIDR_PREFIX = "203.0.113."
def analyze_cloudtrail_log(event_record):
"""
CloudTrailの1レコードを受け取り、不審な挙動がないか相関分析を行う
"""
try:
event_name = event_record.get("eventName")
user_identity = event_record.get("userIdentity", {})
source_ip = event_record.get("sourceIPAddress")
event_time = event_record.get("eventTime")
# 1. 高リスクなIAMアクションかチェック
if event_name in HIGH_RISK_ACTIONS:
# 2. ユーザータイプの検証（ルートアカウントや予期せぬIAMユーザーの利用を検知）
principal_id = user_identity.get("principalId", "Unknown")
user_name = user_identity.get("userName", "AssumedRole/Service")
# 3. 送信元IPの検証（社内からのアクセスか？）
is_external_ip = not source_ip.startswith(TRUSTED_CIDR_PREFIX) if source_ip else True
if is_external_ip:
# 外部IPからの高リスク操作は「重大インシデント（P1）」としてフラグ立て
alert_payload = {
"severity": "CRITICAL",
"timestamp": event_time,
"eventName": event_name,
"userName": user_name,
"sourceIP": source_ip,
"principalId": principal_id,
"message": f"🚨 [要警戒] 外部IPから高リスクなIAM操作が検出されました！"
}
dispatch_security_alert(alert_payload)
return True
except Exception as e:
logger.error(f"ログ解析中にエラーが発生しました: {str(e)}", exc_info=True)
return False
def dispatch_security_alert(alert_data):
"""
検知したアラートをセキュアに外部連携（SlackやSIEM）へ送信する関数
"""
# 実務ではここでSNSトピックの発行や、SIEM（Splunk/Datadog）へのWebhook送信を行う
logger.critical(json.dumps(alert_data, ensure_ascii=False))
# 例: Boto3を使用してSNSへ緊急通知を発出する場合
# sns_client = boto3.client('sns', region_name='ap-northeast-1')
# sns_client.publish(
# TopicArn='arn:aws:sns:ap-northeast-1:123456789012:SecurityAlertsTopic',
# Message=json.dumps(alert_data),
# Subject='【緊急】セキュリティインシデントの可能性を検知'
# )
# --- 動作検証用のモックデータ（実際のSIEMストリームやLambdaハンドラーから渡される想定） ---
if __name__ == "__main__":
sample_event = {
"eventVersion": "1.08",
"eventTime": datetime.now(timezone.utc).isoformat(),
"eventSource": "iam.amazonaws.com",
"eventName": "AttachRolePolicy",
"userIdentity": {
"type": "IAMUser",
"principalId": "AIDAXXXXXXXXXXXXXXXXX",
"userName": "web-app-worker"
},
"sourceIPAddress": "198.51.100.42", # 外部の不審なIPアドレス
"awsRegion": "ap-northeast-1"
}
logger.info("相関分析エンジンを起動しました。サンプルログの検証を開始します。")
is_threat_detected = analyze_cloudtrail_log(sample_event)
if is_threat_detected:
logger.warning("テスト実行: 脅威が正常に検知され、アラートロジックが発火しました。")—
4. 現場のチーフからのアドバイス：運用を継続させるための極意
コードを書いて終わり、ではない。ここからが腕の見せ所だ。
SIEMの相関分析ルールを導入した直後は、いわゆる「False Positive（誤検知の嵐）」に開発チームが疲弊することがよくある。例えば、デプロイツールや自動化スクリプトが正当な理由で一時的なIPからAPIを叩き、アラートが鳴りやまなくなるケースだ。
その時は、次のようにアプローチしてくれ。
1. 例外を「隠す」な、「明示的にタグ付け」しろ：誤検知する正当なプロセスがあるなら、CloudTrailのUser-Agentやリソースタグに適切な識別子を持たせ、相関分析ルールのホワイトリストに正当な理由とともにコードとして残すこと。
2. 「自動封じ込め（Auto-Remediation）」への布石：検知してSlackに通知するだけでなく、ゆくゆくは「重大な相関アラートが検知された瞬間に、該当IAMユーザーのセッションを強制無効化（RevokeActiveSessions）するスクリプト」まで自動連携できるように設計を練っておけ。
セキュリティとは、完璧な防御壁を作ることではなく、「敵の侵入を最短で察知し、被害を最小化し続ける仕組み」の維持だ。
今回の実装と設計思想を、今お前が担当しているプロジェクトのインフラにどう組み込めるか、さっそくコードレビューの視点を取り入れて見直してみてくれ。期待しているぞ。
コメント