こんにちは
今回は表題の通り AWS WAF V2 のレートベースルールでブロックされたIPアドレスをSNSで登録したメールアドレスへ飛ばしてみたので、その方法を紹介します
なお、レートベースの詳細な挙動は説明せず、あくまで全体の構成と動作確認などに焦点を当てたいと思います
それでは、さっそくやっていきましょう
■構成
全体の構成としては下記となります(今回もClaudeCodeくんに構成図を作成してもらったのでちょっと見にくいかもしれません)
構成の肝は、CloudWatch Logs のサブスクリプションフィルターで設定した文字列を検知した場合のみ Lambda をキックし、Lambdaで整形したメッセージをSNSのメールアドレスへ飛ばすことです
また、レートベースルールの閾値は 同一IPアドレスから1分間で10回以上のアクセスがあった場合にブロックするようにしており、ブロックが発生するとCloudWatch Logs へ下記のようなログが出力されます
{
～省略～
"terminatingRuleId": "example-dev-rate-based-rule",
"terminatingRuleType": "RATE_BASED",
"action": "BLOCK",
～省略～
}
■やってみた
※事前にネットワーク回り、EC2、ALB、CloudWatch、SNSは事前に作成しておいてください
また、ブラウザなどのインターネット上からEC2上のWebサーバーへFQDNでアクセス可能な状態にしてください
1. LambdaからSNSを操作するためのIAMポリシーとロールを作成
まずは、LambdaからSNSを操作するためのIAMポリシーとロールを作成します
権限は Publish を利用し、リソースは検証なのですべてを設定します
・ポリシー
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": "sns:Publish",
"Resource": "*"
}
]
}
・ロール
Trusted entities
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
→ ポリシーは先ほど作成したものを選択します
2. Lmabdaの作成
Python 3.13 で下記コードを作成しました
import base64
import gzip
import json
import os
from datetime import datetime, timedelta, timezone
import boto3
sns = boto3.client("sns")
SNS_TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]
# 表示タイムゾーン（既定は日本時間 JST = UTC+9）
_TZ_OFFSETS = {"Asia/Tokyo": 9, "UTC": 0}
_TZ_NAME = os.environ.get("TIMEZONE", "Asia/Tokyo")
_TZ = timezone(timedelta(hours=_TZ_OFFSETS.get(_TZ_NAME, 9)), _TZ_NAME)
# ISO 3166-1 alpha-2 → 日本語国名（主要国。未登録はコードをそのまま表示）
COUNTRY_NAMES = {
"JP": "日本", "US": "アメリカ合衆国", "CN": "中国", "KR": "韓国",
"TW": "台湾", "HK": "香港", "SG": "シンガポール", "IN": "インド",
"TH": "タイ", "VN": "ベトナム", "ID": "インドネシア", "PH": "フィリピン",
"MY": "マレーシア", "RU": "ロシア", "GB": "イギリス", "DE": "ドイツ",
"FR": "フランス", "NL": "オランダ", "BR": "ブラジル", "CA": "カナダ",
"AU": "オーストラリア", "UA": "ウクライナ", "TR": "トルコ", "IR": "イラン",
}
def _country_label(code):
"""国コードを『日本 (JP)』の形式に整形。未登録・不明はコードのみ。"""
if not code:
return "不明"
name = COUNTRY_NAMES.get(code.upper())
return f"{name} ({code})" if name else code
def _format_datetime(epoch_ms):
"""エポックミリ秒を表示タイムゾーンの文字列に整形。"""
if epoch_ms is None:
return "不明"
dt = datetime.fromtimestamp(epoch_ms / 1000, tz=_TZ)
return dt.strftime("%Y-%m-%d %H:%M:%S ") + _TZ_NAME
def _iter_waf_logs(event):
"""CloudWatch Logs サブスクリプションイベントを展開し、各 WAF ログ(dict)を返す。"""
data = event.get("awslogs", {}).get("data")
if not data:
return
decoded = gzip.decompress(base64.b64decode(data))
payload = json.loads(decoded)
for log_event in payload.get("logEvents", []):
try:
yield json.loads(log_event["message"])
except (KeyError, json.JSONDecodeError):
continue
def _build_message(waf_log):
"""WAF ログ 1 件から SNS 送信用の件名・本文を生成。"""
http_request = waf_log.get("httpRequest", {})
client_ip = http_request.get("clientIp", "不明")
country = _country_label(http_request.get("country"))
blocked_at = _format_datetime(waf_log.get("timestamp"))
rule_name = waf_log.get("terminatingRuleId", "不明")
subject = f"[WAF] レートベースルールでブロックしました (IP: {client_ip})"
body = (
"AWS WAF のレートベースルールにより、以下の通信をブロックしました。\n"
"\n"
"----------------------------------------\n"
f"ブロック日時 : {blocked_at}\n"
f"IP アドレス : {client_ip}\n"
f"国名 : {country}\n"
f"ルール名 : {rule_name}\n"
"----------------------------------------\n"
"\n"
"内容をご確認ください。"
)
return subject, body
def lambda_handler(event, context):
published = 0
for waf_log in _iter_waf_logs(event):
subject, body = _build_message(waf_log)
sns.publish(
TopicArn=SNS_TOPIC_ARN,
# SNS の件名は最大 100 文字・ASCII 制約があるため安全な固定件名にする
Subject="[WAF] Rate-based rule blocked an IP",
Message=body,
)
published += 1
print(f"Published: {subject}")
return {"statusCode": 200, "published": published}
→ CloudWatchLogsの中身をデコードし、その中の httpRequest から必要な情報を整形しています
また、SNSのARNを環境変数から設定できるようにしています
次に、IAMロールで先ほど 1 の手順で作成したものを指定し、環境変数でSNS_TOPIC_ARNを指定します
3. CloudWatch Logs でサブスクリプションフィルターを設定します
サブスクリプションフィルターの種類は Create Lambda subscription filter を選択し、
フィルターパターンで RATE_BASED と BLOCK を半角でつなげることで AND 条件を実現しています
下記条件で作成しました
Lambda function：2 で作成したLambda関数
フィルターパターン：RATE_BASED BLOCK
4. 動作確認
それでは、一度レートベースに引っ掛かるようにして通知されるか確認してみます
適当にブラウザから対象のFQDNへF5を連打し、403 Forbidden の画面になることを確認します
次に、Lambda が実行されているか確認します
→ Error count and success rate の Success rate のカウントが1になっているので、Lambda関数自体は実行されてました
最後にメールを確認してみます
→ 秘匿すべき箇所はマスクしていますが、ブロック日時、IPアドレス、国名など最低限の情報が記載されたメールを受信したので、きちんと動作してそうです
なお、今回の構成では CloudWatch Logs にログが出力されるたびにメールが通知されるので、もしこの辺りを調整したい場合は、DynamoDBでの状態保持などが必要になってくると思います
■最後に
いかがでしたでしょうか
割と定番の構成にLambdaとSNSを追加するだけで簡単にブロックされたIPアドレスをメール通知できました
これにより毎回AWSコンソールへ行かなくても、ブロックされたIPアドレスが確認できるので、少し楽になるかなと思います
ただ、最後にも記載したようにもし大量にブロックされた場合、大量のメールが飛んでくる可能性があるので、そのあたりはDynamoDBを追加した状態保持が必要になってくるかなと思います
もし時間があればこっちもやってみようかなと思います
それではまたー