diff --git a/CHANGELOG.md b/CHANGELOG.md index 0070ff0..883bb7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- 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/CLAUDE.md b/CLAUDE.md index d3843fa..f54f156 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -62,7 +62,7 @@ The core stays an in-repo workspace crate on purpose (no git submodule) until it ## Runtime Configuration -Read in `run_from_env` (`src/lib.rs`): `BIND_ADDR` (default `127.0.0.1:8080`), `ADMIN_TOKEN` (write token for `X-Admin-Token`), `ADMIN_TOKENS` (comma-separated `token:actor` pairs for multi-token RBAC with per-token audit actors), `WAF_IDS_STATE_PATH` (optional JSON state file; omitted = seeded in-memory state), `DNSBL_ORIGIN` (default `dnsbl.local`), `EVENT_LIMIT` (default 1000, must be > 0), `RATE_LIMIT` / `RATE_LIMIT_WINDOW`, `CORAZA_LIB_PATH` / `CORAZA_RULES_PATH` / `CORAZA_DIRECTIVES` (optional in-process libcoraza), `CORAZA_WAF_URL` (optional in-path Coraza sidecar), `PROVEN_ENGINE_FAIL_CLOSED` (boolean; default false — set true in production when an engine is set). +Read in `run_from_env` (`src/lib.rs`): `BIND_ADDR` (default `127.0.0.1:8080`), `ADMIN_TOKEN` (write token for `X-Admin-Token`), `ADMIN_TOKENS` (comma-separated `token:actor` pairs for multi-token RBAC with per-token audit actors), `WAF_IDS_STATE_PATH` (optional JSON state file; omitted = seeded in-memory state), `CONTROL_PLANE_DATABASE_URL` (required for non-loopback binds; secret `control_plane_url`), `DNSBL_ORIGIN` (default `dnsbl.local`), `EVENT_LIMIT` (default 1000, must be > 0), `RATE_LIMIT` / `RATE_LIMIT_WINDOW`, `CORAZA_LIB_PATH` / `CORAZA_RULES_PATH` / `CORAZA_DIRECTIVES` (optional in-process libcoraza), `CORAZA_WAF_URL` (optional in-path Coraza sidecar), `PROVEN_ENGINE_FAIL_CLOSED` (boolean; default false — set true in production when an engine is set). PostgreSQL mode starts a leased outbox worker (`GET /api/outbox`, `/healthz.outbox`). ## Key Conventions diff --git a/Cargo.lock b/Cargo.lock index a05a89e..831e74f 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" @@ -104,6 +110,15 @@ version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "block-buffer" version = "0.12.1" @@ -160,7 +175,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.0", "rand_core 0.10.1", ] @@ -170,12 +185,27 @@ 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" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "cpufeatures" version = "0.3.0" @@ -185,6 +215,16 @@ dependencies = [ "libc", ] +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "crypto-common" version = "0.2.2" @@ -203,15 +243,48 @@ 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" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer 0.10.4", + "crypto-common 0.1.7", +] + [[package]] name = "digest" version = "0.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ - "block-buffer", - "const-oid", - "crypto-common", + "block-buffer 0.12.1", + "const-oid 0.10.2", + "crypto-common 0.2.2", "ctutils", ] @@ -254,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" @@ -330,6 +409,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -375,7 +464,7 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" dependencies = [ - "digest", + "digest 0.11.3", ] [[package]] @@ -689,7 +778,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" dependencies = [ "cfg-if", - "digest", + "digest 0.11.3", ] [[package]] @@ -826,7 +915,7 @@ dependencies = [ "md-5", "memchr", "rand 0.10.2", - "sha2", + "sha2 0.11.0", "stringprep", ] @@ -1255,6 +1344,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures 0.2.17", + "digest 0.10.7", +] + [[package]] name = "sha2" version = "0.11.0" @@ -1262,8 +1362,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", - "cpufeatures", - "digest", + "cpufeatures 0.3.0", + "digest 0.11.3", ] [[package]] @@ -1310,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" @@ -1433,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" @@ -1486,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" @@ -1651,6 +1797,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "waf-ids-ai-soc" version = "0.1.0" @@ -1660,10 +1812,13 @@ dependencies = [ "libloading", "proptest", "reqwest", + "rustls", "serde", "serde_json", + "sha2 0.10.9", "tokio", "tokio-postgres", + "tokio-postgres-rustls", "tower", "waf-ids-core", ] @@ -1939,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" @@ -2008,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 88185c4..7b9bc24 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,9 @@ 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] tower = { version = "0.5", features = ["util"] } diff --git a/README.md b/README.md index 47b4466..35bd2d2 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 291a390..b22c7d2 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 b998080..6edc3e5 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,7 +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/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. @@ -58,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 new file mode 100644 index 0000000..cb6337a --- /dev/null +++ b/docs/doctoring/outbox-workers.md @@ -0,0 +1,55 @@ +# Doctoring — transactional outbox and leased workers + +This note grounds issue #81 (external effects leave the PostgreSQL control plane +through a transactional outbox, not request-path retry loops). IEEE/ACM PDFs +are not redistributed. + +## Adopted standards and literature + +PostgreSQL Global Development Group. (2026). *PostgreSQL documentation: Explicit +locking*. https://www.postgresql.org/docs/current/explicit-locking.html + +- **Design impact:** Workers claim `outbox_message` rows with + `FOR UPDATE SKIP LOCKED`. Expired leases are reclaimable. Unrelated tenants + and aggregates are not globally serialized. + +PostgreSQL Global Development Group. (2026). *PostgreSQL documentation: +Transaction isolation*. https://www.postgresql.org/docs/current/transaction-iso.html + +- **Design impact:** The security event (or policy snapshot) and its outbox row + commit in one transaction. A crash after domain commit but before dispatch + leaves a pending message; it cannot invent extra authority. + +Hohpe, G., & Woolf, B. (2003). *Enterprise integration patterns: Designing, +building, and deploying messaging solutions*. Addison-Wesley. + +- **Design impact:** Transactional outbox. Downstream stdout SIEM export is + **at-least-once**. The `outbox_receipt` unique `(tenant_id, idempotency_key)` + is the exactly-once business acknowledgement. Do not call transport delivery + exactly once. + +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 / PW.7 — durable retry, dead-letter, and authorized + replay with audit. Replay is a write (`X-Admin-Token`). + +## Operator next action + +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 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`. `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 551ac4d..7fff5bb 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 0000000..38cbc71 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 0074571..4f5328b 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-23T16:20Z (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,13 +25,20 @@ not “waiting on review/CI time”. | PR | Title | Head | Checks | Reviews | Merge blocker | | --- | --- | --- | --- | --- | --- | -| [#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 this 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. Remaining unresolved: DESTINATION_* env (documented operational-config deviation), hostname-allowlist mixed answers (intended), sidecar loopback needs allowlist in production, pin-cap eviction info | Org 2-approval + self-author. Merge #95 first. Do not re-implement the TCP-peer pin. | +| [#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. | | [#94](https://github.com/ContextualWisdomLab/wardnet/pull/94) | fix(auth): fail closed without write-capable admin on public bind | `f31d960a0b52` (`fix/issue-78-fail-closed-credentials`) | Checks re-ran after readiness-order fix | Author `seonghobae`; Devin COMMENTED | Org 2-approval + self-author. Do not `--admin` merge. Do not re-implement #78. | | [#93](https://github.com/ContextualWisdomLab/wardnet/pull/93) | test(persistence): replace permission-based fault injection with a deterministic seam | `f77eb69748ec` | rust + Security Scan green; **strix FAILURE** (org LiteLLM provider `openai-direct/gpt-5.6-luna`) | Author `seonghobae`; Devin COMMENTED | strix org-provider FAILURE + 2-approval + self-author. Do not rotate review-agent keys. | -| [#92](https://github.com/ContextualWisdomLab/wardnet/pull/92) | build(deps): bump github/codeql-action/upload-sarif from 4.37.6 to 4.37.7 | dependabot `17277d78d5e1` | All green. `--auto` squash already enabled. Copilot review re-requested this hour. | Maintainer APPROVED (1 of 2). | **Second independent APPROVE missing**. `gh pr merge` rejected by ruleset 18156473. | -| [#91](https://github.com/ContextualWisdomLab/wardnet/pull/91) | build(deps): bump futures-util from 0.3.33 to 0.3.34 | dependabot `c4662caebfa1` | All green. `--auto` squash already enabled. Copilot review re-requested this hour. | Maintainer APPROVED (1 of 2). | Same as #92: second independent APPROVE missing. | +| [#92](https://github.com/ContextualWisdomLab/wardnet/pull/92) | build(deps): bump github/codeql-action/upload-sarif from 4.37.6 to 4.37.7 | dependabot `17277d78d5e1` | All green. `--auto` squash already enabled. | Maintainer APPROVED (1 of 2). | **Second independent APPROVE missing**. `gh pr merge` rejected by ruleset 18156473. | +| [#91](https://github.com/ContextualWisdomLab/wardnet/pull/91) | build(deps): bump futures-util from 0.3.33 to 0.3.34 | dependabot `c4662caebfa1` | All green. `--auto` squash already enabled. | Maintainer APPROVED (1 of 2). | Same as #92: second independent APPROVE missing. | | [#90](https://github.com/ContextualWisdomLab/wardnet/pull/90) | feat(observability): export Wardnet events to SIEM and OpenTelemetry | `40f11b93a972` | All green (35). | Author `seonghobae`; CodeRabbit/Devin/GHAS COMMENTED. **0 unresolved threads** on exact head. | Org 2-approval + self-author. | | [#88](https://github.com/ContextualWisdomLab/wardnet/pull/88) | feat(security): reject non-LiteLLM credentials before upstream | `41b21cfe2168` | All green (35). | Author `seonghobae`; CodeRabbit COMMENTED. **0 unresolved threads** on exact head. | Org 2-approval + self-author. | | [#77](https://github.com/ContextualWisdomLab/wardnet/pull/77) | build(rust): pin and track Rust 1.97.1 | `a13c08656177` | rust green; **strix FAILURE**. Same org-provider fail-closed as #93. | Author `seonghobae`; Devin COMMENTED. | strix org-provider FAILURE + 2-approval + self-author. | @@ -52,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** | -| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical** | +| [#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 | @@ -71,20 +78,17 @@ still say `waf-ids-ai-soc`. Kubernetes manifest remains this pass: docs and health copy already mention Wardnet in newer surfaces; wholesale crate rename is deferred (not a merge blocker). -### Proven-engine enforcement (issue #86) — **in-process libcoraza this pass** +### Proven-engine enforcement (issue #86) — **in-process libcoraza shipped, unmerged** Coraza/Suricata ingest still maps proven-engine hits into DNSBL + threat indicators. PR #95 consults a Coraza sidecar on each live `/gateway` -transaction when `CORAZA_WAF_URL` is set. This pass also `dlopen`s -operator-supplied libcoraza (`CORAZA_LIB_PATH` + `CORAZA_RULES_PATH` and/or -`CORAZA_DIRECTIVES`) and evaluates the same live transactions through the -libcoraza C ABI (`src/coraza_inprocess.rs`). In-process wins over sidecar when -both are set. Missing library, missing rules, or an empty ruleset fail -startup before bind. `GET /api/waf/engine-status` and `/healthz.proven_engine` -report `coraza_in_process` / `coraza_sidecar` / `ingest_hints_only`. CI stays -hermetic with a fixture cdylib that exports the same symbols; production -points at a real libcoraza + CRS bundle. Suricata tail/shipper and -detection-quality corpora remain open. +transaction when `CORAZA_WAF_URL` is set. PR #97 `dlopen`s operator-supplied +libcoraza (`CORAZA_LIB_PATH` + `CORAZA_RULES_PATH` and/or `CORAZA_DIRECTIVES`) +and evaluates the same live transactions through the libcoraza C ABI +(`src/coraza_inprocess.rs`). In-process wins over sidecar when both are set. +Missing library, missing rules, or an empty ruleset fail startup before bind. +`GET /api/waf/engine-status` and `/healthz.proven_engine` report +`coraza_in_process` / `coraza_sidecar` / `ingest_hints_only`. Do not re-implement. ### Identity (issue #82, Keyverse) @@ -92,15 +96,44 @@ 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 + RLS snapshot this pass** +### 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. +`/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. +Policy snapshots enqueue `policy.snapshot_replaced`. A leased worker claims +with `FOR UPDATE SKIP LOCKED`, retries with bounded exponential backoff, +dead-letters permanent/exhausted failures, and records unique receipts. +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. `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** @@ -116,7 +149,8 @@ Production sidecar URLs on loopback/private still need `DESTINATION_ALLOWLIST` ### SIEM / OpenTelemetry (issue #85 / PR #90) `/api/events.ndjson` and stdout JSON lines exist on main. Full exporter binary -and OTel sit on PR #90, blocked by the 2-approval ruleset. +and OTel sit on PR #90, blocked by the 2-approval ruleset. The #81 worker now +replays `security_event.recorded` as stdout SIEM with receipts. ### UI-UX / Storybook / Figma @@ -128,6 +162,9 @@ and OTel sit on PR #90, blocked by the 2-approval ruleset. | 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 | +| 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 @@ -139,15 +176,16 @@ future encryption-at-rest. ### Coverage / docstring bar Org 100% line/branch/docstring applies to **changed** surfaces this loop -(libcoraza loader, engine-status in-process fields, startup fail-closed, -gateway consult). 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) 1. **keyverse** — identity for management plane (#82). 2. **contextual-orchestrator** — SOC LLM already optional via - `SOC_LLM_BASE_URL`; keep adapter, do not fork routing. + `SOC_LLM_BASE_URL`; keep adapter, do not fork routing. Next: same outbox + contract. 3. **naruon** / **clearfolio** — document viewer already optional. 4. **TEPP / RankWeave / ThreadWeave / LineageWeave / disksage / fast-mlsirm** — not on the gateway data path; no connector this pass. @@ -192,11 +230,19 @@ tests: `run_from_env_fail_closes_public_bind_without_postgres`, service). Do not re-implement #78, the #86 sidecar/libcoraza slices, or the #79 pin. +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 and this #80 PR merge-ready. Merge order #95 then #96 - then #97 then this. Do not re-implement shipped slices. -3. Next runtime gap if policy still blocks: #81 outbox/workers on this - postgres authority, or rustls / backup drill remainder of #80. +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 ac08132..a00c44d 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 b57968e..f54b092 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 59472dc..25b0b59 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -82,6 +82,9 @@ assert_json_field "$health" 'data["event_limit"] == 5' assert_json_field "$health" 'data["proven_engine"] == "ingest_hints_only"' 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 5fd10d7..793ceef 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -5,10 +5,18 @@ //! production authority. Tenant isolation is default-deny row-level security //! with `FORCE ROW LEVEL SECURITY`; each transaction sets `wardnet.tenant_id`. +use crate::outbox::{ + self, CLAIM_BATCH, DispatchError, EVENT_SECURITY_RECORDED, EVENT_SNAPSHOT_REPLACED, + 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}; +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, @@ -17,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 = 1; +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#" @@ -117,6 +141,41 @@ CREATE TABLE IF NOT EXISTS threat_feed ( CREATE INDEX IF NOT EXISTS security_event_tenant_event ON security_event (tenant_id, event_id); +CREATE TABLE IF NOT EXISTS outbox_message ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + message_id TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + aggregate_version BIGINT NOT NULL, + event_type TEXT NOT NULL, + schema_version INTEGER NOT NULL, + created_unix BIGINT NOT NULL, + payload_json TEXT NOT NULL, + payload_hash TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + message_status TEXT NOT NULL, + lease_owner TEXT, + lease_expires_unix BIGINT, + attempt_count INTEGER NOT NULL, + first_attempt_unix BIGINT, + last_attempt_unix BIGINT, + next_available_unix BIGINT NOT NULL, + terminal_reason TEXT, + PRIMARY KEY (tenant_id, message_id), + UNIQUE (tenant_id, idempotency_key) +); + +CREATE TABLE IF NOT EXISTS outbox_receipt ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + idempotency_key TEXT NOT NULL, + message_id TEXT NOT NULL, + processed_unix BIGINT NOT NULL, + receipt_evidence TEXT NOT NULL, + PRIMARY KEY (tenant_id, idempotency_key) +); + +CREATE INDEX IF NOT EXISTS outbox_message_claim + ON outbox_message (tenant_id, message_status, next_available_unix); + ALTER TABLE tenant_account ENABLE ROW LEVEL SECURITY; ALTER TABLE tenant_account FORCE ROW LEVEL SECURITY; DROP POLICY IF EXISTS tenant_isolation ON tenant_account; @@ -172,8 +231,153 @@ DROP POLICY IF EXISTS tenant_isolation ON threat_feed; CREATE POLICY tenant_isolation ON threat_feed USING (tenant_id = current_setting('wardnet.tenant_id', true)) WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE outbox_message ENABLE ROW LEVEL SECURITY; +ALTER TABLE outbox_message FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON outbox_message; +CREATE POLICY tenant_isolation ON outbox_message + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE outbox_receipt ENABLE ROW LEVEL SECURITY; +ALTER TABLE outbox_receipt FORCE ROW LEVEL SECURITY; +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, @@ -191,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() { @@ -202,51 +405,327 @@ 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(())); + /// Live PostgreSQL snapshot store for one tenant. pub struct PostgresPlane { client: Mutex, tenant_id: String, + /// Processed-outbox retention; mirrors operator `EVENT_LIMIT`. + event_limit: i64, } impl PostgresPlane { pub async fn connect(url: &str) -> Result { + Self::connect_tenant(url, DEFAULT_TENANT_ID).await + } + + 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: DEFAULT_TENANT_ID.to_string(), + 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; client - .batch_execute(MIGRATION_SQL) + .execute("SELECT pg_advisory_lock($1)", &[&MIGRATION_LOCK_KEY]) .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| 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 current_user, current_setting('is_superuser') = 'on'", + &[], ) .await - .map_err(|error| format!("control plane migration version failed: {error}"))?; - Ok(()) + .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| 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| 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. @@ -255,10 +734,205 @@ impl PostgresPlane { load_snapshot(&mut client, &self.tenant_id).await } - /// Replace the tenant snapshot in one transaction (mutation + audit). + /// 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. + pub async fn append_security_event( + &self, + event: &SecurityEvent, + event_limit: usize, + ) -> Result<(), String> { + let mut client = self.client.lock().await; + append_security_event(&mut client, &self.tenant_id, event, event_limit).await + } + + pub async fn drain_once( + &self, + owner: &str, + now_unix: i64, + dispatch: F, + ) -> Result + where + F: Fn(&OutboxMessage) -> Result, + { + let mut client = self.client.lock().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 { + let mut client = self.client.lock().await; + outbox_health(&mut client, &self.tenant_id, now_unix).await + } + + 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, 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)) } } @@ -311,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 @@ -323,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) @@ -503,10 +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}"))?; } - - tx.commit() - .await - .map_err(|error| format!("control plane commit failed: {error}"))?; Ok(()) } @@ -551,7 +1241,7 @@ async fn load_routes( let rows = client .query( "SELECT route_id, path_prefix, upstream_url, enforcement_mode, is_enabled, block_threshold - FROM route_config WHERE tenant_id = $1", + FROM route_config WHERE tenant_id = $1 ORDER BY route_id", &[&tenant_id], ) .await @@ -577,7 +1267,8 @@ async fn load_threats( let rows = client .query( "SELECT indicator_type, indicator_value, indicator_source, severity_name, ttl_seconds - FROM threat_indicator WHERE tenant_id = $1", + FROM threat_indicator WHERE tenant_id = $1 + ORDER BY indicator_type, indicator_value, indicator_source", &[&tenant_id], ) .await @@ -602,7 +1293,7 @@ async fn load_dnsbl( let rows = client .query( "SELECT host_address, response_code, block_reason, entry_source, ttl_seconds, prefix_length - FROM dnsbl_entry WHERE tenant_id = $1", + FROM dnsbl_entry WHERE tenant_id = $1 ORDER BY host_address", &[&tenant_id], ) .await @@ -693,7 +1384,7 @@ async fn load_feeds( let rows = client .query( "SELECT feed_id, feed_source, last_updated_unix, threat_count, dnsbl_count, ttl_seconds - FROM threat_feed WHERE tenant_id = $1", + FROM threat_feed WHERE tenant_id = $1 ORDER BY feed_id", &[&tenant_id], ) .await @@ -711,49 +1402,816 @@ async fn load_feeds( .collect()) } -fn mode_sql(mode: &EnforcementMode) -> &'static str { - match mode { - EnforcementMode::Monitor => "monitor", - EnforcementMode::Block => "block", - } +async fn enqueue_snapshot_outbox( + tx: &Transaction<'_>, + tenant_id: &str, + data: &AppData, +) -> Result<(), String> { + let payload = serde_json::json!({ + "route_count": data.routes.len(), + "threat_count": data.threats.len(), + "dnsbl_count": data.dnsbl.len(), + "event_count": data.events.len(), + "audit_count": data.audit_logs.len(), + "event_sequence": data.next_event_id, + "audit_sequence": data.next_audit_log_id, + }) + .to_string(); + let hash = outbox::payload_hash(&payload); + let (message_id, idempotency_key) = + outbox::snapshot_ids(tenant_id, data.next_event_id, data.next_audit_log_id, &hash); + insert_outbox( + tx, + tenant_id, + &OutboxInsert { + message_id, + aggregate_id: tenant_id.to_string(), + aggregate_version: data.next_audit_log_id as i64, + event_type: EVENT_SNAPSHOT_REPLACED, + created_unix: unix_now_i64(), + payload_json: payload, + payload_hash: hash, + idempotency_key, + }, + ) + .await } -fn parse_mode(value: &str) -> Result { - match value { - "monitor" => Ok(EnforcementMode::Monitor), - "block" => Ok(EnforcementMode::Block), - other => Err(format!("unknown enforcement_mode {other}")), - } -} +async fn append_security_event( + client: &mut Client, + tenant_id: &str, + event: &SecurityEvent, + event_limit: usize, +) -> Result<(), String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane event 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}"))?; -fn severity_sql(severity: &Severity) -> &'static str { - match severity { - Severity::Low => "low", - Severity::Medium => "medium", - Severity::High => "high", - Severity::Critical => "critical", - } -} + let next_event_id = event.id.saturating_add(1) as i64; + tx.execute( + "INSERT INTO tenant_account (tenant_id, event_sequence, audit_sequence) + VALUES ($1, $2, 1) + ON CONFLICT (tenant_id) DO UPDATE SET + event_sequence = GREATEST(tenant_account.event_sequence, EXCLUDED.event_sequence)", + &[&tenant_id, &next_event_id], + ) + .await + .map_err(|error| format!("control plane upsert tenant_account failed: {error}"))?; -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}")), - } + let client_address = event.client_ip.map(|ip| ip.to_string()); + tx.execute( + "INSERT INTO security_event ( + tenant_id, event_id, timestamp_unix, client_address, route_id, + action_name, event_reason, event_score, request_path + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) + ON CONFLICT (tenant_id, event_id) DO NOTHING", + &[ + &tenant_id, + &(event.id as i64), + &(event.timestamp_unix as i64), + &client_address, + &event.route_id, + &event.action, + &event.reason, + &i32::from(event.score), + &event.path, + ], + ) + .await + .map_err(|error| format!("control plane insert security_event failed: {error}"))?; + + let keep_from = next_event_id.saturating_sub(event_limit.max(1) as i64); + tx.execute( + "DELETE FROM security_event WHERE tenant_id = $1 AND event_id < $2", + &[&tenant_id, &keep_from], + ) + .await + .map_err(|error| format!("control plane event retention failed: {error}"))?; + + let payload = serde_json::to_string(event).expect("SecurityEvent is JSON-serializable"); + let hash = outbox::payload_hash(&payload); + let (message_id, idempotency_key) = outbox::security_event_ids(tenant_id, event.id); + insert_outbox( + &tx, + tenant_id, + &OutboxInsert { + message_id, + aggregate_id: event.id.to_string(), + aggregate_version: event.id as i64, + event_type: EVENT_SECURITY_RECORDED, + created_unix: event.timestamp_unix as i64, + payload_json: payload, + payload_hash: hash, + idempotency_key, + }, + ) + .await?; + prune_processed_outbox(&tx, tenant_id, event_limit as i64).await?; + + tx.commit() + .await + .map_err(|error| format!("control plane event commit failed: {error}"))?; + Ok(()) } -fn edition_sql(edition: &ProductEdition) -> &'static str { - match edition { - ProductEdition::Community => "community", - ProductEdition::Evaluation => "evaluation", - ProductEdition::Enterprise => "enterprise", - } +struct OutboxInsert { + message_id: String, + aggregate_id: String, + aggregate_version: i64, + event_type: &'static str, + created_unix: i64, + payload_json: String, + payload_hash: String, + idempotency_key: String, } -fn parse_edition(value: &str) -> Result { +async fn insert_outbox( + tx: &Transaction<'_>, + tenant_id: &str, + row: &OutboxInsert, +) -> 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, attempt_count, next_available_unix + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,0,$7) + ON CONFLICT (tenant_id, idempotency_key) DO NOTHING", + &[ + &tenant_id, + &row.message_id, + &row.aggregate_id, + &row.aggregate_version, + &row.event_type, + &SCHEMA_VERSION, + &row.created_unix, + &row.payload_json, + &row.payload_hash, + &row.idempotency_key, + &STATUS_PENDING, + ], + ) + .await + .map_err(|error| format!("control plane insert outbox_message failed: {error}"))?; + Ok(()) +} + +async fn drain_once( + client: &mut Client, + tenant_id: &str, + owner: &str, + now_unix: i64, + keep: i64, + dispatch: F, +) -> Result +where + F: Fn(&OutboxMessage) -> Result, +{ + let claimed = claim_batch(client, tenant_id, owner, now_unix).await?; + 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, + keep, + ) + .await?; + processed += 1; + continue; + } + match dispatch(&message) { + Ok(evidence) => { + ack_processed(client, tenant_id, &message, &evidence, now_unix, keep).await?; + processed += 1; + } + Err(error) => { + fail_claimed(client, tenant_id, &message, now_unix, &error).await?; + } + } + } + Ok(processed) +} + +async fn claim_batch( + client: &mut Client, + tenant_id: &str, + owner: &str, + now_unix: i64, +) -> Result, String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane claim 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 lease_expires = now_unix.saturating_add(LEASE_SECONDS); + let rows = tx + .query( + "WITH picked AS ( + SELECT message_id FROM outbox_message + WHERE tenant_id = $1 + AND ( + (message_status = $2 AND next_available_unix <= $5) + OR (message_status = $3 AND COALESCE(lease_expires_unix, 0) <= $5) + ) + ORDER BY aggregate_id, aggregate_version, created_unix + FOR UPDATE SKIP LOCKED + LIMIT $6 + ) + UPDATE outbox_message AS message + SET message_status = $3, + lease_owner = $4, + lease_expires_unix = $7, + attempt_count = message.attempt_count + 1, + first_attempt_unix = COALESCE(message.first_attempt_unix, $5), + last_attempt_unix = $5 + FROM picked + WHERE message.tenant_id = $1 AND message.message_id = picked.message_id + RETURNING 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", + &[ + &tenant_id, + &STATUS_PENDING, + &STATUS_LEASED, + &owner, + &now_unix, + &CLAIM_BATCH, + &lease_expires, + ], + ) + .await + .map_err(|error| format!("control plane claim outbox failed: {error}"))?; + let messages = rows + .iter() + .map(|row| row_to_outbox(row, tenant_id)) + .collect(); + tx.commit() + .await + .map_err(|error| format!("control plane claim commit failed: {error}"))?; + Ok(messages) +} + +fn row_to_outbox(row: &tokio_postgres::Row, tenant_id: &str) -> OutboxMessage { + OutboxMessage { + message_id: row.get(0), + tenant_id: tenant_id.to_string(), + aggregate_id: row.get(1), + aggregate_version: row.get(2), + event_type: row.get(3), + schema_version: row.get(4), + created_unix: row.get(5), + payload_json: row.get(6), + payload_hash: row.get(7), + idempotency_key: row.get(8), + message_status: row.get(9), + lease_owner: row.get(10), + lease_expires_unix: row.get(11), + attempt_count: row.get(12), + first_attempt_unix: row.get(13), + last_attempt_unix: row.get(14), + next_available_unix: row.get(15), + terminal_reason: row.get(16), + } +} + +async fn receipt_exists( + client: &mut Client, + tenant_id: &str, + idempotency_key: &str, +) -> Result { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane receipt 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 row = tx + .query_opt( + "SELECT 1 FROM outbox_receipt WHERE tenant_id = $1 AND idempotency_key = $2", + &[&tenant_id, &idempotency_key], + ) + .await + .map_err(|error| format!("control plane load outbox_receipt failed: {error}"))?; + tx.commit() + .await + .map_err(|error| format!("control plane receipt commit failed: {error}"))?; + Ok(row.is_some()) +} + +async fn ack_processed( + client: &mut Client, + tenant_id: &str, + message: &OutboxMessage, + evidence: &str, + now_unix: i64, + keep: i64, +) -> Result<(), String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane ack 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}"))?; + tx.execute( + "INSERT INTO outbox_receipt ( + tenant_id, idempotency_key, message_id, processed_unix, receipt_evidence + ) VALUES ($1,$2,$3,$4,$5) + ON CONFLICT (tenant_id, idempotency_key) DO NOTHING", + &[ + &tenant_id, + &message.idempotency_key, + &message.message_id, + &now_unix, + &evidence, + ], + ) + .await + .map_err(|error| format!("control plane insert outbox_receipt failed: {error}"))?; + tx.execute( + "UPDATE outbox_message + SET message_status = $3, lease_owner = NULL, lease_expires_unix = NULL, + terminal_reason = NULL, next_available_unix = $4 + WHERE tenant_id = $1 AND message_id = $2", + &[ + &tenant_id, + &message.message_id, + &STATUS_PROCESSED, + &now_unix, + ], + ) + .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}"))?; + Ok(()) +} + +async fn fail_claimed( + client: &mut Client, + tenant_id: &str, + message: &OutboxMessage, + now_unix: i64, + error: &DispatchError, +) -> Result<(), String> { + let dead = outbox::should_dead_letter(message.attempt_count, error); + let status = if dead { + STATUS_DEAD_LETTER + } else { + STATUS_PENDING + }; + let next = if dead { + now_unix + } else { + outbox::next_available_unix(now_unix, message.attempt_count, &message.message_id) + }; + let reason = error.as_str(); + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane fail 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}"))?; + tx.execute( + "UPDATE outbox_message + SET message_status = $3, lease_owner = NULL, lease_expires_unix = NULL, + next_available_unix = $4, terminal_reason = $5 + WHERE tenant_id = $1 AND message_id = $2", + &[&tenant_id, &message.message_id, &status, &next, &reason], + ) + .await + .map_err(|error| format!("control plane fail outbox_message failed: {error}"))?; + tx.commit() + .await + .map_err(|error| format!("control plane fail commit failed: {error}"))?; + Ok(()) +} + +async fn outbox_health( + client: &mut Client, + tenant_id: &str, + now_unix: i64, +) -> Result { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane health 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 row = tx + .query_one( + "SELECT + COUNT(*) FILTER (WHERE message_status = $2), + COUNT(*) FILTER (WHERE message_status = $3), + COUNT(*) FILTER (WHERE message_status = $4), + MIN(created_unix) FILTER ( + WHERE message_status IN ($2, $3) + ) + FROM outbox_message WHERE tenant_id = $1", + &[ + &tenant_id, + &STATUS_PENDING, + &STATUS_LEASED, + &STATUS_DEAD_LETTER, + ], + ) + .await + .map_err(|error| format!("control plane outbox health failed: {error}"))?; + let oldest: Option = row.get(3); + tx.commit() + .await + .map_err(|error| format!("control plane health commit failed: {error}"))?; + Ok(OutboxHealth { + status: "ready".to_string(), + pending: row.get(0), + leased: row.get(1), + dead_letter: row.get(2), + oldest_age_seconds: oldest.map(|created| now_unix.saturating_sub(created).max(0)), + }) +} + +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 + .map_err(|error| format!("control plane 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 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 + FROM outbox_message WHERE tenant_id = $1 + ORDER BY CASE message_status + WHEN 'dead_letter' THEN 0 + WHEN 'pending' THEN 1 + WHEN 'leased' THEN 2 + ELSE 3 + END, + created_unix DESC, message_id DESC + LIMIT $2", + &[&tenant_id, &limit], + ) + .await + .map_err(|error| format!("control plane list outbox failed: {error}"))?; + let messages = rows + .iter() + .map(|row| row_to_outbox(row, tenant_id)) + .collect(); + tx.commit() + .await + .map_err(|error| format!("control plane list commit failed: {error}"))?; + Ok(messages) +} + +async fn replay_dead_letter( + client: &mut Client, + tenant_id: &str, + message_id: &str, + now_unix: i64, +) -> Result<(), String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane replay 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 updated = tx + .execute( + "UPDATE outbox_message + SET message_status = $3, lease_owner = NULL, lease_expires_unix = NULL, + attempt_count = 0, next_available_unix = $4, terminal_reason = NULL + WHERE tenant_id = $1 AND message_id = $2 AND message_status = $5", + &[ + &tenant_id, + &message_id, + &STATUS_PENDING, + &now_unix, + &STATUS_DEAD_LETTER, + ], + ) + .await + .map_err(|error| format!("control plane replay outbox failed: {error}"))?; + if updated != 1 { + tx.rollback() + .await + .map_err(|error| format!("control plane replay rollback failed: {error}"))?; + return Err(format!("outbox message {message_id} is not in dead_letter")); + } + tx.commit() + .await + .map_err(|error| format!("control plane replay commit failed: {error}"))?; + Ok(()) +} + +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() +} + +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) +} + +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", + } +} + +fn parse_edition(value: &str) -> Result { match value { "community" => Ok(ProductEdition::Community), "evaluation" => Ok(ProductEdition::Evaluation), @@ -795,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] @@ -814,14 +2334,19 @@ mod tests { "security_event", "audit_record", "threat_feed", + "outbox_message", + "outbox_receipt", "schema_migration", ] { assert!(MIGRATION_SQL.contains(table), "missing table {table}"); } 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)")); assert!( !MIGRATION_SQL.contains("json_blob"), "do not dump AppData as one JSON column" @@ -851,5 +2376,700 @@ 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_limited(LIST_LIMIT) + .await + .expect("list snapshot outbox"); + assert!( + messages + .iter() + .any(|message| message.event_type == EVENT_SNAPSHOT_REPLACED), + "snapshot persist must enqueue an outbox row" + ); + } + + fn test_database_url() -> Option { + std::env::var("CONTROL_PLANE_TEST_DATABASE_URL") + .ok() + .filter(|url| !url.trim().is_empty()) + } + + fn unique_tenant(label: &str) -> String { + format!( + "{label}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("clock after epoch") + .as_nanos() + ) + } + + fn sample_event(id: u64, path: &str) -> SecurityEvent { + SecurityEvent { + id, + timestamp_unix: 1_700_000_000, + client_ip: Some("198.51.100.20".parse().expect("documentation IP")), + route_id: Some("demo".into()), + action: "blocked".into(), + reason: "fixture".into(), + score: 80, + path: path.into(), + } + } + + #[tokio::test] + async fn postgres_appends_event_and_outbox_atomically() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-append"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("test database must accept the control plane"); + plane + .save(&AppData::seeded()) + .await + .expect("seed tenant snapshot"); + let setup_now = unix_now_i64().saturating_add(60); + let _ = plane + .drain_once("setup", setup_now, |_| Ok("setup".into())) + .await + .expect("ack snapshot outbox"); + let event = sample_event(7, "/gateway/login"); + plane + .append_security_event(&event, 1_000) + .await + .expect("append event + outbox"); + let loaded = plane.load().await.expect("load").expect("tenant exists"); + assert!( + loaded + .events + .iter() + .any(|row| row.id == 7 && row.path == "/gateway/login"), + "event must round-trip unmasked" + ); + assert_eq!(loaded.next_event_id, 8); + let messages = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list outbox"); + let recorded = messages + .iter() + .find(|message| message.event_type == EVENT_SECURITY_RECORDED) + .expect("security event outbox row"); + assert!(recorded.payload_json.contains("198.51.100.20")); + assert!(recorded.payload_json.contains("/gateway/login")); + assert_eq!(recorded.message_status, STATUS_PENDING); + plane + .append_security_event(&event, 1_000) + .await + .expect("idempotent retry of same event id"); + let again = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list after retry"); + assert_eq!( + again + .iter() + .filter(|message| message.event_type == EVENT_SECURITY_RECORDED) + .count(), + 1, + "duplicate event id must not enqueue a second outbox row" + ); + } + + #[tokio::test] + async fn postgres_outbox_worker_is_idempotent_and_dead_letters() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-worker"); + 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"); + plane + .append_security_event(&sample_event(1, "/one"), 100) + .await + .expect("enqueue"); + + let dispatched = std::sync::Arc::new(std::sync::Mutex::new(Vec::::new())); + let seen = dispatched.clone(); + let processed = plane + .drain_once("worker-a", now, move |message| { + seen.lock() + .expect("dispatcher lock") + .push(message.message_id.clone()); + Ok(format!("ack:{}", message.payload_hash)) + }) + .await + .expect("first drain"); + assert_eq!(processed, 1); + assert_eq!(dispatched.lock().expect("dispatcher lock").len(), 1); + + let processed_again = plane + .drain_once("worker-a", now.saturating_add(30), |_| { + panic!("processed messages must not be claimed again") + }) + .await + .expect("second drain"); + assert_eq!(processed_again, 0); + + plane + .append_security_event(&sample_event(2, "/poison"), 100) + .await + .expect("poison enqueue"); + let _ = plane + .drain_once("worker-a", now.saturating_add(60), |_| { + Err(crate::outbox::DispatchError::Permanent("malformed".into())) + }) + .await + .expect("dead-letter drain"); + let health = plane + .outbox_health(now.saturating_add(60)) + .await + .expect("health"); + assert_eq!(health.status, "ready"); + assert_eq!(health.dead_letter, 1); + + let dead = plane + .list_outbox_limited(LIST_LIMIT) + .await + .expect("list") + .into_iter() + .find(|message| message.message_status == STATUS_DEAD_LETTER) + .expect("dead letter row"); + plane + .replay_dead_letter(&dead.message_id, now.saturating_add(90)) + .await + .expect("authorized replay"); + let replayed = plane + .drain_once( + "worker-b", + now.saturating_add(90), + |_| Ok("replayed".into()), + ) + .await + .expect("replay drain"); + assert_eq!(replayed, 1); + } + + #[tokio::test] + async fn postgres_expired_lease_is_reclaimed_and_skip_locked_is_exclusive() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-lease"); + let plane_a = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("plane a"); + let plane_b = PostgresPlane::connect_tenant(&url, &tenant) + .await + .expect("plane b"); + plane_a.save(&AppData::seeded()).await.expect("seed"); + let now = unix_now_i64().saturating_add(60); + let _ = plane_a + .drain_once("setup", now, |_| Ok("setup".into())) + .await + .expect("ack snapshot outbox"); + plane_a + .append_security_event(&sample_event(3, "/lease"), 100) + .await + .expect("enqueue"); + + let first = plane_a + .drain_once("worker-a", now, |_| { + Err(crate::outbox::DispatchError::Transient("timeout".into())) + }) + .await + .expect("lease then fail transient"); + assert_eq!(first, 0); + 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) + .expect("event still queued"); + assert_eq!(pending.message_status, STATUS_PENDING); + assert!(pending.next_available_unix > now); + + let later = pending.next_available_unix; + let reclaimed = plane_b + .drain_once("worker-b", later, |_| Ok("reclaimed".into())) + .await + .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 df89626..6fd6c27 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,6 +42,7 @@ mod credentials; mod destination; mod misp_import; mod opencti_import; +mod outbox; mod proven_engine; mod stix_import; mod suricata_eve; @@ -136,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(), @@ -345,8 +348,44 @@ impl AppState { proven_engine: self.proven_engine.mode().to_string(), proven_engine_fail_closed: self.proven_engine.fail_closed, destination_mode: self.destination.mode().to_string(), + outbox: if self.control_plane.is_some() { + "ready".to_string() + } else { + "disabled".to_string() + }, + outbox_pending: 0, + 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, } } + + async fn health_status_live(&self) -> HealthStatus { + let mut health = self.health_status(); + let Some(plane) = &self.control_plane else { + return health; + }; + match plane.outbox_health(now_unix() as i64).await { + Ok(stats) => { + health.outbox = stats.status; + health.outbox_pending = stats.pending; + health.outbox_leased = stats.leased; + health.outbox_dead_letter = stats.dead_letter; + health.outbox_oldest_age_seconds = stats.oldest_age_seconds; + } + Err(_) => health.outbox = "error".to_string(), + } + if let Ok(count) = plane.event_partition_count().await { + health.event_partitions = count; + } + health + } } #[derive(Debug, Clone)] @@ -473,6 +512,16 @@ pub struct HealthStatus { pub proven_engine_fail_closed: bool, /// `production` (fail-closed classes) or `development` (loopback class permitted). pub destination_mode: String, + /// `ready` when the PostgreSQL outbox is the authority; `disabled` on file/memory. + pub outbox: String, + pub outbox_pending: i64, + 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"; @@ -545,6 +594,10 @@ pub fn build_app(state: AppState) -> Router { .route("/api/dnsbl", get(list_dnsbl).post(create_dnsbl)) .route("/api/events", get(list_events)) .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)) @@ -907,7 +960,7 @@ async fn soc_analyze( } async fn healthz(State(state): State) -> Json { - Json(state.health_status()) + Json(state.health_status_live().await) } /// Build/version metadata for deployment verification. @@ -1092,6 +1145,208 @@ async fn list_audit_logs(State(state): State, headers: HeaderMap) -> R Json(state.inner.read().await.audit_logs.clone()).into_response() } +#[derive(Serialize)] +struct OutboxListView { + status: String, + limit: usize, + messages: Vec, +} + +async fn list_outbox(State(state): State, headers: HeaderMap) -> Response { + 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_limited(limit as i64).await { + Ok(messages) => Json(OutboxListView { + status: "ready".to_string(), + limit, + messages, + }) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +async fn replay_outbox( + State(state): State, + PathParam(message_id): PathParam, + 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, + "outbox replay requires the PostgreSQL control plane", + ); + }; + let actor = audit_actor(&state, &headers); + if let Err(message) = plane + .replay_dead_letter(&message_id, now_unix() as i64) + .await + { + return error(StatusCode::BAD_REQUEST, message); + } + match state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + "replay_outbox", + "outbox_message", + message_id.clone(), + ); + }) + .await + { + Ok(()) => Json(serde_json::json!({ + "status": "pending", + "message_id": message_id + })) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +#[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())) @@ -2161,7 +2416,7 @@ async fn support_bundle(State(state): State) -> Json { let generated_at_unix = now_unix(); Json(SupportBundle { generated_at_unix, - health: state.health_status(), + health: state.health_status_live().await, kpis: kpi_snapshot_at(&data, generated_at_unix), commercial: data.commercial.clone(), readiness: commercial_readiness_snapshot_at(&data, generated_at_unix), @@ -2569,29 +2824,37 @@ async fn record_event( let action = action.to_string(); let path = path.to_string(); let event_limit = state.event_limit; - if let Err(error) = state - .mutate_and_persist(|data| { - let id = data.next_event_id; - data.next_event_id += 1; - let event = SecurityEvent { - id, - timestamp_unix: now_unix(), - client_ip, - route_id, - action, - reason, - score, - path, - }; - // Structured stdout log line for SIEM / log-collector ingestion. - // ponytail: one println per recorded event — fine at gateway volumes; - // add async batching if event throughput ever becomes a bottleneck. - println!("{}", security_event_log_line(&event)); - data.events.push(event); - enforce_event_limit(data, event_limit); - }) - .await - { + let _guard = state.persist_lock.lock().await; + let (event, previous) = { + let mut data = state.inner.write().await; + let previous = data.clone(); + let id = data.next_event_id; + data.next_event_id += 1; + let event = SecurityEvent { + id, + timestamp_unix: now_unix(), + client_ip, + route_id, + action, + reason, + score, + path, + }; + data.events.push(event.clone()); + enforce_event_limit(&mut data, event_limit); + (event, previous) + }; + let persist = if let Some(plane) = &state.control_plane { + plane.append_security_event(&event, event_limit).await + } else { + // File/memory has no leased worker; emit the SIEM line on the request path. + println!("{}", security_event_log_line(&event)); + let snapshot = state.inner.read().await.clone(); + state.persist_snapshot(&snapshot).await + }; + if let Err(error) = persist { + let mut data = state.inner.write().await; + *data = previous; eprintln!("failed to persist security event: {error}"); } } @@ -3072,6 +3335,15 @@ 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…