feat(governance): confidential computing, MoE routing, PQC WORM — runnable assurance

Extend the runnable-assurance suite into four net-new verifiable domains:

- Confidential computing (env-01): rego/attestation_gate.rego enforces SEV-SNP/TDX
  + vTPM PCR_MATCH admission (golden measurement, TCB anti-rollback, fresh nonce),
  with structured denial reasons; 9 OPA tests (21/21 total). TLA+
  AdmissionWithAttestation proves no T0 workload runs without valid attestation
  and that TCB rollback / PCR drift force eviction (TLC, 64 states, no error).

- MoE routing stability (rte-01): routing/sara_acr_router.py implements SARA
  (load-aware gating) + ACR (capacity regulation); demonstrates baseline expert
  collapse (entropy 0.38, load ratio 5.6) vs stabilized (entropy 0.99, ratio 1.25)
  satisfying entropy/load/drop invariants; 4 pytests.

- PQC WORM (cry-02): kafka/pqc_worm_logger_v2.py replaces the HMAC placeholder with
  real CRYSTALS-Dilithium (ML-DSA-65 / FIPS 204) signatures + tamper-evident hash
  chain + S3 Object Lock COMPLIANCE retention; verify_chain() detects entry
  mutation, batch reorder, and signature forgery; 6 pytests.

- OSCAL: new catalog_sentinel_v24_env_rte.json adding ENV and RTE control groups,
  each backed by a runnable artifact.

run_runnable_assurance.sh now runs 8 checks (all PASS); CI + docs + requirements
updated. No regressions in existing governance tests.

Author: OneFineStarstuff

.github/workflows/runnable-assurance.yml

@@ -40,7 +40,7 @@ jobs:
           java-version: '17'
 
       - name: Install Python deps
-        run: pip install pyyaml jsonschema
+        run: pip install pyyaml jsonschema dilithium-py pytest
 
       - name: Install OPA
         run: |
@@ -71,5 +71,10 @@ jobs:
           circom circuits/src1_concentration_bound.circom --r1cs --wasm --sym --O0 -o circuits/
           circom circuits/src_fair1_reason_code_check.circom --r1cs --wasm --sym --O0 -o circuits/
 
+      - name: Unit tests (routing + PQC WORM)
+        run: |
+          pytest governance_artifacts/routing/test_sara_acr_router.py -q
+          pytest governance_artifacts/kafka/test_pqc_worm_logger_v2.py -q
+
       - name: Run runnable assurance suite
         run: bash governance_artifacts/run_runnable_assurance.sh

governance_artifacts/RUNNABLE_ASSURANCE.md

@@ -23,11 +23,21 @@ Runs all five checks below and fails fast on any error.
 
 | # | Check | Tool | Backs OSCAL control | Regime anchor |
 |---|-------|------|---------------------|---------------|
-| 1 | Deny-by-default release gate + high-impact credit gate | `opa test` (12 tests) | release-gate semantics; `con-07` quorum | SR 11-7, EU AI Act Art. 14, ECOA, GDPR Art. 22 |
+| 1 | Release gate + credit gate + confidential-computing attestation gate (PCR_MATCH) | `opa test` (21 tests) | release-gate, `con-07`, `env-01` | SR 11-7, EU AI Act Art. 14/15, ECOA, GDPR Art. 22, DORA |
 | 2 | Containment one-way ratchet & terminal-actuation quorum | TLA+ `tlc2.TLC` | `con-04`, `con-07` | EU AI Act Art. 14, DORA resilience testing |
-| 3 | GC-IR cross-target conformance (policy ⇔ circuit ⇔ expectation) | `opa eval` + Circom witness | obligation `ob-ecoa-adverse-reason-codes` | ECOA, GDPR Art. 22, EU AI Act Art. 13 |
-| 4 | Systemic-risk concentration bound (HHI) zk proof | Circom + Groth16 (snarkjs) | `cry-05` | Basel op-risk, systemic telemetry |
-| 5 | Governance artifact schema validation | Python validator | manifest/schema integrity | OSCAL, evidence logging (EU AI Act Art. 12) |
+| 3 | Attested admission — no T0 workload runs without fresh valid attestation; TCB rollback / PCR drift force eviction | TLA+ `tlc2.TLC` | `env-01` | EU AI Act Art. 15, DORA ICT risk, NIST AI RMF |
+| 4 | GC-IR cross-target conformance (policy ⇔ circuit ⇔ expectation) | `opa eval` + Circom witness | obligation `ob-ecoa-adverse-reason-codes` | ECOA, GDPR Art. 22, EU AI Act Art. 13 |
+| 5 | Systemic-risk concentration bound (HHI) zk proof | Circom + Groth16 (snarkjs) | `cry-05` | Basel op-risk, systemic telemetry |
+| 6 | SARA/ACR MoE routing stabilization invariants (entropy / load balance / drop) | Python simulator + pytest | `rte-01` | EU AI Act Art. 15 robustness, SR 11-7 |
+| 7 | PQC WORM audit log — real CRYSTALS-Dilithium (ML-DSA-65) signatures + tamper-evident hash chain + S3 Object Lock retention | Python (`dilithium-py`) + pytest | `cry-02` | DORA, EU AI Act Art. 12 logging |
+| 8 | Governance artifact schema validation | Python validator | manifest/schema integrity | OSCAL, evidence logging (EU AI Act Art. 12) |
+
+### New control groups (`oscal/catalog_sentinel_v24_env_rte.json`)
+
+- **ENV — Confidential Computing & Attested Execution**: `env-01` (hardware-attested
+  admission for T0/T1 via SEV-SNP / TDX + vTPM PCR_MATCH; runtime TCB-rollback and
+  PCR-drift eviction), `env-02` (enclave-bound ML-DSA key custody).
+- **RTE — MoE Routing Stability**: `rte-01` (SARA/ACR stabilization invariants).
 
 ## 1. OPA policy tests — `rego/`
 
@@ -90,6 +100,57 @@ cd governance_artifacts/zk && bash run_src1_proof.sh
 > production-secure. A production deployment requires a multi-party trusted setup
 > (or a transparent system such as PLONK/STARK as noted in the schema enum).
 
+## 6. Confidential-computing attestation gate — `rego/attestation_gate.rego` + `tla/AdmissionWithAttestation.tla`
+
+The `PCR_MATCH=TRUE` assertion that recurs throughout the master docs is now
+*enforced*, not merely stated. The Rego gate (`sentinel.attestation`) admits a
+T0/T1 workload only when it presents a SEV-SNP or TDX report with a verified
+signature, fresh anti-replay nonce, a launch measurement in the golden registry,
+platform TCB at/above the ratified minimum (no rollback), and a vTPM PCR quote
+matching the policy digest. The TLA+ spec proves the *temporal* guarantee: across
+all 64 initial evidence combinations, no workload reaches `RUNNING` without a
+valid attestation, and runtime TCB rollback or PCR drift forces `EVICTED`.
+
+```bash
+opa test governance_artifacts/rego/                       # includes 9 attestation tests
+cd governance_artifacts/tla
+java -cp tools/tla2tools.jar tlc2.TLC -config AdmissionWithAttestation.cfg AdmissionWithAttestation.tla
+```
+
+## 7. SARA/ACR MoE routing stabilization — `routing/sara_acr_router.py`
+
+Defines and demonstrates two stack-specific mechanisms (not external standards):
+**SARA** (Stabilized Adaptive Routing — load-aware gating bias + temperature) and
+**ACR** (Adaptive Capacity Regulation — per-expert capacity factor with overflow
+handling). The simulator shows that under skewed gating a naive top-k router
+collapses (normalised entropy ≈ 0.38, load ratio ≈ 5.6) and *violates* the
+`rte-01` invariants, while SARA+ACR holds entropy ≈ 0.99 and load ratio ≈ 1.25,
+*satisfying* all invariants (entropy ≥ 0.80, load ratio ≤ 1.60, drop ≤ 0.02).
+
+```bash
+python3 governance_artifacts/routing/sara_acr_router.py
+pytest governance_artifacts/routing/test_sara_acr_router.py -q   # 4 tests
+```
+
+## 8. PQC WORM audit log — `kafka/pqc_worm_logger_v2.py`
+
+Replaces the original HMAC placeholder with **real CRYSTALS-Dilithium (ML-DSA-65,
+FIPS 204)** signatures over canonical batch payloads, linked in a tamper-evident
+**hash chain** (`prev_batch_hash`), with an S3 Object Lock COMPLIANCE-mode
+retention record per batch. `verify_chain()` re-validates every signature and link
+and returns a supervisory-ready report; the demo proves that entry mutation,
+batch reordering, and signature forgery are all detected.
+
+```bash
+python3 governance_artifacts/kafka/pqc_worm_logger_v2.py
+pytest governance_artifacts/kafka/test_pqc_worm_logger_v2.py -q  # 6 tests
+```
+
+> ML-DSA-65 here is provided by the pure-Python `dilithium-py` reference
+> implementation — correct and FIPS-204-aligned, but **not** constant-time or
+> side-channel-hardened. Production signing belongs in the env-02 enclave using a
+> validated cryptographic module.
+
 ## Reproducing from a clean checkout
 
 ```bash
@@ -101,7 +162,7 @@ curl -L -o ~/.local/bin/circom https://github.com/iden3/circom/releases/download
 # TLA+ tools
 curl -L -o governance_artifacts/tla/tools/tla2tools.jar https://github.com/tlaplus/tlaplus/releases/download/v1.7.4/tla2tools.jar
 # Python
-pip install pyyaml jsonschema
+pip install pyyaml jsonschema dilithium-py
 # Run everything
 bash governance_artifacts/run_runnable_assurance.sh
 ```

governance_artifacts/kafka/pqc_worm_logger_v2.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""
+PQC WORM Logger v2 — CRYSTALS-Dilithium (ML-DSA-65 / FIPS 204) signed audit log.
+================================================================================
+Backs OSCAL control cry-02 (hybrid PQC signatures on governance event envelopes)
+and the Kafka/S3 Object Lock WORM evidence pipeline.
+
+Improvements over the original pqc_worm_logger.py (which used an HMAC placeholder):
+
+  * REAL post-quantum signatures via ML-DSA-65 (CRYSTALS-Dilithium), the exact
+    algorithm named in cry-02. Each batch is signed; verification uses the public
+    key only (asymmetric, unlike the prior HMAC).
+  * Tamper-evident HASH CHAIN: each batch records prev_batch_hash, so any
+    reordering, deletion, or mutation of historic batches is detectable.
+  * WORM semantics modelled: an immutable "retention" record (S3 Object Lock
+    COMPLIANCE mode + retain-until date) accompanies each committed batch.
+  * verify_chain() re-validates every signature AND the hash linkage; returns a
+    machine-readable report suitable for supervisory evidence.
+
+Falls back is intentionally absent: if dilithium_py is unavailable the import
+fails loudly rather than silently downgrading crypto.
+"""
+from __future__ import annotations
+import hashlib
+import json
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from dilithium_py.ml_dsa import ML_DSA_65
+
+ALG = "ML-DSA-65"  # CRYSTALS-Dilithium, FIPS 204
+RETENTION_YEARS = 7  # Basel/DORA-style retention default
+
+
+def _canon(obj: Any) -> bytes:
+    """Deterministic canonical JSON for signing/hashing."""
+    return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _sha256(b: bytes) -> str:
+    return "sha256:" + hashlib.sha256(b).hexdigest()
+
+
+@dataclass
+class CommittedBatch:
+    batch_id: str
+    timestamp: str
+    entries: list[dict]
+    prev_batch_hash: str
+    payload_hash: str
+    signature_hex: str
+    retention: dict  # S3 Object Lock model
+
+    def to_dict(self) -> dict:
+        return {
+            "batch_id": self.batch_id,
+            "timestamp": self.timestamp,
+            "entries": self.entries,
+            "prev_batch_hash": self.prev_batch_hash,
+            "payload_hash": self.payload_hash,
+            "signature_alg": ALG,
+            "signature_hex": self.signature_hex,
+            "retention": self.retention,
+        }
+
+
+@dataclass
+class PQCWormLoggerV2:
+    bucket: str = "kacg-gsifi-worm-evidence-prod"
+    batch_size_threshold: int = 10
+    _pk: bytes = field(default=None, repr=False)
+    _sk: bytes = field(default=None, repr=False)
+    _pending: list[dict] = field(default_factory=list, repr=False)
+    _chain: list[CommittedBatch] = field(default_factory=list, repr=False)
+    _genesis: str = "sha256:" + "0" * 64
+
+    def __post_init__(self):
+        if self._pk is None or self._sk is None:
+            self._pk, self._sk = ML_DSA_65.keygen()
+
+    @property
+    def public_key_fingerprint(self) -> str:
+        return _sha256(self._pk)
+
+    def add_entry(self, entry: dict) -> CommittedBatch | None:
+        self._pending.append(entry)
+        if len(self._pending) >= self.batch_size_threshold:
+            return self.commit_batch()
+        return None
+
+    def commit_batch(self) -> CommittedBatch | None:
+        if not self._pending:
+            return None
+        entries = self._pending
+        self._pending = []
+
+        prev_hash = self._chain[-1].payload_hash if self._chain else self._genesis
+        ts = datetime.now(timezone.utc).isoformat()
+        batch_id = hashlib.sha256(f"{ts}{len(self._chain)}".encode()).hexdigest()[:16]
+
+        # Payload binds entries + the previous hash (chain linkage).
+        payload = {"batch_id": batch_id, "timestamp": ts,
+                   "entries": entries, "prev_batch_hash": prev_hash}
+        payload_bytes = _canon(payload)
+        payload_hash = _sha256(payload_bytes)
+
+        # REAL ML-DSA signature over the canonical payload.
+        signature = ML_DSA_65.sign(self._sk, payload_bytes)
+
+        retain_until = (datetime.now(timezone.utc)
+                        + timedelta(days=365 * RETENTION_YEARS)).isoformat()
+        retention = {
+            "mode": "COMPLIANCE",            # S3 Object Lock COMPLIANCE mode
+            "retain_until": retain_until,
+            "legal_hold": False,
+            "bucket": self.bucket,
+        }
+
+        batch = CommittedBatch(
+            batch_id=batch_id, timestamp=ts, entries=entries,
+            prev_batch_hash=prev_hash, payload_hash=payload_hash,
+            signature_hex=signature.hex(), retention=retention,
+        )
+        self._chain.append(batch)
+        return batch
+
+    def verify_chain(self) -> dict:
+        """Re-verify every signature and the hash linkage. Returns a report."""
+        errors: list[str] = []
+        prev = self._genesis
+        for i, b in enumerate(self._chain):
+            if b.prev_batch_hash != prev:
+                errors.append(f"batch[{i}] {b.batch_id}: broken hash chain link")
+            payload = {"batch_id": b.batch_id, "timestamp": b.timestamp,
+                       "entries": b.entries, "prev_batch_hash": b.prev_batch_hash}
+            payload_bytes = _canon(payload)
+            if _sha256(payload_bytes) != b.payload_hash:
+                errors.append(f"batch[{i}] {b.batch_id}: payload hash mismatch")
+            if not ML_DSA_65.verify(self._pk, payload_bytes, bytes.fromhex(b.signature_hex)):
+                errors.append(f"batch[{i}] {b.batch_id}: ML-DSA signature INVALID")
+            prev = b.payload_hash
+        return {
+            "alg": ALG,
+            "public_key_fingerprint": self.public_key_fingerprint,
+            "batches": len(self._chain),
+            "status": "VERIFIED" if not errors else "FAILED",
+            "errors": errors,
+        }
+
+
+def _demo() -> int:
+    log = PQCWormLoggerV2(batch_size_threshold=3)
+    for i in range(7):
+        log.add_entry({
+            "event_id": f"evt-{i:03d}",
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "control_id": "cry-02",
+            "decision": ["allow", "deny", "escalate"][i % 3],
+        })
+    log.commit_batch()  # flush remainder
+
+    report = log.verify_chain()
+    print("PQC WORM Logger v2 —", ALG)
+    print(f"  public key fingerprint: {report['public_key_fingerprint'][:23]}...")
+    print(f"  committed batches      : {report['batches']}")
+    print(f"  chain verification     : {report['status']}")
+    assert report["status"] == "VERIFIED", report
+
+    # Tamper test: mutate a historic entry and confirm detection.
+    log._chain[0].entries[0]["decision"] = "TAMPERED"
+    bad = log.verify_chain()
+    print(f"  after tamper           : {bad['status']} ({len(bad['errors'])} error(s))")
+    assert bad["status"] == "FAILED", "tamper went undetected!"
+    print("  RESULT: signatures + hash chain verify; tampering detected")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(_demo())

governance_artifacts/kafka/test_pqc_worm_logger_v2.py

@@ -0,0 +1,63 @@
+"""Tests for PQC WORM Logger v2 (ML-DSA-65 signed, hash-chained audit log)."""
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+import pytest  # noqa: E402
+
+from pqc_worm_logger_v2 import PQCWormLoggerV2, ALG  # noqa: E402
+
+
+def _fill(log, n):
+    for i in range(n):
+        log.add_entry({"event_id": f"e{i}", "control_id": "cry-02", "decision": "allow"})
+    log.commit_batch()
+
+
+def test_alg_is_ml_dsa_65():
+    assert ALG == "ML-DSA-65"
+
+
+def test_chain_verifies_clean():
+    log = PQCWormLoggerV2(batch_size_threshold=3)
+    _fill(log, 7)
+    report = log.verify_chain()
+    assert report["status"] == "VERIFIED"
+    assert report["batches"] == 3
+    assert not report["errors"]
+
+
+def test_retention_is_compliance_worm():
+    log = PQCWormLoggerV2(batch_size_threshold=2)
+    _fill(log, 2)
+    batch = log._chain[0]
+    assert batch.retention["mode"] == "COMPLIANCE"
+    assert "retain_until" in batch.retention
+
+
+def test_tamper_entry_detected():
+    log = PQCWormLoggerV2(batch_size_threshold=2)
+    _fill(log, 4)
+    log._chain[0].entries[0]["decision"] = "TAMPERED"
+    report = log.verify_chain()
+    assert report["status"] == "FAILED"
+
+
+def test_chain_reorder_detected():
+    log = PQCWormLoggerV2(batch_size_threshold=2)
+    _fill(log, 6)
+    # Swap two batches -> hash linkage breaks.
+    log._chain[0], log._chain[1] = log._chain[1], log._chain[0]
+    report = log.verify_chain()
+    assert report["status"] == "FAILED"
+
+
+def test_signature_forgery_detected():
+    log = PQCWormLoggerV2(batch_size_threshold=2)
+    _fill(log, 2)
+    sig = bytearray(bytes.fromhex(log._chain[0].signature_hex))
+    sig[0] ^= 0xFF
+    log._chain[0].signature_hex = sig.hex()
+    report = log.verify_chain()
+    assert report["status"] == "FAILED"