From 42ea5522de8521fcff96c48fabf289aa713584d0 Mon Sep 17 00:00:00 2001 From: egerev Date: Tue, 27 Oct 2026 15:08:10 +0800 Subject: [PATCH] feat(delivery): dispatcher in production on an isolated egress path, lanes, heartbeats and the delivery gate, schema 50 (ANYM-534) - deploy/compose.delivery.yaml: dispatcher with its own DSN, read-only, an internal network shared only with PostgreSQL and its own egress bridge dk-delivery0. - scripts/delivery_egress_check.py: host firewall rules and the in-container check. - Separate operator and self-service lanes, one in-flight attempt per key, least recently served key first, bounded queue per key, separate DNS slots. - Migration 050 component_heartbeats (RLS per role); API, matcher and dispatcher report version and schema; `set-delivery-features --enable|--disable`. - production.md runbook for 050, the overlays, firewall, check and enable. Co-Authored-By: Claude Opus 5.5 Signed-off-by: egerev --- .claude/rules/migrations.md | 4 +- ARCHITECTURE.md | 2 +- CLAUDE.md | 2 +- README.md | 2 +- RELEASING.md | 2 +- SECURITY.md | 4 +- SUBSCRIPTIONS.md | 29 +- deploy/compose.delivery.yaml | 76 ++++ docs/operations/production.md | 97 +++++ scripts/delivery_egress_check.py | 170 +++++++++ .../messaging/messaging_service/__main__.py | 10 +- services/messaging/messaging_service/app.py | 14 +- .../messaging/messaging_service/backup.py | 5 +- .../messaging_service/backup_r2_verifier.py | 2 +- .../backup_restore_worker.py | 2 +- .../messaging_service/delivery_endpoints.py | 28 +- .../messaging_service/delivery_policy.py | 21 +- .../messaging_service/delivery_role.sql | 7 + .../messaging_service/delivery_worker.py | 81 ++++- .../messaging_service/event_delivery.py | 119 +++++- .../messaging_service/gateway_role.sql | 7 + .../messaging/messaging_service/heartbeats.py | 48 +++ .../migrations/050_component_heartbeats.sql | 30 ++ services/messaging/messaging_service/store.py | 11 +- .../messaging_service/subscription_role.sql | 7 + .../messaging_service/subscription_worker.py | 7 + tests/test_archive.py | 2 +- tests/test_backup_r2_verifier.py | 2 +- tests/test_backup_r2_verifier_native.py | 2 +- tests/test_backup_restore_worker.py | 6 +- tests/test_capability_availability.py | 19 + tests/test_chat_role_schema.py | 8 +- tests/test_collectors.py | 2 +- tests/test_compose_runtime.py | 20 +- tests/test_contact_cards.py | 6 +- tests/test_coverage_backup.py | 4 +- tests/test_coverage_deployment.py | 2 +- tests/test_coverage_subscription_deploy.py | 62 ++++ tests/test_db_role_isolation.py | 2 +- tests/test_delivery_egress_check.py | 113 ++++++ tests/test_delivery_endpoints.py | 84 +++++ tests/test_event_delivery.py | 340 ++++++++++++++++++ tests/test_extend_key.py | 14 +- tests/test_feedback_status.py | 4 +- tests/test_key_operation.py | 4 +- tests/test_lifecycle_backup.py | 6 +- tests/test_message_purge.py | 4 +- tests/test_production.py | 23 ++ tests/test_recovery.py | 4 +- tests/test_reply_queue.py | 2 +- tests/test_reply_queue_controls.py | 4 +- tests/test_reviews.py | 2 +- tests/test_subscription_backup_roles.py | 6 +- tests/test_subscription_retention.py | 2 +- tests/test_whatsapp_native_migration.py | 8 +- 55 files changed, 1430 insertions(+), 114 deletions(-) create mode 100644 deploy/compose.delivery.yaml create mode 100644 scripts/delivery_egress_check.py create mode 100644 services/messaging/messaging_service/heartbeats.py create mode 100644 services/messaging/messaging_service/migrations/050_component_heartbeats.sql create mode 100644 tests/test_delivery_egress_check.py diff --git a/.claude/rules/migrations.md b/.claude/rules/migrations.md index 7207a035..b118ef9d 100644 --- a/.claude/rules/migrations.md +++ b/.claude/rules/migrations.md @@ -5,8 +5,8 @@ paths: # Messaging SQL migrations - Applied migrations are immutable: `Store.migrate()` compares each file's sha256 with `archive_migrations` and raises `migration_checksum_mismatch`. Never edit, rename, reorder or reformat an existing file. -- Schema changes get a new next-numbered file (current head: `049_delivery_endpoints.sql`, schema 49). -- In the same change: register it in the `migrate()` list, raise the `version>49` runtime ceiling, update backup `TABLES`/`V*_TABLES` and the restore range in `backup.py`, and update role grants (`*_role.sql`). +- Schema changes get a new next-numbered file (current head: `050_component_heartbeats.sql`, schema 50). +- In the same change: register it in the `migrate()` list, raise the `version>50` runtime ceiling, update backup `TABLES`/`V*_TABLES` and the restore range in `backup.py`, and update role grants (`*_role.sql`). - Check roles/ACL and a repeated `migrate()` (CONTRIBUTING.md). Update the documented schema version (README, RELEASING). - Keep source revisions, original provider IDs, archive IDs and acknowledgement semantics stable. Older runtimes deliberately reject a newer database. - Expand, then contract: a migration works with both the deployed and the new code; drop, rename or narrow only in a later release, once no deployed code uses the old shape. Merging an expand migration is AUTO+review; destructive SQL is OWNER (`scripts/change_path.py`). The agent of a task (a ready card) applies its migration after the backup gate of docs/operations/production.md (AGENTS.md section 2). Reserve the next number in the PR title while you hold the `schema` lane. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index e0a66f11..eeba7e30 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -27,7 +27,7 @@ principal sources. | AnyMsg WhatsApp Connector | Primary whatsmeow adapter `whatsapp/whatsmeow/` for new connections; whatsapp-web.js fallback and rollback adapter `whatsapp/*.mjs`; `whatsapp_collector.py`, pairing/identity modules, `whatsapp_capabilities.py`, WhatsApp deployment examples | Synthetic provider/IPC tests for both adapters, command lanes, history bounds, identity evidence, account provider identity, passive receipts and media handling | | AnyMsg Media Storage & Index | `blobs.py`, `storage_admission.py`, `media_policy.py`, `media_requests.py`, `attachment_metadata.py`, `uploads.py`, `file_extract.py`, `file_index.py`, `file_lifecycle.py`, indexer and file-lifecycle SQL roles, `deploy/compose.file-lifecycle.yaml` | Original integrity, scoped retrieval, bounded original requests and staged uploads, extraction safety, passages, provenance, derivative invalidation and reuse, operator-only original residency | | AnyMsg AI Enrichment | `enrichment.py`, `enrichment_providers.py`, `codex_enrichment.py`, `file_understanding.py`, `resource_admission.py`, enrichment SQL role, file-understanding deploy overlay and role, enrichment requirements and root `enrichment.*.example.json` | Budgets, isolated processing, profile validation, result fences, independent file descriptions and untrusted model output | -| AnyMsg Subscriptions & Events | `subscriptions.py`, `subscription_worker.py`, `subscription_consumption.py`, `processing_events.py`, `event_delivery.py`, `delivery_worker.py`, subscription and delivery SQL roles, `packages/wake-relay/` (host receiver) | Key ownership, committed frontier, lifecycle/version checks, fair matching, atomic checkpointing, receipt-bound ACK, signed hints that never ACK, artifact references and the host relay that turns a hint into one assistant run | +| AnyMsg Subscriptions & Events | `subscriptions.py`, `subscription_worker.py`, `subscription_consumption.py`, `processing_events.py`, `event_delivery.py`, `delivery_worker.py`, `heartbeats.py`, subscription and delivery SQL roles, `deploy/compose.subscriptions.yaml`, `deploy/compose.delivery.yaml`, `scripts/delivery_egress_check.py`, `packages/wake-relay/` (host receiver) | Key ownership, committed frontier, lifecycle/version checks, fair matching, atomic checkpointing, receipt-bound ACK, signed hints that never ACK, artifact references and the host relay that turns a hint into one assistant run | | AnyMsg Backup & Recovery | `backup.py`, crypto, runner, worker, receiver, R2 receiver/verifier, restore worker and recovery modules; R2 examples, verifier systemd units and R2 requirements | Consistent table/blob manifests, encrypted transfer, restore equality, quarantine and retention anchors | | AnyMsg Deployment & Verification | SQL migrations, `schema.sql`, `__init__.py` version, `release.py`, `telemetry.py`, root `Dockerfile`, Compose overlays, messaging packaging (`services/messaging/` Dockerfile, `pyproject.toml`, `hatch_build.py`, lockfiles), root `uv.lock`, release and branch-protection examples in `deploy/`, `services/release-router/`, `.github/`, `scripts/`, `tools/dev/` | Schema checksums, least privilege, images/packages, release routing, synthetic network drills and authentic CI evidence | | AnyMsg Python Client & Operator CLI | `services/messaging/anymsg/`, `__main__.py` operator entry point, root `client_check.py` and `media_benchmark.py`, packaged assets | Safe HTTP client, token-file CLI, source assets and explicit owner administration | diff --git a/CLAUDE.md b/CLAUDE.md index d2d48105..da31952b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -9,7 +9,7 @@ compatibility meaning too. Every service keeps its own lockfile, environment, process and state. ## Layout -- `services/messaging`: AnyMsg server (Python 3.13, PostgreSQL 18, schema 49) +- `services/messaging`: AnyMsg server (Python 3.13, PostgreSQL 18, schema 50) - `services/gateway`: common MCP gateway (HTTP only, imports no provider code) - `services/calendar`: family calendar (Google API/CalDAV), has its own AGENTS.md - `services/calls`: voice calls; `services/calls/frontend`: Vite UI diff --git a/README.md b/README.md index 3c2c6e1a..5223ea63 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ use SHA-256 deduplication; references to every source message stay separate. [Conversation view and compatibility](CONVERSATION.md) explain message kinds, ordering uncertainty and the last known dialogue turn. -Development version: **0.9.19 — experimental**, PostgreSQL schema 49. +Development version: **0.9.19 — experimental**, PostgreSQL schema 50. The expanded live-operation catalog covers messages, registered attachments, reactions, groups, contacts and profile fields. Access requires explicit per-key diff --git a/RELEASING.md b/RELEASING.md index 5632a392..77b66fad 100644 --- a/RELEASING.md +++ b/RELEASING.md @@ -3,7 +3,7 @@ The public repository is a fresh source export, not the Personal OS git history. Never publish correspondence, operational state, backups, session data or tokens. Python and npm distribution versions must match `messaging_service.__version__`. -The current source version is 0.9.19 (experimental), with PostgreSQL schema 49. +The current source version is 0.9.19 (experimental), with PostgreSQL schema 50. The published npm/PyPI release remains 0.9.9 until a separate registry release. Sending is implemented in the source with explicit account opt-in and per-key permissions; see [SENDING.md](SENDING.md). diff --git a/SECURITY.md b/SECURITY.md index f9aa9dc4..eed60a2c 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -26,7 +26,9 @@ agent's own headers; never message text, names or file content. SSRF controls: h on port 443, every resolved address public (no private, loopback, CGNAT, multicast, NAT64, 6to4, Teredo or IPv4-mapped answers), resolution again before each attempt with the address pinned, TLS hostname verification, no redirects, no proxies, and the -host policy rechecked before each attempt. Signing secrets and header values are +host policy rechecked before each attempt. The dispatcher (ANYM-534) shares a network +only with PostgreSQL and egresses over its own bridge, which host rules limit to DNS on one configured resolver and +public port 443 (no host, tailnet, private, link-local, metadata or IPv6 address). Signing secrets and header values are write-only for the API role and never returned, logged or audited. This is privilege separation, not encryption: database owners and anyone who can decrypt a backup can read them. Details: [SUBSCRIPTIONS.md](SUBSCRIPTIONS.md#self-service-webhook-endpoints-schema-49). diff --git a/SUBSCRIPTIONS.md b/SUBSCRIPTIONS.md index 2c379105..6589a048 100644 --- a/SUBSCRIPTIONS.md +++ b/SUBSCRIPTIONS.md @@ -242,9 +242,22 @@ A receiver HTTP success is not a verified host turn; all hints and probes report `host_turn_verified=false`. Host adapters must separately prove that integration. Run `python -m messaging_service.delivery_worker` using only the dispatcher DSN -and private receiver secret mounts. `--once` is a bounded operator pass. Keep the -matcher on its existing internal-only network; public HTTPS egress belongs to a -separate dispatcher deployment. No dispatcher is started by installing the +and private receiver secret mounts; in production `deploy/compose.delivery.yaml` +runs it (docs/operations/production.md, migration 050). `--once` is one serial +bounded operator pass. Otherwise it runs separate lanes (ANYM-534): one for operator +receivers, with its own DNS lookup slots, and three for self-service endpoints. An +endpoint lane leases at most one attempt per key at a time, counting the key's +endpoint tests, and serves the least recently served key first. A key holds at +most 32 queued endpoint hints across its subscriptions and hint kinds: past it a +due hint stays held (`held_reason` `budget`) and fires on a later tick, and +`test_subscription` answers `rate_limited` (events stay fetchable). So endpoints that hang for the full timeout never delay operator +receivers, and a failing endpoint never stops a lane. The dispatcher heartbeat is +written every 30 s and only while every lane finished a pass within 60 s. The +dispatcher reaches PostgreSQL only over an internal network that holds nothing +else, and the internet over its own bridge, which host rules restrict to DNS on +the one configured resolver and public port 443; host-bound packets are dropped on +both of its bridges (`scripts/delivery_egress_check.py`). Keep the matcher on its +existing internal-only network. No dispatcher is started by installing the package. Operator `--disable-receiver UUID` uses the same authorization fence as in-flight delivery and cancels queued hints. No internal/private IP exceptions are enabled. @@ -467,7 +480,11 @@ the switch, the host policy and the limits. both), with the same version check and fence as a receiver. **Switch.** Every endpoint tool, the `delivery` field and endpoint staging answer -`delivery_disabled` until the operator enables the feature (S2b). With the switch +`delivery_disabled` until the operator enables the feature with +`python -m messaging_service set-delivery-features --enable` (ANYM-534). That refuses +until the API, the matcher and the dispatcher report this version and schema +through fresh heartbeats (`component_heartbeats`, under 120 s); `--disable` is the +rollback. With the switch off no endpoint hint is staged, and turning it off cancels queued endpoint hints and tests. The webhook transport in capabilities then shows `operational: false` with reason `delivery_disabled`. @@ -484,8 +501,8 @@ suffix matches on a DNS label boundary (`example.com` matches `a.example.com`, n `badexample.com`) and deny wins. A missing or invalid file, unknown keys or an empty allowlist without `allow_any_public: true` leave the policy unset: registration answers `delivery_policy_unset` and the webhook transport shows -reason `delivery_policy_unset`. Set `allow_any_public: true` only after S2b's -isolation test passes. The policy is read again before every attempt; a host that +reason `delivery_policy_unset`. Set `allow_any_public: true` only after the +dispatcher's egress check passes (production.md, migration 050). The policy is read again before every attempt; a host that leaves it cancels the hint (`endpoint_policy_refused`). After editing the file, run `python -m messaging_service reload-delivery-policy`: it waits for every in-flight POST and registration that read the old file, then cancels queued hints and tests of diff --git a/deploy/compose.delivery.yaml b/deploy/compose.delivery.yaml new file mode 100644 index 00000000..875243b8 --- /dev/null +++ b/deploy/compose.delivery.yaml @@ -0,0 +1,76 @@ +# Opt-in overlay (ANYM-534), after compose.subscriptions.yaml. Install delivery_role.sql as owner and +# configure a separate LOGIN inheriting messaging_event_delivery; never use the gateway or matcher DSN. +# The dispatcher reaches PostgreSQL only over delivery_db (internal bridge dk-deliverydb0, no API, matcher, +# gateway or collector) and the internet only over its own bridge dk-delivery0. The host rules from +# `scripts/delivery_egress_check.py rules` drop host-bound packets on both bridges and restrict dk-delivery0 to +# DNS on DELIVERY_RESOLVER and public 443 (docs/operations/production.md). An explicit `dns` makes Docker's +# embedded resolver query DELIVERY_RESOLVER from the container namespace, so those rules apply to DNS too. +services: + delivery: + image: anymsg-delivery:${MESSAGING_VERSION:-0.9.19} + build: + context: .. + target: runtime + command: [python, -m, messaging_service.delivery_worker] + init: true + restart: unless-stopped + depends_on: + db: {condition: service_healthy} + environment: + DATABASE_URL_FILE: /run/secrets/database_url + DELIVERY_POLICY_FILE: /run/secrets/delivery-policy.json + volumes: + - type: bind + source: ./private/delivery.dsn + target: /run/secrets/database_url + read_only: true + bind: {create_host_path: false} + - type: bind + source: ./private/delivery-policy.json + target: /run/secrets/delivery-policy.json + read_only: true + bind: {create_host_path: false} + # Operator receiver secret files (subscription_receivers.secret_file names /run/secrets/receivers/). + - type: bind + source: ./private/delivery-receivers + target: /run/secrets/receivers + read_only: true + bind: {create_host_path: false} + networks: [delivery_db, delivery_egress] + dns: ["${DELIVERY_RESOLVER:-1.1.1.1}"] + read_only: true + user: "10001:10001" + cap_drop: [ALL] + security_opt: [no-new-privileges:true] + tmpfs: ["/tmp:size=16m,noexec,nosuid"] + mem_limit: 256m + cpu_shares: 512 + pids_limit: 64 + stop_grace_period: 15s + logging: + driver: json-file + options: {max-size: 5m, max-file: "2"} + db: + networks: [delivery_db] + # Self-service registration checks the same host allowlist as the dispatcher. + api: + environment: + DELIVERY_POLICY_FILE: /run/secrets/delivery-policy.json + volumes: + - type: bind + source: ./private/delivery-policy.json + target: /run/secrets/delivery-policy.json + read_only: true + bind: {create_host_path: false} + +networks: + delivery_db: + internal: true + enable_ipv6: false + driver_opts: + com.docker.network.bridge.name: dk-deliverydb0 + delivery_egress: + driver: bridge + enable_ipv6: false + driver_opts: + com.docker.network.bridge.name: dk-delivery0 diff --git a/docs/operations/production.md b/docs/operations/production.md index e2c26780..eceefb15 100644 --- a/docs/operations/production.md +++ b/docs/operations/production.md @@ -632,6 +632,103 @@ ran before it, from `//.before.json`. It does not call `promotable()` (the previous revision is no longer current `main`), does not run or reverse migrations and does not restore data: when the release changed the schema or data, follow the module's backup and migration runbook instead. +Messaging migration 050 (`manual`, ANYM-534) adds `component_heartbeats`: one row each for the +API, the matcher and the dispatcher (package version, schema), each role limited to its own row by +row-level security. Additive; older runtimes write nothing there. With it the matcher and the +delivery dispatcher run in production. Nothing here runs during an active call (check read-only +first). +1. Fresh verified backup (the backup gate), then `migrate`; the pinned backup and restore runtimes + move to the schema-50 release. +2. Role SQL (reviewed with the release): `gateway_role.sql`, `subscription_role.sql` and + `delivery_role.sql`. Logins as in step 4 of 045: `messaging_subscriptions_runtime` (`LOGIN IN + ROLE messaging_subscriptions`) and `messaging_event_delivery_runtime` (`LOGIN IN ROLE + messaging_event_delivery`), DSNs in `private/subscriptions.dsn` and `private/delivery.dsn`, + mode 0600, owned like the other DSNs. Undo: stop the service, `DROP ROLE `, remove the file. +3. Private files, owned by uid 10001: `private/delivery-policy.json` (mode 0600; SUBSCRIPTIONS.md, + Host policy) and the directory `private/delivery-receivers/` (mode 0700, its files 0600) holding + operator receiver secrets; a receiver's `secret_file` is `/run/secrets/receivers/`. +4. Resolver and host egress rules (a host step the owner approves). Set `DELIVERY_RESOLVER=` in + the project's env file, `` being the host's current upstream resolver (`/etc/resolv.conf`, or + `/run/systemd/resolve/resolv.conf` behind systemd-resolved; not a new third party). The overlay + passes it as the dispatcher's `dns`, so Docker's embedded resolver queries it from the container + namespace, where the rules apply. `python3 scripts/delivery_egress_check.py rules --resolver ` + (the same IP) prints the commands; as root, apply them before the dispatcher first starts (they + match bridge names, which may not exist yet) and save them with the host's firewall persistence. + They drop host-bound packets from both bridges, `dk-delivery0` (egress) and `dk-deliverydb0` (the + internal database network, which still has a host gateway address), and, from `dk-delivery0`, the + tailnet (100.64.0.0/10), RFC 1918, link-local and metadata (169.254.0.0/16), loopback and every + IPv6 packet; they allow DNS to `` and TCP 443 to public addresses. Undo: `rules --undo`. +5. First enrollment (owner-authorized, once; `production.py` cannot do it: a selected service must + already run, `snapshot`, and databases are never restarted). The host is on a release with this + code, applied by `production.py`; `I=deedkit-production-runtime:` is its verified image + (`docker image inspect` it; never `--build`). `F` is the project's Compose arguments (its `files` + and env files) plus `-f compose.subscriptions.yaml -f compose.delivery.yaml`. + 1. `docker tag $I anymsg-subscriptions:enroll-` and `docker tag $I anymsg-delivery:enroll-` + (new tags, never a production tag). + 2. `MESSAGING_VERSION=enroll- docker compose $F up --no-start --no-build --pull never --no-deps + subscriptions delivery` creates both networks and containers without starting them. + 3. `docker network connect --alias db _delivery_db `: the running database + joins the dispatcher's network; its backend attachment and identity are unchanged. + 4. `docker compose $F start subscriptions delivery`. + 5. Append both overlays, in this order, to the project's `files` in `/etc/deedkit/production.json` + and select their services (backup of the file first): + ```json + "files": ["...existing files...", "/srv/deedkit/deploy/compose.subscriptions.yaml", + "/srv/deedkit/deploy/compose.delivery.yaml"], + "services": {"api": "runtime", "subscriptions": "runtime", "delivery": "runtime"} + ``` + The serving API is plan B, which reads its own `/etc/deedkit/http-production.json` + (`seamless_release.py --config`, its own `prepare` snapshots): append the same two overlays, + in the same order, to the messaging project's `files` there too (back it up first) and keep + its `services` exactly the API and the gateway. Autodeploy pauses on these configuration + changes; under the pause file apply the next green `main` SHA S2 by hand: + `production.py prepare --sha S2` and `apply --sha S2` (plan A: both enrollment containers become + release images), then `seamless_release.py prepare --sha S2` and `promote --sha S2` (plan B), + `status`; check that the serving API container has `DELIVERY_POLICY_FILE` and its mount + (`docker inspect`); remove the pause. Later deploys include them. + Undo before the S2 apply, in reverse: restore both configuration backups; `docker compose $F + stop subscriptions delivery` and `rm subscriptions delivery`; `docker network disconnect + _delivery_db `; `docker network rm _delivery_db + _delivery_egress`; `docker image rm` both enroll tags. After it, the receipts name the + overlays and S2 can never be prepared again, so the undo is a release at a newer green `main` + SHA: restore both configuration backups, stop and remove both services, the same hand apply + (plan A and plan B) at that SHA, then disconnect and remove the networks. Until such a SHA + exists, `set-delivery-features --disable` and `docker compose $F stop delivery` hold. +6. Check they run: `docker compose $F ps subscriptions delivery` shows both running with no + restarts; their logs show no repeated `subscription_match_failed`, `delivery_lane_failed` or + `delivery_heartbeat_failed`; `SELECT component,version,schema_version,clock_timestamp()-beat_at + FROM component_heartbeats` (operator DSN) has three rows younger than 120 s on the deployed + version and schema 50; an agent's `get_capabilities` shows the webhook transport + `operational: true` once a receiver or endpoint is configured. +7. Connectivity check (the receipt keeps the JSON). Blocked targets are IP literals read with + `docker inspect`/`ip addr` first; a name, an unresolvable or malformed target fails. The matcher + has no listener, so start a bounded synthetic one in its namespace: `docker compose $F exec -T + subscriptions python -I - listen --port 18080 --seconds 300 < scripts/delivery_egress_check.py &`. + `` is a public resolver other than `DELIVERY_RESOLVER` (8.8.8.8, or 9.9.9.9 if that is the + configured one). First the positive control from the `api` container: every listener answers. + ```sh + docker compose $F exec -T api python -I - check --open :18080 --open :13240 \ + --open :22 --open :22 --open-dns < scripts/delivery_egress_check.py + ``` + Then from the dispatcher; it exits 0 only when every open target connects or answers and every + blocked one times out or has no route: + ```sh + docker compose $F exec -T delivery python -I - check --open db:5432 --open example.com:443 \ + --open-dns $DELIVERY_RESOLVER --blocked :8080 --blocked :13240 \ + --blocked :18080 --blocked :22 --blocked :22 \ + --blocked :22 --blocked :22 \ + --blocked 169.254.169.254:80 --blocked :53 --blocked-dns \ + --blocked [2606:4700:4700::1111]:443 < scripts/delivery_egress_check.py + ``` + `example.com` resolves through Docker's embedded resolver (explicit `dns`); `--open-dns` and + `--blocked-dns` query a resolver directly over UDP, `:53` over TCP. Stop the listener after. +8. `python -m messaging_service set-delivery-features --enable` (operator DSN) switches + self-service delivery on. It refuses (exit 1, `component_heartbeat_absent|stale`, + `component_version_mismatch`, `component_schema_mismatch` with the component) until the API, + matcher and dispatcher all report this version on schema 50. Undo: `--disable`; queued + endpoint hints and tests are cancelled under the delivery fence and events stay fetchable. +Rollback: `--disable`, then step 5's undo and `rules --undo`. 050 stays; a schema-49 runtime +ignores the table, but its `migrate` refuses a schema-50 database. Messaging migration 045 (`manual`, ANYM-511) adds the deletion purge: `messages.purged` (defaulted, a catalog change), `chats.purged_at`, the `message_purges`, `chat_tombstones`, `ingest_receipt_links` and `purge_state` tables without foreign keys, guard triggers that keep diff --git a/scripts/delivery_egress_check.py b/scripts/delivery_egress_check.py new file mode 100644 index 00000000..3a949a7d --- /dev/null +++ b/scripts/delivery_egress_check.py @@ -0,0 +1,170 @@ +"""Dispatcher egress isolation (ANYM-534): the host rules and the connectivity check for the receipt. + +`rules --resolver IP` prints the host firewall commands for the dispatcher bridge dk-delivery0 (a host step the +owner approves; nothing here changes a firewall). `check` runs inside the dispatcher container, stdlib only: + + docker compose ... exec -T delivery python -I - check --open db:5432 --open example.com:443 \\ + --blocked 172.18.0.5:8080 ... < scripts/delivery_egress_check.py + +`check --open-dns/--blocked-dns IP` sends one UDP DNS query straight to a resolver; `listen --port N` (in another +container, such as the matcher, which has no listener of its own) is a bounded synthetic positive control. + +Blocked targets are explicit IP:PORT pairs with a known listener (production.md lists them), never names: a name +the dispatcher cannot resolve sends no packet and proves nothing. An open target passes only when TCP connects. A +blocked target passes only when nothing answers (timeout, no route): a refused connection means the address +answered, which a DROP rule never does, and an unresolved or malformed target fails. +""" +from __future__ import annotations + +import argparse +import ipaddress +import json +import socket +import sys +import time + +BRIDGE = 'dk-delivery0' +# The internal database bridge still has a host gateway address: host listeners answer on it unless dropped. +DB_BRIDGE = 'dk-deliverydb0' +CHAIN = 'DEEDKIT-DELIVERY' +# Host-local, private, tailnet (CGNAT), link-local (with metadata 169.254.169.254), multicast and reserved. +DROPPED = ('0.0.0.0/8', '10.0.0.0/8', '100.64.0.0/10', '127.0.0.0/8', '169.254.0.0/16', '172.16.0.0/12', + '192.0.0.0/24', '192.168.0.0/16', '198.18.0.0/15', '224.0.0.0/4', '240.0.0.0/4') + + +def rules(resolver): + allow = [f'iptables -A {CHAIN} -d {resolver} -p {proto} --dport 53 -j RETURN' for proto in ('udp', 'tcp')] + return ([f'iptables -N {CHAIN}', f'iptables -F {CHAIN}', *allow] + + [f'iptables -A {CHAIN} -d {network} -j DROP' for network in DROPPED] + + [f'iptables -A {CHAIN} -p tcp --dport 443 -j RETURN', f'iptables -A {CHAIN} -j DROP', + f'iptables -I DOCKER-USER -i {BRIDGE} -j {CHAIN}', + # Packets to the host itself are not forwarded, so DOCKER-USER never sees them; both bridges. + *[f'iptables -I INPUT -i {bridge} -j DROP' for bridge in (BRIDGE, DB_BRIDGE)], + # The bridges have no IPv6 (enable_ipv6: false); these drop ULA, NAT64 and everything else anyway. + f'ip6tables -I DOCKER-USER -i {BRIDGE} -j DROP', + *[f'ip6tables -I INPUT -i {bridge} -j DROP' for bridge in (BRIDGE, DB_BRIDGE)]]) + + +def undo(): + return [f'iptables -D DOCKER-USER -i {BRIDGE} -j {CHAIN}', + *[f'iptables -D INPUT -i {bridge} -j DROP' for bridge in (BRIDGE, DB_BRIDGE)], + f'ip6tables -D DOCKER-USER -i {BRIDGE} -j DROP', + *[f'ip6tables -D INPUT -i {bridge} -j DROP' for bridge in (BRIDGE, DB_BRIDGE)], + f'iptables -F {CHAIN}', f'iptables -X {CHAIN}'] + + +def probe(target, timeout, literal=False): + host, _, port = target.rpartition(':') + try: + if literal: + ipaddress.ip_address(host.strip('[]')) + address = socket.getaddrinfo(host.strip('[]'), int(port), type=socket.SOCK_STREAM)[0][4] + except ValueError: + return 'malformed' + except (OSError, UnicodeError): + return 'unresolved' + try: + with socket.create_connection(address[:2], timeout=timeout): + return 'connected' + except ConnectionRefusedError: + return 'refused' + except TimeoutError: + return 'timeout' + except OSError: + return 'unreachable' + + +def probe_dns(target, timeout): + """One UDP DNS query (example.com A) straight to IP[:PORT], default 53: answered, or why not.""" + host, port = target, '53' + if target.startswith('['): + host, _, rest = target[1:].partition(']') + port = rest.removeprefix(':') or '53' + elif target.count(':') == 1: + host, port = target.split(':') + try: + address = ipaddress.ip_address(host) + port = int(port) + except ValueError: + return 'malformed' + query = bytes.fromhex('d53401000001000000000000076578616d706c6503636f6d0000010001') + with socket.socket(socket.AF_INET6 if address.version == 6 else socket.AF_INET, socket.SOCK_DGRAM) as udp: + udp.settimeout(timeout) + try: + udp.connect((str(address), port)) + udp.send(query) + return 'answered' if udp.recv(512)[:2] == query[:2] else 'malformed' + except ConnectionRefusedError: + return 'refused' + except TimeoutError: + return 'timeout' + except OSError: + return 'unreachable' + + +def check(targets, timeout): + results = [] + for target, expect in targets: + if expect.endswith('dns'): + outcome = probe_dns(target, timeout) + passed = outcome == 'answered' if expect == 'open-dns' else outcome in ('timeout', 'unreachable') + results.append({'target': target, 'expect': expect, 'outcome': outcome, 'passed': passed}) + continue + outcome = probe(target, timeout, literal=expect == 'blocked') + passed = outcome == 'connected' if expect == 'open' else outcome in ('timeout', 'unreachable') + results.append({'target': target, 'expect': expect, 'outcome': outcome, 'passed': passed}) + return {'passed': bool(results) and all(item['passed'] for item in results), 'results': results} + + +def serve(port, seconds): + """Accept and close TCP connections on every address for a bounded time; prints the bound port first.""" + deadline = time.monotonic() + seconds + with socket.create_server(('', port), backlog=16) as server: + print(json.dumps({'listening': server.getsockname()[1]}), flush=True) + while (left := deadline - time.monotonic()) > 0: + server.settimeout(left) + try: + server.accept()[0].close() + except TimeoutError: + break + return 0 + + +class Target(argparse.Action): + def __call__(self, parser, namespace, value, option_string=None): + namespace.targets.append((value, self.const)) + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[0]) + commands = parser.add_subparsers(dest='command', required=True) + rule = commands.add_parser('rules', help='print the host firewall commands') + rule.add_argument('--resolver', type=ipaddress.IPv4Address, help='the DNS resolver the containers use') + rule.add_argument('--undo', action='store_true', help='print the commands that remove the rules') + probe_parser = commands.add_parser('check', help='probe targets from inside the dispatcher container') + probe_parser.set_defaults(targets=[]) + probe_parser.add_argument('--open', action=Target, const='open', metavar='HOST:PORT') + probe_parser.add_argument('--blocked', action=Target, const='blocked', metavar='IP:PORT') + probe_parser.add_argument('--open-dns', action=Target, const='open-dns', metavar='IP[:PORT]', + help='a UDP DNS query to this resolver must be answered') + probe_parser.add_argument('--blocked-dns', action=Target, const='blocked-dns', metavar='IP[:PORT]', + help='a UDP DNS query to this resolver must get no answer') + probe_parser.add_argument('--timeout', type=float, default=3) + listen = commands.add_parser('listen', help='a synthetic TCP listener for a positive control (another container)') + listen.add_argument('--port', type=int, required=True) + listen.add_argument('--seconds', type=float, default=120) + args = parser.parse_args(argv) + if args.command == 'listen': + return serve(args.port, args.seconds) + if args.command == 'rules': + if not args.undo and args.resolver is None: + parser.error('rules needs --resolver') + print('\n'.join(undo() if args.undo else rules(args.resolver))) + return 0 + report = check(args.targets, args.timeout) + print(json.dumps(report)) + return 0 if report['passed'] else 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/services/messaging/messaging_service/__main__.py b/services/messaging/messaging_service/__main__.py index 1e135167..46a6cafc 100644 --- a/services/messaging/messaging_service/__main__.py +++ b/services/messaging/messaging_service/__main__.py @@ -38,7 +38,7 @@ def main(): 'set-messenger-operations','extend-key','set-feedback-status','export-feedback','release-retention-hold', 'set-subscription-retention','link-person','unlink-person','delete-person','list-people','key-info', 'principal-info','delivery-endpoints-list','disable-delivery-endpoint','enable-delivery-endpoint', - 'reload-delivery-policy']) + 'reload-delivery-policy','set-delivery-features']) parser.add_argument('--principal') parser.add_argument('--accounts',nargs='+') parser.add_argument('--key-file',type=Path) @@ -267,6 +267,14 @@ def main(): elif args.command=='delivery-endpoints-list': # ANYM-519: metadata and fingerprints only; never a secret or header value. print(json.dumps(store.list_delivery_endpoints_admin(args.principal))) + elif args.command=='set-delivery-features': + # ANYM-534: --enable refuses unless the API, matcher and dispatcher heartbeats agree on this version; + # --disable is the rollback: it cancels queued endpoint hints and tests under the fence. + if not (args.enable or args.disable): + parser.error('set-delivery-features requires --enable or --disable') + result=store.set_delivery_features(args.enable,verify_components=True) + print(json.dumps(result)) + if 'error' in result:raise SystemExit(1) elif args.command=='reload-delivery-policy': # ANYM-519: a host policy change counts as made when this returns (exclusive delivery fence). print(json.dumps(store.reload_delivery_policy())) diff --git a/services/messaging/messaging_service/app.py b/services/messaging/messaging_service/app.py index 493bcce4..4cf5b505 100644 --- a/services/messaging/messaging_service/app.py +++ b/services/messaging/messaging_service/app.py @@ -1300,11 +1300,21 @@ async def sweep_uploads(): await run_in_threadpool(store.cleanup_file_uploads) except Exception:log.error('upload_cleanup_unavailable') await asyncio.sleep(60) + async def heartbeat(): + # ANYM-534: the API's version heartbeat gates set-delivery-features; record() never raises on SQL errors. + from .heartbeats import HEARTBEAT_S,record + while True: + try: + if release is None or release.phase == 'active': + await run_in_threadpool(record,store,'api') + except Exception:log.error('heartbeat_unavailable') + await asyncio.sleep(HEARTBEAT_S) async with sdk_lifespan(application) as state: - task=asyncio.create_task(sweep_uploads()) if hasattr(store,'cleanup_file_uploads') else None + tasks=[asyncio.create_task(job()) for job,ready in ((sweep_uploads,hasattr(store,'cleanup_file_uploads')), + (heartbeat,hasattr(store,'pool'))) if ready] try:yield state finally: - if task: + for task in tasks: task.cancel() with suppress(asyncio.CancelledError):await task app.router.lifespan_context=lifespan diff --git a/services/messaging/messaging_service/backup.py b/services/messaging/messaging_service/backup.py index fa968aeb..a6bf0d8a 100644 --- a/services/messaging/messaging_service/backup.py +++ b/services/messaging/messaging_service/backup.py @@ -61,8 +61,9 @@ V45_TABLES={'message_purges':'message','chat_tombstones':'id','ingest_receipt_links':'digest','purge_state':'name'} V48_TABLES={'subscription_dispatcher_heartbeat':'id'} V49_TABLES={'delivery_endpoints':'id','delivery_endpoint_tests':'id'} +V50_TABLES={'component_heartbeats':'component'} # Every schema this runtime can capture, verify and restore (one ceiling for capture and verify_bundle). -SCHEMAS=range(2,50) +SCHEMAS=range(2,51) V26_TABLES={'subscriptions':'id','subscription_sources':'subscription,account,chat', 'subscription_matches':'subscription,event'} V17_TABLES={'message_conflicts':'receipt_digest'} @@ -86,7 +87,7 @@ def database_manifest(c): **(V35_TABLES if version>=35 else {}), **(V37_TABLES if version>=37 else {}), **(V38_TABLES if version>=38 else {}),**(V41_TABLES if version>=41 else {}),**(V42_TABLES if version>=42 else {}),**(V43_TABLES if version>=43 else {}), - **(V44_TABLES if version>=44 else {}),**(V45_TABLES if version>=45 else {}),**(V48_TABLES if version>=48 else {}),**(V49_TABLES if version>=49 else {})}.items(): + **(V44_TABLES if version>=44 else {}),**(V45_TABLES if version>=45 else {}),**(V48_TABLES if version>=48 else {}),**(V49_TABLES if version>=49 else {}),**(V50_TABLES if version>=50 else {})}.items(): query=sql.SQL('SELECT row_to_json(t) AS record FROM {} t ORDER BY {}').format( sql.Identifier(name),sql.SQL(',').join(sql.Identifier(part) for part in order.split(','))) sha=hashlib.sha256(); count=0 diff --git a/services/messaging/messaging_service/backup_r2_verifier.py b/services/messaging/messaging_service/backup_r2_verifier.py index f1210423..4781f88f 100644 --- a/services/messaging/messaging_service/backup_r2_verifier.py +++ b/services/messaging/messaging_service/backup_r2_verifier.py @@ -290,7 +290,7 @@ def preflight_disk(config,receipt,docker_root): def make_proof(receipt,facts,run): if (not isinstance(facts,dict) or set(facts)!={'v','status','schema_version','manifest_sha256','checks'} or type(facts['v']) is not int or facts['v']!=1 or facts['status']!='verified' or - type(facts['schema_version']) is not int or facts['schema_version'] not in range(2,50) or + type(facts['schema_version']) is not int or facts['schema_version'] not in range(2,51) or facts['manifest_sha256']!=receipt['wire']['snapshot_sha256'] or not isinstance(facts['checks'],dict) or set(facts['checks'])!=r2.CHECKS or any(v is not True for v in facts['checks'].values())): raise ValueError('invalid_helper_result') diff --git a/services/messaging/messaging_service/backup_restore_worker.py b/services/messaging/messaging_service/backup_restore_worker.py index e02feeae..3659f7cf 100644 --- a/services/messaging/messaging_service/backup_restore_worker.py +++ b/services/messaging/messaging_service/backup_restore_worker.py @@ -34,7 +34,7 @@ def restored_schema_version(store,manifest_version): row=connection.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone() actual=row.get('value') if row else None if (not isinstance(actual,str) or not actual.isascii() or not actual.isdigit() or len(actual)>3 or - int(actual) not in range(2,50) or type(manifest_version) is not int or int(actual)!=manifest_version): + int(actual) not in range(2,51) or type(manifest_version) is not int or int(actual)!=manifest_version): raise ValueError('restored_schema_mismatch') return int(actual) diff --git a/services/messaging/messaging_service/delivery_endpoints.py b/services/messaging/messaging_service/delivery_endpoints.py index 6699a526..248de0b8 100644 --- a/services/messaging/messaging_service/delivery_endpoints.py +++ b/services/messaging/messaging_service/delivery_endpoints.py @@ -547,10 +547,17 @@ async def finish_inline_test(store, token, result, *, wait_s=10.0): return result -def set_delivery_features(store, enabled): - """Operator only. Disabling cancels every queued endpoint hint and test under each endpoint's lock.""" +def set_delivery_features(store, enabled, *, verify_components=False): + """Operator only. Disabling cancels every queued endpoint hint and test under each endpoint's lock. With + verify_components (the CLI), enabling refuses unless the API, matcher and dispatcher heartbeats are fresh and + report this version on the current schema (ANYM-534); the refusal names the first component and changes nothing.""" with store.pool.connection() as c: c.execute('SELECT pg_advisory_xact_lock(hashtextextended(%s,0))', (FEATURES_LOCK,)) + if enabled and verify_components: + from .heartbeats import mismatch + problem = mismatch(c) + if problem is not None: + return problem c.execute('''INSERT INTO probe_metadata(name,value) VALUES('delivery_features',%s) ON CONFLICT(name) DO UPDATE SET value=EXCLUDED.value''', ('enabled' if enabled else 'disabled',)) cancelled = 0 @@ -701,17 +708,22 @@ def _compact(value): return json.dumps(value, sort_keys=True, separators=(',', ':')).encode() -def dispatch_tests(store, *, limit=8, transport, resolver): - """One attempt per staged test; the result is the endpoint's health. Same fence as hints.""" +def dispatch_tests(store, *, limit=8, transport, resolver, served=None): + """One attempt per staged test; the result is the endpoint's health. Same fence as hints. + + ANYM-534: picked by the endpoint lanes' scheduler (next_endpoint_item; served: the lanes' shared + {(principal, key_id): turn}), so a key's test never runs beside its hint or ahead of a waiting key's turn. + """ + from .event_delivery import next_endpoint_item from .subscriptions import _worker_identity stats = dict(delivered=0, failed=0, cancelled=0) + strict, served = served is not None, {} if served is None else served for _ in range(limit): lease = uuid4() with store.pool.connection() as c: - test = c.execute('''SELECT t.*,e.principal,e.key_id FROM delivery_endpoint_tests t - JOIN delivery_endpoints e ON e.id=t.endpoint - WHERE t.state='pending' OR t.state='leased' AND t.lease_until<=clock_timestamp() - ORDER BY t.created_at,t.id LIMIT 1 FOR UPDATE OF t SKIP LOCKED''').fetchone() + # Without lanes (served None, the serial pass) tests skip keys whose next item is a hint. + test = next_endpoint_item(c, served, 'test', '''SELECT t.*,e.principal,e.key_id FROM delivery_endpoint_tests t + JOIN delivery_endpoints e ON e.id=t.endpoint WHERE t.id=%s FOR UPDATE OF t SKIP LOCKED''', strict) if test is None: break if c.execute("SELECT %s now: return due, reason slot = budget_slot(c, row, now) - return (slot, 'budget') if slot else (None, None) + if slot: + return slot, 'budget' + # Under budget_slot's per-key lock, so the count and the insert are one step for every writer. + from .event_delivery import endpoint_queue_full + return (now + QUEUE_RETRY, 'budget') if endpoint and endpoint_queue_full(c, row) else (None, None) def _bypass(policy, reasons): @@ -211,12 +220,13 @@ def _emit(c, row, binding, window, kind, through, bypass): def _settle(c, row, binding, window, now, through): policy = binding['policy'] or {} + endpoint = binding['receiver'] is None emitted = 0 if window['first_match_at'] is not None: due, reason = _due(policy, window) # The current policy decides the bypass from the range's reasons, also after a reconfiguration. bypass = _bypass(policy, window['reasons']) - window['due_at'], window['held_reason'] = _gate(c, row, policy, due, reason, bypass, now) + window['due_at'], window['held_reason'] = _gate(c, row, policy, due, reason, bypass, now, endpoint) if window['due_at'] is None: emitted += _emit(c, row, binding, window, 'events', window['through_position'], bypass) window['hinted_through'] = max(window['hinted_through'] or 0, window['through_position']) @@ -227,7 +237,8 @@ def _settle(c, row, binding, window, now, through): window['silence_state'], window['silence_fired_at'] = 'fired', now if window['silence_state'] == 'fired': bypass = 'silence' in policy.get('quiet_bypass', ()) - window['silence_due_at'], window['silence_held_reason'] = _gate(c, row, policy, now, None, bypass, now) + window['silence_due_at'], window['silence_held_reason'] = _gate(c, row, policy, now, None, bypass, now, + endpoint) if window['silence_due_at'] is None: emitted += _emit(c, row, binding, window, 'silence', through, bypass) window['silence_state'] = 'hinted' diff --git a/services/messaging/messaging_service/delivery_role.sql b/services/messaging/messaging_service/delivery_role.sql index 2ea12b38..903f0d61 100644 --- a/services/messaging/messaging_service/delivery_role.sql +++ b/services/messaging/messaging_service/delivery_role.sql @@ -31,3 +31,10 @@ GRANT SELECT(subscription,acked_through) ON subscription_consumption TO messagin GRANT EXECUTE ON FUNCTION public.subscription_hint_references(uuid,bigint,bigint,text[],text[]) TO messaging_event_delivery; GRANT UPDATE(state,lease_token,lease_until,http_status,latency_ms,error_code,finished_at) ON delivery_endpoint_tests TO messaging_event_delivery; +-- Schema 50 (ANYM-534): only this component's heartbeat row (row-level security). +GRANT SELECT,INSERT,UPDATE ON component_heartbeats TO messaging_event_delivery; +DO $$ BEGIN + IF NOT EXISTS (SELECT FROM pg_policies WHERE tablename='component_heartbeats' AND policyname='messaging_event_delivery_beat') THEN + CREATE POLICY messaging_event_delivery_beat ON component_heartbeats TO messaging_event_delivery USING (component='dispatcher') WITH CHECK (component='dispatcher'); + END IF; +END $$; diff --git a/services/messaging/messaging_service/delivery_worker.py b/services/messaging/messaging_service/delivery_worker.py index ff3bf52e..b005bcc1 100644 --- a/services/messaging/messaging_service/delivery_worker.py +++ b/services/messaging/messaging_service/delivery_worker.py @@ -5,9 +5,81 @@ from pathlib import Path import signal import threading -from .event_delivery import dispatch_hints, register_receiver, disable_receiver +import time +from . import heartbeats +from .event_delivery import (OPERATOR_DNS_SLOTS, _addresses, _transport, beat, dispatch_hints, disable_receiver, + endpoint_tests, register_receiver) from .store import Store +HEARTBEAT_S = heartbeats.HEARTBEAT_S +# A lane that has not finished a pass for this long stops the heartbeat, so capabilities report webhook down. +LANE_STALL_S = 60 +ENDPOINT_LANES = 3 + + +def pulse(store, progress, *, now=None): + """Write both heartbeats only while every lane has finished a pass within LANE_STALL_S (ANYM-534).""" + now = time.monotonic() if now is None else now + if any(now - last > LANE_STALL_S for last in list(progress.values())): + return False + alive = beat(store) + return heartbeats.record(store, 'dispatcher') and alive + + +def _lane(store, stop, name, progress, interval, step): + while not stop.is_set(): + busy = False + try: + result = step() + busy = any(result.get(key) for key in ('delivered', 'retried', 'dead_letter', 'cancelled', 'deferred')) + if busy or result.get('endpoint_tests'): + print(json.dumps(dict(result, lane=name), default=str), flush=True) + except Exception as error: + # One endpoint's failure (or a database blip) never stops the lane; its lease expires and retries. + print(json.dumps({'error': 'delivery_lane_failed', 'lane': name, 'class': type(error).__name__}), flush=True) + finally: + progress[name] = time.monotonic() + if not busy: + stop.wait(interval) + + +def run(store, stop, *, transport=_transport, resolver=None, interval=2): + """Operator receivers and self-service endpoints in separate lanes until stop is set (ANYM-534).""" + operator_resolver = resolver or (lambda host, port: _addresses(host, port, slots=OPERATOR_DNS_SLOTS)) + resolver = resolver or _addresses + served = {} + def endpoint_step(first): + def step(): + result = dispatch_hints(store, lane='endpoint', limit=1, served=served, transport=transport, + resolver=resolver) + if first: + # One test per pass, in the lanes' shared key order: a key's tests never run a batch ahead. + tests = endpoint_tests(store, transport=transport, resolver=resolver, limit=1, served=served) + if tests is not None: + result['endpoint_tests'] = tests + return result + return step + steps = {'operator': lambda: dispatch_hints(store, lane='operator', transport=transport, resolver=operator_resolver)} + steps.update({f'endpoint-{n}': endpoint_step(n == 0) for n in range(ENDPOINT_LANES)}) + started = time.monotonic() + progress = dict.fromkeys(steps, started) + threads = [threading.Thread(target=_lane, args=(store, stop, name, progress, interval, step), daemon=True, + name='deedkit-delivery-' + name) for name, step in steps.items()] + for thread in threads: + thread.start() + try: + while not stop.is_set(): + try: + pulse(store, progress) + except Exception as error: + # Local stdout only: the class, never the message (it can hold a DSN or an endpoint). + print(json.dumps({'error': 'delivery_heartbeat_failed', 'class': type(error).__name__}), flush=True) + stop.wait(HEARTBEAT_S) + finally: + stop.set() + for thread in threads: + thread.join() + def main(): parser = argparse.ArgumentParser(description=__doc__) @@ -36,11 +108,10 @@ def main(): if args.disable_receiver: print(json.dumps(disable_receiver(store, args.disable_receiver))) return - while not stop.is_set(): + if args.once: print(json.dumps(dispatch_hints(store)), flush=True) - if args.once: - break - stop.wait(2) + return + run(store, stop) finally: store.close() diff --git a/services/messaging/messaging_service/event_delivery.py b/services/messaging/messaging_service/event_delivery.py index 64387c57..9197e748 100644 --- a/services/messaging/messaging_service/event_delivery.py +++ b/services/messaging/messaging_service/event_delivery.py @@ -21,7 +21,18 @@ _DNS_SLOTS = threading.BoundedSemaphore(4) +# ANYM-534: operator receivers resolve from their own slots, so hanging agent endpoint lookups never starve them. +OPERATOR_DNS_SLOTS = threading.BoundedSemaphore(2) _DNS_TIMEOUT_S = 2 +# ANYM-534: queued self-service hints per key; past it, a subscription that already has one queued stages no more. +ENDPOINT_QUEUE_PER_KEY = 32 +ENDPOINT_LANE_LOCK = 'deedkit-delivery-endpoint-lane' +# A live leased endpoint hint or endpoint test of the key: one attempt in flight per key across both (ANYM-534). +KEY_IN_FLIGHT = '''(EXISTS (SELECT 1 FROM subscription_outbox f JOIN subscriptions k ON k.id=f.subscription + WHERE f.endpoint IS NOT NULL AND f.state='leased' AND f.lease_until>clock_timestamp() + AND k.principal={principal} AND k.key_id={key_id}) + OR EXISTS (SELECT 1 FROM delivery_endpoint_tests f JOIN delivery_endpoints k ON k.id=f.endpoint + WHERE f.state='leased' AND f.lease_until>clock_timestamp() AND k.principal={principal} AND k.key_id={key_id}))''' # ANYM-519: is_global accepts these; they translate to or embed another (possibly internal) address. _REFUSED_V6 = tuple(ipaddress.ip_network(net) for net in ('64:ff9b::/96', '64:ff9b:1::/48', '2002::/16', '2001::/32')) @@ -33,10 +44,11 @@ def _refused(address): address.teredo or any(address in net for net in _REFUSED_V6)) -def _addresses(host, port): +def _addresses(host, port, *, slots=None): # libc DNS may outlive a socket deadline. Bound response latency and retain # one of four slots until that lookup really ends, without an unbounded queue. - if not _DNS_SLOTS.acquire(blocking=False): + slots = _DNS_SLOTS if slots is None else slots + if not slots.acquire(blocking=False): raise OSError('receiver_dns_busy') finished = threading.Event() result = {} @@ -47,12 +59,12 @@ def lookup(): except OSError: result['error'] = True finally: - _DNS_SLOTS.release() + slots.release() finished.set() try: threading.Thread(target=lookup, daemon=True, name='deedkit-receiver-dns').start() except BaseException: - _DNS_SLOTS.release() + slots.release() raise if not finished.wait(_DNS_TIMEOUT_S) or result.get('error'): raise OSError('receiver_dns_unavailable') @@ -232,11 +244,25 @@ def _hint_target(c, row): return binding if endpoint_stageable(c, binding['endpoint']) else None +def endpoint_queue_full(c, row): + """True when the key holds ENDPOINT_QUEUE_PER_KEY queued endpoint hints of any kind and subscription. + + Callers hold the key's subscription-hints lock (budget_slot), so no writer passes the count concurrently. + """ + return c.execute('''SELECT count(*)>=%s AS full + FROM subscription_outbox o JOIN subscriptions s ON s.id=o.subscription WHERE o.endpoint IS NOT NULL + AND o.state IN ('pending','leased') AND s.principal=%s AND s.key_id=%s''', + (ENDPOINT_QUEUE_PER_KEY, row['principal'], row['key_id'])).fetchone()['full'] + + def enqueue_hint(c, row, through, *, kind='events', request_id=''): binding = _hint_target(c, row) if binding is None: return None target = 'receiver' if binding['receiver'] is not None else 'endpoint' + if target == 'endpoint' and endpoint_queue_full(c, row): + from .store import APIError + raise APIError('rate_limited', 429) return c.execute('''INSERT INTO subscription_outbox(subscription,subscription_version,''' + target + ''', through_position,kind,request_id) VALUES(%s,%s,%s,%s,%s,%s) ON CONFLICT(subscription,subscription_version,kind,through_position,request_id) @@ -412,32 +438,89 @@ def delivery_transports(c, p): return rows -def dispatch_hints(store, *, limit=16, transport=_transport, resolver=_addresses): - """Leased at-least-once hints, independently reauthorized before each send.""" - from .policy import lock_principal - from .store import bounded - bounded(limit, 100) - beat(store) - stats = dict(delivered=0, retried=0, dead_letter=0, cancelled=0, deferred=0) - holds = [] - # ANYM-519: rotation promotion and endpoint tests; a failure there never stops operator hints. +def endpoint_tests(store, *, transport=_transport, resolver=_addresses, limit=8, served=None): + """ANYM-519 rotation promotion and endpoint tests; a failure there never stops hints (None).""" try: from .delivery_endpoints import dispatch_tests, promote_rotations, schema_ready with store.pool.connection() as c: ready = schema_ready(c) and promote_rotations(c) is not None if ready: - stats['endpoint_tests'] = dispatch_tests(store, transport=transport, resolver=resolver) + return dispatch_tests(store, transport=transport, resolver=resolver, limit=limit, served=served) except Exception as error: from .telemetry import capture capture(error, 'endpoint_tests_failed') print(json.dumps({'error': 'endpoint_tests_failed', 'class': type(error).__name__}), flush=True) + return None + + +_DUE = "(o.state='pending' AND o.next_attempt_at<=clock_timestamp() OR o.state='leased' AND o.lease_until<=clock_timestamp())" + + +# Each key's oldest due endpoint hint or pending endpoint test, keys with an attempt in flight left out (ANYM-534). +_ENDPOINT_QUEUE = '''SELECT DISTINCT ON (q.principal,q.key_id) q.kind,q.id,q.principal,q.key_id FROM ( + SELECT 'hint' AS kind,o.id,s.principal,s.key_id,o.next_attempt_at AS due,o.created_at + FROM subscription_outbox o JOIN subscriptions s ON s.id=o.subscription WHERE o.endpoint IS NOT NULL AND ''' + _DUE + ''' + UNION ALL SELECT 'test',t.id,e.principal,e.key_id,t.created_at,t.created_at + FROM delivery_endpoint_tests t JOIN delivery_endpoints e ON e.id=t.endpoint + WHERE t.state='pending' OR t.state='leased' AND t.lease_until<=clock_timestamp()) q + WHERE NOT ''' + KEY_IN_FLIGHT.format(principal='q.principal', key_id='q.key_id') + ''' + ORDER BY q.principal,q.key_id,q.due,q.created_at,q.id''' + + +def next_endpoint_item(c, served, kind, lock, strict=True): + """The endpoint lanes' one scheduler over hints and endpoint tests: lock the item of the least recently served key. + + One endpoint lane (or the endpoint test pass) picks at a time, so a key never has two attempts in flight. + Strict: when the next key's item is of the other kind, None, so neither queue overtakes a waiting key (the + first endpoint lane runs a hint pick and a test pick each pass). Non-strict skips to this kind's keys. + """ + c.execute('SELECT pg_advisory_xact_lock(hashtext(%s))', (ENDPOINT_LANE_LOCK,)) + for row in sorted(c.execute(_ENDPOINT_QUEUE).fetchall(), key=lambda row: served.get((row['principal'], row['key_id']), 0)): + if row['kind'] != kind: + if strict: + return None + continue + picked = c.execute(lock, (row['id'],)).fetchone() + if picked is not None: + served[(row['principal'], row['key_id'])] = max(served.values(), default=0) + 1 + return picked + return None + + +def _next_endpoint_hint(c, served): + """ANYM-534 endpoint lane: the due hint of the least recently served key with no attempt in flight.""" + return next_endpoint_item(c, served, 'hint', 'SELECT * FROM subscription_outbox WHERE id=%s FOR UPDATE SKIP LOCKED') + + +def dispatch_hints(store, *, limit=16, transport=_transport, resolver=_addresses, lane=None, served=None): + """Leased at-least-once hints, independently reauthorized before each send. + + lane None is one serial pass over every hint (with the heartbeat and endpoint tests, as before ANYM-534); + 'operator' takes only operator receiver hints and 'endpoint' only self-service ones, one key at a time, + least recently served key first (served: the lane's shared {(principal, key_id): turn}). + """ + from .policy import lock_principal + from .store import bounded + bounded(limit, 100) + if lane not in (None, 'operator', 'endpoint'): + raise ValueError('unknown_lane') + stats = dict(delivered=0, retried=0, dead_letter=0, cancelled=0, deferred=0) + if lane is None: + beat(store) + tests = endpoint_tests(store, transport=transport, resolver=resolver) + if tests is not None: + stats['endpoint_tests'] = tests + served = {} if served is None else served + holds = [] for _ in range(limit): lease = uuid4() with store.pool.connection() as c: - hint = c.execute('''SELECT * FROM subscription_outbox WHERE - (state='pending' AND next_attempt_at<=clock_timestamp() OR - state='leased' AND lease_until<=clock_timestamp()) - ORDER BY next_attempt_at,created_at,id LIMIT 1 FOR UPDATE SKIP LOCKED''').fetchone() + if lane == 'endpoint': + hint = _next_endpoint_hint(c, served) + else: + hint = c.execute('''SELECT o.* FROM subscription_outbox o WHERE ''' + _DUE + + (" AND o.receiver IS NOT NULL" if lane == 'operator' else '') + + ''' ORDER BY o.next_attempt_at,o.created_at,o.id LIMIT 1 FOR UPDATE SKIP LOCKED''').fetchone() if hint is None: break if hint['attempts'] >= 8: diff --git a/services/messaging/messaging_service/gateway_role.sql b/services/messaging/messaging_service/gateway_role.sql index f596f50b..7fc0e85c 100644 --- a/services/messaging/messaging_service/gateway_role.sql +++ b/services/messaging/messaging_service/gateway_role.sql @@ -94,3 +94,10 @@ GRANT UPDATE(header_names,headers_fingerprint,header_values,pending_secret,pendi signing_secret,state,version,updated_at) ON delivery_endpoints TO messaging_gateway; GRANT SELECT,INSERT ON delivery_endpoint_tests TO messaging_gateway; GRANT UPDATE(state,finished_at,error_code) ON delivery_endpoint_tests TO messaging_gateway; +-- Schema 50 (ANYM-534): only this component's heartbeat row (row-level security). +GRANT SELECT,INSERT,UPDATE ON component_heartbeats TO messaging_gateway; +DO $$ BEGIN + IF NOT EXISTS (SELECT FROM pg_policies WHERE tablename='component_heartbeats' AND policyname='messaging_gateway_beat') THEN + CREATE POLICY messaging_gateway_beat ON component_heartbeats TO messaging_gateway USING (component='api') WITH CHECK (component='api'); + END IF; +END $$; diff --git a/services/messaging/messaging_service/heartbeats.py b/services/messaging/messaging_service/heartbeats.py new file mode 100644 index 00000000..a28d087a --- /dev/null +++ b/services/messaging/messaging_service/heartbeats.py @@ -0,0 +1,48 @@ +"""Component heartbeats (schema 50, ANYM-534): the API, matcher and dispatcher report their version. + +`set-delivery-features --enable` turns self-service delivery on only while all three are fresh and agree, so a +component still running an older release keeps the feature off. A failed write never stops the component. +""" +from __future__ import annotations + +COMPONENTS = ('api', 'matcher', 'dispatcher') +HEARTBEAT_S = 30 +MAX_AGE_S = 120 + + +def record(store, component): + import psycopg + from . import __version__ + from .store import RUNTIME_SCHEMA + try: + with store.pool.connection() as c: + c.execute('''INSERT INTO component_heartbeats(component,version,schema_version,beat_at) + VALUES(%s,%s,%s,clock_timestamp()) ON CONFLICT(component) DO UPDATE SET version=EXCLUDED.version, + schema_version=EXCLUDED.schema_version,beat_at=EXCLUDED.beat_at''', (component, __version__, RUNTIME_SCHEMA)) + return True + except psycopg.Error: + # Includes a schema before 50 (no table) and a role without the grant. + return False + + +def mismatch(c): + """None when every component is fresh and runs this version on the current schema, else the first problem.""" + from . import __version__ + from .store import RUNTIME_SCHEMA, schema_version + current = schema_version(c) + if current != RUNTIME_SCHEMA: + return {'error': 'component_schema_mismatch', 'component': 'operator', 'schema_version': current} + rows = {row['component']: row for row in c.execute('''SELECT component,version,schema_version, + clock_timestamp()-beat_at<=make_interval(secs=>%s) AS fresh FROM component_heartbeats''', (MAX_AGE_S,))} + for name in COMPONENTS: + row = rows.get(name) + if row is None: + return {'error': 'component_heartbeat_absent', 'component': name} + if not row['fresh']: + return {'error': 'component_heartbeat_stale', 'component': name} + if row['version'] != __version__: + return {'error': 'component_version_mismatch', 'component': name, 'version': row['version'], + 'expected': __version__} + if row['schema_version'] != current: + return {'error': 'component_schema_mismatch', 'component': name, 'schema_version': row['schema_version']} + return None diff --git a/services/messaging/messaging_service/migrations/050_component_heartbeats.sql b/services/messaging/messaging_service/migrations/050_component_heartbeats.sql new file mode 100644 index 00000000..25f20b55 --- /dev/null +++ b/services/messaging/messaging_service/migrations/050_component_heartbeats.sql @@ -0,0 +1,30 @@ +-- rollout: manual +-- Component heartbeats (ANYM-534): the API, the matcher and the dispatcher each refresh one row with their +-- package version and the newest schema they know. `set-delivery-features --enable` refuses unless all three +-- are fresh and agree. Additive: older runtimes write nothing here. Manual only because of the grants below. +-- Row-level security keeps each role on its own row, so one component can never report for another. +SET LOCAL lock_timeout='2s'; +CREATE TABLE component_heartbeats ( + component text PRIMARY KEY CHECK(component IN ('api','matcher','dispatcher')), + version text NOT NULL CHECK(version ~ '^[0-9A-Za-z.+-]{1,64}$'), + schema_version integer NOT NULL CHECK(schema_version>0), + beat_at timestamptz NOT NULL +); +ALTER TABLE component_heartbeats ENABLE ROW LEVEL SECURITY; +DO $$ BEGIN + IF EXISTS (SELECT FROM pg_roles WHERE rolname='messaging_gateway') THEN + GRANT SELECT,INSERT,UPDATE ON component_heartbeats TO messaging_gateway; + CREATE POLICY messaging_gateway_beat ON component_heartbeats TO messaging_gateway + USING (component='api') WITH CHECK (component='api'); + END IF; + IF EXISTS (SELECT FROM pg_roles WHERE rolname='messaging_subscriptions') THEN + GRANT SELECT,INSERT,UPDATE ON component_heartbeats TO messaging_subscriptions; + CREATE POLICY messaging_subscriptions_beat ON component_heartbeats TO messaging_subscriptions + USING (component='matcher') WITH CHECK (component='matcher'); + END IF; + IF EXISTS (SELECT FROM pg_roles WHERE rolname='messaging_event_delivery') THEN + GRANT SELECT,INSERT,UPDATE ON component_heartbeats TO messaging_event_delivery; + CREATE POLICY messaging_event_delivery_beat ON component_heartbeats TO messaging_event_delivery + USING (component='dispatcher') WITH CHECK (component='dispatcher'); + END IF; +END $$; diff --git a/services/messaging/messaging_service/store.py b/services/messaging/messaging_service/store.py index dba75d6a..f854a041 100644 --- a/services/messaging/messaging_service/store.py +++ b/services/messaging/messaging_service/store.py @@ -107,6 +107,7 @@ def digest(value: str) -> str: KEY_OPERATION_SCHEMA=44 # api_key_operations (ANYM-495) +RUNTIME_SCHEMA=50 # the newest schema this runtime knows; migrate() refuses a newer database def revocable_token(value): @@ -1117,10 +1118,10 @@ def delivery_endpoint_test(self, token, endpoint_id, request_id): with self.authorized(token) as (c, p): return endpoint_test(c, p, endpoint_id, request_id) - def set_delivery_features(self, enabled): - # Operator only (ANYM-534 adds the CLI); disabling cancels queued endpoint hints and tests. + def set_delivery_features(self, enabled, *, verify_components=False): + # Operator only (`set-delivery-features`, ANYM-534); disabling cancels queued endpoint hints and tests. from .delivery_endpoints import set_delivery_features - return set_delivery_features(self, enabled) + return set_delivery_features(self, enabled, verify_components=verify_components) def reload_delivery_policy(self): # Operator only, after editing DELIVERY_POLICY_FILE: waits for in-flight POSTs, cancels refused hosts' hints. @@ -1248,8 +1249,8 @@ def migrate(self): lock_replay_tables(c) c.execute(Path(__file__).with_name('schema.sql').read_text()) version=int(c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']) - if version>49: raise RuntimeError('database_newer_than_runtime') - for target,name in [(2,'002_archive.sql'),(3,'003_collectors.sql'),(4,'004_whatsapp.sql'),(5,'005_access.sql'),(6,'006_reviews.sql'),(7,'007_search_indexes.sql'),(8,'008_people.sql'),(9,'009_file_index.sql'),(10,'010_search_audit.sql'),(11,'011_telegram_window.sql'),(12,'012_search.sql'),(13,'013_file_search.sql'),(14,'014_collector_health.sql'),(15,'015_enrichment.sql'),(16,'016_file_content.sql'),(17,'017_chat_evidence.sql'),(18,'018_collector_operations.sql'),(19,'019_whatsapp_identities.sql'),(20,'020_processing_admission.sql'),(21,'021_media_storage.sql'),(22,'022_outgoing.sql'),(23,'023_receipts.sql'),(24,'024_formatting.sql'),(25,'025_operations.sql'),(26,'026_subscriptions.sql'),(27,'027_reply_queue.sql'),(28,'028_send_reference.sql'),(29,'029_file_uploads.sql'),(30,'030_whatsapp_native_identities.sql'),(31,'031_media_requests.sql'),(32,'032_event_delivery.sql'),(33,'033_file_lifecycle.sql'),(34,'034_file_understanding.sql'),(35,'035_cold_archive.sql'),(36,'036_extend_key_audit.sql'),(37,'037_narrow_worker_roles.sql'),(38,'038_hint_policy.sql'),(39,'039_feedback_status.sql'),(40,'040_subscription_retention.sql'),(41,'041_reply_queue_controls.sql'),(42,'042_read_state_freshness.sql'),(43,'043_person_links.sql'),(44,'044_key_operation_id.sql'),(45,'045_message_purge.sql'),(46,'046_chat_roles.sql'),(47,'047_contact_search.sql'),(48,'048_dispatcher_heartbeat.sql'),(49,'049_delivery_endpoints.sql')]: + if version>RUNTIME_SCHEMA: raise RuntimeError('database_newer_than_runtime') + for target,name in [(2,'002_archive.sql'),(3,'003_collectors.sql'),(4,'004_whatsapp.sql'),(5,'005_access.sql'),(6,'006_reviews.sql'),(7,'007_search_indexes.sql'),(8,'008_people.sql'),(9,'009_file_index.sql'),(10,'010_search_audit.sql'),(11,'011_telegram_window.sql'),(12,'012_search.sql'),(13,'013_file_search.sql'),(14,'014_collector_health.sql'),(15,'015_enrichment.sql'),(16,'016_file_content.sql'),(17,'017_chat_evidence.sql'),(18,'018_collector_operations.sql'),(19,'019_whatsapp_identities.sql'),(20,'020_processing_admission.sql'),(21,'021_media_storage.sql'),(22,'022_outgoing.sql'),(23,'023_receipts.sql'),(24,'024_formatting.sql'),(25,'025_operations.sql'),(26,'026_subscriptions.sql'),(27,'027_reply_queue.sql'),(28,'028_send_reference.sql'),(29,'029_file_uploads.sql'),(30,'030_whatsapp_native_identities.sql'),(31,'031_media_requests.sql'),(32,'032_event_delivery.sql'),(33,'033_file_lifecycle.sql'),(34,'034_file_understanding.sql'),(35,'035_cold_archive.sql'),(36,'036_extend_key_audit.sql'),(37,'037_narrow_worker_roles.sql'),(38,'038_hint_policy.sql'),(39,'039_feedback_status.sql'),(40,'040_subscription_retention.sql'),(41,'041_reply_queue_controls.sql'),(42,'042_read_state_freshness.sql'),(43,'043_person_links.sql'),(44,'044_key_operation_id.sql'),(45,'045_message_purge.sql'),(46,'046_chat_roles.sql'),(47,'047_contact_search.sql'),(48,'048_dispatcher_heartbeat.sql'),(49,'049_delivery_endpoints.sql'),(50,'050_component_heartbeats.sql')]: migration=Path(__file__).with_name('migrations')/name sha=hashlib.sha256(migration.read_bytes()).hexdigest() if version= heartbeats.HEARTBEAT_S: + heartbeats.record(store, 'matcher') + last_beat = time.monotonic() except Exception as error: capture(error, 'subscription_match_failed') # No SQL, DSN, content, selectors or subscription IDs in logs. diff --git a/tests/test_archive.py b/tests/test_archive.py index 2e085488..8cd9e441 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -290,7 +290,7 @@ def test_migration_checksum_and_restart_preserve_revisions(archive): ar.ingest(event()); s.migrate(); s.migrate() assert len(s.get_updates(key,'main')['items'])==1 with s.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' saved=c.execute('SELECT sha256 FROM archive_migrations WHERE version=2').fetchone()['sha256'] c.execute("UPDATE archive_migrations SET sha256='wrong' WHERE version=2") try: diff --git a/tests/test_backup_r2_verifier.py b/tests/test_backup_r2_verifier.py index 22874645..7ff24d04 100644 --- a/tests/test_backup_r2_verifier.py +++ b/tests/test_backup_r2_verifier.py @@ -34,7 +34,7 @@ def test_server_verifier_proof_requires_exact_helper_facts_and_cloud_identity(): # spec: ANYM-511 -@pytest.mark.parametrize('version,accepted',[(48,True),(49,True),(50,False)],ids=['n-1','current','next']) +@pytest.mark.parametrize('version,accepted',[(49,True),(50,True),(51,False)],ids=['n-1','current','next']) def test_server_verifier_proof_accepts_current_schema_and_rejects_the_next(version,accepted): v=module();receipt={'wire':{'id':'20261016T091703Z-fedcba9876543210','bytes':100,'sha256':'a'*64,'snapshot_sha256':'b'*64}} facts={'v':1,'status':'verified','schema_version':version,'manifest_sha256':'b'*64,'checks':{k:True for k in r2.CHECKS}} diff --git a/tests/test_backup_r2_verifier_native.py b/tests/test_backup_r2_verifier_native.py index 5e5bf4e9..ccc03ba7 100644 --- a/tests/test_backup_r2_verifier_native.py +++ b/tests/test_backup_r2_verifier_native.py @@ -112,7 +112,7 @@ def get_object(self,**kwargs):return {'ContentLength':cipher.stat().st_size,'Bod 'snapshot_sha256':snapshot,'max_plaintext_bytes':receipt['wire']['bytes']})) facts=engine.restore(config,{'bucket':'synthetic'},Cloud(),receipt,restore_run,target_payload,target_secrets) proof=verifier.make_proof(receipt,facts,restore_run) - assert proof['verification']['schema_version']==49 + assert proof['verification']['schema_version']==50 assert proof['verification']['checks']=={name:True for name in r2.CHECKS} helper=engine.inspect_owned('helper',verifier.names(restore_run)['helper'],restore_run) network=engine.inspect_owned('network',verifier.names(restore_run)['network'],restore_run) diff --git a/tests/test_backup_restore_worker.py b/tests/test_backup_restore_worker.py index 54fe8d12..039afda2 100644 --- a/tests/test_backup_restore_worker.py +++ b/tests/test_backup_restore_worker.py @@ -145,8 +145,8 @@ def test_large_stream_keeps_original_ciphertext_source_distinct_from_tar_members assert hashlib.sha256((tmp_path/'large-restore/files'/digest).read_bytes()).hexdigest()==digest -# The id '36-36' is a Tiden case ID; since schema 49 the first unsupported schema is 50. -@pytest.mark.parametrize('actual,manifest',[('34',31),('31',34),pytest.param('50',50,id='36-36'),('corrupt',34),(None,34)]) +# The id '36-36' is a Tiden case ID; since schema 50 the first unsupported schema is 51. +@pytest.mark.parametrize('actual,manifest',[('34',31),('31',34),pytest.param('51',51,id='36-36'),('corrupt',34),(None,34)]) def test_restored_database_schema_must_match_manifest_before_quarantine_or_proof(actual,manifest): from messaging_service import backup_restore_worker as worker class Lease: @@ -165,7 +165,7 @@ def __exit__(self,*args):pass # spec: ANYM-511 -@pytest.mark.parametrize('version',[48,49],ids=['n-1','current']) +@pytest.mark.parametrize('version',[49,50],ids=['n-1','current']) def test_restored_database_schema_accepts_the_current_and_previous_schema(version): from messaging_service import backup_restore_worker as worker class Lease: diff --git a/tests/test_capability_availability.py b/tests/test_capability_availability.py index d13be696..b8c80bbf 100644 --- a/tests/test_capability_availability.py +++ b/tests/test_capability_availability.py @@ -242,3 +242,22 @@ def test_webhook_policy_unset_then_a_bound_self_service_endpoint(seeded,tmp_path assert 'register_delivery_endpoint' not in s.capabilities(s.issue_key('endpoint-legacy',['telegram-demo']))['available'] finally: s.set_delivery_features(False) + + +# spec: ANYM-534 +def test_a_lane_worker_pulse_keeps_webhook_operational_for_120_seconds(seeded,tmp_path): + import time + from messaging_service import delivery_worker + from test_event_delivery import identity + s,_,_=seeded + token=identity(s) + bound_receiver(s,token,tmp_path) + with s.pool.connection() as c:c.execute('DELETE FROM subscription_dispatcher_heartbeat WHERE id=1') + now=time.monotonic() + assert delivery_worker.pulse(s,{'operator':now,'endpoint-0':now},now=now) + assert transports(s,token)['webhook']['operational'] is True + heartbeat(s,delivery_worker.HEARTBEAT_S*3) + assert transports(s,token)['webhook']['operational'] is True + heartbeat(s,121) + webhook=transports(s,token)['webhook'] + assert webhook['operational'] is False and webhook['reason']=='dispatcher_heartbeat_stale' diff --git a/tests/test_chat_role_schema.py b/tests/test_chat_role_schema.py index 0c7e3e01..e2a45c73 100644 --- a/tests/test_chat_role_schema.py +++ b/tests/test_chat_role_schema.py @@ -41,7 +41,7 @@ def test_migration_046_adds_nullable_role_columns_and_repeats_over_schema45_rows assert schema(store)=='45' and role_columns(store)=={} with store.pool.connection() as c:before=c.execute('SELECT * FROM chats WHERE id=%s',(chat,)).fetchone() store.migrate();store.migrate() - assert schema(store)=='49' and role_columns(store)==COLUMNS + assert schema(store)=='50' and role_columns(store)==COLUMNS with store.pool.connection() as c: after=c.execute('SELECT * FROM chats WHERE id=%s',(chat,)).fetchone() assert c.execute('SELECT count(*) AS n FROM archive_migrations WHERE version=46').fetchone()['n']==1 @@ -56,7 +56,7 @@ def test_schema46_backup_carries_role_columns_and_verifies(seeded,tmp_path): role_observed_at=clock_timestamp() WHERE id=%s''',(chat,)) create_snapshot(store,tmp_path/'capture',DUMP) manifest=verify_bundle(tmp_path/'capture') - assert manifest['schema_version']==49 and len(manifest['tables'])==78 + assert manifest['schema_version']==50 and len(manifest['tables'])==79 assert verify_snapshot(store,tmp_path/'capture')['status']=='verified' # The chats digest covers the role columns: a changed role no longer matches the snapshot. with store.pool.connection() as c:c.execute("UPDATE chats SET my_role='admin' WHERE id=%s",(chat,)) @@ -99,7 +99,7 @@ def test_prepared_chat_queries_survive_migration_046_on_a_serving_connection(dat commands._chat(c,p,'roles-tg',chat);outgoing.target(c,p,'roles-tg',chat,'') c.commit() store.migrate() # another pool connection - assert schema(store)=='49' + assert schema(store)=='50' assert commands._chat(c,p,'roles-tg',chat)['provider_chat']=='-1001' assert outgoing.target(c,p,'roles-tg',chat,'')[0]['platform']=='telegram' @@ -119,7 +119,7 @@ def test_real_schema45_dump_restores_verifies_and_quarantines_before_migration_0 assert not c.execute('SELECT 1 FROM account_policies WHERE enabled OR send_enabled').fetchone() assert not c.execute('SELECT 1 FROM api_keys WHERE NOT revoked').fetchone() restored.migrate() - assert schema(restored)=='49' and role_columns(restored)==COLUMNS + assert schema(restored)=='50' and role_columns(restored)==COLUMNS with restored.pool.connection() as c: row=c.execute('SELECT provider_chat,my_role,can_post FROM chats WHERE id=%s',(chat,)).fetchone() assert row=={'provider_chat':'-1001','my_role':None,'can_post':None} diff --git a/tests/test_collectors.py b/tests/test_collectors.py index f7b46d8a..70560169 100644 --- a/tests/test_collectors.py +++ b/tests/test_collectors.py @@ -768,7 +768,7 @@ def test_backup_covers_collector_positions_and_pending_deletions(worker,tmp_path directory=tmp_path/'snapshot' create_snapshot(s,directory,[sys.executable,'-c','print("unit-dump-stub")']) manifest=json.loads((directory/'manifest.json').read_text()) - assert manifest['schema_version']==49 + assert manifest['schema_version']==50 assert manifest['tables']['collector_chats']['rows']==1 assert manifest['tables']['provider_tombstones']['rows']==1 assert verify_snapshot(s,directory)['status']=='verified' diff --git a/tests/test_compose_runtime.py b/tests/test_compose_runtime.py index 478c0705..2b5aceb8 100644 --- a/tests/test_compose_runtime.py +++ b/tests/test_compose_runtime.py @@ -21,16 +21,18 @@ REPO = Path(__file__).resolve().parents[1] BASE = ["compose.yaml", "compose.collectors.yaml", "compose.whatsapp.yaml", "compose.indexer.yaml", "compose.enrichment.yaml", "compose.subscriptions.yaml", "compose.operator.yaml", - "compose.file-lifecycle.yaml", "compose.file-understanding.yaml", "compose.message-purge.yaml"] + "compose.file-lifecycle.yaml", "compose.file-understanding.yaml", "compose.message-purge.yaml", + "compose.delivery.yaml"] NATIVE = "compose.whatsapp-native.yaml" APPLICATIONS = {"api", "telegram", "whatsapp1", "whatsapp2", "indexer", "enricher", "enricher-local", "operator", "subscriptions", "file-lifecycle", "cold-archive", "cold-restore", - "file-understanding", "message-purge"} + "file-understanding", "message-purge", "delivery"} COLLECTORS = {"telegram", "whatsapp1", "whatsapp2"} SECRET_DIRECTORIES = ["processing-locks", "telegram-session", "whatsapp1-session", "whatsapp2-session", - "issued-keys", "cold-archive-identity", "cold-archive-ssh", "enrichment-models"] + "issued-keys", "cold-archive-identity", "cold-archive-ssh", "enrichment-models", "delivery-receivers"] SECRET_FILES = ["app.dsn", "collector.dsn", "indexer.dsn", "owner.dsn", "subscriptions.dsn", "enrichment.dsn", "enrichment-local.dsn", "file-lifecycle.dsn", "cold-archive.dsn", "file-understanding.dsn", "message-purge.dsn", + "delivery.dsn", "delivery-policy.json", "enrichment.keys", "enrichment.json", "enrichment-local.json", "file-understanding.json", "cold-archive.json", "telegram.json", "telegram.credentials.json", "telegram.permit.json", "whatsapp1.json", "whatsapp1.permit.json", "whatsapp2.json", "whatsapp2.permit.json"] @@ -159,6 +161,10 @@ def networks_of(container): return set(container["NetworkSettings"]["Networks"]) +def members_of(containers): + return {name: networks_of(container) for name, container in containers.items()} + + def assert_runtime(stack, declared, native): containers = stack.containers() assert set(containers) == APPLICATIONS | {"db"} @@ -169,9 +175,13 @@ def assert_runtime(stack, declared, native): db = containers["db"] assert db["State"]["Running"] assert db["HostConfig"]["LogConfig"] == {"Type": "json-file", "Config": {"max-size": "5m", "max-file": "2"}} - assert networks_of(db) == {f"{stack.project}_backend"} + assert networks_of(db) == {f"{stack.project}_backend", f"{stack.project}_delivery_db"} assert stack.network("backend")["Internal"] is True - members = {name: networks_of(container) for name, container in containers.items()} + # ANYM-534: the dispatcher shares only the database network, and its egress bridge with nobody. + assert {name for name, joined in members_of(containers).items() if f"{stack.project}_delivery_db" in joined} == {"db", "delivery"} + assert networks_of(containers["delivery"]) == {f"{stack.project}_delivery_db", f"{stack.project}_delivery_egress"} + assert stack.network("delivery_db")["Internal"] is True and stack.network("delivery_egress")["Internal"] is False + members = members_of(containers) assert {name for name, joined in members.items() if f"{stack.project}_collector_egress" in joined} == COLLECTORS assert {name for name, joined in members.items() if f"{stack.project}_cold_archive_egress" in joined} == {"cold-archive", "cold-restore"} assert stack.network("collector_egress")["Internal"] is False diff --git a/tests/test_contact_cards.py b/tests/test_contact_cards.py index ff31b92c..25c1294e 100644 --- a/tests/test_contact_cards.py +++ b/tests/test_contact_cards.py @@ -288,7 +288,7 @@ def test_migration_047_indexes_only_contact_metadata_and_repeats_over_schema46_r assert c.execute("SELECT to_regclass('message_metadata_contact') AS i").fetchone()['i'] is None store.migrate();store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' assert c.execute('SELECT count(*) AS n FROM archive_migrations WHERE version=47').fetchone()['n']==1 index=c.execute('''SELECT i.indisvalid,i.indisready,pg_get_expr(i.indpred,i.indrelid) AS predicate FROM pg_index i WHERE i.indexrelid='message_metadata_contact'::regclass''').fetchone() @@ -314,7 +314,7 @@ def test_migration_047_keeps_an_index_built_before_it(database,tmp_path): before=c.execute("SELECT 'message_metadata_contact'::regclass::oid AS i").fetchone()['i'] store.migrate();store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' assert c.execute('SELECT count(*) AS n FROM archive_migrations WHERE version=47').fetchone()['n']==1 assert c.execute("SELECT 'message_metadata_contact'::regclass::oid AS i").fetchone()['i']==before @@ -332,6 +332,6 @@ def test_migration_047_refuses_an_unusable_prebuilt_index(database,tmp_path,shap c.execute('DROP INDEX message_metadata_contact') store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' assert c.execute('''SELECT indisvalid FROM pg_index WHERE indexrelid='message_metadata_contact'::regclass''').fetchone()['indisvalid'] diff --git a/tests/test_coverage_backup.py b/tests/test_coverage_backup.py index 339b3b68..eaf24d49 100644 --- a/tests/test_coverage_backup.py +++ b/tests/test_coverage_backup.py @@ -608,7 +608,7 @@ def test_actual_postgres_age_receiver_restore_drill_preserves_source_and_quarant input=(tmp_path / 'restored/database.dump').read_bytes(), check=True, capture_output=True) restored = Store(make_conninfo(source.pool.conninfo, dbname=restored_name), tmp_path / 'restored/files', 'http://testserver') - assert verify_snapshot(restored, tmp_path / 'restored')['tables'] == 78 + assert verify_snapshot(restored, tmp_path / 'restored')['tables'] == 79 quarantine_restore(restored) with restored.pool.connection() as connection: keys = connection.execute('SELECT revoked FROM api_keys').fetchall() @@ -616,7 +616,7 @@ def test_actual_postgres_age_receiver_restore_drill_preserves_source_and_quarant assert keys and all(row['revoked'] for row in keys) assert policies and all(not row['enabled'] for row in policies) # The original live database, CAS and fake cold session are unchanged. - assert verify_snapshot(source, tmp_path / 'restored')['tables'] == 78 + assert verify_snapshot(source, tmp_path / 'restored')['tables'] == 79 assert all(hashlib.sha256((source.files / digest).read_bytes()).hexdigest() == value for digest, value in source_hashes.items()) assert (tmp_path / 'session/example.session').read_bytes() == b'FAKE_SESSION_BYTES' diff --git a/tests/test_coverage_deployment.py b/tests/test_coverage_deployment.py index 5d6efc6b..49565db7 100644 --- a/tests/test_coverage_deployment.py +++ b/tests/test_coverage_deployment.py @@ -67,7 +67,7 @@ def test_application_units_enforce_sandbox_resource_and_log_bounds(): assert set(applications) == {'api', 'db', 'telegram', 'whatsapp1', 'whatsapp2', 'indexer', 'enricher', 'enricher-local', 'operator', 'subscriptions', 'calendar-db', 'calendar-api', 'calendar-worker', 'gateway', 'calls', - 'file-lifecycle', 'file-understanding', 'cold-archive', 'cold-restore', 'message-purge'} + 'file-lifecycle', 'file-understanding', 'cold-archive', 'cold-restore', 'message-purge', 'delivery'} for name, unit in applications.items(): if name in {'db', 'calendar-db'}: continue diff --git a/tests/test_coverage_subscription_deploy.py b/tests/test_coverage_subscription_deploy.py index e00d2fc4..c3bd5aa1 100644 --- a/tests/test_coverage_subscription_deploy.py +++ b/tests/test_coverage_subscription_deploy.py @@ -76,3 +76,65 @@ def render(*files): assert service['stop_grace_period'] == '15s' assert service['logging'] == {'driver': 'json-file', 'options': {'max-size': '5m', 'max-file': '2'}} + + +# spec: ANYM-534 +@pytest.mark.skipif( + shutil.which('docker') is None and not os.environ.get('CI'), + reason='Compose rendering needs the host Docker CLI; the isolated test image intentionally has no Docker CLI or socket.', +) +def test_delivery_overlay_isolates_the_dispatcher_database_path_and_egress(tmp_path): + assert shutil.which('docker') is not None, 'CI must provide the Docker CLI for the Compose deployment proof' + root = Path(__file__).resolve().parents[1] + env = {**os.environ, 'PROBE_PUBLIC_URL': 'https://example.invalid', 'PROBE_ALLOWED_HOSTS': 'example.invalid'} + raw = yaml.safe_load((root / 'deploy/compose.delivery.yaml').read_text()) + for name in ('delivery', 'api'): + # Raw Compose defaults host-path creation to true: require explicit false on every bind. + assert all(volume['bind']['create_host_path'] is False for volume in raw['services'][name]['volumes']), name + base = yaml.safe_load((root / 'deploy/compose.yaml').read_text()) + fixture_env = tmp_path / 'postgres.synthetic.env' + fixture_env.write_text('POSTGRES_USER=synthetic\nPOSTGRES_PASSWORD=synthetic\nPOSTGRES_DB=synthetic\n') + base['services']['db']['env_file'] = str(fixture_env) + fixture_base = tmp_path / 'compose.synthetic.yaml' + fixture_base.write_text(yaml.safe_dump(base, sort_keys=False)) + command = ['docker', 'compose', '--project-directory', str(root / 'deploy'), '-f', str(fixture_base)] + for name in ('compose.subscriptions.yaml', 'compose.delivery.yaml'): + command += ['-f', str(root / 'deploy' / name)] + result = subprocess.run(command + ['config', '--format', 'json'], env=env, text=True, capture_output=True, timeout=20) + assert result.returncode == 0, f'Compose render failed ({result.returncode}): {result.stderr}' + config = json.loads(result.stdout) + services, networks = config['services'], config['networks'] + service = services['delivery'] + assert service['command'] == ['python', '-m', 'messaging_service.delivery_worker'] + assert service['build']['target'] == 'runtime' and service['init'] is True and service['restart'] == 'unless-stopped' + # A database-only internal network plus its own egress bridge; never the backend network of the API and matcher. + assert set(service['networks']) == {'delivery_db', 'delivery_egress'} + assert {name for name, item in services.items() if 'delivery_db' in item['networks']} == {'db', 'delivery'} + assert {name for name, item in services.items() if 'delivery_egress' in item['networks']} == {'delivery'} + assert set(services['db']['networks']) == {'backend', 'delivery_db'} + assert set(services['subscriptions']['networks']) == {'backend'} + assert networks['delivery_db']['internal'] is True and networks['delivery_db'].get('enable_ipv6') is False + assert networks['delivery_db']['driver_opts'] == {'com.docker.network.bridge.name': 'dk-deliverydb0'} + # Explicit dns: Docker queries it from the container namespace, where the host rules apply. + assert service['dns'] == ['1.1.1.1'] + assert not networks['delivery_egress'].get('internal') and networks['delivery_egress'].get('enable_ipv6') is False + assert networks['delivery_egress']['driver_opts'] == {'com.docker.network.bridge.name': 'dk-delivery0'} + assert not service.get('ports') + assert service['environment'] == {'DATABASE_URL_FILE': '/run/secrets/database_url', + 'DELIVERY_POLICY_FILE': '/run/secrets/delivery-policy.json'} + mounts = {volume['target']: volume for volume in service['volumes']} + assert set(mounts) == {'/run/secrets/database_url', '/run/secrets/delivery-policy.json', '/run/secrets/receivers'} + assert all(volume['type'] == 'bind' and volume['read_only'] is True for volume in mounts.values()) + assert Path(mounts['/run/secrets/database_url']['source']) == root / 'deploy/private/delivery.dsn' + assert Path(mounts['/run/secrets/delivery-policy.json']['source']) == root / 'deploy/private/delivery-policy.json' + assert Path(mounts['/run/secrets/receivers']['source']) == root / 'deploy/private/delivery-receivers' + assert service['read_only'] is True and service['user'] == '10001:10001' and service['cap_drop'] == ['ALL'] + assert 'no-new-privileges:true' in service['security_opt'] + assert service['tmpfs'] == ['/tmp:size=16m,noexec,nosuid'] + assert int(service['mem_limit']) == 256 * 1024**2 and service['cpu_shares'] == 512 and service['pids_limit'] == 64 + assert service['logging'] == {'driver': 'json-file', 'options': {'max-size': '5m', 'max-file': '2'}} + # The API reads the same host policy for registration; nothing else of the API changes. + api = services['api'] + assert api['environment']['DELIVERY_POLICY_FILE'] == '/run/secrets/delivery-policy.json' + policy = [volume for volume in api['volumes'] if volume['target'] == '/run/secrets/delivery-policy.json'] + assert len(policy) == 1 and policy[0]['read_only'] is True diff --git a/tests/test_db_role_isolation.py b/tests/test_db_role_isolation.py index 322a4518..15d65c1d 100644 --- a/tests/test_db_role_isolation.py +++ b/tests/test_db_role_isolation.py @@ -451,7 +451,7 @@ def test_upgrade_and_rollback_keep_feed_order_and_cursors(schema36_store, monkey store.migrate() store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' assert c.execute('SELECT * FROM archive_migrations WHERE version<=36 ORDER BY version').fetchall() == migrations assert feed_state(store) == before try: diff --git a/tests/test_delivery_egress_check.py b/tests/test_delivery_egress_check.py new file mode 100644 index 00000000..0703f15c --- /dev/null +++ b/tests/test_delivery_egress_check.py @@ -0,0 +1,113 @@ +"""The dispatcher's egress check and host rules (ANYM-534); local sockets only, no firewall is changed.""" +import json +from pathlib import Path +import socket +import subprocess +import sys + +SCRIPT = Path(__file__).resolve().parents[1] / 'scripts/delivery_egress_check.py' + + +def run(*args): + result = subprocess.run([sys.executable, '-I', str(SCRIPT), *args], text=True, capture_output=True, timeout=30) + return result.returncode, result.stdout + + +# spec: ANYM-534 +def test_check_passes_only_when_open_targets_connect_and_blocked_ones_never_answer(): + with socket.socket() as closed: + # A port nobody holds: every OS refuses it (a bound, non-listening one times out on macOS). + closed.bind(('127.0.0.1', 0)) + refused_target = f'127.0.0.1:{closed.getsockname()[1]}' + with socket.socket() as listener: + listener.bind(('127.0.0.1', 0)) + listener.listen() + open_target = f'127.0.0.1:{listener.getsockname()[1]}' + # TEST-NET-1 has no route or no answer: what a DROP looks like from inside the dispatcher. + code, out = run('check', '--timeout', '1', '--open', open_target, '--blocked', '192.0.2.1:8080') + report = json.loads(out) + assert code == 0 and report['passed'] is True + assert [(r['target'], r['expect'], r['passed']) for r in report['results']] == [ + (open_target, 'open', True), ('192.0.2.1:8080', 'blocked', True)] + assert report['results'][0]['outcome'] == 'connected' and report['results'][1]['outcome'] in ('timeout', 'unreachable') + # A name sends no packet when it does not resolve, so a blocked target must be an IP literal: names and + # malformed targets fail instead of passing as "nothing answered". + code, out = run('check', '--timeout', '1', '--blocked', 'api.invalid:8080', '--blocked', 'api:8080', + '--blocked', '127.0.0.1:http') + report = json.loads(out) + assert code == 1 and [(r['outcome'], r['passed']) for r in report['results']] == [('malformed', False)] * 3 + # A refused connection proves the address answers: a DROP rule never sends a reset, so this fails. + code, out = run('check', '--timeout', '1', '--blocked', refused_target, '--blocked', open_target) + report = json.loads(out) + assert code == 1 and report['passed'] is False + assert [(r['outcome'], r['passed']) for r in report['results']] == [('refused', False), ('connected', False)] + # An open target that does not connect fails too. + code, out = run('check', '--timeout', '1', '--open', 'db.invalid:5432') + assert code == 1 and json.loads(out)['results'][0]['outcome'] == 'unresolved' + + +# spec: ANYM-534 +def test_dns_probes_need_an_answer_or_silence_and_the_listener_is_a_positive_control(): + import threading + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as resolver, \ + socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as silent: + resolver.bind(('127.0.0.1', 0)) + silent.bind(('127.0.0.1', 0)) + def answer(): + query, peer = resolver.recvfrom(512) + resolver.sendto(query[:2] + b'\x81\x80' + query[4:], peer) + thread = threading.Thread(target=answer, daemon=True) + thread.start() + allowed, other = (f'127.0.0.1:{s.getsockname()[1]}' for s in (resolver, silent)) + code, out = run('check', '--timeout', '1', '--open-dns', allowed, '--blocked-dns', other) + thread.join(5) + assert code == 0, out + assert [(r['expect'], r['outcome']) for r in json.loads(out)['results']] == [('open-dns', 'answered'), ('blocked-dns', 'timeout')] + # A resolver that does not answer (the port is closed now) fails as open; a name is malformed. + code, out = run('check', '--timeout', '1', '--open-dns', other, '--blocked-dns', 'resolver.invalid') + results = json.loads(out)['results'] + assert code == 1 and results[0]['outcome'] in ('refused', 'timeout') and not results[0]['passed'] + assert (results[1]['outcome'], results[1]['passed']) == ('malformed', False) + # The synthetic listener (run in the matcher container) accepts until its time is up. + listener = subprocess.Popen([sys.executable, '-I', str(SCRIPT), 'listen', '--port', '0', '--seconds', '10'], + stdout=subprocess.PIPE, text=True) + try: + port = json.loads(listener.stdout.readline())['listening'] + code, out = run('check', '--timeout', '1', '--open', f'127.0.0.1:{port}') + assert code == 0 and json.loads(out)['results'][0]['outcome'] == 'connected' + finally: + listener.kill() + listener.wait(5) + + +# spec: ANYM-534 +def test_rules_drop_private_tailnet_metadata_and_host_and_allow_only_dns_and_443(): + code, out = run('rules', '--resolver', '1.1.1.1') + assert code == 0 + lines = out.splitlines() + chain = [line for line in lines if line.startswith('iptables -A DEEDKIT-DELIVERY')] + for network in ('10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16', '100.64.0.0/10', '169.254.0.0/16', '127.0.0.0/8'): + assert f'iptables -A DEEDKIT-DELIVERY -d {network} -j DROP' in chain, network + dns = [line for line in chain if '--dport 53' in line] + assert dns and all('-d 1.1.1.1' in line and line.endswith('-j RETURN') for line in dns) + # DNS first, then the private drops, then public 443, then drop everything else. + assert chain.index(dns[-1]) < chain.index('iptables -A DEEDKIT-DELIVERY -d 10.0.0.0/8 -j DROP') + assert chain[-2:] == ['iptables -A DEEDKIT-DELIVERY -p tcp --dport 443 -j RETURN', 'iptables -A DEEDKIT-DELIVERY -j DROP'] + assert 'iptables -I DOCKER-USER -i dk-delivery0 -j DEEDKIT-DELIVERY' in lines + # The host itself (INPUT, not forwarded) and every IPv6 packet (ULA, NAT64) are dropped. + # The internal database bridge has a host gateway address too: host-bound packets drop on both bridges. + for bridge in ('dk-delivery0', 'dk-deliverydb0'): + assert f'iptables -I INPUT -i {bridge} -j DROP' in lines and f'ip6tables -I INPUT -i {bridge} -j DROP' in lines + assert 'ip6tables -I DOCKER-USER -i dk-delivery0 -j DROP' in lines + # The undo removes every inserted rule and the chain. + code, out = run('rules', '--undo') + assert code == 0 + undo = out.splitlines() + inserted = [line.replace(' -I ', ' -D ', 1) for line in lines if ' -I ' in line] + assert sorted(line for line in undo if ' -D ' in line) == sorted(inserted) + assert undo[-2:] == ['iptables -F DEEDKIT-DELIVERY', 'iptables -X DEEDKIT-DELIVERY'] + # A resolver inside a dropped range (a tailnet DNS) is allowed only on port 53, before the drops. + code, out = run('rules', '--resolver', '100.100.100.100') + assert code == 0 and 'iptables -A DEEDKIT-DELIVERY -d 100.100.100.100 -p udp --dport 53 -j RETURN' in out + code, _ = run('rules', '--resolver', 'not-an-address') + assert code != 0 diff --git a/tests/test_delivery_endpoints.py b/tests/test_delivery_endpoints.py index b530386f..fa0251af 100644 --- a/tests/test_delivery_endpoints.py +++ b/tests/test_delivery_endpoints.py @@ -688,3 +688,87 @@ def receiver(url, body, headers, **kwargs): with store.pool.connection() as c: c.execute("UPDATE delivery_endpoints SET state='auto_paused' WHERE id=%s", (eid,)) assert store.update_delivery_endpoint(token, eid, version + 1, 'pause')['state'] == 'agent_paused' + + +def features_cli(store, tmp_path, monkeypatch, capsys, *args): + """Run `python -m messaging_service set-delivery-features` on the test database; (exit code, JSON).""" + import sys + import messaging_service.__main__ as cli + dsn = tmp_path / 'operator.dsn' + dsn.write_text(store.pool.conninfo) + dsn.chmod(0o600) + class BorrowedStore: + def __getattr__(self, name): + return getattr(store, name) + def close(self): + pass # The seeded fixture owns this connection pool. + monkeypatch.setattr(cli, 'Store', lambda *_: BorrowedStore()) + monkeypatch.setenv('DATABASE_URL_FILE', str(dsn)) + monkeypatch.setattr(sys, 'argv', ['messaging_service', 'set-delivery-features', *args]) + capsys.readouterr() + try: + cli.main() + code = 0 + except SystemExit as exit: + code = exit.code + return code, json.loads(capsys.readouterr().out.strip().splitlines()[-1]) + + +def beats(store, **components): + """Synthetic heartbeats: component=(version, schema_version, age_s); None deletes the row.""" + with store.pool.connection() as c: + for name, value in components.items(): + c.execute('DELETE FROM component_heartbeats WHERE component=%s', (name,)) + if value is not None: + c.execute('''INSERT INTO component_heartbeats(component,version,schema_version,beat_at) + VALUES(%s,%s,%s,clock_timestamp()-make_interval(secs=>%s))''', (name, *value)) + + +def gate_state(store): + with store.pool.connection() as c: + row = c.execute("SELECT value FROM probe_metadata WHERE name='delivery_features'").fetchone() + return row and row['value'] + + +# spec: ANYM-534 +def test_enable_refuses_until_api_matcher_and_dispatcher_report_one_version(seeded, policy, tmp_path, monkeypatch, + capsys): + from messaging_service import __version__ + from messaging_service.store import RUNTIME_SCHEMA + store = seeded[0] + current = (__version__, RUNTIME_SCHEMA, 0) + store.set_delivery_features(False) + try: + # An older matcher writes no heartbeat at all; one that does reports its own version. + for matcher, reason in ((None, 'component_heartbeat_absent'), (('0.0.1', RUNTIME_SCHEMA, 0), 'component_version_mismatch'), + ((__version__, RUNTIME_SCHEMA - 1, 0), 'component_schema_mismatch'), + ((__version__, RUNTIME_SCHEMA, 121), 'component_heartbeat_stale')): + beats(store, api=current, dispatcher=current, matcher=matcher) + code, result = features_cli(store, tmp_path, monkeypatch, capsys, '--enable') + assert code != 0 and result['error'] == reason and result['component'] == 'matcher', matcher + assert gate_state(store) == 'disabled' + beats(store, matcher=current) + assert features_cli(store, tmp_path, monkeypatch, capsys, '--enable') == (0, {'delivery_features': 'enabled', + 'cancelled': 0}) + assert gate_state(store) == 'enabled' + finally: + store.set_delivery_features(False) + beats(store, api=None, matcher=None, dispatcher=None) + + +# spec: ANYM-534 +def test_disable_cancels_queued_endpoint_hints_and_events_stay_fetchable(gate, policy, tmp_path, monkeypatch, capsys): + store = gate + token = identity(store) + endpoint = register(store, token) + sid = store.create_subscription(token, ['telegram-demo-chat'], delivery={'endpoint_id': endpoint['endpoint_id']})[ + 'subscription_id'] + event, = emit(store) + match_subscriptions(store) + code, result = features_cli(store, tmp_path, monkeypatch, capsys, '--disable') + assert code == 0 and result['delivery_features'] == 'disabled' and result['cancelled'] >= 1 + with store.pool.connection() as c: + assert not c.execute("""SELECT 1 FROM subscription_outbox WHERE endpoint=%s AND state IN ('pending','leased')""", + (endpoint['endpoint_id'],)).fetchone() + assert dispatch_hints(store, transport=lambda *a, **k: pytest.fail('POST after disable'), resolver=PUBLIC)['delivered'] == 0 + assert [item['event_id'] for item in store.fetch_subscription_events(token, sid)['items']] == [event] diff --git a/tests/test_event_delivery.py b/tests/test_event_delivery.py index cc6a9d2e..096d848c 100644 --- a/tests/test_event_delivery.py +++ b/tests/test_event_delivery.py @@ -882,3 +882,343 @@ def transport(*args, **kwargs): assert after['cancelled'] == 1 finally: store.set_delivery_features(False) + + +PUBLIC = lambda *a, **k: ['93.184.216.34'] # noqa: E731 + + +def endpoint_keys(store, tmp_path, monkeypatch, subscriptions): + """One key per item, each with one bound self-service endpoint and that many subscriptions (ANYM-534).""" + import base64 + policy = tmp_path / 'delivery-policy.json' + policy.write_text(json.dumps({'allow_hosts': ['example.test']})) + policy.chmod(0o600) + monkeypatch.setenv('DELIVERY_POLICY_FILE', str(policy)) + store.set_delivery_features(True) + permissions = ['read', 'subscription-write', 'subscription-ack', 'delivery-write'] + keys = [] + for n, count in enumerate(subscriptions): + token = store.issue_key(f'lane-agent-{n}', ['telegram-demo'], permissions=permissions, key_permissions=permissions) + secret = 'whsec_' + base64.b64encode(b'synthetic-lane-secret-%02d-0123456789' % n).decode() + endpoint = store.register_delivery_endpoint(token, f'lane-request-{n}', f'https://slow{n}.example.test/hook', + secret, resolver=PUBLIC) + sids = [store.create_subscription(token, ['telegram-demo-chat'], label=f'Lane {n}.{i}', + delivery={'endpoint_id': endpoint['endpoint_id']})['subscription_id'] for i in range(count)] + keys.append((token, endpoint['endpoint_id'], sids)) + with store.pool.connection() as c: + # Only hints are under test here, not the inline registration tests. + c.execute("UPDATE delivery_endpoint_tests SET state='cancelled' WHERE state<>'cancelled'") + emit(store) + match_subscriptions(store) + return keys + + +# spec: ANYM-534 +def test_operator_receiver_hint_is_on_time_while_four_endpoints_hang(seeded, tmp_path, monkeypatch): + import threading + import time + from messaging_service import delivery_worker + store, _, _ = seeded + try: + endpoint_keys(store, tmp_path, monkeypatch, [1, 1, 1, 1, 1]) + token = identity(store) + sid = subscription(store, token) + rid, _ = receiver(store, token, tmp_path) + call(store, token, 'configure_subscription_delivery', sid, 1, rid) + # The endpoint hints are older, so a serial dispatcher would POST all of them first. + emit(store) + match_subscriptions(store) + due, sent, hung = time.monotonic(), {}, [] + def transport(url, body, headers, *, resolved, timeout): + if url.startswith('https://slow0.'): + raise RuntimeError('synthetic endpoint bug') + if url.startswith('https://slow'): + hung.append(url) + time.sleep(timeout) + raise TimeoutError + sent[url] = time.monotonic() + return 204 + stop = threading.Event() + worker = threading.Thread(target=delivery_worker.run, args=(store, stop), + kwargs={'transport': transport, 'resolver': PUBLIC, 'interval': 0.2}) + worker.start() + try: + while 'https://hooks.example.test/wake' not in sent and time.monotonic() < due + 15: + time.sleep(0.05) + assert worker.is_alive(), 'an endpoint failure stopped the dispatcher' + finally: + stop.set() + worker.join(30) + assert not worker.is_alive() + assert sent['https://hooks.example.test/wake'] - due < 15 and hung + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_endpoint_lane_keeps_one_attempt_per_key_in_flight_and_serves_keys_in_turn(seeded, tmp_path, monkeypatch): + import threading + import time + from messaging_service.event_delivery import dispatch_hints + store, _, _ = seeded + try: + endpoint_keys(store, tmp_path, monkeypatch, [3, 1]) + served, inflight, release = {}, [], threading.Event() + def blocking(url, body, headers, **kwargs): + inflight.append(url) + assert release.wait(10) + return 204 + def lane(): + dispatch_hints(store, lane='endpoint', limit=1, served=served, transport=blocking, resolver=PUBLIC) + threads = [] + for n in (1, 2): + threads.append(threading.Thread(target=lane)) + threads[-1].start() + deadline = time.monotonic() + 10 + while len(inflight) < n: + assert time.monotonic() < deadline + time.sleep(0.02) + # Both keys have one attempt in flight: key 0's other two hints wait, and the operator lane sees none. + idle = dispatch_hints(store, lane='endpoint', limit=1, served=served, + transport=lambda *a, **k: pytest.fail('a second attempt for a key in flight'), resolver=PUBLIC) + assert idle['delivered'] == idle['retried'] == 0 + assert dispatch_hints(store, lane='operator', transport=lambda *a, **k: pytest.fail('endpoint hint in the operator lane'), + resolver=PUBLIC)['delivered'] == 0 + release.set() + for thread in threads: + thread.join(10) + assert sorted(url.split('.')[0] for url in inflight) == ['https://slow0', 'https://slow1'] + rest = [] + for _ in range(3): + dispatch_hints(store, lane='endpoint', limit=1, served=served, + transport=lambda url, *a, **k: rest.append(url) or 204, resolver=PUBLIC) + assert [url.split('.')[0] for url in rest] == ['https://slow0', 'https://slow0'] + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_the_least_recently_served_key_goes_first(seeded, tmp_path, monkeypatch): + from messaging_service.event_delivery import dispatch_hints + store, _, _ = seeded + try: + endpoint_keys(store, tmp_path, monkeypatch, [3, 2]) + served, order = {}, [] + for _ in range(5): + dispatch_hints(store, lane='endpoint', limit=1, served=served, + transport=lambda url, *a, **k: order.append(url.split('.')[0]) or 204, resolver=PUBLIC) + # Key 0's three hints are as old as key 1's, yet the keys take turns. + assert order in (['https://slow0', 'https://slow1', 'https://slow0', 'https://slow1', 'https://slow0'], + ['https://slow1', 'https://slow0', 'https://slow1', 'https://slow0', 'https://slow0']) + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_a_key_endpoint_queue_is_bounded_and_events_stay_fetchable(seeded, tmp_path, monkeypatch): + from messaging_service import event_delivery + from messaging_service.delivery_policy import flush_hints + from messaging_service.event_delivery import dispatch_hints + store, _, _ = seeded + monkeypatch.setattr(event_delivery, 'ENDPOINT_QUEUE_PER_KEY', 2) + try: + [(token, _, sids)] = endpoint_keys(store, tmp_path, monkeypatch, [3]) + def queued(): + with store.pool.connection() as c: + return c.execute("""SELECT count(*) AS n FROM subscription_outbox WHERE endpoint IS NOT NULL + AND state IN ('pending','leased')""").fetchone()['n'] + def held(): + with store.pool.connection() as c: + return c.execute("""SELECT subscription FROM subscription_hint_windows WHERE held_reason='budget' + AND first_match_at IS NOT NULL""").fetchall() + # The bound is key-wide, across subscriptions: the third subscription's wake is held, not dropped. + assert queued() == 2 and len(held()) == 1 + for _ in range(3): + emit(store) + match_subscriptions(store) + assert queued() <= 2 + # An explicit test of a full key is refused like an over-budget one. + with pytest.raises(APIError, match='rate_limited'): + call(store, token, 'test_subscription', sids[0], 'full-queue-probe') + assert queued() == 2 + assert len(call(store, token, 'fetch_subscription_events', sids[0])['items']) == 4 + # After each drain the held windows fire on their next tick, never more than the bound at once. + def drain_and_tick(): + dispatch_hints(store, lane='endpoint', limit=8, transport=lambda *a, **k: 204, resolver=PUBLIC) + assert queued() == 0 + with store.pool.connection() as c: + c.execute("UPDATE subscription_hint_windows SET due_at=clock_timestamp() WHERE due_at IS NOT NULL") + flush_hints(store) + drain_and_tick() + assert queued() == 2 and len(held()) == 1 + drain_and_tick() + assert queued() == 1 and not held() + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_a_key_never_has_an_endpoint_test_and_a_hint_in_flight_together(seeded, tmp_path, monkeypatch): + import base64 + from messaging_service.event_delivery import dispatch_hints, endpoint_tests + store, _, _ = seeded + try: + [(token, _, _)] = endpoint_keys(store, tmp_path, monkeypatch, [1]) + # A second endpoint of the same key stages its registration test: a different URL, the same key. + with store.pool.connection() as c: + # Past the per-key registration and test rate limits (one a minute). + c.execute("UPDATE delivery_endpoints SET created_at=created_at-interval '2 minutes' WHERE state<>'deleted'") + c.execute("UPDATE delivery_endpoint_tests SET created_at=created_at-interval '2 minutes' WHERE state='cancelled'") + store.register_delivery_endpoint(token, 'lane-request-second', 'https://second.example.test/hook', + 'whsec_' + base64.b64encode(b'synthetic-second-secret-0123456789').decode(), resolver=PUBLIC) + def tests(state): + with store.pool.connection() as c: + return c.execute('SELECT count(*) AS n FROM delivery_endpoint_tests WHERE state=%s', (state,)).fetchone()['n'] + assert tests('pending') == 1 + seen = [] + def hint_transport(url, *a, **k): + # While the key's hint is in flight, the test pass leaves the key's test pending. + seen.append(('hint', url)) + assert endpoint_tests(store, transport=lambda *a, **k: pytest.fail('test POST beside the hint'), + resolver=PUBLIC)['delivered'] == 0 + return 204 + assert dispatch_hints(store, lane='endpoint', limit=1, served={}, transport=hint_transport, + resolver=PUBLIC)['delivered'] == 1 + assert tests('pending') == 1 + emit(store) + match_subscriptions(store) + def test_transport(url, *a, **k): + # And while the key's test is in flight, the endpoint lane leaves the key's hint pending. + seen.append(('test', url)) + assert dispatch_hints(store, lane='endpoint', limit=1, served={}, + transport=lambda *a, **k: pytest.fail('hint POST beside the test'), resolver=PUBLIC)['delivered'] == 0 + return 204 + assert endpoint_tests(store, transport=test_transport, resolver=PUBLIC)['delivered'] == 1 + assert [kind for kind, _ in seen] == ['hint', 'test'] and seen[1][1] == 'https://second.example.test/hook' + # An expired test lease is not in flight: the hint goes. + with store.pool.connection() as c: + c.execute("""INSERT INTO delivery_endpoint_tests(endpoint,request_id,state,lease_token,lease_until) + SELECT endpoint,'expired-lease','leased',gen_random_uuid(),clock_timestamp()-interval '1 second' + FROM delivery_endpoint_tests LIMIT 1""") + assert dispatch_hints(store, lane='endpoint', limit=1, served={}, transport=lambda *a, **k: 204, + resolver=PUBLIC)['delivered'] == 1 + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_endpoint_tests_take_the_lanes_least_recently_served_key_order(seeded, tmp_path, monkeypatch): + from messaging_service.event_delivery import dispatch_hints, endpoint_tests + store, _, _ = seeded + try: + [(_, first, _), (_, second, _)] = endpoint_keys(store, tmp_path, monkeypatch, [1, 0]) + served, order = {}, [] + record = lambda url, *a, **k: order.append(url.split('.')[0]) or 204 # noqa: E731 + assert dispatch_hints(store, lane='endpoint', limit=1, served=served, transport=record, resolver=PUBLIC)['delivered'] == 1 + with store.pool.connection() as c: + # Key 0's two tests are older than key 1's. + for endpoint, request, age in ((first, 'fairness-a1', 30), (first, 'fairness-a2', 20), (second, 'fairness-b1', 10)): + c.execute("""INSERT INTO delivery_endpoint_tests(endpoint,request_id,created_at) + VALUES(%s,%s,clock_timestamp()-make_interval(secs=>%s))""", (endpoint, request, age)) + # Key 0 was just served by its hint: key 1's newer test goes before key 0's older ones. + assert endpoint_tests(store, transport=record, resolver=PUBLIC, limit=3, served=served)['delivered'] == 3 + assert order == ['https://slow0', 'https://slow1', 'https://slow0', 'https://slow0'] + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_hints_and_endpoint_tests_share_one_key_order(seeded, tmp_path, monkeypatch): + from messaging_service.event_delivery import dispatch_hints, endpoint_tests + store, _, _ = seeded + try: + [(_, first, _), (_, second, _)] = endpoint_keys(store, tmp_path, monkeypatch, [1, 1]) + with store.pool.connection() as c: + keys = [tuple(c.execute('SELECT principal,key_id FROM delivery_endpoints WHERE id=%s', (endpoint,)).fetchone().values()) + for endpoint in (first, second)] + c.execute("""INSERT INTO delivery_endpoint_tests(endpoint,request_id,created_at) + VALUES(%s,'shared-order-a1',clock_timestamp()-interval '1 minute')""", (first,)) + order = [] + record = lambda url, *a, **k: order.append(url.split('.')[0]) or 204 # noqa: E731 + def hint(served): + return dispatch_hints(store, lane='endpoint', limit=1, served=served, transport=record, resolver=PUBLIC)['delivered'] + def test(served): + return endpoint_tests(store, transport=record, resolver=PUBLIC, limit=1, served=served)['delivered'] + # Key 0 was just served; key 1 waits with a hint only: key 0's test does not overtake it. + served = {keys[0]: 1} + assert test(served) == 0 and hint(served) == 1 and order == ['https://slow1'] + assert test(served) == 1 and order[-1] == 'https://slow0' + # Now key 1 waits with a test only, key 0 with a hint: key 0's hint does not overtake it. + with store.pool.connection() as c: + c.execute("""INSERT INTO delivery_endpoint_tests(endpoint,request_id,created_at) + VALUES(%s,'shared-order-b1',clock_timestamp()-interval '1 minute')""", (second,)) + served = {keys[0]: 1, keys[1]: 0} + assert hint(served) == 0 and test(served) == 1 and order[-1] == 'https://slow1' + assert hint(served) == 1 and order[-1] == 'https://slow0' + finally: + store.set_delivery_features(False) + + +# spec: ANYM-534 +def test_dispatcher_heartbeat_reports_version_and_stops_when_a_lane_stalls(seeded): + import time + from messaging_service import __version__, delivery_worker + from messaging_service.store import RUNTIME_SCHEMA + store, _, _ = seeded + assert delivery_worker.HEARTBEAT_S == 30 + def rows(): + with store.pool.connection() as c: + return (c.execute('SELECT * FROM subscription_dispatcher_heartbeat').fetchall(), + c.execute("SELECT * FROM component_heartbeats WHERE component='dispatcher'").fetchall()) + with store.pool.connection() as c: + c.execute('DELETE FROM subscription_dispatcher_heartbeat WHERE id=1') + c.execute("DELETE FROM component_heartbeats WHERE component IS NOT NULL") + now = time.monotonic() + assert delivery_worker.pulse(store, {'operator': now - 5, 'endpoint-0': now - delivery_worker.LANE_STALL_S - 1}, + now=now) is False + assert rows() == ([], []) + assert delivery_worker.pulse(store, {'operator': now - 5, 'endpoint-0': now}, now=now) is True + legacy, [component] = rows() + assert len(legacy) == 1 + assert (component['version'], component['schema_version']) == (__version__, RUNTIME_SCHEMA) + + +# spec: ANYM-534 +def test_each_role_writes_only_its_own_component_heartbeat(seeded): + from contextlib import contextmanager + from copy import copy + import psycopg + from psycopg import sql + from messaging_service.heartbeats import record + store, _, _ = seeded + files = {'dispatcher': ('delivery_role.sql', 'messaging_event_delivery'), + 'matcher': ('subscription_role.sql', 'messaging_subscriptions'), + 'api': ('gateway_role.sql', 'messaging_gateway')} + roles = [] + try: + for component, (filename, name) in files.items(): + role = 'synthetic_' + component + '_' + uuid4().hex + roles.append(role) + ddl = Path('services/messaging/messaging_service', filename).read_text().replace(name, role) + with store.pool.connection() as c: + c.execute(ddl) + class RestrictedPool: + @contextmanager + def connection(self, role=role): + with store.pool.connection() as c: + c.execute(sql.SQL('SET LOCAL ROLE {}').format(sql.Identifier(role))) + yield c + worker = copy(store) + worker.pool = RestrictedPool() + assert record(worker, component) is True + for other in set(files) - {component}: + with pytest.raises(psycopg.errors.InsufficientPrivilege): + with worker.pool.connection() as c: + c.execute('''INSERT INTO component_heartbeats(component,version,schema_version,beat_at) + VALUES(%s,'0.0.0',1,clock_timestamp()) ON CONFLICT(component) DO UPDATE + SET beat_at=EXCLUDED.beat_at''', (other,)) + finally: + with store.pool.connection() as c: + for role in roles: + c.execute(sql.SQL('DROP OWNED BY {}').format(sql.Identifier(role))) + c.execute(sql.SQL('DROP ROLE {}').format(sql.Identifier(role))) diff --git a/tests/test_extend_key.py b/tests/test_extend_key.py index aa7a8c36..88cfb783 100644 --- a/tests/test_extend_key.py +++ b/tests/test_extend_key.py @@ -305,7 +305,7 @@ def test_migration_036_adds_extend_key_audit_and_keeps_schema35_rows(schema35_st store.migrate() store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' assert c.execute('SELECT * FROM archive_migrations WHERE version<=35 ORDER BY version').fetchall() == migrations assert c.execute('SELECT * FROM audit_log ORDER BY id').fetchall() == audit_before assert {row['operation'] for row in audit_before} <= {'issue_key', 'revoke_key', 'capabilities'} @@ -341,7 +341,7 @@ def migrate(): holder.close() store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' with pytest.raises(psycopg.errors.CheckViolation), c.transaction(): c.execute("INSERT INTO audit_log(principal,operation,outcome) VALUES('agent-busy','not_an_operation','ok')") @@ -416,7 +416,7 @@ def test_migrate_gives_up_on_a_busy_messages_table_without_stalling_writers(seed replay_attempt, lambda c: c.execute('LOCK TABLE messages IN ROW EXCLUSIVE MODE')) assert outcome == ['lock_not_available'] assert writer_seconds < 5 and migrate_seconds < 10 - assert schema_state(store) == ('49', 49, True) + assert schema_state(store) == ('50', 50, True) store.migrate() @@ -434,7 +434,7 @@ def test_migrate_lock_timeout_rolls_back_migrations_it_already_applied(database, # Nothing of 017-021 stays: same version, no new migration rows, no new table. assert schema_state(store) == ('16', 16, False) store.migrate() - assert schema_state(store) == ('49', 49, True) + assert schema_state(store) == ('50', 50, True) # spec: ANYM-422 @@ -447,9 +447,9 @@ def test_migrate_never_holds_one_replay_lock_while_waiting_for_another(seeded, r replay_attempt, lambda c: c.execute('LOCK TABLE messages IN ROW EXCLUSIVE MODE')) assert outcome == ['lock_not_available'] assert writer_seconds < 1 and migrate_seconds < 10 - assert schema_state(store) == ('49', 49, True) + assert schema_state(store) == ('50', 50, True) store.migrate() - assert schema_state(store) == ('49', 49, True) + assert schema_state(store) == ('50', 50, True) # spec: ANYM-422 @@ -458,7 +458,7 @@ def test_migrate_replays_replay_tables_missing_from_a_partial_restore(database, partial = base[:base.index('CREATE TABLE IF NOT EXISTS deliveries')] with old_schema_store(database, tmp_path, 1, base=partial) as store: store.migrate() - assert schema_state(store) == ('49', 49, True) + assert schema_state(store) == ('50', 50, True) with store.pool.connection() as c: assert c.execute("""SELECT to_regclass('deliveries') IS NOT NULL AND to_regclass('download_grants') IS NOT NULL AS present""").fetchone()['present'] diff --git a/tests/test_feedback_status.py b/tests/test_feedback_status.py index d71edc06..094fad64 100644 --- a/tests/test_feedback_status.py +++ b/tests/test_feedback_status.py @@ -89,7 +89,7 @@ def test_feedback_status_migration(schema38_store): store.migrate() store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' assert c.execute('SELECT * FROM archive_migrations WHERE version<=38 ORDER BY version').fetchall() == migrations assert c.execute('SELECT sha256 FROM archive_migrations WHERE version=39').fetchone()['sha256'] == \ hashlib.sha256(SOURCE.joinpath('migrations/039_feedback_status.sql').read_bytes()).hexdigest() @@ -144,7 +144,7 @@ def test_feedback_status_backup_round_trip(seeded, tmp_path): docker = ['docker', 'exec', '-i', '-e', 'PGPASSWORD=' + values['password'], container] create_snapshot(source, tmp_path / 'capture', [*docker, 'pg_dump', '--format=custom', '--dbname=' + container_dsn]) manifest = verify_bundle(tmp_path / 'capture') - assert manifest['schema_version'] == 49 and manifest['tables']['agent_feedback']['rows'] == 3 + assert manifest['schema_version'] == 50 and manifest['tables']['agent_feedback']['rows'] == 3 name = 'anymsg_feedback_' + uuid4().hex + '_restore_test' restored = None created = False diff --git a/tests/test_key_operation.py b/tests/test_key_operation.py index c20e8daa..a552a626 100644 --- a/tests/test_key_operation.py +++ b/tests/test_key_operation.py @@ -201,7 +201,7 @@ def schema_43(store): finally: with store.pool.connection() as c: c.execute(MIGRATION_044.read_text()) - c.execute("UPDATE probe_metadata SET value='49' WHERE name='schema_version'") + c.execute("UPDATE probe_metadata SET value='50' WHERE name='schema_version'") def rows_any(store): @@ -223,7 +223,7 @@ def rows_any(store): # spec: ANYM-495 @pytest.mark.parametrize('command', list(CHECKS)) -@pytest.mark.parametrize('version,answer', [(49, 'ok'), (43, 'schema_not_ready')], ids=['current', 'n-1']) +@pytest.mark.parametrize('version,answer', [(50, 'ok'), (43, 'schema_not_ready')], ids=['current', 'n-1']) def test_check_validates_and_reports_the_schema_and_writes_nothing(seeded, run, tmp_path, command, version, answer): store = seeded[0] with store.pool.connection() as c: diff --git a/tests/test_lifecycle_backup.py b/tests/test_lifecycle_backup.py index 0b8180bb..4007447a 100644 --- a/tests/test_lifecycle_backup.py +++ b/tests/test_lifecycle_backup.py @@ -55,7 +55,7 @@ def test_reference_capture_retains_durable_fence_and_schema_inventory(backup_see create_snapshot(store,tmp_path/'capture',[sys.executable,'-c','print("synthetic dump")'], reference_blobs=True,backup_fence='synthetic-reference-backup') manifest=verify_bundle(tmp_path/'capture',blob_root=store.files) - assert manifest['schema_version']==49 + assert manifest['schema_version']==50 assert manifest['backup_fence']=='synthetic-reference-backup' assert {'attachment_residency','attachment_eviction_journal','attachment_backup_fences', 'attachment_understanding','file_understanding_profiles'}<=set(manifest['tables']) @@ -235,7 +235,7 @@ def connection(self): self.calls+=1 if self.calls==1: with store.pool.connection() as c: - c.execute("UPDATE probe_metadata SET value='49' WHERE name='schema_version'") + c.execute("UPDATE probe_metadata SET value='50' WHERE name='schema_version'") with pytest.raises(RuntimeError,match='backup_capture_unfenced'): create_snapshot(SimpleNamespace(pool=UpgradePool(),files=store.files),tmp_path/'capture', [sys.executable,'-c','print("synthetic dump")']) @@ -292,7 +292,7 @@ def test_schema35_absence_names_the_verified_cold_copy(backup_seeded,tmp_path): try: create_snapshot(store,tmp_path/'capture',[sys.executable,'-c','print("synthetic dump")']) manifest=verify_bundle(tmp_path/'capture') - assert manifest['schema_version']==49 and manifest['tables']['attachment_cold_copies']['rows']==1 + assert manifest['schema_version']==50 and manifest['tables']['attachment_cold_copies']['rows']==1 assert manifest['absent_blobs']==[{**expected,'cold_object_key':key}] assert verify_snapshot(store,tmp_path/'capture')['status']=='verified' # The restore check still requires the managed-absent original to be absent. diff --git a/tests/test_message_purge.py b/tests/test_message_purge.py index 8066371b..0bd74e09 100644 --- a/tests/test_message_purge.py +++ b/tests/test_message_purge.py @@ -732,7 +732,7 @@ def test_migration_refuses_an_unusable_events_message_index(database,tmp_path,in c.execute('DROP INDEX events_message');c.execute('CREATE INDEX events_message ON events(message)') store.migrate() with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' # A small database also gets the visibility probe's index in place (production: release step 2). assert c.execute("SELECT to_regclass('messages_unpurged') IS NOT NULL AS ok").fetchone()['ok'] @@ -781,7 +781,7 @@ def writers(): assert errors==[] and migrate_errors==[] and outcome==['migrated'] assert max(waits)<0.5,waits with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value']=='50' # spec: ANYM-511 diff --git a/tests/test_production.py b/tests/test_production.py index 99833c34..992eb4d6 100644 --- a/tests/test_production.py +++ b/tests/test_production.py @@ -4888,3 +4888,26 @@ def test_a_missing_configuration_file_is_named_in_the_refusal(tmp_path, monkeypa # Plain lines with spaces around = and paired quotes read as systemd reads them. (tmp_path / "autodeploy.env").write_text('; note\nOTHER = "x y"\nDEEDKIT_DEPLOY_CONFIG = "/etc/deedkit/b.json"\n') assert p.active_config() == "/etc/deedkit/b.json" + + +# spec: ANYM-534 +def test_config_promotes_the_matcher_and_dispatcher_overlays(tmp_path, monkeypatch): + settings, _, _ = fixture(tmp_path, monkeypatch) + project = settings["projects"][0] + for name in ("compose.subscriptions.yaml", "compose.delivery.yaml"): + (Path(project["directory"]) / name).write_text("services: {}\n") + project["files"].append(str(Path(project["directory"]) / name)) + project["services"].update({"subscriptions": "runtime", "delivery": "runtime"}) + filename = tmp_path / "production.json" + filename.write_text(json.dumps(settings)) + filename.chmod(0o600) + loaded = p.config(filename)["projects"][0] + command = p.compose(loaded) + files = [command[i + 1] for i, part in enumerate(command) if part == "-f"] + assert files == project["files"] and files[-2:] == [str(tmp_path / "compose.subscriptions.yaml"), + str(tmp_path / "compose.delivery.yaml")] + assert {"subscriptions", "delivery"} <= set(loaded["services"]) + # The documented host file list names both overlays (docs/operations/production.md). + docs = (Path(__file__).resolve().parents[1] / "docs/operations/production.md").read_text() + assert "compose.subscriptions.yaml" in docs and "compose.delivery.yaml" in docs + assert '"subscriptions": "runtime"' in docs and '"delivery": "runtime"' in docs diff --git a/tests/test_recovery.py b/tests/test_recovery.py index 09e7962a..0f4ce098 100644 --- a/tests/test_recovery.py +++ b/tests/test_recovery.py @@ -35,7 +35,7 @@ def recovery_fixture(store,path): def test_full_recovery_roundtrip_quarantines_without_source_paths(seeded,tmp_path): s,*_=seeded;cfg=recovery_fixture(s,tmp_path) create_snapshot(s,tmp_path/'snapshot',DUMP,cfg) - assert verify_snapshot(s,tmp_path/'snapshot')['tables']==78 + assert verify_snapshot(s,tmp_path/'snapshot')['tables']==79 text=(tmp_path/'snapshot/manifest.json').read_text();manifest=json.loads(text) assert manifest['format']==2 and manifest['includes_provider_sessions'] is True assert str(tmp_path) not in text and 'FAKE_SESSION_BYTES' not in text @@ -176,7 +176,7 @@ def test_encrypted_bundle_shape_checks_with_transport_stub(seeded,tmp_path): create_snapshot(s,target,DUMP,cfg) encrypt_snapshot(target,tmp_path/'bundle','fake',age=str(age)) decrypt_snapshot(tmp_path/'bundle','fake',tmp_path/'restored',age=str(age)) - assert verify_snapshot(s,tmp_path/'restored')['tables']==78 + assert verify_snapshot(s,tmp_path/'restored')['tables']==79 # Valid tar framing cannot smuggle a path or quietly omit a recovery blob. for attack in ['path','duplicate','omit']: with tarfile.open(tmp_path/'bundle','r:') as src,tarfile.open(tmp_path/attack,'w:') as out: diff --git a/tests/test_reply_queue.py b/tests/test_reply_queue.py index 80c9ce4d..05bd90f3 100644 --- a/tests/test_reply_queue.py +++ b/tests/test_reply_queue.py @@ -390,7 +390,7 @@ def test_backup_hashes_exact_owner_observations_and_key_local_dismissal(inbox): observe(c,'queue-test','100',{'observed_at':datetime.now(timezone.utc).isoformat(), 'unread_count':0,'read_inbox_max_id':1,'basis':'telegram_dialog'}) manifest=database_manifest(c) - assert len(manifest)==78 + assert len(manifest)==79 assert manifest['reply_dismissals']['rows']==manifest['chat_read_observations']['rows']==1 c.execute('UPDATE chat_read_observations SET unread_count=1 WHERE chat=%s',(chat,)) changed=database_manifest(c) diff --git a/tests/test_reply_queue_controls.py b/tests/test_reply_queue_controls.py index 7f620d0a..bc041f9f 100644 --- a/tests/test_reply_queue_controls.py +++ b/tests/test_reply_queue_controls.py @@ -398,7 +398,7 @@ def test_reply_controls_surface_and_backup(inbox): with s.pool.connection() as c: manifest = database_manifest(c) assert manifest['reply_mutes']['rows'] == 1 and manifest['reply_snoozes']['rows'] == 0 - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' assert s.list_reply_dismissals(key)['items'] == [] and mid @@ -572,7 +572,7 @@ def controls(store): docker = ['docker', 'exec', '-i', '-e', 'PGPASSWORD=' + values['password'], container] create_snapshot(s, tmp_path / 'capture', [*docker, 'pg_dump', '--format=custom', '--dbname=' + container_dsn]) manifest = verify_bundle(tmp_path / 'capture') - assert manifest['schema_version'] == 49 + assert manifest['schema_version'] == 50 assert [manifest['tables'][t]['rows'] for t in ('reply_dismissals', 'reply_snoozes', 'reply_mutes')] == [1, 1, 1] name = 'anymsg_reply_' + uuid4().hex + '_restore_test' restored = None diff --git a/tests/test_reviews.py b/tests/test_reviews.py index fcf3e437..117b7cbb 100644 --- a/tests/test_reviews.py +++ b/tests/test_reviews.py @@ -244,7 +244,7 @@ def connection(self): with gateway.pool.connection() as c:c.execute(sql) with s.pool.connection() as c: manifest=database_manifest(c) - assert len(manifest)==78 and manifest['reviews']['rows']==manifest['review_sources']['rows']==1 + assert len(manifest)==79 and manifest['reviews']['rows']==manifest['review_sources']['rows']==1 finally: with s.pool.connection() as c:c.execute('DROP OWNED BY '+role);c.execute('DROP ROLE '+role) diff --git a/tests/test_subscription_backup_roles.py b/tests/test_subscription_backup_roles.py index 0b037554..7b1fdfef 100644 --- a/tests/test_subscription_backup_roles.py +++ b/tests/test_subscription_backup_roles.py @@ -29,9 +29,9 @@ def test_schema26_manifest_includes_definitions_sources_and_exact_match_referenc _, definition, event = _subscription(store, 'synthetic-manifest-subscriber') match_subscriptions(store) with store.pool.connection() as connection: - assert connection.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert connection.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' manifest = database_manifest(connection) - assert len(manifest) == 78 + assert len(manifest) == 79 assert all(manifest[name]['rows'] == 1 for name in ( 'subscriptions', 'subscription_sources', 'subscription_matches')) assert all(len(manifest[name]['sha256']) == 64 for name in ( @@ -114,7 +114,7 @@ def test_native_schema26_quarantine_closes_versions_preserves_refs_and_revokes_r assert connection.execute('SELECT count(*) AS n FROM subscription_receipts').fetchone()['n'] == 0 assert after['subscription_consumption'] == before['subscription_consumption'] assert after['agent_feedback'] == before['agent_feedback'] - assert len(after) == 78 and after['subscriptions']['rows'] == 2 + assert len(after) == 79 and after['subscriptions']['rows'] == 2 assert all(row['state'] == 'closed' and row['version'] == before_versions[str(row['id'])] + 1 for row in definitions) assert after['subscription_sources'] == before['subscription_sources'] diff --git a/tests/test_subscription_retention.py b/tests/test_subscription_retention.py index 809630d3..93b3c5c3 100644 --- a/tests/test_subscription_retention.py +++ b/tests/test_subscription_retention.py @@ -402,7 +402,7 @@ def test_schema38_database_migrates_with_floor_zero_and_quarantine_sets_the_hold restored.set_subscription_retention(True) restored.migrate() with restored.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' row = c.execute('SELECT retention_floor,retention_hold FROM subscriptions WHERE id=%s', (sid,)).fetchone() assert row == {'retention_floor': 0, 'retention_hold': False} assert database_manifest(c)['subscription_matches'] == legacy['subscription_matches'] diff --git a/tests/test_whatsapp_native_migration.py b/tests/test_whatsapp_native_migration.py index 1efc6fb0..97e0d444 100644 --- a/tests/test_whatsapp_native_migration.py +++ b/tests/test_whatsapp_native_migration.py @@ -80,7 +80,7 @@ def test_schema29_upgrade_preserves_webjs_and_replays_both_native_provenances(sc for value in pending: assert state.apply(value)['outcome'] == 'duplicate' with store.pool.connection() as c: - assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '49' + assert c.execute("SELECT value FROM probe_metadata WHERE name='schema_version'").fetchone()['value'] == '50' assert c.execute('SELECT * FROM archive_migrations WHERE version<=29 ORDER BY version').fetchall() == migrations assert c.execute("SELECT * FROM whatsapp_identity_observations WHERE provenance='webjs_contact_cache_v1'").fetchall() == before assert {r['provenance'] for r in c.execute('SELECT provenance FROM whatsapp_identity_observations')} == PROVENANCES @@ -94,10 +94,10 @@ def test_schema29_upgrade_preserves_webjs_and_replays_both_native_provenances(sc bundle = tmp_path / 'schema30-snapshot' create_snapshot(store, bundle, [sys.executable, '-c', 'print("SYNTHETIC_DUMP")']) manifest = verify_bundle(bundle) - assert manifest['schema_version'] == 49 + assert manifest['schema_version'] == 50 assert manifest['tables']['whatsapp_identity_observations']['rows'] == 3 - assert len(manifest['tables']) == 78 - future = {**manifest, 'schema_version': 50} + assert len(manifest['tables']) == 79 + future = {**manifest, 'schema_version': 51} bundle.joinpath('manifest.json').write_text(json.dumps(future)) with pytest.raises(ValueError, match='unsupported_backup'): verify_bundle(bundle)