From c9a405ba80bd652a33a2fd5c4026a026e8686a0a Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 13:10:40 +0100 Subject: [PATCH 1/4] feat: sign and deliver MCP events safely Callback URLs are client-chosen, so every webhook POST is an attacker-steerable request. Add an SSRF-safe egress path that pins the connection to a checked global address while keeping SNI and Host, and a Standard Webhooks dispatcher for verification handshakes and bounded, re-signed event delivery retries. --- pyproject.toml | 1 + src/mcp_server_appwrite/events/delivery.py | 336 +++++++++++++++++ src/mcp_server_appwrite/events/egress.py | 379 +++++++++++++++++++ tests/unit/test_events_delivery.py | 414 +++++++++++++++++++++ tests/unit/test_events_egress.py | 352 ++++++++++++++++++ uv.lock | 2 + 6 files changed, 1484 insertions(+) create mode 100644 src/mcp_server_appwrite/events/delivery.py create mode 100644 src/mcp_server_appwrite/events/egress.py create mode 100644 tests/unit/test_events_delivery.py create mode 100644 tests/unit/test_events_egress.py diff --git a/pyproject.toml b/pyproject.toml index e7b81f3..a88be07 100644 --- a/pyproject.toml +++ b/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", diff --git a/src/mcp_server_appwrite/events/delivery.py b/src/mcp_server_appwrite/events/delivery.py new file mode 100644 index 0000000..b3f4286 --- /dev/null +++ b/src/mcp_server_appwrite/events/delivery.py @@ -0,0 +1,336 @@ +"""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,``, +where ``key`` is the base64-decoded part of the client's ``whsec_`` secret. +During rotation the header carries one space-separated signature per secret so +the receiver can verify with either. 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, Sequence +from dataclasses import dataclass, field +from enum import StrEnum +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(keys: Sequence[bytes], message_id: str, timestamp: int, body: bytes) -> str: + """Standard Webhooks ``webhook-signature`` value, one entry per key.""" + content = f"{message_id}.{timestamp}.".encode() + body + return " ".join( + f"{SIGNATURE_VERSION}," + + base64.b64encode(hmac.new(key, content, hashlib.sha256).digest()).decode() + for key in keys + ) + + +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. + + ``secrets`` holds every active ``whsec_`` secret, newest first; more than + one during a rotation grace window. + """ + + url: str + subscription: str + secrets: tuple[str, ...] + keys: tuple[bytes, ...] = field(init=False, repr=False, compare=False) + + def __post_init__(self) -> None: + if not self.secrets: + raise InvalidSecretError("At least one secret is required") + keys = tuple(decode_secret(secret) for secret in self.secrets) + object.__setattr__(self, "keys", keys) + + +@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.""" + timestamp = int(self._clock()) + return { + "Content-Type": CONTENT_TYPE, + ID_HEADER: message_id, + TIMESTAMP_HEADER: str(timestamp), + SIGNATURE_HEADER: sign(callback.keys, 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, 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, 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"" diff --git a/src/mcp_server_appwrite/events/egress.py b/src/mcp_server_appwrite/events/egress.py new file mode 100644 index 0000000..61c9638 --- /dev/null +++ b/src/mcp_server_appwrite/events/egress.py @@ -0,0 +1,379 @@ +"""SSRF-safe outbound HTTP for webhook callbacks. + +Callback URLs are chosen by MCP clients, so every request we send to one is an +attacker-steerable request from inside our network. This module is the only +path those requests take, for verification handshakes and event deliveries +alike, and it enforces the MCP Events "Webhook Security" rules: + +* ``https`` only. Plain ``http`` is accepted solely through the explicit + ``allow_loopback`` constructor flag, which tests use to reach a local server; + nothing reads it from the environment. +* We resolve DNS ourselves and refuse the request if *any* resolved address is + not globally routable (IANA IPv4/IPv6 special-purpose registries, plus the + IPv6 forms that embed an IPv4 address). +* The check happens at connect time, inside the httpcore network backend, and + the socket is opened to the address we just checked. The hostname never + reaches a second resolver, so DNS rebinding between check and connect is not + possible. TLS SNI and the ``Host`` header still carry the original hostname, + because httpcore derives both from the request URL, not from the address the + backend connected to. +* Redirects are never followed, the whole request is bounded by one deadline, + response bodies are read up to a small cap, and concurrent requests to one + destination host are limited so a single subscriber cannot monopolise egress. +""" + +from __future__ import annotations + +import ipaddress +import socket +import ssl +import typing +from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping +from contextlib import asynccontextmanager +from dataclasses import dataclass +from types import TracebackType + +import anyio +import httpcore +import httpx + +IPAddress = ipaddress.IPv4Address | ipaddress.IPv6Address + +# Maps (host, port) to every address the host resolves to. +Resolver = Callable[[str, int], Awaitable[list[IPAddress]]] + +SECURE_SCHEME = "https" +INSECURE_SCHEME = "http" + +# Total wall-clock budget for one request: connect, TLS, send and read. +TIMEOUT_SECONDS = 10.0 + +# Receivers answer with an acknowledgement or a challenge echo; anything past +# this is discarded unread. +BODY_LIMIT_BYTES = 16 * 1024 + +# Concurrent in-flight requests allowed per destination host. +CONCURRENCY = 4 + +# Special-purpose ranges that are never a valid webhook destination. The +# ``ipaddress`` predicates already cover most of these; listing them keeps the +# policy explicit and independent of Python patch-level registry updates. +BLOCKED_NETWORKS: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = ( + ipaddress.ip_network("0.0.0.0/8"), # "this network" + ipaddress.ip_network("10.0.0.0/8"), # private + ipaddress.ip_network("100.64.0.0/10"), # carrier-grade NAT + ipaddress.ip_network("127.0.0.0/8"), # loopback + ipaddress.ip_network("169.254.0.0/16"), # link-local, cloud metadata + ipaddress.ip_network("172.16.0.0/12"), # private + ipaddress.ip_network("192.0.0.0/24"), # IETF protocol assignments + ipaddress.ip_network("192.0.2.0/24"), # documentation + ipaddress.ip_network("192.88.99.0/24"), # deprecated 6to4 relay anycast + ipaddress.ip_network("192.168.0.0/16"), # private + ipaddress.ip_network("198.18.0.0/15"), # benchmarking + ipaddress.ip_network("198.51.100.0/24"), # documentation + ipaddress.ip_network("203.0.113.0/24"), # documentation + ipaddress.ip_network("224.0.0.0/4"), # multicast + ipaddress.ip_network("240.0.0.0/4"), # reserved, includes broadcast + ipaddress.ip_network("::/96"), # unspecified, loopback, IPv4-compatible + ipaddress.ip_network("64:ff9b:1::/48"), # local-use NAT64 + ipaddress.ip_network("100::/64"), # discard-only + ipaddress.ip_network("2001::/23"), # IETF protocol assignments, Teredo + ipaddress.ip_network("2001:db8::/32"), # documentation + ipaddress.ip_network("2002::/16"), # 6to4, routes via arbitrary relays + ipaddress.ip_network("3fff::/20"), # documentation + ipaddress.ip_network("fc00::/7"), # unique local + ipaddress.ip_network("fe80::/10"), # link-local + ipaddress.ip_network("fec0::/10"), # deprecated site-local + ipaddress.ip_network("ff00::/8"), # multicast +) + +# Well-known NAT64 prefix: the low 32 bits are the IPv4 destination. +NAT64_NETWORK = ipaddress.IPv6Network("64:ff9b::/96") + + +class DestinationError(Exception): + """The callback URL or an address it resolves to is not a permitted target.""" + + +def permitted(address: IPAddress, *, allow_loopback: bool = False) -> bool: + """Whether ``address`` is a globally routable webhook destination. + + IPv6 forms that carry an IPv4 destination (IPv4-mapped and NAT64) are + judged by the embedded address, so ``::ffff:10.0.0.1`` is as blocked as + ``10.0.0.1``. ``allow_loopback`` admits loopback only (for tests); every + other special-purpose range stays blocked. + """ + if isinstance(address, ipaddress.IPv6Address): + if address.ipv4_mapped is not None: + return permitted(address.ipv4_mapped, allow_loopback=allow_loopback) + if address in NAT64_NETWORK: + embedded = ipaddress.IPv4Address(int(address) & 0xFFFFFFFF) + return permitted(embedded) + if allow_loopback and address.is_loopback: + return True + if any(address in network for network in BLOCKED_NETWORKS): + return False + return address.is_global and not ( + address.is_private + or address.is_loopback + or address.is_link_local + or address.is_multicast + or address.is_reserved + or address.is_unspecified + ) + + +def validate_url(url: str, *, allow_loopback: bool = False) -> httpx.URL: + """Parse ``url`` and enforce the scheme and host rules for callbacks. + + This is the syntactic half of the policy; address checks happen when the + connection is opened. Raises :class:`DestinationError`. + """ + try: + parsed = httpx.URL(url) + except (httpx.InvalidURL, TypeError) as error: + raise DestinationError("Callback URL is not a valid URL") from error + schemes = {SECURE_SCHEME, INSECURE_SCHEME} if allow_loopback else {SECURE_SCHEME} + if parsed.scheme not in schemes: + raise DestinationError("Callback URL must use https") + if not parsed.host: + raise DestinationError("Callback URL must include a host") + if parsed.userinfo: + raise DestinationError("Callback URL must not include credentials") + return parsed + + +async def resolve(host: str, port: int) -> list[IPAddress]: + """Resolve ``host`` to every address it has, without connecting.""" + results = await anyio.getaddrinfo(host, port, type=socket.SOCK_STREAM) + addresses: list[IPAddress] = [] + for *_, sockaddr in results: + address = ipaddress.ip_address(sockaddr[0]) + if address not in addresses: + addresses.append(address) + return addresses + + +class Backend(httpcore.AsyncNetworkBackend): + """httpcore network backend that connects only to checked addresses. + + httpcore hands us the URL's hostname; we resolve it, reject the whole + request if any address is not permitted, and open the TCP stream to the + checked IP. httpcore then starts TLS with ``server_hostname`` taken from + the request origin, so SNI and certificate validation use the hostname. + """ + + def __init__( + self, + resolver: Resolver = resolve, + *, + allow_loopback: bool = False, + network: httpcore.AsyncNetworkBackend | None = None, + ) -> None: + self._resolver = resolver + self._allow_loopback = allow_loopback + # httpcore exports AnyIOBackend behind an import guard, which hides its + # base class from type checkers; anyio is a hard dependency here. + self._network = network or typing.cast( + httpcore.AsyncNetworkBackend, httpcore.AnyIOBackend() + ) + + async def addresses(self, host: str, port: int) -> list[IPAddress]: + """Resolve ``host`` and return its addresses if all are permitted.""" + addresses = await self._resolver(host, port) + if not addresses: + raise DestinationError("Callback host did not resolve") + for address in addresses: + if not permitted(address, allow_loopback=self._allow_loopback): + raise DestinationError("Callback host resolves to a non-public address") + return addresses + + async def connect_tcp( + self, + host: str, + port: int, + timeout: float | None = None, + local_address: str | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + with anyio.fail_after(timeout): + addresses = await self.addresses(host, port) + failure: Exception | None = None + for address in addresses: + try: + return await self._network.connect_tcp( + str(address), + port, + timeout=timeout, + local_address=local_address, + socket_options=socket_options, + ) + except (httpcore.ConnectError, httpcore.ConnectTimeout) as error: + failure = error + assert failure is not None + raise failure + + async def connect_unix_socket( + self, + path: str, + timeout: float | None = None, + socket_options: Iterable[httpcore.SOCKET_OPTION] | None = None, + ) -> httpcore.AsyncNetworkStream: + raise DestinationError("Unix sockets are not a webhook destination") + + async def sleep(self, seconds: float) -> None: + await self._network.sleep(seconds) + + +class Transport(httpx.AsyncHTTPTransport): + """httpx transport whose connection pool dials through :class:`Backend`. + + Proxies are deliberately unsupported: a proxy would resolve the hostname + itself and defeat the connect-time address check. + """ + + def __init__( + self, + backend: Backend, + *, + ssl_context: ssl.SSLContext | None = None, + ) -> None: + super().__init__(trust_env=False) + self._pool = httpcore.AsyncConnectionPool( + ssl_context=ssl_context or httpx.create_ssl_context(trust_env=False), + network_backend=backend, + ) + + +class Limiter: + """Bounds concurrent requests per destination host. + + Entries exist only while a host has requests waiting or in flight, so the + map does not grow with the number of distinct callback hosts seen. + """ + + def __init__(self, capacity: int) -> None: + if capacity < 1: + raise ValueError("capacity must be at least 1") + self._capacity = capacity + self._semaphores: dict[str, anyio.Semaphore] = {} + self._users: dict[str, int] = {} + + @asynccontextmanager + async def slot(self, host: str) -> AsyncIterator[None]: + semaphore = self._semaphores.get(host) + if semaphore is None: + semaphore = anyio.Semaphore(self._capacity) + self._semaphores[host] = semaphore + self._users[host] = self._users.get(host, 0) + 1 + try: + async with semaphore: + yield + finally: + self._users[host] -= 1 + if self._users[host] == 0: + del self._users[host] + del self._semaphores[host] + + +@dataclass(frozen=True) +class Response: + """Status and (capped) body of a callback response.""" + + status: int + body: bytes + + +class Egress: + """The single outbound HTTP path for webhook callbacks. + + Create one per process and share it; it owns a pooled ``httpx`` client. + ``transport`` replaces the SSRF-checking transport entirely and exists for + tests that stub the network with ``httpx.MockTransport``. + """ + + def __init__( + self, + *, + timeout: float = TIMEOUT_SECONDS, + body_limit: int = BODY_LIMIT_BYTES, + concurrency: int = CONCURRENCY, + resolver: Resolver = resolve, + allow_loopback: bool = False, + ssl_context: ssl.SSLContext | None = None, + network: httpcore.AsyncNetworkBackend | None = None, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self._timeout = timeout + self._body_limit = body_limit + self._allow_loopback = allow_loopback + self._limiter = Limiter(concurrency) + self._backend = Backend( + resolver, allow_loopback=allow_loopback, network=network + ) + self._client = httpx.AsyncClient( + transport=transport or Transport(self._backend, ssl_context=ssl_context), + follow_redirects=False, + timeout=httpx.Timeout(timeout), + trust_env=False, + ) + + async def check(self, url: str) -> None: + """Validate ``url`` and its current DNS answers without connecting. + + Lets the subscribe path reject an obviously bad callback early. It is + not a substitute for the connect-time check, which runs on every + request regardless. + """ + parsed = validate_url(url, allow_loopback=self._allow_loopback) + port = parsed.port or (443 if parsed.scheme == SECURE_SCHEME else 80) + with anyio.fail_after(self._timeout): + await self._backend.addresses(parsed.host, port) + + async def post( + self, url: str, content: bytes, headers: Mapping[str, str] + ) -> Response: + """POST ``content`` to ``url`` under the egress policy. + + Raises :class:`DestinationError` for a forbidden URL or address, + ``TimeoutError`` when the deadline passes, and ``httpx.TransportError`` + for network and TLS failures. + """ + parsed = validate_url(url, allow_loopback=self._allow_loopback) + # Identity encoding keeps the body cap meaningful: a compressed reply + # cannot expand past it. + request_headers = {**headers, "Accept-Encoding": "identity"} + async with self._limiter.slot(parsed.host): + with anyio.fail_after(self._timeout): + async with self._client.stream( + "POST", parsed, content=content, headers=request_headers + ) as response: + body = await self._read(response) + return Response(status=response.status_code, body=body) + + async def _read(self, response: httpx.Response) -> bytes: + # Raw bytes, not decoded ones: a body sent with Content-Encoding despite + # our identity request is cut at the cap instead of being inflated. + if response.is_stream_consumed: + # In-memory responses (e.g. from httpx.MockTransport) arrive read. + return response.content[: self._body_limit] + body = bytearray() + async for chunk in response.aiter_raw(): + body.extend(chunk) + if len(body) >= self._body_limit: + break + return bytes(body[: self._body_limit]) + + async def aclose(self) -> None: + await self._client.aclose() + + async def __aenter__(self) -> typing.Self: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + await self.aclose() diff --git a/tests/unit/test_events_delivery.py b/tests/unit/test_events_delivery.py new file mode 100644 index 0000000..ffb4512 --- /dev/null +++ b/tests/unit/test_events_delivery.py @@ -0,0 +1,414 @@ +import base64 +import hashlib +import hmac +import json +import unittest + +import httpx + +from mcp_server_appwrite.events import delivery, egress + +# Official Standard Webhooks signing vector, shared by every reference library, +# e.g. libraries/python/tests/test_webhooks.py::test_sign_function in +# https://github.com/standard-webhooks/standard-webhooks +VECTOR_SECRET = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw" +VECTOR_ID = "msg_p5jXN8AQM9LWM0D4loKWxJek" +VECTOR_TIMESTAMP = 1614265330 +VECTOR_BODY = b'{"test": 2432232314}' +VECTOR_SIGNATURE = "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=" + +URL = "https://hooks.example.com/mcp" +SUBSCRIPTION = "sub_0123456789abcdef" +SECRET = "whsec_" + base64.b64encode(bytes(range(32))).decode() +ROTATED = "whsec_" + base64.b64encode(bytes(range(100, 132))).decode() +EVENT = delivery.Event( + id="evt_1", + name="tablesdb.row.created", + timestamp="2026-10-09T12:00:00Z", + data={"row_id": "r1", "table_id": "tickets"}, +) + + +def verify(secret: str, headers: httpx.Headers, body: bytes, now: int) -> bool: + """Receiver-side Standard Webhooks check, written from the spec text.""" + key = base64.b64decode(secret.split("_", 1)[1]) + message_id = headers["webhook-id"] + timestamp = headers["webhook-timestamp"] + if abs(now - int(timestamp)) > 300: + return False + expected = base64.b64encode( + hmac.new( + key, message_id.encode() + b"." + timestamp.encode() + b"." + body, "sha256" + ).digest() + ) + for entry in headers["webhook-signature"].split(" "): + version, _, signature = entry.partition(",") + if version == "v1" and hmac.compare_digest(signature.encode(), expected): + return True + return False + + +class Clock: + def __init__(self, start: float = 1_760_000_000.0) -> None: + self.now = start + + def __call__(self) -> float: + return self.now + + +class Recorder: + """MockTransport handler replaying scripted responses and keeping requests.""" + + def __init__(self, *responses: httpx.Response | Exception) -> None: + self.responses = list(responses) + self.requests: list[httpx.Request] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + response = ( + self.responses.pop(0) if len(self.responses) > 1 else self.responses[0] + ) + if isinstance(response, Exception): + raise response + return response + + +class SigningTest(unittest.TestCase): + def test_matches_standard_webhooks_vector(self) -> None: + key = delivery.decode_secret(VECTOR_SECRET) + signature = delivery.sign([key], VECTOR_ID, VECTOR_TIMESTAMP, VECTOR_BODY) + self.assertEqual(signature, VECTOR_SIGNATURE) + + def test_vector_round_trips_through_independent_verifier(self) -> None: + headers = httpx.Headers( + { + "webhook-id": VECTOR_ID, + "webhook-timestamp": str(VECTOR_TIMESTAMP), + "webhook-signature": VECTOR_SIGNATURE, + } + ) + self.assertTrue(verify(VECTOR_SECRET, headers, VECTOR_BODY, VECTOR_TIMESTAMP)) + self.assertFalse( + verify(VECTOR_SECRET, headers, VECTOR_BODY + b" ", VECTOR_TIMESTAMP) + ) + + def test_signature_uses_the_decoded_key(self) -> None: + key = bytes(range(32)) + content = b"m.1." + b"{}" + expected = base64.b64encode( + hmac.new(key, content, hashlib.sha256).digest() + ).decode() + self.assertEqual(delivery.sign([key], "m", 1, b"{}"), "v1," + expected) + + +class SecretTest(unittest.TestCase): + def test_accepts_bounds(self) -> None: + for size in (24, 32, 64): + with self.subTest(size=size): + secret = "whsec_" + base64.b64encode(b"k" * size).decode() + self.assertEqual(delivery.decode_secret(secret), b"k" * size) + + def test_rejects_invalid_secrets(self) -> None: + cases = [ + "", + "whsec_", + "MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", + "WHSEC_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", + "whsec_not*base64!", + "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaS", # bad padding + "whsec_" + base64.b64encode(b"k" * 23).decode(), + "whsec_" + base64.b64encode(b"k" * 65).decode(), + "whsec_ " + base64.b64encode(b"k" * 32).decode(), + None, + 42, + ] + for secret in cases: + with self.subTest(secret=secret): + with self.assertRaises(delivery.InvalidSecretError): + delivery.decode_secret(secret) # type: ignore[arg-type] + + def test_callback_requires_valid_secrets(self) -> None: + with self.assertRaises(delivery.InvalidSecretError): + delivery.Callback(URL, SUBSCRIPTION, ()) + with self.assertRaises(delivery.InvalidSecretError): + delivery.Callback(URL, SUBSCRIPTION, ("whsec_short",)) + + +class VerificationTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.callback = delivery.Callback(URL, SUBSCRIPTION, (SECRET,)) + self.clock = Clock() + + async def run_verify(self, handler) -> None: + transport = httpx.MockTransport(handler) + async with egress.Egress(transport=transport) as client: + dispatcher = delivery.Dispatcher(client, clock=self.clock) + await dispatcher.verify(self.callback) + + async def assert_reason(self, handler, reason: delivery.Reason) -> None: + with self.assertRaises(delivery.CallbackError) as raised: + await self.run_verify(handler) + self.assertEqual(raised.exception.reason, reason) + + async def test_echoed_challenge_succeeds(self) -> None: + seen: list[httpx.Request] = [] + + def echo(request: httpx.Request) -> httpx.Response: + seen.append(request) + payload = json.loads(request.content) + return httpx.Response(200, json={"challenge": payload["challenge"]}) + + await self.run_verify(echo) + request = seen[0] + payload = json.loads(request.content) + self.assertEqual(payload["type"], "verification") + self.assertGreaterEqual(len(payload["challenge"]), 32) + self.assertTrue(request.headers["webhook-id"].startswith("msg_verification_")) + self.assertEqual(request.headers["x-mcp-subscription-id"], SUBSCRIPTION) + self.assertEqual(request.headers["content-type"], "application/json") + self.assertTrue( + verify(SECRET, request.headers, request.content, int(self.clock.now)) + ) + + async def test_challenges_are_single_use(self) -> None: + challenges: list[str] = [] + + def echo(request: httpx.Request) -> httpx.Response: + challenges.append(json.loads(request.content)["challenge"]) + return httpx.Response(200, json={"challenge": challenges[-1]}) + + await self.run_verify(echo) + await self.run_verify(echo) + self.assertNotEqual(challenges[0], challenges[1]) + + async def test_mismatched_challenge_fails(self) -> None: + await self.assert_reason( + lambda request: httpx.Response(200, json={"challenge": "wrong"}), + delivery.Reason.CHALLENGE_FAILED, + ) + + async def test_malformed_bodies_fail(self) -> None: + for content in [b"", b"not json", b"[]", b'{"challenge": 1}', b"{}"]: + with self.subTest(content=content): + await self.assert_reason( + lambda request, content=content: httpx.Response( + 200, content=content + ), + delivery.Reason.CHALLENGE_FAILED, + ) + + async def test_non_2xx_echo_fails(self) -> None: + def echo(status: int): + def handler(request: httpx.Request) -> httpx.Response: + challenge = json.loads(request.content)["challenge"] + return httpx.Response(status, json={"challenge": challenge}) + + return handler + + await self.assert_reason(echo(404), delivery.Reason.HTTP_4XX) + await self.assert_reason(echo(503), delivery.Reason.HTTP_5XX) + + async def test_redirect_is_not_followed(self) -> None: + recorder = Recorder( + httpx.Response(307, headers={"Location": "https://169.254.169.254/"}) + ) + await self.assert_reason(recorder, delivery.Reason.CHALLENGE_FAILED) + self.assertEqual(len(recorder.requests), 1) + + async def test_transport_failures_are_categorized(self) -> None: + await self.assert_reason( + Recorder(httpx.ReadTimeout("slow")), delivery.Reason.TIMEOUT + ) + await self.assert_reason( + Recorder(httpx.ConnectError("refused")), + delivery.Reason.CONNECTION_REFUSED, + ) + + async def test_forbidden_destination_is_connection_refused(self) -> None: + self.callback = delivery.Callback( + "http://hooks.example.com/", SUBSCRIPTION, (SECRET,) + ) + recorder = Recorder(httpx.Response(200)) + await self.assert_reason(recorder, delivery.Reason.CONNECTION_REFUSED) + self.assertEqual(recorder.requests, []) + + +class DeliveryTest(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self.callback = delivery.Callback(URL, SUBSCRIPTION, (SECRET,)) + self.clock = Clock() + self.sleeps: list[float] = [] + + async def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.clock.now += seconds + + async def run_deliver( + self, + recorder: Recorder, + event: delivery.Event = EVENT, + sample: float = 0.5, + ) -> delivery.Result: + transport = httpx.MockTransport(recorder) + async with egress.Egress(transport=transport) as client: + dispatcher = delivery.Dispatcher( + client, sleep=self.sleep, clock=self.clock, sample=lambda: sample + ) + return await dispatcher.deliver(self.callback, event) + + async def test_delivers_one_signed_event(self) -> None: + recorder = Recorder(httpx.Response(200)) + result = await self.run_deliver(recorder) + self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 1, 200)) + request = recorder.requests[0] + self.assertEqual(request.method, "POST") + self.assertEqual( + request.content, + b'{"eventId":"evt_1","name":"tablesdb.row.created",' + b'"timestamp":"2026-10-09T12:00:00Z",' + b'"data":{"row_id":"r1","table_id":"tickets"},"cursor":null}', + ) + self.assertEqual(request.headers["webhook-id"], "evt_1") + self.assertEqual(request.headers["x-mcp-subscription-id"], SUBSCRIPTION) + self.assertEqual(request.headers["content-type"], "application/json") + self.assertTrue( + verify(SECRET, request.headers, request.content, int(self.clock.now)) + ) + self.assertEqual(self.sleeps, []) + + async def test_retries_with_backoff_and_re_signs(self) -> None: + recorder = Recorder( + httpx.Response(503), + httpx.Response(429), + httpx.Response(202), + ) + result = await self.run_deliver(recorder) + self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 3, 202)) + self.assertEqual(self.sleeps, [30.0, 120.0]) + timestamps = [ + request.headers["webhook-timestamp"] for request in recorder.requests + ] + self.assertEqual(len(set(timestamps)), 3) + signatures = { + request.headers["webhook-signature"] for request in recorder.requests + } + self.assertEqual(len(signatures), 3) + self.assertEqual( + {r.headers["webhook-id"] for r in recorder.requests}, {"evt_1"} + ) + for request in recorder.requests: + sent = int(request.headers["webhook-timestamp"]) + self.assertTrue(verify(SECRET, request.headers, request.content, sent)) + + async def test_abandons_after_bounded_attempts(self) -> None: + recorder = Recorder(httpx.Response(500)) + result = await self.run_deliver(recorder, sample=0.999999) + self.assertEqual( + result, + delivery.Result( + delivery.Outcome.ABANDONED, 4, 500, delivery.Reason.HTTP_5XX + ), + ) + self.assertEqual(len(recorder.requests), 4) + self.assertEqual(len(self.sleeps), 3) + # Worst-case jitter plus four full egress timeouts stays within 15 min. + worst = sum(self.sleeps) + 4 * egress.TIMEOUT_SECONDS + self.assertLessEqual(worst, 15 * 60) + + async def test_jitter_bounds(self) -> None: + policy = delivery.RetryPolicy() + self.assertAlmostEqual(policy.delay(0, 0.0), 24.0) + self.assertAlmostEqual(policy.delay(0, 1.0), 36.0) + self.assertEqual(policy.attempts, 4) + with self.assertRaises(ValueError): + delivery.RetryPolicy(jitter=1.0) + with self.assertRaises(ValueError): + delivery.RetryPolicy(delays=(-1.0,)) + + async def test_gone_and_too_large_are_final(self) -> None: + for status in (410, 413): + with self.subTest(status=status): + self.sleeps.clear() + recorder = Recorder(httpx.Response(status)) + result = await self.run_deliver(recorder) + self.assertEqual( + result, + delivery.Result( + delivery.Outcome.REJECTED, 1, status, delivery.Reason.HTTP_4XX + ), + ) + self.assertEqual(len(recorder.requests), 1) + self.assertEqual(self.sleeps, []) + + async def test_transport_errors_are_retried(self) -> None: + recorder = Recorder( + httpx.ConnectError("refused"), + httpx.ReadTimeout("slow"), + httpx.Response(200), + ) + result = await self.run_deliver(recorder) + self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 3, 200)) + + async def test_last_error_category_is_reported(self) -> None: + result = await self.run_deliver(Recorder(httpx.ReadTimeout("slow"))) + self.assertEqual(result.outcome, delivery.Outcome.ABANDONED) + self.assertEqual(result.reason, delivery.Reason.TIMEOUT) + self.assertIsNone(result.status) + + async def test_forbidden_destination_is_not_retried(self) -> None: + self.callback = delivery.Callback( + "http://hooks.example.com/", SUBSCRIPTION, (SECRET,) + ) + recorder = Recorder(httpx.Response(200)) + result = await self.run_deliver(recorder) + self.assertEqual( + result, + delivery.Result( + delivery.Outcome.REJECTED, + 1, + reason=delivery.Reason.CONNECTION_REFUSED, + ), + ) + self.assertEqual(recorder.requests, []) + + async def test_dual_signs_during_rotation(self) -> None: + self.callback = delivery.Callback(URL, SUBSCRIPTION, (ROTATED, SECRET)) + recorder = Recorder(httpx.Response(200)) + await self.run_deliver(recorder) + request = recorder.requests[0] + headers, body = request.headers, request.content + self.assertEqual(len(headers["webhook-signature"].split(" ")), 2) + now = int(self.clock.now) + self.assertTrue(verify(SECRET, headers, body, now)) + self.assertTrue(verify(ROTATED, headers, body, now)) + other = "whsec_" + base64.b64encode(b"\x07" * 32).decode() + self.assertFalse(verify(other, headers, body, now)) + + async def test_payload_size_cap(self) -> None: + overhead = len( + delivery.Event(EVENT.id, EVENT.name, EVENT.timestamp, {"blob": ""}).encode() + ) + fits = delivery.Event( + EVENT.id, + EVENT.name, + EVENT.timestamp, + {"blob": "a" * (delivery.MAX_BODY_BYTES - overhead)}, + ) + self.assertEqual(len(fits.encode()), delivery.MAX_BODY_BYTES) + result = await self.run_deliver(Recorder(httpx.Response(200)), fits) + self.assertEqual(result.outcome, delivery.Outcome.DELIVERED) + + too_large = delivery.Event( + EVENT.id, + EVENT.name, + EVENT.timestamp, + {"blob": "a" * (delivery.MAX_BODY_BYTES - overhead + 1)}, + ) + recorder = Recorder(httpx.Response(200)) + with self.assertRaises(delivery.PayloadTooLargeError): + await self.run_deliver(recorder, too_large) + self.assertEqual(recorder.requests, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_events_egress.py b/tests/unit/test_events_egress.py new file mode 100644 index 0000000..f421cd0 --- /dev/null +++ b/tests/unit/test_events_egress.py @@ -0,0 +1,352 @@ +import asyncio +import datetime +import ipaddress +import ssl +import tempfile +import unittest +from pathlib import Path + +import httpcore +import httpx +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import NameOID + +from mcp_server_appwrite.events import delivery, egress + +HOSTNAME = "callback.test" +PUBLIC = ipaddress.ip_address("93.184.216.34") +PRIVATE = ipaddress.ip_address("10.0.0.7") + +BLOCKED = [ + "0.0.0.0", + "0.1.2.3", + "10.1.2.3", + "100.64.0.1", + "100.127.255.254", + "127.0.0.1", + "127.8.8.8", + "169.254.169.254", + "172.16.0.1", + "172.31.255.255", + "192.0.0.8", + "192.0.2.1", + "192.88.99.1", + "192.168.1.1", + "198.18.0.1", + "198.51.100.1", + "203.0.113.1", + "224.0.0.1", + "239.255.255.250", + "240.0.0.1", + "255.255.255.255", + "::", + "::1", + "::8.8.8.8", # IPv4-compatible (deprecated) + "::ffff:127.0.0.1", # IPv4-mapped loopback + "::ffff:10.0.0.1", # IPv4-mapped private + "::ffff:169.254.169.254", # IPv4-mapped metadata service + "::ffff:100.64.0.1", # IPv4-mapped CGNAT + "64:ff9b::a00:1", # NAT64 of 10.0.0.1 + "64:ff9b::7f00:1", # NAT64 of 127.0.0.1 + "64:ff9b::a9fe:a9fe", # NAT64 of 169.254.169.254 + "64:ff9b:1::1", # local-use NAT64 + "2002:a00:1::1", # 6to4 of 10.0.0.1 + "2002:7f00:1::1", # 6to4 of 127.0.0.1 + "2001::1", # Teredo + "2001:db8::1", # documentation + "3fff::1", # documentation + "100::1", # discard-only + "fc00::1", # unique local + "fd12:3456::1", # unique local + "fe80::1", # link-local + "febf::1", # link-local + "fec0::1", # site-local + "ff02::1", # multicast +] + +ALLOWED = [ + "8.8.8.8", + "1.1.1.1", + "93.184.216.34", + "2606:4700:4700::1111", + "2a00:1450:4001::200e", + "::ffff:8.8.8.8", # IPv4-mapped public + "64:ff9b::808:808", # NAT64 of 8.8.8.8 +] + + +def static(*addresses: ipaddress.IPv4Address | ipaddress.IPv6Address): + async def resolver(host: str, port: int): + return list(addresses) + + return resolver + + +class SpyNetwork(httpcore.AsyncNetworkBackend): + """Records connect targets and refuses every connection.""" + + def __init__(self) -> None: + self.targets: list[tuple[str, int]] = [] + + async def connect_tcp( + self, host, port, timeout=None, local_address=None, socket_options=None + ): + self.targets.append((host, port)) + raise httpcore.ConnectError("refused by spy") + + +class PermittedTest(unittest.TestCase): + def test_blocks_special_purpose_addresses(self) -> None: + for text in BLOCKED: + with self.subTest(address=text): + self.assertFalse(egress.permitted(ipaddress.ip_address(text))) + + def test_allows_global_addresses(self) -> None: + for text in ALLOWED: + with self.subTest(address=text): + self.assertTrue(egress.permitted(ipaddress.ip_address(text))) + + def test_allow_loopback_admits_only_loopback(self) -> None: + for text in ["127.0.0.1", "::1", "::ffff:127.0.0.1"]: + with self.subTest(address=text): + self.assertTrue( + egress.permitted(ipaddress.ip_address(text), allow_loopback=True) + ) + for text in ["10.0.0.1", "169.254.169.254", "fd00::1", "64:ff9b::7f00:1"]: + with self.subTest(address=text): + self.assertFalse( + egress.permitted(ipaddress.ip_address(text), allow_loopback=True) + ) + + +class ValidateUrlTest(unittest.TestCase): + def test_accepts_https(self) -> None: + url = egress.validate_url("https://hooks.example.com/a?b=c") + self.assertEqual(url.host, "hooks.example.com") + + def test_rejects_other_schemes_and_shapes(self) -> None: + for url in [ + "http://hooks.example.com/", + "ftp://hooks.example.com/", + "file:///etc/passwd", + "https:///path", + "https://user:pass@hooks.example.com/", + "not a url", + ]: + with self.subTest(url=url): + with self.assertRaises(egress.DestinationError): + egress.validate_url(url) + + def test_http_only_with_explicit_loopback_flag(self) -> None: + url = egress.validate_url("http://127.0.0.1:8080/", allow_loopback=True) + self.assertEqual(url.scheme, "http") + + +class ConnectTargetTest(unittest.IsolatedAsyncioTestCase): + async def test_connects_to_the_checked_address(self) -> None: + spy = SpyNetwork() + async with egress.Egress(resolver=static(PUBLIC), network=spy) as client: + with self.assertRaises(httpx.ConnectError): + await client.post("https://hooks.example.com/x", b"{}", {}) + self.assertEqual(spy.targets, [(str(PUBLIC), 443)]) + + async def test_any_private_answer_rejects_the_host(self) -> None: + spy = SpyNetwork() + resolver = static(PUBLIC, PRIVATE) + async with egress.Egress(resolver=resolver, network=spy) as client: + with self.assertRaises(egress.DestinationError): + await client.post("https://hooks.example.com/x", b"{}", {}) + self.assertEqual(spy.targets, []) + + async def test_dns_rebinding_is_rejected_at_connect(self) -> None: + answers = [[PUBLIC], [PRIVATE]] + + async def rebinding(host: str, port: int): + return answers.pop(0) + + spy = SpyNetwork() + async with egress.Egress(resolver=rebinding, network=spy) as client: + await client.check("https://rebind.example.com/hook") + with self.assertRaises(egress.DestinationError): + await client.post("https://rebind.example.com/hook", b"{}", {}) + self.assertEqual(spy.targets, []) + + async def test_check_rejects_private_answers(self) -> None: + async with egress.Egress(resolver=static(PRIVATE)) as client: + with self.assertRaises(egress.DestinationError): + await client.check("https://internal.example.com/") + + async def test_unresolvable_host_is_rejected(self) -> None: + async with egress.Egress(resolver=static(), network=SpyNetwork()) as client: + with self.assertRaises(egress.DestinationError): + await client.post("https://nowhere.example.com/", b"{}", {}) + + +class LimitsTest(unittest.IsolatedAsyncioTestCase): + async def test_total_timeout(self) -> None: + async def slow(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(5) + return httpx.Response(200) + + transport = httpx.MockTransport(slow) + async with egress.Egress(timeout=0.05, transport=transport) as client: + with self.assertRaises(TimeoutError): + await client.post("https://hooks.example.com/", b"{}", {}) + + async def test_response_body_is_capped(self) -> None: + transport = httpx.MockTransport( + lambda request: httpx.Response(200, content=b"x" * 100_000) + ) + async with egress.Egress(body_limit=1024, transport=transport) as client: + response = await client.post("https://hooks.example.com/", b"{}", {}) + self.assertEqual(len(response.body), 1024) + + async def test_concurrency_is_limited_per_host(self) -> None: + active: dict[str, int] = {} + peak: dict[str, int] = {} + + async def handler(request: httpx.Request) -> httpx.Response: + host = request.url.host + active[host] = active.get(host, 0) + 1 + peak[host] = max(peak.get(host, 0), active[host]) + await asyncio.sleep(0.01) + active[host] -= 1 + return httpx.Response(200) + + transport = httpx.MockTransport(handler) + async with egress.Egress(concurrency=2, transport=transport) as client: + await asyncio.gather( + *[client.post("https://a.example.com/", b"", {}) for _ in range(6)], + *[client.post("https://b.example.com/", b"", {}) for _ in range(6)], + ) + self.assertEqual(client._limiter._semaphores, {}) + self.assertEqual(peak, {"a.example.com": 2, "b.example.com": 2}) + + +def certificate(directory: Path) -> tuple[Path, Path]: + key = ec.generate_private_key(ec.SECP256R1()) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, HOSTNAME)]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=1)) + .not_valid_after(now + datetime.timedelta(hours=1)) + .add_extension(x509.SubjectAlternativeName([x509.DNSName(HOSTNAME)]), False) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), True) + .sign(key, hashes.SHA256()) + ) + certificate_path = directory / "cert.pem" + key_path = directory / "key.pem" + certificate_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + key_path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + ) + return certificate_path, key_path + + +class LocalTLSServerTest(unittest.IsolatedAsyncioTestCase): + """A real TLS server on 127.0.0.1 that only knows itself as callback.test.""" + + async def asyncSetUp(self) -> None: + self.directory = tempfile.TemporaryDirectory() + certificate_path, key_path = certificate(Path(self.directory.name)) + self.server_names: list[str | None] = [] + self.requests: list[tuple[str, dict[str, str]]] = [] + + server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + server_context.load_cert_chain(certificate_path, key_path) + server_context.sni_callback = self._record_sni + self.server = await asyncio.start_server( + self._handle, "127.0.0.1", 0, ssl=server_context + ) + self.port = self.server.sockets[0].getsockname()[1] + + self.trusting = ssl.create_default_context(cafile=str(certificate_path)) + self.resolver = static(ipaddress.ip_address("127.0.0.1")) + + async def asyncTearDown(self) -> None: + self.server.close() + await self.server.wait_closed() + self.directory.cleanup() + + def _record_sni(self, connection, server_name, context) -> None: + self.server_names.append(server_name) + + async def _handle(self, reader, writer) -> None: + try: + head = await reader.readuntil(b"\r\n\r\n") + except (asyncio.IncompleteReadError, ConnectionResetError): + writer.close() + return + lines = head.decode().split("\r\n") + path = lines[0].split(" ")[1] + headers = { + name.strip().lower(): value.strip() + for name, value in (line.split(":", 1) for line in lines[1:] if line) + } + await reader.readexactly(int(headers.get("content-length", "0"))) + self.requests.append((path, headers)) + if path == "/redirect": + status = "302 Found" + extra = f"Location: https://{HOSTNAME}:{self.port}/ok\r\n" + body = b"" + else: + status, extra, body = "200 OK", "", b'{"ok":true}' + writer.write( + f"HTTP/1.1 {status}\r\n{extra}Content-Length: {len(body)}\r\n" + "Connection: close\r\n\r\n".encode() + body + ) + await writer.drain() + writer.close() + + def url(self, path: str) -> str: + return f"https://{HOSTNAME}:{self.port}{path}" + + async def test_sni_and_host_keep_the_hostname(self) -> None: + async with egress.Egress( + resolver=self.resolver, allow_loopback=True, ssl_context=self.trusting + ) as client: + response = await client.post(self.url("/ok"), b"{}", {}) + self.assertEqual(response.status, 200) + self.assertEqual(response.body, b'{"ok":true}') + self.assertEqual(self.server_names, [HOSTNAME]) + self.assertEqual(self.requests[0][1]["host"], f"{HOSTNAME}:{self.port}") + + async def test_redirects_are_not_followed(self) -> None: + async with egress.Egress( + resolver=self.resolver, allow_loopback=True, ssl_context=self.trusting + ) as client: + response = await client.post(self.url("/redirect"), b"{}", {}) + self.assertEqual(response.status, 302) + self.assertEqual([path for path, _ in self.requests], ["/redirect"]) + + async def test_loopback_is_blocked_without_the_flag(self) -> None: + async with egress.Egress( + resolver=self.resolver, ssl_context=self.trusting + ) as client: + with self.assertRaises(egress.DestinationError): + await client.post(self.url("/ok"), b"{}", {}) + self.assertEqual(self.requests, []) + + async def test_untrusted_certificate_is_a_tls_error(self) -> None: + # The server side logs the aborted handshake; that noise is expected. + asyncio.get_running_loop().set_exception_handler(lambda loop, context: None) + async with egress.Egress(resolver=self.resolver, allow_loopback=True) as client: + with self.assertRaises(httpx.ConnectError) as raised: + await client.post(self.url("/ok"), b"{}", {}) + self.assertIs(delivery.classify(raised.exception), delivery.Reason.TLS_ERROR) + self.assertEqual(self.requests, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/uv.lock b/uv.lock index dafc3b7..43eadd6 100644 --- a/uv.lock +++ b/uv.lock @@ -699,6 +699,7 @@ dependencies = [ { name = "appwrite-console" }, { name = "cryptography" }, { name = "docstring-parser" }, + { name = "httpcore" }, { name = "httpx" }, { name = "mcp", extra = ["cli"] }, { name = "mcp-event-subscriptions" }, @@ -740,6 +741,7 @@ requires-dist = [ { name = "bcrypt", marker = "extra == 'integration'", specifier = ">=4.1.2" }, { name = "cryptography", specifier = ">=44.0.0" }, { name = "docstring-parser", specifier = ">=0.16" }, + { name = "httpcore", specifier = ">=1.0" }, { name = "httpx", specifier = ">=0.27.0" }, { name = "mcp", extras = ["cli"], specifier = ">=2.3,<3" }, { name = "mcp-event-subscriptions", specifier = ">=0.1.0a1,<0.2" }, From 4964cbb8dac96ca3499cf3eed40555eeb514da1a Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Fri, 9 Oct 2026 14:40:53 +0100 Subject: [PATCH 2/4] test: keep only delivery and egress tests that e2e cannot reach Signing, retries, final statuses, TLS and redirect handling, timeouts and per-host limits are covered through the ingress e2e flows (#134). What stays needs a controlled resolver or address table, runs only from subscribe (PR 5), or waits on the production retry schedule. --- tests/unit/test_events_delivery.py | 187 +++-------------------------- tests/unit/test_events_egress.py | 185 +++------------------------- 2 files changed, 38 insertions(+), 334 deletions(-) diff --git a/tests/unit/test_events_delivery.py b/tests/unit/test_events_delivery.py index ffb4512..7a2daf0 100644 --- a/tests/unit/test_events_delivery.py +++ b/tests/unit/test_events_delivery.py @@ -1,5 +1,19 @@ +"""Delivery behavior that no e2e flow can reach yet. + +Signing (checked by the official ``standardwebhooks`` library), the delivery +body and headers, re-signed retries with a short injected policy, 410/413 not +retried, dual signing during secret rotation and the 256 KiB cap are covered +end to end through the ingress in ``tests/e2e/test_events_ingress.py`` (#134). +What stays here: + +* ``whsec_`` secret validation and the verification handshake: only + ``events/subscribe`` (PR 5 of #127) calls them. Replace these with its e2e + flow. +* The production retry schedule, whose real-time waits (30 s, 2 min, 8 min) + are too long to run end to end. +""" + import base64 -import hashlib import hmac import json import unittest @@ -8,19 +22,9 @@ from mcp_server_appwrite.events import delivery, egress -# Official Standard Webhooks signing vector, shared by every reference library, -# e.g. libraries/python/tests/test_webhooks.py::test_sign_function in -# https://github.com/standard-webhooks/standard-webhooks -VECTOR_SECRET = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw" -VECTOR_ID = "msg_p5jXN8AQM9LWM0D4loKWxJek" -VECTOR_TIMESTAMP = 1614265330 -VECTOR_BODY = b'{"test": 2432232314}' -VECTOR_SIGNATURE = "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE=" - URL = "https://hooks.example.com/mcp" SUBSCRIPTION = "sub_0123456789abcdef" SECRET = "whsec_" + base64.b64encode(bytes(range(32))).decode() -ROTATED = "whsec_" + base64.b64encode(bytes(range(100, 132))).decode() EVENT = delivery.Event( id="evt_1", name="tablesdb.row.created", @@ -73,34 +77,6 @@ def __call__(self, request: httpx.Request) -> httpx.Response: return response -class SigningTest(unittest.TestCase): - def test_matches_standard_webhooks_vector(self) -> None: - key = delivery.decode_secret(VECTOR_SECRET) - signature = delivery.sign([key], VECTOR_ID, VECTOR_TIMESTAMP, VECTOR_BODY) - self.assertEqual(signature, VECTOR_SIGNATURE) - - def test_vector_round_trips_through_independent_verifier(self) -> None: - headers = httpx.Headers( - { - "webhook-id": VECTOR_ID, - "webhook-timestamp": str(VECTOR_TIMESTAMP), - "webhook-signature": VECTOR_SIGNATURE, - } - ) - self.assertTrue(verify(VECTOR_SECRET, headers, VECTOR_BODY, VECTOR_TIMESTAMP)) - self.assertFalse( - verify(VECTOR_SECRET, headers, VECTOR_BODY + b" ", VECTOR_TIMESTAMP) - ) - - def test_signature_uses_the_decoded_key(self) -> None: - key = bytes(range(32)) - content = b"m.1." + b"{}" - expected = base64.b64encode( - hmac.new(key, content, hashlib.sha256).digest() - ).decode() - self.assertEqual(delivery.sign([key], "m", 1, b"{}"), "v1," + expected) - - class SecretTest(unittest.TestCase): def test_accepts_bounds(self) -> None: for size in (24, 32, 64): @@ -233,7 +209,10 @@ async def test_forbidden_destination_is_connection_refused(self) -> None: self.assertEqual(recorder.requests, []) -class DeliveryTest(unittest.IsolatedAsyncioTestCase): +class RetryScheduleTest(unittest.IsolatedAsyncioTestCase): + """The production schedule spans ~12 minutes, too long to wait for in e2e, + which runs a short injected policy instead.""" + def setUp(self) -> None: self.callback = delivery.Callback(URL, SUBSCRIPTION, (SECRET,)) self.clock = Clock() @@ -256,50 +235,6 @@ async def run_deliver( ) return await dispatcher.deliver(self.callback, event) - async def test_delivers_one_signed_event(self) -> None: - recorder = Recorder(httpx.Response(200)) - result = await self.run_deliver(recorder) - self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 1, 200)) - request = recorder.requests[0] - self.assertEqual(request.method, "POST") - self.assertEqual( - request.content, - b'{"eventId":"evt_1","name":"tablesdb.row.created",' - b'"timestamp":"2026-10-09T12:00:00Z",' - b'"data":{"row_id":"r1","table_id":"tickets"},"cursor":null}', - ) - self.assertEqual(request.headers["webhook-id"], "evt_1") - self.assertEqual(request.headers["x-mcp-subscription-id"], SUBSCRIPTION) - self.assertEqual(request.headers["content-type"], "application/json") - self.assertTrue( - verify(SECRET, request.headers, request.content, int(self.clock.now)) - ) - self.assertEqual(self.sleeps, []) - - async def test_retries_with_backoff_and_re_signs(self) -> None: - recorder = Recorder( - httpx.Response(503), - httpx.Response(429), - httpx.Response(202), - ) - result = await self.run_deliver(recorder) - self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 3, 202)) - self.assertEqual(self.sleeps, [30.0, 120.0]) - timestamps = [ - request.headers["webhook-timestamp"] for request in recorder.requests - ] - self.assertEqual(len(set(timestamps)), 3) - signatures = { - request.headers["webhook-signature"] for request in recorder.requests - } - self.assertEqual(len(signatures), 3) - self.assertEqual( - {r.headers["webhook-id"] for r in recorder.requests}, {"evt_1"} - ) - for request in recorder.requests: - sent = int(request.headers["webhook-timestamp"]) - self.assertTrue(verify(SECRET, request.headers, request.content, sent)) - async def test_abandons_after_bounded_attempts(self) -> None: recorder = Recorder(httpx.Response(500)) result = await self.run_deliver(recorder, sample=0.999999) @@ -325,90 +260,6 @@ async def test_jitter_bounds(self) -> None: with self.assertRaises(ValueError): delivery.RetryPolicy(delays=(-1.0,)) - async def test_gone_and_too_large_are_final(self) -> None: - for status in (410, 413): - with self.subTest(status=status): - self.sleeps.clear() - recorder = Recorder(httpx.Response(status)) - result = await self.run_deliver(recorder) - self.assertEqual( - result, - delivery.Result( - delivery.Outcome.REJECTED, 1, status, delivery.Reason.HTTP_4XX - ), - ) - self.assertEqual(len(recorder.requests), 1) - self.assertEqual(self.sleeps, []) - - async def test_transport_errors_are_retried(self) -> None: - recorder = Recorder( - httpx.ConnectError("refused"), - httpx.ReadTimeout("slow"), - httpx.Response(200), - ) - result = await self.run_deliver(recorder) - self.assertEqual(result, delivery.Result(delivery.Outcome.DELIVERED, 3, 200)) - - async def test_last_error_category_is_reported(self) -> None: - result = await self.run_deliver(Recorder(httpx.ReadTimeout("slow"))) - self.assertEqual(result.outcome, delivery.Outcome.ABANDONED) - self.assertEqual(result.reason, delivery.Reason.TIMEOUT) - self.assertIsNone(result.status) - - async def test_forbidden_destination_is_not_retried(self) -> None: - self.callback = delivery.Callback( - "http://hooks.example.com/", SUBSCRIPTION, (SECRET,) - ) - recorder = Recorder(httpx.Response(200)) - result = await self.run_deliver(recorder) - self.assertEqual( - result, - delivery.Result( - delivery.Outcome.REJECTED, - 1, - reason=delivery.Reason.CONNECTION_REFUSED, - ), - ) - self.assertEqual(recorder.requests, []) - - async def test_dual_signs_during_rotation(self) -> None: - self.callback = delivery.Callback(URL, SUBSCRIPTION, (ROTATED, SECRET)) - recorder = Recorder(httpx.Response(200)) - await self.run_deliver(recorder) - request = recorder.requests[0] - headers, body = request.headers, request.content - self.assertEqual(len(headers["webhook-signature"].split(" ")), 2) - now = int(self.clock.now) - self.assertTrue(verify(SECRET, headers, body, now)) - self.assertTrue(verify(ROTATED, headers, body, now)) - other = "whsec_" + base64.b64encode(b"\x07" * 32).decode() - self.assertFalse(verify(other, headers, body, now)) - - async def test_payload_size_cap(self) -> None: - overhead = len( - delivery.Event(EVENT.id, EVENT.name, EVENT.timestamp, {"blob": ""}).encode() - ) - fits = delivery.Event( - EVENT.id, - EVENT.name, - EVENT.timestamp, - {"blob": "a" * (delivery.MAX_BODY_BYTES - overhead)}, - ) - self.assertEqual(len(fits.encode()), delivery.MAX_BODY_BYTES) - result = await self.run_deliver(Recorder(httpx.Response(200)), fits) - self.assertEqual(result.outcome, delivery.Outcome.DELIVERED) - - too_large = delivery.Event( - EVENT.id, - EVENT.name, - EVENT.timestamp, - {"blob": "a" * (delivery.MAX_BODY_BYTES - overhead + 1)}, - ) - recorder = Recorder(httpx.Response(200)) - with self.assertRaises(delivery.PayloadTooLargeError): - await self.run_deliver(recorder, too_large) - self.assertEqual(recorder.requests, []) - if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_events_egress.py b/tests/unit/test_events_egress.py index f421cd0..9293a48 100644 --- a/tests/unit/test_events_egress.py +++ b/tests/unit/test_events_egress.py @@ -1,21 +1,28 @@ -import asyncio -import datetime +"""SSRF rules that need a controlled resolver or address table. + +Against a real local HTTPS receiver, the ingress e2e flows (#134) cover: SNI and +``Host`` keep the hostname while the socket dials the checked address, +redirects are not followed, loopback is refused without the test-only flag, an +untrusted certificate is a ``tls_error``, the total timeout, and the +per-host concurrency limit. What stays here cannot be produced end to end: + +* The blocklist across IPv4, IPv6, IPv4-mapped, NAT64, 6to4 and Teredo forms: + a test machine cannot route to most of these addresses. +* Mixed public and private DNS answers and DNS rebinding: they need a resolver + that answers differently between check and connect. +* URL shapes the subscribe path (PR 5) will reject before any delivery. +* The response body cap: delivery discards response bodies, so it is + invisible on the wire. +""" + import ipaddress -import ssl -import tempfile import unittest -from pathlib import Path import httpcore import httpx -from cryptography import x509 -from cryptography.hazmat.primitives import hashes, serialization -from cryptography.hazmat.primitives.asymmetric import ec -from cryptography.x509.oid import NameOID -from mcp_server_appwrite.events import delivery, egress +from mcp_server_appwrite.events import egress -HOSTNAME = "callback.test" PUBLIC = ipaddress.ip_address("93.184.216.34") PRIVATE = ipaddress.ip_address("10.0.0.7") @@ -184,17 +191,7 @@ async def test_unresolvable_host_is_rejected(self) -> None: await client.post("https://nowhere.example.com/", b"{}", {}) -class LimitsTest(unittest.IsolatedAsyncioTestCase): - async def test_total_timeout(self) -> None: - async def slow(request: httpx.Request) -> httpx.Response: - await asyncio.sleep(5) - return httpx.Response(200) - - transport = httpx.MockTransport(slow) - async with egress.Egress(timeout=0.05, transport=transport) as client: - with self.assertRaises(TimeoutError): - await client.post("https://hooks.example.com/", b"{}", {}) - +class BodyCapTest(unittest.IsolatedAsyncioTestCase): async def test_response_body_is_capped(self) -> None: transport = httpx.MockTransport( lambda request: httpx.Response(200, content=b"x" * 100_000) @@ -203,150 +200,6 @@ async def test_response_body_is_capped(self) -> None: response = await client.post("https://hooks.example.com/", b"{}", {}) self.assertEqual(len(response.body), 1024) - async def test_concurrency_is_limited_per_host(self) -> None: - active: dict[str, int] = {} - peak: dict[str, int] = {} - - async def handler(request: httpx.Request) -> httpx.Response: - host = request.url.host - active[host] = active.get(host, 0) + 1 - peak[host] = max(peak.get(host, 0), active[host]) - await asyncio.sleep(0.01) - active[host] -= 1 - return httpx.Response(200) - - transport = httpx.MockTransport(handler) - async with egress.Egress(concurrency=2, transport=transport) as client: - await asyncio.gather( - *[client.post("https://a.example.com/", b"", {}) for _ in range(6)], - *[client.post("https://b.example.com/", b"", {}) for _ in range(6)], - ) - self.assertEqual(client._limiter._semaphores, {}) - self.assertEqual(peak, {"a.example.com": 2, "b.example.com": 2}) - - -def certificate(directory: Path) -> tuple[Path, Path]: - key = ec.generate_private_key(ec.SECP256R1()) - name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, HOSTNAME)]) - now = datetime.datetime.now(datetime.timezone.utc) - cert = ( - x509.CertificateBuilder() - .subject_name(name) - .issuer_name(name) - .public_key(key.public_key()) - .serial_number(x509.random_serial_number()) - .not_valid_before(now - datetime.timedelta(minutes=1)) - .not_valid_after(now + datetime.timedelta(hours=1)) - .add_extension(x509.SubjectAlternativeName([x509.DNSName(HOSTNAME)]), False) - .add_extension(x509.BasicConstraints(ca=True, path_length=None), True) - .sign(key, hashes.SHA256()) - ) - certificate_path = directory / "cert.pem" - key_path = directory / "key.pem" - certificate_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) - key_path.write_bytes( - key.private_bytes( - serialization.Encoding.PEM, - serialization.PrivateFormat.PKCS8, - serialization.NoEncryption(), - ) - ) - return certificate_path, key_path - - -class LocalTLSServerTest(unittest.IsolatedAsyncioTestCase): - """A real TLS server on 127.0.0.1 that only knows itself as callback.test.""" - - async def asyncSetUp(self) -> None: - self.directory = tempfile.TemporaryDirectory() - certificate_path, key_path = certificate(Path(self.directory.name)) - self.server_names: list[str | None] = [] - self.requests: list[tuple[str, dict[str, str]]] = [] - - server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - server_context.load_cert_chain(certificate_path, key_path) - server_context.sni_callback = self._record_sni - self.server = await asyncio.start_server( - self._handle, "127.0.0.1", 0, ssl=server_context - ) - self.port = self.server.sockets[0].getsockname()[1] - - self.trusting = ssl.create_default_context(cafile=str(certificate_path)) - self.resolver = static(ipaddress.ip_address("127.0.0.1")) - - async def asyncTearDown(self) -> None: - self.server.close() - await self.server.wait_closed() - self.directory.cleanup() - - def _record_sni(self, connection, server_name, context) -> None: - self.server_names.append(server_name) - - async def _handle(self, reader, writer) -> None: - try: - head = await reader.readuntil(b"\r\n\r\n") - except (asyncio.IncompleteReadError, ConnectionResetError): - writer.close() - return - lines = head.decode().split("\r\n") - path = lines[0].split(" ")[1] - headers = { - name.strip().lower(): value.strip() - for name, value in (line.split(":", 1) for line in lines[1:] if line) - } - await reader.readexactly(int(headers.get("content-length", "0"))) - self.requests.append((path, headers)) - if path == "/redirect": - status = "302 Found" - extra = f"Location: https://{HOSTNAME}:{self.port}/ok\r\n" - body = b"" - else: - status, extra, body = "200 OK", "", b'{"ok":true}' - writer.write( - f"HTTP/1.1 {status}\r\n{extra}Content-Length: {len(body)}\r\n" - "Connection: close\r\n\r\n".encode() + body - ) - await writer.drain() - writer.close() - - def url(self, path: str) -> str: - return f"https://{HOSTNAME}:{self.port}{path}" - - async def test_sni_and_host_keep_the_hostname(self) -> None: - async with egress.Egress( - resolver=self.resolver, allow_loopback=True, ssl_context=self.trusting - ) as client: - response = await client.post(self.url("/ok"), b"{}", {}) - self.assertEqual(response.status, 200) - self.assertEqual(response.body, b'{"ok":true}') - self.assertEqual(self.server_names, [HOSTNAME]) - self.assertEqual(self.requests[0][1]["host"], f"{HOSTNAME}:{self.port}") - - async def test_redirects_are_not_followed(self) -> None: - async with egress.Egress( - resolver=self.resolver, allow_loopback=True, ssl_context=self.trusting - ) as client: - response = await client.post(self.url("/redirect"), b"{}", {}) - self.assertEqual(response.status, 302) - self.assertEqual([path for path, _ in self.requests], ["/redirect"]) - - async def test_loopback_is_blocked_without_the_flag(self) -> None: - async with egress.Egress( - resolver=self.resolver, ssl_context=self.trusting - ) as client: - with self.assertRaises(egress.DestinationError): - await client.post(self.url("/ok"), b"{}", {}) - self.assertEqual(self.requests, []) - - async def test_untrusted_certificate_is_a_tls_error(self) -> None: - # The server side logs the aborted handshake; that noise is expected. - asyncio.get_running_loop().set_exception_handler(lambda loop, context: None) - async with egress.Egress(resolver=self.resolver, allow_loopback=True) as client: - with self.assertRaises(httpx.ConnectError) as raised: - await client.post(self.url("/ok"), b"{}", {}) - self.assertIs(delivery.classify(raised.exception), delivery.Reason.TLS_ERROR) - self.assertEqual(self.requests, []) - if __name__ == "__main__": unittest.main() From 321f72b49d381bb0ab8efd38223d8d41dd3de0e2 Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Sun, 11 Oct 2026 21:04:14 +0100 Subject: [PATCH 3/4] fix: confine test-only http to loopback and bound queued egress allow_loopback widened the allowed schemes to http for every host, so an embedder enabling it could send signed event data unencrypted to a public callback. Plain http now needs a loopback host (localhost or a loopback IP) whose every resolved address is loopback. The per-host slot was acquired before the request deadline started, so a request queued behind slow ones to the same host could wait without bound and go out with a signature timestamped long before it was sent. One deadline now covers the slot wait and the request, and Egress.post takes a headers factory that is called only once the slot is held, right before sending. --- src/mcp_server_appwrite/events/delivery.py | 8 +- src/mcp_server_appwrite/events/egress.py | 68 ++++++++--- tests/unit/test_events_egress.py | 125 ++++++++++++++++++++- 3 files changed, 180 insertions(+), 21 deletions(-) diff --git a/src/mcp_server_appwrite/events/delivery.py b/src/mcp_server_appwrite/events/delivery.py index b3f4286..f996823 100644 --- a/src/mcp_server_appwrite/events/delivery.py +++ b/src/mcp_server_appwrite/events/delivery.py @@ -39,6 +39,7 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass, field from enum import StrEnum +from functools import partial from typing import Any import anyio @@ -264,7 +265,8 @@ def __init__( def headers( self, callback: Callback, message_id: str, body: bytes ) -> dict[str, str]: - """Signed headers for one attempt, stamped with the current time.""" + """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, @@ -281,7 +283,7 @@ async def verify(self, callback: Callback) -> None: message_id = VERIFICATION_ID_PREFIX + secrets.token_hex(12) try: response = await self._egress.post( - callback.url, body, self.headers(callback, message_id, body) + callback.url, body, partial(self.headers, callback, message_id, body) ) except TRANSPORT_ERRORS as error: raise CallbackError(classify(error)) from error @@ -306,7 +308,7 @@ async def deliver(self, callback: Callback, event: Event) -> Result: await self._sleep(self._policy.delay(attempt - 2, self._sample())) try: response = await self._egress.post( - callback.url, body, self.headers(callback, event.id, body) + callback.url, body, partial(self.headers, callback, event.id, body) ) except DestinationError: return Result( diff --git a/src/mcp_server_appwrite/events/egress.py b/src/mcp_server_appwrite/events/egress.py index 61c9638..8ba346a 100644 --- a/src/mcp_server_appwrite/events/egress.py +++ b/src/mcp_server_appwrite/events/egress.py @@ -6,8 +6,9 @@ alike, and it enforces the MCP Events "Webhook Security" rules: * ``https`` only. Plain ``http`` is accepted solely through the explicit - ``allow_loopback`` constructor flag, which tests use to reach a local server; - nothing reads it from the environment. + ``allow_loopback`` constructor flag, which tests use to reach a local server, + and even then only for a loopback host (``localhost`` or a loopback IP) whose + every address is loopback; nothing reads the flag from the environment. * We resolve DNS ourselves and refuse the request if *any* resolved address is not globally routable (IANA IPv4/IPv6 special-purpose registries, plus the IPv6 forms that embed an IPv4 address). @@ -17,8 +18,8 @@ possible. TLS SNI and the ``Host`` header still carry the original hostname, because httpcore derives both from the request URL, not from the address the backend connected to. -* Redirects are never followed, the whole request is bounded by one deadline, - response bodies are read up to a small cap, and concurrent requests to one +* Redirects are never followed, the whole request is bounded by one deadline + that also covers the wait for a per-host slot, response bodies are read up to a small cap, and concurrent requests to one destination host are limited so a single subscriber cannot monopolise egress. """ @@ -45,6 +46,9 @@ SECURE_SCHEME = "https" INSECURE_SCHEME = "http" +LOOPBACK_NAME = "localhost" +"""The one hostname (and, per RFC 6761, its subdomains) treated as loopback.""" + # Total wall-clock budget for one request: connect, TLS, send and read. TIMEOUT_SECONDS = 10.0 @@ -123,6 +127,21 @@ def permitted(address: IPAddress, *, allow_loopback: bool = False) -> bool: ) +def loopback_host(host: str) -> bool: + """Whether ``host`` names this machine: ``localhost``, a subdomain of it, + or a loopback IP literal. Resolution is checked separately, at connect.""" + name = host.lower().rstrip(".") + if name == LOOPBACK_NAME or name.endswith("." + LOOPBACK_NAME): + return True + try: + address = ipaddress.ip_address(name) + except ValueError: + return False + if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped: + return address.ipv4_mapped.is_loopback + return address.is_loopback + + def validate_url(url: str, *, allow_loopback: bool = False) -> httpx.URL: """Parse ``url`` and enforce the scheme and host rules for callbacks. @@ -133,11 +152,16 @@ def validate_url(url: str, *, allow_loopback: bool = False) -> httpx.URL: parsed = httpx.URL(url) except (httpx.InvalidURL, TypeError) as error: raise DestinationError("Callback URL is not a valid URL") from error - schemes = {SECURE_SCHEME, INSECURE_SCHEME} if allow_loopback else {SECURE_SCHEME} - if parsed.scheme not in schemes: + if parsed.scheme not in (SECURE_SCHEME, INSECURE_SCHEME): raise DestinationError("Callback URL must use https") if not parsed.host: raise DestinationError("Callback URL must include a host") + if parsed.scheme == INSECURE_SCHEME and not ( + allow_loopback and loopback_host(parsed.host) + ): + # Plain http is a test escape hatch for a local server only; signed + # event data never travels unencrypted to another machine. + raise DestinationError("Callback URL must use https") if parsed.userinfo: raise DestinationError("Callback URL must not include credentials") return parsed @@ -154,6 +178,12 @@ async def resolve(host: str, port: int) -> list[IPAddress]: return addresses +def _loopback(address: IPAddress) -> bool: + if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped: + return address.ipv4_mapped.is_loopback + return address.is_loopback + + class Backend(httpcore.AsyncNetworkBackend): """httpcore network backend that connects only to checked addresses. @@ -179,11 +209,17 @@ def __init__( ) async def addresses(self, host: str, port: int) -> list[IPAddress]: - """Resolve ``host`` and return its addresses if all are permitted.""" + """Resolve ``host`` and return its addresses if all are permitted. + + A loopback host (the only one plain ``http`` may reach) must resolve + to loopback alone, so ``localhost`` mapped elsewhere is refused.""" addresses = await self._resolver(host, port) if not addresses: raise DestinationError("Callback host did not resolve") + local = self._allow_loopback and loopback_host(host) for address in addresses: + if local and not _loopback(address): + raise DestinationError("Loopback host resolves to another machine") if not permitted(address, allow_loopback=self._allow_loopback): raise DestinationError("Callback host resolves to a non-public address") return addresses @@ -331,20 +367,26 @@ async def check(self, url: str) -> None: await self._backend.addresses(parsed.host, port) async def post( - self, url: str, content: bytes, headers: Mapping[str, str] + self, url: str, content: bytes, headers: Callable[[], Mapping[str, str]] ) -> Response: """POST ``content`` to ``url`` under the egress policy. + One deadline covers the wait for a per-host slot and the request, so a + request queued behind slow ones still ends within the timeout. + ``headers`` is called once the slot is held, right before sending, so + a signature timestamp is taken when the request goes out, not when it + was queued. + Raises :class:`DestinationError` for a forbidden URL or address, ``TimeoutError`` when the deadline passes, and ``httpx.TransportError`` for network and TLS failures. """ parsed = validate_url(url, allow_loopback=self._allow_loopback) - # Identity encoding keeps the body cap meaningful: a compressed reply - # cannot expand past it. - request_headers = {**headers, "Accept-Encoding": "identity"} - async with self._limiter.slot(parsed.host): - with anyio.fail_after(self._timeout): + with anyio.fail_after(self._timeout): + async with self._limiter.slot(parsed.host): + # Identity encoding keeps the body cap meaningful: a + # compressed reply cannot expand past it. + request_headers = {**headers(), "Accept-Encoding": "identity"} async with self._client.stream( "POST", parsed, content=content, headers=request_headers ) as response: diff --git a/tests/unit/test_events_egress.py b/tests/unit/test_events_egress.py index 9293a48..03b5e6d 100644 --- a/tests/unit/test_events_egress.py +++ b/tests/unit/test_events_egress.py @@ -13,11 +13,19 @@ * URL shapes the subscribe path (PR 5) will reject before any delivery. * The response body cap: delivery discards response bodies, so it is invisible on the wire. +* Plain ``http`` under ``allow_loopback``: the flag is test-only, and no test + can safely prove that a public ``http`` callback is refused by sending to + one, so the scheme and resolution rules are checked here. +* The deadline covering the wait for a per-host slot, and headers (the + signature timestamp) built only once the slot is held, at a precision the + wire's whole-second ``webhook-timestamp`` cannot show. """ import ipaddress +import time import unittest +import anyio import httpcore import httpx @@ -151,12 +159,119 @@ def test_http_only_with_explicit_loopback_flag(self) -> None: self.assertEqual(url.scheme, "http") +class LoopbackHttpTest(unittest.IsolatedAsyncioTestCase): + """``allow_loopback`` admits plain ``http`` for a loopback host only.""" + + def test_http_needs_a_loopback_host(self) -> None: + for url in ( + "http://localhost:8080/x", + "http://api.localhost/x", + "http://127.8.8.8/x", + "http://[::1]/x", + ): + with self.subTest(url=url): + egress.validate_url(url, allow_loopback=True) + for url in ( + "http://hooks.example.com/x", + "http://93.184.216.34/x", + "http://10.0.0.7/x", + "http://localhost.example.com/x", + ): + with self.subTest(url=url): + with self.assertRaises(egress.DestinationError): + egress.validate_url(url, allow_loopback=True) + with self.assertRaises(egress.DestinationError): + egress.validate_url("http://localhost/x") + egress.validate_url("https://hooks.example.com/x", allow_loopback=True) + + async def test_loopback_name_must_resolve_to_loopback(self) -> None: + loopback = ipaddress.ip_address("127.0.0.1") + for answers in ((PUBLIC,), (loopback, PUBLIC)): + with self.subTest(answers=answers): + spy = SpyNetwork() + async with egress.Egress( + resolver=static(*answers), network=spy, allow_loopback=True + ) as client: + with self.assertRaises(egress.DestinationError): + await client.post("http://localhost:9/x", b"{}", dict) + self.assertEqual(spy.targets, []) + spy = SpyNetwork() + async with egress.Egress( + resolver=static(loopback), network=spy, allow_loopback=True + ) as client: + with self.assertRaises(httpx.ConnectError): + await client.post("http://localhost:9/x", b"{}", dict) + self.assertEqual(spy.targets, [("127.0.0.1", 9)]) + + +class SlowTransport(httpx.AsyncBaseTransport): + """Answers every request after ``delay`` seconds, recording arrivals.""" + + def __init__(self, delay: float) -> None: + self.delay = delay + self.arrivals: list[float] = [] + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + self.arrivals.append(time.monotonic()) + await anyio.sleep(self.delay) + return httpx.Response(200, content=b"{}") + + +class DeadlineTest(unittest.IsolatedAsyncioTestCase): + async def test_waiting_for_a_slot_counts_against_the_deadline(self) -> None: + # Four 80 ms requests to one host, one slot, a 100 ms timeout: only + # the first fits. With the deadline started after the slot, all four + # succeeded after ~330 ms. + transport = SlowTransport(0.08) + outcomes: list[str] = [] + + async def post(client: egress.Egress) -> None: + try: + await client.post("https://hooks.example.com/", b"{}", dict) + outcomes.append("ok") + except TimeoutError: + outcomes.append("timeout") + + started = time.monotonic() + async with egress.Egress( + concurrency=1, timeout=0.1, transport=transport + ) as client: + async with anyio.create_task_group() as group: + for _ in range(4): + group.start_soon(post, client) + elapsed = time.monotonic() - started + self.assertEqual(sorted(outcomes), ["ok", "timeout", "timeout", "timeout"]) + self.assertLess(elapsed, 0.2) + + async def test_headers_are_built_once_the_slot_is_held(self) -> None: + transport = SlowTransport(0.08) + built: list[float] = [] + + def headers() -> dict[str, str]: + built.append(time.monotonic()) + return {} + + async with egress.Egress( + concurrency=1, timeout=1.0, transport=transport + ) as client: + async with anyio.create_task_group() as group: + for _ in range(4): + group.start_soon( + client.post, "https://hooks.example.com/", b"{}", headers + ) + self.assertEqual(len(built), 4) + for stamped, sent in zip(built, transport.arrivals, strict=True): + self.assertLess(sent - stamped, 0.03) + # Queued requests were stamped as they went out, not when queued. + self.assertGreater(built[-1] - built[0], 0.2) + + class ConnectTargetTest(unittest.IsolatedAsyncioTestCase): async def test_connects_to_the_checked_address(self) -> None: spy = SpyNetwork() async with egress.Egress(resolver=static(PUBLIC), network=spy) as client: with self.assertRaises(httpx.ConnectError): - await client.post("https://hooks.example.com/x", b"{}", {}) + await client.post("https://hooks.example.com/x", b"{}", dict) self.assertEqual(spy.targets, [(str(PUBLIC), 443)]) async def test_any_private_answer_rejects_the_host(self) -> None: @@ -164,7 +279,7 @@ async def test_any_private_answer_rejects_the_host(self) -> None: resolver = static(PUBLIC, PRIVATE) async with egress.Egress(resolver=resolver, network=spy) as client: with self.assertRaises(egress.DestinationError): - await client.post("https://hooks.example.com/x", b"{}", {}) + await client.post("https://hooks.example.com/x", b"{}", dict) self.assertEqual(spy.targets, []) async def test_dns_rebinding_is_rejected_at_connect(self) -> None: @@ -177,7 +292,7 @@ async def rebinding(host: str, port: int): async with egress.Egress(resolver=rebinding, network=spy) as client: await client.check("https://rebind.example.com/hook") with self.assertRaises(egress.DestinationError): - await client.post("https://rebind.example.com/hook", b"{}", {}) + await client.post("https://rebind.example.com/hook", b"{}", dict) self.assertEqual(spy.targets, []) async def test_check_rejects_private_answers(self) -> None: @@ -188,7 +303,7 @@ async def test_check_rejects_private_answers(self) -> None: async def test_unresolvable_host_is_rejected(self) -> None: async with egress.Egress(resolver=static(), network=SpyNetwork()) as client: with self.assertRaises(egress.DestinationError): - await client.post("https://nowhere.example.com/", b"{}", {}) + await client.post("https://nowhere.example.com/", b"{}", dict) class BodyCapTest(unittest.IsolatedAsyncioTestCase): @@ -197,7 +312,7 @@ async def test_response_body_is_capped(self) -> None: lambda request: httpx.Response(200, content=b"x" * 100_000) ) async with egress.Egress(body_limit=1024, transport=transport) as client: - response = await client.post("https://hooks.example.com/", b"{}", {}) + response = await client.post("https://hooks.example.com/", b"{}", dict) self.assertEqual(len(response.body), 1024) From 656ac206794db282d6a18836b9e7163f6e830bbe Mon Sep 17 00:00:00 2001 From: Chirag Aggarwal Date: Sun, 11 Oct 2026 22:12:30 +0100 Subject: [PATCH 4/4] refactor: deliver with one secret and keep it out of repr A subscription record carries one delivery.secret, so a Callback holds one secret and signs once; the multi-secret path had nothing left to feed it. The secret is no longer part of a Callback's repr. --- src/mcp_server_appwrite/events/delivery.py | 36 +++++++++------------- tests/unit/test_events_delivery.py | 26 ++++++++++------ 2 files changed, 30 insertions(+), 32 deletions(-) diff --git a/src/mcp_server_appwrite/events/delivery.py b/src/mcp_server_appwrite/events/delivery.py index f996823..77d4002 100644 --- a/src/mcp_server_appwrite/events/delivery.py +++ b/src/mcp_server_appwrite/events/delivery.py @@ -5,8 +5,8 @@ every POST carries ``webhook-id``, ``webhook-timestamp`` (Unix seconds) and ``webhook-signature: v1,``, where ``key`` is the base64-decoded part of the client's ``whsec_`` secret. -During rotation the header carries one space-separated signature per secret so -the receiver can verify with either. MCP adds ``X-MCP-Subscription-Id``. +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: @@ -36,7 +36,7 @@ import secrets import ssl import time -from collections.abc import Awaitable, Callable, Mapping, Sequence +from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field from enum import StrEnum from functools import partial @@ -125,14 +125,11 @@ def decode_secret(secret: str) -> bytes: return key -def sign(keys: Sequence[bytes], message_id: str, timestamp: int, body: bytes) -> str: - """Standard Webhooks ``webhook-signature`` value, one entry per 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 - return " ".join( - f"{SIGNATURE_VERSION}," - + base64.b64encode(hmac.new(key, content, hashlib.sha256).digest()).decode() - for key in keys - ) + digest = hmac.new(key, content, hashlib.sha256).digest() + return f"{SIGNATURE_VERSION},{base64.b64encode(digest).decode()}" def encode(payload: Mapping[str, Any]) -> bytes: @@ -142,22 +139,17 @@ def encode(payload: Mapping[str, Any]) -> bytes: @dataclass(frozen=True) class Callback: - """Where and how to deliver for one subscription. - - ``secrets`` holds every active ``whsec_`` secret, newest first; more than - one during a rotation grace window. - """ + """Where and how to deliver for one subscription.""" url: str subscription: str - secrets: tuple[str, ...] - keys: tuple[bytes, ...] = field(init=False, repr=False, compare=False) + 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: - if not self.secrets: - raise InvalidSecretError("At least one secret is required") - keys = tuple(decode_secret(secret) for secret in self.secrets) - object.__setattr__(self, "keys", keys) + object.__setattr__(self, "key", decode_secret(self.secret)) @dataclass(frozen=True) @@ -272,7 +264,7 @@ def headers( "Content-Type": CONTENT_TYPE, ID_HEADER: message_id, TIMESTAMP_HEADER: str(timestamp), - SIGNATURE_HEADER: sign(callback.keys, message_id, timestamp, body), + SIGNATURE_HEADER: sign(callback.key, message_id, timestamp, body), SUBSCRIPTION_HEADER: callback.subscription, } diff --git a/tests/unit/test_events_delivery.py b/tests/unit/test_events_delivery.py index 7a2daf0..9958599 100644 --- a/tests/unit/test_events_delivery.py +++ b/tests/unit/test_events_delivery.py @@ -2,15 +2,16 @@ Signing (checked by the official ``standardwebhooks`` library), the delivery body and headers, re-signed retries with a short injected policy, 410/413 not -retried, dual signing during secret rotation and the 256 KiB cap are covered -end to end through the ingress in ``tests/e2e/test_events_ingress.py`` (#134). -What stays here: +retried and the 256 KiB cap are covered end to end through the ingress in +``tests/e2e/test_events_ingress.py`` (#134). What stays here: * ``whsec_`` secret validation and the verification handshake: only ``events/subscribe`` (PR 5 of #127) calls them. Replace these with its e2e flow. * The production retry schedule, whose real-time waits (30 s, 2 min, 8 min) are too long to run end to end. +* That a callback's ``repr`` never shows its secret: nothing on the wire + shows a ``repr``. """ import base64 @@ -103,16 +104,21 @@ def test_rejects_invalid_secrets(self) -> None: with self.assertRaises(delivery.InvalidSecretError): delivery.decode_secret(secret) # type: ignore[arg-type] - def test_callback_requires_valid_secrets(self) -> None: + def test_callback_requires_a_valid_secret(self) -> None: with self.assertRaises(delivery.InvalidSecretError): - delivery.Callback(URL, SUBSCRIPTION, ()) - with self.assertRaises(delivery.InvalidSecretError): - delivery.Callback(URL, SUBSCRIPTION, ("whsec_short",)) + delivery.Callback(URL, SUBSCRIPTION, "whsec_short") + + +class CallbackTest(unittest.TestCase): + def test_repr_hides_the_secret(self) -> None: + callback = delivery.Callback(URL, SUBSCRIPTION, SECRET) + self.assertNotIn(SECRET.removeprefix("whsec_"), repr(callback)) + self.assertIn(SUBSCRIPTION, repr(callback)) class VerificationTest(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: - self.callback = delivery.Callback(URL, SUBSCRIPTION, (SECRET,)) + self.callback = delivery.Callback(URL, SUBSCRIPTION, SECRET) self.clock = Clock() async def run_verify(self, handler) -> None: @@ -202,7 +208,7 @@ async def test_transport_failures_are_categorized(self) -> None: async def test_forbidden_destination_is_connection_refused(self) -> None: self.callback = delivery.Callback( - "http://hooks.example.com/", SUBSCRIPTION, (SECRET,) + "http://hooks.example.com/", SUBSCRIPTION, SECRET ) recorder = Recorder(httpx.Response(200)) await self.assert_reason(recorder, delivery.Reason.CONNECTION_REFUSED) @@ -214,7 +220,7 @@ class RetryScheduleTest(unittest.IsolatedAsyncioTestCase): which runs a short injected policy instead.""" def setUp(self) -> None: - self.callback = delivery.Callback(URL, SUBSCRIPTION, (SECRET,)) + self.callback = delivery.Callback(URL, SUBSCRIPTION, SECRET) self.clock = Clock() self.sleeps: list[float] = []