diff --git a/AGENTS.md b/AGENTS.md index e37506c..77b0ba2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,7 +34,7 @@ Source lives in `src/mcp_server_appwrite/`: | `docs_search.py` | In-process semantic docs search (`appwrite_search_docs`) over a prebuilt index. | | `docs_source.py` | Fetches the published docs (`/docs/llms.txt` + per-page `.md` exports on appwrite.io) and chunks them for the index build. | | `docs_index.py` | Writes the index artifact deterministically and atomically: page and chunk hashes, reuse of cached chunk vectors, build lock, and the added/changed/removed page report. | -| `events/` | MCP Events (behind `flags.EVENTS`, HTTP only): `catalog.py` (event catalog), `protocol.py` (capability + `events/list`), `envelope.py` (subscription state sealed with the `mcp-event-subscriptions` JWE profile), `egress.py` + `delivery.py` (SSRF-safe signed delivery), `projection.py` + `ingress.py` (Appwrite webhook ingress at `/appwrite/webhooks/{id}`). See [docs/events.md](docs/events.md). | +| `events/` | MCP Events (behind `flags.EVENTS`, HTTP only): `catalog.py` (event catalog), `protocol.py` (capability, `events/list` and the subscribe methods), `subscriptions.py` + `webhooks.py` (`events/subscribe` / `events/unsubscribe`: one managed Appwrite webhook per subscription), `envelope.py` (subscription state sealed with the `mcp-event-subscriptions` JWE profile), `egress.py` + `delivery.py` (SSRF-safe signed delivery), `projection.py` + `ingress.py` (Appwrite webhook ingress at `/appwrite/webhooks/{id}`). See [docs/events.md](docs/events.md). | | `telemetry.py` | OpenTelemetry metrics layer (OTLP/HTTP). No-op unless an OTLP endpoint is configured and the transport is `http`. | | `data/` | Committed docs index artifact (`docs_index.npz`: vectors, chunk map, chunk hashes, embedded `meta.json`), shipped in the wheel/image. | diff --git a/README.md b/README.md index 1f446da..1621ada 100644 --- a/README.md +++ b/README.md @@ -321,7 +321,7 @@ for per-client setup. - [Documentation search](docs/documentation-search.md) — the in-process `appwrite_search_docs` tool and how to rebuild its index. - [Events](docs/events.md) — MCP Events: webhook subscriptions to changes in - your Appwrite project (in progress, behind a flag). + your Appwrite project (in development, not available yet). - [Self-hosted Appwrite](docs/self-hosted.md) — run the server locally with a project API key. - [Local development](docs/development.md) — running, testing, and debugging the diff --git a/docs/development.md b/docs/development.md index 86db250..56f13c2 100644 --- a/docs/development.md +++ b/docs/development.md @@ -41,8 +41,9 @@ docker compose up --build > [`mcp-subscription+jwe`](https://github.com/ChiragAgg5k/mcp-event-subscriptions/blob/main/profiles/encrypted-envelope.md) > envelope, with the keys in `MCP_EVENTS_SEALING_KEYS` (`: random bytes>`, comma separated, first one active). Generate a key with -> `openssl rand -base64 32`. To rotate, put a new key first and remove the old -> one once the longest subscription TTL (24h) has passed. +> `openssl rand -base64 32`. To rotate, add the new key last and roll that out, +> then move it first, then remove the old one 25 hours after that reached every +> replica (see "Key rotation" in [events.md](events.md)). **`uv` directly — HTTP:** diff --git a/docs/events.md b/docs/events.md index a57de98..9f43003 100644 --- a/docs/events.md +++ b/docs/events.md @@ -1,10 +1,10 @@ # Events -> **In progress.** Events are being built behind the `--events` flag -> ([#127](https://github.com/appwrite/mcp/issues/127)). Today the server -> advertises the capability, lists the catalog, and receives Appwrite webhooks -> and forwards them as events; `events/subscribe` and `events/unsubscribe` are -> not available yet. +> **Not a supported feature yet.** Events sit behind the `--events` testing +> flag ([#127](https://github.com/appwrite/mcp/issues/127)), which is off in +> production. They cannot work on Appwrite Cloud until +> [appwrite/appwrite#14293](https://github.com/appwrite/appwrite/issues/14293) +> lets a webhook's `authPassword` hold the sealed subscription. [MCP Events](https://developers.openai.com/plugins/build/mcp-events) let an MCP client such as ChatGPT subscribe to things that happen in an Appwrite project @@ -12,19 +12,13 @@ and receive a signed webhook when they occur, for example "when a function execution fails in project X, read the logs and open a fix". Events are served only by the hosted HTTP server (`mcp.appwrite.io`), and v1 -delivers only by webhook. Each subscription will be backed by one Appwrite -webhook in the subscriber's project, so the MCP server keeps no subscription -store. +delivers only by webhook. Each subscription is backed by one Appwrite webhook in +the subscriber's project, so the MCP server keeps no subscription store. -## Enabling +## Testing flag -```bash -MCP_PUBLIC_URL=http://localhost:8000 \ - uv run mcp-server-appwrite --transport http --events 1 -``` - -`MCP_EVENTS=1` works too. See [flags.md](flags.md#--events--mcp-events-in-progress) -for quick checks with `curl`. +For testers only: [flags.md](flags.md#--events--mcp-events-in-progress) shows +how to turn the flag on locally and check it with `curl`. With the flag on, the server advertises the capability in both shapes that clients read today: @@ -77,6 +71,178 @@ they are kept behind constants in | Unsupported | `-32014` | | | CallbackEndpointError | `-32015` | `reason`: `challenge_failed`, `timeout`, `connection_refused`, `tls_error`, `http_4xx`, `http_5xx` | +Failures on our side, including Appwrite answering `5xx`, are `-32603`. + +## Subscribe and unsubscribe + +``` +events/subscribe {name, arguments, delivery: {mode: "webhook", url, secret}, cursor, ttlMs?, maxAgeMs?} + -> {id, refreshBefore, cursor: null, truncated: false} +events/unsubscribe {name, arguments, delivery: {url}} + -> {} +``` + +The server keeps nothing. A subscription is one Appwrite webhook in the +subscriber's project, written with the subscriber's own OAuth token, and the +subscription id is that webhook's `$id`: `sub_` + 32 hex characters of a +SHA-256 over the principal (OAuth `iss`, `sub`, `client_id`), the callback URL, +the event name and the arguments. Subscribing again with the same values is a +refresh of the same webhook, and two users subscribing the same URL get two +subscriptions. + +`events/subscribe`, in order: + +1. Looks the event up (`-32011`, `data.kind = "event"`) and checks its + arguments (`-32602`). `delivery.mode` must be `webhook` (`-32014`). The + callback must be `https`, without credentials, and resolve only to public + addresses (`-32602`). The secret must be `whsec_` + base64 of 24 to 64 bytes + (`-32602`). +2. Needs a user behind the token: no `sub` is `-32012`. No token at all never + reaches JSON-RPC (HTTP `401`). A token whose granted scopes include none of + `project:webhooks.write`, `project:all` or `all` is `-32012` here, before + the callback gets any request (listing webhooks below proves only read + access, and Appwrite has no side-effect-free write check). +3. Grants a TTL and seals the subscription (see [Envelope](#envelope)). A + callback URL too long for the envelope is `-32602`, and so is a delivery the + subscription record refuses (a URL with a fragment, a secret whose base64 + is not canonical). +4. Reads the event's resource in the project with the caller's token: + + | Event | Read | Scope | + | --- | --- | --- | + | `functions.execution.failed`, `functions.deployment.completed` | `GET /functions/{function_id}` | `functions.read` | + | `sites.deployment.completed` | `GET /sites/{site_id}` | `sites.read` | + | `tablesdb.row.created` | `GET /tablesdb/{database_id}/tables/{table_id}` | `tables.read` | + | `storage.file.created` | `GET /storage/buckets/{bucket_id}` | `buckets.read` | + | `users.user.created` | `GET /users` with `limit(1)` | `users.read` | + + Appwrite checks access to the project before it looks the resource up. So a + `404` for the resource means the caller is in the project and it does not + exist: NotFound with `data.kind = "resource"`. A `401` or `403` (no access + to the project, or a missing scope) is Forbidden (`-32012`), and so is a + project that does not exist, so project ids cannot be probed. Every refresh + repeats the read, so revoked access ends a subscription at its next refresh + at the latest. +5. Lists the project's webhooks, which also proves the webhook scopes, and + deletes this principal's long-expired subscriptions (see below). +6. Runs the verification handshake: a signed + `{"type":"verification","challenge":…}` that the callback must echo in a + `2xx`. A failure is `-32015` with the reason. The handshake runs on every + subscribe and refresh: the server has no verification cache, and it cannot + read the previous subscription back from Appwrite to know whether the + callback or secret changed. One extra signed POST per refresh is cheap, and + it proves the callback still wants deliveries under the secret it is about + to get. +7. Creates the webhook, or rewrites it on a refresh (which also re-enables a + webhook Appwrite paused after failed deliveries). + +`events/unsubscribe` recomputes the id from the caller and the request and +deletes that webhook if it is a managed one. It returns `{}` when there was +nothing to delete, so it is idempotent, and it can only ever reach the caller's +own subscriptions because the id includes the principal. A project that does +not exist is `-32012`, as on subscribe, so project ids cannot be probed. + +### TTL + +| Suggestion (`ttlMs`) | Grant | +| --- | --- | +| absent or `null` | 1 hour. `null` asks for no expiry, which needs durable storage this server does not have. | +| below 5 minutes | 5 minutes (floor) | +| 5 minutes to 24 hours | as suggested | +| above 24 hours | 24 hours (cap) | + +The subscription expires when the grant runs out. `refreshBefore` is a tenth of +the grant earlier (6 minutes for the default hour, 30 seconds at the floor), so +a slow refresh never lets deliveries lapse. The cap also bounds how long revoked +access can keep receiving events. `refreshBefore` is never `null`. `cursor` is +always `null` and `truncated` always `false`: Appwrite webhooks cannot replay, +so `maxAgeMs` is accepted and has no effect. + +### The managed webhook + +| Field | Value | +| --- | --- | +| `$id` | The subscription id | +| `name` | `MCP events \| \| until \| owner ` | +| `url` | `{MCP_PUBLIC_URL}/appwrite/webhooks/{id}` | +| `events` | The event's Appwrite patterns | +| `tls` | `true` | +| `authUsername` | `mcp-events`, a fixed non-secret value. Appwrite sends Basic auth only when both username and password are set. | +| `authPassword` | The sealed envelope, and nothing else | +| `secret` | The derived signing key for `X-Appwrite-Webhook-Signature` | + +Appwrite never returns `authPassword`, so the envelope cannot be read back. The +name carries the two facts cleanup needs instead: when the subscription expires +and whose it is. It is ASCII because Appwrite sends it back in the +`X-Appwrite-Webhook-Name` header. A webhook counts as managed only if its id is +a subscription id and its name parses as this label. Anything else in the +project is never changed or deleted, and a webhook that has a subscription's id +but a different name (renamed in the Console) makes subscribe fail with +`-32012` rather than be overwritten. + +Before writing, subscribe deletes the caller's managed webhooks whose label says +they expired more than 10 minutes ago. It only touches the caller's own (matched +by the owner tag) and only long-expired ones, so a tag collision between two +users could at worst remove an expired webhook whose deliveries the ingress +already drops. Appwrite has no conditional delete, so a late refresh of that +same subscription could race the cleanup. The grace period keeps a subscription +that is still being refreshed out of the candidates, and each candidate is read +again right before the delete and kept if its name or `$updatedAt` changed +since the listing. What remains is a refresh landing between that read and the +delete, of a subscription left expired for over 10 minutes; its next refresh +creates the webhook again. Expired +subscriptions of a user who never comes back stay until that user subscribes +again in the project, or the webhook is deleted in the Console. Appwrite keeps +delivering to them, and the ingress answers `200` and drops them. + +### Secret rotation + +A refresh with a new `delivery.secret` replaces the secret. The previous one +cannot be kept for dual signing, because it exists only in the old envelope, +which Appwrite never returns. Deliveries already in flight were opened from the +old envelope, so their retries keep signing with the old secret. Receivers +should accept both secrets for a short while after rotating, as Standard +Webhooks receivers do. + +### Failure handling + +| Appwrite answer | Result | +| --- | --- | +| `401 general_unauthorized_scope` on `/webhooks` | `-32012`, naming `project:webhooks.read` and `project:webhooks.write` | +| `401` / `403` otherwise on `/webhooks` | `-32012` | +| `404 project_not_found` on `/webhooks` (unsubscribe from a missing project) | `-32012` | +| `403 additional_resource_not_allowed` (plan limit) | `-32013`, `data: {"limit": "webhooks", "max": }` | +| `5xx`, or anything else | `-32603` | + +Creating a webhook is a single call. A refresh sets the signing key first, +because the update endpoint does not take it, and then rewrites the webhook +with the new envelope. If that second call fails, the webhook keeps its previous +envelope with the new signing key. The ingress accepts the signing key of every +ring key, and the previous envelope's sealing key stays in the ring until that +envelope expires, so deliveries continue; the error (`-32603`) tells the client +to retry. In the other order, a failure could leave a new envelope with a +signing key derived from a sealing key that is retired while the envelope is +still live, and deliveries would be rejected until the next refresh. + +### Free plan + +Appwrite Cloud allows 2 webhooks per project on Free, 50 on Start and unlimited +on Pro and Scale, and every subscription uses one. At the limit, subscribe +returns `-32013` and tells Free users to upgrade to Pro or unsubscribe from +another event. Cleanup runs before the create, so expired subscriptions of the +caller never hold a slot. + +### OAuth scopes + +| Scope | For | +| --- | --- | +| `project:webhooks.read` | Listing the project's webhooks (refresh, cleanup, unsubscribe) | +| `project:webhooks.write` | Creating, rewriting and deleting the subscription's webhook | +| `project:functions.read`, `project:sites.read`, `project:tables.read`, `project:buckets.read`, `project:users.read` | The authorization read for each event (see above) | + +The server already advertises these. A user can narrow them at consent, and +the error message names the scope to grant. + ## Ingress Each subscription is one Appwrite webhook in the subscriber's project. It points @@ -222,14 +388,29 @@ request that fails authentication gets `401`. A legitimate webhook cannot get a `401` in normal operation: -- **Key rotation.** A new sealing key goes first in `MCP_EVENTS_SEALING_KEYS` - and the old one stays until the longest subscription TTL has passed, so - webhooks written with the old key still verify and open. Replicas that - refresh the same subscription during a rolling rotation can interleave their - writes and leave an envelope from one key next to a signing secret from the - other; that still verifies, because the signature is checked against every - key in the ring. After the old key is removed, only expired subscriptions - still name it. Their webhooks are orphans awaiting cleanup, so they get `200` (`retired_key`) instead of being paused. +- **Key rotation.** Rotate in three deploys, each rolled out to every replica + before the next: + 1. Add the new key *after* the current one (`k1:…,k2:…`). Every replica can + now open `k2` envelopes, but `k1` still seals. + 2. Move the new key first (`k2:…,k1:…`). Refreshes now seal with `k2`. A + replica still on step 1 opens them, so no delivery that reaches it is + dropped as `retired_key`. + 3. Once 25 hours have passed since step 2 reached every replica, remove the + old key (`k2:…`). + + Sealing with a key that some replica does not have yet would make that + replica drop the subscription's deliveries as `retired_key`. Replicas that + refresh the same subscription during step 2 can interleave their writes and + leave an envelope from one key next to a signing secret from the other; that + still verifies, because the signature is checked against every key in the + ring. Only a replica still on step 1 writes an old-key signing secret, so + every envelope paired with one was written before step 2 finished, give or + take one refresh, and expires within the longest TTL (24 hours) after that; + the extra hour covers refreshes in flight. After the old key is removed, only + expired subscriptions still name it or hold a secret derived from it. Their + webhooks are orphans awaiting cleanup, so they get `200` (`retired_key`, or + `expired` because expiry is checked before the signature) instead of being + paused. - **Public URL.** The signature covers the registered URL, and every envelope is bound to `MCP_PUBLIC_URL`. Changing it breaks every subscription until it refreshes, so keep it fixed while subscriptions exist. @@ -248,6 +429,13 @@ lost: delivery is at most once, like Appwrite webhooks themselves. | --- | --- | | `mcp.events.ingress` | `outcome` (`accepted`, `dropped`, `rejected`), `reason`, `event` | | `mcp.events.deliveries` | `event`, `outcome` (`delivered`, `rejected`, `abandoned`, `too_large`, `error`), `reason` | +| `mcp.events.subscriptions` | `operation` (`subscribe`, `unsubscribe`, `cleanup`), `outcome` (`created`, `refreshed`, `removed`, `absent`, `error`), `event`, `reason` (the JSON-RPC code of an error) | + +`events/subscribe` and `events/unsubscribe` are also counted in +`mcp.messages.received` and `mcp.jsonrpc.errors`, like `tools/call`. Errors on +our side (`-32603`) go to Sentry with the same tags as `tools/call` plus +`event.name` and `appwrite.project_id`. Errors the caller caused are counted but +not reported. ## Testing @@ -261,12 +449,16 @@ from CI. The suite needs no credentials and runs on every PR: uv run --group e2e python -m unittest discover -s tests/e2e -v ``` -The ingress flows also play the parties around the server on localhost: -Appwrite's webhooks worker signs and posts recorded-shape bodies -(`tests/e2e/fixtures/appwrite/`) to the real ingress, and deliveries go through -the real dispatcher and egress to an HTTPS receiver that checks each one with -the official `standardwebhooks` library. Counters are read from the server's -real OTLP export. +The flows also play the parties around the server on localhost. An Appwrite +REST emulator (`support.Cloud`) serves the resource reads and `/v1/webhooks` +with Appwrite's scopes, error types and plan limit, so `events/subscribe` and +`events/unsubscribe` write real webhooks into it. Appwrite's webhooks worker +fires those stored webhooks, or hand-built ones, with recorded-shape bodies +(`tests/e2e/fixtures/appwrite/`) at the real ingress, and deliveries go through +the real dispatcher and egress to an HTTPS receiver that answers the +verification handshake and checks every request with the official +`standardwebhooks` library. Counters are read from the server's real OTLP +export. A live end-to-end run against Cloud staging (real OAuth tokens, real Appwrite webhooks) is planned once diff --git a/docs/flags.md b/docs/flags.md index 8e67e3d..d776d47 100644 --- a/docs/flags.md +++ b/docs/flags.md @@ -85,11 +85,14 @@ token exchange against the real authorization server. ### `--events` — MCP Events (in progress) Turns on [MCP Events](events.md): `server/discover` (and legacy `initialize`) -advertise the events capability, the server answers `events/list` with the -event catalog, and the Appwrite webhook ingress is mounted at -`/appwrite/webhooks/{id}`. It is off by default while the feature is built -([#127](https://github.com/appwrite/mcp/issues/127)); `events/subscribe` is not -available yet. Any of `1`, `true`, `yes` or `on` enables it. With the flag on, +advertise the events capability, the server answers `events/list`, +`events/subscribe` and `events/unsubscribe`, and the Appwrite webhook ingress +is mounted at `/appwrite/webhooks/{id}`. It is a testing flag, off by default +and in production, while the feature is built +([#127](https://github.com/appwrite/mcp/issues/127)): subscriptions cannot work +on Appwrite Cloud until +[appwrite/appwrite#14293](https://github.com/appwrite/appwrite/issues/14293) +ships. Any of `1`, `true`, `yes` or `on` enables it. With the flag on, `MCP_EVENTS_SEALING_KEYS` is required and the server refuses to start without it. @@ -128,5 +131,6 @@ curl -s -X POST http://localhost:8000/appwrite/webhooks/sub_0123456789abcdef0123 # {"status":"rejected","reason":"credentials"} ``` -With the flag off, neither capability appears, `events/list` returns `-32601` -(method not found) and the ingress route returns `404`. +With the flag off, neither capability appears, `events/list`, +`events/subscribe` and `events/unsubscribe` return `-32601` (method not found) +and the ingress route returns `404`. diff --git a/src/mcp_server_appwrite/events/catalog.py b/src/mcp_server_appwrite/events/catalog.py index e3ab9b7..fa346fc 100644 --- a/src/mcp_server_appwrite/events/catalog.py +++ b/src/mcp_server_appwrite/events/catalog.py @@ -14,6 +14,9 @@ with ``appwrite_call_tool``. * ``status`` — the payload status filter. Appwrite webhooks filter only by event name, so the ingress applies this one before delivering. +* ``access`` — the Appwrite read that proves a caller may watch the event. + ``events/subscribe`` makes it with the caller's own token on every subscribe + and refresh, so a refresh doubles as a periodic access check. The JSON Schemas served by ``events/list`` are derived from those fields, and :meth:`Event.validate` enforces exactly what the input schema promises. @@ -28,6 +31,8 @@ from types import MappingProxyType from typing import Any +from appwrite_console.query import Query + from .errors import EventsError ID_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9_-]{0,35}$" @@ -139,6 +144,27 @@ def matches(self, status: object, arguments: Mapping[str, Any]) -> bool: return wanted is None or wanted == status +@dataclass(frozen=True) +class Access: + """The least-privilege Appwrite read that proves access to an event's + resource: ``GET {path}`` in the subscribed project, which needs ``scope``. + + ``path`` and ``resource`` (how errors name what was read) have + ``{argument}`` placeholders like the pattern templates; ``params`` go in + the query string.""" + + path: str + scope: str + resource: str + params: Mapping[str, Any] = MappingProxyType({}) + + def resolve(self, arguments: Mapping[str, str]) -> str: + return self.path.format(**arguments) + + def describe(self, arguments: Mapping[str, str]) -> str: + return self.resource.format(**arguments) + + PROJECT = Argument("project_id", "Appwrite project ID to watch.") """Required by every event; it is never part of the Appwrite pattern because project webhooks are already scoped to one project.""" @@ -152,6 +178,7 @@ class Event: description: str templates: tuple[str, ...] payload: tuple[PayloadField, ...] + access: Access arguments: tuple[Argument, ...] = () status: StatusFilter | None = None @@ -221,6 +248,10 @@ def _timestamp(name: str, description: str) -> PayloadField: choices=(Status.READY, Status.FAILED), ) +_FUNCTION = Access( + "/functions/{function_id}", "functions.read", "function {function_id}" +) + _DEPLOYMENT_FILTER = StatusFilter( accepted=(Status.READY, Status.FAILED), argument=_DEPLOYMENT_STATUS.name ) @@ -277,6 +308,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: ), _timestamp("created_at", "When the execution started."), ), + access=_FUNCTION, status=StatusFilter(accepted=(Status.FAILED,)), ), Event( @@ -288,6 +320,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: ), templates=("functions.{function_id}.deployments.*.update",), payload=_deployment_payload("function_id", "function"), + access=_FUNCTION, status=_DEPLOYMENT_FILTER, ), Event( @@ -299,6 +332,7 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: ), templates=("sites.{site_id}.deployments.*.update",), payload=_deployment_payload("site_id", "site"), + access=Access("/sites/{site_id}", "sites.read", "site {site_id}"), status=_DEPLOYMENT_FILTER, ), Event( @@ -315,6 +349,11 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: PayloadField("row_id", "string", "ID of the new row."), _timestamp("created_at", "When the row was created."), ), + access=Access( + "/tablesdb/{database_id}/tables/{table_id}", + "tables.read", + "table {table_id} in database {database_id}", + ), ), Event( name="storage.file.created", @@ -332,6 +371,9 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: ), _timestamp("created_at", "When the file was uploaded."), ), + access=Access( + "/storage/buckets/{bucket_id}", "buckets.read", "bucket {bucket_id}" + ), ), Event( name="users.user.created", @@ -341,6 +383,14 @@ def _deployment_payload(resource: str, kind: str) -> tuple[PayloadField, ...]: PayloadField("user_id", "string", "ID of the new user."), _timestamp("created_at", "When the user was created."), ), + # Listing one user is the narrowest read that proves the caller may + # see users; the agent needs users.read to act on the event anyway. + access=Access( + "/users", + "users.read", + "users", + MappingProxyType({"queries": [Query.limit(1)], "total": False}), + ), ), ) """The v1 catalog, in the order ``events/list`` serves it.""" diff --git a/src/mcp_server_appwrite/events/envelope.py b/src/mcp_server_appwrite/events/envelope.py index 69d68c0..4930971 100644 --- a/src/mcp_server_appwrite/events/envelope.py +++ b/src/mcp_server_appwrite/events/envelope.py @@ -28,8 +28,10 @@ Sealing keys come from ``MCP_EVENTS_SEALING_KEYS`` (``:``, comma separated, first one active). Every key in the ring opens envelopes, so a -rotation adds the new key in front and drops the old one once every subscription -has refreshed (the TTL cap bounds how long that takes). +rotation first adds the new key behind the active one on every replica, then +moves it in front, and drops the old one once every subscription has refreshed +(the TTL cap bounds how long that takes). Sealing with a key some replica does +not hold yet would make that replica drop the deliveries as ``retired_key``. """ from __future__ import annotations diff --git a/src/mcp_server_appwrite/events/errors.py b/src/mcp_server_appwrite/events/errors.py index 720d19c..e4aeca9 100644 --- a/src/mcp_server_appwrite/events/errors.py +++ b/src/mcp_server_appwrite/events/errors.py @@ -27,13 +27,15 @@ from mcp import MCPError -# Malformed arguments, callback URL or secret: the JSON-RPC code, unchanged by -# SEP-3415. Re-exported so callers take every events code from this module. -from mcp.types import INVALID_PARAMS +# Malformed arguments, callback URL or secret, and failures on our side (such +# as Appwrite answering 5xx): the JSON-RPC codes, unchanged by SEP-3415. +# Re-exported so callers take every events code from this module. +from mcp.types import INTERNAL_ERROR, INVALID_PARAMS __all__ = [ "CALLBACK_ENDPOINT", "FORBIDDEN", + "INTERNAL_ERROR", "INVALID_PARAMS", "NOT_FOUND", "RESOURCE_EXHAUSTED", @@ -101,3 +103,14 @@ def unsupported(cls, message: str) -> EventsError: @classmethod def callback_endpoint(cls, message: str, *, reason: CallbackFailure) -> EventsError: return cls(CALLBACK_ENDPOINT, message, {"reason": reason.value}) + + @classmethod + def internal(cls, message: str) -> EventsError: + return cls(INTERNAL_ERROR, message) + + @property + def expected(self) -> bool: + """Whether the caller caused this error (bad input, no access, a quota, + an unreachable callback) rather than a failure on our side. Only + unexpected errors are reported to Sentry.""" + return self.code != INTERNAL_ERROR diff --git a/src/mcp_server_appwrite/events/ingress.py b/src/mcp_server_appwrite/events/ingress.py index c85c243..e186024 100644 --- a/src/mcp_server_appwrite/events/ingress.py +++ b/src/mcp_server_appwrite/events/ingress.py @@ -259,6 +259,23 @@ def from_env(cls) -> Ingress: keys are missing, so a misconfigured server fails at startup.""" return cls(Keyring.from_env(), Egress(), base=public_base_url()) + @property + def keyring(self) -> Keyring: + return self._keyring + + @property + def egress(self) -> Egress: + return self._egress + + @property + def dispatcher(self) -> Dispatcher: + return self._dispatcher + + @property + def server(self) -> str: + """This deployment as envelopes are bound to it: ``MCP_PUBLIC_URL``.""" + return self._base + @property def route(self) -> Route: return Route(PATH, endpoint=self.handle, methods=["POST"]) diff --git a/src/mcp_server_appwrite/events/protocol.py b/src/mcp_server_appwrite/events/protocol.py index 0ad59c2..f9e9ac0 100644 --- a/src/mcp_server_appwrite/events/protocol.py +++ b/src/mcp_server_appwrite/events/protocol.py @@ -10,9 +10,13 @@ today: ``capabilities.events = {}`` for ChatGPT and ``capabilities.extensions["io.modelcontextprotocol/events"]`` for SEP-3415. Delete the middleware once the SDK models the capability. -* **Methods.** ``events/list`` is registered with ``Server.add_request_handler``; - the SDK adds ``resultType: "complete"`` to custom-method results on the - 2026-07-28 protocol. +* **Methods.** ``events/list``, ``events/subscribe`` and ``events/unsubscribe`` + are registered with ``Server.add_request_handler``; the SDK adds + ``resultType: "complete"`` to custom-method results on the 2026-07-28 + protocol. Subscribe and unsubscribe are counted like ``tools/call`` + (``mcp.messages.received`` / ``mcp.jsonrpc.errors``), and failures on our + side go to Sentry with the same tags; errors the caller caused (bad input, + no access, a quota, an unreachable callback) are counted but not reported. Events are served only on the hosted HTTP transport and only behind the ``events`` flag (``MCP_EVENTS``); see :func:`enabled`. @@ -20,16 +24,26 @@ from __future__ import annotations -from typing import Any +import time +from collections.abc import Awaitable, Callable, Mapping +from typing import Any, TypeVar +from mcp import MCPError from mcp.server import Server, ServerRequestContext from mcp.server.context import CallNext, HandlerResult from mcp.types import RequestParams from pydantic import ConfigDict -from .. import flags -from .catalog import EVENTS, Event +from .. import error_monitoring, flags, telemetry +from .catalog import CATALOG, EVENTS, Event from .errors import EventsError +from .subscriptions import ( + WEBHOOK, + Operation, + SubscribeParams, + Subscriptions, + UnsubscribeParams, +) CAPABILITY = "events" """Key under ``capabilities`` that ChatGPT reads.""" @@ -40,10 +54,9 @@ EXTENSION_SETTINGS: dict[str, Any] = {"listChanged": False} """The catalog is static, so there is never a list-changed notification.""" -DELIVERY_WEBHOOK = "webhook" -"""The only delivery mode served (v1 is webhook-only, like ChatGPT).""" - LIST_METHOD = "events/list" +SUBSCRIBE_METHOD = "events/subscribe" +UNSUBSCRIBE_METHOD = "events/unsubscribe" TRANSPORT = "http" """Events need the hosted ingress, so stdio never advertises them.""" @@ -72,7 +85,7 @@ def describe(event: Event) -> dict[str, Any]: return { "name": event.name, "description": event.description, - "delivery": [DELIVERY_WEBHOOK], + "delivery": [WEBHOOK], "inputSchema": event.input_schema, "payloadSchema": event.payload_schema, } @@ -116,7 +129,105 @@ async def _handle_list( return list_events(params.cursor) -def register(server: Server[Any]) -> None: - """Advertise events and serve ``events/list`` on ``server``.""" +Identify = Callable[[ServerRequestContext[Any, Any]], Mapping[str, Any]] +"""Sentry tags naming the client behind a request (``mcp.client.name``, …).""" + +_Params = TypeVar("_Params", SubscribeParams, UnsubscribeParams) + + +def _observed( + method: str, + operation: Operation, + handler: Callable[[_Params], Awaitable[dict[str, Any]]], + identify: Identify, +) -> Callable[[ServerRequestContext[Any, Any], _Params], Awaitable[dict[str, Any]]]: + """``handler`` with the telemetry and error reporting ``tools/call`` has.""" + + async def handle( + context: ServerRequestContext[Any, Any], params: _Params + ) -> dict[str, Any]: + started = time.monotonic() + name = params.name + event = name if isinstance(name, str) and name in CATALOG else None + try: + result = await handler(params) + except Exception as error: + # A bug is reported, and its text never sent to the client. + failure = ( + error + if isinstance(error, MCPError) + else EventsError.internal("Internal error") + ) + code = failure.code + telemetry.record_message( + method, + "error", + time.monotonic() - started, + error_code=code, + error_message=type(error).__name__, + ) + telemetry.record_subscription(operation, "error", event, str(code)) + if not (isinstance(error, EventsError) and error.expected): + _report(error, method, event, params, context, identify) + if failure is error: + raise + raise failure from error + telemetry.record_message(method, "success", time.monotonic() - started) + return result + + return handle + + +def _report( + error: Exception, + method: str, + event: str | None, + params: SubscribeParams | UnsubscribeParams, + context: ServerRequestContext[Any, Any], + identify: Identify, +) -> None: + arguments = params.arguments if isinstance(params.arguments, Mapping) else {} + project = arguments.get("project_id") + try: + client = dict(identify(context)) + except Exception: + client = {} + tags = { + "mcp.method": method, + "transport": TRANSPORT, + **client, + **({"event.name": event} if event else {}), + **({"appwrite.project_id": project} if isinstance(project, str) else {}), + } + error_monitoring.capture_exception( + error, + tags=tags, + context={"mcp": {"method": method, "event": event, "transport": TRANSPORT}}, + transaction=f"mcp.{method}:{event}" if event else f"mcp.{method}", + ) + + +def register( + server: Server[Any], subscriptions: Subscriptions, identify: Identify +) -> None: + """Advertise events and serve ``events/list``, ``events/subscribe`` and + ``events/unsubscribe`` on ``server``.""" server.add_request_handler(LIST_METHOD, ListParams, _handle_list) + server.add_request_handler( + SUBSCRIBE_METHOD, + SubscribeParams, + _observed( + SUBSCRIBE_METHOD, Operation.SUBSCRIBE, subscriptions.subscribe, identify + ), + ) + server.add_request_handler( + UNSUBSCRIBE_METHOD, + UnsubscribeParams, + _observed( + UNSUBSCRIBE_METHOD, + Operation.UNSUBSCRIBE, + subscriptions.unsubscribe, + identify, + ), + ) server.middleware.append(CapabilityMiddleware()) diff --git a/src/mcp_server_appwrite/events/subscriptions.py b/src/mcp_server_appwrite/events/subscriptions.py new file mode 100644 index 0000000..4255375 --- /dev/null +++ b/src/mcp_server_appwrite/events/subscriptions.py @@ -0,0 +1,628 @@ +"""``events/subscribe`` and ``events/unsubscribe`` on a stateless server. + +A subscription is one managed Appwrite webhook in the subscriber's project (see +:mod:`.webhooks`), and nothing about it is stored here. ``events/subscribe``: + +1. Validates the request: a catalog event (``-32011``), its arguments, a + ``webhook`` delivery (``-32014`` otherwise), an ``https`` callback that + resolves only to public addresses and a ``whsec_`` secret (``-32602``). +2. Takes the principal from the caller's OAuth token (``-32012`` without one), + and refuses a token whose granted scopes cannot write webhooks (``-32012``) + before anything reaches the callback. +3. Grants a TTL (:class:`Grant`) and seals the subscription into an envelope + for this server (an oversized one, or one the subscription record refuses, + is ``-32602``). +4. Authorizes: reads the event's resource in the project with the caller's own + token (:class:`.catalog.Access`). Every refresh repeats it, so access that + was revoked ends the subscription at the next refresh at the latest. +5. Lists the project's webhooks (which also proves the webhook scopes) and + deletes this principal's managed ones that expired more than + :data:`CLEANUP_GRACE_MS` ago and are unchanged when read again, so stale + subscriptions do not hold Free-plan slots. Expiry and owner come from the + webhook name (:class:`.webhooks.Label`), because the envelope cannot be + read back. +6. Runs the verification handshake against the callback, every time + (``-32015`` with ``data.reason`` on failure). The server keeps no + per-``(principal, url)`` cache and cannot read the previous envelope back + from Appwrite, so it cannot tell a known callback from a new one; one signed + POST per refresh is cheap and always proves the callback still wants + deliveries under the secret it is about to get. +7. Creates the subscription's webhook, or rewrites it on refresh. + +A refresh with a new ``delivery.secret`` replaces the secret. The previous one +cannot be kept for dual signing: it lives only in the old envelope, which +Appwrite never returns. Deliveries already in flight were opened from the old +envelope and keep signing with the old secret through their retries. + +Errors from Appwrite map onto the events codes; see :func:`_access_error` and +:func:`_webhook_error`. Creating is one call. A refresh writes the signing key +first and the envelope second, so a webhook never holds a signing key derived +from an older sealing key than its envelope's. When the second call fails, the +webhook keeps its previous envelope with the new signing key: the ingress +accepts the key of every sealing key in the ring, and the previous envelope's +sealing key stays in the ring until that envelope expires, so the webhook keeps +delivering until then. The error is reported so the client retries. The other +order could leave a new envelope with a signing key derived from a sealing key +that is retired while that envelope is still live. + +``events/unsubscribe`` recomputes the id from the principal and the request and +deletes that webhook if it is a managed one. A missing webhook is still +success, so unsubscribe is idempotent; a missing project is ``-32012``, as on +subscribe. The id is bound to the principal, so a +caller can only ever reach their own subscriptions. +""" + +from __future__ import annotations + +import sys +import time +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from enum import StrEnum +from typing import Any + +import httpx +from anyio import to_thread +from appwrite_console.client import Client +from appwrite_console.exception import AppwriteException +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.types import RequestParams +from pydantic import ConfigDict + +from .. import telemetry +from .catalog import Event, lookup +from .delivery import Callback, CallbackError, InvalidSecretError, decode_secret +from .egress import SECURE_SCHEME, DestinationError +from .envelope import ( + EnvelopeTooLarge, + InvalidSubscription, + Principal, + Subscription, + subscription_id, + timestamp, +) +from .errors import CallbackFailure, EventsError +from .ingress import Ingress +from .webhooks import Label, Webhooks, managed, owner + +WEBHOOK = "webhook" +"""The only delivery mode served (v1 is webhook-only, like ChatGPT).""" + +TTL_DEFAULT_MS = 60 * 60 * 1000 +"""Granted when the client suggests no TTL, or asks for none (``ttlMs: null``): +a server MUST NOT grant no-expiry without durable storage, and this one has +none.""" + +TTL_MIN_MS = 5 * 60 * 1000 +"""Shorter suggestions are raised to this floor, which keeps refresh traffic +(and the verification POST each refresh sends) bounded.""" + +TTL_MAX_MS = 24 * 60 * 60 * 1000 +"""Longer suggestions are capped here. A refresh is also the access re-check, +so this bounds how long revoked access can keep receiving events.""" + +REFRESH_SHARE = 10 +"""``refreshBefore`` comes this fraction (1/10) of the grant before the +envelope expires: 6 minutes for the default hour, 30 s at the floor. The +margin absorbs a slow refresh without letting deliveries lapse.""" + +CLEANUP_GRACE_MS = 10 * 60 * 1000 +"""How long past its expiry a managed webhook must be before a subscribe +cleans it up. A client still refreshing that subscription would have done so +well before; past this it has gone, and its deliveries are already dropped.""" + +PROJECT = "project_id" + +SCOPE_PREFIX = "project:" +WEBHOOK_SCOPES = ("webhooks.read", "webhooks.write") +WRITE_GRANTS = frozenset({"all", "project:all", "project:webhooks.write"}) +"""OAuth scopes any one of which lets a token write project webhooks.""" + + +class Operation(StrEnum): + SUBSCRIBE = "subscribe" + UNSUBSCRIBE = "unsubscribe" + CLEANUP = "cleanup" + + +class Outcome(StrEnum): + CREATED = "created" + REFRESHED = "refreshed" + REMOVED = "removed" + ABSENT = "absent" + + +class AppwriteError(StrEnum): + """Appwrite error ``type`` values the mapping depends on.""" + + UNAUTHORIZED_SCOPE = "general_unauthorized_scope" + PROJECT_NOT_FOUND = "project_not_found" + ALREADY_EXISTS = "webhook_already_exists" + PLAN_LIMIT = "additional_resource_not_allowed" + """Cloud's plan limit on webhooks per project (403).""" + + +class SubscribeParams(RequestParams): + """``events/subscribe`` params. Every member is checked by hand so each + problem gets its own error code and message; unknown members are + tolerated, as the draft specification may add some.""" + + model_config = ConfigDict(extra="allow") + + name: Any = None + arguments: Any = None + delivery: Any = None + cursor: Any = None + ttl_ms: Any = None + max_age_ms: Any = None + + +class UnsubscribeParams(RequestParams): + """``events/unsubscribe`` params.""" + + model_config = ConfigDict(extra="allow") + + name: Any = None + arguments: Any = None + delivery: Any = None + + +@dataclass(frozen=True) +class Grant: + """A granted subscription lifetime, in Unix epoch milliseconds.""" + + expires: int + """When the envelope stops delivering.""" + refresh: int + """``refreshBefore``: when the client must have refreshed by.""" + + @classmethod + def of(cls, suggested: int | None, now: int) -> Grant: + """Grant for a suggested TTL (``None``: no or a null suggestion).""" + if suggested is None: + ttl = TTL_DEFAULT_MS + else: + ttl = min(max(suggested, TTL_MIN_MS), TTL_MAX_MS) + expires = now + ttl + return cls(expires=expires, refresh=expires - ttl // REFRESH_SHARE) + + +ClientFactory = Callable[[str], Client] +"""Builds the Appwrite client for the current request's token, scoped to a +project (``server.resolve_client``).""" + + +@dataclass(frozen=True) +class State: + """What :meth:`Subscriptions._prepare` learned about the project.""" + + webhooks: Webhooks + exists: bool + """Whether the subscription's webhook is already there (a refresh).""" + count: int + """Webhooks in the project after cleanup.""" + + +@dataclass(frozen=True) +class Target: + """What a subscribe or unsubscribe request names, validated.""" + + event: Event + arguments: dict[str, str] + delivery: Mapping[str, Any] + + @property + def project(self) -> str: + return self.arguments[PROJECT] + + +class Subscriptions: + """Serves ``events/subscribe`` and ``events/unsubscribe`` for one + :class:`Ingress`, whose keyring, URL, egress and dispatcher it shares.""" + + def __init__( + self, + ingress: Ingress, + client: ClientFactory, + *, + clock: Callable[[], float] = time.time, + ) -> None: + self._ingress = ingress + self._client = client + self._clock = clock + + def _now(self) -> int: + return int(self._clock() * 1000) + + async def subscribe(self, params: SubscribeParams) -> dict[str, Any]: + target = _target(params.name, params.arguments, params.delivery) + mode = target.delivery.get("mode", WEBHOOK) + if mode != WEBHOOK: + raise EventsError.unsupported( + f"Delivery mode {mode!r} is not supported; use 'webhook'" + ) + callback = _callback(target.delivery) + secret = target.delivery.get("secret") + if not isinstance(secret, str): + raise EventsError.invalid_params("delivery.secret is required") + try: + decode_secret(secret) + except InvalidSecretError as error: + raise EventsError.invalid_params(f"delivery.secret: {error}") from error + suggested = _ttl(params) + _max_age(params.max_age_ms) + principal = _principal() + _writable() + + grant = Grant.of(suggested, self._now()) + subscription = Subscription.create( + project=target.project, + name=target.event.name, + arguments=target.arguments, + callback=callback, + secret=secret, + expires=grant.expires, + refresh=grant.refresh, + principal=principal, + ) + try: + envelope = self._ingress.keyring.seal(subscription, self._ingress.server) + except EnvelopeTooLarge as error: + raise EventsError.invalid_params( + "delivery.url is too long to fit in the subscription" + ) from error + except InvalidSubscription as error: + raise EventsError.invalid_params( + "delivery is not a valid subscription: the url must be https " + "without credentials or a fragment, and the secret canonical " + "padded base64" + ) from error + await self._check(callback) + + state = await to_thread.run_sync( + self._prepare, target, subscription, abandon_on_cancel=True + ) + await self._verify(subscription) + outcome = await to_thread.run_sync( + self._write, state, target, subscription, envelope, abandon_on_cancel=True + ) + telemetry.record_subscription(Operation.SUBSCRIBE, outcome, target.event.name) + _log(f"{subscription.id}: {outcome} {target.event.name} until {grant.expires}") + return { + "id": subscription.id, + "refreshBefore": timestamp(grant.refresh), + "cursor": None, + "truncated": False, + } + + async def unsubscribe(self, params: UnsubscribeParams) -> dict[str, Any]: + target = _target(params.name, params.arguments, params.delivery) + url = target.delivery.get("url") + if not isinstance(url, str) or not url: + raise EventsError.invalid_params("delivery.url is required") + principal = _principal() + id = subscription_id(principal, url, target.event.name, target.arguments) + outcome = await to_thread.run_sync( + self._remove, target, id, abandon_on_cancel=True + ) + telemetry.record_subscription(Operation.UNSUBSCRIBE, outcome, target.event.name) + _log(f"{id}: unsubscribe {outcome}") + return {} + + async def _check(self, url: str) -> None: + """Refuse a callback whose host is not public before anything is sent + to Appwrite or to the callback. The connect-time check still runs on + every request.""" + try: + await self._ingress.egress.check(url) + except DestinationError as error: + raise EventsError.invalid_params(f"delivery.url: {error}") from error + except TimeoutError as error: + raise EventsError.callback_endpoint( + "The callback host did not resolve in time", + reason=CallbackFailure.TIMEOUT, + ) from error + except OSError as error: + raise EventsError.invalid_params( + "delivery.url: Callback host did not resolve" + ) from error + + async def _verify(self, subscription: Subscription) -> None: + callback = Callback(subscription.callback, subscription.id, subscription.secret) + try: + await self._ingress.dispatcher.verify(callback) + except CallbackError as error: + raise EventsError.callback_endpoint( + f"The callback URL failed verification ({error.reason})", + reason=error.reason, + ) from error + + def _project_client(self, project: str) -> Client: + try: + return self._client(project) + except RuntimeError as error: + raise EventsError.forbidden( + "Event subscriptions need an Appwrite OAuth access token" + ) from error + + def _authorize(self, target: Target) -> Client: + """The project client, once the caller has proven they may read the + event's resource.""" + client = self._project_client(target.project) + access = target.event.access + try: + client.call( + "get", + access.resolve(target.arguments), + headers={"accept": "application/json"}, + params=dict(access.params), + ) + except AppwriteException as error: + raise _access_error(error, target) from error + return client + + def _prepare(self, target: Target, subscription: Subscription) -> State: + """Everything that needs Appwrite before the handshake: authorize, + read the project's webhooks (which also proves the webhook scopes) and + clean up this principal's expired ones. A refusal here costs the + callback nothing.""" + webhooks = Webhooks(self._authorize(target)) + try: + existing = webhooks.listing() + except AppwriteException as error: + raise _webhook_error(error, target.project, 0) from error + current = next((w for w in existing if w.id == subscription.id), None) + if current is not None and managed(current) is None: + raise _foreign(subscription.id, target.project) + removed = self._clean(webhooks, existing, subscription) + return State(webhooks, current is not None, len(existing) - removed) + + def _write( + self, + state: State, + target: Target, + subscription: Subscription, + envelope: str, + ) -> Outcome: + webhooks, count = state.webhooks, state.count + label = Label( + target.event.name, subscription.expires, owner(subscription.principal) + ) + fields: dict[str, Any] = { + "url": self._ingress.url(subscription.id), + "label": label, + "events": target.event.patterns(target.arguments), + "password": envelope, + } + secret = self._ingress.keyring.signing_key(subscription.id) + if not state.exists: + try: + webhooks.create(subscription.id, secret=secret, **fields) + return Outcome.CREATED + except AppwriteException as error: + if error.type != AppwriteError.ALREADY_EXISTS: + raise _webhook_error(error, target.project, count) from error + # Created concurrently by another subscribe: rewrite it below. + try: + current = webhooks.get(subscription.id) + except AppwriteException as error: + raise _webhook_error(error, target.project, count) from error + if current is not None and managed(current) is None: + raise _foreign(subscription.id, target.project) + # The signing key goes first; see the module docstring for why. + try: + webhooks.sign(subscription.id, secret) + except AppwriteException as error: + raise _webhook_error(error, target.project, count) from error + try: + webhooks.update(subscription.id, **fields) + except AppwriteException as error: + # The new signing key is written and the previous envelope stays. + # Report the failure so the client retries, but do not turn a + # transient error into a lost subscription. + raise _webhook_error(error, target.project, count) from error + return Outcome.REFRESHED + + def _clean( + self, webhooks: Webhooks, existing: list[Any], subscription: Subscription + ) -> int: + """Delete this principal's long-expired managed webhooks; return how + many went. Best effort: a failed delete is retried on the next + subscribe. + + Appwrite has no conditional delete, so a refresh of a candidate can + race this. Two guards keep that from deleting a live subscription: + only webhooks expired for longer than :data:`CLEANUP_GRACE_MS` are + candidates, and each is read again right before the delete and kept + if its name or ``$updatedAt`` changed since the listing. What remains + is a refresh landing between that read and the delete, of a + subscription its client left expired for over ten minutes; the client + then gets its webhook back on the following refresh.""" + tag = owner(subscription.principal) + cutoff = self._now() - CLEANUP_GRACE_MS + removed = 0 + for webhook in existing: + label = managed(webhook) + if label is None or webhook.id == subscription.id: + continue + if label.owner != tag or not label.expired(cutoff): + continue + try: + current = webhooks.get(webhook.id) + except AppwriteException as error: + _log(f"{webhook.id}: cleanup failed ({error.code} {error.type})") + continue + if current is None or (current.name, current.updatedat) != ( + webhook.name, + webhook.updatedat, + ): + continue + try: + webhooks.delete(webhook.id) + except AppwriteException as error: + _log(f"{webhook.id}: cleanup failed ({error.code} {error.type})") + continue + removed += 1 + telemetry.record_subscription( + Operation.CLEANUP, Outcome.REMOVED, label.event + ) + _log(f"{webhook.id}: removed expired {label.event}") + return removed + + def _remove(self, target: Target, id: str) -> Outcome: + webhooks = Webhooks(self._project_client(target.project)) + try: + webhook = webhooks.get(id) + if webhook is None or managed(webhook) is None: + return Outcome.ABSENT + return Outcome.REMOVED if webhooks.delete(id) else Outcome.ABSENT + except AppwriteException as error: + raise _webhook_error(error, target.project, 0) from error + + +def _target(name: object, arguments: object, delivery: object) -> Target: + event = lookup(name) + validated = event.validate(arguments) + if not isinstance(delivery, Mapping): + raise EventsError.invalid_params("delivery must be an object") + return Target(event, validated, delivery) + + +def _callback(delivery: Mapping[str, Any]) -> str: + url = delivery.get("url") + if not isinstance(url, str) or not url: + raise EventsError.invalid_params("delivery.url is required") + try: + scheme = httpx.URL(url).scheme + except (httpx.InvalidURL, TypeError) as error: + raise EventsError.invalid_params("delivery.url is not a valid URL") from error + if scheme != SECURE_SCHEME: + raise EventsError.invalid_params("delivery.url must use https") + return url + + +def _ttl(params: SubscribeParams) -> int | None: + """The suggested TTL in milliseconds; ``None`` when it is absent or + ``null`` (both get the default grant).""" + value = params.ttl_ms + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + raise EventsError.invalid_params("ttlMs must be a positive integer or null") + return value + + +def _max_age(value: object) -> None: + """``maxAgeMs`` bounds replay. Appwrite webhooks have none, so it is only + checked for shape.""" + if value is None: + return + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise EventsError.invalid_params( + "maxAgeMs must be a non-negative integer or null" + ) + + +def _writable() -> None: + """Refuse a token whose granted scopes cannot write webhooks before the + callback gets a verification POST. + + Listing the webhooks proves ``webhooks.read`` only, and Appwrite has no + side-effect-free way to prove ``webhooks.write``; the token's ``scope`` + claim does. A token that carries no scopes at all is left to Appwrite, + which refuses the write itself.""" + token = get_access_token() + scopes = set(token.scopes) if token is not None else set() + if not scopes or scopes & WRITE_GRANTS: + return + raise EventsError.forbidden(_scopes_message()) + + +def _scopes_message() -> str: + scopes = " and ".join(SCOPE_PREFIX + scope for scope in WEBHOOK_SCOPES) + return ( + f"Event subscriptions need the {scopes} scopes. Reconnect the " + "Appwrite connector and grant them." + ) + + +def _principal() -> str: + """Digest of the OAuth principal behind the current request.""" + token = get_access_token() + claims = (token.claims or {}) if token is not None else {} + issuer = claims.get("iss") + subject = (token.subject if token is not None else None) or claims.get("sub") + client = token.client_id if token is not None else None + if not all(isinstance(part, str) and part for part in (issuer, subject, client)): + raise EventsError.forbidden( + "Event subscriptions need an Appwrite OAuth access token for a user" + ) + return Principal(str(issuer), str(subject), str(client)).digest + + +def _foreign(id: str, project: str) -> EventsError: + """The subscription's webhook id is taken by a webhook whose name is not + ours (someone renamed it). It is left alone.""" + return EventsError.forbidden( + f"Webhook {id} in project {project} is not managed by MCP events. " + "Delete it in the Appwrite Console, then subscribe again." + ) + + +def _access_error(error: AppwriteException, target: Target) -> EventsError: + """Map a failed authorization read. + + Appwrite authenticates the caller against the project before it looks up + the resource: a token without access to the project gets ``401`` (or + ``404 project_not_found`` for a project that does not exist, which is + reported the same way so project ids cannot be probed). So a ``404`` for + the resource itself means the caller has project access and the function, + site, table or bucket is missing: NotFound with ``data.kind = "resource"``. + """ + access = target.event.access + resource = access.describe(target.arguments) + code = error.code if isinstance(error.code, int) else 0 + if code == 404 and error.type != AppwriteError.PROJECT_NOT_FOUND: + return EventsError.not_found( + f"Appwrite has no {resource} in project {target.project}", + kind="resource", + ) + if code in (401, 403, 404): + if error.type == AppwriteError.UNAUTHORIZED_SCOPE: + return EventsError.forbidden( + f"Reading {resource} needs the {SCOPE_PREFIX}{access.scope} scope. " + "Reconnect the Appwrite connector and grant it." + ) + return EventsError.forbidden( + f"You do not have access to {resource} in project {target.project}" + ) + return EventsError.internal( + f"Appwrite failed to read {resource} (HTTP {code or 'error'})" + ) + + +def _webhook_error(error: AppwriteException, project: str, count: int) -> EventsError: + """Map a failed webhook call. ``count`` is how many webhooks the project + has, which is the plan's maximum when the plan limit is what failed.""" + code = error.code if isinstance(error.code, int) else 0 + if error.type == AppwriteError.PLAN_LIMIT: + return EventsError.resource_exhausted( + f"Project {project} has reached its plan's limit of {count} webhooks, " + "and every event subscription uses one. On the Free plan, upgrade to " + "Pro for unlimited webhooks, or unsubscribe from another event first.", + limit="webhooks", + maximum=count, + ) + # A project that does not exist is reported like one the caller cannot + # reach, as on subscribe, so project ids cannot be probed. + if code in (401, 403) or error.type == AppwriteError.PROJECT_NOT_FOUND: + if error.type == AppwriteError.UNAUTHORIZED_SCOPE: + return EventsError.forbidden(_scopes_message()) + return EventsError.forbidden(f"You cannot manage webhooks in project {project}") + return EventsError.internal( + f"Appwrite could not save the subscription (HTTP {code or 'error'})" + ) + + +def _log(message: str) -> None: + print(f"[appwrite-mcp][events] {message}", file=sys.stderr, flush=True) diff --git a/src/mcp_server_appwrite/events/webhooks.py b/src/mcp_server_appwrite/events/webhooks.py new file mode 100644 index 0000000..406c32e --- /dev/null +++ b/src/mcp_server_appwrite/events/webhooks.py @@ -0,0 +1,213 @@ +"""The Appwrite project webhooks that back MCP event subscriptions. + +Every subscription is one webhook in the subscriber's own project, written with +the subscriber's OAuth token (``project:webhooks.read`` and +``project:webhooks.write``). This module owns what such a webhook looks like +and how it is recognized later: + +* ``$id`` is the subscription id (``sub_`` + 32 hex chars, see + :func:`.envelope.subscription_id`), so a refresh is an update and an + unsubscribe a delete by id. +* ``name`` is a :class:`Label`, for example + ``MCP events | tablesdb.row.created | until 2026-10-09T11:00:00Z | owner OPxPhaa1``. + Appwrite stores ``authPassword`` write-only, so the sealed envelope (and the + expiry inside it) can never be read back. The label repeats the two facts a + later subscribe needs to clean up without opening it: when the subscription + expires and a short tag of who made it. It is also what a person sees in the + Console. +* ``authUsername`` is :data:`USERNAME`, a fixed non-secret value. Appwrite sends + Basic auth only when both username and password are set, and the password is + the envelope. +* ``secret`` is the derived Appwrite signing key. Current Appwrite returns it + only from create and from the secret update, so a refresh always sets it + again rather than comparing. + +A webhook is **managed** only when its id is a subscription id *and* its name +parses as a label. Anything else in the project belongs to the user and is +never changed or deleted. +""" + +from __future__ import annotations + +import math +import re +from dataclasses import dataclass +from datetime import UTC, datetime + +from appwrite_console.client import Client +from appwrite_console.exception import AppwriteException +from appwrite_console.models.webhook import Webhook +from appwrite_console.query import Query +from appwrite_console.services.webhooks import Webhooks as Service + +from .envelope import valid_subscription_id + +USERNAME = "mcp-events" +"""``authUsername`` of every managed webhook. Not a secret: it only makes +Appwrite send the envelope (the password) as Basic auth.""" + +PREFIX = "MCP events" +SEPARATOR = " | " +"""ASCII on purpose: Appwrite's worker sends the name back in the +``X-Appwrite-Webhook-Name`` header of every delivery.""" +UNTIL = "until " +OWNER = "owner " +STAMP = "%Y-%m-%dT%H:%M:%SZ" + +OWNER_CHARS = 8 +"""Characters of the principal digest kept in the label. Enough to keep +principals apart within one project; a collision could only ever delete +another principal's *expired* webhook, whose deliveries are already dropped.""" + +OWNER_PATTERN = re.compile(rf"[A-Za-z0-9_-]{{{OWNER_CHARS}}}") + +NAME_MAX = 128 +"""Appwrite's limit on a webhook name. The longest label (the longest catalog +event name) is under 100 characters.""" + +PAGE = 100 +"""Webhooks per listing request.""" + +NOT_FOUND = "webhook_not_found" +ALREADY_EXISTS = "webhook_already_exists" + + +def owner(principal: str) -> str: + """The label tag for a principal digest.""" + return principal[:OWNER_CHARS] + + +@dataclass(frozen=True) +class Label: + """The ``name`` of a managed webhook.""" + + event: str + expires: int + """Expiry as Unix epoch milliseconds. Written rounded up to the second, so + the label never claims an earlier expiry than the envelope.""" + owner: str + + def __str__(self) -> str: + seconds = math.ceil(self.expires / 1000) + stamp = datetime.fromtimestamp(seconds, UTC).strftime(STAMP) + return SEPARATOR.join((PREFIX, self.event, UNTIL + stamp, OWNER + self.owner)) + + @classmethod + def parse(cls, name: str) -> Label | None: + """The label ``name`` spells, or ``None`` if it is not one.""" + parts = name.split(SEPARATOR) + if len(parts) != 4: + return None + prefix, event, until, tag = parts + if prefix != PREFIX or not until.startswith(UNTIL): + return None + if not tag.startswith(OWNER): + return None + tag = tag.removeprefix(OWNER) + if OWNER_PATTERN.fullmatch(tag) is None or not event: + return None + try: + stamp = datetime.strptime(until.removeprefix(UNTIL), STAMP) + except ValueError: + return None + expires = int(stamp.replace(tzinfo=UTC).timestamp()) * 1000 + return cls(event=event, expires=expires, owner=tag) + + def expired(self, now: int) -> bool: + return now >= self.expires + + +def managed(webhook: Webhook) -> Label | None: + """The label of a webhook this server created, or ``None`` for anything + else in the project.""" + if not valid_subscription_id(webhook.id): + return None + return Label.parse(webhook.name) + + +class Webhooks: + """Webhook calls in one project, made with the caller's client. + + Errors other than "not found" propagate as ``AppwriteException`` for the + caller to translate.""" + + def __init__(self, client: Client) -> None: + self._service = Service(client) + + def listing(self) -> list[Webhook]: + """Every webhook in the project.""" + webhooks: list[Webhook] = [] + while True: + page = self._service.list( + queries=[Query.limit(PAGE), Query.offset(len(webhooks))] + ).webhooks + webhooks.extend(page) + if len(page) < PAGE: + return webhooks + + def get(self, id: str) -> Webhook | None: + try: + return self._service.get(id) + except AppwriteException as error: + if error.type == NOT_FOUND: + return None + raise + + def create( + self, + id: str, + *, + url: str, + label: Label, + events: list[str], + password: str, + secret: str, + ) -> None: + self._service.create( + id, + url, + str(label), + events, + enabled=True, + tls=True, + auth_username=USERNAME, + auth_password=password, + secret=secret, + ) + + def update( + self, + id: str, + *, + url: str, + label: Label, + events: list[str], + password: str, + ) -> None: + """Rewrite a managed webhook. Re-enables it if Appwrite paused it after + failed deliveries, as a refresh should. The update endpoint does not + take the secret; see :meth:`sign`.""" + self._service.update( + id, + str(label), + url, + events, + enabled=True, + tls=True, + auth_username=USERNAME, + auth_password=password, + ) + + def sign(self, id: str, secret: str) -> None: + """Set the webhook's signing key.""" + self._service.update_secret(id, secret) + + def delete(self, id: str) -> bool: + """Delete a webhook; ``False`` when it was already gone.""" + try: + self._service.delete(id) + except AppwriteException as error: + if error.type == NOT_FOUND: + return False + raise + return True diff --git a/src/mcp_server_appwrite/http_app.py b/src/mcp_server_appwrite/http_app.py index b4f4ad9..2b91d48 100644 --- a/src/mcp_server_appwrite/http_app.py +++ b/src/mcp_server_appwrite/http_app.py @@ -468,7 +468,11 @@ def build_app(ingress: Ingress | None = None) -> Starlette: telemetry.init_telemetry("http", SERVER_VERSION) tools_manager = build_catalog_tools_manager() operator = build_operator(tools_manager, store_results=False) - server = build_mcp_server(operator, transport="http") + if events_protocol.enabled("http"): + ingress = ingress or Ingress.from_env() + else: + ingress = None + server = build_mcp_server(operator, transport="http", ingress=ingress) # Streamable HTTP with SSE responses (the MCP SDK/ecosystem default). Stateless, # so each request opens and closes its own short-lived stream — no session to pin. @@ -483,11 +487,6 @@ async def handle_mcp(scope: Scope, receive: Receive, send: Send) -> None: mcp_endpoint = RequireBearer(MCPIdentityMiddleware(handle_mcp)) - if events_protocol.enabled("http"): - ingress = ingress or Ingress.from_env() - else: - ingress = None - @contextlib.asynccontextmanager async def lifespan(app: Starlette): async with contextlib.AsyncExitStack() as stack: diff --git a/src/mcp_server_appwrite/server.py b/src/mcp_server_appwrite/server.py index 136d6d5..6881f59 100644 --- a/src/mcp_server_appwrite/server.py +++ b/src/mcp_server_appwrite/server.py @@ -75,6 +75,8 @@ from .docs_search import DocsSearch from .error_classification import HostedBinaryResponseTooLarge, is_response_parse_error from .events import protocol as events +from .events.ingress import Ingress +from .events.subscriptions import Subscriptions from .operator import Operator, _parse_tool_name from .service import Service from .tool_manager import ToolManager @@ -1265,7 +1267,12 @@ def build_instructions(transport: str = "http", *, docs_enabled: bool = True) -> ) -def build_mcp_server(operator: Operator, *, transport: str = "http") -> Server: +def build_mcp_server( + operator: Operator, *, transport: str = "http", ingress: Ingress | None = None +) -> Server: + """The low-level MCP server. With events enabled, ``events/subscribe`` + writes webhooks that point at ``ingress`` (built from the environment when + not given) and verifies callbacks through its egress.""" _configure_uploads(transport) instructions = build_instructions( transport, docs_enabled=bool(getattr(operator, "docs_enabled", False)) @@ -1518,7 +1525,11 @@ def tool_input_schema(name: str) -> dict[str, Any] | None: get_tool_input_schema=tool_input_schema, ) if events.enabled(transport): - events.register(server) + events.register( + server, + Subscriptions(ingress or Ingress.from_env(), resolve_client), + lambda ctx: _mcp_request_context(ctx).tags, + ) return server diff --git a/src/mcp_server_appwrite/telemetry.py b/src/mcp_server_appwrite/telemetry.py index 37eed0d..277d248 100644 --- a/src/mcp_server_appwrite/telemetry.py +++ b/src/mcp_server_appwrite/telemetry.py @@ -294,6 +294,14 @@ def _build_instruments(meter: Any, transport: str, version: str) -> None: unit="{event}", description="MCP event deliveries to subscriber callbacks by outcome.", ) + _instruments["events_subscriptions"] = meter.create_counter( + "mcp.events.subscriptions", + unit="{subscription}", + description=( + "events/subscribe and events/unsubscribe outcomes, and expired " + "subscription webhooks cleaned up, by operation and event." + ), + ) _instruments["token_usage"] = meter.create_counter( "mcp.token.usage", @@ -716,6 +724,19 @@ def record_ingress(outcome: str, reason: str | None, event: str | None) -> None: ) +def record_subscription( + operation: str, outcome: str, event: str | None, reason: str | None = None +) -> None: + """One subscription operation: ``subscribe`` (created, refreshed, error), + ``unsubscribe`` (removed, absent, error) or ``cleanup`` (removed). ``event`` + is a catalog event name, or ``None`` when the request named none.""" + _safe_add( + "events_subscriptions", + 1, + {"operation": operation, "outcome": outcome, "event": event, "reason": reason}, + ) + + def record_event_delivery(event: str, outcome: str, reason: str | None) -> None: """How one MCP event delivery to a subscriber callback ended.""" _safe_add( diff --git a/tests/e2e/support.py b/tests/e2e/support.py index cf1af69..fe2a34d 100644 --- a/tests/e2e/support.py +++ b/tests/e2e/support.py @@ -8,15 +8,21 @@ client drops capabilities it does not model, such as ``events``). The one stub is the OAuth token verifier: Cloud OAuth is not reachable from CI, -so a fixed bearer token is accepted in place of an Appwrite access token, the +so a few fixed bearer tokens (:data:`SUBJECTS`) are accepted in place of +Appwrite access tokens, with the claims the real verifier produces. This is the same seam the unit tests of ``http_app`` use. Everything behind it (routing, auth middleware, the MCP session manager, the low-level server and its middleware) is the production code path. -The parties around the server are real processes on localhost too: +The parties around the server are real servers on localhost too: +* :class:`Cloud` plays the Appwrite REST API the server calls with the + caller's token (``APPWRITE_ENDPOINT``): resource reads and ``/v1/webhooks``, + with Appwrite-shaped responses, scopes, plan limits and injectable failures. * :class:`Appwrite` plays Appwrite's webhooks worker: it signs and posts - deliveries exactly as ``src/Appwrite/Platform/Workers/Webhooks.php`` does. + deliveries exactly as ``src/Appwrite/Platform/Workers/Webhooks.php`` does, + either for a given webhook or, with :meth:`Appwrite.fire`, for every + webhook stored in :class:`Cloud` that listens to the event. * :class:`Receiver` plays ChatGPT's webhook endpoint: an HTTPS server with a self-signed certificate that checks every delivery with the official ``standardwebhooks`` library and echoes verification challenges. @@ -33,6 +39,7 @@ import itertools import json import os +import re import secrets import socket import ssl @@ -46,7 +53,7 @@ from types import TracebackType from typing import Any from unittest import mock -from urllib.parse import urlsplit +from urllib.parse import parse_qs, urlsplit import httpx import uvicorn @@ -66,9 +73,29 @@ from mcp_server_appwrite.http_app import build_app TOKEN = "e2e-access-token" -"""The bearer token the stubbed verifier accepts.""" +"""The bearer token the stubbed verifier accepts, for :data:`SUBJECT`.""" + +OTHER_TOKEN = "e2e-other-access-token" +"""A token for a second user, :data:`OTHER_SUBJECT`.""" + +SUBJECTLESS_TOKEN = "e2e-subjectless-access-token" +"""A token that names no user (no ``sub``), like a client-credentials token.""" CLIENT_ID = "e2e-client" +ISSUER = "https://cloud.appwrite.io/v1/oauth2/console" +SUBJECT = "66b1c2d3e4f5a6b7c8d9" +OTHER_SUBJECT = "77c2d3e4f5a6b7c8d9e0" + +SUBJECTS: dict[str, str | None] = { + TOKEN: SUBJECT, + OTHER_TOKEN: OTHER_SUBJECT, + SUBJECTLESS_TOKEN: None, +} +"""Every token the stubbed verifier accepts, and the user it names.""" + +GRANTED: dict[str, list[str]] = {} +"""The OAuth ``scope`` claim of a token, for tests that set one; others carry +none.""" PROTOCOL_VERSION = "2026-07-28" LEGACY_PROTOCOL_VERSION = "2025-11-25" @@ -80,6 +107,7 @@ STARTUP_SECONDS = 15.0 CONTROLLED = ( + "APPWRITE_ENDPOINT", "MCP_EVENTS", "MCP_EVENTS_SEALING_KEYS", "MCP_PUBLIC_URL", @@ -94,9 +122,22 @@ async def _verify_token(_verifier: Any, token: str) -> AccessToken | None: - if token != TOKEN: + """What ``AppwriteTokenVerifier`` returns for a valid Cloud token: the + console project in ``project_id`` and the issuer, user and client that + make up the events principal.""" + if token not in SUBJECTS: return None - return AccessToken(token=token, client_id=CLIENT_ID, scopes=[]) + subject = SUBJECTS[token] + claims: dict[str, Any] = {"iss": ISSUER, "project_id": "console"} + if subject is not None: + claims["sub"] = subject + return AccessToken( + token=token, + client_id=CLIENT_ID, + scopes=list(GRANTED.get(token, [])), + subject=subject, + claims=claims, + ) class Server: @@ -621,6 +662,7 @@ class Webhook: user: str password: str name: str = "MCP events" + enabled: bool = True def fired(event: str) -> list[str]: @@ -709,3 +751,473 @@ def deliver( return self._http.request( method, urlsplit(webhook.url).path, content=body, headers=request_headers ) + + def fire( + self, + cloud: Cloud, + project: str, + event: str, + payload: Mapping[str, Any] | bytes, + ) -> list[httpx.Response]: + """What Appwrite does when ``event`` happens in ``project``: one + delivery to every enabled webhook stored in ``cloud`` whose events + include one of the forms Appwrite generates for it.""" + forms = set(fired(event)) + return [ + self.deliver(webhook, event, payload) + for webhook in cloud.webhooks(project) + if webhook.enabled and forms & set(webhook.events) + ] + + +SCOPES = frozenset( + { + "functions.read", + "sites.read", + "tables.read", + "buckets.read", + "users.read", + "webhooks.read", + "webhooks.write", + } +) +"""Every Appwrite scope the events methods need.""" + +CUSTOM_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,35}") + +AUTH_PASSWORD_MAX = 2048 +"""What appwrite/appwrite#14293 must allow for ``authPassword`` (today 256).""" + + +@dataclass +class Project: + """One Appwrite project in :class:`Cloud`.""" + + id: str + functions: set[str] = field(default_factory=set) + sites: set[str] = field(default_factory=set) + tables: set[tuple[str, str]] = field(default_factory=set) + """``(database id, table id)`` pairs.""" + buckets: set[str] = field(default_factory=set) + webhook_limit: int | None = None + """Webhooks the plan allows (Free: 2); ``None`` for unlimited.""" + webhooks: dict[str, dict[str, Any]] = field(default_factory=dict) + """Stored webhook documents, write-only fields included.""" + + +@dataclass +class Failure: + """An injected Appwrite error for the next ``times`` matching requests.""" + + method: str + path: re.Pattern[str] + status: int + type: str + times: int + + +@dataclass +class Interleaving: + """Another client's work, run once before the next matching request.""" + + method: str + path: re.Pattern[str] + action: Callable[[], None] + + +class _Refusal(Exception): + def __init__(self, status: int, type: str, message: str) -> None: + super().__init__(message) + self.status = status + self.type = type + self.message = message + + +def _stamp() -> str: + """Now, in Appwrite's response format (``2026-10-09T10:00:00.250+00:00``).""" + return datetime.datetime.now(datetime.UTC).isoformat(timespec="milliseconds") + + +class Cloud: + """The Appwrite REST API at ``http://127.0.0.1:/v1``, as the server + reaches it with a caller's OAuth token. + + Only what ``events/subscribe`` and ``events/unsubscribe`` call is served: + the console project lookup ``resolve_client`` makes for regions, the + resource reads that authorize a subscription, and ``/v1/webhooks``. Each + request is checked the way Appwrite checks it: an unknown project is + ``404 project_not_found``, a token without access to the project is + ``401 user_unauthorized``, a missing scope is + ``401 general_unauthorized_scope``. Webhooks follow the current API: the + password is write-only, the secret is returned only by create and by the + secret update, an update keeps the password unless it sends one, and a + project at its plan limit refuses a create with + ``403 additional_resource_not_allowed``. + """ + + VERSION = "1.8.0" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._grants: dict[str, tuple[frozenset[str], frozenset[str]]] = {} + self._projects: dict[str, Project] = {} + self._failures: list[Failure] = [] + self._interleavings: list[Interleaving] = [] + self.requests: list[tuple[str, str, str | None]] = [] + """``(method, path, bearer token)`` of every request, in order.""" + cloud = self + + class Handler(BaseHTTPRequestHandler): + def _serve(self) -> None: + length = int(self.headers.get("Content-Length", "0") or 0) + body = self.rfile.read(length) if length else b"" + headers = {name.lower(): value for name, value in self.headers.items()} + status, payload = cloud._handle(self.command, self.path, headers, body) + content = b"" if payload is None else json.dumps(payload).encode() + self.send_response(status) + # Appwrite answers 204 with an HTML content type, which the + # Python SDK returns as raw bytes. + kind = ( + "text/html; charset=UTF-8" + if payload is None + else "application/json" + ) + self.send_header("Content-Type", kind) + self.send_header("Content-Length", str(len(content))) + self.end_headers() + self.wfile.write(content) + + do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = _serve + + def log_message(self, format: str, *args: Any) -> None: + pass + + self._server = _QuietServer(("127.0.0.1", 0), Handler) + threading.Thread(target=self._server.serve_forever, daemon=True).start() + self.endpoint = f"http://127.0.0.1:{self._server.server_address[1]}/v1" + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + + def add(self, project: Project) -> Project: + with self._lock: + self._projects[project.id] = project + return project + + def grant( + self, token: str, projects: Iterable[str], scopes: Iterable[str] = SCOPES + ) -> None: + """Let ``token`` reach ``projects`` with ``scopes``.""" + with self._lock: + self._grants[token] = (frozenset(projects), frozenset(scopes)) + + def fail( + self, + method: str, + path: str, + status: int = 500, + type: str = "general_server_error", + *, + times: int = 1, + ) -> None: + """Answer the next ``times`` requests matching ``method`` and the + ``path`` regular expression with an Appwrite error.""" + with self._lock: + self._failures.append( + Failure(method, re.compile(path), status, type, times) + ) + + def before(self, method: str, path: str, action: Callable[[], None]) -> None: + """Run ``action`` once, right before the next request matching + ``method`` and the ``path`` expression is handled, while that request + waits: how a concurrent client's calls land between two of a + server's. ``action`` may call this API itself.""" + with self._lock: + self._interleavings.append(Interleaving(method, re.compile(path), action)) + + def store(self, project: str, document: Mapping[str, Any]) -> None: + """Put a webhook straight into the project, as a person in the Console + (or an earlier subscribe) would have left it.""" + stamp = _stamp() + with self._lock: + self._projects[project].webhooks[document["$id"]] = { + "$createdAt": stamp, + "$updatedAt": stamp, + "events": [], + "tls": True, + "authUsername": "", + "httpPass": "", + "signatureKey": secrets.token_hex(64), + "enabled": True, + "logs": "", + "attempts": 0, + **document, + } + + def documents(self, project: str) -> dict[str, dict[str, Any]]: + """The stored webhook documents, write-only fields included.""" + with self._lock: + return { + id: dict(document) + for id, document in self._projects[project].webhooks.items() + } + + def webhooks(self, project: str) -> list[Webhook]: + """The project's webhooks as Appwrite's worker reads them.""" + return [ + Webhook( + id=id, + url=document["url"], + secret=document["signatureKey"], + events=tuple(document["events"]), + project=project, + user=document["authUsername"], + password=document["httpPass"], + name=document["name"], + enabled=document["enabled"], + ) + for id, document in self.documents(project).items() + ] + + def calls(self, method: str, path: str) -> int: + """How many requests matched ``method`` and the ``path`` expression.""" + pattern = re.compile(path) + with self._lock: + return sum( + 1 + for (made, requested, _) in self.requests + if made == method and pattern.fullmatch(requested) + ) + + def _handle( + self, method: str, target: str, headers: Mapping[str, str], body: bytes + ) -> tuple[int, Any]: + parts = urlsplit(target) + path = parts.path + query = parse_qs(parts.query) + authorization = headers.get("authorization", "") + token = authorization[7:] if authorization.startswith("Bearer ") else None + with self._lock: + interleaving = next( + ( + each + for each in self._interleavings + if each.method == method and each.path.fullmatch(path) + ), + None, + ) + if interleaving is not None: + self._interleavings.remove(interleaving) + if interleaving is not None: + interleaving.action() + with self._lock: + self.requests.append((method, path, token)) + try: + for failure in self._failures: + if failure.method == method and failure.path.fullmatch(path): + failure.times -= 1 + if failure.times <= 0: + self._failures.remove(failure) + raise _Refusal(failure.status, failure.type, "Injected") + payload = json.loads(body) if body else {} + return self._route(method, path, query, headers, token, payload) + except _Refusal as refusal: + return refusal.status, { + "message": refusal.message, + "code": refusal.status, + "type": refusal.type, + "version": self.VERSION, + } + + def _route( + self, + method: str, + path: str, + query: Mapping[str, list[str]], + headers: Mapping[str, str], + token: str | None, + payload: Mapping[str, Any], + ) -> tuple[int, Any]: + grant = self._grants.get(token or "") + region = re.fullmatch(r"/v1/projects/([^/]+)", path) + if method == "GET" and region: + if grant is None or region.group(1) not in grant[0]: + raise _Refusal(404, "project_not_found", "Project not found") + return 200, {"$id": region.group(1), "region": "default"} + + project = self._projects.get(headers.get("x-appwrite-project", "")) + if project is None: + raise _Refusal(404, "project_not_found", "Project not found") + if grant is None or project.id not in grant[0]: + raise _Refusal( + 401, "user_unauthorized", "The current user is not authorized" + ) + scopes = grant[1] + + def need(scope: str) -> None: + if scope not in scopes: + raise _Refusal( + 401, + "general_unauthorized_scope", + f'jane@example.com (role: owner) missing scopes (["{scope}"])', + ) + + if match := re.fullmatch(r"/v1/functions/([^/]+)", path): + need("functions.read") + if match.group(1) not in project.functions: + raise _Refusal(404, "function_not_found", "Function not found") + return 200, {"$id": match.group(1)} + if match := re.fullmatch(r"/v1/sites/([^/]+)", path): + need("sites.read") + if match.group(1) not in project.sites: + raise _Refusal(404, "site_not_found", "Site not found") + return 200, {"$id": match.group(1)} + if match := re.fullmatch(r"/v1/tablesdb/([^/]+)/tables/([^/]+)", path): + need("tables.read") + database, table = match.groups() + if database not in {db for db, _ in project.tables}: + raise _Refusal(404, "database_not_found", "Database not found") + if (database, table) not in project.tables: + raise _Refusal(404, "table_not_found", "Table not found") + return 200, {"$id": table, "databaseId": database} + if match := re.fullmatch(r"/v1/storage/buckets/([^/]+)", path): + need("buckets.read") + if match.group(1) not in project.buckets: + raise _Refusal(404, "storage_bucket_not_found", "Bucket not found") + return 200, {"$id": match.group(1)} + if path == "/v1/users" and method == "GET": + need("users.read") + return 200, {"total": 0, "users": []} + if path == "/v1/webhooks": + if method == "GET": + need("webhooks.read") + return 200, self._list(project, query) + need("webhooks.write") + return 201, self._create(project, payload) + if match := re.fullmatch(r"/v1/webhooks/([^/]+)(/secret)?", path): + id, secret = match.groups() + document = project.webhooks.get(id) + need("webhooks.read" if method == "GET" else "webhooks.write") + if document is None: + raise _Refusal(404, "webhook_not_found", "Webhook not found") + if method == "GET" and not secret: + return 200, _model(id, document) + if method == "PUT" and not secret: + return 200, self._update(document, payload, id) + if method == "PATCH" and secret: + document["signatureKey"] = payload.get("secret") or secrets.token_hex( + 64 + ) + document["$updatedAt"] = _stamp() + return 200, _model(id, document, secret=True) + if method == "DELETE" and not secret: + del project.webhooks[id] + return 204, None + raise _Refusal(404, "general_route_not_found", f"No route for {path}") + + def _list(self, project: Project, query: Mapping[str, list[str]]) -> Any: + limit, offset = 25, 0 + for key, values in query.items(): + if key.startswith("queries["): + parsed = json.loads(values[0]) + if parsed["method"] == "limit": + limit = parsed["values"][0] + elif parsed["method"] == "offset": + offset = parsed["values"][0] + documents = list(project.webhooks.items()) + page = documents[offset : offset + limit] + return { + "total": len(documents), + "webhooks": [_model(id, document) for id, document in page], + } + + def _create(self, project: Project, payload: Mapping[str, Any]) -> Any: + id = payload.get("webhookId", "") + if not CUSTOM_ID.fullmatch(id): + raise _Refusal(400, "general_argument_invalid", "Invalid webhookId") + if id in project.webhooks: + raise _Refusal(409, "webhook_already_exists", "Webhook already exists") + if ( + project.webhook_limit is not None + and len(project.webhooks) >= project.webhook_limit + ): + raise _Refusal( + 403, + "additional_resource_not_allowed", + "The maximum number of webhooks allowed for the selected plan " + "has reached. Upgrade to increase the limit.", + ) + secret = payload.get("secret") or secrets.token_hex(64) + _check(payload, secret) + stamp = _stamp() + document = { + "$createdAt": stamp, + "$updatedAt": stamp, + "name": payload["name"], + "url": payload["url"], + "events": list(payload["events"]), + "tls": payload.get("tls", False), + "authUsername": payload.get("authUsername", ""), + "httpPass": payload.get("authPassword", ""), + "signatureKey": secret, + "enabled": payload.get("enabled", True), + "logs": "", + "attempts": 0, + } + project.webhooks[id] = document + return _model(id, document, secret=True) + + def _update( + self, document: dict[str, Any], payload: Mapping[str, Any], id: str + ) -> Any: + _check(payload, None) + same = payload["url"] == document["url"] and ( + payload.get("tls", False) or not document["tls"] + ) + password = payload.get("authPassword") + if password is not None or not same: + document["httpPass"] = password or "" + enabled = payload.get("enabled", True) + document.update( + name=payload["name"], + url=payload["url"], + events=list(payload["events"]), + tls=payload.get("tls", False), + authUsername=payload.get("authUsername", ""), + enabled=enabled, + ) + if enabled: + document["attempts"] = 0 + document["$updatedAt"] = _stamp() + return _model(id, document) + + +def _check(payload: Mapping[str, Any], secret: str | None) -> None: + """Appwrite's parameter validators for a webhook write.""" + if len(payload.get("name", "")) > 128: + raise _Refusal(400, "general_argument_invalid", "Invalid name") + if len(payload.get("authPassword") or "") > AUTH_PASSWORD_MAX: + raise _Refusal(400, "general_argument_invalid", "Invalid authPassword") + if secret is not None and not 8 <= len(secret) <= 256: + raise _Refusal(400, "general_argument_invalid", "Invalid secret") + + +def _model(id: str, document: Mapping[str, Any], *, secret: bool = False) -> Any: + """The Webhook response model: the password is never returned, the + signing key only by create and the secret update.""" + return { + "$id": id, + "$createdAt": document["$createdAt"], + "$updatedAt": document["$updatedAt"], + "name": document["name"], + "url": document["url"], + "events": document["events"], + "tls": document["tls"], + "authUsername": document["authUsername"], + "authPassword": "", + "secret": document["signatureKey"] if secret else "", + "enabled": document["enabled"], + "logs": document["logs"], + "attempts": document["attempts"], + } diff --git a/tests/e2e/test_events_ingress.py b/tests/e2e/test_events_ingress.py index 8334b91..82cc156 100644 --- a/tests/e2e/test_events_ingress.py +++ b/tests/e2e/test_events_ingress.py @@ -7,11 +7,13 @@ every request with the official ``standardwebhooks`` library. Counters are read from the OTLP metrics the server exports (:class:`support.Collector`). -Until ``events/subscribe`` exists (PR 5 of #127), each test creates the webhook -the way subscribe will: the subscription sealed with the server's sealing keys -into ``authPassword``, the signing key derived from them as the webhook -``secret``, the event's Appwrite patterns as its events. Replace -:meth:`IngressFlow.subscribe` with a real ``events/subscribe`` call then. +The full loop through ``events/subscribe`` (webhook written by the server, +fired by Appwrite) is in ``test_events_subscribe.py``. The flows here build the +webhook by hand, exactly as subscribe writes it (the subscription sealed into +``authPassword``, the derived signing key as ``secret``, the event's Appwrite +patterns as its events), because they need states subscribe never produces: +expired or retired-key envelopes, events outside the catalog, and envelopes +copied, tampered with or sealed for another server. The servers under test use the production egress with two test settings, injected through ``build_app(ingress=...)``: loopback callbacks are allowed and @@ -57,6 +59,7 @@ Subscription, ) from mcp_server_appwrite.events.ingress import Ingress +from mcp_server_appwrite.events.webhooks import USERNAME FIXTURES = Path(__file__).parent / "fixtures" / "appwrite" PROJECT = "6630f1a2b3c4d5e6f7a8" @@ -184,8 +187,8 @@ def subscribe( server: str = PUBLIC_URL, target: Receiver | None = None, ) -> tuple[Webhook, str, str]: - """Create the webhook PR 5's ``events/subscribe`` will create, sealed - for ``server``. Returns it with the callback URL and the subscription's + """Create the webhook ``events/subscribe`` would create, sealed for + ``server``. Returns it with the callback URL and the subscription's secret, which the receiver checks each delivery against.""" secret = whsec() callback = (target or receiver).endpoint((secret,), replies) @@ -215,7 +218,7 @@ def subscribe( ), events=events, project=PROJECT, - user="mcp", + user=USERNAME, password=keyring.seal(subscription, server), ) return webhook, callback, secret diff --git a/tests/e2e/test_events_protocol.py b/tests/e2e/test_events_protocol.py index db375a4..4f21408 100644 --- a/tests/e2e/test_events_protocol.py +++ b/tests/e2e/test_events_protocol.py @@ -87,11 +87,6 @@ def test_client_discovers_and_lists_events(self): self.assertEqual(status, 400, message) self.assertEqual(message["error"]["code"], -32602) - for method in ("events/subscribe", "events/unsubscribe"): - with self.subTest(step=method): - _, message = client.call(method) - self.assertEqual(message["error"]["code"], -32601) - def check_catalog(self, result): self.assertEqual(result["resultType"], "complete") self.assertNotIn("nextCursor", result) @@ -175,8 +170,9 @@ def test_nothing_is_advertised_or_served(self): self.assertEqual(status, 200, message) self.assertNotIn("events", message["result"]["capabilities"]) - _, message = client.call("events/list") - self.assertEqual(message["error"]["code"], -32601) + for method in ("events/list", "events/subscribe", "events/unsubscribe"): + _, message = client.call(method) + self.assertEqual(message["error"]["code"], -32601, method) # The Appwrite webhook ingress is not mounted either. response = httpx.post( diff --git a/tests/e2e/test_events_subscribe.py b/tests/e2e/test_events_subscribe.py new file mode 100644 index 0000000..c810925 --- /dev/null +++ b/tests/e2e/test_events_subscribe.py @@ -0,0 +1,1005 @@ +"""``events/subscribe`` and ``events/unsubscribe`` through the real hosted server. + +Every party around the server runs on localhost: :class:`support.Cloud` is the +Appwrite REST API the server calls with the caller's token, the subscriber's +HTTPS endpoint (:class:`support.Receiver`) answers the verification handshake +and checks every request with the official ``standardwebhooks`` library, and +Appwrite's webhooks worker (:meth:`support.Appwrite.fire`) delivers from the +webhooks the server stored in :class:`support.Cloud`, with their real +``authPassword``, secret and URL. That closes the loop: subscribe, handshake, +webhook stored, Appwrite event, ingress, signed delivery, unsubscribe, webhook +gone, no delivery. + +Like the ingress flows, the server uses the production ingress with an egress +that may reach the local receiver and trusts its certificate, and a 1 s +request timeout. +""" + +import base64 +import json +import re +import secrets +import socket +import time +import unittest +from datetime import datetime +from pathlib import Path +from typing import Any + +import httpx +from jsonschema import Draft202012Validator +from support import ( + CLIENT_ID, + GRANTED, + ISSUER, + OTHER_SUBJECT, + OTHER_TOKEN, + PUBLIC_URL, + SCOPES, + SUBJECT, + SUBJECTLESS_TOKEN, + SUBJECTS, + TOKEN, + Appwrite, + Client, + Cloud, + Project, + Receiver, + Reply, + Server, + collector, + whsec, +) + +from mcp_server_appwrite.auth import public_base_url +from mcp_server_appwrite.events.catalog import lookup +from mcp_server_appwrite.events.delivery import Dispatcher, RetryPolicy +from mcp_server_appwrite.events.egress import Egress +from mcp_server_appwrite.events.envelope import ( + Context, + Keyring, + Principal, + key_id, + subscription_id, + timestamp, +) +from mcp_server_appwrite.events.ingress import Ingress + +FIXTURES = Path(__file__).parent / "fixtures" / "appwrite" +KEYS = "k1:" + base64.b64encode(bytes(range(64, 96))).decode() +ROTATED = "k2:" + base64.b64encode(bytes(range(128, 160))).decode() +"""The next sealing key of a rotation.""" +TIMEOUT = 1.0 +SETTLE = 0.5 +"""Seconds to wait before asserting that nothing (more) was received.""" + +MAIN = "6630f1a2b3c4d5e6f7a8" +FREE = "7740f1a2b3c4d5e6f7a8" +"""A Free-plan project: at most two webhooks.""" +TIDY = "8850f1a2b3c4d5e6f7a8" +"""A project with other people's webhooks in it.""" +MISSING = "9960f1a2b3c4d5e6f7a8" +"""A project that does not exist.""" + +READ_ONLY = "e2e-read-only" +"""Granted every scope but ``webhooks.read`` / ``webhooks.write``.""" +NO_FUNCTIONS = "e2e-no-functions" +"""Granted every scope but ``functions.read``.""" +STRANGER = "e2e-stranger" +"""A user who is not a member of any project.""" +NO_WRITE = "e2e-no-webhook-write" +"""Granted, and carrying in its ``scope`` claim, every scope but +``webhooks.write``.""" + +PRINCIPAL = Principal(ISSUER, SUBJECT, CLIENT_ID).digest +OTHER_PRINCIPAL = Principal(ISSUER, OTHER_SUBJECT, CLIENT_ID).digest + +ROWS = ("tablesdb.row.created", {"database_id": "main", "table_id": "support_tickets"}) +ROW_EVENT = "tablesdb.main.tables.support_tickets.rows.row_8f2a.create" + +CASES = [ + ("functions.execution.failed", {"function_id": "fn_billing"}, "execution_sync_failed", "functions.fn_billing.executions.68e7a1b2c3d4e5f6a7b8.create", "/v1/functions/fn_billing"), + ("functions.deployment.completed", {"function_id": "fn_billing", "status": "ready"}, "deployment_function_ready", "functions.fn_billing.deployments.68e79f00aa11bb22cc33.update", "/v1/functions/fn_billing"), + ("sites.deployment.completed", {"site_id": "site_docs"}, "deployment_site_ready", "sites.site_docs.deployments.68e7c000aa11bb22cc55.update", "/v1/sites/site_docs"), + (*ROWS, "row_created", ROW_EVENT, "/v1/tablesdb/main/tables/support_tickets"), + ("storage.file.created", {"bucket_id": "uploads"}, "file_created", "buckets.uploads.files.file_invoice.create", "/v1/storage/buckets/uploads"), + ("users.user.created", {}, "user_created", "users.user_jane.create", "/v1/users"), +] # fmt: skip +"""Per catalog event: arguments, the recorded Appwrite body and event that +should reach the subscriber, and the read that authorizes the subscription.""" + +LABEL = re.compile( + r"MCP events \| (?P[a-z.]+) \| until " + r"(?P\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ) \| owner (?P[A-Za-z0-9_-]{8})" +) + +SUBSCRIPTIONS = "mcp.events.subscriptions" + +INVALID_PARAMS = -32602 +INTERNAL_ERROR = -32603 +NOT_FOUND = -32011 +FORBIDDEN = -32012 +RESOURCE_EXHAUSTED = -32013 +UNSUPPORTED = -32014 +CALLBACK_ENDPOINT = -32015 + +receiver: Receiver +untrusted: Receiver +cloud: Cloud + + +def setUpModule() -> None: + global receiver, untrusted, cloud + receiver = Receiver() + untrusted = Receiver() + cloud = Cloud() + SUBJECTS.update( + { + READ_ONLY: "88d3e4f5a6b7c8d9e0f1", + NO_FUNCTIONS: "99e4f5a6b7c8d9e0f1a2", + STRANGER: "aaf5a6b7c8d9e0f1a2b3", + NO_WRITE: "bb06b7c8d9e0f1a2b3c4", + } + ) + # The main user's token names its scopes, so subscribe's write-scope check + # passes on a real grant; the others carry no scope claim. + GRANTED[TOKEN] = sorted(f"project:{scope}" for scope in SCOPES) + GRANTED[NO_WRITE] = sorted( + f"project:{scope}" for scope in SCOPES - {"webhooks.write"} + ) + cloud.add( + Project( + MAIN, + functions={"fn_billing"}, + sites={"site_docs"}, + tables={("main", "support_tickets"), ("main", "orders")}, + buckets={"uploads"}, + ) + ) + cloud.add( + Project( + FREE, + tables={("main", "tickets"), ("main", "orders")}, + webhook_limit=2, + ) + ) + cloud.add(Project(TIDY, tables={("main", "support_tickets")})) + cloud.grant(TOKEN, {MAIN, FREE, TIDY}) + cloud.grant(OTHER_TOKEN, {MAIN, TIDY}) + cloud.grant(READ_ONLY, {MAIN}, SCOPES - {"webhooks.read", "webhooks.write"}) + cloud.grant(NO_FUNCTIONS, {MAIN}, SCOPES - {"functions.read"}) + cloud.grant(STRANGER, set()) + cloud.grant(NO_WRITE, {MAIN}, SCOPES - {"webhooks.write"}) + + +def tearDownModule() -> None: + receiver.close() + untrusted.close() + cloud.close() + for token in (READ_ONLY, NO_FUNCTIONS, STRANGER, NO_WRITE): + SUBJECTS.pop(token, None) + GRANTED.clear() + + +def fixture(name: str) -> dict[str, Any]: + return json.loads((FIXTURES / f"{name}.json").read_text()) + + +def injected() -> dict[str, Any]: + """``build_app`` arguments: the production ingress, keyring and public URL + from the environment, with an egress that may reach the local receiver.""" + egress = Egress(allow_loopback=True, ssl_context=receiver.trust(), timeout=TIMEOUT) + policy = RetryPolicy(delays=(0.3,), jitter=0.0) + return { + "ingress": Ingress( + Keyring.from_env(), + egress, + Dispatcher(egress, policy), + base=public_base_url(), + ) + } + + +def environment() -> dict[str, str]: + return { + "MCP_EVENTS": "1", + # The new key is already in the ring, behind the active one: step 1 + # of a rotation, so these servers open what a step-2 replica seals. + "MCP_EVENTS_SEALING_KEYS": f"{KEYS},{ROTATED}", + "APPWRITE_ENDPOINT": cloud.endpoint, + } + + +def request( + name: str, + arguments: dict[str, str], + url: str, + secret: str | None, + *, + project: str = MAIN, + **extra: Any, +) -> dict[str, Any]: + """``events/subscribe`` params as ChatGPT sends them.""" + delivery: dict[str, Any] = {"mode": "webhook", "url": url} + if secret is not None: + delivery["secret"] = secret + return { + "name": name, + "arguments": {"project_id": project, **arguments}, + "delivery": delivery, + "cursor": None, + **extra, + } + + +def target( + name: str, arguments: dict[str, str], url: str, *, project: str = MAIN +) -> dict[str, Any]: + """``events/unsubscribe`` params.""" + return { + "name": name, + "arguments": {"project_id": project, **arguments}, + "delivery": {"url": url}, + } + + +def moment(text: str) -> float: + return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp() + + +def label(event: str, until: float, owner: str) -> str: + """A managed webhook name, as an earlier subscribe would have written it.""" + stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(until)) + return f"MCP events | {event} | until {stamp} | owner {owner}" + + +class SubscribeFlow(unittest.TestCase): + """A client subscribes, receives events, refreshes and unsubscribes.""" + + server: Server + appwrite: Appwrite + client: Client + + @classmethod + def setUpClass(cls) -> None: + cls.server = Server(environment(), injected).__enter__() + cls.appwrite = Appwrite(cls.server) + + @classmethod + def tearDownClass(cls) -> None: + cls.appwrite.close() + cls.server.__exit__(None, None, None) + + def call(self, method: str, params: dict[str, Any], token: str) -> dict[str, Any]: + """The result, without the server info the SDK puts in ``_meta``.""" + with Client(self.server.url, token) as client: + status, message = client.call(method, params) + self.assertEqual(status, 200, message) + self.assertNotIn("error", message) + return {k: v for k, v in message["result"].items() if k != "_meta"} + + def subscribe( + self, params: dict[str, Any], *, token: str = TOKEN + ) -> dict[str, Any]: + return self.call("events/subscribe", params, token) + + def refused( + self, method: str, params: dict[str, Any], code: int, *, token: str = TOKEN + ) -> dict[str, Any]: + with Client(self.server.url, token) as client: + _, message = client.call(method, params) + self.assertIn("error", message, message) + self.assertEqual(message["error"]["code"], code, message) + return message["error"] + + def schemas(self) -> dict[str, Draft202012Validator]: + """Validators for the ``inputSchema`` each event publishes.""" + events = self.call("events/list", {}, TOKEN)["events"] + return {e["name"]: Draft202012Validator(e["inputSchema"]) for e in events} + + def unsubscribe(self, params: dict[str, Any], *, token: str = TOKEN) -> None: + result = self.call("events/unsubscribe", params, token) + self.assertEqual(result, {"resultType": "complete"}) + + def granted(self, result: dict[str, Any], ttl: float, started: float) -> None: + """``refreshBefore`` is 9/10 of the ``ttl`` (seconds) after ``started``.""" + refresh = moment(result["refreshBefore"]) + self.assertRegex( + result["refreshBefore"], r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z$" + ) + self.assertAlmostEqual(refresh, started + ttl * 0.9, delta=5) + + def ours(self, project: str = MAIN) -> dict[str, dict[str, Any]]: + """The stored webhooks this server manages in ``project``.""" + return { + id: document + for id, document in cloud.documents(project).items() + if LABEL.fullmatch(document["name"]) + } + + def accepted(self, response) -> str: + self.assertEqual(response.status_code, 200, response.text) + body = response.json() + self.assertEqual(body["status"], "accepted", body) + return body["eventId"] + + def nothing_more(self, callback: str, count: int) -> None: + time.sleep(SETTLE) + self.assertEqual(len(receiver.received(callback)), count, callback) + + def test_subscribe_receive_refresh_rotate_unsubscribe(self): + counts = { + outcome: collector().count(SUBSCRIPTIONS, outcome=outcome) + for outcome in ("created", "refreshed", "removed", "absent") + } + old, new = whsec(), whsec(64) + callback = receiver.endpoint((old, new)) + name, arguments = ROWS + + # Subscribe: the callback is verified, then the webhook is stored. + started = time.time() + result = self.subscribe(request(name, arguments, callback, old)) + self.assertEqual( + set(result), {"id", "refreshBefore", "cursor", "truncated", "resultType"} + ) + id = result["id"] + self.assertRegex(id, r"^sub_[0-9a-f]{32}$") + self.assertEqual( + id, + subscription_id( + PRINCIPAL, callback, name, {"project_id": MAIN, **arguments} + ), + ) + self.assertIsNone(result["cursor"]) + self.assertFalse(result["truncated"]) + self.assertEqual(result["resultType"], "complete") + self.granted(result, 3600, started) + + handshake = receiver.wait(callback, 1)[0] + challenge = handshake.json() + self.assertEqual(set(challenge), {"type", "challenge"}) + self.assertEqual(challenge["type"], "verification") + self.assertRegex(handshake.headers["webhook-id"], r"^msg_verification_") + self.assertEqual(handshake.headers["x-mcp-subscription-id"], id) + self.assertEqual(handshake.verified, (True, False)) + + webhook = cloud.documents(MAIN)[id] + self.assertEqual(webhook["url"], f"{PUBLIC_URL}/appwrite/webhooks/{id}") + self.assertEqual( + webhook["events"], ["tablesdb.main.tables.support_tickets.rows.*.create"] + ) + self.assertTrue(webhook["tls"]) + self.assertTrue(webhook["enabled"]) + self.assertEqual(webhook["authUsername"], "mcp-events") + self.assertEqual(webhook["signatureKey"], Keyring.parse(KEYS).signing_key(id)) + # The password is the JWE envelope, sealed with k1 for this server, the + # project and the webhook id, and carrying the granted lease. + self.assertEqual(key_id(webhook["httpPass"]), "k1") + sealed = Keyring.parse(KEYS).open( + webhook["httpPass"], Context(server=PUBLIC_URL, tenant=MAIN, resource=id) + ) + self.assertEqual( + (sealed.id, sealed.callback, sealed.secret), (id, callback, old) + ) + self.assertEqual(timestamp(sealed.refresh), result["refreshBefore"]) + self.assertAlmostEqual(sealed.expires / 1000, started + 3600, delta=5) + match = LABEL.fullmatch(webhook["name"]) + assert match is not None, webhook["name"] + self.assertEqual(match["event"], name) + self.assertEqual(match["owner"], PRINCIPAL[:8]) + self.assertAlmostEqual(moment(match["until"]), started + 3600, delta=5) + # The client's secret and callback are only inside the sealed envelope. + for text in (old, callback): + self.assertNotIn(text, json.dumps(webhook)) + # Writes were made with the caller's own token. + self.assertTrue( + all(token == TOKEN for method, path, token in cloud.requests[-4:]) + ) + + # Appwrite fires the stored webhook; the event reaches the callback. + responses = self.appwrite.fire(cloud, MAIN, ROW_EVENT, fixture("row_created")) + self.assertEqual(len(responses), 1) + event_id = self.accepted(responses[0]) + delivery = receiver.wait(callback, 2)[1] + self.assertEqual(delivery.headers["webhook-id"], event_id) + self.assertEqual(delivery.headers["x-mcp-subscription-id"], id) + self.assertEqual(delivery.verified, (True, False)) + self.assertEqual(delivery.json()["data"]["row_id"], "row_8f2a") + # Not an event for this table: Appwrite does not fire the webhook. + other = "tablesdb.main.tables.orders.rows.row_1.create" + self.assertEqual( + self.appwrite.fire(cloud, MAIN, other, fixture("row_created")), [] + ) + + # Refresh: same id, one webhook, a new grant each time. Suggestions + # are clamped to 5 minutes .. 24 hours; null gets the default hour. + # The order of the arguments does not change the subscription. + managed = set(self.ours()) + reordered = {"project_id": MAIN, **dict(reversed(arguments.items()))} + reordered = dict(reversed(reordered.items())) + for suggestion, ttl in ((60_000, 300), (10 * 86_400_000, 86_400), (None, 3600)): + with self.subTest(ttlMs=suggestion): + started = time.time() + params = request(name, arguments, callback, old, ttlMs=suggestion) + if suggestion is None: + params["arguments"] = reordered + result = self.subscribe(params) + self.assertEqual(result["id"], id) + self.granted(result, ttl, started) + self.assertEqual(set(self.ours()), managed) + until = LABEL.fullmatch(cloud.documents(MAIN)[id]["name"]) + assert until is not None + self.assertAlmostEqual(moment(until["until"]), started + ttl, delta=5) + # Every refresh verifies the callback again (no verification cache), + # each time with a fresh challenge. + handshakes = [ + r.json()["challenge"] + for r in receiver.received(callback) + if r.json().get("type") == "verification" + ] + self.assertEqual(len(handshakes), 4) + self.assertEqual(len(set(handshakes)), 4) + + # Rotate: the new secret replaces the old one, for the handshake and + # for every later delivery. + result = self.subscribe(request(name, arguments, callback, new)) + self.assertEqual(result["id"], id) + self.assertEqual(receiver.wait(callback, 6)[5].verified, (False, True)) + responses = self.appwrite.fire(cloud, MAIN, ROW_EVENT, fixture("row_created")) + self.accepted(responses[0]) + self.assertEqual(receiver.wait(callback, 7)[6].verified, (False, True)) + + # Another user's unsubscribe for the same event and URL is a different + # subscription id: nothing of ours is touched. + self.unsubscribe(target(name, arguments, callback), token=OTHER_TOKEN) + self.assertIn(id, cloud.documents(MAIN)) + + # Unsubscribe: the webhook is gone, so Appwrite has nothing to fire. + self.unsubscribe(target(name, arguments, callback)) + self.assertNotIn(id, cloud.documents(MAIN)) + self.assertEqual( + self.appwrite.fire(cloud, MAIN, ROW_EVENT, fixture("row_created")), [] + ) + # Idempotent: unsubscribing again is still {}. + self.unsubscribe(target(name, arguments, callback)) + self.nothing_more(callback, 7) + + for outcome, increase in ( + ("created", 1), + ("refreshed", 4), + ("removed", 1), + ("absent", 2), + ): + with self.subTest(counter=outcome): + self.assertEqual( + collector().count(SUBSCRIPTIONS, outcome=outcome), + counts[outcome] + increase, + ) + + def test_refreshes_interleaved_across_a_key_rotation_still_deliver(self): + # Step 2 of a rotation, rolling out: this server (the old replica, + # ring k1,k2) still seals with k1; the new replica (ring k2,k1) seals + # with k2. + secret = whsec() + callback = receiver.endpoint((secret,)) + params = request(*ROWS, callback, secret) + id = self.subscribe(params)["id"] + rotated = {**environment(), "MCP_EVENTS_SEALING_KEYS": f"{ROTATED},{KEYS}"} + k2 = Keyring.parse(f"{ROTATED},{KEYS}").signing_key(id) + with Server(rotated, injected) as replica: + + def refresh(server: Server) -> dict[str, Any]: + with Client(server.url, TOKEN) as client: + _, message = client.call("events/subscribe", params) + return message + + # A refresh writes the signing key before the envelope, so when + # the envelope write fails the k1 envelope stays with the k2 + # signing key, never a k2 envelope with a k1 signing key that + # outlives k1 in the ring. + cloud.fail("PUT", rf"/v1/webhooks/{id}", 500) + message = refresh(replica) + self.assertEqual(message["error"]["code"], INTERNAL_ERROR, message) + stored = cloud.documents(MAIN)[id] + self.assertEqual(key_id(stored["httpPass"]), "k1") + self.assertEqual(stored["signatureKey"], k2) + + # Both replicas refresh the same subscription and their writes + # interleave: new signing key, old signing key, old envelope, new + # envelope last. The k2 envelope keeps the k1 signing key, which + # is why step 3 waits a full TTL after step 2 reached every + # replica: no step-1 replica writes a k1 signing key after that. + def old_replica_refreshes() -> None: + message = refresh(self.server) + assert "error" not in message, message + + cloud.before("PUT", rf"/v1/webhooks/{id}", old_replica_refreshes) + message = refresh(replica) + self.assertNotIn("error", message, message) + stored = cloud.documents(MAIN)[id] + self.assertEqual(key_id(stored["httpPass"]), "k2") + self.assertEqual( + stored["signatureKey"], Keyring.parse(KEYS).signing_key(id) + ) + + # The load balancer can hand Appwrite's delivery to either + # replica. Both open the k2 envelope (the old replica holds k2 + # since step 1) and both accept the k1 signing key. + received = 4 # the handshakes of the subscribe and three refreshes + for server in (replica, self.server): + with self.subTest(replica=server.url): + appwrite = Appwrite(server) + responses = appwrite.fire( + cloud, MAIN, ROW_EVENT, fixture("row_created") + ) + appwrite.close() + self.assertEqual(len(responses), 1) + event_id = self.accepted(responses[0]) + received += 1 + delivery = receiver.wait(callback, received)[received - 1] + self.assertEqual(delivery.headers["webhook-id"], event_id) + self.assertEqual(delivery.verified, (True,)) + self.unsubscribe(target(*ROWS, callback)) + self.nothing_more(callback, 6) + + def test_every_event_subscribes_and_is_authorized(self): + managed = set(self.ours()) + schemas = self.schemas() + ids = set() + # Secrets span the 24..64 bytes Standard Webhooks allows. + sizes = (24, 32, 64, 24, 32, 64) + for (name, arguments, body, event, read), size in zip( + CASES, sizes, strict=True + ): + with self.subTest(event=name): + secret = whsec(size) + callback = receiver.endpoint((secret,)) + # What the server accepts, the published schema accepts. + self.assertTrue( + schemas[name].is_valid({"project_id": MAIN, **arguments}) + ) + reads = cloud.calls("GET", read) + result = self.subscribe(request(name, arguments, callback, secret)) + ids.add(result["id"]) + # The resource was read with the caller's token first. + self.assertEqual(cloud.calls("GET", read), reads + 1) + webhook = cloud.documents(MAIN)[result["id"]] + self.assertEqual( + webhook["events"], + lookup(name).patterns({"project_id": MAIN, **arguments}), + ) + responses = self.appwrite.fire(cloud, MAIN, event, fixture(body)) + self.assertEqual(len(responses), 1) + event_id = self.accepted(responses[0]) + received = receiver.wait(callback, 2)[1] + self.assertEqual(received.headers["webhook-id"], event_id) + self.assertEqual(received.json()["name"], name) + self.unsubscribe(target(name, arguments, callback)) + self.assertEqual(len(ids), len(CASES)) + + # The resource must exist: NotFound, kind "resource". + secret = whsec() + callback = receiver.endpoint((secret,)) + for name, arguments in ( + ("functions.execution.failed", {"function_id": "fn_gone"}), + ("sites.deployment.completed", {"site_id": "site_gone"}), + ("tablesdb.row.created", {"database_id": "main", "table_id": "gone"}), + ("tablesdb.row.created", {"database_id": "gone", "table_id": "orders"}), + ("storage.file.created", {"bucket_id": "gone"}), + ): + with self.subTest(missing=arguments): + error = self.refused( + "events/subscribe", + request(name, arguments, callback, secret), + NOT_FOUND, + ) + self.assertEqual(error["data"], {"kind": "resource"}) + + # The caller must be able to read it: Forbidden, whether they lack + # the project, the scope, or the project does not exist at all. + execution = ("functions.execution.failed", {"function_id": "fn_billing"}) + error = self.refused( + "events/subscribe", + request(*execution, callback, secret), + FORBIDDEN, + token=STRANGER, + ) + self.assertIn(MAIN, error["message"]) + error = self.refused( + "events/subscribe", + request(*execution, callback, secret), + FORBIDDEN, + token=NO_FUNCTIONS, + ) + self.assertIn("project:functions.read", error["message"]) + self.refused( + "events/subscribe", + request(*execution, callback, secret, project=MISSING), + FORBIDDEN, + ) + # Unsubscribe from a project that does not exist is Forbidden too, + # not an internal error, and not the idempotent {} of a missing + # webhook. + self.refused( + "events/unsubscribe", + target(*execution, callback, project=MISSING), + FORBIDDEN, + ) + # A token with no user behind it has no principal to subscribe for. + self.refused( + "events/subscribe", + request(*execution, callback, secret), + FORBIDDEN, + token=SUBJECTLESS_TOKEN, + ) + self.refused( + "events/unsubscribe", + target(*execution, callback), + FORBIDDEN, + token=SUBJECTLESS_TOKEN, + ) + # No token at all never reaches JSON-RPC. + response = httpx.post( + f"{self.server.url}/", + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "events/subscribe", + "params": request(*ROWS, callback, secret), + }, + headers={"Accept": "application/json, text/event-stream"}, + ) + self.assertEqual(response.status_code, 401) + # None of the refusals contacted the callback or wrote a webhook. + self.nothing_more(callback, 0) + self.assertEqual(set(self.ours()), managed) + + def test_invalid_requests_are_refused_before_anything_is_sent(self): + secret = whsec() + callback = receiver.endpoint((secret,)) + name, arguments = ROWS + base = request(name, arguments, callback, secret) + requests = len(cloud.requests) + + def changed(**delivery: Any) -> dict[str, Any]: + return {**base, "delivery": {**base["delivery"], **delivery}} + + key = base64.b64encode(b"k" * 32).decode() + invalid = { + "unknown argument": request(name, {**arguments, "extra": "x"}, callback, secret), + "no project": {**base, "arguments": dict(arguments)}, + "arguments not an object": {**base, "arguments": ["main"]}, + "no delivery": {**base, "delivery": None}, + "no url": changed(url=None), + "http url": changed(url=callback.replace("https://", "http://")), + "ftp url": changed(url="ftp://hooks.example.com/"), + "file url": changed(url="file:///etc/passwd"), + "url without host": changed(url="https:///path"), + "not a url": changed(url="not a url"), + "private url": changed(url="https://10.0.0.7/hooks"), + "metadata url": changed(url="https://169.254.169.254/latest"), + "credentials in url": changed(url=callback.replace("https://", "https://u:p@")), + "fragment in url": changed(url=callback + "#events"), + "no secret": changed(secret=None), + "secret not a string": changed(secret=42), + "empty secret": changed(secret=""), + "bare prefix": changed(secret="whsec_"), + "secret without prefix": changed(secret=key), + "uppercase prefix": changed(secret="WHSEC_" + key), + "space after prefix": changed(secret="whsec_ " + key), + "secret not base64": changed(secret="whsec_not*base64!"), + "secret bad padding": changed(secret="whsec_" + key[:-1]), + # Decodes, but the subscription record wants canonical base64. + "secret not canonical": changed(secret="whsec_" + key[:-2] + "t="), + "secret of 23 bytes": changed(secret="whsec_" + base64.b64encode(b"k" * 23).decode()), + "secret of 65 bytes": changed(secret="whsec_" + base64.b64encode(b"k" * 65).decode()), + "envelope too large": changed(url=callback + "?pad=" + "a" * 2000), + "ttl not a number": {**base, "ttlMs": "1h"}, + "ttl negative": {**base, "ttlMs": -1}, + "maxAge negative": {**base, "maxAgeMs": -5}, + } # fmt: skip + # Arguments the published input schema must reject as well. + schema_rejects = {"unknown argument", "no project", "arguments not an object"} + # IDs that would widen an Appwrite event pattern, or are not IDs, in + # every ID argument of every event; statuses outside the choices. + ids: tuple[object, ...] = ("*", "a*", "a.b", "fn1.executions", "", "-lead", "_lead", "a b", "a" * 37, None, 7) # fmt: skip + statuses: tuple[object, ...] = ("building", "READY", "*", None, 7) + for event_name, event_arguments, *_ in CASES: + full = {"project_id": MAIN, **event_arguments} + for argument in full: + for bad in statuses if argument == "status" else ids: + params = request(event_name, {}, callback, secret) + params["arguments"] = {**full, argument: bad} + label_ = f"{event_name} {argument}={bad!r}" + invalid[label_] = params + schema_rejects.add(label_) + schemas = self.schemas() + for label_, params in invalid.items(): + with self.subTest(invalid=label_): + self.refused("events/subscribe", params, INVALID_PARAMS) + if label_ in schema_rejects: + # The server refuses what the published schema rejects. + schema = schemas[params["name"]] + self.assertFalse(schema.is_valid(params["arguments"])) + + error = self.refused( + "events/subscribe", {**base, "name": "tablesdb.row.deleted"}, NOT_FOUND + ) + self.assertEqual(error["data"], {"kind": "event"}) + self.refused( + "events/unsubscribe", + {**target(*ROWS, callback), "name": "nope"}, + NOT_FOUND, + ) + self.refused( + "events/unsubscribe", + target(name, {**arguments, "table_id": "*"}, callback), + INVALID_PARAMS, + ) + for mode in ("poll", "push"): + with self.subTest(mode=mode): + self.refused("events/subscribe", changed(mode=mode), UNSUPPORTED) + + # Nothing reached Appwrite or the callback. + self.assertEqual(len(cloud.requests), requests) + self.nothing_more(callback, 0) + + def test_handshake_failures_store_nothing(self): + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + closed = probe.getsockname()[1] + secret = whsec() + elsewhere = receiver.endpoint((secret,)) + cases = [ + # An endpoint that cannot verify the request does not echo. + ("challenge_failed", receiver.endpoint((whsec(),))), + # A redirect is never followed, even to an endpoint that would echo. + ("challenge_failed", receiver.endpoint((secret,), (Reply(307, location=elsewhere),))), + ("http_4xx", receiver.endpoint((secret,), (Reply(404),))), + ("http_5xx", receiver.endpoint((secret,), (Reply(503),))), + ("timeout", receiver.endpoint((secret,), (Reply(delay=TIMEOUT + 1),))), + ("tls_error", untrusted.endpoint((secret,))), + ("connection_refused", f"https://localhost:{closed}/hooks/gone"), + ] # fmt: skip + managed = set(self.ours()) + for reason, callback in cases: + with self.subTest(reason=reason, callback=callback): + error = self.refused( + "events/subscribe", + request(*ROWS, callback, secret), + CALLBACK_ENDPOINT, + ) + self.assertEqual(error["data"], {"reason": reason}) + # Raw endpoint responses never surface. + self.assertNotIn('"ok"', error["message"]) + self.assertEqual(set(self.ours()), managed) + self.nothing_more(elsewhere, 0) + + def test_appwrite_refusals_leave_no_partial_state(self): + secret = whsec() + callback = receiver.endpoint((secret,)) + managed = set(self.ours()) + internal = collector().count("mcp.jsonrpc.errors", error_code="-32603") + forbidden = collector().count( + SUBSCRIPTIONS, operation="subscribe", outcome="error", reason="-32012" + ) + + # Without the webhook scopes the request stops before the handshake. + error = self.refused( + "events/subscribe", + request(*ROWS, callback, secret), + FORBIDDEN, + token=READ_ONLY, + ) + self.assertIn("project:webhooks.read", error["message"]) + self.assertIn("project:webhooks.write", error["message"]) + self.refused( + "events/unsubscribe", target(*ROWS, callback), FORBIDDEN, token=READ_ONLY + ) + # A token that may list webhooks but not write them is refused from + # its scope claim before Appwrite or the callback hears of it. + requests = len(cloud.requests) + error = self.refused( + "events/subscribe", + request(*ROWS, callback, secret), + FORBIDDEN, + token=NO_WRITE, + ) + self.assertIn("project:webhooks.write", error["message"]) + self.assertEqual(len(cloud.requests), requests) + self.nothing_more(callback, 0) + + # Appwrite failing on its side is an internal error, and leaves nothing. + for method, path in ( + ("GET", r"/v1/tablesdb/main/tables/support_tickets"), + ("GET", r"/v1/webhooks"), + ("POST", r"/v1/webhooks"), + ): + with self.subTest(failing=f"{method} {path}"): + cloud.fail(method, path, 503) + self.refused( + "events/subscribe", + request(*ROWS, callback, secret), + INTERNAL_ERROR, + ) + self.assertEqual(set(self.ours()), managed) + + # A refresh whose webhook update fails is an internal error, but the + # webhook stays: its previous envelope still opens and the signing key + # written before it verifies, so a transient failure does not end the + # subscription. + result = self.subscribe(request(*ROWS, callback, secret)) + cloud.fail("PUT", r"/v1/webhooks/[^/]+", 500) + self.refused( + "events/subscribe", request(*ROWS, callback, secret), INTERNAL_ERROR + ) + self.assertIn(result["id"], cloud.documents(MAIN)) + seen = len(receiver.received(callback)) + responses = self.appwrite.fire(cloud, MAIN, ROW_EVENT, fixture("row_created")) + self.assertEqual(len(responses), 1) + event_id = self.accepted(responses[0]) + delivery = receiver.wait(callback, seen + 1)[seen] + self.assertEqual(delivery.headers["webhook-id"], event_id) + self.unsubscribe(target(*ROWS, callback)) + + # Counted like tools/call: four internal errors, two refusals. + self.assertEqual( + collector().count("mcp.jsonrpc.errors", error_code="-32603"), internal + 4 + ) + self.assertEqual( + collector().count( + SUBSCRIPTIONS, operation="subscribe", outcome="error", reason="-32012" + ), + forbidden + 2, + ) + + def test_free_plan_limit_and_expired_cleanup(self): + secret = whsec() + callback = receiver.endpoint((secret,)) + tickets = ( + "tablesdb.row.created", + {"database_id": "main", "table_id": "tickets"}, + ) + orders = ("tablesdb.row.created", {"database_id": "main", "table_id": "orders"}) + + # The Free project holds the user's own webhook and one of ours that + # expired an hour ago: the slot it holds is freed before creating. + cloud.store( + FREE, + { + "$id": "slack", + "name": "Slack alerts", + "url": "https://hooks.example.com/a", + }, + ) + expired = "sub_" + secrets.token_hex(16) + cloud.store( + FREE, + { + "$id": expired, + "name": label( + "tablesdb.row.created", time.time() - 3600, PRINCIPAL[:8] + ), + "url": f"{PUBLIC_URL}/appwrite/webhooks/{expired}", + }, + ) + cleaned = collector().count(SUBSCRIPTIONS, operation="cleanup") + result = self.subscribe(request(*tickets, callback, secret, project=FREE)) + self.assertEqual(set(cloud.documents(FREE)), {"slack", result["id"]}) + self.assertEqual( + collector().count(SUBSCRIPTIONS, operation="cleanup"), cleaned + 1 + ) + + # A second subscription does not fit: ResourceExhausted, nothing sent + # beyond the handshake, and the user's webhook is untouched. + error = self.refused( + "events/subscribe", + request(*orders, callback, secret, project=FREE), + RESOURCE_EXHAUSTED, + ) + self.assertEqual(error["data"], {"limit": "webhooks", "max": 2}) + self.assertIn("Pro", error["message"]) + self.assertIn("unsubscribe", error["message"]) + self.assertEqual(set(cloud.documents(FREE)), {"slack", result["id"]}) + + # Unsubscribing frees the slot. + self.unsubscribe(target(*tickets, callback, project=FREE)) + self.subscribe(request(*orders, callback, secret, project=FREE)) + self.unsubscribe(target(*orders, callback, project=FREE)) + self.assertEqual(set(cloud.documents(FREE)), {"slack"}) + + def test_webhooks_this_server_did_not_create_are_never_touched(self): + secret = whsec() + callback = receiver.endpoint((secret,)) + name, arguments = ROWS + now = time.time() + past, recent, future = now - 3600, now - 60, now + 3600 + ours_expired = "sub_" + secrets.token_hex(16) + ours_recent = "sub_" + secrets.token_hex(16) + ours_live = "sub_" + secrets.token_hex(16) + theirs_expired = "sub_" + secrets.token_hex(16) + renamed = "sub_" + secrets.token_hex(16) + # A long-expired subscription of ours whose client refreshes it while + # this subscribe is cleaning up. + late_secret = whsec() + late_callback = receiver.endpoint((late_secret,)) + late = subscription_id( + PRINCIPAL, late_callback, name, {"project_id": TIDY, **arguments} + ) + seeded = { + # The user's webhooks, including ones that copy our label, one + # with an id of the right shape but not hex. + "slack": "Slack alerts", + "lookalike": label(name, past, PRINCIPAL[:8]), + "sub_" + "z" * 32: label(name, past, PRINCIPAL[:8]), + renamed: "Renamed by a person", + # Another user's expired subscription, one of ours still live, and + # one of ours that expired within the 10 minute grace period. + theirs_expired: label(name, past, OTHER_PRINCIPAL[:8]), + ours_live: label(name, future, PRINCIPAL[:8]), + ours_recent: label(name, recent, PRINCIPAL[:8]), + late: label(name, past, PRINCIPAL[:8]), + # Ours and long expired: the only one cleanup removes. + ours_expired: label(name, past, PRINCIPAL[:8]), + } + for id, title in seeded.items(): + cloud.store(TIDY, {"$id": id, "name": title, "url": "https://x.example"}) + before = cloud.documents(TIDY) + + # The late refresh lands after cleanup listed the project, right + # before it re-reads that webhook to delete it. + def refresh() -> None: + self.subscribe( + request(name, arguments, late_callback, late_secret, project=TIDY) + ) + + cloud.before("GET", rf"/v1/webhooks/{late}", refresh) + result = self.subscribe( + request(name, arguments, callback, secret, project=TIDY) + ) + after = cloud.documents(TIDY) + self.assertEqual(set(after), set(seeded) - {ours_expired} | {result["id"]}) + for id in set(seeded) - {ours_expired, late}: + self.assertEqual(after[id], before[id], id) + # The refreshed subscription survived with its new expiry. + refreshed = LABEL.fullmatch(after[late]["name"]) + assert refreshed is not None + self.assertGreater(moment(refreshed["until"]), now) + self.unsubscribe(target(name, arguments, callback, project=TIDY)) + self.unsubscribe(target(name, arguments, late_callback, project=TIDY)) + + # A webhook that holds the subscription's id but not our name (renamed + # in the Console) is neither rewritten nor deleted. + id = subscription_id( + PRINCIPAL, callback, name, {"project_id": TIDY, **arguments} + ) + cloud.store(TIDY, {"$id": id, "name": "Renamed", "url": "https://x.example"}) + error = self.refused( + "events/subscribe", + request(name, arguments, callback, secret, project=TIDY), + FORBIDDEN, + ) + self.assertIn(id, error["message"]) + self.unsubscribe(target(name, arguments, callback, project=TIDY)) + self.assertEqual(cloud.documents(TIDY)[id]["name"], "Renamed") + + +class ProductionEgressFlow(unittest.TestCase): + """The server as production builds it refuses loopback callbacks.""" + + def test_loopback_callback_is_invalid(self): + secret = whsec() + callback = receiver.endpoint((secret,)) + requests = len(cloud.requests) + with Server(environment()) as server, server.client() as client: + _, message = client.call( + "events/subscribe", request(*ROWS, callback, secret) + ) + self.assertEqual(message["error"]["code"], INVALID_PARAMS, message) + self.assertEqual(len(cloud.requests), requests) + time.sleep(SETTLE) + self.assertEqual(receiver.received(callback), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_events_catalog.py b/tests/unit/test_events_catalog.py deleted file mode 100644 index 124e74a..0000000 --- a/tests/unit/test_events_catalog.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Server-side argument validation of the events catalog. - -Nothing reaches ``Event.patterns`` or ``lookup`` over HTTP until -``events/subscribe`` lands (PR 5 of #127). The published schemas are covered -end to end in ``tests/e2e/test_events_protocol.py``; replace this module with -the subscribe e2e flow once it exists. -""" - -import unittest - -from mcp_server_appwrite.events import errors -from mcp_server_appwrite.events.catalog import EVENTS, lookup -from mcp_server_appwrite.events.errors import EventsError - -VALID = { - "functions.execution.failed": {"function_id": "fn1"}, - "functions.deployment.completed": {"function_id": "fn1"}, - "sites.deployment.completed": {"site_id": "site-1"}, - "tablesdb.row.created": {"database_id": "db_1", "table_id": "tickets"}, - "storage.file.created": {"bucket_id": "uploads"}, - "users.user.created": {}, -} - -# Values that would widen an Appwrite event pattern or are not Appwrite IDs. -BAD_IDS = ["*", "a*", "a.b", "fn1.executions", "", "-lead", "_lead", "a b", "a" * 37] - - -class ArgumentValidationTests(unittest.TestCase): - def test_ids_can_never_widen_a_pattern(self): - for event in EVENTS: - arguments = {"project_id": "p1", **VALID[event.name]} - patterns = event.patterns(arguments) - self.assertTrue(patterns) - for pattern in patterns: - self.assertNotIn("p1", pattern.split(".")) - for argument in event.parameters: - if argument.choices: - continue - for bad in BAD_IDS: - with self.subTest(event=event.name, argument=argument.name): - with self.assertRaises(EventsError) as caught: - event.patterns({**arguments, argument.name: bad}) - self.assertEqual(caught.exception.code, errors.INVALID_PARAMS) - - def test_unknown_event_is_not_found(self): - with self.assertRaises(EventsError) as caught: - lookup("users.user.deleted") - self.assertEqual(caught.exception.code, errors.NOT_FOUND) - self.assertEqual(caught.exception.data, {"kind": "event"}) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/unit/test_events_delivery.py b/tests/unit/test_events_delivery.py index b207aa9..12bedbd 100644 --- a/tests/unit/test_events_delivery.py +++ b/tests/unit/test_events_delivery.py @@ -1,22 +1,17 @@ -"""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``. +"""The production retry schedule, whose real-time waits (30 s, 2 min, 8 min) +are too long to run end to end, and that a callback's ``repr`` never shows its +secret (nothing on the wire shows a ``repr``). + +Everything else about delivery is covered end to end. 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 go through the ingress in +``tests/e2e/test_events_ingress.py``. ``whsec_`` validation and every outcome +of the verification handshake go through ``events/subscribe`` in +``tests/e2e/test_events_subscribe.py``. """ import base64 -import hmac -import json import unittest import httpx @@ -35,25 +30,6 @@ ) -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 @@ -79,37 +55,6 @@ def __call__(self, request: httpx.Request) -> httpx.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) @@ -117,105 +62,6 @@ def test_repr_hides_the_secret(self) -> None: 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: CallbackFailure) -> 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"}), - CallbackFailure.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 - ), - CallbackFailure.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), CallbackFailure.HTTP_4XX) - await self.assert_reason(echo(503), CallbackFailure.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, CallbackFailure.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")), CallbackFailure.TIMEOUT - ) - await self.assert_reason( - Recorder(httpx.ConnectError("refused")), - CallbackFailure.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, CallbackFailure.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.""" diff --git a/tests/unit/test_events_egress.py b/tests/unit/test_events_egress.py index 03b5e6d..4cf83fd 100644 --- a/tests/unit/test_events_egress.py +++ b/tests/unit/test_events_egress.py @@ -4,13 +4,14 @@ ``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: +per-host concurrency limit. The subscribe e2e flows cover callback URL shapes +(scheme, credentials, missing host, private and metadata addresses). 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 @@ -136,29 +137,6 @@ def test_allow_loopback_admits_only_loopback(self) -> None: ) -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.""" diff --git a/tests/unit/test_events_envelope.py b/tests/unit/test_events_envelope.py index 84371d6..fdd424a 100644 --- a/tests/unit/test_events_envelope.py +++ b/tests/unit/test_events_envelope.py @@ -2,7 +2,10 @@ Sealing, opening, tampering (401), server, project and webhook binding, key rotation across server restarts and expiry are covered end to end through the -ingress in ``tests/e2e/test_events_ingress.py`` (PR 4 of #127). What stays here: +ingress in ``tests/e2e/test_events_ingress.py``. Subscription ids +(deterministic, bound to the principal, independent of argument order, valid +Appwrite custom ids) and the derived webhook secret are covered through +``events/subscribe`` in ``tests/e2e/test_events_subscribe.py``. What stays here: * The published ``mcp-subscription+jwe`` conformance vector from mcp-event-subscriptions, opened through this adapter. Every envelope in the @@ -11,9 +14,6 @@ * The Appwrite signature against a vector computed by PHP, the way Appwrite's webhooks worker does it. The e2e harness signs deliveries with its own implementation of the same formula, so only this vector ties both to PHP. -* Subscription id determinism and its Appwrite custom-id charset. Nothing - derives an id from a subscribe request until ``events/subscribe`` (PR 5), - whose e2e flow should replace these tests. * The envelope size budget that appwrite/appwrite#14293 must accommodate: worst-case inputs that no realistic flow would send. """ @@ -27,7 +27,6 @@ from mcp_server_appwrite.events.envelope import ( ENVELOPE_BUDGET, - SUBSCRIPTION_ID_MAX, Context, EnvelopeError, EnvelopeFailure, @@ -37,8 +36,6 @@ Subscription, appwrite_signature, milliseconds, - subscription_id, - valid_subscription_id, verify_appwrite_signature, ) @@ -115,89 +112,6 @@ def test_opens_the_published_vector(self): self.assertNotIn(record["delivery"]["secret"], repr(ours)) -class SubscriptionIdTests(unittest.TestCase): - def test_deterministic(self): - first = subscription_id(PRINCIPAL, "https://a.test/x", "users.user.created", {}) - second = subscription_id( - PRINCIPAL, "https://a.test/x", "users.user.created", {} - ) - self.assertEqual(first, second) - - def test_argument_order_does_not_matter(self): - forward = subscription_id( - PRINCIPAL, - "https://a.test/x", - "tablesdb.row.created", - {"project_id": "p", "database_id": "d", "table_id": "t"}, - ) - backward = subscription_id( - PRINCIPAL, - "https://a.test/x", - "tablesdb.row.created", - {"table_id": "t", "database_id": "d", "project_id": "p"}, - ) - self.assertEqual(forward, backward) - - def test_every_input_changes_the_id(self): - base = ("principal", "https://a.test/x", "users.user.created", {"a": "1"}) - variants = ( - ("other", *base[1:]), - (base[0], "https://a.test/y", *base[2:]), - (*base[:2], "storage.file.created", base[3]), - (*base[:3], {"a": "2"}), - ) - original = subscription_id(*base) - for variant in variants: - with self.subTest(variant=variant): - self.assertNotEqual(subscription_id(*variant), original) - - def test_is_a_valid_appwrite_custom_id(self): - # Mirrors utopia-php/database Key + Appwrite CustomId: at most 36 chars - # of [A-Za-z0-9._-], not starting with "_", "." or "-". - for index in range(200): - value = subscription_id(f"principal-{index}", "https://a.test", "e", {}) - with self.subTest(value=value): - self.assertLessEqual(len(value), SUBSCRIPTION_ID_MAX) - self.assertRegex(value, r"^[A-Za-z0-9][A-Za-z0-9._-]*$") - self.assertTrue(valid_subscription_id(value)) - - def test_rejects_foreign_ids(self): - for value in ( - "unique()", - "_sub", - "sub_short", - "x" * 36, - "sub_" + "g" * 31 + "!", - # Right length and charset for Appwrite, but not a SHA-256 prefix. - "sub_" + "z" * 32, - "sub_" + "A" * 32, - "sub_" + "0" * 31 + ".", - "sub_" + "0" * 33, - ): - with self.subTest(value=value): - self.assertFalse(valid_subscription_id(value)) - - def test_principal_digest_is_stable_and_short(self): - again = Principal( - issuer="https://cloud.appwrite.io/v1/oauth2/console", - subject="66b1c2d3e4f5a6b7c8d9", - client="chatgpt-connector", - ).digest - other = Principal( - issuer="https://cloud.appwrite.io/v1/oauth2/console", - subject="66b1c2d3e4f5a6b7c8d9", - client="another-client", - ).digest - self.assertEqual(PRINCIPAL, again) - self.assertNotEqual(PRINCIPAL, other) - self.assertEqual(len(PRINCIPAL), 22) - - def test_signing_key_fits_the_appwrite_webhook_secret(self): - # Appwrite's webhook `secret` param is Text(256, 8). - signing = ring(("k1", key(1))).signing_key("sub_" + "0" * 32) - self.assertRegex(signing, r"^[0-9a-f]{64}$") - - class AppwriteSignatureTests(unittest.TestCase): # Produced with PHP 8.5, the way Appwrite's webhooks worker signs # (src/Appwrite/Platform/Workers/Webhooks.php):