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