diff --git a/CHANGELOG.md b/CHANGELOG.md index 6beb98d9..883bb7e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security -- PostgreSQL control-plane mutations enqueue a transactional outbox row in the same transaction (issue #81). Security events append incrementally instead of rewriting the snapshot. A leased worker claims with `FOR UPDATE SKIP LOCKED`, retries with bounded backoff, dead-letters exhausted/permanent failures, and records unique receipts. Stdout SIEM export is at-least-once; the receipt is the exactly-once ack. `/healthz.outbox` and `GET /api/outbox` are operator-visible; `POST /api/outbox/{id}/replay` requeues dead letters with audit. File/memory adapters report `outbox=disabled`. +- PostgreSQL `security_event` is HASH-partitioned by `tenant_id` (8 children). Unpartitioned tables convert in place and keep unmasked client IPs and paths. `/healthz.event_partitions` reports the child count (0 on file/memory). Logical restore still accepts schema 2 through the current migration version; HASH does not change the snapshot shape. +- PostgreSQL control-plane runtime is `wardnet_runtime` (NOSUPERUSER, NOBYPASSRLS, not the table owner). Migrations run as the login role, then `SET ROLE` so FORCE RLS binds even when the URL user is a superuser. Missing `wardnet.tenant_id` yields no rows. DDL (`DROP TABLE`, `DISABLE ROW LEVEL SECURITY`) is denied. Logical restore accepts schema 2 through the current migration version so a role-only upgrade cannot void the last pre-upgrade backup. +- PostgreSQL control-plane logical backup and isolated restore drill (issue #80 remainder). `GET /api/backup` exports a hashed tenant snapshot (policy, events, outbox, receipts). `POST /api/backup` restores after schema and payload-hash checks. `POST /api/backup/drill` restores into an isolated tenant, compares unmasked invariants, and drops the drill tenant. Declared RPO is the last successful export; declared RTO is 60 seconds. File/memory adapters report `/healthz.backup=disabled`. Client IPs, paths, and actor names stay unmasked. +- PostgreSQL control-plane mutations enqueue a transactional outbox row in the same transaction (issue #81). Security events append incrementally instead of rewriting the snapshot. A leased worker claims with `FOR UPDATE SKIP LOCKED`, retries with bounded backoff, dead-letters exhausted/permanent failures, and records unique receipts. Stdout SIEM export is at-least-once; the receipt is the exactly-once ack. `/healthz.outbox` and `GET /api/outbox` are operator-visible; `POST /api/outbox/{id}/replay` requeues dead letters with audit. File/memory adapters report `outbox=disabled`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (dead letters and pending first). Processed `outbox_message` rows are pruned to that same cap on append, snapshot save, and worker ack; receipts stay as the exactly-once ack. +- Control-plane PostgreSQL URLs honor `sslmode=require` / `verify-ca` / `verify-full` with rustls and Mozilla roots (certificates always verified). tokio-postgres 0.7 only parses `require`, so verification modes are rewritten to `require` before connect. `sslmode=allow` / `prefer` are rejected so the process cannot silently drop to plaintext. - Production (non-loopback) binds fail closed without `CONTROL_PLANE_DATABASE_URL`. PostgreSQL is the production control-plane authority (3NF two-word tables, default-deny row-level security, snapshot persist in one transaction). Loopback still uses the JSON file / memory adapter. `/healthz.persistence` reports `postgres`, `file`, or `memory`. The URL is a secret and is bootstrapped into the credential registry. - Live `/gateway` transactions consult in-process libcoraza when `CORAZA_LIB_PATH` is set (with `CORAZA_RULES_PATH` and/or `CORAZA_DIRECTIVES`). Missing library, missing rules, or an empty ruleset fail startup before bind. Otherwise a Coraza sidecar is consulted when `CORAZA_WAF_URL` is set. The sidecar response is parsed with the existing Coraza audit adapter (OWASP CRS authority, not a hand-rolled engine). Engine outage is fail-closed when `PROVEN_ENGINE_FAIL_CLOSED` is true. `GET /api/waf/engine-status` and `/healthz.proven_engine` report `coraza_in_process`, `coraza_sidecar`, or `ingest_hints_only`. - Fail-closed destination policy on every outbound `http`/`https` call, including the Coraza sidecar URL (gateway upstream, threat-intel fetch, Clearfolio, SOC LLM). Private/loopback/metadata classes are denied unless `DESTINATION_ALLOWLIST` (or loopback development) permits them; `DESTINATION_DENYLIST` wins. Clients ignore ambient HTTP proxies and do not follow redirects. CIDR allowlist matches apply per resolved address; CIDR entries authorize non-default ports; IPv6 site-local (`fec0::/10`) is denied; invalid CIDR prefixes fail startup; `/healthz.destination_mode` reports the policy class. Blocking DNS runs on `spawn_blocking` with a 2s timeout. Persistence and destination-list validation complete before the readiness line is printed. After a host is allowed, the HTTP client connects only to those evaluated addresses (Host/SNI unchanged) so a rebinding answer cannot bypass the policy. diff --git a/Cargo.lock b/Cargo.lock index 3c910412..831e74ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -83,6 +83,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64ct" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" + [[package]] name = "bit-set" version = "0.8.0" @@ -179,6 +185,12 @@ version = "0.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" +[[package]] +name = "const-oid" +version = "0.9.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" + [[package]] name = "const-oid" version = "0.10.2" @@ -231,6 +243,29 @@ dependencies = [ "cmov", ] +[[package]] +name = "der" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" +dependencies = [ + "const-oid 0.9.6", + "der_derive", + "flagset", + "zeroize", +] + +[[package]] +name = "der_derive" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "digest" version = "0.10.7" @@ -248,7 +283,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.1", - "const-oid", + "const-oid 0.10.2", "crypto-common 0.2.2", "ctutils", ] @@ -292,6 +327,12 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "flagset" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" + [[package]] name = "fnv" version = "1.0.7" @@ -1369,6 +1410,16 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "spki" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" +dependencies = [ + "base64ct", + "der", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -1492,6 +1543,27 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" +[[package]] +name = "tls_codec" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de2e01245e2bb89d6f05801c564fa27624dbd7b1846859876c7dad82e90bf6b" +dependencies = [ + "tls_codec_derive", + "zeroize", +] + +[[package]] +name = "tls_codec_derive" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "tokio" version = "1.53.1" @@ -1545,6 +1617,21 @@ dependencies = [ "whoami", ] +[[package]] +name = "tokio-postgres-rustls" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4c2ad44aa0ae96db89c4742212ed41645b2f597311ff6e1945542a4d9fadc2fb" +dependencies = [ + "rustls", + "sha2 0.11.0", + "tokio", + "tokio-postgres", + "tokio-rustls", + "webpki-roots", + "x509-cert", +] + [[package]] name = "tokio-rustls" version = "0.26.4" @@ -1725,11 +1812,13 @@ dependencies = [ "libloading", "proptest", "reqwest", + "rustls", "serde", "serde_json", "sha2 0.10.9", "tokio", "tokio-postgres", + "tokio-postgres-rustls", "tower", "waf-ids-core", ] @@ -2005,6 +2094,18 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "x509-cert" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94" +dependencies = [ + "const-oid 0.9.6", + "der", + "spki", + "tls_codec", +] + [[package]] name = "yoke" version = "0.8.3" @@ -2074,6 +2175,20 @@ name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" +dependencies = [ + "zeroize_derive", +] + +[[package]] +name = "zeroize_derive" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c50655cbb0fe3fc43170059e702f1ce5e19b84cec58dc87b037a09935c2f328" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] [[package]] name = "zerotrie" diff --git a/Cargo.toml b/Cargo.toml index 839f999c..7b9bc242 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,8 @@ serde_json = "1" tokio = { version = "1", features = ["fs", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] } waf-ids-core = { path = "crates/waf-ids-core" } tokio-postgres = { version = "0.7", default-features = false, features = ["runtime"] } +tokio-postgres-rustls = { version = "0.14", default-features = false, features = ["ring", "webpki-roots"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } sha2 = "0.10" [dev-dependencies] diff --git a/README.md b/README.md index 47b44669..35bd2d23 100644 --- a/README.md +++ b/README.md @@ -68,9 +68,9 @@ Useful environment variables: - `ADMIN_TOKEN`: optional write token for management writes via `X-Admin-Token` - `DESTINATION_ALLOWLIST` / `DESTINATION_DENYLIST`: comma-separated hosts, `*.suffix`, or CIDRs for outbound `http`/`https`. Denylist wins. CIDR matches apply per resolved address and also authorize non-default ports. Loopback/private/metadata/site-local destinations are denied unless allowlisted (loopback development still permits loopback-class destinations). After a host is allowed, outbound HTTP connects only to those evaluated addresses (original Host/SNI). `/healthz.destination_mode` reports `production` or `development`. - `WAF_IDS_STATE_PATH`: optional JSON state path for loopback/community. When omitted, the service runs with seeded in-memory state. Production (non-loopback) binds require `CONTROL_PLANE_DATABASE_URL` instead. -- `CONTROL_PLANE_DATABASE_URL`: PostgreSQL URL for the production control plane (`postgres://…`). Secret; prefer `WAF_IDS_CREDENTIALS_PATH` key `control_plane_url`. TLS `sslmode=require` is fail-closed until rustls is wired. `/healthz.persistence` reports `postgres` when connected. +- `CONTROL_PLANE_DATABASE_URL`: PostgreSQL URL for the production control plane (`postgres://…`). Secret; prefer `WAF_IDS_CREDENTIALS_PATH` key `control_plane_url`. `sslmode=require` / `verify-full` uses rustls with Mozilla roots (certificates always verified). `sslmode=disable` or omitted is plaintext. `allow`/`prefer` are rejected. After migrate, the session runs as `wardnet_runtime` (NOSUPERUSER, NOBYPASSRLS). `security_event` is HASH-partitioned by `tenant_id`. `/healthz.persistence` reports `postgres` when connected; `/healthz.event_partitions` reports the child count. - `DNSBL_ORIGIN`: DNSBL zone origin, default `dnsbl.local` -- `EVENT_LIMIT`: retained event count, default `1000`; must be greater than zero +- `EVENT_LIMIT`: retained event count, default `1000`; must be greater than zero. Also caps `GET /api/outbox` and processed outbox-row retention. - `CORAZA_LIB_PATH` / `CORAZA_RULES_PATH` / `CORAZA_DIRECTIVES`: optional in-process libcoraza. A missing library or empty ruleset fails startup. `/healthz.proven_engine` reports `coraza_in_process`. - `CORAZA_WAF_URL`: optional Coraza sidecar URL used when libcoraza is not loaded - `PROVEN_ENGINE_FAIL_CLOSED`: when true, a configured engine outage returns 503 instead of degrading to builtin scoring diff --git a/crates/waf-ids-core/src/lib.rs b/crates/waf-ids-core/src/lib.rs index 291a3900..b22c7d24 100644 --- a/crates/waf-ids-core/src/lib.rs +++ b/crates/waf-ids-core/src/lib.rs @@ -1177,7 +1177,7 @@ fn buyer_evidence_endpoints() -> Vec { "GET", "/healthz", "application/json", - "runtime health, persistence mode, DNSBL origin, and event retention limit", + "runtime health, persistence mode, DNSBL origin, event retention, and HASH event partitions", true, ), buyer_evidence_endpoint( diff --git a/docs/architecture.md b/docs/architecture.md index 0abf4937..6edc3e53 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,8 +29,8 @@ flowchart LR - `src/main.rs`: process startup and operator configuration from `BIND_ADDR`, `ADMIN_TOKEN`, `WAF_IDS_STATE_PATH`, `DNSBL_ORIGIN`, and `EVENT_LIMIT`. - `src/lib.rs`: Axum app, routing, management APIs, optional JSON persistence, gateway handler, upstream proxying, admin console, support bundle assembly, NDJSON event export, and in-crate HTTP tests. Persistence, destination-list, and sidecar settings validate before the readiness line is printed. -- `src/control_plane.rs`: PostgreSQL production authority (issue #80). Non-loopback binds require `CONTROL_PLANE_DATABASE_URL`. Tenant isolation is default-deny RLS. The JSON file adapter remains loopback/community only. -- `src/outbox.rs`: transactional outbox + leased workers (issue #81). Security events append incrementally with an outbox row in the same transaction. Workers claim with `SKIP LOCKED`. `GET /api/outbox` and `/healthz.outbox` are operator-visible. +- `src/control_plane.rs`: PostgreSQL production authority (issue #80). Non-loopback binds require `CONTROL_PLANE_DATABASE_URL`. Tenant isolation is default-deny RLS under `wardnet_runtime` (not superuser/owner). `sslmode=require` uses rustls. `security_event` is HASH-partitioned by `tenant_id` (`/healthz.event_partitions`). The JSON file adapter remains loopback/community only. +- `src/outbox.rs`: transactional outbox + leased workers (issue #81). Security events append incrementally with an outbox row in the same transaction. Workers claim with `SKIP LOCKED`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (processed rows pruned; receipts kept). `/healthz.outbox` is operator-visible. - `src/destination.rs`: fail-closed outbound URL policy (issue #79) for every `http`/`https` send. CIDR allowlist exceptions are per resolved address; blocking DNS is offloaded from Tokio workers. The outbound HTTP client DNS resolver returns only addresses that already passed policy (TCP peer pin / DNS-rebinding TOCTOU close). - `crates/waf-ids-core`: reusable domain models plus validation, upsert, scoring, DNSBL zone export, event retention, threat-feed freshness, KPI snapshot, and commercial readiness logic. - `/admin`: embedded web console. @@ -59,7 +59,7 @@ flowchart LR - `WAF_IDS_STATE_PATH` enables JSON state persistence for standalone/loopback operation. Without it, the service uses seeded in-memory state. Production binds require PostgreSQL (`CONTROL_PLANE_DATABASE_URL`). - File-backed writes use temporary sibling files followed by atomic rename. Management API mutations roll back in memory if the state file cannot be replaced. - Block mode is route-scoped to avoid global accidental enforcement. -- JSON persistence is a baseline durability mechanism, not a substitute for a production database, backup plan, or audited change workflow. +- JSON persistence is a baseline durability mechanism, not a substitute for a production database. PostgreSQL mode exports a hashed logical snapshot (`GET /api/backup`) and runs an isolated restore drill (`POST /api/backup/drill`). - Commercial readiness is a runtime evidence model for buyer pilots, not a legal revenue recognition or compliance certification system. - The reusable core remains in-repo as a workspace crate. A git submodule is intentionally deferred until an independently versioned engine, SDK, or adapter needs a separate release lifecycle. diff --git a/docs/doctoring/outbox-workers.md b/docs/doctoring/outbox-workers.md index a4f69307..cb6337a3 100644 --- a/docs/doctoring/outbox-workers.md +++ b/docs/doctoring/outbox-workers.md @@ -40,11 +40,16 @@ https://doi.org/10.6028/NIST.SP.800-218 Production binds already require `CONTROL_PLANE_DATABASE_URL`. On that path: - `GET /healthz` reports `outbox=ready` plus pending/leased/dead-letter counts -- `GET /api/outbox` (admin read) lists messages +- `GET /api/outbox` (admin read) lists at most `EVENT_LIMIT` messages + (dead letters, then pending, then leased, then processed) - `POST /api/outbox/{message_id}/replay` (admin write) requeues dead letters +Processed `outbox_message` rows are pruned to the operator `EVENT_LIMIT` on +append, snapshot save, and worker ack. `outbox_receipt` rows stay; they are +the exactly-once ack. Dead letters are never pruned. + Loopback file/memory adapters keep in-process stdout SIEM and report -`outbox=disabled`. Remaining: rustls for the control-plane connection, a -non-owner runtime role, backup/restore drill, HASH partitioning, and additional -consumers (TAXII poll, Clearfolio, contextual-orchestrator) on the same -message/receipt contract. +`outbox=disabled`. `security_event` HASH partitioning is on the PostgreSQL +plane. Remaining consumers: TAXII poll, Clearfolio, contextual-orchestrator +on the same message/receipt contract. Backup/restore drill is on the +PostgreSQL plane. diff --git a/docs/doctoring/postgres-control-plane.md b/docs/doctoring/postgres-control-plane.md index 551ac4df..7fff5bbf 100644 --- a/docs/doctoring/postgres-control-plane.md +++ b/docs/doctoring/postgres-control-plane.md @@ -26,18 +26,60 @@ Transaction isolation*. https://www.postgresql.org/docs/current/transaction-iso. commits in one transaction so a policy mutation cannot land without its audit records. +PostgreSQL Global Development Group. (2026). *PostgreSQL documentation: Table +partitioning*. https://www.postgresql.org/docs/current/ddl-partitioning.html + +- **Design impact:** `security_event` is `PARTITION BY HASH (tenant_id)` with + eight children so tenant-scoped SOC queries prune and high-volume appends do + not share one btree. The partition key is part of the primary key. Logical + backups stay a tenant snapshot; HASH is an on-disk layout, not a restore + schema bump that voids prior artifacts. + National Institute of Standards and Technology. (2022). *Secure Software Development Framework (SSDF) version 1.1* (NIST SP 800-218). https://doi.org/10.6028/NIST.SP.800-218 - **Design impact:** PW.1 — fail closed when a production bind has no - control-plane URL, when the URL is not `postgres://`, or when TLS - `sslmode=require` is requested before rustls is wired. + control-plane URL, when the URL is not `postgres://`, or when + `sslmode=allow` / `prefer` could silently drop to plaintext. `require` / + `verify-full` use rustls with Mozilla roots; certificates are always + verified. ## Operator next action Set `CONTROL_PLANE_DATABASE_URL` (or credentials-file key `control_plane_url`) -before binding a non-loopback address. `/healthz.persistence` reports -`postgres`. Loopback still uses `WAF_IDS_STATE_PATH` or in-memory state. -Remaining: rustls, non-owner runtime role, backup/restore drill, HASH -partitioning for `security_event`, optimistic concurrency. +before binding a non-loopback address. Use `sslmode=require` or +`sslmode=verify-full` for rustls. `/healthz.persistence` reports `postgres`. +Loopback still uses `WAF_IDS_STATE_PATH` or in-memory state. + +After migrations, the session `SET ROLE`s to `wardnet_runtime` (NOSUPERUSER, +NOBYPASSRLS, not table owner) so FORCE RLS binds. Provision that role and +`GRANT` it to the login user if the URL user cannot `CREATE ROLE`. Missing +`wardnet.tenant_id` yields no rows. + +`GET /api/backup` (admin read) exports a hashed logical snapshot stamped with +the current `MIGRATION_VERSION`. `POST /api/backup` restores after schema-version +and payload-hash checks. Role-only migrations (v3 `wardnet_runtime`) and HASH +layout (v4) do not change the logical snapshot shape, so `verify()` accepts +schema versions `MIN_RESTORABLE_SCHEMA_VERSION` (2) through the current version +rather than rejecting pre-upgrade snapshots. `POST /api/backup/drill` restores into an +isolated tenant, compares invariants, and drops the drill rows. Declared RPO: +last successful export (`on-demand-logical-snapshot`). Declared RTO: 60 seconds. +`/healthz.backup` is `ready` on PostgreSQL. + +National Institute of Standards and Technology. (2010). *Contingency planning +guide for federal information systems* (NIST SP 800-34 rev. 1). +https://doi.org/10.6028/NIST.SP.800-34r1 +(`docs/papers/nist-sp-800-34r1-contingency-planning.pdf`, public domain) + +- **Design impact:** CP-2 / CP-4 — declared RPO/RTO and an automated restore + drill into an isolated environment. The artifact is application-level (not + `pg_dump`) so RLS tenant context is preserved and secrets (admin tokens, + database URL) are never copied. + +`security_event` is HASH-partitioned by `tenant_id` (8 children) so tenant +queries prune and high-volume appends do not share one btree. Existing +unpartitioned tables convert under `pg_advisory_lock`; rows keep unmasked +client IPs and paths. `/healthz.event_partitions` reports the child count. + +Remaining: optimistic concurrency. diff --git a/docs/papers/nist-sp-800-34r1-contingency-planning.pdf b/docs/papers/nist-sp-800-34r1-contingency-planning.pdf new file mode 100644 index 00000000..38cbc717 Binary files /dev/null and b/docs/papers/nist-sp-800-34r1-contingency-planning.pdf differ diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 2867a9de..677fe74d 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,6 +1,6 @@ # Product and technical gap baseline -Snapshot date: 2026-08-23T17:06Z (exact-head inventory of then-open GitHub PRs +Snapshot date: 2026-08-23T19:00Z (exact-head inventory of then-open GitHub PRs and Issues plus operator-perceptible gaps). Update this file on every hourly loop. Commercial contract and `/api/commercial/readiness` remain **2B KRW**. The @@ -25,8 +25,13 @@ not “waiting on review/CI time”. | PR | Title | Head | Checks | Reviews | Merge blocker | | --- | --- | --- | --- | --- | --- | -| [#99](https://github.com/ContextualWisdomLab/wardnet/pull/99) | feat(store): transactional outbox and leased workers | `feat/issue-81-outbox-workers` stacked on #98 | local fmt/test/clippy + two `/healthz` smokes + postgres `/healthz.outbox=ready` this hour | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 first. Do not `--admin`. Do not re-implement the postgres gate. | -| [#98](https://github.com/ContextualWisdomLab/wardnet/pull/98) | feat(store): require PostgreSQL as the production control plane | `ea621985e276` (`feat/issue-80-postgres-control-plane`) stacked on #97 | rust + fuzz green at last snapshot; Devin 7 threads (full-snapshot rewrite, ORDER BY, TLS, RLS owner, reconnect) | Author this pass; Devin COMMENTED | Org 2-approval + self-author. ORDER BY + incremental event persist addressed on #99. Remaining rustls / non-owner role / backup are #80 remainder. Do not `--admin`. | +| [#104](https://github.com/ContextualWisdomLab/wardnet/pull/104) | feat(store): HASH-partition security_event by tenant | `feat/issue-80-event-hash-partition` stacked on #103 | local fmt/test/clippy + two `scripts/smoke.sh` + live postgres `/healthz.event_partitions=8` | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 then #103 first. Do not `--admin`. Do not re-implement runtime role. | +| [#103](https://github.com/ContextualWisdomLab/wardnet/pull/103) | feat(store): non-owner PostgreSQL runtime role after migrate | `feat/issue-80-runtime-role` stacked on #100 | still-valid Devin restore-window finding fixed this pass (`MIN_RESTORABLE_SCHEMA_VERSION=2`); local fmt/test/clippy + smokes | Author this pass; Devin COMMENTED (v3 backup voiding addressed) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 first. Do not `--admin`. Do not re-implement rustls, outbox, retention, backup, or HASH. | +| [#102](https://github.com/ContextualWisdomLab/wardnet/pull/102) | feat(store): logical backup and isolated restore drill | squash-merged into #100 (`321e792`) | prior hour | Author prior hour | Folded into rustls stack. Do not re-implement. | +| [#101](https://github.com/ContextualWisdomLab/wardnet/pull/101) | feat(store): bound outbox listing and prune processed rows | `feat/issue-81-outbox-retention` (`0c2167a`) stacked on #100 | still-valid Devin prune-cap finding fixed this pass (`EVENT_LIMIT` on save/ack) | Author; Devin COMMENTED (prune thread addressed) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 first. Do not `--admin`. | +| [#100](https://github.com/ContextualWisdomLab/wardnet/pull/100) | feat(store): rustls for production PostgreSQL `sslmode=require` | `feat/issue-80-postgres-rustls` stacked on #99 | local fmt/test/clippy + two `/healthz` smokes; live `sslmode=require` fails closed against plaintext postgres | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 first. Do not `--admin`. Do not re-implement the postgres gate or outbox. | +| [#99](https://github.com/ContextualWisdomLab/wardnet/pull/99) | feat(store): transactional outbox and leased workers | `feat/issue-81-outbox-workers` stacked on #98 | local fmt/test/clippy + two `/healthz` smokes + postgres `/healthz.outbox=ready` prior hour | Author; Devin COMMENTED (unbounded list closed on #101) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 first. Do not `--admin`. | +| [#98](https://github.com/ContextualWisdomLab/wardnet/pull/98) | feat(store): require PostgreSQL as the production control plane | `ea621985e276` (`feat/issue-80-postgres-control-plane`) stacked on #97 | rust + fuzz green at last snapshot; Devin 7 threads (full-snapshot rewrite, ORDER BY, TLS, RLS owner, reconnect) | Author this pass; Devin COMMENTED | Org 2-approval + self-author. ORDER BY + incremental event persist on #99; rustls on #100; backup/restore this pass. Remaining non-owner role. Do not `--admin`. | | [#97](https://github.com/ContextualWisdomLab/wardnet/pull/97) | feat(waf): evaluate live gateway transactions with in-process libcoraza | `feat/issue-86-in-process-libcoraza` stacked on #96 | local fmt/test/clippy + two `/healthz` smokes prior hour | Author this pass | Org 2-approval + self-author. Merge #95 then #96 first. Do not `--admin`. Do not re-implement sidecar or pin. | | [#96](https://github.com/ContextualWisdomLab/wardnet/pull/96) | feat(security): fail-closed destination policy for outbound HTTP | `7cacaf135179` (`feat/issue-79-destination-policy`) stacked on #95 | rust + fuzz green at last snapshot; remaining Devin threads are info/KV-deviation | Author this pass; Devin/Codex COMMENTED | Org 2-approval + self-author. Merge #95 first. Do not re-implement the TCP-peer pin. | | [#95](https://github.com/ContextualWisdomLab/wardnet/pull/95) | feat(waf): consult Coraza sidecar on live gateway transactions | `ba9ee3a0b142` (`feat/issue-86-in-path-coraza`) | rust + Security Scan green at last snapshot | Author this pass; Devin/Codex COMMENTED | Org 2-approval + self-author. Do not re-implement sidecar slice. | @@ -54,8 +59,8 @@ by ruleset `18156473` (not by failing Checks). Do not `--admin` merge. | [#84](https://github.com/ContextualWisdomLab/wardnet/issues/84) | [P1] Build an immutable signed release, promotion, and rollback pipeline | high | | [#83](https://github.com/ContextualWisdomLab/wardnet/issues/83) | [P1] Add bounded distributed admission control, trusted client attribution, and overload behavior | high | | [#82](https://github.com/ContextualWisdomLab/wardnet/issues/82) | [P1] Integrate Keyverse identity, tenant authorization, consent, and human approval evidence | high (blocked) | -| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice this pass** | -| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical — gate on #98; rustls/backup remainder** | +| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice on #99; bounded list/retention on #101** | +| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical — gate on #98; rustls/backup on #100; non-owner role on #103; HASH partitions this pass** | | [#79](https://github.com/ContextualWisdomLab/wardnet/issues/79) | [P0] Enforce a fail-closed destination policy for all outbound traffic | **critical — closed in runtime on #96** | | [#78](https://github.com/ContextualWisdomLab/wardnet/issues/78) | [P0] Fail closed when management credentials are absent | **critical — closed in runtime on #94** | | [#75](https://github.com/ContextualWisdomLab/wardnet/issues/75) | Rename Kubernetes manifest to wardnet.yaml after external-secret hardening lands | medium | @@ -91,17 +96,30 @@ Management auth is shared secrets (`X-Admin-Token`) plus optional multi-token RBAC. Keyverse (OIDC/SCIM/FIDO2) is not wired. Fail-closed (#78) is the prerequisite shipped on PR #94. -### Durable control plane (issue #80) — **production gate on #98** +### Durable control plane (issue #80) — **production gate on #98; rustls on #100; backup on #102; runtime role on #103; HASH this pass** PostgreSQL is required for non-loopback binds (`CONTROL_PLANE_DATABASE_URL`). `src/control_plane.rs` migrates 3NF two-word tables with default-deny RLS (`FORCE ROW LEVEL SECURITY`, `wardnet.tenant_id`). Snapshot persist is one transaction. JSON file / memory remain loopback/community only. -`/healthz.persistence` is `postgres` | `file` | `memory`. Remaining: rustls, -non-owner role, backup/restore drill, event HASH partitioning, optimistic -concurrency. - -### Transactional outbox (issue #81) — **first slice this pass** +`/healthz.persistence` is `postgres` | `file` | `memory`. `sslmode=require` +/ `verify-ca` / `verify-full` use rustls with Mozilla roots (certificates +always verified; stricter than libpq `require`). `allow` / `prefer` are +rejected. `GET /api/backup` exports a hashed logical snapshot; `POST /api/backup` +restores after schema and payload-hash checks; `POST /api/backup/drill` restores +into an isolated tenant, compares unmasked invariants, and drops the drill +tenant. Declared RPO: last successful export. Declared RTO: 60s. +`/healthz.backup` is `ready` on PostgreSQL, `disabled` on file/memory. +Runtime is `wardnet_runtime` (NOSUPERUSER, NOBYPASSRLS) after migrate. +Logical restore accepts schema 2 through the current migration version +(`MIN_RESTORABLE_SCHEMA_VERSION`); role-only and HASH-layout migrations do not +void pre-upgrade snapshots. `security_event` is `PARTITION BY HASH (tenant_id)` +with 8 children. Unpartitioned tables convert in place under `pg_advisory_lock`. +`/healthz.event_partitions` is 8 on PostgreSQL, 0 on file/memory. Client IPs +and paths stay unmasked across convert. Remaining: optimistic concurrency. +Physical/PITR backups stay a DBA concern. + +### Transactional outbox (issue #81) — **first slice on #99; retention on #101** On the PostgreSQL authority, security events append (`security_event` + `outbox_message`) in one transaction instead of rewriting every table. @@ -112,8 +130,10 @@ Stdout SIEM export is **at-least-once**; the receipt is the exactly-once ack. Operator-visible: `/healthz.outbox` (`ready`|`disabled`), pending/leased/ dead-letter counts, `GET /api/outbox` (admin read), `POST /api/outbox/{id}/replay` (admin write + audit). Client IPs and paths in payloads are not masked. -File/memory adapters stay `outbox=disabled` with in-process stdout. Remaining -consumers: TAXII poll, Clearfolio, contextual-orchestrator on the same contract. +File/memory adapters stay `outbox=disabled` with in-process stdout. `GET /api/outbox` +is bounded to `EVENT_LIMIT`; processed rows prune to that cap on append, snapshot +save, and worker ack; dead letters stay. Remaining consumers: TAXII poll, +Clearfolio, contextual-orchestrator on the same contract. ### Fail-closed credentials (issue #78) — **closed on PR #94** @@ -142,7 +162,9 @@ replays `security_event.recorded` as stdout SIEM with receipts. | Figma Code Connect | Not used | | Ten UI-UX areas | Inventoried in `docs/ui-ux/storybook-scene-inventory.md` | | Node Storybook | **Not hosted in `/admin`** (embedded-console architecture). File:// inventory is the scene/edge-case contract this pass. | -| Outbox card | Embedded `/admin` Outbox section this pass | +| Outbox card | Embedded `/admin` Outbox section | +| Backup card | Embedded `/admin` Control-plane backup section | +| Event partitions KPI | Embedded `/admin` KPI tile from `/healthz.event_partitions` this pass | ### CSAP / SOC 2 vs PII unmasking @@ -154,9 +176,9 @@ future encryption-at-rest. ### Coverage / docstring bar Org 100% line/branch/docstring applies to **changed** surfaces this loop -(outbox schema, claim/ack/dead-letter/replay, incremental event persist, -health/API, admin card). Remaining holes on untouched handlers stay listed -for later loops. +(HASH convert SQL, restorable schema window, `/healthz.event_partitions`, +admin KPI tile, probe upgrade preserving unmasked IPs/paths). Remaining +holes on untouched handlers stay listed for later loops. ### Ecosystem connectors (leverage order) @@ -170,24 +192,19 @@ for later loops. ## This loop’s shipped gap -Issue **#81** first slice (transactional outbox + leased workers) on the #80 -PostgreSQL authority. Also the still-valid #98 ORDER BY on load queries and -incremental security-event persist (gateway path no longer rewrites the whole -snapshot). Operator-visible: `/healthz.outbox=ready` on postgres, -`GET /api/outbox`, admin Outbox card. Driving tests: -`postgres_appends_event_and_outbox_atomically`, -`postgres_outbox_worker_is_idempotent_and_dead_letters`, -`postgres_expired_lease_is_reclaimed_and_skip_locked_is_exclusive`, -`outbox_api_is_admin_authenticated_and_disabled_without_postgres`. -Do not re-implement #78, the #86 sidecar/libcoraza slices, the #79 pin, or -the #80 production postgres gate. +Issue **#80** remaining: HASH-partition `security_event` by `tenant_id` (8 +children) stacked on #103. Still-valid #103 Devin finding: `verify()` now +accepts schema 2..=current so a role-only/HASH-layout upgrade cannot void +the last pre-upgrade logical backup. Do not re-implement #78, sidecar, pin, +libcoraza, the postgres gate, outbox, rustls, retention, backup/restore, or +the runtime role. ## Next hourly loop (do, do not report) 1. Second independent APPROVE on #91/#92. Do not `--admin`. -2. Keep #94/#95/#96/#97/#98 and this #81 PR merge-ready. Merge order #95 then - #96 then #97 then #98 then this. Do not re-implement shipped slices. -3. Next runtime gap if policy still blocks: rustls / backup drill remainder of - #80, or additional #81 consumers (TAXII / Clearfolio / orchestrator) on this - outbox. +2. Keep #94/#95/#96/#97/#98/#99/#100/#103 and #104 merge-ready. + Merge order #94 independently; #95 then #96 then #97 then #98 then #99 + then #100 then #103 then #104. +3. Next runtime gap if policy still blocks: extra #81 consumers (TAXII / + Clearfolio / orchestrator) or optimistic concurrency. 4. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. diff --git a/docs/runbooks/operations.md b/docs/runbooks/operations.md index ac081326..a00c44dd 100644 --- a/docs/runbooks/operations.md +++ b/docs/runbooks/operations.md @@ -67,6 +67,33 @@ Expected fields: - `event_limit`: retained security event count - `credentials_source`: `file`, `env`, or `none` - `admin_auth_configured`: whether any admin write token is configured +- `backup`: `ready` on PostgreSQL (logical export/restore available) or `disabled` on file/memory +- `event_partitions`: HASH child count for `security_event` (8 on PostgreSQL, 0 on file/memory) + +## Control-plane backup and restore drill + +PostgreSQL mode (`/healthz.persistence=postgres`) is the only authority that +can export or restore. File/memory adapters report `/healthz.backup=disabled`. + +Declared RPO: last successful `GET /api/backup`. Declared RTO: 60 seconds for +the isolated drill. + +```bash +# Export a hashed tenant snapshot (admin read token). Client IPs and paths stay unmasked. +curl -fsS -H "X-Admin-Token: $ADMIN_TOKEN" http://127.0.0.1:8080/api/backup > backup.json + +# Isolated restore drill (does not replace the live tenant). +curl -fsS -H "X-Admin-Token: $ADMIN_TOKEN" -X POST http://127.0.0.1:8080/api/backup/drill + +# Restore the live tenant from an artifact (admin write). Schema and payload-hash +# mismatches fail closed. The action is audited. +curl -fsS -H "X-Admin-Token: $ADMIN_TOKEN" -H 'content-type: application/json' \ + -d @backup.json -X POST http://127.0.0.1:8080/api/backup +``` + +The artifact does not contain admin tokens or `CONTROL_PLANE_DATABASE_URL`. +Physical/PITR backups remain a DBA concern; this is the application-level +recovery path with tenant RLS preserved. ## Smoke Test @@ -100,7 +127,7 @@ This baseline is suitable for local and controlled lab deployments. Internet-fac - TLS termination and identity-aware admin access - upstream allowlists and egress controls -- durable database storage with backups +- durable database storage with backups (PostgreSQL logical export at `GET /api/backup`; isolated restore drill at `POST /api/backup/drill`; declared RPO is last successful export, declared RTO is 60s) - SSO/OIDC federation (multi-token RBAC with readonly role and audit-log auth are available) - asynchronous event persistence or a database-backed event store for high-throughput gateway traffic - Detection-quality corpora and Suricata EVE tail/shipper remain open. In-process libcoraza (`CORAZA_LIB_PATH` + `CORAZA_RULES_PATH` or `CORAZA_DIRECTIVES`) evaluates each live `/gateway` transaction; otherwise HTTP sidecar consult at `CORAZA_WAF_URL`. Audit ingest at `POST /api/waf/coraza/audit` still fuses block hits into DNSBL/`client_ip` indicators. Set `PROVEN_ENGINE_FAIL_CLOSED=true` in production so an engine outage does not silently allow traffic. diff --git a/docs/security/threat-model.md b/docs/security/threat-model.md index b57968ed..f54b0928 100644 --- a/docs/security/threat-model.md +++ b/docs/security/threat-model.md @@ -24,7 +24,7 @@ | --- | --- | --- | --- | | Unauthorized management write | Route takeover or false blocking | `X-Admin-Token` write gate; multi-token RBAC with actor labels and readonly role; audit log for successful writes | SSO/OIDC, mTLS or identity proxy, SCIM | | Malicious threat feed import | False positives or broad blocks | Validation, route-scoped enforcement | Source signing, feed confidence, staged promotion | -| State file corruption | Startup failure or stale policy | JSON parse failure surfaces startup error; production binds require PostgreSQL (`src/control_plane.rs`) with RLS | Backup/restore drill, TLS, non-owner runtime role | +| State file corruption | Startup failure or stale policy | JSON parse failure surfaces startup error; production binds require PostgreSQL (`src/control_plane.rs`) with RLS; runtime is `wardnet_runtime` (not superuser/owner); `sslmode=require` uses rustls; hashed logical backup plus isolated restore drill (`GET /api/backup`, `POST /api/backup/drill`) | Physical/PITR backups owned by the DBA | | Upstream SSRF through routes | Internal network exposure | Scheme validation plus fail-closed destination policy (`src/destination.rs`): deny loopback/private/link-local/metadata unless allowlisted; denylist wins; no ambient HTTP proxy; no redirects. After evaluation, HTTP connects only to those IPs (Host/SNI preserved). Coraza sidecar URLs use the same policy. | Kubernetes NetworkPolicy egress as defense in depth | | Gateway DoS | Availability loss | Rust memory safety, event retention limit | Rate limits, body limits, async event sink | | DNSBL abuse | Reputation damage | Loopback response-code validation | Authoritative DNS service, signing, publisher workflow | diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 9530cbc8..25b0b59c 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -84,6 +84,7 @@ assert_json_field "$health" 'data["proven_engine_fail_closed"] is False' assert_json_field "$health" 'data["destination_mode"] == "development"' assert_json_field "$health" 'data["outbox"] == "disabled"' assert_json_field "$health" 'data["outbox_pending"] == 0' +assert_json_field "$health" 'data["event_partitions"] == 0' engine_status="$(curl -fsS "$BASE_URL/api/waf/engine-status")" assert_json_field "$engine_status" 'data["mode"] == "ingest_hints_only"' diff --git a/src/control_plane.rs b/src/control_plane.rs index f5df2d90..793ceef6 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -7,13 +7,16 @@ use crate::outbox::{ self, CLAIM_BATCH, DispatchError, EVENT_SECURITY_RECORDED, EVENT_SNAPSHOT_REPLACED, - LEASE_SECONDS, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, STATUS_LEASED, - STATUS_PENDING, STATUS_PROCESSED, + LEASE_SECONDS, LIST_LIMIT, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, + STATUS_LEASED, STATUS_PENDING, STATUS_PROCESSED, }; +use serde::{Deserialize, Serialize}; use std::net::IpAddr; use std::str::FromStr; +use std::time::Instant; use tokio::sync::Mutex; use tokio_postgres::{Client, GenericClient, NoTls, Transaction}; +use tokio_postgres_rustls::MakeRustlsConnect; use waf_ids_core::{ AppData, AuditLogEntry, CommercialProfile, DnsblEntry, EnforcementMode, LicenseStatus, ProductEdition, RouteConfig, SecurityEvent, Severity, ThreatFeedStatus, ThreatIndicator, @@ -22,7 +25,23 @@ use waf_ids_core::{ /// Default tenant used until Keyverse supplies claims (#82). pub const DEFAULT_TENANT_ID: &str = "local-lab"; -const MIGRATION_VERSION: i32 = 2; +const MIGRATION_VERSION: i32 = 4; +/// Oldest logical-backup schema that restores on this binary. +/// +/// v3 only provisions `wardnet_runtime`. v4 HASH-partitions `security_event` +/// without changing the logical snapshot shape. A snapshot exported at schema +/// 2 is restorable here so those upgrades cannot void the declared RPO. +const MIN_RESTORABLE_SCHEMA_VERSION: i32 = 2; +/// HASH partitions for `security_event` (by `tenant_id`). +pub const EVENT_PARTITION_MODULUS: i32 = 8; +/// Non-owner, non-superuser role used after migrations so FORCE RLS binds. +pub const RUNTIME_ROLE: &str = "wardnet_runtime"; +/// Declared RPO: last successful `GET /api/backup` (on-demand logical snapshot). +pub const BACKUP_RPO: &str = "on-demand-logical-snapshot"; +/// Declared RTO budget for an isolated restore drill. +pub const BACKUP_RTO_BUDGET_MS: u64 = 60_000; +/// Session advisory lock for schema/partition DDL (cross-process). +const MIGRATION_LOCK_KEY: i64 = 80_201_680; /// Recoverable forward migration. Two-word snake_case names, 3NF, RLS. pub const MIGRATION_SQL: &str = r#" @@ -226,8 +245,139 @@ DROP POLICY IF EXISTS tenant_isolation ON outbox_receipt; CREATE POLICY tenant_isolation ON outbox_receipt USING (tenant_id = current_setting('wardnet.tenant_id', true)) WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'wardnet_runtime') THEN + CREATE ROLE wardnet_runtime + NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOLOGIN + NOBYPASSRLS NOREPLICATION; + END IF; +END +$$; +GRANT wardnet_runtime TO CURRENT_USER; +GRANT USAGE ON SCHEMA public TO wardnet_runtime; +REVOKE CREATE ON SCHEMA public FROM wardnet_runtime; +GRANT SELECT, INSERT, UPDATE, DELETE ON + tenant_account, tenant_profile, route_config, threat_indicator, + dnsbl_entry, security_event, audit_record, threat_feed, + outbox_message, outbox_receipt TO wardnet_runtime; +GRANT SELECT ON schema_migration TO wardnet_runtime; "#; +fn hash_partition_sql_for(parent: &str) -> Result { + if parent.is_empty() + || !parent.as_bytes()[0].is_ascii_lowercase() + || !parent + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') + { + return Err("hash partition parent must be a lowercase SQL identifier".to_string()); + } + let modulus = EVENT_PARTITION_MODULUS; + let unpartitioned = format!("{parent}_unpartitioned"); + Ok(format!( + r#" +DO $hash$ +DECLARE + kind "char"; + i integer; +BEGIN + SELECT c.relkind INTO kind + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = current_schema() + AND c.relname = '{parent}'; + + IF kind = 'r' THEN + ALTER TABLE {parent} RENAME TO {unpartitioned}; + DROP INDEX IF EXISTS {parent}_tenant_event; + CREATE TABLE {parent} ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + event_id BIGINT NOT NULL, + timestamp_unix BIGINT NOT NULL, + client_address TEXT, + route_id TEXT, + action_name TEXT NOT NULL, + event_reason TEXT NOT NULL, + event_score INTEGER NOT NULL, + request_path TEXT NOT NULL, + PRIMARY KEY (tenant_id, event_id) + ) PARTITION BY HASH (tenant_id); + i := 0; + WHILE i < {modulus} LOOP + EXECUTE format( + 'CREATE TABLE {parent}_p%s PARTITION OF {parent} FOR VALUES WITH (MODULUS {modulus}, REMAINDER %s)', + i, i + ); + i := i + 1; + END LOOP; + INSERT INTO {parent} ( + tenant_id, event_id, timestamp_unix, client_address, route_id, + action_name, event_reason, event_score, request_path + ) + SELECT tenant_id, event_id, timestamp_unix, client_address, route_id, + action_name, event_reason, event_score, request_path + FROM {unpartitioned}; + DROP TABLE {unpartitioned}; + ALTER TABLE {parent} ENABLE ROW LEVEL SECURITY; + ALTER TABLE {parent} FORCE ROW LEVEL SECURITY; + DROP POLICY IF EXISTS tenant_isolation ON {parent}; + CREATE POLICY tenant_isolation ON {parent} + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'wardnet_runtime') THEN + GRANT SELECT, INSERT, UPDATE, DELETE ON {parent} TO wardnet_runtime; + i := 0; + WHILE i < {modulus} LOOP + EXECUTE format( + 'GRANT SELECT, INSERT, UPDATE, DELETE ON {parent}_p%s TO wardnet_runtime', + i + ); + i := i + 1; + END LOOP; + END IF; + ELSIF kind = 'p' THEN + i := 0; + WHILE i < {modulus} LOOP + IF NOT EXISTS ( + SELECT 1 FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = current_schema() + AND c.relname = format('{parent}_p%s', i) + ) THEN + EXECUTE format( + 'CREATE TABLE {parent}_p%s PARTITION OF {parent} FOR VALUES WITH (MODULUS {modulus}, REMAINDER %s)', + i, i + ); + END IF; + i := i + 1; + END LOOP; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'wardnet_runtime') THEN + GRANT SELECT, INSERT, UPDATE, DELETE ON {parent} TO wardnet_runtime; + i := 0; + WHILE i < {modulus} LOOP + IF EXISTS ( + SELECT 1 FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = current_schema() + AND c.relname = format('{parent}_p%s', i) + ) THEN + EXECUTE format( + 'GRANT SELECT, INSERT, UPDATE, DELETE ON {parent}_p%s TO wardnet_runtime', + i + ); + END IF; + i := i + 1; + END LOOP; + END IF; + END IF; +END +$hash$; +"#, + )) +} + /// Fail closed when a non-loopback bind has no control-plane URL. pub fn require_postgres_for_bind( bind_addr: &str, @@ -245,8 +395,7 @@ pub fn require_postgres_for_bind( } } -/// Structural URL checks. TLS (`sslmode=require`) is fail-closed until rustls -/// is wired. Password stays in the registry, not logs. +/// Structural URL checks. Password stays in the registry, not logs. pub fn parse_database_url(raw: &str) -> Result { let raw = raw.trim(); if raw.is_empty() { @@ -256,14 +405,176 @@ pub fn parse_database_url(raw: &str) -> Result { if !(lower.starts_with("postgres://") || lower.starts_with("postgresql://")) { return Err("CONTROL_PLANE_DATABASE_URL must be a postgres:// URL".to_string()); } - if lower.contains("sslmode=require") || lower.contains("sslmode=verify") { - return Err( - "CONTROL_PLANE_DATABASE_URL TLS (sslmode=require/verify) is not wired yet".to_string(), - ); - } + ssl_mode(raw)?; Ok(raw.to_string()) } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SslMode { + Disable, + Require, +} + +/// `disable` (or omitted) uses plaintext. `require` / `verify-ca` / +/// `verify-full` use rustls with Mozilla roots (certificates are always +/// verified — stricter than libpq `require`). `allow` / `prefer` are rejected +/// because they can silently drop to plaintext. +fn ssl_mode(raw: &str) -> Result { + let lower = raw.to_ascii_lowercase(); + let Some((_, query)) = lower.split_once('?') else { + return Ok(SslMode::Disable); + }; + for part in query.split('&').flat_map(|chunk| chunk.split('#')) { + let Some((key, value)) = part.split_once('=') else { + continue; + }; + if key != "sslmode" { + continue; + } + return match value { + "disable" => Ok(SslMode::Disable), + "require" | "verify-ca" | "verify-full" => Ok(SslMode::Require), + other => Err(format!( + "unsupported sslmode {other}; use disable or require/verify-full" + )), + }; + } + Ok(SslMode::Disable) +} + +/// tokio-postgres 0.7 only parses `disable` / `prefer` / `require`. Map the +/// libpq verification modes we already treat as `Require` so rustls can +/// still verify certificates. +fn rewrite_sslmode_for_tokio(raw: &str) -> String { + let Some((head, query)) = raw.split_once('?') else { + return raw.to_string(); + }; + let rewritten = query + .split('&') + .map(|part| { + let Some((key, value)) = part.split_once('=') else { + return part.to_string(); + }; + if key.eq_ignore_ascii_case("sslmode") + && (value.eq_ignore_ascii_case("verify-ca") + || value.eq_ignore_ascii_case("verify-full")) + { + format!("{key}=require") + } else { + part.to_string() + } + }) + .collect::>() + .join("&"); + format!("{head}?{rewritten}") +} + +fn rustls_connector() -> Result { + rustls::crypto::ring::default_provider() + .install_default() + .ok(); + Ok(MakeRustlsConnect::with_webpki_roots()) +} + +async fn assume_runtime_role(client: &Client) -> Result<(), String> { + let current: String = client + .query_one("SELECT current_user", &[]) + .await + .map_err(|error| format!("control plane current_user failed: {error}"))? + .get(0); + if current != RUNTIME_ROLE { + client + .batch_execute( + "DO $$ + BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'wardnet_runtime') THEN + CREATE ROLE wardnet_runtime + NOSUPERUSER NOCREATEDB NOCREATEROLE NOINHERIT NOLOGIN + NOBYPASSRLS NOREPLICATION; + END IF; + END + $$; + GRANT wardnet_runtime TO CURRENT_USER; + GRANT USAGE ON SCHEMA public TO wardnet_runtime; + REVOKE CREATE ON SCHEMA public FROM wardnet_runtime; + GRANT SELECT, INSERT, UPDATE, DELETE ON + tenant_account, tenant_profile, route_config, threat_indicator, + dnsbl_entry, security_event, audit_record, threat_feed, + outbox_message, outbox_receipt TO wardnet_runtime; + GRANT SELECT ON schema_migration TO wardnet_runtime; + SET ROLE wardnet_runtime;", + ) + .await + .map_err(|error| { + format!( + "control plane runtime role {RUNTIME_ROLE} failed (provision NOSUPERUSER NOBYPASSRLS and GRANT it to the login role): {error}" + ) + })?; + } + let row = client + .query_one("SELECT current_user, current_setting('is_superuser')", &[]) + .await + .map_err(|error| format!("control plane runtime identity failed: {error}"))?; + let user: String = row.get(0); + let superuser: String = row.get(1); + if user != RUNTIME_ROLE { + return Err(format!( + "control plane must run as {RUNTIME_ROLE}, current_user is {user}" + )); + } + if superuser == "on" { + return Err(format!( + "control plane role {RUNTIME_ROLE} must not be a superuser" + )); + } + Ok(()) +} + +async fn apply_schema(client: &Client) -> Result<(), String> { + let applied = match client + .query_one( + "SELECT COALESCE(MAX(migration_version), 0) FROM schema_migration", + &[], + ) + .await + { + Ok(row) => row.get::<_, i32>(0), + Err(_) => 0, + }; + if applied < MIGRATION_VERSION { + client + .batch_execute(MIGRATION_SQL) + .await + .map_err(|error| format!("control plane migration failed: {error:?}"))?; + client + .execute( + "INSERT INTO schema_migration (migration_version) VALUES ($1) ON CONFLICT (migration_version) DO NOTHING", + &[&MIGRATION_VERSION], + ) + .await + .map_err(|error| format!("control plane migration version failed: {error}"))?; + } + let hash_sql = hash_partition_sql_for("security_event")?; + client + .batch_execute(&hash_sql) + .await + .map_err(|error| format!("control plane event hash partition failed: {error:?}"))?; + Ok(()) +} + +async fn event_partition_count(client: &Client) -> Result { + let row = client + .query_one( + "SELECT COUNT(*)::bigint + FROM pg_partition_tree('security_event'::regclass) + WHERE level = 1", + &[], + ) + .await + .map_err(|error| format!("control plane event partition count failed: {error}"))?; + Ok(row.get(0)) +} + /// Serializes schema application across connections (DROP/CREATE POLICY is not concurrent-safe). static MIGRATION_GATE: std::sync::LazyLock> = std::sync::LazyLock::new(|| Mutex::new(())); @@ -271,6 +582,8 @@ static MIGRATION_GATE: std::sync::LazyLock> = std::sync::LazyLock::new pub struct PostgresPlane { client: Mutex, tenant_id: String, + /// Processed-outbox retention; mirrors operator `EVENT_LIMIT`. + event_limit: i64, } impl PostgresPlane { @@ -280,48 +593,139 @@ impl PostgresPlane { pub async fn connect_tenant(url: &str, tenant_id: &str) -> Result { let url = parse_database_url(url)?; - let (client, connection) = tokio_postgres::connect(&url, NoTls) - .await - .map_err(|error| format!("control plane connect failed: {error}"))?; - tokio::spawn(async move { - let _ = connection.await; - }); + let mode = ssl_mode(&url)?; + let connect_url = rewrite_sslmode_for_tokio(&url); + let (client, connection) = match mode { + SslMode::Disable => { + let (client, connection) = tokio_postgres::connect(&connect_url, NoTls) + .await + .map_err(|error| format!("control plane connect failed: {error}"))?; + ( + client, + tokio::spawn(async move { + let _ = connection.await; + }), + ) + } + SslMode::Require => { + let tls = rustls_connector()?; + let (client, connection) = tokio_postgres::connect(&connect_url, tls) + .await + .map_err(|error| format!("control plane TLS connect failed: {error}"))?; + ( + client, + tokio::spawn(async move { + let _ = connection.await; + }), + ) + } + }; + std::mem::drop(connection); let plane = Self { client: Mutex::new(client), tenant_id: tenant_id.to_string(), + event_limit: LIST_LIMIT, }; plane.migrate().await?; + plane.assume_runtime_role().await?; Ok(plane) } + /// Use the operator-configured `EVENT_LIMIT` for processed-outbox retention. + pub fn with_event_limit(mut self, event_limit: usize) -> Self { + self.event_limit = event_limit.max(1) as i64; + self + } + async fn migrate(&self) -> Result<(), String> { let _gate = MIGRATION_GATE.lock().await; let client = self.client.lock().await; - let applied = match client + client + .execute("SELECT pg_advisory_lock($1)", &[&MIGRATION_LOCK_KEY]) + .await + .map_err(|error| format!("control plane migration lock failed: {error}"))?; + let result = apply_schema(&client).await; + let _ = client + .execute("SELECT pg_advisory_unlock($1)", &[&MIGRATION_LOCK_KEY]) + .await; + result + } + + async fn assume_runtime_role(&self) -> Result<(), String> { + let client = self.client.lock().await; + assume_runtime_role(&client).await + } + + #[cfg(test)] + async fn runtime_identity(&self) -> Result<(String, bool), String> { + let client = self.client.lock().await; + let row = client .query_one( - "SELECT COALESCE(MAX(migration_version), 0) FROM schema_migration", + "SELECT current_user, current_setting('is_superuser') = 'on'", &[], ) .await - { - Ok(row) => row.get::<_, i32>(0), - Err(_) => 0, - }; - if applied >= MIGRATION_VERSION { - return Ok(()); - } - client - .batch_execute(MIGRATION_SQL) + .map_err(|error| error.to_string())?; + Ok((row.get(0), row.get(1))) + } + + #[cfg(test)] + async fn unscoped_route_count(&self) -> Result { + let mut client = self.client.lock().await; + let tx = client + .transaction() .await - .map_err(|error| format!("control plane migration failed: {error:?}"))?; - client - .execute( - "INSERT INTO schema_migration (migration_version) VALUES ($1) ON CONFLICT (migration_version) DO NOTHING", - &[&MIGRATION_VERSION], + .map_err(|error| error.to_string())?; + let count: i64 = tx + .query_one("SELECT COUNT(*)::bigint FROM route_config", &[]) + .await + .map_err(|error| error.to_string())? + .get(0); + tx.rollback().await.map_err(|error| error.to_string())?; + Ok(count) + } + + pub async fn event_partition_count(&self) -> Result { + let client = self.client.lock().await; + event_partition_count(&client).await + } + + #[cfg(test)] + async fn security_event_tableoid(&self) -> Result { + let mut client = self.client.lock().await; + let tx = client + .transaction() + .await + .map_err(|error| error.to_string())?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&self.tenant_id], + ) + .await + .map_err(|error| error.to_string())?; + let row = tx + .query_one( + "SELECT tableoid::regclass::text FROM security_event WHERE tenant_id = $1 LIMIT 1", + &[&self.tenant_id], ) .await - .map_err(|error| format!("control plane migration version failed: {error}"))?; - Ok(()) + .map_err(|error| error.to_string())?; + tx.commit().await.map_err(|error| error.to_string())?; + Ok(row.get(0)) + } + + #[cfg(test)] + async fn runtime_ddl_is_denied(&self) -> Result { + let client = self.client.lock().await; + let drop_denied = client + .batch_execute("DROP TABLE route_config") + .await + .is_err(); + let disable_denied = client + .batch_execute("ALTER TABLE route_config DISABLE ROW LEVEL SECURITY") + .await + .is_err(); + Ok(drop_denied && disable_denied) } /// Load the tenant snapshot, or `None` when the tenant has no rows yet. @@ -333,7 +737,7 @@ impl PostgresPlane { /// Replace the tenant snapshot in one transaction (mutation + audit + outbox). pub async fn save(&self, data: &AppData) -> Result<(), String> { let mut client = self.client.lock().await; - save_snapshot(&mut client, &self.tenant_id, data).await + save_snapshot(&mut client, &self.tenant_id, data, self.event_limit).await } /// Append one security event and its outbox row without rewriting the snapshot. @@ -356,7 +760,15 @@ impl PostgresPlane { F: Fn(&OutboxMessage) -> Result, { let mut client = self.client.lock().await; - drain_once(&mut client, &self.tenant_id, owner, now_unix, dispatch).await + drain_once( + &mut client, + &self.tenant_id, + owner, + now_unix, + self.event_limit, + dispatch, + ) + .await } pub async fn outbox_health(&self, now_unix: i64) -> Result { @@ -364,15 +776,164 @@ impl PostgresPlane { outbox_health(&mut client, &self.tenant_id, now_unix).await } - pub async fn list_outbox(&self) -> Result, String> { + pub async fn list_outbox_limited(&self, limit: i64) -> Result, String> { let mut client = self.client.lock().await; - list_outbox(&mut client, &self.tenant_id).await + list_outbox(&mut client, &self.tenant_id, limit.max(1)).await } pub async fn replay_dead_letter(&self, message_id: &str, now_unix: i64) -> Result<(), String> { let mut client = self.client.lock().await; replay_dead_letter(&mut client, &self.tenant_id, message_id, now_unix).await } + + /// Tenant-scoped logical backup. Client IPs, paths, and actor names stay unmasked. + pub async fn logical_backup(&self) -> Result { + let mut client = self.client.lock().await; + export_backup(&mut client, &self.tenant_id).await + } + + /// Restore a verified artifact into this tenant. Fail closed on schema or hash mismatch. + pub async fn restore_logical_backup(&self, backup: &ControlPlaneBackup) -> Result<(), String> { + backup.verify()?; + let mut client = self.client.lock().await; + restore_backup(&mut client, &self.tenant_id, backup).await + } + + /// Restore into an isolated tenant, compare invariants, then drop the drill tenant. + pub async fn restore_drill(&self) -> Result { + let started = Instant::now(); + let backup = self.logical_backup().await?; + let isolated = format!("restore-drill-{}-{}", std::process::id(), unix_now_i64()); + let mut client = self.client.lock().await; + restore_backup(&mut client, &isolated, &backup).await?; + let restored = export_backup(&mut client, &isolated).await?; + drop_tenant(&mut client, &isolated).await?; + let source_hash = backup.semantic_hash()?; + let restored_hash = restored.semantic_hash()?; + let passed = source_hash == restored_hash + && restored.snapshot.routes == backup.snapshot.routes + && restored.snapshot.events == backup.snapshot.events + && restored.snapshot.threats == backup.snapshot.threats + && restored.snapshot.dnsbl == backup.snapshot.dnsbl + && restored.outbox.len() == backup.outbox.len() + && restored.receipts.len() == backup.receipts.len(); + let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX); + Ok(BackupDrillReport { + passed, + duration_ms, + rpo: BACKUP_RPO.to_string(), + rto_budget_ms: BACKUP_RTO_BUDGET_MS, + source_hash, + restored_hash, + route_count: backup.snapshot.routes.len(), + event_count: backup.snapshot.events.len(), + outbox_count: backup.outbox.len(), + receipt_count: backup.receipts.len(), + isolated_tenant_id: isolated, + }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct OutboxReceiptRow { + pub tenant_id: String, + pub idempotency_key: String, + pub message_id: String, + pub processed_unix: i64, + pub receipt_evidence: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ControlPlaneBackup { + pub schema_version: i32, + pub tenant_id: String, + pub created_unix: i64, + pub snapshot: AppData, + pub outbox: Vec, + pub receipts: Vec, + pub payload_hash: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct BackupDrillReport { + pub passed: bool, + pub duration_ms: u64, + pub rpo: String, + pub rto_budget_ms: u64, + pub source_hash: String, + pub restored_hash: String, + pub route_count: usize, + pub event_count: usize, + pub outbox_count: usize, + pub receipt_count: usize, + pub isolated_tenant_id: String, +} + +impl ControlPlaneBackup { + fn unsigned_json(&self) -> Result { + let mut unsigned = self.clone(); + unsigned.payload_hash.clear(); + serde_json::to_string(&unsigned) + .map_err(|error| format!("backup serialize failed: {error}")) + } + + fn seal(mut self) -> Result { + self.payload_hash.clear(); + let json = self.unsigned_json()?; + self.payload_hash = outbox::payload_hash(&json); + Ok(self) + } + + /// Fail closed when the schema is unsupported or the artifact was tampered with. + pub fn verify(&self) -> Result<(), String> { + if self.schema_version < MIN_RESTORABLE_SCHEMA_VERSION + || self.schema_version > MIGRATION_VERSION + { + return Err(format!( + "backup schema_version {} is unsupported; accepted {MIN_RESTORABLE_SCHEMA_VERSION}..={MIGRATION_VERSION}", + self.schema_version + )); + } + if self.tenant_id.trim().is_empty() { + return Err("backup tenant_id must be non-empty".to_string()); + } + let expected = outbox::payload_hash(&self.unsigned_json()?); + if expected != self.payload_hash { + return Err("backup payload_hash does not match contents".to_string()); + } + Ok(()) + } + + fn semantic_hash(&self) -> Result { + let mut snapshot = self.snapshot.clone(); + snapshot.commercial.tenant_id.clear(); + let mut outbox: Vec<_> = self + .outbox + .iter() + .map(|message| { + ( + message.idempotency_key.clone(), + message.payload_hash.clone(), + message.message_status.clone(), + message.payload_json.clone(), + ) + }) + .collect(); + outbox.sort(); + let mut receipts: Vec<_> = self + .receipts + .iter() + .map(|row| (row.idempotency_key.clone(), row.receipt_evidence.clone())) + .collect(); + receipts.sort(); + let body = serde_json::json!({ + "snapshot": snapshot, + "outbox": outbox, + "receipts": receipts, + }) + .to_string(); + Ok(outbox::payload_hash(&body)) + } } async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result, String> { @@ -424,7 +985,12 @@ async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result Result<(), String> { +async fn save_snapshot( + client: &mut Client, + tenant_id: &str, + data: &AppData, + keep: i64, +) -> Result<(), String> { let tx = client .transaction() .await @@ -436,6 +1002,21 @@ async fn save_snapshot(client: &mut Client, tenant_id: &str, data: &AppData) -> .await .map_err(|error| format!("control plane tenant context failed: {error}"))?; + write_snapshot_rows(&tx, tenant_id, data).await?; + enqueue_snapshot_outbox(&tx, tenant_id, data).await?; + prune_processed_outbox(&tx, tenant_id, keep).await?; + + tx.commit() + .await + .map_err(|error| format!("control plane commit failed: {error}"))?; + Ok(()) +} + +async fn write_snapshot_rows( + tx: &Transaction<'_>, + tenant_id: &str, + data: &AppData, +) -> Result<(), String> { tx.execute( "INSERT INTO tenant_account (tenant_id, event_sequence, audit_sequence) VALUES ($1, $2, $3) @@ -616,12 +1197,6 @@ async fn save_snapshot(client: &mut Client, tenant_id: &str, data: &AppData) -> .await .map_err(|error| format!("control plane insert threat_feed failed: {error}"))?; } - - enqueue_snapshot_outbox(&tx, tenant_id, data).await?; - - tx.commit() - .await - .map_err(|error| format!("control plane commit failed: {error}"))?; Ok(()) } @@ -938,6 +1513,7 @@ async fn append_security_event( }, ) .await?; + prune_processed_outbox(&tx, tenant_id, event_limit as i64).await?; tx.commit() .await @@ -992,6 +1568,7 @@ async fn drain_once( tenant_id: &str, owner: &str, now_unix: i64, + keep: i64, dispatch: F, ) -> Result where @@ -1001,13 +1578,21 @@ where let mut processed = 0; for message in claimed { if receipt_exists(client, tenant_id, &message.idempotency_key).await? { - ack_processed(client, tenant_id, &message, "duplicate-receipt", now_unix).await?; + ack_processed( + client, + tenant_id, + &message, + "duplicate-receipt", + now_unix, + keep, + ) + .await?; processed += 1; continue; } match dispatch(&message) { Ok(evidence) => { - ack_processed(client, tenant_id, &message, &evidence, now_unix).await?; + ack_processed(client, tenant_id, &message, &evidence, now_unix, keep).await?; processed += 1; } Err(error) => { @@ -1143,6 +1728,7 @@ async fn ack_processed( message: &OutboxMessage, evidence: &str, now_unix: i64, + keep: i64, ) -> Result<(), String> { let tx = client .transaction() @@ -1183,6 +1769,7 @@ async fn ack_processed( ) .await .map_err(|error| format!("control plane ack outbox_message failed: {error}"))?; + prune_processed_outbox(&tx, tenant_id, keep).await?; tx.commit() .await .map_err(|error| format!("control plane ack commit failed: {error}"))?; @@ -1280,7 +1867,37 @@ async fn outbox_health( }) } -async fn list_outbox(client: &mut Client, tenant_id: &str) -> Result, String> { +async fn prune_processed_outbox( + client: &C, + tenant_id: &str, + keep: i64, +) -> Result<(), String> { + let keep = keep.max(1); + client + .execute( + "DELETE FROM outbox_message + WHERE tenant_id = $1 + AND message_status = $2 + AND message_id IN ( + SELECT message_id FROM ( + SELECT message_id FROM outbox_message + WHERE tenant_id = $1 AND message_status = $2 + ORDER BY created_unix DESC, message_id DESC + OFFSET $3 + ) old_processed + )", + &[&tenant_id, &STATUS_PROCESSED, &keep], + ) + .await + .map_err(|error| format!("control plane prune outbox_message failed: {error}"))?; + Ok(()) +} + +async fn list_outbox( + client: &mut Client, + tenant_id: &str, + limit: i64, +) -> Result, String> { let tx = client .transaction() .await @@ -1298,8 +1915,15 @@ async fn list_outbox(client: &mut Client, tenant_id: &str) -> Result i64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i64 -} - -fn mode_sql(mode: &EnforcementMode) -> &'static str { - match mode { - EnforcementMode::Monitor => "monitor", - EnforcementMode::Block => "block", - } -} - -fn parse_mode(value: &str) -> Result { - match value { - "monitor" => Ok(EnforcementMode::Monitor), - "block" => Ok(EnforcementMode::Block), - other => Err(format!("unknown enforcement_mode {other}")), - } -} - -fn severity_sql(severity: &Severity) -> &'static str { - match severity { - Severity::Low => "low", - Severity::Medium => "medium", - Severity::High => "high", - Severity::Critical => "critical", +async fn export_backup(client: &mut Client, tenant_id: &str) -> Result { + let snapshot = load_snapshot(client, tenant_id) + .await? + .ok_or_else(|| format!("tenant {tenant_id} has no snapshot to back up"))?; + let outbox = list_outbox(client, tenant_id, i64::MAX).await?; + let receipts = list_receipts(client, tenant_id).await?; + ControlPlaneBackup { + schema_version: MIGRATION_VERSION, + tenant_id: tenant_id.to_string(), + created_unix: unix_now_i64(), + snapshot, + outbox, + receipts, + payload_hash: String::new(), } + .seal() } -fn parse_severity(value: &str) -> Result { - match value { - "low" => Ok(Severity::Low), - "medium" => Ok(Severity::Medium), - "high" => Ok(Severity::High), - "critical" => Ok(Severity::Critical), - other => Err(format!("unknown severity_name {other}")), - } +async fn list_receipts( + client: &mut Client, + tenant_id: &str, +) -> Result, String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane receipt list transaction failed: {error}"))?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane tenant context failed: {error}"))?; + let rows = tx + .query( + "SELECT idempotency_key, message_id, processed_unix, receipt_evidence + FROM outbox_receipt WHERE tenant_id = $1 + ORDER BY processed_unix, idempotency_key", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane list outbox_receipt failed: {error}"))?; + let receipts = rows + .iter() + .map(|row| OutboxReceiptRow { + tenant_id: tenant_id.to_string(), + idempotency_key: row.get(0), + message_id: row.get(1), + processed_unix: row.get(2), + receipt_evidence: row.get(3), + }) + .collect(); + tx.commit() + .await + .map_err(|error| format!("control plane receipt list commit failed: {error}"))?; + Ok(receipts) } -fn edition_sql(edition: &ProductEdition) -> &'static str { - match edition { - ProductEdition::Community => "community", - ProductEdition::Evaluation => "evaluation", +async fn restore_backup( + client: &mut Client, + tenant_id: &str, + backup: &ControlPlaneBackup, +) -> Result<(), String> { + backup.verify()?; + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane restore transaction failed: {error}"))?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane tenant context failed: {error}"))?; + write_snapshot_rows(&tx, tenant_id, &backup.snapshot).await?; + for table in ["outbox_receipt", "outbox_message"] { + tx.execute( + &format!("DELETE FROM {table} WHERE tenant_id = $1"), + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane delete {table} failed: {error}"))?; + } + for message in &backup.outbox { + insert_restored_outbox(&tx, tenant_id, message).await?; + } + for receipt in &backup.receipts { + tx.execute( + "INSERT INTO outbox_receipt ( + tenant_id, idempotency_key, message_id, processed_unix, receipt_evidence + ) VALUES ($1,$2,$3,$4,$5)", + &[ + &tenant_id, + &receipt.idempotency_key, + &receipt.message_id, + &receipt.processed_unix, + &receipt.receipt_evidence, + ], + ) + .await + .map_err(|error| format!("control plane restore outbox_receipt failed: {error}"))?; + } + tx.commit() + .await + .map_err(|error| format!("control plane restore commit failed: {error}"))?; + Ok(()) +} + +async fn insert_restored_outbox( + tx: &Transaction<'_>, + tenant_id: &str, + message: &OutboxMessage, +) -> Result<(), String> { + tx.execute( + "INSERT INTO outbox_message ( + tenant_id, message_id, aggregate_id, aggregate_version, event_type, + schema_version, created_unix, payload_json, payload_hash, idempotency_key, + message_status, lease_owner, lease_expires_unix, attempt_count, + first_attempt_unix, last_attempt_unix, next_available_unix, terminal_reason + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18)", + &[ + &tenant_id, + &message.message_id, + &message.aggregate_id, + &message.aggregate_version, + &message.event_type, + &message.schema_version, + &message.created_unix, + &message.payload_json, + &message.payload_hash, + &message.idempotency_key, + &message.message_status, + &message.lease_owner, + &message.lease_expires_unix, + &message.attempt_count, + &message.first_attempt_unix, + &message.last_attempt_unix, + &message.next_available_unix, + &message.terminal_reason, + ], + ) + .await + .map_err(|error| format!("control plane restore outbox_message failed: {error}"))?; + Ok(()) +} + +async fn drop_tenant(client: &mut Client, tenant_id: &str) -> Result<(), String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane drop-tenant transaction failed: {error}"))?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane tenant context failed: {error}"))?; + for table in [ + "outbox_receipt", + "outbox_message", + "threat_feed", + "audit_record", + "security_event", + "dnsbl_entry", + "threat_indicator", + "route_config", + "tenant_profile", + "tenant_account", + ] { + tx.execute( + &format!("DELETE FROM {table} WHERE tenant_id = $1"), + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane drop {table} failed: {error}"))?; + } + tx.commit() + .await + .map_err(|error| format!("control plane drop-tenant commit failed: {error}"))?; + Ok(()) +} + +fn unix_now_i64() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} + +fn mode_sql(mode: &EnforcementMode) -> &'static str { + match mode { + EnforcementMode::Monitor => "monitor", + EnforcementMode::Block => "block", + } +} + +fn parse_mode(value: &str) -> Result { + match value { + "monitor" => Ok(EnforcementMode::Monitor), + "block" => Ok(EnforcementMode::Block), + other => Err(format!("unknown enforcement_mode {other}")), + } +} + +fn severity_sql(severity: &Severity) -> &'static str { + match severity { + Severity::Low => "low", + Severity::Medium => "medium", + Severity::High => "high", + Severity::Critical => "critical", + } +} + +fn parse_severity(value: &str) -> Result { + match value { + "low" => Ok(Severity::Low), + "medium" => Ok(Severity::Medium), + "high" => Ok(Severity::High), + "critical" => Ok(Severity::Critical), + other => Err(format!("unknown severity_name {other}")), + } +} + +fn edition_sql(edition: &ProductEdition) -> &'static str { + match edition { + ProductEdition::Community => "community", + ProductEdition::Evaluation => "evaluation", ProductEdition::Enterprise => "enterprise", } } @@ -1448,12 +2253,74 @@ mod tests { } #[test] - fn database_url_rejects_non_postgres_and_tls_until_wired() { + fn database_url_rejects_non_postgres_and_ambiguous_sslmode() { parse_database_url("").unwrap_err(); parse_database_url("mysql://x").unwrap_err(); - parse_database_url("postgres://wardnet@127.0.0.1/wardnet?sslmode=require").unwrap_err(); + parse_database_url("postgres://wardnet@127.0.0.1/wardnet?sslmode=prefer").unwrap_err(); + parse_database_url("postgres://wardnet@127.0.0.1/wardnet?sslmode=allow").unwrap_err(); + parse_database_url("postgres://wardnet@127.0.0.1/wardnet?sslmode=require").unwrap(); + parse_database_url("postgres://wardnet@127.0.0.1/wardnet?sslmode=verify-full").unwrap(); parse_database_url("postgres://wardnet@127.0.0.1/wardnet").unwrap(); parse_database_url("postgresql://wardnet@127.0.0.1/wardnet?sslmode=disable").unwrap(); + assert_eq!( + ssl_mode("postgres://wardnet@127.0.0.1/wardnet").unwrap(), + SslMode::Disable + ); + assert_eq!( + ssl_mode("postgres://wardnet@127.0.0.1/wardnet?sslmode=require").unwrap(), + SslMode::Require + ); + assert_eq!( + ssl_mode("postgres://wardnet@127.0.0.1/wardnet?sslmode=verify-full").unwrap(), + SslMode::Require + ); + assert_eq!( + rewrite_sslmode_for_tokio( + "postgres://wardnet:sslmode=verify-full@127.0.0.1/wardnet?sslmode=verify-full" + ), + "postgres://wardnet:sslmode=verify-full@127.0.0.1/wardnet?sslmode=require" + ); + assert_eq!( + rewrite_sslmode_for_tokio( + "postgres://wardnet@127.0.0.1/wardnet?sslmode=verify-ca&connect_timeout=5" + ), + "postgres://wardnet@127.0.0.1/wardnet?sslmode=require&connect_timeout=5" + ); + use std::str::FromStr; + tokio_postgres::Config::from_str( + "postgres://wardnet@127.0.0.1/wardnet?sslmode=verify-full", + ) + .expect_err("tokio-postgres 0.7 rejects verify-full"); + tokio_postgres::Config::from_str(&rewrite_sslmode_for_tokio( + "postgres://wardnet@127.0.0.1/wardnet?sslmode=verify-full", + )) + .expect("rewritten verify-full must parse as require"); + } + + #[tokio::test] + async fn require_tls_fails_closed_against_plaintext_postgres() { + let Ok(url) = std::env::var("CONTROL_PLANE_TEST_DATABASE_URL") else { + return; + }; + if url.trim().is_empty() { + return; + } + let separator = if url.contains('?') { '&' } else { '?' }; + for mode in ["require", "verify-ca", "verify-full"] { + let tls_url = format!("{url}{separator}sslmode={mode}"); + let error = match PostgresPlane::connect(&tls_url).await { + Ok(_) => panic!("plaintext CI postgres must not satisfy rustls ({mode})"), + Err(error) => error, + }; + assert!( + !error.to_ascii_lowercase().contains("invalid value"), + "{mode} must not fail as a tokio-postgres config parse: {error}" + ); + assert!( + error.contains("TLS") || error.contains("ssl") || error.contains("certificate"), + "operator must see a TLS failure for {mode}, not a silent plaintext fallback: {error}" + ); + } } #[test] @@ -1475,6 +2342,8 @@ mod tests { } assert!(MIGRATION_SQL.contains("FORCE ROW LEVEL SECURITY")); assert!(MIGRATION_SQL.contains("wardnet.tenant_id")); + assert!(MIGRATION_SQL.contains("wardnet_runtime")); + assert!(MIGRATION_SQL.contains("NOBYPASSRLS")); assert!(MIGRATION_SQL.contains("PRIMARY KEY (tenant_id, route_id)")); assert!(MIGRATION_SQL.contains("REFERENCES tenant_account")); assert!(MIGRATION_SQL.contains("UNIQUE (tenant_id, idempotency_key)")); @@ -1507,7 +2376,10 @@ mod tests { assert_eq!(loaded.dnsbl, seeded.dnsbl); assert_eq!(loaded.next_event_id, seeded.next_event_id); assert_eq!(loaded.commercial.tenant_id, DEFAULT_TENANT_ID); - let messages = plane.list_outbox().await.expect("list snapshot outbox"); + let messages = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list snapshot outbox"); assert!( messages .iter() @@ -1578,7 +2450,10 @@ mod tests { "event must round-trip unmasked" ); assert_eq!(loaded.next_event_id, 8); - let messages = plane.list_outbox().await.expect("list outbox"); + let messages = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list outbox"); let recorded = messages .iter() .find(|message| message.event_type == EVENT_SECURITY_RECORDED) @@ -1590,7 +2465,10 @@ mod tests { .append_security_event(&event, 1_000) .await .expect("idempotent retry of same event id"); - let again = plane.list_outbox().await.expect("list after retry"); + let again = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list after retry"); assert_eq!( again .iter() @@ -1661,7 +2539,7 @@ mod tests { assert_eq!(health.dead_letter, 1); let dead = plane - .list_outbox() + .list_outbox_limited(LIST_LIMIT) .await .expect("list") .into_iter() @@ -1712,7 +2590,10 @@ mod tests { .await .expect("lease then fail transient"); assert_eq!(first, 0); - let listed = plane_a.list_outbox().await.expect("list after fail"); + let listed = plane_a + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list after fail"); let pending = listed .iter() .find(|message| message.event_type == EVENT_SECURITY_RECORDED) @@ -1727,4 +2608,468 @@ mod tests { .expect("expired/next-available reclaim"); assert_eq!(reclaimed, 1); } + + #[tokio::test] + async fn postgres_outbox_list_is_bounded_and_prunes_processed() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-bound"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("test database"); + plane.save(&AppData::seeded()).await.expect("seed"); + let now = unix_now_i64().saturating_add(60); + let _ = plane + .drain_once("setup", now, |_| Ok("setup".into())) + .await + .expect("ack snapshot outbox"); + for id in 1..=5 { + plane + .append_security_event(&sample_event(id, "/bound"), 3) + .await + .expect("enqueue"); + } + let processed = plane + .drain_once("worker-bound", now.saturating_add(30), |_| Ok("ack".into())) + .await + .expect("drain pending"); + assert_eq!(processed, 5); + plane + .append_security_event(&sample_event(6, "/bound-tail"), 3) + .await + .expect("append that prunes processed"); + let listed = plane + .list_outbox_limited(100) + .await + .expect("list after prune"); + assert_eq!( + listed + .iter() + .filter(|message| message.message_status == STATUS_PROCESSED) + .count(), + 3, + "processed outbox rows must be retained like EVENT_LIMIT" + ); + assert_eq!( + listed + .iter() + .filter(|message| message.message_status == STATUS_PENDING) + .count(), + 1 + ); + let bounded = plane.list_outbox_limited(2).await.expect("bounded list"); + assert_eq!(bounded.len(), 2); + assert_eq!(bounded[0].message_status, STATUS_PENDING); + let _ = plane + .drain_once("worker-bound", now.saturating_add(60), |_| { + Err(crate::outbox::DispatchError::Permanent("malformed".into())) + }) + .await + .expect("dead-letter the tail"); + let health = plane + .outbox_health(now.saturating_add(60)) + .await + .expect("health after poison"); + assert_eq!(health.dead_letter, 1); + for id in 7..=10 { + plane + .append_security_event(&sample_event(id, "/bound-more"), 3) + .await + .expect("more events after dead letter"); + let _ = plane + .drain_once("worker-bound", now.saturating_add(90 + id as i64), |_| { + Ok("ack".into()) + }) + .await + .expect("drain extra"); + } + let after = plane + .list_outbox_limited(100) + .await + .expect("list dead letter"); + assert!( + after + .iter() + .any(|message| message.message_status == STATUS_DEAD_LETTER), + "dead letters must not be pruned" + ); + } + + #[tokio::test] + async fn postgres_ack_and_save_prune_to_configured_event_limit() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-ack-limit"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("test database") + .with_event_limit(2); + plane.save(&AppData::seeded()).await.expect("seed"); + let now = unix_now_i64().saturating_add(60); + let _ = plane + .drain_once("setup", now, |_| Ok("setup".into())) + .await + .expect("ack snapshot outbox"); + for id in 1..=4 { + plane + .append_security_event(&sample_event(id, "/ack-limit"), 1_000) + .await + .expect("enqueue pending"); + } + let processed = plane + .drain_once("worker-ack-limit", now.saturating_add(30), |_| { + Ok("ack".into()) + }) + .await + .expect("drain pending"); + assert_eq!(processed, 4); + let listed = plane + .list_outbox_limited(100) + .await + .expect("list after ack prune"); + assert_eq!( + listed + .iter() + .filter(|message| message.message_status == STATUS_PROCESSED) + .count(), + 2, + "ack must prune processed rows to the configured EVENT_LIMIT, not LIST_LIMIT" + ); + plane + .save(&AppData::seeded()) + .await + .expect("save must use the same retention cap"); + let after_save = plane + .list_outbox_limited(100) + .await + .expect("list after save prune"); + assert_eq!( + after_save + .iter() + .filter(|message| message.message_status == STATUS_PROCESSED) + .count(), + 2, + "save_snapshot must prune processed rows to EVENT_LIMIT" + ); + } + + #[test] + fn backup_verify_fails_closed_on_schema_and_hash() { + let backup = ControlPlaneBackup { + schema_version: MIGRATION_VERSION, + tenant_id: "local-lab".into(), + created_unix: 1, + snapshot: AppData::seeded(), + outbox: Vec::new(), + receipts: Vec::new(), + payload_hash: String::new(), + } + .seal() + .expect("seal"); + assert!(backup.verify().is_ok()); + + let mut prior = backup.clone(); + prior.schema_version = MIN_RESTORABLE_SCHEMA_VERSION; + let prior = prior.seal().expect("re-seal compatible prior schema"); + assert!( + prior.verify().is_ok(), + "role-only schema 3 must restore schema-{MIN_RESTORABLE_SCHEMA_VERSION} logical backups" + ); + + let mut too_old = backup.clone(); + too_old.schema_version = MIN_RESTORABLE_SCHEMA_VERSION - 1; + assert!( + too_old + .verify() + .expect_err("older than restorable window") + .contains("unsupported") + ); + + let mut bad_schema = backup.clone(); + bad_schema.schema_version = MIGRATION_VERSION + 1; + assert!( + bad_schema + .verify() + .expect_err("future schema") + .contains("unsupported") + ); + + let mut bad_hash = backup.clone(); + bad_hash.payload_hash = "deadbeef".into(); + assert!( + bad_hash + .verify() + .expect_err("tamper") + .contains("payload_hash") + ); + } + + #[tokio::test] + async fn postgres_backup_restore_drill_preserves_unmasked_invariants() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("backup-drill"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("test database") + .with_event_limit(10); + let mut seeded = AppData::seeded(); + seeded.events.push(sample_event(1, "/backup-restore")); + seeded.next_event_id = 2; + plane.save(&seeded).await.expect("seed with unmasked event"); + let now = unix_now_i64().saturating_add(60); + plane + .append_security_event(&sample_event(2, "/backup-path"), 10) + .await + .expect("enqueue"); + let _ = plane + .drain_once("backup-worker", now, |_| Ok("backup-ack".into())) + .await + .expect("process one"); + let backup = plane.logical_backup().await.expect("export backup"); + backup.verify().expect("self-hash"); + assert!( + backup + .snapshot + .events + .iter() + .any(|event| event.path == "/backup-restore" + && event.client_ip.map(|ip| ip.to_string()) == Some("198.51.100.20".into())), + "backup must keep client IPs and paths unmasked" + ); + assert!( + backup + .outbox + .iter() + .any(|message| message.payload_json.contains("198.51.100.20")), + "outbox payloads must keep client IPs unmasked" + ); + + let isolated = unique_tenant("backup-restore-target"); + let target = PostgresPlane::connect_tenant(&url, &isolated) + .await + .expect("isolated restore tenant"); + target + .restore_logical_backup(&backup) + .await + .expect("restore into isolated tenant"); + let restored = target.logical_backup().await.expect("re-export restored"); + assert_eq!(restored.snapshot.routes, backup.snapshot.routes); + assert_eq!(restored.snapshot.events, backup.snapshot.events); + assert_eq!(restored.outbox.len(), backup.outbox.len()); + assert_eq!(restored.receipts.len(), backup.receipts.len()); + assert_eq!( + restored.semantic_hash().expect("restored hash"), + backup.semantic_hash().expect("source hash") + ); + + let report = plane.restore_drill().await.expect("isolated drill"); + assert!(report.passed, "drill must match source and restored hashes"); + assert!( + report.duration_ms <= BACKUP_RTO_BUDGET_MS, + "drill duration {}ms exceeds declared RTO {}ms", + report.duration_ms, + BACKUP_RTO_BUDGET_MS + ); + assert_eq!(report.rpo, BACKUP_RPO); + assert!( + plane + .load() + .await + .expect("source tenant still loads") + .is_some(), + "drill must not drop the production tenant" + ); + } + + #[tokio::test] + async fn postgres_runtime_role_is_not_superuser_and_rls_default_denies() { + let Some(url) = test_database_url() else { + return; + }; + let tenant_a = unique_tenant("runtime-a"); + let tenant_b = unique_tenant("runtime-b"); + let plane_a = PostgresPlane::connect_tenant(&url, &tenant_a) + .await + .expect("plane a"); + let (role, superuser) = plane_a.runtime_identity().await.expect("runtime identity"); + assert_eq!(role, RUNTIME_ROLE); + assert!(!superuser, "runtime role must not be a superuser"); + assert!( + plane_a.runtime_ddl_is_denied().await.expect("ddl probe"), + "runtime role must not DROP TABLE or DISABLE ROW LEVEL SECURITY" + ); + plane_a + .save(&AppData::seeded()) + .await + .expect("save tenant a"); + assert_eq!( + plane_a + .unscoped_route_count() + .await + .expect("unscoped count"), + 0, + "missing wardnet.tenant_id must yield no rows under FORCE RLS" + ); + let plane_b = PostgresPlane::connect_tenant(&url, &tenant_b) + .await + .expect("plane b"); + assert!( + plane_b.load().await.expect("load tenant b").is_none(), + "tenant b must not observe tenant a rows" + ); + let loaded_a = plane_a + .load() + .await + .expect("load tenant a") + .expect("tenant a rows"); + assert!(!loaded_a.routes.is_empty()); + } + + #[test] + fn hash_partition_parent_rejects_injection() { + assert!(hash_partition_sql_for("security_event").is_ok()); + assert!(hash_partition_sql_for("event_hash_probe").is_ok()); + assert!(hash_partition_sql_for("event;drop").is_err()); + assert!(hash_partition_sql_for("Event").is_err()); + assert!(hash_partition_sql_for("").is_err()); + let sql = hash_partition_sql_for("security_event").expect("sql"); + assert!(sql.contains("PARTITION BY HASH (tenant_id)")); + assert!(sql.contains(&format!("MODULUS {EVENT_PARTITION_MODULUS}"))); + } + + async fn owner_client(url: &str) -> tokio_postgres::Client { + let (client, connection) = tokio_postgres::connect(url, tokio_postgres::NoTls) + .await + .expect("owner connect"); + tokio::spawn(async move { + let _ = connection.await; + }); + client + } + + #[tokio::test] + async fn postgres_security_event_is_hash_partitioned() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("event-hash"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("plane"); + assert_eq!( + plane + .event_partition_count() + .await + .expect("partition count"), + i64::from(EVENT_PARTITION_MODULUS) + ); + let mut seeded = AppData::seeded(); + seeded.events.push(sample_event(1, "/hash-path")); + seeded.next_event_id = 2; + plane.save(&seeded).await.expect("seed with unmasked event"); + let loaded = plane.load().await.expect("load").expect("snapshot"); + assert!( + loaded.events.iter().any(|event| event.path == "/hash-path" + && event.client_ip.map(|ip| ip.to_string()) == Some("198.51.100.20".into())), + "HASH partitions must keep client IPs and paths unmasked" + ); + let tableoid = plane + .security_event_tableoid() + .await + .expect("child tableoid"); + assert!( + tableoid.starts_with("security_event_p"), + "row must land in a HASH child, got {tableoid}" + ); + } + + #[tokio::test] + async fn postgres_hash_partition_convert_preserves_unmasked_probe_rows() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("hash-probe"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("plane so tenant_account and role exist"); + plane.save(&AppData::seeded()).await.expect("seed tenant"); + + let client = owner_client(&url).await; + client + .batch_execute( + "DROP TABLE IF EXISTS event_hash_probe CASCADE; + CREATE TABLE event_hash_probe ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + event_id BIGINT NOT NULL, + timestamp_unix BIGINT NOT NULL, + client_address TEXT, + route_id TEXT, + action_name TEXT NOT NULL, + event_reason TEXT NOT NULL, + event_score INTEGER NOT NULL, + request_path TEXT NOT NULL, + PRIMARY KEY (tenant_id, event_id) + );", + ) + .await + .expect("unpartitioned probe"); + client + .execute( + "INSERT INTO event_hash_probe ( + tenant_id, event_id, timestamp_unix, client_address, route_id, + action_name, event_reason, event_score, request_path + ) VALUES ($1, 1, 1700000000, '198.51.100.20', 'demo', 'blocked', 'fixture', 80, '/probe-path')", + &[&tenant], + ) + .await + .expect("probe row"); + let sql = hash_partition_sql_for("event_hash_probe").expect("probe sql"); + client + .batch_execute(&sql) + .await + .expect("convert unpartitioned probe"); + let kind: String = client + .query_one( + "SELECT relkind::text FROM pg_class WHERE relname = 'event_hash_probe'", + &[], + ) + .await + .expect("kind") + .get(0); + assert_eq!(kind, "p"); + let children: i64 = client + .query_one( + "SELECT COUNT(*)::bigint FROM pg_partition_tree('event_hash_probe'::regclass) WHERE level = 1", + &[], + ) + .await + .expect("children") + .get(0); + assert_eq!(children, i64::from(EVENT_PARTITION_MODULUS)); + let row = client + .query_one( + "SELECT client_address, request_path, tableoid::regclass::text + FROM event_hash_probe WHERE event_id = 1", + &[], + ) + .await + .expect("converted row"); + let ip: String = row.get(0); + let path: String = row.get(1); + let tableoid: String = row.get(2); + assert_eq!(ip, "198.51.100.20"); + assert_eq!(path, "/probe-path"); + assert!( + tableoid.starts_with("event_hash_probe_p"), + "converted row must land in a HASH child, got {tableoid}" + ); + client + .batch_execute("DROP TABLE IF EXISTS event_hash_probe CASCADE") + .await + .expect("drop probe"); + } } diff --git a/src/lib.rs b/src/lib.rs index 03b4ee4e..6d7526a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -137,7 +137,9 @@ impl AppState { /// Load from PostgreSQL, seeding the tenant snapshot when empty. pub async fn load_postgres(config: AppConfig, database_url: &str) -> Result { - let plane = control_plane::PostgresPlane::connect(database_url).await?; + let plane = control_plane::PostgresPlane::connect(database_url) + .await? + .with_event_limit(config.event_limit); let mut data = match plane.load().await? { Some(loaded) => loaded, None => AppData::seeded(), @@ -355,6 +357,12 @@ impl AppState { outbox_leased: 0, outbox_dead_letter: 0, outbox_oldest_age_seconds: None, + backup: if self.control_plane.is_some() { + "ready".to_string() + } else { + "disabled".to_string() + }, + event_partitions: 0, } } @@ -373,6 +381,9 @@ impl AppState { } Err(_) => health.outbox = "error".to_string(), } + if let Ok(count) = plane.event_partition_count().await { + health.event_partitions = count; + } health } } @@ -507,6 +518,10 @@ pub struct HealthStatus { pub outbox_leased: i64, pub outbox_dead_letter: i64, pub outbox_oldest_age_seconds: Option, + /// `ready` when PostgreSQL logical backup/restore is the authority; `disabled` on file/memory. + pub backup: String, + /// HASH child count for `security_event` (0 on file/memory). + pub event_partitions: i64, } const PHISHING_DATABASE_DEFAULT_FEED_ID: &str = "phishing-database-active"; @@ -581,6 +596,8 @@ pub fn build_app(state: AppState) -> Router { .route("/api/audit-logs", get(list_audit_logs)) .route("/api/outbox", get(list_outbox)) .route("/api/outbox/{message_id}/replay", post(replay_outbox)) + .route("/api/backup", get(get_backup).post(restore_backup)) + .route("/api/backup/drill", post(backup_drill)) .route("/api/events.ndjson", get(events_ndjson)) .route("/api/kpis", get(kpis)) .route("/api/signatures", get(list_signatures)) @@ -1131,6 +1148,7 @@ async fn list_audit_logs(State(state): State, headers: HeaderMap) -> R #[derive(Serialize)] struct OutboxListView { status: String, + limit: usize, messages: Vec, } @@ -1138,16 +1156,19 @@ async fn list_outbox(State(state): State, headers: HeaderMap) -> Respo if !admin_authenticated(&state, &headers) { return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); } + let limit = state.event_limit.max(1); let Some(plane) = &state.control_plane else { return Json(OutboxListView { status: "disabled".to_string(), + limit, messages: Vec::new(), }) .into_response(); }; - match plane.list_outbox().await { + match plane.list_outbox_limited(limit as i64).await { Ok(messages) => Json(OutboxListView { status: "ready".to_string(), + limit, messages, }) .into_response(), @@ -1197,6 +1218,135 @@ async fn replay_outbox( } } +#[derive(Serialize)] +struct BackupView { + status: String, + rpo: String, + rto_budget_ms: u64, + artifact: Option, +} + +async fn get_backup(State(state): State, headers: HeaderMap) -> Response { + if !admin_authenticated(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return Json(BackupView { + status: "disabled".to_string(), + rpo: control_plane::BACKUP_RPO.to_string(), + rto_budget_ms: control_plane::BACKUP_RTO_BUDGET_MS, + artifact: None, + }) + .into_response(); + }; + match plane.logical_backup().await { + Ok(artifact) => Json(BackupView { + status: "ready".to_string(), + rpo: control_plane::BACKUP_RPO.to_string(), + rto_budget_ms: control_plane::BACKUP_RTO_BUDGET_MS, + artifact: Some(artifact), + }) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +async fn restore_backup( + State(state): State, + headers: HeaderMap, + Json(backup): Json, +) -> Response { + if !admin_authorized(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return error( + StatusCode::SERVICE_UNAVAILABLE, + "backup restore requires the PostgreSQL control plane", + ); + }; + if let Err(message) = backup.verify() { + return error(StatusCode::BAD_REQUEST, message); + } + if let Err(message) = plane.restore_logical_backup(&backup).await { + return error(StatusCode::BAD_REQUEST, message); + } + match plane.load().await { + Ok(Some(loaded)) => { + *state.inner.write().await = loaded; + } + Ok(None) => { + return error( + StatusCode::INTERNAL_SERVER_ERROR, + "restore committed but tenant snapshot is empty", + ); + } + Err(message) => return error(StatusCode::INTERNAL_SERVER_ERROR, message), + } + let actor = audit_actor(&state, &headers); + match state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + "restore_backup", + "control_plane_backup", + backup.payload_hash.clone(), + ); + }) + .await + { + Ok(_) => Json(serde_json::json!({ + "status": "restored", + "schema_version": backup.schema_version, + "payload_hash": backup.payload_hash, + })) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +async fn backup_drill(State(state): State, headers: HeaderMap) -> Response { + if !admin_authorized(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return error( + StatusCode::SERVICE_UNAVAILABLE, + "backup drill requires the PostgreSQL control plane", + ); + }; + let report = match plane.restore_drill().await { + Ok(report) => report, + Err(message) => return error(StatusCode::INTERNAL_SERVER_ERROR, message), + }; + let actor = audit_actor(&state, &headers); + let outcome = if report.passed { + "backup_drill" + } else { + "backup_drill_failed" + }; + if let Err(message) = state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + outcome, + "control_plane_backup", + report.source_hash.clone(), + ); + }) + .await + { + return error(StatusCode::INTERNAL_SERVER_ERROR, message); + } + if report.passed { + (StatusCode::OK, Json(report)).into_response() + } else { + (StatusCode::INTERNAL_SERVER_ERROR, Json(report)).into_response() + } +} + async fn kpis(State(state): State) -> Json { let data = state.inner.read().await; Json(kpi_snapshot_at(&data, now_unix())) @@ -3137,6 +3287,14 @@ input,select{font:inherit;min-height:44px;padding:0 12px;border:1px solid var(--

Audit log

Loading…

Outbox

PostgreSQL leased workers for external effects. File/memory adapters report disabled. Client IPs and paths are not masked.

Loading…
+

Control-plane backup

+

On-demand PostgreSQL logical snapshot. Restore drill uses an isolated tenant and does not mask client IPs, paths, or actors. File/memory adapters report disabled. Declared RPO is last successful export; declared RTO is 60s.

+
Loading…
+
+ +
+ 

+    

     
Evidence manifest
Loading…

     
SOC event export (ndjson)
Loading…