Amazon Kinesis Data FirehoseからHTTPエンドポイントに配信したAWS WAFのログをAWS Lambdaで取り出してみた

クラウドインテグレーション2部の山下です。

AWS WAFのログは、2022年10月現在、以下のいずれかに送ることができます。

Amazon CloudWatch Logs（以下、CloudWatch Logs）
Amazon Kinesis Data Firehose（以下、Kinesis Firehose）
Amazon S3（以下、S3）

上記のうち、Kinesis Firehoseに送信すれば、そこから配信ストリームで様々な送信先にログを配信することが可能です。

私はこれまでCloudWatch LogsとS3にしかAWS WAFログを送ったことがなかったので、今回は、AWS WAFログをKinesis FirehoseからHTTPエンドポイント（Amazon API Gateway、以下API Gateway）へ送信し、バックエンド（AWS Lambda、以下Lambda）で中身を取り出してみたいと思います。

今回の検証構成
Kinesis Firehoseから配信したデータをLambdaで取り出す
文字列を辞書型・リスト型に変換する
AWS WAFログ部分を取り出し、BASE64デコードする
S3のバックアップデータとの比較
おわりに

今回の検証構成

今回は下図の構成で検証します。

AWS WAFでログを有効化し、送信先をKinesis Firehoseに設定します。

Kinesis Firehoseでは配信ストリームをHTTPエンドポイントにして、API GatewayのURLを指定します。それ以外はデフォルト設定です。また、配信データは全てS3にバックアップし、後でLambdaの表示と差分が無いか確認できるようにします。

余談ですが、上記の構成・設定をすることで、結果的に、HTTPエンドポイントとS3の両方にAWS WAFログを格納することが可能です。

API GatewayではLambdaプロキシ統合を使用します。統合リクエストでの変換は行わず、処理は基本Lambdaで書くことにします。検証なので、メソッドリクエストでの認可や検証も特に入れずに設定します。

Lambdaでは、受け取ったデータからAWS WAFのログ部分のみを抜き出して表示します。ランタイムはPython3.9です。

なお、LambdaにはKinesis Firehose用のブループリントや公式のサンプルコードも用意されていますが、今回はログの見え方を確認しながら少しずつコードをリファクタリングしてみたいので、あえて一からコードを書いてみます。

Kinesis Firehoseから配信したデータをLambdaで取り出す

とりあえず、Kinesis Firehoseから配信したAWS WAFのログを、実際にLambdaで取り出してみます。 Kinesis FirehoseからHTTPエンドポイントへ送付されるデータのフォーマットは、以下の公式ドキュメントに記載されています。

docs.aws.amazon.com

bodyの部分に配信データ（AWS WAFのログ）が格納されているようなので、まずはLambda関数に、イベントデータのbody部分をそのまま表示する処理と、body部分のデータ型を表示する処理を記述します。これでLambdaの実行ログにイベントデータとデータ型が表示されるので、公式ドキュメントのフォーマットと比較ができます。

import json

def lambda_handler(event, context):
    
    event_body = event['body']
    print (event_body)
    print (type(event_body))
    return {
        'statusCode': 200,
        'body': json.dumps(event)
    }

次に、Kinesis FirehoseからAWS WAFログを配信し、Lambdaの実行ログを見ると、以下のようになっていました。

{
    "requestId": "01591343-d026-436d-8f56-e08aeb9ad76a",
    "timestamp": 1666613584186,
    "records": [
        {
            "data": "eyJ0aW（中略）PSJ9fQo="
        },
        {
            "data": "eyJ0aW（中略）PSJ9fQo="
        }
    ]
}
<class 'str'>

公式ドキュメントのサンプルと同様の形式になっています。また、データ型は文字列型となっています。

data部分には、一見すると意味不明の、かなり長文の文字列が表示されました。これはバグが発生しているわけではなく、Kinesisの配信データがBASE64でエンコードされるためです。

以下は公式ドキュメントからの引用です。

The data of this record,in Base64. Note that empty records are permitted in Firehose. The maximum allowed size of the data, before Base64 encoding, is 1024000 bytes; the maximum length of this field is therefore 1365336 chars. （翻訳）このレコードのデータ (Base64)。空レコードも Firehose で許可されるので注意してください。許可される、Base64エンコード前の最大データサイズは 1,024,000バイトです。したがって、このフィールドの最大文字長は1,365,336文字です。

Appendix - HTTP Endpoint Delivery Request and Response Specifications - Amazon Kinesis Data Firehose

（Request Format > Body - Schema の data > description 参照）

そのため、この長文文字列の部分がAWS WAFのログであり、BASE64でデコードすれば中身が読めるはずです。

文字列を辞書型・リスト型に変換する

次はbodyからAWS WAFログの部分だけを取り出したいのですが、body全体が文字列のままだと扱いづらいので、Pythonオブジェクト（辞書型・リスト型）で解釈しなおしてみます。

import json

def lambda_handler(event, context):
    
    event_body = event['body']
    event_body_dict = json.loads(event_body)
    print (event_body_dict)
    print (type(event_body_dict))
    return {
        'statusCode': 200,
        'body': json.dumps(event)
    }

再度AWS WAFのログをKinesis Firehoseから配信し、Lambdaのログを見ると、今度はデータ型が辞書型になっていました。

{'requestId': '788cdd7a-a5fb-4862-8f6e-45cc7a0a04e3', 'timestamp': 1666616118467, 'records': [{'data': 'eyJ0a（中略）9fQo='}, {'data': 'eyJ0aW（中略）J9fQo='},{'data': 'eyJ0aW（中略）J9fQo='}]}
<class 'dict'>

AWS WAFログ部分を取り出し、BASE64デコードする

データ型を変換したことで、データの取り出しがしやすくなりました。ここで改めてPythonオブジェクトとしてbody全体を見てみると、下記の通り、「辞書①＞リスト＞辞書②」のネストされた形になっています。

{"requestId":"ID","timestamp":1234567890123,"records":[{"data":"AWS WAFログ"},{"data":"AWS WAFログ"}]}

辞書②のkeyは"data"で固定なので、あとは以下の処理でいけそうです。

辞書①のkey「record」を指定して、配列を取り出す
配列に対して繰り返し以下を実施
- 辞書②のkey「data」を指定して、AWS WAFログを取り出す
- AWS WAFログをBASE64デコードして表示

上記を踏まえ、Lambda関数を以下の通り更新しました。

import json
import base64

def lambda_handler(event, context):
    
    event_body = event['body']
    event_body_dict = json.loads(event_body)
    data_list = event_body_dict['records']
    
    for i in data_list:
        data64 = i['data']
        print(base64.b64decode(data64).decode())
    
    return {
        'statusCode': 200,
        'body': json.dumps(event)
    }

再度Lambddaのログを見てみます。

{
    "timestamp": 1666617261968,
    "formatVersion": 1,
    "webaclId": "arn:aws:wafv2:ap-northeast-1:XXXXXXXXXXXX:regional/webacl/kinesis-test/a113e553-49b2-4822-8a97-7d38223e17cb",
    "terminatingRuleId": "allow-from-my-pc",
    "terminatingRuleType": "REGULAR",
    "action": "ALLOW",
    "terminatingRuleMatchDetails": [],
    "httpSourceName": "APIGW",
    "httpSourceId": "XXXXXXXXXXXX:XXXXXXXXXX:dev",
    "ruleGroupList": [],
    "rateBasedRuleList": [],
    "nonTerminatingMatchingRules": [],
    "requestHeadersInserted": null,
    "responseCodeSent": null,
    "httpRequest": {
        "clientIp": "XXX.XXX.XXX.XXX",
        "country": "JP",
    （中略）
        "httpMethod": "GET",
        "requestId": "agtjNH1ctjMFawA="
    }
}

どうやらAWS WAFログが表示できたようです。

リストに複数のログがある場合でも、全て表示できているようです。