feat: sign and deliver MCP events safely

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "uvicorn[standard]>=0.30.0",
     "httpx>=0.27.0",
     "cryptography>=44.0.0",
+    "httpcore>=1.0",
     "mcp-event-subscriptions>=0.1.0a1,<0.2",
     "pyjwt[crypto]>=2.9.0",
     "numpy>=2.0",

src/mcp_server_appwrite/events/delivery.py

@@ -0,0 +1,330 @@
+"""Sign and deliver MCP events to webhook callbacks.
+
+MCP webhook delivery is a profile of Standard Webhooks
+(https://github.com/standard-webhooks/standard-webhooks/blob/main/spec/standard-webhooks.md):
+every POST carries ``webhook-id``, ``webhook-timestamp`` (Unix seconds) and
+``webhook-signature: v1,<base64 HMAC-SHA256(key, "{id}.{timestamp}.{body}")>``,
+where ``key`` is the base64-decoded part of the client's ``whsec_`` secret.
+A subscription has one secret, so the header carries one signature. MCP adds
+``X-MCP-Subscription-Id``.
+
+Two kinds of POST go out through :class:`Dispatcher`, both over the SSRF-safe
+:mod:`egress` path:
+
+* a **verification handshake** before a callback is used: a signed
+  ``{"type": "verification", "challenge": ...}`` body that the receiver must
+  echo in a 2xx JSON response, compared in constant time;
+* **event deliveries**: one event per POST, ``{eventId, name, timestamp, data,
+  cursor: null}`` serialized once, at most 256 KiB. Failed attempts retry with
+  bounded, jittered exponential backoff and are re-signed with a fresh
+  timestamp each time; ``410`` and ``413`` are final.
+
+Retries run in-process and nothing is persisted: Appwrite webhooks have no
+replay, so a delivery that exhausts its retries is abandoned (``cursor`` stays
+``null``). Failures are reported only as the spec's categories, never as raw
+endpoint responses, so a callback URL cannot be used as a response oracle.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import hmac
+import json
+import random
+import secrets
+import ssl
+import time
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass, field
+from enum import StrEnum
+from functools import partial
+from typing import Any
+
+import anyio
+import httpx
+
+from .egress import DestinationError, Egress
+
+SECRET_PREFIX = "whsec_"
+SECRET_MIN_BYTES = 24
+SECRET_MAX_BYTES = 64
+SIGNATURE_VERSION = "v1"
+
+ID_HEADER = "webhook-id"
+TIMESTAMP_HEADER = "webhook-timestamp"
+SIGNATURE_HEADER = "webhook-signature"
+SUBSCRIPTION_HEADER = "X-MCP-Subscription-Id"
+CONTENT_TYPE = "application/json"
+
+VERIFICATION_TYPE = "verification"
+VERIFICATION_ID_PREFIX = "msg_verification_"
+CHALLENGE_BYTES = 32
+
+# Delivery profile: receivers and intermediaries may reject larger bodies.
+MAX_BODY_BYTES = 256 * 1024
+
+# Receivers return 410 to refuse an event for good and 413 for a body they
+# will never accept; retrying either cannot succeed.
+FINAL_STATUSES = frozenset({410, 413})
+
+
+class Reason(StrEnum):
+    """Failure categories shared by ``data.reason`` and ``lastError``."""
+
+    CHALLENGE_FAILED = "challenge_failed"
+    TIMEOUT = "timeout"
+    CONNECTION_REFUSED = "connection_refused"
+    TLS_ERROR = "tls_error"
+    HTTP_4XX = "http_4xx"
+    HTTP_5XX = "http_5xx"
+
+
+class CallbackError(Exception):
+    """The callback endpoint failed verification; ``reason`` is a :class:`Reason`."""
+
+    def __init__(self, reason: str) -> None:
+        super().__init__(f"Callback endpoint verification failed: {reason}")
+        self.reason = reason
+
+
+class InvalidSecretError(ValueError):
+    """A delivery secret is not ``whsec_`` + base64 of 24-64 bytes."""
+
+
+class PayloadTooLargeError(ValueError):
+    """An event body exceeds :data:`MAX_BODY_BYTES` and was not sent."""
+
+
+class Outcome(StrEnum):
+    DELIVERED = "delivered"
+    # The receiver (410/413) or the egress policy refused the event; not retried.
+    REJECTED = "rejected"
+    # Every attempt failed with a retryable error.
+    ABANDONED = "abandoned"
+
+
+def decode_secret(secret: str) -> bytes:
+    """Validate a ``whsec_`` secret and return its HMAC key bytes.
+
+    The subscribe handler calls this to reject bad ``delivery.secret`` values
+    with ``InvalidParams`` before anything is stored or sent.
+    """
+    if not isinstance(secret, str) or not secret.startswith(SECRET_PREFIX):
+        raise InvalidSecretError(f"Secret must start with {SECRET_PREFIX}")
+    encoded = secret.removeprefix(SECRET_PREFIX)
+    try:
+        key = base64.b64decode(encoded, validate=True)
+    except (binascii.Error, ValueError) as error:
+        raise InvalidSecretError("Secret must be base64 after the prefix") from error
+    if not SECRET_MIN_BYTES <= len(key) <= SECRET_MAX_BYTES:
+        raise InvalidSecretError(
+            f"Secret must decode to {SECRET_MIN_BYTES}-{SECRET_MAX_BYTES} bytes"
+        )
+    return key
+
+
+def sign(key: bytes, message_id: str, timestamp: int, body: bytes) -> str:
+    """Standard Webhooks ``webhook-signature`` value."""
+    content = f"{message_id}.{timestamp}.".encode() + body
+    digest = hmac.new(key, content, hashlib.sha256).digest()
+    return f"{SIGNATURE_VERSION},{base64.b64encode(digest).decode()}"
+
+
+def encode(payload: Mapping[str, Any]) -> bytes:
+    """Compact JSON, serialized once so the signed bytes are the sent bytes."""
+    return json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode()
+
+
+@dataclass(frozen=True)
+class Callback:
+    """Where and how to deliver for one subscription."""
+
+    url: str
+    subscription: str
+    secret: str = field(repr=False)
+    """The client's ``whsec_`` secret. Kept out of ``repr``, like its key, so
+    logging a callback never leaks it."""
+    key: bytes = field(init=False, repr=False, compare=False)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "key", decode_secret(self.secret))
+
+
+@dataclass(frozen=True)
+class Event:
+    """One event occurrence. ``timestamp`` is ISO 8601 with a timezone."""
+
+    id: str
+    name: str
+    timestamp: str
+    data: Mapping[str, Any]
+
+    def encode(self) -> bytes:
+        return encode(
+            {
+                "eventId": self.id,
+                "name": self.name,
+                "timestamp": self.timestamp,
+                "data": dict(self.data),
+                "cursor": None,
+            }
+        )
+
+
+@dataclass(frozen=True)
+class RetryPolicy:
+    """Backoff between delivery attempts.
+
+    ``delays`` are the waits before each retry, so attempts are
+    ``len(delays) + 1``. Each wait is scaled by a random factor in
+    ``[1 - jitter, 1 + jitter]``. The defaults give 4 attempts with at most
+    ~12.6 minutes of waiting, inside the spec's 15-minute window even with
+    every attempt hitting the 10 s egress timeout.
+    """
+
+    delays: tuple[float, ...] = (30.0, 120.0, 480.0)
+    jitter: float = 0.2
+
+    def __post_init__(self) -> None:
+        if any(delay < 0 for delay in self.delays):
+            raise ValueError("delays must not be negative")
+        if not 0 <= self.jitter < 1:
+            raise ValueError("jitter must be in [0, 1)")
+
+    @property
+    def attempts(self) -> int:
+        return len(self.delays) + 1
+
+    def delay(self, retry: int, sample: float) -> float:
+        """Wait before retry number ``retry`` (0-based); ``sample`` in [0, 1)."""
+        return self.delays[retry] * (1 + self.jitter * (2 * sample - 1))
+
+
+@dataclass(frozen=True)
+class Result:
+    """How a delivery ended. ``reason`` is set unless delivered."""
+
+    outcome: Outcome
+    attempts: int
+    status: int | None = None
+    reason: Reason | None = None
+
+
+def classify(error: BaseException) -> Reason:
+    """Map a transport failure to its reported category."""
+    if isinstance(error, (TimeoutError, httpx.TimeoutException)):
+        return Reason.TIMEOUT
+    cause: BaseException | None = error
+    while cause is not None:
+        if isinstance(cause, ssl.SSLError):
+            return Reason.TLS_ERROR
+        cause = cause.__cause__ or cause.__context__
+    return Reason.CONNECTION_REFUSED
+
+
+def status_reason(status: int) -> Reason:
+    return Reason.HTTP_5XX if status >= 500 else Reason.HTTP_4XX
+
+
+# Failures raised by Egress.post; anything else is a bug and propagates.
+TRANSPORT_ERRORS = (DestinationError, TimeoutError, httpx.TransportError, OSError)
+
+
+class Dispatcher:
+    """Verifies callbacks and delivers events through one :class:`Egress`.
+
+    ``sleep``, ``clock`` and ``sample`` (jitter source) are injectable so tests can drive the
+    retry schedule without waiting.
+    """
+
+    def __init__(
+        self,
+        egress: Egress,
+        policy: RetryPolicy | None = None,
+        *,
+        sleep: Callable[[float], Awaitable[None]] = anyio.sleep,
+        clock: Callable[[], float] = time.time,
+        sample: Callable[[], float] = random.random,
+    ) -> None:
+        self._egress = egress
+        self._policy = policy or RetryPolicy()
+        self._sleep = sleep
+        self._clock = clock
+        self._sample = sample
+
+    def headers(
+        self, callback: Callback, message_id: str, body: bytes
+    ) -> dict[str, str]:
+        """Signed headers for one attempt, stamped with the current time. The
+        egress builds them once the request holds its slot, right before sending."""
+        timestamp = int(self._clock())
+        return {
+            "Content-Type": CONTENT_TYPE,
+            ID_HEADER: message_id,
+            TIMESTAMP_HEADER: str(timestamp),
+            SIGNATURE_HEADER: sign(callback.key, message_id, timestamp, body),
+            SUBSCRIPTION_HEADER: callback.subscription,
+        }
+
+    async def verify(self, callback: Callback) -> None:
+        """Run the verification handshake once; raise :class:`CallbackError`."""
+        challenge = secrets.token_urlsafe(CHALLENGE_BYTES)
+        body = encode({"type": VERIFICATION_TYPE, "challenge": challenge})
+        message_id = VERIFICATION_ID_PREFIX + secrets.token_hex(12)
+        try:
+            response = await self._egress.post(
+                callback.url, body, partial(self.headers, callback, message_id, body)
+            )
+        except TRANSPORT_ERRORS as error:
+            raise CallbackError(classify(error)) from error
+        if response.status >= 400:
+            raise CallbackError(status_reason(response.status))
+        if not 200 <= response.status < 300:
+            raise CallbackError(Reason.CHALLENGE_FAILED)
+        if not hmac.compare_digest(_echo(response.body), challenge.encode()):
+            raise CallbackError(Reason.CHALLENGE_FAILED)
+
+    async def deliver(self, callback: Callback, event: Event) -> Result:
+        """Deliver ``event``, retrying per the policy, and report the outcome."""
+        body = event.encode()
+        if len(body) > MAX_BODY_BYTES:
+            raise PayloadTooLargeError(
+                f"Event body is {len(body)} bytes; the limit is {MAX_BODY_BYTES}"
+            )
+        reason: Reason | None = None
+        status: int | None = None
+        for attempt in range(1, self._policy.attempts + 1):
+            if attempt > 1:
+                await self._sleep(self._policy.delay(attempt - 2, self._sample()))
+            try:
+                response = await self._egress.post(
+                    callback.url, body, partial(self.headers, callback, event.id, body)
+                )
+            except DestinationError:
+                return Result(
+                    Outcome.REJECTED, attempt, reason=Reason.CONNECTION_REFUSED
+                )
+            except TRANSPORT_ERRORS as error:
+                reason, status = classify(error), None
+                continue
+            status = response.status
+            if 200 <= status < 300:
+                return Result(Outcome.DELIVERED, attempt, status)
+            reason = status_reason(status)
+            if status in FINAL_STATUSES:
+                return Result(Outcome.REJECTED, attempt, status, reason)
+        return Result(Outcome.ABANDONED, self._policy.attempts, status, reason)
+
+
+def _echo(body: bytes) -> bytes:
+    """The ``challenge`` a verification response echoed, or empty bytes."""
+    try:
+        payload = json.loads(body)
+    except ValueError:
+        return b""
+    if not isinstance(payload, dict):
+        return b""
+    challenge = payload.get("challenge")
+    return challenge.encode() if isinstance(challenge, str) else b""