diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bf6f303..ebf4d129 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +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`. +- PostgreSQL control-plane mutations enqueue a transactional outbox row in the same transaction (issue #81). Security events append incrementally instead of rewriting the snapshot. A leased worker claims with `FOR UPDATE SKIP LOCKED`, retries with bounded backoff, dead-letters exhausted/permanent failures, and records unique receipts. Stdout SIEM export is at-least-once; the receipt is the exactly-once ack. `/healthz.outbox` and `GET /api/outbox` are operator-visible; `POST /api/outbox/{id}/replay` requeues dead letters with audit. File/memory adapters report `outbox=disabled`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (dead letters and pending first). Processed `outbox_message` rows are pruned to that same cap on append, snapshot save, and worker ack; receipts stay as the exactly-once ack. - Control-plane PostgreSQL URLs honor `sslmode=require` / `verify-ca` / `verify-full` with rustls and Mozilla roots (certificates always verified). tokio-postgres 0.7 only parses `require`, so verification modes are rewritten to `require` before connect. `sslmode=allow` / `prefer` are rejected so the process cannot silently drop to plaintext. - Production (non-loopback) binds fail closed without `CONTROL_PLANE_DATABASE_URL`. PostgreSQL is the production control-plane authority (3NF two-word tables, default-deny row-level security, snapshot persist in one transaction). Loopback still uses the JSON file / memory adapter. `/healthz.persistence` reports `postgres`, `file`, or `memory`. The URL is a secret and is bootstrapped into the credential registry. - Live `/gateway` transactions consult in-process libcoraza when `CORAZA_LIB_PATH` is set (with `CORAZA_RULES_PATH` and/or `CORAZA_DIRECTIVES`). Missing library, missing rules, or an empty ruleset fail startup before bind. Otherwise a Coraza sidecar is consulted when `CORAZA_WAF_URL` is set. The sidecar response is parsed with the existing Coraza audit adapter (OWASP CRS authority, not a hand-rolled engine). Engine outage is fail-closed when `PROVEN_ENGINE_FAIL_CLOSED` is true. `GET /api/waf/engine-status` and `/healthz.proven_engine` report `coraza_in_process`, `coraza_sidecar`, or `ingest_hints_only`. diff --git a/README.md b/README.md index 5f79aecf..3df88fab 100644 --- a/README.md +++ b/README.md @@ -70,7 +70,7 @@ Useful environment variables: - `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`. `sslmode=require` / `verify-full` uses rustls with Mozilla roots (certificates always verified). `sslmode=disable` or omitted is plaintext. `allow`/`prefer` are rejected. `/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 +- `EVENT_LIMIT`: retained event count, default `1000`; must be greater than zero. Also caps `GET /api/outbox` and processed outbox-row retention. - `CORAZA_LIB_PATH` / `CORAZA_RULES_PATH` / `CORAZA_DIRECTIVES`: optional in-process libcoraza. A missing library or empty ruleset fails startup. `/healthz.proven_engine` reports `coraza_in_process`. - `CORAZA_WAF_URL`: optional Coraza sidecar URL used when libcoraza is not loaded - `PROVEN_ENGINE_FAIL_CLOSED`: when true, a configured engine outage returns 503 instead of degrading to builtin scoring diff --git a/docs/architecture.md b/docs/architecture.md index 0db74d85..a9b2bde1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -30,7 +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. `sslmode=require` uses rustls. 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/outbox.rs`: transactional outbox + leased workers (issue #81). Security events append incrementally with an outbox row in the same transaction. Workers claim with `SKIP LOCKED`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (processed rows pruned; receipts kept). `/healthz.outbox` is operator-visible. - `src/destination.rs`: fail-closed outbound URL policy (issue #79) for every `http`/`https` send. CIDR allowlist exceptions are per resolved address; blocking DNS is offloaded from Tokio workers. The outbound HTTP client DNS resolver returns only addresses that already passed policy (TCP peer pin / DNS-rebinding TOCTOU close). - `crates/waf-ids-core`: reusable domain models plus validation, upsert, scoring, DNSBL zone export, event retention, threat-feed freshness, KPI snapshot, and commercial readiness logic. - `/admin`: embedded web console. diff --git a/docs/doctoring/outbox-workers.md b/docs/doctoring/outbox-workers.md index 9e1bee7b..d7abaea5 100644 --- a/docs/doctoring/outbox-workers.md +++ b/docs/doctoring/outbox-workers.md @@ -40,9 +40,14 @@ https://doi.org/10.6028/NIST.SP.800-218 Production binds already require `CONTROL_PLANE_DATABASE_URL`. On that path: - `GET /healthz` reports `outbox=ready` plus pending/leased/dead-letter counts -- `GET /api/outbox` (admin read) lists messages +- `GET /api/outbox` (admin read) lists at most `EVENT_LIMIT` messages + (dead letters, then pending, then leased, then processed) - `POST /api/outbox/{message_id}/replay` (admin write) requeues dead letters +Processed `outbox_message` rows are pruned to the operator `EVENT_LIMIT` on +append, snapshot save, and worker ack. `outbox_receipt` rows stay; they are +the exactly-once ack. Dead letters are never pruned. + Loopback file/memory adapters keep in-process stdout SIEM and report `outbox=disabled`. Remaining: a non-owner runtime role, backup/restore drill, HASH partitioning, and additional diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index f334cbc3..3674d57e 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,6 +1,6 @@ # Product and technical gap baseline -Snapshot date: 2026-08-23T17:24Z (exact-head inventory of then-open GitHub PRs +Snapshot date: 2026-08-23T17:35Z (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,6 +25,7 @@ not “waiting on review/CI time”. | PR | Title | Head | Checks | Reviews | Merge blocker | | --- | --- | --- | --- | --- | --- | +| [#101](https://github.com/ContextualWisdomLab/wardnet/pull/101) | feat(store): bound outbox listing and prune processed rows | `feat/issue-81-outbox-retention` stacked on #100 | local fmt/test/clippy + two `/healthz` smokes; live `postgres_outbox_list_is_bounded_and_prunes_processed` | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 first. Do not `--admin`. Do not re-implement rustls or the outbox first slice. | | [#100](https://github.com/ContextualWisdomLab/wardnet/pull/100) | feat(store): rustls for production PostgreSQL `sslmode=require` | `feat/issue-80-postgres-rustls` stacked on #99 | local fmt/test/clippy + two `/healthz` smokes; live `sslmode=require` fails closed against plaintext postgres | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 first. Do not `--admin`. Do not re-implement the postgres gate or outbox. | | [#99](https://github.com/ContextualWisdomLab/wardnet/pull/99) | feat(store): transactional outbox and leased workers | `feat/issue-81-outbox-workers` stacked on #98 | local fmt/test/clippy + two `/healthz` smokes + postgres `/healthz.outbox=ready` prior hour | Author; Devin COMMENTED (4 threads: unbounded list still-valid; 3 info) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 first. Do not `--admin`. Do not re-implement the postgres gate. Bounded `list_outbox` is #81 remainder, not this TLS slice. | | [#98](https://github.com/ContextualWisdomLab/wardnet/pull/98) | feat(store): require PostgreSQL as the production control plane | `ea621985e276` (`feat/issue-80-postgres-control-plane`) stacked on #97 | rust + fuzz green at last snapshot; Devin 7 threads (full-snapshot rewrite, ORDER BY, TLS, RLS owner, reconnect) | Author this pass; Devin COMMENTED | Org 2-approval + self-author. ORDER BY + incremental event persist addressed on #99. rustls this pass; remaining non-owner role / backup are #80 remainder. Do not `--admin`. | @@ -115,8 +116,10 @@ Stdout SIEM export is **at-least-once**; the receipt is the exactly-once ack. Operator-visible: `/healthz.outbox` (`ready`|`disabled`), pending/leased/ dead-letter counts, `GET /api/outbox` (admin read), `POST /api/outbox/{id}/replay` (admin write + audit). Client IPs and paths in payloads are not masked. -File/memory adapters stay `outbox=disabled` with in-process stdout. Remaining -consumers: TAXII poll, Clearfolio, contextual-orchestrator on the same contract. +File/memory adapters stay `outbox=disabled` with in-process stdout. `GET /api/outbox` +is bounded to `EVENT_LIMIT`; processed rows prune to that cap; dead letters stay. +Remaining consumers: TAXII poll, Clearfolio, contextual-orchestrator on the same +contract. ### Fail-closed credentials (issue #78) — **closed on PR #94** @@ -173,19 +176,16 @@ for later loops. ## This loop’s shipped gap -Issue **#80** rustls remainder. `sslmode=require` / `verify-full` uses rustls -with Mozilla roots (certificates always verified). `allow` / `prefer` fail -closed. Driving tests: `database_url_rejects_non_postgres_and_ambiguous_sslmode`, -`require_tls_fails_closed_against_plaintext_postgres` (CI plaintext postgres -must not silently fall back), `binary_fail_closes_when_control_plane_sslmode_is_ambiguous`. -Do not re-implement #78, sidecar, pin, libcoraza, the postgres gate, or the -#81 outbox slice. +Issue **#81** still-valid #99 finding: bounded `GET /api/outbox` (`EVENT_LIMIT`) +and prune of processed `outbox_message` rows (receipts and dead letters stay). +Stacked on #100 rustls. Do not re-implement #78, sidecar, pin, libcoraza, the +postgres gate, the #81 first outbox slice, or rustls. ## Next hourly loop (do, do not report) 1. Second independent APPROVE on #91/#92. Do not `--admin`. -2. Keep #94/#95/#96/#97/#98/#99 and this rustls PR merge-ready. Merge order - #95 then #96 then #97 then #98 then #99 then this. +2. Keep #94/#95/#96/#97/#98/#99/#100 and this retention PR merge-ready. Merge + order #95 then #96 then #97 then #98 then #99 then #100 then this. 3. Next runtime gap if policy still blocks: backup/restore remainder of #80, or additional #81 consumers (TAXII / Clearfolio / orchestrator). 4. Refresh this file’s PR/Issue tables from `gh pr list` / `gh issue list`. diff --git a/src/control_plane.rs b/src/control_plane.rs index 4d3a8d86..0beab68b 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -7,8 +7,8 @@ use crate::outbox::{ self, CLAIM_BATCH, DispatchError, EVENT_SECURITY_RECORDED, EVENT_SNAPSHOT_REPLACED, - LEASE_SECONDS, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, STATUS_LEASED, - STATUS_PENDING, STATUS_PROCESSED, + LEASE_SECONDS, LIST_LIMIT, OutboxHealth, OutboxMessage, SCHEMA_VERSION, STATUS_DEAD_LETTER, + STATUS_LEASED, STATUS_PENDING, STATUS_PROCESSED, }; use std::net::IpAddr; use std::str::FromStr; @@ -334,6 +334,8 @@ static MIGRATION_GATE: std::sync::LazyLock> = std::sync::LazyLock::new pub struct PostgresPlane { client: Mutex, tenant_id: String, + /// Processed-outbox retention; mirrors operator `EVENT_LIMIT`. + event_limit: i64, } impl PostgresPlane { @@ -374,11 +376,18 @@ impl PostgresPlane { let plane = Self { client: Mutex::new(client), tenant_id: tenant_id.to_string(), + event_limit: LIST_LIMIT, }; plane.migrate().await?; Ok(plane) } + /// Use the operator-configured `EVENT_LIMIT` for processed-outbox retention. + pub fn with_event_limit(mut self, event_limit: usize) -> Self { + self.event_limit = event_limit.max(1) as i64; + self + } + async fn migrate(&self) -> Result<(), String> { let _gate = MIGRATION_GATE.lock().await; let client = self.client.lock().await; @@ -418,7 +427,7 @@ impl PostgresPlane { /// Replace the tenant snapshot in one transaction (mutation + audit + outbox). pub async fn save(&self, data: &AppData) -> Result<(), String> { let mut client = self.client.lock().await; - save_snapshot(&mut client, &self.tenant_id, data).await + save_snapshot(&mut client, &self.tenant_id, data, self.event_limit).await } /// Append one security event and its outbox row without rewriting the snapshot. @@ -441,7 +450,15 @@ impl PostgresPlane { F: Fn(&OutboxMessage) -> Result, { let mut client = self.client.lock().await; - drain_once(&mut client, &self.tenant_id, owner, now_unix, dispatch).await + drain_once( + &mut client, + &self.tenant_id, + owner, + now_unix, + self.event_limit, + dispatch, + ) + .await } pub async fn outbox_health(&self, now_unix: i64) -> Result { @@ -449,9 +466,9 @@ impl PostgresPlane { outbox_health(&mut client, &self.tenant_id, now_unix).await } - pub async fn list_outbox(&self) -> Result, String> { + pub async fn list_outbox_limited(&self, limit: i64) -> Result, String> { let mut client = self.client.lock().await; - list_outbox(&mut client, &self.tenant_id).await + list_outbox(&mut client, &self.tenant_id, limit.max(1)).await } pub async fn replay_dead_letter(&self, message_id: &str, now_unix: i64) -> Result<(), String> { @@ -509,7 +526,12 @@ async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result Result<(), String> { +async fn save_snapshot( + client: &mut Client, + tenant_id: &str, + data: &AppData, + keep: i64, +) -> Result<(), String> { let tx = client .transaction() .await @@ -703,6 +725,7 @@ async fn save_snapshot(client: &mut Client, tenant_id: &str, data: &AppData) -> } enqueue_snapshot_outbox(&tx, tenant_id, data).await?; + prune_processed_outbox(&tx, tenant_id, keep).await?; tx.commit() .await @@ -1023,6 +1046,7 @@ async fn append_security_event( }, ) .await?; + prune_processed_outbox(&tx, tenant_id, event_limit as i64).await?; tx.commit() .await @@ -1077,6 +1101,7 @@ async fn drain_once( tenant_id: &str, owner: &str, now_unix: i64, + keep: i64, dispatch: F, ) -> Result where @@ -1086,13 +1111,21 @@ where let mut processed = 0; for message in claimed { if receipt_exists(client, tenant_id, &message.idempotency_key).await? { - ack_processed(client, tenant_id, &message, "duplicate-receipt", now_unix).await?; + ack_processed( + client, + tenant_id, + &message, + "duplicate-receipt", + now_unix, + keep, + ) + .await?; processed += 1; continue; } match dispatch(&message) { Ok(evidence) => { - ack_processed(client, tenant_id, &message, &evidence, now_unix).await?; + ack_processed(client, tenant_id, &message, &evidence, now_unix, keep).await?; processed += 1; } Err(error) => { @@ -1228,6 +1261,7 @@ async fn ack_processed( message: &OutboxMessage, evidence: &str, now_unix: i64, + keep: i64, ) -> Result<(), String> { let tx = client .transaction() @@ -1268,6 +1302,7 @@ async fn ack_processed( ) .await .map_err(|error| format!("control plane ack outbox_message failed: {error}"))?; + prune_processed_outbox(&tx, tenant_id, keep).await?; tx.commit() .await .map_err(|error| format!("control plane ack commit failed: {error}"))?; @@ -1365,7 +1400,37 @@ async fn outbox_health( }) } -async fn list_outbox(client: &mut Client, tenant_id: &str) -> Result, String> { +async fn prune_processed_outbox( + client: &C, + tenant_id: &str, + keep: i64, +) -> Result<(), String> { + let keep = keep.max(1); + client + .execute( + "DELETE FROM outbox_message + WHERE tenant_id = $1 + AND message_status = $2 + AND message_id IN ( + SELECT message_id FROM ( + SELECT message_id FROM outbox_message + WHERE tenant_id = $1 AND message_status = $2 + ORDER BY created_unix DESC, message_id DESC + OFFSET $3 + ) old_processed + )", + &[&tenant_id, &STATUS_PROCESSED, &keep], + ) + .await + .map_err(|error| format!("control plane prune outbox_message failed: {error}"))?; + Ok(()) +} + +async fn list_outbox( + client: &mut Client, + tenant_id: &str, + limit: i64, +) -> Result, String> { let tx = client .transaction() .await @@ -1383,8 +1448,15 @@ async fn list_outbox(client: &mut Client, tenant_id: &str) -> Result Result { - let plane = control_plane::PostgresPlane::connect(database_url).await?; + let plane = control_plane::PostgresPlane::connect(database_url) + .await? + .with_event_limit(config.event_limit); let mut data = match plane.load().await? { Some(loaded) => loaded, None => AppData::seeded(), @@ -1131,6 +1133,7 @@ async fn list_audit_logs(State(state): State, headers: HeaderMap) -> R #[derive(Serialize)] struct OutboxListView { status: String, + limit: usize, messages: Vec, } @@ -1138,16 +1141,19 @@ async fn list_outbox(State(state): State, headers: HeaderMap) -> Respo if !admin_authenticated(&state, &headers) { return error(StatusCode::UNAUTHORIZED, "missing or invalid X-Admin-Token"); } + let limit = state.event_limit.max(1); let Some(plane) = &state.control_plane else { return Json(OutboxListView { status: "disabled".to_string(), + limit, messages: Vec::new(), }) .into_response(); }; - match plane.list_outbox().await { + match plane.list_outbox_limited(limit as i64).await { Ok(messages) => Json(OutboxListView { status: "ready".to_string(), + limit, messages, }) .into_response(), @@ -7448,6 +7454,10 @@ mod tests { ) .await; assert_eq!(body["status"], "disabled"); + assert_eq!( + body["limit"].as_u64(), + Some(AppConfig::DEFAULT_EVENT_LIMIT as u64) + ); assert_eq!(body["messages"], serde_json::json!([])); let health: HealthStatus = diff --git a/src/outbox.rs b/src/outbox.rs index 8ea626ac..a7d7ab97 100644 --- a/src/outbox.rs +++ b/src/outbox.rs @@ -14,6 +14,8 @@ pub const SCHEMA_VERSION: i32 = 1; pub const MAX_ATTEMPTS: i32 = 8; pub const LEASE_SECONDS: i64 = 30; pub const CLAIM_BATCH: i64 = 16; +/// Cap for `GET /api/outbox` and processed-row retention (mirrors `EVENT_LIMIT`). +pub const LIST_LIMIT: i64 = 1_000; pub const STATUS_PENDING: &str = "pending"; pub const STATUS_LEASED: &str = "leased"; pub const STATUS_PROCESSED: &str = "processed";