From ea621985e27691a6b80629b36888e4defcc361c4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 24 Aug 2026 01:20:14 +0900 Subject: [PATCH 1/3] feat(store): require PostgreSQL as the production control plane Non-loopback binds fail closed without CONTROL_PLANE_DATABASE_URL. Migrations create 3NF two-word tables with default-deny RLS. Snapshot persist commits policy rows and audit records in one transaction. Loopback still uses the JSON file or memory adapter. --- .github/workflows/ci.yml | 16 + CHANGELOG.md | 1 + Cargo.lock | 343 ++++++++- Cargo.toml | 1 + README.md | 3 +- docs/architecture.md | 3 +- docs/doctoring/postgres-control-plane.md | 43 ++ docs/product-technical-gap-baseline.md | 53 +- docs/security/threat-model.md | 2 +- scripts/smoke.sh | 1 + src/control_plane.rs | 855 +++++++++++++++++++++++ src/credentials.rs | 18 +- src/lib.rs | 72 +- tests/binary.rs | 58 ++ 14 files changed, 1426 insertions(+), 43 deletions(-) create mode 100644 docs/doctoring/postgres-control-plane.md create mode 100644 src/control_plane.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index df092755..2b64f2ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,6 +11,22 @@ permissions: jobs: rust: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: wardnet + POSTGRES_PASSWORD: wardnet + POSTGRES_DB: wardnet + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U wardnet -d wardnet" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + env: + CONTROL_PLANE_TEST_DATABASE_URL: postgres://wardnet:wardnet@localhost:5432/wardnet steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 - uses: dtolnay/rust-toolchain@4be7066ada62dd38de10e7b70166bc74ed198c30 # stable diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c1ef4d2..0070ff0d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,5 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Production (non-loopback) binds fail closed without `CONTROL_PLANE_DATABASE_URL`. PostgreSQL is the production control-plane authority (3NF two-word tables, default-deny row-level security, snapshot persist in one transaction). Loopback still uses the JSON file / memory adapter. `/healthz.persistence` reports `postgres`, `file`, or `memory`. The URL is a secret and is bootstrapped into the credential registry. - Live `/gateway` transactions consult in-process libcoraza when `CORAZA_LIB_PATH` is set (with `CORAZA_RULES_PATH` and/or `CORAZA_DIRECTIVES`). Missing library, missing rules, or an empty ruleset fail startup before bind. Otherwise a Coraza sidecar is consulted when `CORAZA_WAF_URL` is set. The sidecar response is parsed with the existing Coraza audit adapter (OWASP CRS authority, not a hand-rolled engine). Engine outage is fail-closed when `PROVEN_ENGINE_FAIL_CLOSED` is true. `GET /api/waf/engine-status` and `/healthz.proven_engine` report `coraza_in_process`, `coraza_sidecar`, or `ingest_hints_only`. - Fail-closed destination policy on every outbound `http`/`https` call, including the Coraza sidecar URL (gateway upstream, threat-intel fetch, Clearfolio, SOC LLM). Private/loopback/metadata classes are denied unless `DESTINATION_ALLOWLIST` (or loopback development) permits them; `DESTINATION_DENYLIST` wins. Clients ignore ambient HTTP proxies and do not follow redirects. CIDR allowlist matches apply per resolved address; CIDR entries authorize non-default ports; IPv6 site-local (`fec0::/10`) is denied; invalid CIDR prefixes fail startup; `/healthz.destination_mode` reports the policy class. Blocking DNS runs on `spawn_blocking` with a 2s timeout. Persistence and destination-list validation complete before the readiness line is printed. After a host is allowed, the HTTP client connects only to those evaluated addresses (Host/SNI unchanged) so a rebinding answer cannot bypass the policy. diff --git a/Cargo.lock b/Cargo.lock index cb42a020..a05a89e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,17 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -93,12 +104,27 @@ version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b588b76d00fde79687d7646a9b5bdf3cc0f655e0bbd080335a95d7e96f3587da" +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "bumpalo" version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.12.1" @@ -138,6 +164,18 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "cmov" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a" + +[[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.3.0" @@ -147,6 +185,36 @@ dependencies = [ "libc", ] +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "ctutils" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d5515a3834141de9eafb9717ad39eea8247b5674e6066c404e8c4b365d2a29e" +dependencies = [ + "cmov", +] + +[[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", + "ctutils", +] + [[package]] name = "displaydoc" version = "0.2.7" @@ -168,6 +236,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "fallible-iterator" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443176a9f2c162692bd3d352d745ef9413eec5782a80d8fd6f8a1ac692a07f7" + [[package]] name = "fastrand" version = "2.5.0" @@ -202,6 +276,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "262590f4fe6afeb0bc83be1daa64e52657fe185690a958af7f3ad0e92085c5ae" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -264,7 +339,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "wasm-bindgen", ] @@ -294,6 +369,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "hmac" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6303bc9732ae41b04cb554b844a762b4115a61bfaa81e3e83050991eeb56863f" +dependencies = [ + "digest", +] + [[package]] name = "http" version = "1.5.0" @@ -339,6 +423,15 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "1.11.0" @@ -541,6 +634,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "libredox" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" +dependencies = [ + "libc", +] + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -553,6 +655,15 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.33" @@ -571,6 +682,16 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest", +] + [[package]] name = "memchr" version = "2.8.3" @@ -600,7 +721,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -613,24 +734,113 @@ dependencies = [ "autocfg", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + [[package]] name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "percent-encoding" version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "phf" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c1562dc717473dbaa4c1f85a36410e03c047b2e7df7f45ee938fbef64ae7fadf" +dependencies = [ + "phf_shared", + "serde", +] + +[[package]] +name = "phf_shared" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e57fef6bc5981e38c2ce2d63bfa546861309f875b8a75f092d1d54ae2d64f266" +dependencies = [ + "siphasher", +] + [[package]] name = "pin-project-lite" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "postgres-protocol" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08808e3c483c46e999108051c78334f473d5adb59d78bb80a1268c7e6aa6c514" +dependencies = [ + "base64", + "byteorder", + "bytes", + "fallible-iterator", + "hmac", + "md-5", + "memchr", + "rand 0.10.2", + "sha2", + "stringprep", +] + +[[package]] +name = "postgres-types" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "851ca9db4932932d69f3ea811b1abe63087a0f740a47692619dd40d4899b68be" +dependencies = [ + "bytes", + "fallible-iterator", + "postgres-protocol", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -824,6 +1034,15 @@ dependencies = [ "rand_core 0.9.5", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "regex-syntax" version = "0.8.11" @@ -964,6 +1183,12 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "serde" version = "1.0.229" @@ -1030,6 +1255,17 @@ dependencies = [ "serde", ] +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1046,6 +1282,12 @@ dependencies = [ "libc", ] +[[package]] +name = "siphasher" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" + [[package]] name = "slab" version = "0.4.12" @@ -1074,6 +1316,17 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "stringprep" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b4df3d392d81bd458a8a621b8bffbd2302a12ffe288a9d931670948749463b1" +dependencies = [ + "unicode-bidi", + "unicode-normalization", + "unicode-properties", +] + [[package]] name = "subtle" version = "2.6.1" @@ -1207,6 +1460,32 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio-postgres" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a528f7d280f6d5b9cd149635c8705b0dd049754bc67d81d31fa25169a93809d3" +dependencies = [ + "async-trait", + "byteorder", + "bytes", + "fallible-iterator", + "futures-channel", + "futures-util", + "log", + "parking_lot", + "percent-encoding", + "phf", + "pin-project-lite", + "postgres-protocol", + "postgres-types", + "rand 0.10.2", + "socket2", + "tokio", + "tokio-util", + "whoami", +] + [[package]] name = "tokio-rustls" version = "0.26.4" @@ -1226,6 +1505,7 @@ dependencies = [ "bytes", "futures-core", "futures-sink", + "libc", "pin-project-lite", "tokio", ] @@ -1302,6 +1582,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "unarray" version = "0.1.4" @@ -1314,12 +1600,33 @@ version = "2.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" +[[package]] +name = "unicode-bidi" +version = "0.3.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c1cb5db39152898a79168971543b1cb5020dff7fe43c8dc468b0885f5e29df5" + [[package]] name = "unicode-ident" version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-normalization" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fd4f6878c9cb28d874b009da9e8d183b5abc80117c40bbd187a1fde336be6e8" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-properties" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d" + [[package]] name = "untrusted" version = "0.9.0" @@ -1356,6 +1663,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-postgres", "tower", "waf-ids-core", ] @@ -1394,6 +1702,15 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" +[[package]] +name = "wasi" +version = "0.14.7+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c" +dependencies = [ + "wasip2", +] + [[package]] name = "wasip2" version = "1.0.4+wasi-0.2.12" @@ -1403,6 +1720,15 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasite" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42" +dependencies = [ + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "wasm-bindgen" version = "0.2.126" @@ -1500,6 +1826,19 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "wasite", + "web-sys", +] + [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 48e4d416..88185c4d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ serde = { version = "1", features = ["derive"] } 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"] } [dev-dependencies] tower = { version = "0.5", features = ["util"] } diff --git a/README.md b/README.md index 1d7bb12d..47b44669 100644 --- a/README.md +++ b/README.md @@ -67,7 +67,8 @@ Useful environment variables: - `BIND_ADDR`: listen address, default `127.0.0.1:8080` - `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. When omitted, the service runs with seeded in-memory state. +- `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. - `DNSBL_ORIGIN`: DNSBL zone origin, default `dnsbl.local` - `EVENT_LIMIT`: retained event count, default `1000`; must be greater than zero - `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`. diff --git a/docs/architecture.md b/docs/architecture.md index c38c5797..b998080b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,6 +29,7 @@ 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/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. @@ -54,7 +55,7 @@ flowchart LR - Default bind address is localhost. - Remote management requires `ADMIN_TOKEN` plus external TLS and identity controls. -- `WAF_IDS_STATE_PATH` enables JSON state persistence for standalone operation. Without it, the service uses seeded in-memory state. +- `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. diff --git a/docs/doctoring/postgres-control-plane.md b/docs/doctoring/postgres-control-plane.md new file mode 100644 index 00000000..551ac4df --- /dev/null +++ b/docs/doctoring/postgres-control-plane.md @@ -0,0 +1,43 @@ +# Doctoring — PostgreSQL control plane + +This note grounds issue #80 (PostgreSQL is the production authority; the JSON +file adapter remains loopback/community only). IEEE/ACM PDFs are not +redistributed. + +## Adopted standards and literature + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: Row +security policies*. https://www.postgresql.org/docs/18/ddl-rowsecurity.html + +- **Design impact:** Every tenant table uses `ENABLE` + `FORCE ROW LEVEL + SECURITY` and a default-deny policy keyed on `wardnet.tenant_id`. Missing + tenant context yields no rows. + +PostgreSQL Global Development Group. (2026). *PostgreSQL documentation: +Constraints*. https://www.postgresql.org/docs/current/ddl-constraints.html + +- **Design impact:** Primary keys include `tenant_id`. Foreign keys point at + `tenant_account`. Two-word snake_case names. + +PostgreSQL Global Development Group. (2026). *PostgreSQL documentation: +Transaction isolation*. https://www.postgresql.org/docs/current/transaction-iso.html + +- **Design impact:** Snapshot replace (routes, indicators, DNSBL, events, audit) + commits in one transaction so a policy mutation cannot land without its audit + records. + +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. + +## 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. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index d6e0f4a5..8c3715c0 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:01Z (exact-head inventory of then-open GitHub PRs +Snapshot date: 2026-08-23T16:20Z (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 @@ -92,11 +92,15 @@ 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) +### Durable control plane (issue #80) — **production gate + RLS snapshot this pass** -Optional JSON file + atomic rename. Not PostgreSQL, no tenant isolation, no -migrations, no hot-partition strategy. 3NF/snake_case two-word names apply when -the store lands. +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. ### Fail-closed credentials (issue #78) — **closed on PR #94** @@ -150,30 +154,21 @@ loops. ## This loop’s shipped gap -Issue **#86** in-process libcoraza remainder ([#97](https://github.com/ContextualWisdomLab/wardnet/pull/97), stacked on #96). -Operator-visible: `CORAZA_LIB_PATH` + `CORAZA_RULES_PATH`/`CORAZA_DIRECTIVES`; -`/healthz.proven_engine=coraza_in_process`; `GET /api/waf/engine-status` -`in_process_configured` / `in_process_rules`; missing library fails before -readiness (`tests/binary.rs::binary_fail_closes_when_libcoraza_path_is_missing`). -Driving tests: `gateway_blocks_live_request_from_in_process_libcoraza` and -`stub_engine_blocks_crs_probe_and_allows_clean`. Two real smokes this hour: -default `/healthz` + `/admin` (`ingest_hints_only`); stub-loaded `/healthz` + -`/api/commercial/readiness` (`coraza_in_process`, `target_sale_value_krw` -2_000_000_000). Do not re-implement #78, the #86 sidecar slice, or the #79 pin. +Issue **#80** first slice (PostgreSQL production authority). Non-loopback binds +fail closed without `CONTROL_PLANE_DATABASE_URL`. Operator-visible: +`/healthz.persistence=postgres`; credentials key `control_plane_url`. Driving +tests: `run_from_env_fail_closes_public_bind_without_postgres`, +`binary_fail_closes_non_loopback_listen_without_postgres`, +`binary_fail_closes_when_control_plane_url_is_not_postgres`, +`postgres_roundtrip_seeded_snapshot_when_database_url_is_set` (CI postgres +service). Do not re-implement #78, the #86 sidecar/libcoraza slices, or the +#79 pin. ## Next hourly loop (do, do not report) -1. Second independent APPROVE on #91/#92 (Copilot re-requested; still 1/2; - `--auto` already enabled). Merge if exact-HEAD second independent APPROVE - exists. Do not `--admin`. -2. Keep #94/#95/#96/#97 merge-ready. Merge order #95 then #96 then #97. Do - not re-implement #78, the #86 sidecar, or the #79 pin. -3. Strix FAILURE on #72/#77/#93 is org LiteLLM provider infra, not wardnet - code; do not rotate keys. Watch ContextualWisdomLab/.github branch - `codex/strix-fail-closed-provider-evidence`. -4. Sticky opencode `CHANGES_REQUESTED` on #72 head `6881f47` — review job does - not post APPROVE. -5. Next runtime gap if policy still blocks: #80 durable PostgreSQL control - plane, or #81 outbox/workers, or #86 detection-quality corpora / Suricata - tail. -6. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. +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. +4. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. diff --git a/docs/security/threat-model.md b/docs/security/threat-model.md index 5c843271..b57968ed 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 | Database, backup, schema migration | +| 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 | | 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 22563d43..59472dc1 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -34,6 +34,7 @@ start_server() { WAF_IDS_STATE_PATH="$STATE_FILE" \ DNSBL_ORIGIN="dnsbl.test" \ EVENT_LIMIT="5" \ + CONTROL_PLANE_DATABASE_URL="" \ cargo run --quiet ) >"$LOG_FILE" 2>&1 & SERVER_PID="$!" diff --git a/src/control_plane.rs b/src/control_plane.rs new file mode 100644 index 00000000..5fd10d7e --- /dev/null +++ b/src/control_plane.rs @@ -0,0 +1,855 @@ +//! PostgreSQL control plane (issue #80). +//! +//! Production (non-loopback) binds require a control-plane URL. The JSON file +//! adapter remains for loopback/community use and is never selected as the +//! production authority. Tenant isolation is default-deny row-level security +//! with `FORCE ROW LEVEL SECURITY`; each transaction sets `wardnet.tenant_id`. + +use std::net::IpAddr; +use std::str::FromStr; +use tokio::sync::Mutex; +use tokio_postgres::{Client, GenericClient, NoTls}; +use waf_ids_core::{ + AppData, AuditLogEntry, CommercialProfile, DnsblEntry, EnforcementMode, LicenseStatus, + ProductEdition, RouteConfig, SecurityEvent, Severity, ThreatFeedStatus, ThreatIndicator, +}; + +/// Default tenant used until Keyverse supplies claims (#82). +pub const DEFAULT_TENANT_ID: &str = "local-lab"; + +const MIGRATION_VERSION: i32 = 1; + +/// Recoverable forward migration. Two-word snake_case names, 3NF, RLS. +pub const MIGRATION_SQL: &str = r#" +CREATE TABLE IF NOT EXISTS schema_migration ( + migration_version INTEGER PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS tenant_account ( + tenant_id TEXT PRIMARY KEY, + event_sequence BIGINT NOT NULL, + audit_sequence BIGINT NOT NULL +); + +CREATE TABLE IF NOT EXISTS tenant_profile ( + tenant_id TEXT PRIMARY KEY REFERENCES tenant_account (tenant_id), + deployment_id TEXT NOT NULL, + edition_name TEXT NOT NULL, + license_status TEXT NOT NULL, + license_id TEXT, + licensee_name TEXT, + licensed_until_unix BIGINT, + licensed_node_count INTEGER, + annual_contract_value_krw BIGINT, + support_contact TEXT NOT NULL, + feature_list TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS route_config ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + route_id TEXT NOT NULL, + path_prefix TEXT NOT NULL, + upstream_url TEXT NOT NULL, + enforcement_mode TEXT NOT NULL, + is_enabled BOOLEAN NOT NULL, + block_threshold INTEGER, + PRIMARY KEY (tenant_id, route_id) +); + +CREATE TABLE IF NOT EXISTS threat_indicator ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + indicator_type TEXT NOT NULL, + indicator_value TEXT NOT NULL, + indicator_source TEXT NOT NULL, + severity_name TEXT NOT NULL, + ttl_seconds BIGINT NOT NULL, + PRIMARY KEY (tenant_id, indicator_type, indicator_value, indicator_source) +); + +CREATE TABLE IF NOT EXISTS dnsbl_entry ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + host_address TEXT NOT NULL, + response_code TEXT NOT NULL, + block_reason TEXT NOT NULL, + entry_source TEXT NOT NULL, + ttl_seconds BIGINT NOT NULL, + prefix_length SMALLINT, + PRIMARY KEY (tenant_id, host_address) +); + +CREATE TABLE IF NOT EXISTS security_event ( + 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) +); + +CREATE TABLE IF NOT EXISTS audit_record ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + audit_id BIGINT NOT NULL, + timestamp_unix BIGINT NOT NULL, + actor_name TEXT NOT NULL, + action_name TEXT NOT NULL, + resource_name TEXT NOT NULL, + resource_id TEXT NOT NULL, + action_outcome TEXT NOT NULL, + PRIMARY KEY (tenant_id, audit_id) +); + +CREATE TABLE IF NOT EXISTS threat_feed ( + tenant_id TEXT NOT NULL REFERENCES tenant_account (tenant_id), + feed_id TEXT NOT NULL, + feed_source TEXT NOT NULL, + last_updated_unix BIGINT NOT NULL, + threat_count INTEGER NOT NULL, + dnsbl_count INTEGER NOT NULL, + ttl_seconds BIGINT NOT NULL, + PRIMARY KEY (tenant_id, feed_id) +); + +CREATE INDEX IF NOT EXISTS security_event_tenant_event + ON security_event (tenant_id, event_id); + +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; +CREATE POLICY tenant_isolation ON tenant_account + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE tenant_profile ENABLE ROW LEVEL SECURITY; +ALTER TABLE tenant_profile FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON tenant_profile; +CREATE POLICY tenant_isolation ON tenant_profile + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE route_config ENABLE ROW LEVEL SECURITY; +ALTER TABLE route_config FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON route_config; +CREATE POLICY tenant_isolation ON route_config + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE threat_indicator ENABLE ROW LEVEL SECURITY; +ALTER TABLE threat_indicator FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON threat_indicator; +CREATE POLICY tenant_isolation ON threat_indicator + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE dnsbl_entry ENABLE ROW LEVEL SECURITY; +ALTER TABLE dnsbl_entry FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON dnsbl_entry; +CREATE POLICY tenant_isolation ON dnsbl_entry + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE security_event ENABLE ROW LEVEL SECURITY; +ALTER TABLE security_event FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON security_event; +CREATE POLICY tenant_isolation ON security_event + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE audit_record ENABLE ROW LEVEL SECURITY; +ALTER TABLE audit_record FORCE ROW LEVEL SECURITY; +DROP POLICY IF EXISTS tenant_isolation ON audit_record; +CREATE POLICY tenant_isolation ON audit_record + USING (tenant_id = current_setting('wardnet.tenant_id', true)) + WITH CHECK (tenant_id = current_setting('wardnet.tenant_id', true)); + +ALTER TABLE threat_feed ENABLE ROW LEVEL SECURITY; +ALTER TABLE threat_feed FORCE ROW LEVEL SECURITY; +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)); +"#; + +/// Fail closed when a non-loopback bind has no control-plane URL. +pub fn require_postgres_for_bind( + bind_addr: &str, + database_url: Option<&str>, +) -> Result<(), String> { + if crate::bind_is_loopback(bind_addr) { + return Ok(()); + } + match database_url.map(str::trim).filter(|value| !value.is_empty()) { + Some(_) => Ok(()), + None => Err( + "production bind requires CONTROL_PLANE_DATABASE_URL; JSON file state is not production authority" + .to_string(), + ), + } +} + +/// Structural URL checks. TLS (`sslmode=require`) is fail-closed until rustls +/// is wired. Password stays in the registry, not logs. +pub fn parse_database_url(raw: &str) -> Result { + let raw = raw.trim(); + if raw.is_empty() { + return Err("CONTROL_PLANE_DATABASE_URL is empty".to_string()); + } + let lower = raw.to_ascii_lowercase(); + 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(), + ); + } + Ok(raw.to_string()) +} + +/// Live PostgreSQL snapshot store for one tenant. +pub struct PostgresPlane { + client: Mutex, + tenant_id: String, +} + +impl PostgresPlane { + pub async fn connect(url: &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 plane = Self { + client: Mutex::new(client), + tenant_id: DEFAULT_TENANT_ID.to_string(), + }; + plane.migrate().await?; + Ok(plane) + } + + async fn migrate(&self) -> Result<(), String> { + let client = self.client.lock().await; + 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}"))?; + Ok(()) + } + + /// Load the tenant snapshot, or `None` when the tenant has no rows yet. + pub async fn load(&self) -> Result, String> { + let mut client = self.client.lock().await; + load_snapshot(&mut client, &self.tenant_id).await + } + + /// Replace the tenant snapshot in one transaction (mutation + audit). + 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 + } +} + +async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result, String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane load 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 account = tx + .query_opt( + "SELECT event_sequence, audit_sequence FROM tenant_account WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load tenant_account failed: {error}"))?; + let Some(account) = account else { + tx.rollback() + .await + .map_err(|error| format!("control plane load rollback failed: {error}"))?; + return Ok(None); + }; + + let commercial = load_commercial(&tx, tenant_id).await?; + let routes = load_routes(&tx, tenant_id).await?; + let threats = load_threats(&tx, tenant_id).await?; + let dnsbl = load_dnsbl(&tx, tenant_id).await?; + let events = load_events(&tx, tenant_id).await?; + let audit_logs = load_audit(&tx, tenant_id).await?; + let threat_feeds = load_feeds(&tx, tenant_id).await?; + tx.commit() + .await + .map_err(|error| format!("control plane load commit failed: {error}"))?; + + Ok(Some(AppData { + routes, + threats, + dnsbl, + events, + next_event_id: account.get::<_, i64>(0) as u64, + audit_logs, + next_audit_log_id: account.get::<_, i64>(1) as u64, + commercial, + threat_feeds, + })) +} + +async fn save_snapshot(client: &mut Client, tenant_id: &str, data: &AppData) -> Result<(), String> { + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane 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 tenant_account (tenant_id, event_sequence, audit_sequence) + VALUES ($1, $2, $3) + ON CONFLICT (tenant_id) DO UPDATE SET + event_sequence = EXCLUDED.event_sequence, + audit_sequence = EXCLUDED.audit_sequence", + &[ + &tenant_id, + &(data.next_event_id as i64), + &(data.next_audit_log_id as i64), + ], + ) + .await + .map_err(|error| format!("control plane upsert tenant_account failed: {error}"))?; + + for table in [ + "route_config", + "threat_indicator", + "dnsbl_entry", + "security_event", + "audit_record", + "threat_feed", + "tenant_profile", + ] { + tx.execute( + &format!("DELETE FROM {table} WHERE tenant_id = $1"), + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane delete {table} failed: {error}"))?; + } + + let features = serde_json::to_string(&data.commercial.features) + .expect("feature list is JSON-serializable"); + tx.execute( + "INSERT INTO tenant_profile ( + tenant_id, deployment_id, edition_name, license_status, license_id, + licensee_name, licensed_until_unix, licensed_node_count, + annual_contract_value_krw, support_contact, feature_list + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)", + &[ + &tenant_id, + &data.commercial.deployment_id, + &edition_sql(&data.commercial.edition), + &license_sql(&data.commercial.license_status), + &data.commercial.license_id, + &data.commercial.licensee, + &data.commercial.licensed_until_unix.map(|v| v as i64), + &data.commercial.licensed_node_count.map(|v| v as i32), + &data.commercial.annual_contract_value_krw.map(|v| v as i64), + &data.commercial.support_contact, + &features, + ], + ) + .await + .map_err(|error| format!("control plane insert tenant_profile failed: {error}"))?; + + for route in &data.routes { + tx.execute( + "INSERT INTO route_config ( + tenant_id, route_id, path_prefix, upstream_url, enforcement_mode, + is_enabled, block_threshold + ) VALUES ($1,$2,$3,$4,$5,$6,$7)", + &[ + &tenant_id, + &route.id, + &route.path_prefix, + &route.upstream, + &mode_sql(&route.mode), + &route.enabled, + &route.block_threshold.map(i32::from), + ], + ) + .await + .map_err(|error| format!("control plane insert route_config failed: {error}"))?; + } + + for threat in &data.threats { + tx.execute( + "INSERT INTO threat_indicator ( + tenant_id, indicator_type, indicator_value, indicator_source, + severity_name, ttl_seconds + ) VALUES ($1,$2,$3,$4,$5,$6)", + &[ + &tenant_id, + &threat.indicator_type, + &threat.value, + &threat.source, + &severity_sql(&threat.severity), + &(threat.ttl_seconds as i64), + ], + ) + .await + .map_err(|error| format!("control plane insert threat_indicator failed: {error}"))?; + } + + for entry in &data.dnsbl { + let address = entry.address.to_string(); + tx.execute( + "INSERT INTO dnsbl_entry ( + tenant_id, host_address, response_code, block_reason, entry_source, + ttl_seconds, prefix_length + ) VALUES ($1,$2,$3,$4,$5,$6,$7)", + &[ + &tenant_id, + &address, + &entry.code, + &entry.reason, + &entry.source, + &(entry.ttl_seconds as i64), + &entry.prefix_len.map(i16::from), + ], + ) + .await + .map_err(|error| format!("control plane insert dnsbl_entry failed: {error}"))?; + } + + for event in &data.events { + 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)", + &[ + &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}"))?; + } + + for audit in &data.audit_logs { + tx.execute( + "INSERT INTO audit_record ( + tenant_id, audit_id, timestamp_unix, actor_name, action_name, + resource_name, resource_id, action_outcome + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8)", + &[ + &tenant_id, + &(audit.id as i64), + &(audit.timestamp_unix as i64), + &audit.actor, + &audit.action, + &audit.resource, + &audit.resource_id, + &audit.outcome, + ], + ) + .await + .map_err(|error| format!("control plane insert audit_record failed: {error}"))?; + } + + for feed in &data.threat_feeds { + tx.execute( + "INSERT INTO threat_feed ( + tenant_id, feed_id, feed_source, last_updated_unix, threat_count, + dnsbl_count, ttl_seconds + ) VALUES ($1,$2,$3,$4,$5,$6,$7)", + &[ + &tenant_id, + &feed.feed_id, + &feed.source, + &(feed.last_updated_unix as i64), + &(feed.threat_count as i32), + &(feed.dnsbl_count as i32), + &(feed.ttl_seconds as i64), + ], + ) + .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(()) +} + +async fn load_commercial( + client: &C, + tenant_id: &str, +) -> Result { + let row = client + .query_opt( + "SELECT deployment_id, edition_name, license_status, license_id, licensee_name, + licensed_until_unix, licensed_node_count, annual_contract_value_krw, + support_contact, feature_list + FROM tenant_profile WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load tenant_profile failed: {error}"))?; + let Some(row) = row else { + return Ok(CommercialProfile::seeded()); + }; + let features: String = row.get(9); + Ok(CommercialProfile { + tenant_id: tenant_id.to_string(), + deployment_id: row.get(0), + edition: parse_edition(row.get(1))?, + license_status: parse_license(row.get(2))?, + license_id: row.get(3), + licensee: row.get(4), + licensed_until_unix: row.get::<_, Option>(5).map(|v| v as u64), + licensed_node_count: row.get::<_, Option>(6).map(|v| v as u32), + annual_contract_value_krw: row.get::<_, Option>(7).map(|v| v as u64), + support_contact: row.get(8), + features: serde_json::from_str(&features) + .map_err(|error| format!("control plane feature_list is not JSON: {error}"))?, + }) +} + +async fn load_routes( + client: &C, + tenant_id: &str, +) -> Result, String> { + let rows = client + .query( + "SELECT route_id, path_prefix, upstream_url, enforcement_mode, is_enabled, block_threshold + FROM route_config WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load route_config failed: {error}"))?; + rows.iter() + .map(|row| { + Ok(RouteConfig { + id: row.get(0), + path_prefix: row.get(1), + upstream: row.get(2), + mode: parse_mode(row.get(3))?, + enabled: row.get(4), + block_threshold: row.get::<_, Option>(5).map(|v| v as u16), + }) + }) + .collect() +} + +async fn load_threats( + client: &C, + tenant_id: &str, +) -> Result, String> { + let rows = client + .query( + "SELECT indicator_type, indicator_value, indicator_source, severity_name, ttl_seconds + FROM threat_indicator WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load threat_indicator failed: {error}"))?; + rows.iter() + .map(|row| { + Ok(ThreatIndicator { + indicator_type: row.get(0), + value: row.get(1), + source: row.get(2), + severity: parse_severity(row.get(3))?, + ttl_seconds: row.get::<_, i64>(4) as u64, + }) + }) + .collect() +} + +async fn load_dnsbl( + client: &C, + tenant_id: &str, +) -> Result, String> { + let rows = client + .query( + "SELECT host_address, response_code, block_reason, entry_source, ttl_seconds, prefix_length + FROM dnsbl_entry WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load dnsbl_entry failed: {error}"))?; + rows.iter() + .map(|row| { + let address: String = row.get(0); + Ok(DnsblEntry { + address: IpAddr::from_str(&address) + .map_err(|error| format!("control plane host_address {address}: {error}"))?, + code: row.get(1), + reason: row.get(2), + source: row.get(3), + ttl_seconds: row.get::<_, i64>(4) as u64, + prefix_len: row.get::<_, Option>(5).map(|v| v as u8), + }) + }) + .collect() +} + +async fn load_events( + client: &C, + tenant_id: &str, +) -> Result, String> { + let rows = client + .query( + "SELECT event_id, timestamp_unix, client_address, route_id, action_name, + event_reason, event_score, request_path + FROM security_event WHERE tenant_id = $1 ORDER BY event_id", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load security_event failed: {error}"))?; + rows.iter() + .map(|row| { + let client_address: Option = row.get(2); + Ok(SecurityEvent { + id: row.get::<_, i64>(0) as u64, + timestamp_unix: row.get::<_, i64>(1) as u64, + client_ip: client_address + .map(|value| { + IpAddr::from_str(&value).map_err(|error| { + format!("control plane client_address {value}: {error}") + }) + }) + .transpose()?, + route_id: row.get(3), + action: row.get(4), + reason: row.get(5), + score: row.get::<_, i32>(6) as u16, + path: row.get(7), + }) + }) + .collect() +} + +async fn load_audit( + client: &C, + tenant_id: &str, +) -> Result, String> { + let rows = client + .query( + "SELECT audit_id, timestamp_unix, actor_name, action_name, resource_name, + resource_id, action_outcome + FROM audit_record WHERE tenant_id = $1 ORDER BY audit_id", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load audit_record failed: {error}"))?; + Ok(rows + .iter() + .map(|row| AuditLogEntry { + id: row.get::<_, i64>(0) as u64, + timestamp_unix: row.get::<_, i64>(1) as u64, + actor: row.get(2), + action: row.get(3), + resource: row.get(4), + resource_id: row.get(5), + outcome: row.get(6), + }) + .collect()) +} + +async fn load_feeds( + client: &C, + tenant_id: &str, +) -> Result, String> { + 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", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane load threat_feed failed: {error}"))?; + Ok(rows + .iter() + .map(|row| ThreatFeedStatus { + feed_id: row.get(0), + source: row.get(1), + last_updated_unix: row.get::<_, i64>(2) as u64, + threat_count: row.get::<_, i32>(3) as usize, + dnsbl_count: row.get::<_, i32>(4) as usize, + ttl_seconds: row.get::<_, i64>(5) as u64, + }) + .collect()) +} + +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), + "enterprise" => Ok(ProductEdition::Enterprise), + other => Err(format!("unknown edition_name {other}")), + } +} + +fn license_sql(status: &LicenseStatus) -> &'static str { + match status { + LicenseStatus::Unlicensed => "unlicensed", + LicenseStatus::Evaluation => "evaluation", + LicenseStatus::Active => "active", + LicenseStatus::Expired => "expired", + } +} + +fn parse_license(value: &str) -> Result { + match value { + "unlicensed" => Ok(LicenseStatus::Unlicensed), + "evaluation" => Ok(LicenseStatus::Evaluation), + "active" => Ok(LicenseStatus::Active), + "expired" => Ok(LicenseStatus::Expired), + other => Err(format!("unknown license_status {other}")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn production_bind_requires_control_plane_url() { + require_postgres_for_bind("0.0.0.0:8080", None).unwrap_err(); + require_postgres_for_bind("0.0.0.0:8080", Some("postgres://wardnet@127.0.0.1/wardnet")) + .unwrap(); + require_postgres_for_bind("127.0.0.1:8080", None).unwrap(); + require_postgres_for_bind("[::1]:8080", None).unwrap(); + } + + #[test] + fn database_url_rejects_non_postgres_and_tls_until_wired() { + 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").unwrap(); + parse_database_url("postgresql://wardnet@127.0.0.1/wardnet?sslmode=disable").unwrap(); + } + + #[test] + fn migration_sql_is_3nf_rls_and_two_word_names() { + for table in [ + "tenant_account", + "tenant_profile", + "route_config", + "threat_indicator", + "dnsbl_entry", + "security_event", + "audit_record", + "threat_feed", + "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("PRIMARY KEY (tenant_id, route_id)")); + assert!(MIGRATION_SQL.contains("REFERENCES tenant_account")); + assert!( + !MIGRATION_SQL.contains("json_blob"), + "do not dump AppData as one JSON column" + ); + } + + #[tokio::test] + async fn postgres_roundtrip_seeded_snapshot_when_database_url_is_set() { + let Ok(url) = std::env::var("CONTROL_PLANE_TEST_DATABASE_URL") else { + return; + }; + if url.trim().is_empty() { + return; + } + let plane = PostgresPlane::connect(&url) + .await + .expect("test database must accept the control plane"); + let seeded = AppData::seeded(); + plane.save(&seeded).await.expect("save seeded snapshot"); + let loaded = plane + .load() + .await + .expect("load snapshot") + .expect("tenant rows must exist after save"); + assert_eq!(loaded.routes, seeded.routes); + assert_eq!(loaded.threats, seeded.threats); + 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); + } +} diff --git a/src/credentials.rs b/src/credentials.rs index 02b7f39e..66223ab5 100644 --- a/src/credentials.rs +++ b/src/credentials.rs @@ -11,6 +11,7 @@ use std::{collections::HashMap, io::ErrorKind, path::Path}; /// Well-known secret keys loaded into the registry at bootstrap. pub const CRED_ADMIN_TOKEN: &str = "admin_token"; pub const CRED_ADMIN_TOKENS: &str = "admin_tokens"; +pub const CRED_CONTROL_PLANE_URL: &str = "control_plane_url"; /// Where secret-bearing credentials were loaded from (never includes values). #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -73,6 +74,7 @@ impl CredentialRegistry { credentials_path: Option<&Path>, env_admin_token: Option, env_admin_tokens: Option, + env_control_plane_url: Option, ) -> Result { let mut values = HashMap::new(); let mut from_file = false; @@ -88,7 +90,7 @@ impl CredentialRegistry { path.display() ) })?; - for key in [CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS] { + for key in [CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL] { if let Some(raw) = file_map.get(key) { let text = json_value_as_nonempty_string(raw); if let Some(text) = text { @@ -120,6 +122,12 @@ impl CredentialRegistry { values.insert(CRED_ADMIN_TOKENS.to_string(), tokens); from_env = true; } + if !values.contains_key(CRED_CONTROL_PLANE_URL) + && let Some(url) = env_control_plane_url.filter(|value| !value.is_empty()) + { + values.insert(CRED_CONTROL_PLANE_URL.to_string(), url); + from_env = true; + } let source = if from_file { CredentialSource::File @@ -159,6 +167,7 @@ mod tests { None, Some("secret".to_string()), Some("tok:alice".to_string()), + None, ) .unwrap(); assert_eq!(registry.source(), CredentialSource::Env); @@ -173,7 +182,7 @@ mod tests { #[test] fn bootstrap_empty_when_no_secrets() { let registry = - CredentialRegistry::bootstrap_secrets(None, None, Some(String::new())).unwrap(); + CredentialRegistry::bootstrap_secrets(None, None, Some(String::new()), None).unwrap(); assert_eq!(registry.source(), CredentialSource::None); assert!(!registry.has_admin_auth()); } @@ -202,6 +211,7 @@ mod tests { Some(&path), Some("from-env".to_string()), Some("envtok:env".to_string()), + None, ) .unwrap(); assert_eq!(registry.source(), CredentialSource::File); @@ -232,6 +242,7 @@ mod tests { Some(&path), Some("ignored".to_string()), Some("envtok:bob".to_string()), + None, ) .unwrap(); assert_eq!(registry.source(), CredentialSource::File); @@ -258,6 +269,7 @@ mod tests { Some(&path), Some("env-secret".to_string()), None, + None, ) .unwrap(); assert_eq!(registry.source(), CredentialSource::Env); @@ -280,7 +292,7 @@ mod tests { std::fs::create_dir_all(&dir).unwrap(); let path = dir.join("credentials.json"); std::fs::write(&path, "not-json").unwrap(); - let err = CredentialRegistry::bootstrap_secrets(Some(&path), None, None).unwrap_err(); + let err = CredentialRegistry::bootstrap_secrets(Some(&path), None, None, None).unwrap_err(); assert!(err.contains("not valid JSON")); let _ = std::fs::remove_dir_all(&dir); } diff --git a/src/lib.rs b/src/lib.rs index ec443e29..cd7fcc44 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -35,6 +35,7 @@ pub use waf_ids_core::{ ThreatIndicator, export_dnsbl_zone, ip_in_network, reverse_ipv4_for_dnsbl, score_request, }; +mod control_plane; mod coraza_audit; mod coraza_inprocess; mod credentials; @@ -45,7 +46,10 @@ mod proven_engine; mod stix_import; mod suricata_eve; mod taxii; -pub use credentials::{CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CredentialRegistry, CredentialSource}; +pub use credentials::{ + CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL, CredentialRegistry, + CredentialSource, +}; pub use destination::{DestinationPolicy, HostResolver, SystemHostResolver}; pub use proven_engine::{ProvenEngineConfig, ProvenEngineOutcome}; @@ -84,6 +88,8 @@ pub struct AppState { /// Addresses that already passed policy; the HTTP clients resolve through /// this pin board instead of a second OS DNS lookup. pins: Arc, + /// PostgreSQL snapshot store. `None` keeps the JSON-file / memory adapter. + control_plane: Option>, } /// Configuration for the optional LLM-backed SOC analysis. Points at an @@ -128,6 +134,24 @@ impl AppState { Ok(Self::new(data, config)) } + /// 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 mut data = match plane.load().await? { + Some(loaded) => loaded, + None => AppData::seeded(), + }; + let event_limit = config.event_limit.max(1); + enforce_event_limit(&mut data, event_limit); + plane.save(&data).await?; + Ok(Self::new(data, config).with_control_plane(Arc::new(plane))) + } + + fn with_control_plane(mut self, plane: Arc) -> Self { + self.control_plane = Some(plane); + self + } + fn new(data: AppData, config: AppConfig) -> Self { let pins = Arc::new(destination::DestinationPins::default()); Self { @@ -151,6 +175,7 @@ impl AppState { destination: DestinationPolicy::production(), resolver: Arc::new(SystemHostResolver), pins, + control_plane: None, } } @@ -294,6 +319,9 @@ impl AppState { } async fn persist_snapshot(&self, data: &AppData) -> Result<(), String> { + if let Some(plane) = &self.control_plane { + return plane.save(data).await; + } let Some(path) = self.state_path.as_deref() else { return Ok(()); }; @@ -303,7 +331,9 @@ impl AppState { fn health_status(&self) -> HealthStatus { HealthStatus { status: "ok".to_string(), - persistence: if self.state_path.is_some() { + persistence: if self.control_plane.is_some() { + "postgres".to_string() + } else if self.state_path.is_some() { "file".to_string() } else { "memory".to_string() @@ -429,6 +459,7 @@ pub struct SupportBundle { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct HealthStatus { pub status: String, + /// `postgres` (production authority), `file` (loopback/community), or `memory`. pub persistence: String, pub dnsbl_origin: String, pub event_limit: usize, @@ -3214,7 +3245,7 @@ fn outbound_http_client(pins: Arc) -> reqwest::Cli .expect("failed to build fail-closed outbound HTTP client") } -fn bind_is_loopback(bind_addr: &str) -> bool { +pub(crate) fn bind_is_loopback(bind_addr: &str) -> bool { let trimmed = bind_addr.trim(); if let Ok(addr) = trimmed.parse::() { return addr.ip().is_loopback(); @@ -3263,6 +3294,7 @@ pub async fn run_from_env( credentials_path.as_deref(), std::env::var("ADMIN_TOKEN").ok(), std::env::var("ADMIN_TOKENS").ok(), + std::env::var("CONTROL_PLANE_DATABASE_URL").ok(), )?; let config = AppConfig { admin_token: credentials @@ -3327,9 +3359,17 @@ pub async fn run_from_env( in_process, }; let destination_policy = startup_destination_policy(&bind_addr)?; - let state = AppState::load(config) - .await - .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidData, message))? + let control_plane_url = credentials + .get_credential(CRED_CONTROL_PLANE_URL) + .map(str::to_owned); + control_plane::require_postgres_for_bind(&bind_addr, control_plane_url.as_deref()) + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; + let state = match control_plane_url.as_deref() { + Some(url) => AppState::load_postgres(config, url).await, + None => AppState::load(config).await, + } + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidData, message))?; + let state = state .with_rate_limit(rate_limit, rate_limit_window) .with_admin_tokens(admin_tokens) .with_credentials_source(credentials.source()) @@ -3392,6 +3432,7 @@ mod tests { "PROVEN_ENGINE_FAIL_CLOSED", "DESTINATION_ALLOWLIST", "DESTINATION_DENYLIST", + "CONTROL_PLANE_DATABASE_URL", ] { unsafe { std::env::remove_var(name) }; } @@ -3508,6 +3549,25 @@ mod tests { clear_run_env(); } + #[tokio::test] + async fn run_from_env_fail_closes_public_bind_without_postgres() { + let _guard = ENV_GUARD.lock().await; + clear_run_env(); + unsafe { + std::env::set_var("BIND_ADDR", "0.0.0.0:0"); + std::env::remove_var("CONTROL_PLANE_DATABASE_URL"); + } + let error = run_from_env(Box::pin(std::future::ready(()))) + .await + .expect_err("production bind must require postgres"); + let message = error.to_string(); + assert!( + message.contains("CONTROL_PLANE_DATABASE_URL"), + "operator must see the production authority requirement: {message}" + ); + clear_run_env(); + } + #[tokio::test] async fn run_from_env_rejects_malformed_max_body_bytes() { let _guard = ENV_GUARD.lock().await; diff --git a/tests/binary.rs b/tests/binary.rs index 404eb13d..ec57db7f 100644 --- a/tests/binary.rs +++ b/tests/binary.rs @@ -61,6 +61,7 @@ fn binary_does_not_report_readiness_before_state_validation() { .env_remove("CORAZA_LIB_PATH") .env_remove("CORAZA_RULES_PATH") .env_remove("CORAZA_DIRECTIVES") + .env_remove("CONTROL_PLANE_DATABASE_URL") .output() .expect("spawn gateway binary for startup validation check"); let _ = std::fs::remove_file(&state_path); @@ -112,6 +113,62 @@ fn binary_fail_closes_when_libcoraza_path_is_missing() { ); } +#[test] +fn binary_fail_closes_non_loopback_listen_without_postgres() { + let output = Command::new(env!("CARGO_BIN_EXE_waf-ids-ai-soc")) + .env("BIND_ADDR", "0.0.0.0:0") + .env_remove("CONTROL_PLANE_DATABASE_URL") + .env_remove("WAF_IDS_STATE_PATH") + .env_remove("ADMIN_TOKEN") + .env_remove("ADMIN_TOKENS") + .env_remove("CORAZA_LIB_PATH") + .output() + .expect("spawn gateway binary for production postgres gate"); + assert!( + !output.status.success(), + "production bind without postgres must fail: {:?}", + output.status + ); + let stderr = String::from_utf8_lossy(&output.stderr); + let stdout = String::from_utf8_lossy(&output.stdout); + let combined = format!("{stdout}{stderr}"); + assert!( + combined.contains("CONTROL_PLANE_DATABASE_URL"), + "startup error should name the missing control plane:\n{combined}" + ); + assert!( + !combined.contains("waf-ids-ai-soc listening on"), + "readiness must not be reported without production postgres:\n{combined}" + ); +} + +#[test] +fn binary_fail_closes_when_control_plane_url_is_not_postgres() { + let output = Command::new(env!("CARGO_BIN_EXE_waf-ids-ai-soc")) + .env("BIND_ADDR", "127.0.0.1:0") + .env("CONTROL_PLANE_DATABASE_URL", "mysql://not-postgres") + .env_remove("WAF_IDS_STATE_PATH") + .env_remove("CORAZA_LIB_PATH") + .output() + .expect("spawn gateway binary for control-plane URL check"); + assert!( + !output.status.success(), + "non-postgres URL must fail startup: {:?}", + output.status + ); + let stderr = String::from_utf8_lossy(&output.stderr); + let stdout = String::from_utf8_lossy(&output.stdout); + let combined = format!("{stdout}{stderr}"); + assert!( + combined.contains("postgres://"), + "startup error should require a postgres URL:\n{combined}" + ); + assert!( + !combined.contains("waf-ids-ai-soc listening on"), + "readiness must not be reported for a rejected control-plane URL:\n{combined}" + ); +} + fn spawn_ready_gateway() -> Child { let mut child = Command::new(env!("CARGO_BIN_EXE_waf-ids-ai-soc")) .env("BIND_ADDR", "127.0.0.1:0") @@ -124,6 +181,7 @@ fn spawn_ready_gateway() -> Child { .env_remove("CORAZA_RULES_PATH") .env_remove("CORAZA_DIRECTIVES") .env_remove("CORAZA_WAF_URL") + .env_remove("CONTROL_PLANE_DATABASE_URL") .stdout(Stdio::piped()) .spawn() .expect("spawn gateway binary"); From 3a19ed9b276d3e14a751fbd5a87087aade713a2d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 24 Aug 2026 02:08:07 +0900 Subject: [PATCH 2/3] feat(store): transactional outbox and leased workers Issue #81 first slice on the PostgreSQL control plane. Security events append with an outbox row in one transaction instead of rewriting the snapshot. Workers claim with SKIP LOCKED, retry, dead-letter, and record unique receipts. Stdout SIEM is at-least-once; the receipt is the exactly-once ack. Also deterministic ORDER BY on postgres loads (still-valid #98 finding). Do not re-implement the postgres gate. --- CHANGELOG.md | 1 + CLAUDE.md | 2 +- Cargo.lock | 82 ++- Cargo.toml | 1 + docs/architecture.md | 1 + docs/doctoring/outbox-workers.md | 50 ++ docs/product-technical-gap-baseline.md | 93 ++- scripts/smoke.sh | 2 + src/control_plane.rs | 893 ++++++++++++++++++++++++- src/lib.rs | 241 ++++++- src/outbox.rs | 186 +++++ 11 files changed, 1470 insertions(+), 82 deletions(-) create mode 100644 docs/doctoring/outbox-workers.md create mode 100644 src/outbox.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 0070ff0d..6beb98d9 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 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`. - 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 d3843fac..f54f1569 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 a05a89e4..3c910412 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -104,6 +104,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 +169,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", ] @@ -176,6 +185,15 @@ 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 +203,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 +231,25 @@ dependencies = [ "cmov", ] +[[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", + "block-buffer 0.12.1", "const-oid", - "crypto-common", + "crypto-common 0.2.2", "ctutils", ] @@ -330,6 +368,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 +423,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 +737,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" dependencies = [ "cfg-if", - "digest", + "digest 0.11.3", ] [[package]] @@ -826,7 +874,7 @@ dependencies = [ "md-5", "memchr", "rand 0.10.2", - "sha2", + "sha2 0.11.0", "stringprep", ] @@ -1255,6 +1303,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 +1321,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]] @@ -1651,6 +1710,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" @@ -1662,6 +1727,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2 0.10.9", "tokio", "tokio-postgres", "tower", diff --git a/Cargo.toml b/Cargo.toml index 88185c4d..839f999c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ 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"] } +sha2 = "0.10" [dev-dependencies] tower = { version = "0.5", features = ["util"] } diff --git a/docs/architecture.md b/docs/architecture.md index b998080b..0abf4937 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -30,6 +30,7 @@ flowchart LR - `src/main.rs`: process startup and operator configuration from `BIND_ADDR`, `ADMIN_TOKEN`, `WAF_IDS_STATE_PATH`, `DNSBL_ORIGIN`, and `EVENT_LIMIT`. - `src/lib.rs`: Axum app, routing, management APIs, optional JSON persistence, gateway handler, upstream proxying, admin console, support bundle assembly, NDJSON event export, and in-crate HTTP tests. Persistence, destination-list, and sidecar settings validate before the readiness line is printed. - `src/control_plane.rs`: PostgreSQL production authority (issue #80). Non-loopback binds require `CONTROL_PLANE_DATABASE_URL`. Tenant isolation is default-deny RLS. The JSON file adapter remains loopback/community only. +- `src/outbox.rs`: transactional outbox + leased workers (issue #81). Security events append incrementally with an outbox row in the same transaction. Workers claim with `SKIP LOCKED`. `GET /api/outbox` and `/healthz.outbox` are operator-visible. - `src/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. diff --git a/docs/doctoring/outbox-workers.md b/docs/doctoring/outbox-workers.md new file mode 100644 index 00000000..a4f69307 --- /dev/null +++ b/docs/doctoring/outbox-workers.md @@ -0,0 +1,50 @@ +# 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 messages +- `POST /api/outbox/{message_id}/replay` (admin write) requeues dead letters + +Loopback file/memory adapters keep in-process stdout SIEM and report +`outbox=disabled`. Remaining: rustls for the control-plane connection, a +non-owner runtime role, backup/restore drill, HASH partitioning, and additional +consumers (TAXII poll, Clearfolio, contextual-orchestrator) on the same +message/receipt contract. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 8c3715c0..2867a9de 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-23T17:06Z (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,15 @@ 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. | +| [#99](https://github.com/ContextualWisdomLab/wardnet/pull/99) | feat(store): transactional outbox and leased workers | `feat/issue-81-outbox-workers` stacked on #98 | local fmt/test/clippy + two `/healthz` smokes + postgres `/healthz.outbox=ready` this hour | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 first. Do not `--admin`. Do not re-implement the postgres gate. | +| [#98](https://github.com/ContextualWisdomLab/wardnet/pull/98) | feat(store): require PostgreSQL as the production control plane | `ea621985e276` (`feat/issue-80-postgres-control-plane`) stacked on #97 | rust + fuzz green at last snapshot; Devin 7 threads (full-snapshot rewrite, ORDER BY, TLS, RLS owner, reconnect) | Author this pass; Devin COMMENTED | Org 2-approval + self-author. ORDER BY + incremental event persist addressed on #99. Remaining rustls / non-owner role / backup are #80 remainder. Do not `--admin`. | +| [#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 +54,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 this pass** | +| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical — gate on #98; rustls/backup remainder** | | [#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 +73,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,7 +91,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 + RLS snapshot this pass** +### Durable control plane (issue #80) — **production gate on #98** 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,6 +101,20 @@ transaction. JSON file / memory remain loopback/community only. non-owner role, backup/restore drill, event HASH partitioning, optimistic concurrency. +### Transactional outbox (issue #81) — **first slice this pass** + +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. Remaining +consumers: TAXII poll, Clearfolio, contextual-orchestrator on the same contract. + ### Fail-closed credentials (issue #78) — **closed on PR #94** Shipped on `fix/issue-78-fail-closed-credentials`. Do not re-implement. @@ -116,7 +129,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 +142,7 @@ 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 this pass | ### CSAP / SOC 2 vs PII unmasking @@ -139,36 +154,40 @@ 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. +(outbox schema, claim/ack/dead-letter/replay, incremental event persist, +health/API, admin card). 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. ## This loop’s shipped gap -Issue **#80** first slice (PostgreSQL production authority). Non-loopback binds -fail closed without `CONTROL_PLANE_DATABASE_URL`. Operator-visible: -`/healthz.persistence=postgres`; credentials key `control_plane_url`. Driving -tests: `run_from_env_fail_closes_public_bind_without_postgres`, -`binary_fail_closes_non_loopback_listen_without_postgres`, -`binary_fail_closes_when_control_plane_url_is_not_postgres`, -`postgres_roundtrip_seeded_snapshot_when_database_url_is_set` (CI postgres -service). Do not re-implement #78, the #86 sidecar/libcoraza slices, or the -#79 pin. +Issue **#81** first slice (transactional outbox + leased workers) on the #80 +PostgreSQL authority. Also the still-valid #98 ORDER BY on load queries and +incremental security-event persist (gateway path no longer rewrites the whole +snapshot). Operator-visible: `/healthz.outbox=ready` on postgres, +`GET /api/outbox`, admin Outbox card. Driving tests: +`postgres_appends_event_and_outbox_atomically`, +`postgres_outbox_worker_is_idempotent_and_dead_letters`, +`postgres_expired_lease_is_reclaimed_and_skip_locked_is_exclusive`, +`outbox_api_is_admin_authenticated_and_disabled_without_postgres`. +Do not re-implement #78, the #86 sidecar/libcoraza slices, the #79 pin, or +the #80 production postgres gate. ## 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 and this #81 PR merge-ready. Merge order #95 then + #96 then #97 then #98 then this. Do not re-implement shipped slices. +3. Next runtime gap if policy still blocks: rustls / backup drill remainder of + #80, or additional #81 consumers (TAXII / Clearfolio / orchestrator) on this + outbox. 4. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. diff --git a/scripts/smoke.sh b/scripts/smoke.sh index 59472dc1..9530cbc8 100755 --- a/scripts/smoke.sh +++ b/scripts/smoke.sh @@ -82,6 +82,8 @@ 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' 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 5fd10d7e..f5df2d90 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -5,10 +5,15 @@ //! 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, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, STATUS_LEASED, + STATUS_PENDING, STATUS_PROCESSED, +}; use std::net::IpAddr; use std::str::FromStr; use tokio::sync::Mutex; -use tokio_postgres::{Client, GenericClient, NoTls}; +use tokio_postgres::{Client, GenericClient, NoTls, Transaction}; use waf_ids_core::{ AppData, AuditLogEntry, CommercialProfile, DnsblEntry, EnforcementMode, LicenseStatus, ProductEdition, RouteConfig, SecurityEvent, Severity, ThreatFeedStatus, ThreatIndicator, @@ -17,7 +22,7 @@ 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 = 2; /// Recoverable forward migration. Two-word snake_case names, 3NF, RLS. pub const MIGRATION_SQL: &str = r#" @@ -117,6 +122,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,6 +212,20 @@ 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)); "#; /// Fail closed when a non-loopback bind has no control-plane URL. @@ -210,6 +264,9 @@ pub fn parse_database_url(raw: &str) -> Result { Ok(raw.to_string()) } +/// 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, @@ -218,6 +275,10 @@ pub struct PostgresPlane { 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 @@ -227,18 +288,32 @@ impl PostgresPlane { }); let plane = Self { client: Mutex::new(client), - tenant_id: DEFAULT_TENANT_ID.to_string(), + tenant_id: tenant_id.to_string(), }; plane.migrate().await?; Ok(plane) } async fn migrate(&self) -> Result<(), String> { + let _gate = MIGRATION_GATE.lock().await; let client = self.client.lock().await; + 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 { + return Ok(()); + } client .batch_execute(MIGRATION_SQL) .await - .map_err(|error| format!("control plane migration failed: {error}"))?; + .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", @@ -255,11 +330,49 @@ 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 } + + /// 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, 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(&self) -> Result, String> { + let mut client = self.client.lock().await; + list_outbox(&mut client, &self.tenant_id).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 + } } async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result, String> { @@ -504,6 +617,8 @@ async fn save_snapshot(client: &mut Client, tenant_id: &str, data: &AppData) -> .map_err(|error| format!("control plane insert threat_feed failed: {error}"))?; } + enqueue_snapshot_outbox(&tx, tenant_id, data).await?; + tx.commit() .await .map_err(|error| format!("control plane commit failed: {error}"))?; @@ -551,7 +666,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 +692,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 +718,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 +809,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,6 +827,543 @@ async fn load_feeds( .collect()) } +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 +} + +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}"))?; + + 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}"))?; + + 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?; + + tx.commit() + .await + .map_err(|error| format!("control plane event commit failed: {error}"))?; + Ok(()) +} + +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, +} + +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, + 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).await?; + processed += 1; + continue; + } + match dispatch(&message) { + Ok(evidence) => { + ack_processed(client, tenant_id, &message, &evidence, now_unix).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, +) -> 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}"))?; + 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 list_outbox(client: &mut Client, tenant_id: &str) -> 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 created_unix, aggregate_id, aggregate_version", + &[&tenant_id], + ) + .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(()) +} + +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", @@ -814,6 +1467,8 @@ mod tests { "security_event", "audit_record", "threat_feed", + "outbox_message", + "outbox_receipt", "schema_migration", ] { assert!(MIGRATION_SQL.contains(table), "missing table {table}"); @@ -822,6 +1477,7 @@ mod tests { assert!(MIGRATION_SQL.contains("wardnet.tenant_id")); 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 +1507,224 @@ mod tests { assert_eq!(loaded.dnsbl, seeded.dnsbl); assert_eq!(loaded.next_event_id, seeded.next_event_id); assert_eq!(loaded.commercial.tenant_id, DEFAULT_TENANT_ID); + let messages = plane.list_outbox().await.expect("list snapshot outbox"); + 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().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().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() + .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().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); } } diff --git a/src/lib.rs b/src/lib.rs index cd7fcc44..03b4ee4e 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; @@ -345,7 +346,34 @@ 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, + } + } + + 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(), } + health } } @@ -473,6 +501,12 @@ 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, } const PHISHING_DATABASE_DEFAULT_FEED_ID: &str = "phishing-database-active"; @@ -545,6 +579,8 @@ 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/events.ndjson", get(events_ndjson)) .route("/api/kpis", get(kpis)) .route("/api/signatures", get(list_signatures)) @@ -907,7 +943,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 +1128,75 @@ 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, + 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 Some(plane) = &state.control_plane else { + return Json(OutboxListView { + status: "disabled".to_string(), + messages: Vec::new(), + }) + .into_response(); + }; + match plane.list_outbox().await { + Ok(messages) => Json(OutboxListView { + status: "ready".to_string(), + 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), + } +} + async fn kpis(State(state): State) -> Json { let data = state.inner.read().await; Json(kpi_snapshot_at(&data, now_unix())) @@ -2161,7 +2266,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), @@ -2520,29 +2625,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}"); } } @@ -3023,6 +3136,7 @@ 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…

     
Evidence manifest
Loading…

     
SOC event export (ndjson)
Loading…

     
     
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…