AWS WAFのログをAmazon Data Firehose経由でS3 Tablesに書き込んでみた

import json
import base64
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    # イベント全体をログに出力
    logger.info(f"Received event: {json.dumps(event)}")
    
    # 各レコードを処理
    if 'records' in event:
        for record in event['records']:
            # Base64でエンコードされたデータをデコード
            try:
                data = base64.b64decode(record['data']).decode('utf-8')
                logger.info(f"Record ID: {record['recordId']}, Data: {data}")
            except Exception as e:
                logger.error(f"Error decoding: {str(e)}")
    
    # Firehoseの場合は元のデータをそのまま返す
    if 'records' in event:
        return {
            'records': [{
                'recordId': record['recordId'],
                'result': 'Ok',
                'data': record['data']
            } for record in event['records']]
        }
    
    return {
        'statusCode': 200,
        'body': json.dumps('Processed Firehose records')
    }

{
    "invocationId": "abcdef12-3456-7890-abcd-1234567890ab",
    "deliveryStreamArn": "arn:aws:firehose:ap-northeast-1:123456789012:deliverystream/aws-waf-logs-example-stream",
    "region": "ap-northeast-1",
    "records": [
        {
            "recordId": "shardId-000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
            "approximateArrivalTimestamp": 1742806129515,
            "data": "eyJ0aW1lc3RhbXAiOjE3NDI4MDYxMDk2MjEsImZvcm1hdFZlcnNpb24iOjEsIndlYmFjbElkIjoiYXJuOmF3czp3YWZ2MjphcC1ub3J0aGVhc3QtMToxMjM0NTY3ODkwMTI6cmVnaW9uYWwvd2ViYWNsL0NyZWF0ZWRCeUFMQi13YWZsb2ctZXhhbXBsZS9hYmNkZWYxMi0zNDU2LTc4OTAtYWJjZC0xMjM0NTY3ODkwYWIiLCJ0ZXJtaW5hdGluZ1J1bGVJZCI6IkRlZmF1bHRfQWN0aW9uIiwidGVybWluYXRpbmdSdWxlVHlwZSI6IlJFR1VMQVIiLCJhY3Rpb24iOiJBTExPVyIsInRlcm1pbmF0aW5nUnVsZU1hdGNoRGV0YWlscyI6W10sImh0dHBTb3VyY2VOYW1lIjoiQUxCIiwiaHR0cFNvdXJjZUlkIjoiMTIzNDU2Nzg5MDEyLWFwcC93YWZsb2ctZXhhbXBsZS9hYmNkZWYxMjM0NTYiLCJydWxlR3JvdXBMaXN0IjpbeyJydWxlR3JvdXBJZCI6IkFXUyNBV1NNYW5hZ2VkUnVsZXNBbWF6b25JcFJlcHV0YXRpb25MaXN0IiwidGVybWluYXRpbmdSdWxlIjpudWxsLCJub25UZXJtaW5hdGluZ01hdGNoaW5nUnVsZXMiOltdLCJleGNsdWRlZFJ1bGVzIjpudWxsLCJjdXN0b21lckNvbmZpZyI6bnVsbH0seyJydWxlR3JvdXBJZCI6IkFXUyNBV1NNYW5hZ2VkUnVsZXNDb21tb25SdWxlU2V0IiwidGVybWluYXRpbmdSdWxlIjpudWxsLCJub25UZXJtaW5hdGluZ01hdGNoaW5nUnVsZXMiOltdLCJleGNsdWRlZFJ1bGVzIjpudWxsLCJjdXN0b21lckNvbmZpZyI6bnVsbH0seyJydWxlR3JvdXBJZCI6IkFXUyNBV1NNYW5hZ2VkUnVsZXNLbm93bkJhZElucHV0c1J1bGVTZXQiLCJ0ZXJtaW5hdGluZ1J1bGUiOm51bGwsIm5vblRlcm1pbmF0aW5nTWF0Y2hpbmdSdWxlcyI6W10sImV4Y2x1ZGVkUnVsZXMiOm51bGwsImN1c3RvbWVyQ29uZmlnIjpudWxsfV0sInJhdGVCYXNlZFJ1bGVMaXN0IjpbXSwibm9uVGVybWluYXRpbmdNYXRjaGluZ1J1bGVzIjpbXSwicmVxdWVzdEhlYWRlcnNJbnNlcnRlZCI6bnVsbCwicmVzcG9uc2VDb2RlU2VudCI6bnVsbCwiaHR0cFJlcXVlc3QiOnsiY2xpZW50SXAiOiIxMjMuNDU2Ljc4OS4xMCIsImNvdW50cnkiOiJKUCIsImhlYWRlcnMiOlt7Im5hbWUiOiJIb3N0IiwidmFsdWUiOiJleGFtcGxlLWFwcC5hcC1ub3J0aGVhc3QtMS5lbGIuYW1hem9uYXdzLmNvbSJ9LHsibmFtZSI6IkNvbm5lY3Rpb24iLCJ2YWx1ZSI6ImtlZXAtYWxpdmUifSx7Im5hbWUiOiJQcmFnbWEiLCJ2YWx1ZSI6Im5vLWNhY2hlIn0seyJuYW1lIjoiQ2FjaGUtQ29udHJvbCIsInZhbHVlIjoibm8tY2FjaGUifSx7Im5hbWUiOiJVc2VyLUFnZW50IiwidmFsdWUiOiJNb3ppbGxhLzUuMCAoTWFjaW50b3NoOyBJbnRlbCBNYWMgT1MgWCAxMF8xNV83KSBBcHBsZVdlYktpdC81MzcuMzYgKEtIVE1MLCBsaWtlIEdlY2tvKSBDaHJvbWUvMTM0LjAuMC4wIFNhZmFyaS81MzcuMzYifSx7Im5hbWUiOiJBY2NlcHQiLCJ2YWx1ZSI6ImltYWdlL2F2aWYsaW1hZ2Uvd2VicCxpbWFnZS9hcG5nLGltYWdlL3N2Zyt4bWwsaW1hZ2UvKiwqLyo7cT0wLjgifSx7Im5hbWUiOiJSZWZlcmVyIiwidmFsdWUiOiJodHRwOi8vZXhhbXBsZS1hcHAuYXAtbm9ydGhlYXN0LTEuZWxiLmFtYXpvbmF3cy5jb20vIn0seyJuYW1lIjoiQWNjZXB0LUVuY29kaW5nIiwidmFsdWUiOiJnemlwLCBkZWZsYXRlIn0seyJuYW1lIjoiQWNjZXB0LUxhbmd1YWdlIiwidmFsdWUiOiJqYSxlbi1VUztxPTAuOSxlbjtxPTAuOCJ9XSwidXJpIjoiL2Zhdmljb24uaWNvIiwiYXJncyI6IiIsImh0dHBWZXJzaW9uIjoiSFRUUC8xLjEiLCJodHRwTWV0aG9kIjoiR0VUIiwicmVxdWVzdElkIjoiMS1hYmNkZWYxMi0xMjM0NTY3ODkwYWJjZGVmMTIzNDU2NzgiLCJmcmFnbWVudCI6IiIsInNjaGVtZSI6Imh0dHAiLCJob3N0IjoiZXhhbXBsZS1hcHAuYXAtbm9ydGhlYXN0LTEuZWxiLmFtYXpvbmF3cy5jb20ifX0K"
        }
    ]
}

{
  "timestamp": 1742806109621,
  "formatVersion": 1,
  "webaclId": "arn:aws:wafv2:ap-northeast-1:123456789012:regional/webacl/CreatedByALB-waflog-example/abcdef12-3456-7890-abcd-1234567890ab",
  "terminatingRuleId": "Default_Action",
  "terminatingRuleType": "REGULAR",
  "action": "ALLOW",
  "terminatingRuleMatchDetails": [],
  "httpSourceName": "ALB",
  "httpSourceId": "123456789012-app/waflog-example/abcdef123456",
  "ruleGroupList": [
    {
      "ruleGroupId": "AWS#AWSManagedRulesAmazonIpReputationList",
      "terminatingRule": null,
      "nonTerminatingMatchingRules": [],
      "excludedRules": null,
      "customerConfig": null
    },
    {
      "ruleGroupId": "AWS#AWSManagedRulesCommonRuleSet",
      "terminatingRule": null,
      "nonTerminatingMatchingRules": [],
      "excludedRules": null,
      "customerConfig": null
    },
    {
      "ruleGroupId": "AWS#AWSManagedRulesKnownBadInputsRuleSet",
      "terminatingRule": null,
      "nonTerminatingMatchingRules": [],
      "excludedRules": null,
      "customerConfig": null
    }
  ],
  "rateBasedRuleList": [],
  "nonTerminatingMatchingRules": [],
  "requestHeadersInserted": null,
  "responseCodeSent": null,
  "httpRequest": {
    "clientIp": "123.456.789.10",
    "country": "JP",
    "headers": [
      {
        "name": "Host",
        "value": "example-app.ap-northeast-1.elb.amazonaws.com"
      },
      {
        "name": "Connection",
        "value": "keep-alive"
      },
      {
        "name": "Pragma",
        "value": "no-cache"
      },
      {
        "name": "Cache-Control",
        "value": "no-cache"
      },
      {
        "name": "User-Agent",
        "value": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36"
      },
      {
        "name": "Accept",
        "value": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8"
      },
      {
        "name": "Referer",
        "value": "http://example-app.ap-northeast-1.elb.amazonaws.com/"
      },
      {
        "name": "Accept-Encoding",
        "value": "gzip, deflate"
      },
      {
        "name": "Accept-Language",
        "value": "ja,en-US;q=0.9,en;q=0.8"
      }
    ],
    "uri": "/favicon.ico",
    "args": "",
    "httpVersion": "HTTP/1.1",
    "httpMethod": "GET",
    "requestId": "1-abcdef12-1234567890abcdef12345678",
    "fragment": "",
    "scheme": "http",
    "host": "example-app.ap-northeast-1.elb.amazonaws.com"
  }
}

echo "＜base64エンコードされた文字列＞" | base64 -d | jq

CREATE TABLE `my_s3_namespace`.`waflogs` (
  timestamp bigint,
  formatversion int,
  webaclid string,
  terminatingruleid string,
  terminatingruletype string,
  action string,
  httpsourcename string,
  httpsourceid string,
  httprequest struct<
    clientip: string,
    country: string,
    uri: string,
    args: string,
    httpversion: string,
    httpmethod: string,
    requestid: string,
    host: string,
    headers: array<struct<name: string, value: string>>,
    scheme: string,
    fragment: string
  >,
  rulegrouplist array<struct<
    rulegroupid: string,
    terminatingrule: string,
    nonterminatingmatchingrules: array<string>,
    excludedrules: string,
    customerconfig: string
  >>,
  terminatingrulematchdetails array<string>,
  ratebasedrulelist array<string>,
  nonterminatingmatchingrules array<string>,
  requestheadersinserted string,
  responsecodesent string,
  log_date date
)
PARTITIONED BY (month(`log_date`))
TBLPROPERTIES ('table_type' = 'iceberg')

import base64
import json
import logging
from datetime import datetime, timezone

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    output = []
    
    logger.info(f"Processing {len(event['records'])} records")
    
    for record in event['records']:
        try:
            # Base64でエンコードされたデータをデコード
            payload = base64.b64decode(record['data']).decode('utf-8').strip()
            
            # JSONとして解析
            json_data = json.loads(payload)
            
            # キーを小文字に変換する関数（再帰的に処理）
            def convert_keys_to_lowercase(obj):
                if isinstance(obj, dict):
                    return {k.lower(): convert_keys_to_lowercase(v) for k, v in obj.items()}
                elif isinstance(obj, list):
                    return [convert_keys_to_lowercase(item) for item in obj]
                else:
                    return obj
            
            # すべてのキーを小文字に変換
            result = convert_keys_to_lowercase(json_data)
            
            # タイムスタンプから日付フィールドを生成
            if 'timestamp' in result:
                timestamp_ms = result['timestamp']
                log_date = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc)
                result['log_date'] = log_date.strftime('%Y-%m-%d')
            
            # デバッグ出力
            logger.info(f"Top level keys after transformation: {list(result.keys())}")
            
            # 結果をJSON文字列に変換してBase64エンコード
            data_json = json.dumps(result) + '\n'
            encoded_data = base64.b64encode(data_json.encode('utf-8')).decode('utf-8')
            
            output_record = {
                'recordId': record['recordId'],
                'result': 'Ok',
                'data': encoded_data
            }
        except Exception as e:
            # エラーが発生した場合はログに記録し、元のデータを返す
            logger.error(f"Error processing record: {str(e)}")
            output_record = {
                'recordId': record['recordId'],
                'result': 'ProcessingFailed',
                'data': record['data']
            }
        
        output.append(output_record)
    
    logger.info(f"Successfully processed {len(output)} records")
    return {'records': output}

SELECT 
  timestamp,
  httprequest.uri,
  h.name,
  h.value
FROM my_s3_namespace.waflogs
CROSS JOIN UNNEST(httprequest.headers) AS t(h)
WHERE h.name = 'User-Agent'
  AND log_date BETWEEN CURRENT_DATE - INTERVAL '7' DAY AND CURRENT_DATE
LIMIT 100;

httprequest struct<
  clientip: string,
  country: string,
  uri: string,
  args: string,
  httpversion: string,
  httpmethod: string,
  requestid: string,
  host: string,
  headers: array<struct<name: string, value: string>>,
  scheme: string,
  fragment: string
>