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

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

Inference hooksのAIセキュリティサーバーは、Anthropicが呼び出すHTTPSサービスです。統治対象の推論リクエストごとに、署名付きのPOSTで会話のトランスクリプトが届き、allowかdenyのverdictを返します。Inference hooksはClaude Enterprise組織向けのベータ機能です。管理画面からの有効化・接続手順は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・coworkがあり、接続テストとサーキットブレーカーの復旧チェックではconfig-testが使われます。これはクローズドな列挙ではなく、開放型の文字列として扱う必要があります。信頼境界として使うのは危険で、セキュリティ上重要な判断をこの値だけに委ねてはいけません(Claude Code側でWebhookを受信するサーバーを組む場合の考え方はChannelsでWebhook受信サーバーを自作するも参考になります)。

トランスクリプトは切り詰められずに送られるため、添付ファイルの多い長い会話はリクエストボディが大きくなります。実際にはモデルのコンテキストウィンドウが働いて10MB程度に収まりますが、プロトコル上の上限は64MiBです。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自体は有効です。

Anthropicが読むのは応答ボディの最大64KiBまでで、非圧縮である必要があります。リダイレクトは追跡されず、クッキーは無視されます。ここに書いた以外の未知のフィールドは無視されるので、追加のフィールドを含めても構いません。

トップレベルのtypeがprompt以外の値で届いた場合は、エラーステータスではなくallowを返します。エラーを返すとWebhook失敗になり、続けばサーキットブレーカーの作動要因になるためです。actor.typeの新しい値や未知のブロック種別も、読める項目だけを読んで先へ進めます。

署名を検証する

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

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  # リプレイか、時計がずれている
 
    try:
        key = base64.b64decode(secret.removeprefix("whsec_"), validate=True)
    except ValueError:
        return False  # シークレットの設定ミス。クラッシュさせず拒否する
 
    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が送るすべてのリクエストには署名が付き、接続テストも例外ではありません。管理画面の設定フローが最初の保存でシークレットを生成し、テストはその後に走るためです。署名の無いリクエストは拒否するのが基本です。

例外は、シークレットが必須になる前にInference hooksを有効にした組織です。管理者がシークレットを生成するまで無署名のリクエストが届き続けるので、管理者から生成の連絡を受けるまでは受け入れ、以降は拒否します。管理画面の接続テストが「Signing secret required」と返す場合は、Request signingのGenerate secretでシークレットを作ってから再テストします。

シークレットのローテーションは即座の切り替えですが、切り替え後もおよそ1分間は旧シークレットで署名されたリクエストが届き、すでに送信中のものも残ります。1つのリクエストが両方のシークレットで署名されることは無いので、AIセキュリティサーバー側は新旧どちらの鍵でも検証が通るようにしておきます。

検証関数を手元で試す

上のコードは、署名を自分で作って食わせれば送信元なしで試せます。次のスクリプトは、ランダムな鍵で署名したリクエストを作り、正しいもの・ボディを整形し直したもの・301秒前のタイムスタンプ・署名値が2つ並ぶもの・無署名のものを順に検証します。

python3 selftest.py

# selftest.py(冒頭に上の verify を置く)
import base64, hashlib, hmac, json, os, time
 
kb = os.urandom(24)
secret = "whsec_" + base64.b64encode(kb).decode()
body = json.dumps({"type": "prompt", "request_id": "req_test", "messages": []}).encode()
ts = str(int(time.time()))
payload = f"req_test.{ts}.".encode() + body
sig = "v1," + base64.b64encode(hmac.new(kb, payload, hashlib.sha256).digest()).decode()
h = {"Webhook-Id": "req_test", "Webhook-Timestamp": ts, "Webhook-Signature": sig}
 
print("valid      ", verify(secret, h, body))
print("reformatted", verify(secret, h, json.dumps(json.loads(body), indent=2).encode()))
print("301s old   ", verify(secret, {**h, "Webhook-Timestamp": str(int(ts) - 301)}, body))
print("two values ", verify(secret, {**h, "Webhook-Signature": "v1,AAAA " + sig}, body))
print("unsigned   ", verify(secret, {}, body))

Python 3.12.2で実行した出力は次のとおりです。ヘッダー名をWebhook-Idのように大文字で渡しても通るのは、検索を小文字に揃えているためです。

valid       True
reformatted False
301s old    False
two values  True
unsigned    False

reformattedだけがFalseになるのが、生のバイト列で計算する理由です。同じJSONでもインデントを変えれば別の署名になります。タイムスタンプの検査は301秒前でも落とすので、サーバーの時計がずれていると正規のリクエストまで全部弾きます。

タイムアウトとリトライ

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

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

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

AIセキュリティサーバー起因のWebhook失敗が続くとサーキットブレーカーが作動し、Anthropicはサーバーへの接続を止めてFailure handlingの設定を全リクエストへ適用します。Block the requestを選んでいる組織では、復旧までユーザーがブロックされ続けます。作動時は管理者へclaude.aiの通知センターで知らせが届き、Activity Feedにはinference_hooks_circuit_breaker_trippedが1回の作動につき1件記録されます。記録するには、組織でCompliance APIを有効にしておく必要があります。

復旧チェックの専用ロジックは不要で、通常のリクエストと同じ経路で応答すれば足ります。注意したいのは、サーバーを直した後に管理画面の設定をいじる場合です。作動後にInference hooksの設定を1つでも変えると、シークレットのローテーションを含めて自動チェックが止まります。その場合は管理者がEnforce verdictsをオンに戻してリセットします。サーバーを直した後であれば、設定を変えていなくても、同じ操作でいつでもリセットできます。

管理画面の統計の失敗件数にはネットワークやDNSのエラーも数えられますが、これらはブレーカーを作動させません。件数が多くても作動の表示が空のことがあるのはそのためです。

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

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

• webhook-idで重複排除する: webhook-idヘッダーは配信ごとに一意でボディのrequest_idと一致し、接続失敗によるリトライも同じ値を再利用します。verdictを記録するならこの値をキーにします
• verdictを記録しdenyを突き合わせられるようにする: 返したreference_idとともに各verdictを保存しておくと、Activity Feed(監査ログ)に記録されるinference_hooks_request_deniedの記録と自社側のログを突き合わせられます
• 常にallowを返すアーカイブサーバーとして使う: ポリシーを強制せずトランスクリプトをリアルタイムで残したいだけなら、無条件で{"action": "allow"}を返しつつ応答後に永続化する構成が、Compliance APIをポーリングする代わりのプッシュ型の選択肢になります。応答してから永続化する順序を守れば、往復時間にユーザーの待ち時間を巻き込みません
• 接続元を絞る: リクエストは160.79.106.0/24から届きます。Anthropicの公開アウトバウンドIP範囲の一部で、同じページのインバウンド範囲には含まれません。許可リストに入れるのはアウトバウンド側です。絞り込みは露出を減らす手段にとどまり、このブロックにはInference hooks以外のAnthropicの送信トラフィックも載るので、署名検証の代わりにはなりません
• 応答の速さを測る: 統治対象の全リクエストに、サーバーの往復時間がそのまま乗ります。大きな組織へ広げる前に負荷試験をかけます
• deny_reasonはエンドユーザー向けに書く: ここに書いた文字列(最大500字)がブロックされたユーザーにそのまま見えます。自社チームにしか分からないスキャナーのコードではなく、何を変えればよいかを書いてください

症状からたどる原因

まとめ

署名検証と、denyとWebhook失敗の区別の2点を押さえれば、サーバーの契約はほぼ満たせます。本番に出す前に、検証関数を自作の署名付きリクエストで試し、ボディ上限と時計のずれを確かめておくと、管理画面の接続テストで落ちる原因の大半を先に潰せます。