ようやく自分のいる課でも GitHub 使うようになりましたので、タイトルの通り AWS で GitHub Webhook を受け取って、ごにょごにょするための基盤を作りました。その時の技術メモになります。

完成した物は GitHub でも公開しています。

アーキテクチャ図

ビューアーリクエストの Lambda@Edge で Webhook のペイロードを検証、API Gateway から呼び出される Lambda でペイロードの JSON へ HTTP ヘッダ情報を付与、SNS で Webhook ペイロードを固有の処理へファンアウト、とシンプルな作りになっています。

固有の処理ですが、現在、えらい人が有料プランの契約を検討されているところで Organization での Public レポジトリ作成を禁止できないため、それまでの繋ぎとして、レポジトリが Public になった時に SNS 経由でメール通知する機能と、レポジトリ操作を長期ログとして残すために Firehose で S3 へ保存する機能を用意してます。なにか処理が欲しくなったら、ファンアウト用 SNS にはやす形で Lambda とか増やせばオッケーです。

Lambda@Edge の部分

自身が設定したリクエストのみを受けいれできるようには、下記公式ドキュメントの通り、ハッシュ値の検証が必要となります。

最初は API Gateway の Lambda Authorizer で Webhook の検証ができないかと思いましたが、そこだと HTTP リクエストはヘッダ情報のみ参照可能なので、検証できる場所は

API Gateway 呼び出し先の Lambda
CloudFront を挟んで Lambda@Edge

ぐらいになりそうです。

API Gateway 呼び出し先の Lambda で検証が楽で安上がりですが、検証機能は外に出したく、コスト微増くらいで使い慣れてない技術を使える機会だったので、あえて 2. の方法を使ってます。

Lambda@Edge では環境変数が使えないため、シークレットトークンと後述する API キーは SecureString タイプの Parameter Store へ保存しています。CFn では SecureString タイプのそれを作成できないので、全リソースを一撃作成できないのはやむなしです。

ちなみにシークレットトークンも Python で生成したいときのワンライナーは以下なります。

python -c 'import secrets; print(secrets.token_hex(64))'

検証用のソースコードはこんな感じ。


import hashlib
import hmac
from base64 import b64decode

import boto3


_SSM_AWS_REGION = "us-east-1"
_SSM_PARAM_NAME_GITHUB_WEBHOOK_SECRET_TOKEN = "/github/webhook/secret-token/default"
_SSM_PARAM_NAME_GITHUB_WEBHOOK_APIGW_APIKEY = "/github/webhook/apigw-apikey/default"

_ssm_client = boto3.client("ssm", region_name=_SSM_AWS_REGION)
_GITHUB_WEBHOOK_SECRET_TOKEN = _ssm_client.get_parameter(Name=_SSM_PARAM_NAME_GITHUB_WEBHOOK_SECRET_TOKEN, WithDecryption=True)["Parameter"]["Value"].encode("utf-8")  # pyright: ignore[reportTypedDictNotRequiredAccess]
_GITHUB_WEBHOOK_APIGW_APIKEY = _ssm_client.get_parameter(Name=_SSM_PARAM_NAME_GITHUB_WEBHOOK_APIGW_APIKEY, WithDecryption=True)["Parameter"]["Value"]  # pyright: ignore[reportTypedDictNotRequiredAccess]
_CF_HEADER_X_API_KEY = [{"key": "X-API-Key", "value": _GITHUB_WEBHOOK_APIGW_APIKEY}]


def verify_signature(payload_body: bytes, secret_token: bytes, signature_header: str) -> bool:

    hash_object = hmac.new(secret_token, msg=payload_body, digestmod=hashlib.sha256)
    expected_signature = "sha256=" + hash_object.hexdigest()

    return hmac.compare_digest(expected_signature, signature_header)


def lambda_handler(event, context):

    request = event["Records"][0]["cf"]["request"]
    header_signature = request["headers"].get("x-hub-signature-256")

    if header_signature and request["method"] == "POST" and \
            verify_signature(b64decode(request["body"]["data"]), _GITHUB_WEBHOOK_SECRET_TOKEN, header_signature[0]["value"]):

        request["headers"]["x-api-key"] = _CF_HEADER_X_API_KEY
        return request

    return {"status": "403"}

CloudFront と API Gateway 間のアクセス制限

あたりまえですが、CloudFront 側でペイロードの検証をしたものの、API Gateway 側でも制限をかけないと、なんでもリクエストを受け入れてしまいます。ので、何らかの方法でこの CloudFront からのアクセスのみを API Gateway 側で許可する必要もあります。これは API Gateway の API キーと、リソースポリシーの Referer で実装しています。

Referer での制限は CloudFront 側でシークレットな値 (もちろん GitHub のシークレットとは別) をセットした Referer ヘッダーを追加して、API Gateway ではリソースポリシー等で共用したシークレット値をもつ Referer をセットされたリクエストのみの Action を許可する昔ながらの方法です。

CFn 経由でリソースポリシーを書く際、リソース作成前の API Gateway ID をどうすれば入力できるかなのですが、Resource で "execute-api:なんか" と書くと、execute-api: の部分が REST API のリソースを指し示す ARN へ自動的に展開されるようです。

以下、CFn テンプレートのリソースポリシーを記述している部分になります。

ResourcePolicy:
　CustomStatements: [
　  {
　    "Effect": "Deny",
　    "Principal": "*",
　    "Action": "execute-api:Invoke",
　    "Resource": "execute-api:/*/*/*",
　    "Condition": {
　      "StringNotEquals": {
　        "aws:Referer": !Ref RefererValue
　      }
　    }
　  },
　  {
　    "Effect": "Allow",
　    "Principal": "*",
　    "Action": "execute-api:Invoke",
　    "Resource": "execute-api:/*/*/*",
　    "Condition": {
　      "StringEquals": {
　        "aws:Referer": !Ref RefererValue
　      }
　    }
　  }
　]

と書くと

になります。

API Gateway ではなく、セキュリティグループをもつオリジンへ送信するなら、CloudFront のマネージドプリフィックスリストも使えます。

API キーでの制限ですが、メソッドリクエストで API キーの必要性を true にすることで、使用量プランと関連付けられた API キーをもつリクエストのみ受け入れる、簡易的なアクセス制限に使えます。

ただし、公式ドキュメントには「Don't use API keys for authentication or authorization for your APIs.」と、認証や認可目的の用途は想定してない書き方をしているので、アクセス制御はおまけ程度の機能と考えて、プロダクション用途では本来の用途である使用量プランのために使いましょう。

API Gateway 後ろにある Lambda の部分

当初は API Gateway がら SNS を直接呼ぶ形にしていましたが、後続の処理がイベント種別を判断するため X-GitHub-Event の値が必要そうなので、一旦 Lamdab を挟んで、GitHub Webhook に関連する HTTP ヘッダ情報をペイロードの JSON に追加しています。

ソースコードです。


import json
from os import environ

import boto3


_SNS_TOPIC_ARN = environ["SNS_TOPIC_ARN"]
_sns_client = boto3.client("sns")


def lambda_handler(event, context):

    payload = json.loads(event["body"])
    payload["headers"] = {
        k: v
        for k, v in event["headers"].items()
        if k.startswith("X-GitHub-") or k.startswith("X-Hub-Signature")
    }

    _sns_client.publish(
        Message=json.dumps(payload, ensure_ascii=False, separators=(",", ":"), indent=None),
        TopicArn=_SNS_TOPIC_ARN,
    )

    return {
        "statusCode": 200,
        "body": "{}"
    }

ついでに SNS Subscription Filter を容易に利用できるよう、Publish 時に必要そうな属性を追加してもいいと思います。脳死でヘッダ丸ごとコピーするとアクセス制限用の情報までくっついてしまいますので注意しましょう。

SNS -> KDF の部分

SNS から KDF へ直接配信できるようになって、とりあえずメッセージを S3 へ残したいということを簡単に実現できるようになって便利になりました。Raw メッセージ配信を有効化すると message の値だけが KDF へ送信されるので、Lambda で SNS のメタ情報を取り除く処理すら不要です。

ノーコードで対応できる点は非常に便利なのですが、Direct PUT のデータ処理量カウントは 5KB 切り上げ故、5KB 未満のペイロードを都度都度送信していると思った以上のコストになる......なんてこともあります。

そうなると、バッチサイズをある程度のメッセージをまとめて KDF へ送信する SQS + Lambda を挟んだ方が安上がりになるかもしれません。そこまでするなら安定性を捨てて、Lambda から Firehose 返さず S3 へ保存すればなお安上がりな気もします。要検討です。

KDF の Lambda

approximateArrivalTimestamp でパーティションキーを計算しているだけなので、KDF 受信日時だけでパーティションを切るとはっきり決めているなら不要です。動的パーティションを無効化して配信ストリームを作成すると、それが必要になった際にリソースの再作成が必要になり、状況次第ではかなり面倒くさいことになるので、要件が不明瞭な場合、とりあえずこの Python コードを挟んでいます。

ペイロードに改行がついてなく、jsonl 形式で保存したい場合は

レコード変換用 Lambda で \n を付与
動的パーティショニングで改行の区切り文字を有効

のどちらかで対応することになります。「動的パーティショニングで改行の区切り文字を有効」の方法だと

↑な感じで、エラー出力時に元々つけられる改行へ更に改行が追加されて個人的にチョットキモチワルイので、まわりくどいですがレコード変換 Lambda 側で改行を追加してます。

ソースコードになります。


from base64 import b64decode, b64encode
from datetime import datetime, timezone
from json import loads, dumps


def _conv(record: dict) -> dict:

    payload = loads(b64decode(record["data"]))

    # Do custom processing on the payload here
    timestamp = datetime.fromtimestamp(record["approximateArrivalTimestamp"] / 1000, timezone.utc)

    return {
        "recordId": record["recordId"],
        "result": "Ok",
        "data": b64encode(dumps(payload, ensure_ascii=False, separators=(",", ":"), indent=None).encode("utf-8") + b"\n").decode("utf-8"),
        "metadata": {
            "partitionKeys": {
                "year": timestamp.strftime("%Y"),
                "month": timestamp.strftime("%m"),
                "day": timestamp.strftime("%d"),
                "hour": timestamp.strftime("%H"),
            },
        },
    }


def lambda_handler(event, context) -> dict[str, list[dict]]:
    return {"records": [_conv(record) for record in event["records"]]}

随分昔の話ですが、Python の辞書型はキーの順序を保持するようになったので、GitHub から送信されたペイロードのキー順序は S3 保存時も同じになってます。追加したヘッダーを利用すれば、改ざんチェックみたいなことができそうです。それが特に不要なら保存コスト削減のため、 URL 関連の情報を取り除くのもありです。

レポジトリの Public 化通知 Lambda

イベント元が repository で、publicized か created アクション時にレポジトリが非 Private だったら SNS で通知するだけの Lambda です。

一応これもソースコード貼り付けます。


import json
from os import environ

import boto3


_SNS_TOPIC_ARN = environ["SNS_TOPIC_ARN"]
_sns_client = boto3.client("sns")


def lambda_handler(event, context):

    message = event["Records"][0]["Sns"]["Message"]
    payload = json.loads(message)

    if (payload["headers"]["X-GitHub-Event"] == "repository" and
            payload["action"] in {"publicized", "created", } and
            payload.get("repository", {}).get("private") != True):

        _sns_client.publish(
            Message=message,
            TopicArn=_SNS_TOPIC_ARN,
            Subject=f"GitHub Publicized Repository Notification - {payload['repository']['full_name']}",
        )

こういうちっさいファンクションを積み重ねた後、一気に通して動くのを見ると、ザ・メイキングで生産ラインをみている気分になって私は好きです。AWS バックエンド開発の醍醐味だと思ってます。