Verify the VAPI webhook, so a forged transcript is refused

README.md

@@ -163,16 +163,39 @@ here.
 
 ```python
 from deeptrust.agents import DeepTrust
-from deeptrust.agents.vapi import Bridge
+from deeptrust.agents.vapi import Bridge, WebhookVerificationError
 
-bridge = Bridge(DeepTrust(), api_key=os.environ["VAPI_API_KEY"])
+bridge = Bridge(
+    DeepTrust(),
+    api_key=os.environ["VAPI_API_KEY"],
+    secret=os.environ["VAPI_WEBHOOK_SECRET"],   # see below, do not skip it
+)
 
 @app.post("/vapi/webhook")            # your route, on your server
-async def vapi_webhook(payload: dict):
-    await bridge.handle(payload, user=caller)
+async def vapi_webhook(request: Request, payload: dict):
+    try:
+        await bridge.handle(payload, user=caller, headers=request.headers)
+    except WebhookVerificationError:
+        raise HTTPException(status_code=401)
     return {}
 ```
 
+### Verify the webhook
+
+Your route is a public URL. Anyone who learns it can post a transcript that was
+never said, and it becomes a real call, a real analysis and a real finding in
+your organization. A forged `end-of-call-report` can also end a real call's
+session early.
+
+Set `server.secret` on the assistant, which is the Authorization section of its
+Webhook Server settings. VAPI sends it back in `X-Vapi-Secret` on every request.
+Pass the same value as `secret`, hand `handle` the request headers, and a
+request without it is refused before a single turn is recorded. The compare uses
+`hmac.compare_digest`.
+
+The bridge does not require it, so an existing integration keeps working, but a
+bridge with no `secret` trusts whatever arrives.
+
 VAPI is the mirror image of ElevenLabs. Nobody can hold a socket: VAPI posts
 its server-url events to *your* server, so the adapter is a handler you call
 from your own webhook route rather than a watcher with a loop of its own. Hand

src/deeptrust/agents/vapi.py

@@ -3,13 +3,27 @@
     from deeptrust.agents import DeepTrust
     from deeptrust.agents.vapi import Bridge
 
-    bridge = Bridge(DeepTrust(), api_key=os.environ["VAPI_API_KEY"])
+    bridge = Bridge(
+        DeepTrust(),
+        api_key=os.environ["VAPI_API_KEY"],
+        secret=os.environ["VAPI_WEBHOOK_SECRET"],   # do not skip it
+    )
 
     @app.post("/vapi/webhook")             # your route, your server
-    async def vapi_webhook(payload: dict):
-        await bridge.handle(payload, user=caller)
+    async def vapi_webhook(request: Request, payload: dict):
+        try:
+            await bridge.handle(payload, user=caller, headers=request.headers)
+        except WebhookVerificationError:
+            raise HTTPException(status_code=401)
         return {}
 
+Your route is a public URL, so verify what arrives. VAPI echoes the assistant's
+`server.secret` back in `X-Vapi-Secret` on every request; pass the headers and
+a request without it is refused before a turn is recorded. Without a `secret`
+the bridge trusts whatever arrives, which is enough for anyone who learns your
+URL to invent a call, or to end a real one early with a forged
+`end-of-call-report`.
+
 VAPI's transport is the mirror image of ElevenLabs'. There is no socket anyone
 can hold open: VAPI posts its server-url events to *your* server, and what goes
 back the other way goes to a per-call HTTPS endpoint VAPI mints for that call
@@ -56,13 +70,14 @@ async def vapi_webhook(payload: dict):
 
 from __future__ import annotations
 
-from collections.abc import Callable
+import hmac
+from collections.abc import Callable, Mapping
 from typing import Any
 from urllib.parse import quote, urlsplit
 
 import httpx
 
-from ..errors import ConfigError
+from ..errors import ConfigError, DeepTrustError
 from ..types import Analysis, Nudge, User
 from . import DeepTrust
 from ._session import Session
@@ -72,6 +87,35 @@ async def vapi_webhook(payload: dict):
 #: The only domain a control URL may point at.
 CONTROL_URL_DOMAIN = "vapi.ai"
 
+#: The header VAPI echoes ``server.secret`` back in.
+SECRET_HEADER = "x-vapi-secret"
+
+
+class WebhookVerificationError(DeepTrustError):
+    """The request did not carry the secret the bridge was configured with.
+
+    Raised by :meth:`Bridge.handle` before anything is recorded, so a forged
+    event cannot create a call, a turn or an analysis. Answer it with 401.
+    """
+
+
+def _read_header(headers: Mapping[str, str] | None, name: str) -> str:
+    """One header, case-insensitively, from whatever the framework hands over.
+
+    Starlette and aiohttp pass a case-insensitive mapping and a plain dict
+    works too, so the lookup cannot assume either.
+    """
+    if not headers:
+        return ""
+    direct = headers.get(name)
+    if direct is not None:
+        return str(direct)
+    lowered = name.lower()
+    for key, value in headers.items():
+        if str(key).lower() == lowered:
+            return str(value)
+    return ""
+
 
 def add_message_command(text: str) -> dict[str, Any]:
     """The control-URL body that delivers a nudge as an interrupt.
@@ -100,19 +144,29 @@ def __init__(
         dt: DeepTrust,
         *,
         api_key: str,
+        secret: str | None = None,
         deliver: bool = True,
         on_analysis: Callable[[Analysis], None] | None = None,
         base_url: str = API_BASE_URL,
     ) -> None:
         """`api_key` is a VAPI private key: it reads the call object to find
-        the control URL when an event does not carry one."""
+        the control URL when an event does not carry one.
+
+        `secret` is the assistant's `server.secret`, which VAPI echoes back in
+        `X-Vapi-Secret` on every request. Set it and pass the request headers
+        to `handle`, and a request without it is refused before anything is
+        recorded. Your route is a public URL: without this, anyone who learns
+        it can post a transcript that was never said and have it become a real
+        call, a real analysis and a real finding in your organization.
+        """
         if not api_key:
             raise ConfigError(
                 "Bridge needs a VAPI private API key. It reads the call to "
                 "find monitor.controlUrl, which is where a nudge is sent."
             )
         self._dt = dt
         self._key = api_key
+        self._secret = secret
         self._deliver = deliver
         self._on_analysis = on_analysis
         self._base_url = base_url
@@ -126,6 +180,7 @@ async def handle(
         payload: dict[str, Any],
         *,
         user: User | None = None,
+        headers: Mapping[str, str] | None = None,
     ) -> Analysis | None:
         """Process one server-url event. Returns the analysis it caused, if any.
 
@@ -134,7 +189,15 @@ async def handle(
         call object the control URL is learned from.
 
         Returns None for an event that started no job, which is most of them.
+
+        Raises `WebhookVerificationError` when a secret is configured and the
+        request did not carry it.
         """
+        if not self.verify(headers):
+            raise WebhookVerificationError(
+                "the request did not carry the VAPI secret. Set server.secret "
+                "on the assistant and pass the request headers to handle."
+            )
         message = _message(payload)
         call = message.get("call")
         call = call if isinstance(call, dict) else {}
@@ -175,6 +238,18 @@ async def handle(
                 await self.send_nudge(call_id, nudge)
         return result
 
+    def verify(self, headers: Mapping[str, str] | None) -> bool:
+        """Whether a request carries the configured secret.
+
+        True when no secret is configured, so an existing integration keeps
+        working; the README says what that costs. Compared with
+        `hmac.compare_digest`, so a caller cannot learn the secret one
+        character at a time from how long the refusal took.
+        """
+        if not self._secret:
+            return True
+        return hmac.compare_digest(_read_header(headers, SECRET_HEADER), self._secret)
+
     async def send_nudge(self, call_id: str, nudge: Nudge) -> bool:
         """Send one nudge into a live call. Returns whether VAPI took it.
 

tests/test_adapters.py

@@ -10,12 +10,19 @@
 from typing import Any
 
 import httpx
+import pytest
 import respx
 
 from deeptrust.agents import DeepTrust
 from deeptrust.agents.elevenlabs import Monitor, _read_turn, contextual_update_command
 from deeptrust.agents.livekit import attach
-from deeptrust.agents.vapi import Bridge, add_message_command
+from deeptrust.agents.vapi import (
+    SECRET_HEADER,
+    Bridge,
+    WebhookVerificationError,
+    _read_header,
+    add_message_command,
+)
 from deeptrust.agents.vapi import _read_turn as _vapi_read_turn
 
 BASE = "https://example.test/api/v1"
@@ -473,3 +480,68 @@ async def test_vapi_tool_calls_are_not_answered() -> None:
         assert await bridge.handle(event) is None
 
     assert analyze.call_count == 0
+
+
+# ── webhook verification ─────────────────────────────────────────────────────
+
+
+def _guarded_bridge(secret: str | None = "s3cret") -> Bridge:
+    return Bridge(
+        DeepTrust(api_key="dt_test", base_url=BASE),
+        api_key="vapi_test",
+        secret=secret,
+        deliver=False,
+    )
+
+
+@respx.mock
+async def test_vapi_refuses_a_request_without_the_secret() -> None:
+    analyze = respx.post(f"{BASE}/agents/analyze").mock(
+        return_value=httpx.Response(200, json=ONE_NUDGE)
+    )
+    bridge = _guarded_bridge()
+
+    with pytest.raises(WebhookVerificationError):
+        await bridge.handle(_transcript_event("user", "reset my password"), headers={})
+
+    assert not analyze.called
+    assert bridge.session("call_1") is None
+
+
+@respx.mock
+async def test_vapi_refuses_a_wrong_secret() -> None:
+    bridge = _guarded_bridge()
+
+    with pytest.raises(WebhookVerificationError):
+        await bridge.handle(
+            _transcript_event("user", "reset my password"),
+            headers={SECRET_HEADER: "nope"},
+        )
+
+
+@respx.mock
+async def test_vapi_accepts_the_right_secret() -> None:
+    respx.post(f"{BASE}/agents/analyze").mock(
+        return_value=httpx.Response(200, json=ONE_NUDGE)
+    )
+    bridge = _guarded_bridge()
+
+    result = await bridge.handle(
+        _transcript_event("user", "reset my password"),
+        headers={"X-Vapi-Secret": "s3cret"},
+    )
+
+    assert result is not None
+
+
+async def test_vapi_without_a_secret_keeps_working() -> None:
+    assert _guarded_bridge(secret=None).verify(None) is True
+
+
+async def test_vapi_reads_the_header_case_insensitively() -> None:
+    bridge = _guarded_bridge()
+
+    assert bridge.verify({"X-Vapi-Secret": "s3cret"}) is True
+    assert bridge.verify({"x-vapi-secret": "s3cret"}) is True
+    assert bridge.verify({}) is False
+    assert _read_header(None, SECRET_HEADER) == ""