Webhook が本物である保証はどこにもない — Gemini Webhooks 受信エンドポイントの三層防御設計

CREATE TABLE processed_events (
  dedupe_key  TEXT PRIMARY KEY,   -- イベントID（無ければ合成キー）
  received_at INTEGER NOT NULL    -- UNIX 秒
);

import hmac
import os
import sqlite3
import time
 
from fastapi import FastAPI, Header, HTTPException, Request
 
app = FastAPI()
 
HOOK_TOKEN = os.environ["HOOK_PATH_TOKEN"]      # URL パスに埋める乱数
HOOK_SECRET = os.environ["HOOK_SHARED_SECRET"]  # 通知設定側で付与するヘッダ値
DB_PATH = os.environ.get("HOOK_DB", "webhook.db")
 
 
def db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS processed_events ("
        " dedupe_key TEXT PRIMARY KEY,"
        " received_at INTEGER NOT NULL)"
    )
    conn.execute(
        "CREATE TABLE IF NOT EXISTS work_queue ("
        " id INTEGER PRIMARY KEY AUTOINCREMENT,"
        " operation_name TEXT NOT NULL,"
        " enqueued_at INTEGER NOT NULL,"
        " done INTEGER NOT NULL DEFAULT 0)"
    )
    return conn
 
 
@app.post("/hooks/gemini/{token}")
async def receive(token: str, request: Request,
                  x_hook_secret: str = Header(default="")):
    # 第一層: 到達の検証（パス乱数 + シークレットヘッダを定数時間比較）
    if not hmac.compare_digest(token, HOOK_TOKEN):
        raise HTTPException(status_code=404)
    if not hmac.compare_digest(x_hook_secret, HOOK_SECRET):
        raise HTTPException(status_code=401)
 
    payload = await request.json()
    # ペイロードの形は通知の種類・設定で変わるため、実環境の実物に合わせます
    operation = payload.get("operation", {}).get("name") or payload.get("name", "")
    if not operation:
        # 形が想定外でも 200 で受領し、再送ループを作らない
        return {"status": "ignored"}
    event_id = payload.get("event_id") or f"{operation}:{payload.get('state', '')}"
 
    conn = db()
    with conn:
        cur = conn.execute(
            "INSERT OR IGNORE INTO processed_events (dedupe_key, received_at)"
            " VALUES (?, ?)",
            (event_id, int(time.time())),
        )
        if cur.rowcount == 0:
            return {"status": "duplicate"}  # 第二層: リプレイ・再送はここで畳む
        conn.execute(
            "INSERT INTO work_queue (operation_name, enqueued_at) VALUES (?, ?)",
            (operation, int(time.time())),
        )
    # 第三層: 重い処理はここでやらない。受領だけ返す
    return {"status": "accepted"}

import sqlite3
 
from google import genai
 
client = genai.Client()  # GEMINI_API_KEY は環境変数から
 
 
def fetch_authoritative_state(operation_name: str):
    """ペイロードではなく API を事実として扱う: 状態を取得し直す"""
    return client.operations.get(name=operation_name)
 
 
def run_worker(db_path: str = "webhook.db") -> None:
    conn = sqlite3.connect(db_path)
    rows = conn.execute(
        "SELECT id, operation_name FROM work_queue"
        " WHERE done = 0 ORDER BY id LIMIT 20"
    ).fetchall()
    for row_id, operation_name in rows:
        op = fetch_authoritative_state(operation_name)
        if not getattr(op, "done", False):
            # 偽の「完了」通知はここで実害を失う: 事実がまだなら何もしない
            continue
        handle_completed(op)  # 後段処理（結果の取得・保存など）
        with conn:
            conn.execute("UPDATE work_queue SET done = 1 WHERE id = ?", (row_id,))
 
 
def handle_completed(operation) -> None:
    # 結果の取り出しと保存。ここも冪等に書いておくと再実行に強くなります
    ...