diff --git a/CHANGELOG.md b/CHANGELOG.md index ebf4d129..33722771 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- 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. diff --git a/docs/architecture.md b/docs/architecture.md index a9b2bde1..d5671acf 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -59,7 +59,7 @@ flowchart LR - `WAF_IDS_STATE_PATH` enables JSON state persistence for standalone/loopback operation. Without it, the service uses seeded in-memory state. Production binds require PostgreSQL (`CONTROL_PLANE_DATABASE_URL`). - File-backed writes use temporary sibling files followed by atomic rename. Management API mutations roll back in memory if the state file cannot be replaced. - Block mode is route-scoped to avoid global accidental enforcement. -- JSON persistence is a baseline durability mechanism, not a substitute for a production database, backup plan, or audited change workflow. +- JSON persistence is a baseline durability mechanism, not a substitute for a production database. PostgreSQL mode exports a hashed logical snapshot (`GET /api/backup`) and runs an isolated restore drill (`POST /api/backup/drill`). - Commercial readiness is a runtime evidence model for buyer pilots, not a legal revenue recognition or compliance certification system. - The reusable core remains in-repo as a workspace crate. A git submodule is intentionally deferred until an independently versioned engine, SDK, or adapter needs a separate release lifecycle. diff --git a/docs/doctoring/outbox-workers.md b/docs/doctoring/outbox-workers.md index d7abaea5..e592f6fd 100644 --- a/docs/doctoring/outbox-workers.md +++ b/docs/doctoring/outbox-workers.md @@ -50,6 +50,6 @@ the exactly-once ack. Dead letters are never pruned. Loopback file/memory adapters keep in-process stdout SIEM and report `outbox=disabled`. Remaining: a -non-owner runtime role, backup/restore drill, HASH partitioning, and additional +non-owner runtime role, HASH partitioning, and additional consumers (TAXII poll, Clearfolio, contextual-orchestrator) on the same -message/receipt contract. +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 01ff5bf7..d35d826c 100644 --- a/docs/doctoring/postgres-control-plane.md +++ b/docs/doctoring/postgres-control-plane.md @@ -41,6 +41,23 @@ https://doi.org/10.6028/NIST.SP.800-218 Set `CONTROL_PLANE_DATABASE_URL` (or credentials-file key `control_plane_url`) 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. Remaining: -non-owner runtime role, backup/restore drill, HASH partitioning for -`security_event`, optimistic concurrency. +Loopback still uses `WAF_IDS_STATE_PATH` or in-memory state. + +`GET /api/backup` (admin read) exports a hashed logical snapshot. `POST /api/backup` +restores after schema-version and payload-hash checks. `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. + +Remaining: non-owner runtime role, HASH partitioning for `security_event`, +optimistic concurrency. diff --git a/docs/papers/nist-sp-800-34r1-contingency-planning.pdf b/docs/papers/nist-sp-800-34r1-contingency-planning.pdf new file mode 100644 index 00000000..38cbc717 Binary files /dev/null and b/docs/papers/nist-sp-800-34r1-contingency-planning.pdf differ diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 3674d57e..4c51d86b 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,6 +1,6 @@ # Product and technical gap baseline -Snapshot date: 2026-08-23T17:35Z (exact-head inventory of then-open GitHub PRs +Snapshot date: 2026-08-23T18: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,10 +25,11 @@ not “waiting on review/CI time”. | PR | Title | Head | Checks | Reviews | Merge blocker | | --- | --- | --- | --- | --- | --- | -| [#101](https://github.com/ContextualWisdomLab/wardnet/pull/101) | feat(store): bound outbox listing and prune processed rows | `feat/issue-81-outbox-retention` stacked on #100 | local fmt/test/clippy + two `/healthz` smokes; live `postgres_outbox_list_is_bounded_and_prunes_processed` | Author this pass | 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 or the outbox first slice. | +| [#102](https://github.com/ContextualWisdomLab/wardnet/pull/102) | feat(store): logical backup and isolated restore drill | `feat/issue-80-backup-restore` stacked on #101 | local fmt/test/clippy + two `/healthz` smokes; live `postgres_backup_restore_drill_preserves_unmasked_invariants` | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 then #101 first. Do not `--admin`. Do not re-implement rustls, outbox, or retention. | +| [#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 (4 threads: unbounded list still-valid; 3 info) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 first. Do not `--admin`. Do not re-implement the postgres gate. Bounded `list_outbox` is #81 remainder, not this TLS slice. | -| [#98](https://github.com/ContextualWisdomLab/wardnet/pull/98) | feat(store): require PostgreSQL as the production control plane | `ea621985e276` (`feat/issue-80-postgres-control-plane`) stacked on #97 | rust + fuzz green at last snapshot; Devin 7 threads (full-snapshot rewrite, ORDER BY, TLS, RLS owner, reconnect) | Author this pass; Devin COMMENTED | Org 2-approval + self-author. ORDER BY + incremental event persist addressed on #99. rustls this pass; remaining non-owner role / backup are #80 remainder. Do not `--admin`. | +| [#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. | @@ -56,8 +57,8 @@ by ruleset `18156473` (not by failing Checks). Do not `--admin` merge. | [#84](https://github.com/ContextualWisdomLab/wardnet/issues/84) | [P1] Build an immutable signed release, promotion, and rollback pipeline | high | | [#83](https://github.com/ContextualWisdomLab/wardnet/issues/83) | [P1] Add bounded distributed admission control, trusted client attribution, and overload behavior | high | | [#82](https://github.com/ContextualWisdomLab/wardnet/issues/82) | [P1] Integrate Keyverse identity, tenant authorization, consent, and human approval evidence | high (blocked) | -| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice on #99; bounded list/retention remainder** | -| [#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 this pass; backup remainder** | +| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice on #99; bounded list/retention on #101** | +| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical — gate on #98; rustls on #100; backup/restore this pass; non-owner remainder** | | [#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 | @@ -93,7 +94,7 @@ Management auth is shared secrets (`X-Admin-Token`) plus optional multi-token RBAC. Keyverse (OIDC/SCIM/FIDO2) is not wired. Fail-closed (#78) is the prerequisite shipped on PR #94. -### Durable control plane (issue #80) — **production gate on #98; rustls this pass** +### Durable control plane (issue #80) — **production gate on #98; rustls on #100; backup 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 @@ -102,10 +103,15 @@ transaction. JSON file / memory remain loopback/community only. `/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. Remaining: non-owner role, backup/restore drill, event HASH -partitioning, optimistic concurrency. +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. +Remaining: non-owner runtime role, event HASH partitioning, optimistic +concurrency. Physical/PITR backups stay a DBA concern. -### Transactional outbox (issue #81) — **first slice on #99** +### 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. @@ -117,9 +123,9 @@ 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; dead letters stay. -Remaining consumers: TAXII poll, Clearfolio, contextual-orchestrator on the same -contract. +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** @@ -148,7 +154,8 @@ replays `security_event.recorded` as stdout SIEM with receipts. | Figma Code Connect | Not used | | Ten UI-UX areas | Inventoried in `docs/ui-ux/storybook-scene-inventory.md` | | Node Storybook | **Not hosted in `/admin`** (embedded-console architecture). File:// inventory is the scene/edge-case contract this pass. | -| Outbox card | Embedded `/admin` Outbox section this pass | +| Outbox card | Embedded `/admin` Outbox section | +| Backup card | Embedded `/admin` Control-plane backup section this pass | ### CSAP / SOC 2 vs PII unmasking @@ -160,8 +167,8 @@ future encryption-at-rest. ### Coverage / docstring bar Org 100% line/branch/docstring applies to **changed** surfaces this loop -(outbox schema, claim/ack/dead-letter/replay, incremental event persist, -health/API, admin card). Remaining holes on untouched handlers stay listed +(backup export/verify/restore/drill, health/API, admin card, EVENT_LIMIT +prune on save/ack). Remaining holes on untouched handlers stay listed for later loops. ### Ecosystem connectors (leverage order) @@ -176,16 +183,18 @@ for later loops. ## This loop’s shipped gap -Issue **#81** still-valid #99 finding: bounded `GET /api/outbox` (`EVENT_LIMIT`) -and prune of processed `outbox_message` rows (receipts and dead letters stay). -Stacked on #100 rustls. Do not re-implement #78, sidecar, pin, libcoraza, the -postgres gate, the #81 first outbox slice, or rustls. +Issue **#80** backup/restore remainder: hashed logical snapshot export, +fail-closed restore, and isolated restore drill stacked on #101. Also the +still-valid #101 prune-cap fix (`EVENT_LIMIT` on save and ack). Do not +re-implement #78, sidecar, pin, libcoraza, the postgres gate, the #81 first +outbox slice, rustls, or bounded list/retention. ## Next hourly loop (do, do not report) 1. Second independent APPROVE on #91/#92. Do not `--admin`. -2. Keep #94/#95/#96/#97/#98/#99/#100 and this retention PR merge-ready. Merge - order #95 then #96 then #97 then #98 then #99 then #100 then this. -3. Next runtime gap if policy still blocks: backup/restore remainder of #80, - or additional #81 consumers (TAXII / Clearfolio / orchestrator). +2. Keep #94/#95/#96/#97/#98/#99/#100/#101 and #102 merge-ready. Merge + order #94 independently; #95 then #96 then #97 then #98 then #99 then #100 + then #101 then #102. +3. Next runtime gap if policy still blocks: non-owner runtime role remainder of + #80, or additional #81 consumers (TAXII / Clearfolio / orchestrator). 4. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. diff --git a/docs/runbooks/operations.md b/docs/runbooks/operations.md index ac081326..380311b9 100644 --- a/docs/runbooks/operations.md +++ b/docs/runbooks/operations.md @@ -67,6 +67,32 @@ 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 + +## 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 +126,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 f1103312..be7a2ea0 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; `sslmode=require` uses rustls | Backup/restore drill, 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; `sslmode=require` uses rustls; hashed logical backup plus isolated restore drill (`GET /api/backup`, `POST /api/backup/drill`) | Non-owner runtime role, 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/src/control_plane.rs b/src/control_plane.rs index 0beab68b..9e18660f 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -10,8 +10,10 @@ use crate::outbox::{ LEASE_SECONDS, LIST_LIMIT, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, STATUS_LEASED, STATUS_PENDING, STATUS_PROCESSED, }; +use serde::{Deserialize, Serialize}; use std::net::IpAddr; use std::str::FromStr; +use std::time::Instant; use tokio::sync::Mutex; use tokio_postgres::{Client, GenericClient, NoTls, Transaction}; use tokio_postgres_rustls::MakeRustlsConnect; @@ -24,6 +26,10 @@ use waf_ids_core::{ pub const DEFAULT_TENANT_ID: &str = "local-lab"; const MIGRATION_VERSION: i32 = 2; +/// 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; /// Recoverable forward migration. Two-word snake_case names, 3NF, RLS. pub const MIGRATION_SQL: &str = r#" @@ -475,6 +481,153 @@ impl PostgresPlane { 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 != MIGRATION_VERSION { + return Err(format!( + "backup schema_version {} is unsupported; expected {MIGRATION_VERSION}", + self.schema_version + )); + } + if self.tenant_id.trim().is_empty() { + return Err("backup tenant_id must be non-empty".to_string()); + } + let expected = outbox::payload_hash(&self.unsigned_json()?); + if expected != self.payload_hash { + return Err("backup payload_hash does not match contents".to_string()); + } + Ok(()) + } + + fn semantic_hash(&self) -> Result { + let mut snapshot = self.snapshot.clone(); + snapshot.commercial.tenant_id.clear(); + let mut outbox: Vec<_> = self + .outbox + .iter() + .map(|message| { + ( + message.idempotency_key.clone(), + message.payload_hash.clone(), + message.message_status.clone(), + message.payload_json.clone(), + ) + }) + .collect(); + outbox.sort(); + let mut receipts: Vec<_> = self + .receipts + .iter() + .map(|row| (row.idempotency_key.clone(), row.receipt_evidence.clone())) + .collect(); + receipts.sort(); + let body = serde_json::json!({ + "snapshot": snapshot, + "outbox": outbox, + "receipts": receipts, + }) + .to_string(); + Ok(outbox::payload_hash(&body)) + } } async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result, String> { @@ -543,6 +696,21 @@ async fn save_snapshot( .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) @@ -723,13 +891,6 @@ async fn save_snapshot( .await .map_err(|error| format!("control plane insert threat_feed failed: {error}"))?; } - - 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(()) } @@ -1514,6 +1675,187 @@ async fn replay_dead_letter( 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) @@ -2104,4 +2446,117 @@ mod tests { "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 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" + ); + } } diff --git a/src/lib.rs b/src/lib.rs index e28b522a..1af222a1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -357,6 +357,11 @@ impl AppState { outbox_leased: 0, outbox_dead_letter: 0, outbox_oldest_age_seconds: None, + backup: if self.control_plane.is_some() { + "ready".to_string() + } else { + "disabled".to_string() + }, } } @@ -509,6 +514,8 @@ pub struct HealthStatus { pub outbox_leased: i64, pub outbox_dead_letter: i64, pub outbox_oldest_age_seconds: Option, + /// `ready` when PostgreSQL logical backup/restore is the authority; `disabled` on file/memory. + pub backup: String, } const PHISHING_DATABASE_DEFAULT_FEED_ID: &str = "phishing-database-active"; @@ -583,6 +590,8 @@ pub fn build_app(state: AppState) -> Router { .route("/api/audit-logs", get(list_audit_logs)) .route("/api/outbox", get(list_outbox)) .route("/api/outbox/{message_id}/replay", post(replay_outbox)) + .route("/api/backup", get(get_backup).post(restore_backup)) + .route("/api/backup/drill", post(backup_drill)) .route("/api/events.ndjson", get(events_ndjson)) .route("/api/kpis", get(kpis)) .route("/api/signatures", get(list_signatures)) @@ -1203,6 +1212,135 @@ async fn replay_outbox( } } +#[derive(Serialize)] +struct BackupView { + status: String, + rpo: String, + rto_budget_ms: u64, + artifact: Option, +} + +async fn get_backup(State(state): State, headers: HeaderMap) -> Response { + if !admin_authenticated(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return Json(BackupView { + status: "disabled".to_string(), + rpo: control_plane::BACKUP_RPO.to_string(), + rto_budget_ms: control_plane::BACKUP_RTO_BUDGET_MS, + artifact: None, + }) + .into_response(); + }; + match plane.logical_backup().await { + Ok(artifact) => Json(BackupView { + status: "ready".to_string(), + rpo: control_plane::BACKUP_RPO.to_string(), + rto_budget_ms: control_plane::BACKUP_RTO_BUDGET_MS, + artifact: Some(artifact), + }) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +async fn restore_backup( + State(state): State, + headers: HeaderMap, + Json(backup): Json, +) -> Response { + if !admin_authorized(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return error( + StatusCode::SERVICE_UNAVAILABLE, + "backup restore requires the PostgreSQL control plane", + ); + }; + if let Err(message) = backup.verify() { + return error(StatusCode::BAD_REQUEST, message); + } + if let Err(message) = plane.restore_logical_backup(&backup).await { + return error(StatusCode::BAD_REQUEST, message); + } + match plane.load().await { + Ok(Some(loaded)) => { + *state.inner.write().await = loaded; + } + Ok(None) => { + return error( + StatusCode::INTERNAL_SERVER_ERROR, + "restore committed but tenant snapshot is empty", + ); + } + Err(message) => return error(StatusCode::INTERNAL_SERVER_ERROR, message), + } + let actor = audit_actor(&state, &headers); + match state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + "restore_backup", + "control_plane_backup", + backup.payload_hash.clone(), + ); + }) + .await + { + Ok(_) => Json(serde_json::json!({ + "status": "restored", + "schema_version": backup.schema_version, + "payload_hash": backup.payload_hash, + })) + .into_response(), + Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + } +} + +async fn backup_drill(State(state): State, headers: HeaderMap) -> Response { + if !admin_authorized(&state, &headers) { + return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); + } + let Some(plane) = &state.control_plane else { + return error( + StatusCode::SERVICE_UNAVAILABLE, + "backup drill requires the PostgreSQL control plane", + ); + }; + let report = match plane.restore_drill().await { + Ok(report) => report, + Err(message) => return error(StatusCode::INTERNAL_SERVER_ERROR, message), + }; + let actor = audit_actor(&state, &headers); + let outcome = if report.passed { + "backup_drill" + } else { + "backup_drill_failed" + }; + if let Err(message) = state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + outcome, + "control_plane_backup", + report.source_hash.clone(), + ); + }) + .await + { + return error(StatusCode::INTERNAL_SERVER_ERROR, message); + } + if report.passed { + (StatusCode::OK, Json(report)).into_response() + } else { + (StatusCode::INTERNAL_SERVER_ERROR, Json(report)).into_response() + } +} + async fn kpis(State(state): State) -> Json { let data = state.inner.read().await; Json(kpi_snapshot_at(&data, now_unix())) @@ -3143,6 +3281,14 @@ input,select{font:inherit;min-height:44px;padding:0 12px;border:1px solid var(--

Audit log

Loading…

Outbox

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

Loading…
+

Control-plane backup

+

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

+
Loading…
+
+ +
+ 

+    

     
Evidence manifest
Loading…

     
SOC event export (ndjson)
Loading…