diff --git a/CHANGELOG.md b/CHANGELOG.md index 8903e02..7a473c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- PostgreSQL outbox consumers for TAXII poll, Clearfolio document submit, and contextual-orchestrator SOC analysis (issue #81 remainder). Operator-triggered HTTP leaves through `taxii.collection_polled`, `clearfolio.document_submitted`, and `soc.analysis_requested` with leased-worker retries and unique receipts. Request path returns HTTP 202 and `GET /api/outbox/{message_id}` exposes receipt evidence. Secrets never enter outbox payloads (TAXII bearer lives in the credential registry). File/memory adapters keep the previous synchronous path. Client IPs, paths, indicator values, and actor names stay unmasked. LLM analysis remains advisory and never auto-enforces. + + - PostgreSQL `security_event` is HASH-partitioned by `tenant_id` (8 children). Unpartitioned tables convert in place and keep unmasked client IPs and paths. `/healthz.event_partitions` reports the child count (0 on file/memory). Logical restore still accepts schema 2 through the current migration version; HASH does not change the snapshot shape. - PostgreSQL snapshot persist is optimistic-concurrency: `tenant_account.snapshot_version` must match the loaded token or the write returns a snapshot conflict (HTTP 409). Restores overwrite. File/memory adapters stay single-writer. - PostgreSQL control-plane runtime is `wardnet_runtime` (NOSUPERUSER, NOBYPASSRLS, not the table owner). Migrations run as the login role, then `SET ROLE` so FORCE RLS binds even when the URL user is a superuser. Missing `wardnet.tenant_id` yields no rows. DDL (`DROP TABLE`, `DISABLE ROW LEVEL SECURITY`) is denied. Logical restore accepts schema 2 through the current migration version so a role-only upgrade cannot void the last pre-upgrade backup. diff --git a/docs/architecture.md b/docs/architecture.md index d5a93f9..4bf5619 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 under `wardnet_runtime` (not superuser/owner). Snapshot persist is optimistic-concurrency on `snapshot_version` (HTTP 409). `sslmode=require` uses rustls. `security_event` is HASH-partitioned by `tenant_id` (`/healthz.event_partitions`). The JSON file adapter remains loopback/community only. -- `src/outbox.rs`: transactional outbox + leased workers (issue #81). Security events append incrementally with an outbox row in the same transaction. Workers claim with `SKIP LOCKED`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (processed rows pruned; receipts kept). `/healthz.outbox` is operator-visible. +- `src/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). `GET /api/outbox/{id}` returns receipt evidence for TAXII, Clearfolio, and SOC analysis consumers. `/healthz.outbox` is operator-visible. HTTP consumers release the PostgreSQL client lock before outbound I/O. - `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 cb6337a..3f3c7d2 100644 --- a/docs/doctoring/outbox-workers.md +++ b/docs/doctoring/outbox-workers.md @@ -50,6 +50,9 @@ the exactly-once ack. Dead letters are never pruned. Loopback file/memory adapters keep in-process stdout SIEM and report `outbox=disabled`. `security_event` HASH partitioning is on the PostgreSQL -plane. Remaining consumers: TAXII poll, Clearfolio, contextual-orchestrator -on the same message/receipt contract. Backup/restore drill is on the -PostgreSQL plane. +plane. TAXII poll, Clearfolio submit, and contextual-orchestrator analysis +use the same message/receipt contract on PostgreSQL (`GET /api/outbox/{id}` +for receipt evidence). HTTP dispatch releases the database lock for the +outbound call. Inline TAXII secrets are rejected on the durable path; +`taxii_bearer` is a credential-registry secret. Backup/restore drill is on +the PostgreSQL plane. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index c4cef81..07f1286 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-23T19:20Z (exact-head inventory of then-open GitHub PRs +Snapshot date: 2026-08-23T20: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,7 +25,8 @@ not “waiting on review/CI time”. | PR | Title | Head | Checks | Reviews | Merge blocker | | --- | --- | --- | --- | --- | --- | -| [#105](https://github.com/ContextualWisdomLab/wardnet/pull/105) | feat(store): optimistic concurrency on postgres snapshots | `feat/issue-80-optimistic-concurrency` stacked on #99 | local fmt/test/clippy + two `/healthz` smokes; live `postgres_stale_snapshot_save_conflicts` | Author this pass | Org 2-approval + self-author. #100/#103/#104 folded into #99. Merge #95 then #96 then #97 then #98 then #99 first. Do not `--admin`. | +| [#106](https://github.com/ContextualWisdomLab/wardnet/pull/106) | feat(store): outbox consumers for TAXII, Clearfolio, and orchestrator | `feat/issue-81-outbox-consumers` stacked on #105 | local fmt/test/clippy + smoke.sh + two `/healthz` and `/admin`/`/api/commercial/readiness` (2B KRW) | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #105 first. Do not `--admin`. Do not re-implement OCC or prior store slices. | +| [#105](https://github.com/ContextualWisdomLab/wardnet/pull/105) | feat(store): optimistic concurrency on postgres snapshots | `feat/issue-80-optimistic-concurrency` stacked on #99 | Devin still-valid startup-version 409 fixed this pass (`load_postgres` advances `snapshot_version` after save); local fmt/test/clippy + two `/healthz` smokes | Author; Devin COMMENTED (startup false-conflict addressed, thread resolved) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 first. Do not `--admin`. Do not re-implement HASH/role/backup. | | [#104](https://github.com/ContextualWisdomLab/wardnet/pull/104) | feat(store): HASH-partition security_event by tenant | merged into rustls stack then #99 | prior hour | Author prior hour | Folded into #99. Do not re-implement. | | [#103](https://github.com/ContextualWisdomLab/wardnet/pull/103) | feat(store): non-owner PostgreSQL runtime role after migrate | `feat/issue-80-runtime-role` stacked on #100 | still-valid Devin restore-window finding fixed this pass (`MIN_RESTORABLE_SCHEMA_VERSION=2`); local fmt/test/clippy + smokes | Author this pass; Devin COMMENTED (v3 backup voiding addressed) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 first. Do not `--admin`. Do not re-implement rustls, outbox, retention, backup, or HASH. | | [#102](https://github.com/ContextualWisdomLab/wardnet/pull/102) | feat(store): logical backup and isolated restore drill | squash-merged into #100 (`321e792`) | prior hour | Author prior hour | Folded into rustls stack. Do not re-implement. | @@ -60,8 +61,8 @@ by ruleset `18156473` (not by failing Checks). Do not `--admin` merge. | [#84](https://github.com/ContextualWisdomLab/wardnet/issues/84) | [P1] Build an immutable signed release, promotion, and rollback pipeline | high | | [#83](https://github.com/ContextualWisdomLab/wardnet/issues/83) | [P1] Add bounded distributed admission control, trusted client attribution, and overload behavior | high | | [#82](https://github.com/ContextualWisdomLab/wardnet/issues/82) | [P1] Integrate Keyverse identity, tenant authorization, consent, and human approval evidence | high (blocked) | -| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice on #99; bounded list/retention on #101** | -| [#80](https://github.com/ContextualWisdomLab/wardnet/issues/80) | [P0] Add an authoritative PostgreSQL control plane with tenant isolation and recoverable migrations | **critical — gate on #98; rustls/backup/role/HASH on #99; OCC this pass** | +| [#81](https://github.com/ContextualWisdomLab/wardnet/issues/81) | [P0] Add a transactional outbox and idempotent leased workers for external effects | **critical — first slice on #99; retention on #101; TAXII/Clearfolio/orchestrator consumers 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/role/HASH on #99; OCC on #105** | | [#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 | @@ -117,8 +118,9 @@ Logical restore accepts schema 2 through the current migration version void pre-upgrade snapshots. `security_event` is `PARTITION BY HASH (tenant_id)` with 8 children. Unpartitioned tables convert in place under `pg_advisory_lock`. `/healthz.event_partitions` is 8 on PostgreSQL, 0 on file/memory. Client IPs -and paths stay unmasked across convert. Remaining: optimistic concurrency. -Physical/PITR backups stay a DBA concern. +and paths stay unmasked across convert. Optimistic concurrency is on #105 +(`tenant_account.snapshot_version`, HTTP 409; startup save now advances the +in-memory token). Physical/PITR backups stay a DBA concern. ### Transactional outbox (issue #81) — **first slice on #99; retention on #101** @@ -133,8 +135,13 @@ dead-letter counts, `GET /api/outbox` (admin read), `POST /api/outbox/{id}/repla (admin write + audit). Client IPs and paths in payloads are not masked. File/memory adapters stay `outbox=disabled` with in-process stdout. `GET /api/outbox` is bounded to `EVENT_LIMIT`; processed rows prune to that cap on append, snapshot -save, and worker ack; dead letters stay. Remaining consumers: TAXII poll, -Clearfolio, contextual-orchestrator on the same contract. +save, and worker ack; dead letters stay. TAXII poll, Clearfolio submit, and +contextual-orchestrator analysis enqueue on PostgreSQL (`taxii.collection_polled`, +`clearfolio.document_submitted`, `soc.analysis_requested`) and return HTTP 202. +`GET /api/outbox/{id}` returns receipt evidence. Secrets never enter payloads +(`taxii_bearer` / `soc_llm_token` in the credential registry). File/memory stays +synchronous. LLM analysis is advisory and never auto-enforces. Client IPs, paths, +and indicator values stay unmasked. ### Fail-closed credentials (issue #78) — **closed on PR #94** @@ -184,25 +191,27 @@ 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. Next: same outbox - contract. +2. **contextual-orchestrator** — SOC LLM optional via `SOC_LLM_BASE_URL`; + token from credential registry (`soc_llm_token`). Same outbox contract on + PostgreSQL this pass. Do not fork routing. 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** last remainder: optimistic concurrency on -`tenant_account.snapshot_version`. Stale snapshot persist returns HTTP 409. -Restores overwrite. Do not re-implement #78, sidecar, pin, libcoraza, the -postgres gate, outbox, rustls, retention, backup/restore, runtime role, or HASH. +Issue **#81** extra consumers stacked on #105: TAXII poll, Clearfolio submit, +and contextual-orchestrator SOC analysis go through the leased outbox on +PostgreSQL. `load_postgres` also advances `snapshot_version` after the startup +save so the first management write cannot false-conflict (Devin #105). Do not +re-implement #78, sidecar, pin, libcoraza, the postgres gate, outbox, rustls, +retention, backup/restore, runtime role, HASH, or OCC. ## 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 OCC PR merge-ready. Merge order - #94 independently; #95 then #96 then #97 then #98 then #99 then this. -3. Next runtime gap if policy still blocks: extra #81 consumers (TAXII / - Clearfolio / orchestrator). +2. Keep #94 independently; #95 then #96 then #97 then #98 then #99 then #105 + then #106 merge-ready. Do not `--admin`. +3. Next runtime gap if policy still blocks: signed release/promotion (#84) or + Keyverse identity (#82) after the postgres stack. 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 8212ec3..005fcd7 100644 --- a/src/control_plane.rs +++ b/src/control_plane.rs @@ -11,6 +11,7 @@ use crate::outbox::{ STATUS_LEASED, STATUS_PENDING, STATUS_PROCESSED, }; use serde::{Deserialize, Serialize}; +use std::future::Future; use std::net::IpAddr; use std::str::FromStr; use std::time::Instant; @@ -752,6 +753,7 @@ impl PostgresPlane { append_security_event(&mut client, &self.tenant_id, event, event_limit).await } + #[cfg(test)] pub async fn drain_once( &self, owner: &str, @@ -773,6 +775,160 @@ impl PostgresPlane { .await } + /// Claim due messages, dispatch without holding the client lock, then ack. + /// HTTP consumers must not pin the PostgreSQL connection during outbound I/O. + pub async fn drain_due_async( + &self, + owner: &str, + now_unix: i64, + dispatch: F, + ) -> Result + where + F: Fn(OutboxMessage) -> Fut, + Fut: Future>, + { + let claimed = { + let mut client = self.client.lock().await; + claim_batch(&mut client, &self.tenant_id, owner, now_unix).await? + }; + let mut processed = 0; + for message in claimed { + let duplicate = { + let mut client = self.client.lock().await; + receipt_exists(&mut client, &self.tenant_id, &message.idempotency_key).await? + }; + if duplicate { + let mut client = self.client.lock().await; + ack_processed( + &mut client, + &self.tenant_id, + &message, + "duplicate-receipt", + now_unix, + self.event_limit, + ) + .await?; + processed += 1; + continue; + } + match dispatch(message.clone()).await { + Ok(evidence) => { + let mut client = self.client.lock().await; + ack_processed( + &mut client, + &self.tenant_id, + &message, + &evidence, + now_unix, + self.event_limit, + ) + .await?; + processed += 1; + } + Err(error) => { + let mut client = self.client.lock().await; + fail_claimed(&mut client, &self.tenant_id, &message, now_unix, &error).await?; + } + } + } + Ok(processed) + } + + /// Persist an operator-triggered external effect (TAXII / Clearfolio / SOC). + /// Secrets must not appear in `payload_json`. + pub async fn enqueue_effect( + &self, + event_type: &'static str, + aggregate_id: &str, + payload_json: String, + ) -> Result { + let created_unix = unix_now_i64(); + let hash = outbox::payload_hash(&payload_json); + let unique = format!("{created_unix}:{hash}"); + let (message_id, idempotency_key) = + outbox::effect_ids(event_type, &self.tenant_id, &unique); + let mut client = self.client.lock().await; + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane effect transaction failed: {error}"))?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&self.tenant_id], + ) + .await + .map_err(|error| format!("control plane tenant context failed: {error}"))?; + insert_outbox( + &tx, + &self.tenant_id, + &OutboxInsert { + message_id: message_id.clone(), + aggregate_id: aggregate_id.to_string(), + aggregate_version: created_unix, + event_type, + created_unix, + payload_json, + payload_hash: hash, + idempotency_key, + }, + ) + .await?; + prune_processed_outbox(&tx, &self.tenant_id, self.event_limit).await?; + tx.commit() + .await + .map_err(|error| format!("control plane effect commit failed: {error}"))?; + Ok(message_id) + } + + /// Load one outbox row plus receipt evidence when processed. + pub async fn get_outbox_item( + &self, + message_id: &str, + ) -> Result)>, String> { + let mut client = self.client.lock().await; + let tx = client + .transaction() + .await + .map_err(|error| format!("control plane get-outbox transaction failed: {error}"))?; + tx.execute( + "SELECT set_config('wardnet.tenant_id', $1, true)", + &[&self.tenant_id], + ) + .await + .map_err(|error| format!("control plane tenant context failed: {error}"))?; + let row = tx + .query_opt( + "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 AND message_id = $2", + &[&self.tenant_id, &message_id], + ) + .await + .map_err(|error| format!("control plane get outbox_message failed: {error}"))?; + let Some(row) = row else { + tx.commit() + .await + .map_err(|error| format!("control plane get-outbox commit failed: {error}"))?; + return Ok(None); + }; + let message = row_to_outbox(&row, &self.tenant_id); + let evidence = tx + .query_opt( + "SELECT receipt_evidence FROM outbox_receipt + WHERE tenant_id = $1 AND message_id = $2", + &[&self.tenant_id, &message_id], + ) + .await + .map_err(|error| format!("control plane get outbox_receipt failed: {error}"))? + .map(|row| row.get::<_, String>(0)); + tx.commit() + .await + .map_err(|error| format!("control plane get-outbox commit failed: {error}"))?; + Ok(Some((message, evidence))) + } + 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 @@ -1599,6 +1755,7 @@ async fn insert_outbox( Ok(()) } +#[cfg(test)] async fn drain_once( client: &mut Client, tenant_id: &str, @@ -3147,4 +3304,53 @@ mod tests { assert_eq!(winner.routes[0].path_prefix, "/occ-a"); assert!(winner.snapshot_version > loaded_a.snapshot_version); } + + #[tokio::test] + async fn postgres_enqueues_external_effect_and_async_drain_records_receipt() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("outbox-effect"); + 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"); + let payload = serde_json::json!({ + "objects_url": "https://taxii.example/api1/collections/c/objects/", + "feed_id": "taxii-lab", + "path": "/gateway/login", + "client_ip": "198.51.100.20" + }) + .to_string(); + let message_id = plane + .enqueue_effect(crate::outbox::EVENT_TAXII_POLLED, "taxii-lab", payload) + .await + .expect("enqueue taxii poll"); + let processed = plane + .drain_due_async( + "effect-worker", + now.saturating_add(30), + |message| async move { + assert_eq!(message.event_type, crate::outbox::EVENT_TAXII_POLLED); + assert!(message.payload_json.contains("198.51.100.20")); + assert!(message.payload_json.contains("/gateway/login")); + Ok("taxii-ack:unmasked".into()) + }, + ) + .await + .expect("async drain"); + assert_eq!(processed, 1); + let (message, evidence) = plane + .get_outbox_item(&message_id) + .await + .expect("get item") + .expect("enqueued"); + assert_eq!(message.message_status, STATUS_PROCESSED); + assert_eq!(evidence.as_deref(), Some("taxii-ack:unmasked")); + } } diff --git a/src/credentials.rs b/src/credentials.rs index 66223ab..adc0b1f 100644 --- a/src/credentials.rs +++ b/src/credentials.rs @@ -12,6 +12,8 @@ use std::{collections::HashMap, io::ErrorKind, path::Path}; 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"; +pub const CRED_SOC_LLM_TOKEN: &str = "soc_llm_token"; +pub const CRED_TAXII_BEARER: &str = "taxii_bearer"; /// Where secret-bearing credentials were loaded from (never includes values). #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] @@ -90,7 +92,13 @@ impl CredentialRegistry { path.display() ) })?; - for key in [CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL] { + for key in [ + CRED_ADMIN_TOKEN, + CRED_ADMIN_TOKENS, + CRED_CONTROL_PLANE_URL, + CRED_SOC_LLM_TOKEN, + CRED_TAXII_BEARER, + ] { if let Some(raw) = file_map.get(key) { let text = json_value_as_nonempty_string(raw); if let Some(text) = text { @@ -139,6 +147,21 @@ impl CredentialRegistry { Ok(Self { values, source }) } + + /// Fill a missing secret from bootstrap transport. No-op when the key is + /// already present or the value is empty. Never logs the value. + pub fn load_optional_secret(&mut self, name: &str, env_value: Option) { + if self.values.contains_key(name) { + return; + } + let Some(value) = env_value.filter(|value| !value.is_empty()) else { + return; + }; + self.values.insert(name.to_string(), value); + if self.source == CredentialSource::None { + self.source = CredentialSource::Env; + } + } } fn json_value_as_nonempty_string(value: &serde_json::Value) -> Option { @@ -187,6 +210,22 @@ mod tests { assert!(!registry.has_admin_auth()); } + #[test] + fn load_optional_secret_fills_missing_keys_only() { + let mut registry = + CredentialRegistry::bootstrap_secrets(None, Some("secret".to_string()), None, None) + .unwrap(); + registry.load_optional_secret(CRED_SOC_LLM_TOKEN, Some("llm-token".into())); + registry.load_optional_secret(CRED_SOC_LLM_TOKEN, Some("ignored".into())); + registry.load_optional_secret(CRED_TAXII_BEARER, Some(String::new())); + assert_eq!( + registry.get_credential(CRED_SOC_LLM_TOKEN), + Some("llm-token") + ); + assert!(registry.get_credential(CRED_TAXII_BEARER).is_none()); + assert_eq!(registry.get_credential(CRED_ADMIN_TOKEN), Some("secret")); + } + #[test] fn file_overrides_env_per_key() { let dir = std::env::temp_dir().join(format!( diff --git a/src/lib.rs b/src/lib.rs index 4cc3411..a4c9a9b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -48,8 +48,8 @@ mod stix_import; mod suricata_eve; mod taxii; pub use credentials::{ - CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL, CredentialRegistry, - CredentialSource, + CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL, CRED_SOC_LLM_TOKEN, + CRED_TAXII_BEARER, CredentialRegistry, CredentialSource, }; pub use destination::{DestinationPolicy, HostResolver, SystemHostResolver}; pub use proven_engine::{ProvenEngineConfig, ProvenEngineOutcome}; @@ -81,6 +81,8 @@ pub struct AppState { // Optional LLM SOC-analysis backend (OpenAI-compatible, e.g. the // contextual-orchestrator gateway). `None` unless configured. soc_llm: Option, + /// Optional TAXII Bearer for durable polls. Never logged or written to outbox. + taxii_bearer: Option, /// In-path Coraza consult (in-process libcoraza and/or sidecar). proven_engine: ProvenEngineConfig, /// Fail-closed destination policy for every outbound http/https call. @@ -199,6 +201,7 @@ impl AppState { max_body_bytes: 1_048_576, clearfolio: None, soc_llm: None, + taxii_bearer: None, proven_engine: ProvenEngineConfig::disabled(), destination: DestinationPolicy::production(), resolver: Arc::new(SystemHostResolver), @@ -228,6 +231,12 @@ impl AppState { self } + /// Optional TAXII Bearer from the credential registry. Never written to outbox payloads. + pub fn with_taxii_bearer(mut self, token: Option) -> Self { + self.taxii_bearer = token.filter(|value| !value.is_empty()); + self + } + /// Configure the in-path Coraza adapter (sidecar and/or libcoraza). /// Builder-style. pub fn with_proven_engine(mut self, config: ProvenEngineConfig) -> Self { @@ -623,6 +632,7 @@ pub fn build_app(state: AppState) -> Router { .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}", get(get_outbox_item)) .route("/api/outbox/{message_id}/replay", post(replay_outbox)) .route("/api/backup", get(get_backup).post(restore_backup)) .route("/api/backup/drill", post(backup_drill)) @@ -770,26 +780,84 @@ async fn clearfolio_submit( format!("unknown document kind: {kind}"), ); }; + if state.control_plane.is_some() { + let body_text = String::from_utf8_lossy(&bytes).into_owned(); + let actor = audit_actor(&state, &headers); + let intent = ClearfolioSubmitIntent { + kind: kind.clone(), + filename, + body_text, + actor: actor.clone(), + }; + return enqueue_external_effect( + &state, + actor, + outbox::EVENT_CLEARFOLIO_SUBMITTED, + &kind, + &intent, + ) + .await; + } + match execute_clearfolio_submit(&state, &config, filename, bytes).await { + Ok((status, body)) => clearfolio_bytes_response(status, body), + Err(error) => dispatch_http_error(error), + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct ClearfolioSubmitIntent { + kind: String, + filename: String, + body_text: String, + actor: String, +} + +fn clearfolio_bytes_response(status: u16, body: Vec) -> Response { + let status = StatusCode::from_u16(status).unwrap_or(StatusCode::BAD_GATEWAY); + (status, [("content-type", "application/json")], body).into_response() +} + +fn dispatch_http_error(failure: outbox::DispatchError) -> Response { + let message = failure.as_str(); + let status = match &failure { + outbox::DispatchError::Permanent(text) + if text.contains("destination") || text.starts_with("HTTP 4") => + { + StatusCode::BAD_REQUEST + } + _ => StatusCode::BAD_GATEWAY, + }; + error(status, message.to_string()) +} + +async fn execute_clearfolio_submit( + state: &AppState, + config: &ClearfolioConfig, + filename: String, + bytes: Vec, +) -> Result<(u16, Vec), outbox::DispatchError> { let part = reqwest::multipart::Part::bytes(bytes) .file_name(filename) .mime_str("text/plain") .expect("text/plain is a valid MIME type"); let form = reqwest::multipart::Form::new().part("file", part); let submit_url = clearfolio_submit_url(&config.base_url); - if let Err(message) = state.assert_outbound(&submit_url).await { - return error(StatusCode::BAD_REQUEST, message); - } + state + .assert_outbound(&submit_url) + .await + .map_err(outbox::DispatchError::Permanent)?; let mut request = state.http.post(submit_url).multipart(form); - for (name, value) in clearfolio_tenant_headers(&config) { + for (name, value) in clearfolio_tenant_headers(config) { request = request.header(name, value); } - match request.send().await { - Ok(response) => clearfolio_relay_json(response).await, - Err(err) => error( - StatusCode::BAD_GATEWAY, - format!("clearfolio request failed: {err}"), - ), - } + let response = request.send().await.map_err(|err| { + outbox::DispatchError::Transient(format!("clearfolio request failed: {err}")) + })?; + let status = response.status().as_u16(); + let body = response.bytes().await.unwrap_or_default().to_vec(); + let preview = String::from_utf8_lossy(&body); + outbox::classify_http_status(status, &preview)?; + Ok((status, body)) } /// Proxies one Clearfolio job-status read (tenant headers applied server-side), @@ -946,45 +1014,71 @@ async fn soc_analyze( format!("unknown event id: {}", request.event_id), ); }; - let body = soc_llm_chat_body(&config.model, &event); + if state.control_plane.is_some() { + let actor = audit_actor(&state, &headers); + let intent = SocAnalyzeIntent { + event: event.clone(), + actor: actor.clone(), + }; + return enqueue_external_effect( + &state, + actor, + outbox::EVENT_SOC_ANALYSIS_REQUESTED, + &event.id.to_string(), + &intent, + ) + .await; + } + match execute_soc_analyze(&state, &config, &event).await { + Ok(response) => Json(response).into_response(), + Err(error) => dispatch_http_error(error), + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct SocAnalyzeIntent { + event: SecurityEvent, + actor: String, +} + +async fn execute_soc_analyze( + state: &AppState, + config: &SocLlmConfig, + event: &SecurityEvent, +) -> Result { + let body = soc_llm_chat_body(&config.model, event); let endpoint = format!( "{}/v1/chat/completions", config.base_url.trim_end_matches('/') ); - if let Err(message) = state.assert_outbound(&endpoint).await { - return error(StatusCode::BAD_REQUEST, message); - } + state + .assert_outbound(&endpoint) + .await + .map_err(outbox::DispatchError::Permanent)?; let response = state .http .post(endpoint) .bearer_auth(&config.token) .json(&body) .send() - .await; - match response { - Ok(response) => match response.json::().await { - Ok(json) => match soc_llm_extract_content(&json) { - Some(analysis) => Json(SocAnalyzeResponse { - event_id: event.id, - model: config.model, - analysis, - }) - .into_response(), - None => error( - StatusCode::BAD_GATEWAY, - "llm response missing choices[0].message.content", - ), - }, - Err(err) => error( - StatusCode::BAD_GATEWAY, - format!("llm response read failed: {err}"), - ), - }, - Err(err) => error( - StatusCode::BAD_GATEWAY, - format!("llm request failed: {err}"), - ), - } + .await + .map_err(|err| outbox::DispatchError::Transient(format!("llm request failed: {err}")))?; + let status = response.status().as_u16(); + let json = response.json::().await.map_err(|err| { + outbox::DispatchError::Transient(format!("llm response read failed: {err}")) + })?; + let preview = json.to_string(); + outbox::classify_http_status(status, &preview)?; + let analysis = soc_llm_extract_content(&json).ok_or_else(|| { + outbox::DispatchError::Permanent( + "llm response missing choices[0].message.content".to_string(), + ) + })?; + Ok(SocAnalyzeResponse { + event_id: event.id, + model: config.model.clone(), + analysis, + }) } async fn healthz(State(state): State) -> Json { @@ -1180,6 +1274,20 @@ struct OutboxListView { messages: Vec, } +#[derive(Serialize)] +struct OutboxItemView { + status: String, + message: outbox::OutboxMessage, + receipt_evidence: Option, +} + +#[derive(Serialize)] +struct OutboxAcceptedView { + status: String, + message_id: String, + event_type: String, +} + 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"); @@ -1204,6 +1312,83 @@ async fn list_outbox(State(state): State, headers: HeaderMap) -> Respo } } +async fn get_outbox_item( + State(state): State, + PathParam(message_id): PathParam, + 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 error( + StatusCode::SERVICE_UNAVAILABLE, + "outbox item reads require the PostgreSQL control plane", + ); + }; + match plane.get_outbox_item(&message_id).await { + Ok(Some((message, receipt_evidence))) => Json(OutboxItemView { + status: "ready".to_string(), + message, + receipt_evidence, + }) + .into_response(), + Ok(None) => error( + StatusCode::NOT_FOUND, + format!("unknown outbox message {message_id}"), + ), + Err(message) => persist_error(message), + } +} + +async fn enqueue_external_effect( + state: &AppState, + actor: String, + event_type: &'static str, + aggregate_id: &str, + payload: &impl Serialize, +) -> Response { + let Some(plane) = &state.control_plane else { + return error( + StatusCode::SERVICE_UNAVAILABLE, + "outbox enqueue requires the PostgreSQL control plane", + ); + }; + let payload_json = + serde_json::to_string(payload).expect("outbox effect payload is JSON-serializable"); + match plane + .enqueue_effect(event_type, aggregate_id, payload_json) + .await + { + Ok(message_id) => { + if let Err(error) = state + .mutate_and_persist(|data| { + record_successful_audit_log( + data, + actor, + event_type, + "outbox_message", + message_id.clone(), + ); + }) + .await + { + eprintln!("failed to audit outbox enqueue {message_id}: {error}"); + } + ( + StatusCode::ACCEPTED, + Json(OutboxAcceptedView { + status: "accepted".to_string(), + message_id, + event_type: event_type.to_string(), + }), + ) + .into_response() + } + Err(message) => persist_error(message), + } +} + async fn replay_outbox( State(state): State, PathParam(message_id): PathParam, @@ -1842,6 +2027,16 @@ struct TaxiiPollResult { last_updated_unix: u64, } +#[derive(Debug, Clone, Serialize, Deserialize)] +struct TaxiiPollIntent { + objects_url: String, + feed_id: String, + source: String, + ttl_seconds: u64, + added_after: Option, + actor: String, +} + /// Poll a TAXII 2.1 collection objects endpoint and import STIX indicators. /// Secrets in the request body are never written to audit logs. async fn poll_taxii_collection( @@ -1880,6 +2075,47 @@ async fn poll_taxii_collection( Err(message) => return error(StatusCode::BAD_REQUEST, message), }; + if state.control_plane.is_some() { + let has_inline_secret = request + .bearer_token + .as_deref() + .map(str::trim) + .is_some_and(|value| !value.is_empty()) + || request + .username + .as_deref() + .map(str::trim) + .is_some_and(|value| !value.is_empty()) + || request + .password + .as_deref() + .map(str::trim) + .is_some_and(|value| !value.is_empty()); + if has_inline_secret { + return error( + StatusCode::BAD_REQUEST, + "durable TAXII poll cannot store credentials in the outbox; configure taxii_bearer in the credential registry", + ); + } + let actor = audit_actor(&state, &headers); + let intent = TaxiiPollIntent { + objects_url, + feed_id: request.feed_id.trim().to_string(), + source: request.source.trim().to_string(), + ttl_seconds: request.ttl_seconds, + added_after: request.added_after.clone(), + actor: actor.clone(), + }; + return enqueue_external_effect( + &state, + actor, + outbox::EVENT_TAXII_POLLED, + &intent.feed_id, + &intent, + ) + .await; + } + let body_text = match fetch_taxii_objects( &state, &objects_url, @@ -1983,7 +2219,15 @@ async fn fetch_taxii_objects( PHISHING_DATABASE_FETCH_TIMEOUT_SECS, )); - if let Some(token) = bearer_token.map(str::trim).filter(|s| !s.is_empty()) { + let bearer = bearer_token + .map(str::trim) + .filter(|s| !s.is_empty()) + .or(state + .taxii_bearer + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty())); + if let Some(token) = bearer { request = request.bearer_auth(token); } else if let Some(user) = username.map(str::trim).filter(|s| !s.is_empty()) { request = request.basic_auth(user, password); @@ -2018,6 +2262,63 @@ async fn fetch_taxii_objects( String::from_utf8(bytes).map_err(|error| format!("TAXII response is not valid UTF-8: {error}")) } +fn taxii_fetch_error_to_dispatch(message: String) -> outbox::DispatchError { + let lower = message.to_ascii_lowercase(); + if lower.contains("timed out") + || lower.contains("timeout") + || lower.contains("failed to poll") + || lower.contains("failed to read") + || lower.contains("http 429") + || lower.contains("http 5") + { + outbox::DispatchError::Transient(message) + } else { + outbox::DispatchError::Permanent(message) + } +} + +async fn execute_taxii_poll( + state: &AppState, + intent: &TaxiiPollIntent, +) -> Result { + let body_text = fetch_taxii_objects(state, &intent.objects_url, None, None, None) + .await + .map_err(taxii_fetch_error_to_dispatch)?; + let stix_json = taxii::stix_json_from_taxii_response(&body_text) + .map_err(outbox::DispatchError::Permanent)?; + let material = + stix_import::parse_stix_document(&stix_json, intent.source.trim(), intent.ttl_seconds) + .map_err(outbox::DispatchError::Permanent)?; + let feed = ThreatFeedImport { + feed_id: intent.feed_id.clone(), + source: intent.source.clone(), + ttl_seconds: intent.ttl_seconds, + threats: material.threats, + dnsbl: material.dnsbl, + }; + validate_threat_feed_import(&feed) + .map_err(|message| outbox::DispatchError::Permanent(message.to_string()))?; + let skipped_objects = material.skipped_objects; + let result = + apply_threat_feed_import(state, intent.actor.clone(), "poll_taxii_collection", feed) + .await + .map_err(|message| { + if message.contains("snapshot conflict") { + outbox::DispatchError::Transient(message) + } else { + outbox::DispatchError::Permanent(message) + } + })?; + Ok(TaxiiPollResult { + feed_id: result.feed_id, + objects_url: intent.objects_url.clone(), + upserted_threats: result.upserted_threats, + upserted_dnsbl: result.upserted_dnsbl, + skipped_objects, + last_updated_unix: result.last_updated_unix, + }) +} + /// Ingest Suricata EVE JSON (single object, array, or NDJSON). Admin-auth only. /// Maps `event_type=alert` records into gateway security events for SOC export. #[derive(Debug, Serialize)] @@ -3308,13 +3609,13 @@ input,select{font:inherit;min-height:44px;padding:0 12px;border:1px solid var(--

POST admin-authenticated MISP Event/attribute JSON to /api/threat-intel/misp (optional query: feed_id, source, ttl_seconds). Maps IDS-worthy attributes (ip-src/ip-dst, domain, url, composites, hashes) into threats/DNSBL; attributes with to_ids=false are skipped. Live MISP REST pull is a follow-up.

TAXII 2.1 collection poll

-

POST admin-authenticated JSON to /api/threat-intel/taxii/poll with objects_url (or api_root+collection_id), optional Basic/Bearer credentials, and optional added_after. Fetches TAXII objects, normalizes to STIX, and upserts threats/DNSBL. Credentials are never written to audit logs.

+

POST admin-authenticated JSON to /api/threat-intel/taxii/poll with objects_url (or api_root+collection_id) and optional added_after. PostgreSQL enqueues taxii.collection_polled (202) for the leased worker; file/memory still fetches on the request path. Inline Basic/Bearer is memory-only — durable polls use taxii_bearer in the credential registry. Credentials are never written to outbox payloads or audit logs. Poll GET /api/outbox/{message_id} for receipt evidence. Indicator values stay unmasked.

OpenCTI threat intelligence

POST admin-authenticated OpenCTI GraphQL/list export JSON to /api/threat-intel/opencti (optional query: feed_id, source, ttl_seconds). Maps IPv4/IPv6, Domain-Name, Url, file hashes, and STIX indicators into threats/DNSBL. Live OpenCTI GraphQL pull is a follow-up.