From 3a19ed9b276d3e14a751fbd5a87087aade713a2d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 24 Aug 2026 02:08:07 +0900 Subject: [PATCH 1/2] 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…