Inference hooksのエンドポイントを自作する — verdict応答とWebhook署名検証

Inference hooksのエンドポイントとは何を作ればよいか

Inference hooksのAIセキュリティサーバーは、Anthropicが呼び出すHTTPSサービスです。統治対象の推論リクエストごとに、署名付きのPOSTで会話のトランスクリプトが届き、allowかdenyのverdictを返します。フィールド名・リクエスト形状・ヘッダーはベータ期間中に変わる可能性があると公式ドキュメントが明記しているので、本記事のコード例もその前提で読んでください。管理画面からの有効化・接続手順はShadow modeとサーキットブレーカーの設定手順、機能全体の概要はInference hooksとはにまとめています。

最小構成のサーバーを立てる

動く最小の実装は、リクエストを読んでallowを返すだけのサーバーです。公開のhttps:// URL(自分が管理するドメインのTLS終端リバースプロキシの背後など。ngrokのようなトンネルサービスは非対応)で待ち受け、管理者に接続先として登録してもらいます。

# Python標準ライブラリだけで動く最小サーバー
python server.py

# server.py — 常にallowを返す最小構成(署名検証なし)
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 
class VerdictHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
 
    def do_POST(self):
        # トランスクリプトは数MBになりうるので必ず読み切る
        self.rfile.read(int(self.headers.get("Content-Length", 0)))
        verdict = b'{"action": "allow"}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(verdict)))
        self.end_headers()
        self.wfile.write(verdict)
 
 
ThreadingHTTPServer(("", 8000), VerdictHandler).serve_forever()

このサーバーは署名の無いリクエストも含めて全部を受け付けます。強制を始める前に、必ず後述の署名検証を追加してください。

リクエストを受け取る

Anthropicは管理者が設定したURLへHTTPS POSTを送ります。固定のパスサフィックスは無く、設定したURL全体がエンドポイントです。ホストの要件は、ポート443のhttps://URL、公開ルーティング可能なホスト(プライベート・ループバック・キャリアグレードNATのアドレス範囲は接続時点で拒否)、パブリックCAトラストストアで検証できる証明書、リダイレクトなしで応答すること。リバーストンネルは非対応です。

すべてのリクエストには次の固定ヘッダーが付きます。

これに加えて管理者が設定したカスタムヘッダーと、組織が署名シークレットを持って以降はwebhook-*の署名ヘッダーが付きます。フックイベントはpromptフレームの1種類のみで、統治対象の推論リクエストごとに1回、推論が始まる前に送られます。AIセキュリティサーバーが応答するか判定タイムアウトが経過するまで、Anthropicはリクエストを保持します。

promptフレームの構造

リクエストボディは次のフィールドを持つJSONオブジェクトです。

messagesの各要素はrole(userまたはassistant。ツール結果もuserロール側に入る)と、typeで判別されるcontentブロックの配列を持ちます。ブロック種別はtext(テキスト)、tool_use(ツール名・引数・呼び出しID)、tool_result(ツールの出力テキスト・失敗フラグ・対応するtool_useのID)、attachment(ファイル名・メディアタイプ・サイズ・抽出済みテキスト)の4つです。トランスクリプトにはユーザーから見えている範囲だけが入り、システムプロンプト・ツール定義・Anthropic内部のコンテキスト・Claudeの隠れた思考・生のファイルバイト列は一切含まれません。全ブロックが除外されたターンはまるごと省かれるため、ユーザー・アシスタントの交互構造を前提にしないでください。

source.applicationは既知の値としてclaude-aiとclaude-codeがあり、接続テストではconfig-testが使われます。これはクローズドな列挙ではなく、開放型の文字列として扱う必要があります。信頼境界として使うのは危険で、セキュリティ上重要な判断をこの値だけに委ねてはいけません(Claude Code側でWebhookを受信するサーバーを組む場合の考え方はChannelsでWebhook受信サーバーを自作するも参考になります)。

トランスクリプトは切り詰められずに送られるため、添付ファイルの多い長い会話はリクエストボディが大きくなり、上限は10MBです。nginxのclient_max_body_size(既定1MB)やExpressのexpress.json()(既定100kB)はこれより小さいことが多く、その既定値のままだとボディが拒否されてWebhook失敗になります。Allow the request設定の組織では、拒否されたリクエストは検査なしでモデルへ届いてしまうので、ボディサイズの上限は必ず引き上げてください。

verdictを返す

allow・denyのどちらでもHTTP 200とJSONのverdictボディで応答します。actionフィールドが結果を決めます。

{
  "action": "allow"
}

{
  "action": "deny",
  "deny_reason": "This prompt appears to contain customer payment card data, which your organization's policy does not allow.",
  "reference_id": "scan_01HXPT4R9V"
}

書式の不備でdenyが握りつぶされることはありません。deny_reasonが長すぎれば切り詰められ、reference_idが不正な形式なら静かに落とされますが、action自体は尊重されます。逆にHTTP 200と解釈可能なverdict以外はすべてWebhook失敗として扱われ、組織のFailure handling設定に置き換わります。エラーステータスでdenyを表現することはできません(200以外は失敗であってdenyではない)。allow/deny以外のaction値もWebhook失敗です。Anthropicが読むのは応答ボディの最大64KiBまでで、非圧縮である必要があります。リダイレクトは追跡されず、クッキーは無視されます。ここに書いた以外の未知のフィールドは無視されるので、追加のフィールドを含めても構いません。

署名を検証する

リクエストはStandard Webhooks仕様に沿って3つのヘッダーで署名されます。ヘッダー名は小文字で送られますが、プロキシが大文字小文字を変える可能性があるため、大文字小文字を無視して検索してください。

検証で事故りやすいポイントは2つです。生のバイト列に対して計算すること(JSONのパースや再エンコード前のボディそのもの)。シークレットは標準のbase64デコーダーでデコードすること(whsec_プレフィックスの後ろの値を標準アルファベットの+と/込みでデコードする。URLセーフなデコーダーを使うと、シークレットに+か/が含まれるたびに違う鍵バイト列になる)。

import base64
import hashlib
import hmac
import time
 
TOLERANCE_SECONDS = 300
 
 
def verify(secret: str, headers: dict[str, str], body: bytes) -> bool:
    lowercased = {name.lower(): value for name, value in headers.items()}
    try:
        message_id = lowercased["webhook-id"]
        timestamp = lowercased["webhook-timestamp"]
        signatures = lowercased["webhook-signature"]
    except KeyError:
        return False  # 署名が無い = Anthropicからではない
 
    try:
        signed_at = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - signed_at) > TOLERANCE_SECONDS:
        return False  # リプレイか、時計がずれている
 
    key = base64.b64decode(secret.removeprefix("whsec_"), validate=True)
    payload = f"{message_id}.{timestamp}.".encode() + body
    expected = b"v1," + base64.b64encode(
        hmac.new(key, payload, hashlib.sha256).digest()
    )
 
    return any(
        hmac.compare_digest(expected, candidate.encode())
        for candidate in signatures.split()
    )

組織が署名シークレットを持って以降、Anthropicが送るすべてのリクエストには署名が付きます。したがって署名の無いリクエストは基本的に拒否すべきです。唯一の例外は、組織が初回保存する前に送られる接続テストで、これは署名シークレットがまだ存在しないため無署名で届きます。管理者がシークレットの存在を確認するまでは無署名リクエストを受け入れ、以降は拒否する実装にしてください。

シークレットのローテーションは即座の切り替えですが、切り替え後もおよそ1分間は旧シークレットで署名されたリクエストが届く可能性があり、すでに送信中のものも残ります。切り替え中にこれらのリクエストを弾かないよう、AIセキュリティサーバー側は新旧両方のシークレットからの署名を受け付けるようにしてください。

タイムアウトとリトライ

管理者が設定する判定タイムアウトは1〜10,000ms(既定5,000ms)で、接続・TLSハンドシェイク・リクエスト・応答までの往復全体をこの予算でカバーします。Anthropicは接続の試行が失敗したときに限り、100ms後に1回だけリトライします。リトライは同じタイムアウト予算を共有し、同じwebhook-idと同じ署名を使います。サーバーが一度応答した後は、そのやり取りが再送されることはありません。

タイムアウト・200以外のステータス(リダイレクト含む)・パースできない/大きすぎる応答ボディ・到達不能なエンドポイントは、すべてWebhook失敗として扱われます。Webhook失敗がdenyになることは無く、代わりに組織のFailure handling設定がリクエストをブロックするか通過させるかを決めます。

サーキットブレーカーとの付き合い方

AIセキュリティサーバー起因のWebhook失敗が続くとサーキットブレーカーが作動し、Anthropicはサーバーへの接続を止め、Failure handlingの設定を全リクエストへ適用します。作動から10分後を起点に、Anthropicは組織自身のトラフィックに乗せたテストリクエストをおよそ1分に1回サーバーへ届け、署名も形も通常のリクエストと同じです。これに通常どおり応答してください。allowかdenyの有効なverdictを返せばブレーカーはリセットされ、強制が再開します。Webhook失敗のままならブレーカーは作動を続けます。いずれの場合も、このテストリクエスト自体はそのユーザーに対して処理が進み、verdictは強制されず、失敗してもブロックされません(Block the request設定でも同様です)。

サーバー設計で決めておくべきこと

ワイヤープロトコル以外にも、実運用のサーバーはいくつかの設計判断が必要です。

• webhook-idで重複排除する: webhook-idヘッダーは配信ごとに一意でボディのrequest_idと一致し、接続失敗によるリトライも同じ値を再利用します。verdictを記録するならこの値をキーにします
• verdictを記録しdenyを突き合わせられるようにする: 返したreference_idとともに各verdictを保存しておくと、Activity Feed(監査ログ)に記録されるinference_hooks_request_deniedの記録と自社側のログを突き合わせられます
• 常にallowを返すアーカイブサーバーとして使う: ポリシーを強制せずトランスクリプトをリアルタイムで残したいだけなら、無条件で{"action": "allow"}を返しつつ応答後に永続化する構成が、Compliance APIをポーリングする代わりのプッシュ型の選択肢になります。応答してから永続化する順序を守れば、往復時間にユーザーの待ち時間を巻き込みません
• deny_reasonはエンドユーザー向けに書く: ここに書いた文字列(最大500字)がブロックされたユーザーにそのまま見えます。自社チームにしか分からないスキャナーのコードではなく、何を変えればよいかを書いてください

よくあるつまずき

• JSONにパースしてから署名検証してしまう: 署名は受信した生のバイト列に対して計算されているので、一度パース・再エンコードした文字列で検証すると一致しません
• URLセーフなbase64デコーダーを使ってしまう: シークレットに+や/が含まれる場合、標準アルファベットのデコーダーでないと違う鍵になります
• ボディサイズの上限を上げ忘れる: nginxやExpressの既定値は数百kB〜1MBで、10MBのトランスクリプトを普通に拒否します。拒否されたリクエストは検査なしで通ってしまう組織もあります
• 未知のtypeやブロックをエラーにする: 将来追加されるイベント種別やブロック種別を拒否すると、それ自体がWebhook失敗となりサーキットブレーカーの作動要因になります。知らないフィールドは無視して先へ進めます
• ブレーカーの自動復旧テストに気づかず放置する: テストリクエストは通常のリクエストと見分けが付かない形で届くため、専用の待受ロジックは不要ですが、サーバーを直した後に設定を一切変えていないかを確認してから復旧を待つ必要があります

まとめ

Inference hooksのAIセキュリティサーバーは、署名付きのpromptフレームを受け取り、HTTP 200とallow/denyのJSONで応答するだけの単純な契約です。実装で事故りやすいのは、生バイト列に対する署名検証と標準base64デコード、ボディサイズ上限の引き上げ、Webhook失敗とdenyを混同しないこと、そして未知のフィールドを無視する前方互換性の4点です。サーバーを立てたら、管理画面側の接続手順もあわせて確認してください。