From bb35a0c1157d068d01e3d1f3a305f2405936d176 Mon Sep 17 00:00:00 2001 From: Rui He Date: Wed, 22 Jul 2026 02:19:24 -0400 Subject: [PATCH 1/3] =?UTF-8?q?feat(store):=20tier-2=20value=20ledger=20?= =?UTF-8?q?=E2=80=94=20encrypted-at-rest=20per-step=20inputs=20+=20monitor?= =?UTF-8?q?=20detail=20(M0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Makes each governed tool-use inspectable and reproducible WITHOUT weakening the value-free wire/monitor (docs/pg-provenance-ledger.md, M0). - infra/store/valuecrypt.py: envelope encryption (X25519+HKDF+AES-GCM, no new dep). Fresh per-blob DEK wrapped to a recipient set (node standalone; +mothership at fleet join via rewrap without re-encrypting). values_sha commits the PLAINTEXT. - backend: run_values table (sqlite live; PG inert-parallel, raw BYTEA + JSONB) + record_values/get_values on both adapters; store_selftest interface-conformance covers them. - mirror: new value-free `values` queue op (single drain writer; backend only, never the chain) + values_sha added to _SAFE_EVENT_KEYS. - govd Store: node recipient key at boot (0600, default-on, inert if disabled); record_values encrypts + enqueues at the ONE WS point plaintext exists, binds values_sha into the tier-1 chain step event; decrypt_values for the operator view. New operator-gated endpoint GET /monitor/values/ (node-local decrypt). - monitors: local dashboard run tab + fleetdash run page each gain a "tool-use detail" reveal that decrypts step inputs on request (textContent-only, no inline-onclick sink); fleetdash _EVENT_KEYS + proxy allowlist gain values_sha / monitor/values. - tests: test_value_ledger.py (9) + 3 integrated delegated cases. Full suite green. Secrets never recorded (declared-subset, secret-filtered, *_FILE pointers stay node-local). Off-node at rest = ciphertext only. Co-Authored-By: Claude Fable 5 --- infra/govern/govd.py | 86 ++++++++++++++++- infra/govern/govd_dashboard.html | 25 +++++ infra/store/backend.py | 65 ++++++++++++- infra/store/mirror.py | 15 ++- infra/store/valuecrypt.py | 159 +++++++++++++++++++++++++++++++ infra/tool/fleetdash.py | 50 +++++++++- tests/test_param_delegated.py | 25 +++++ tests/test_value_ledger.py | 105 ++++++++++++++++++++ 8 files changed, 520 insertions(+), 10 deletions(-) create mode 100644 infra/store/valuecrypt.py create mode 100644 tests/test_value_ledger.py diff --git a/infra/govern/govd.py b/infra/govern/govd.py index bdeaa42..3c86f1e 100644 --- a/infra/govern/govd.py +++ b/infra/govern/govd.py @@ -386,6 +386,69 @@ def __init__(self, root, max_runs=MAX_RUNS, cfg=None): # exception-isolated. The in-memory runs + ledger.json remain the authoritative hot-path state. from infra.store.mirror import StoreMirror self.mirror = StoreMirror(self.root, cfg) + self._init_value_ledger(cfg or {}) + + def _init_value_ledger(self, cfg): + """Tier-2 value ledger (docs/pg-provenance-ledger.md): load/gen the node's X25519 recipient key and the + recipient set the per-step var_values are ENCRYPTED to. Standalone: the node's own key is the sole + recipient; a fleet handshake later appends the mothership oversight pubkey. INERT unless enabled — a + node with `value_ledger.enabled` false records exactly as before (no values persisted).""" + self.value_key_file = None + self.value_recipients = [] # raw X25519 pubkeys the blobs are sealed to (>=1 to record) + vc = cfg.get("value_ledger") or {} + if not vc.get("enabled", True): # default-on locally; the tier stays value-free-at-rest regardless + return + try: + from infra.store import valuecrypt + self.value_key_file = os.path.expanduser( + vc.get("node_key_file") or os.path.join(self.root, ".value-key")) + node_pub = valuecrypt.generate_node_key(self.value_key_file) + recips = [node_pub] + for hx in (vc.get("recipients") or []): # extra recipient pubkeys (hex) — e.g. mothership oversight + try: + recips.append(bytes.fromhex(hx)) + except ValueError: + sys.stderr.write("[govd] value-ledger: ignoring malformed recipient pubkey\n") + self.value_recipients = recips + except Exception as e: # never fail server boot on the optional value tier + sys.stderr.write(f"[govd] value ledger unavailable, continuing value-free only: {e}\n") + self.value_key_file, self.value_recipients = None, [] + + def record_values(self, run_id, step, ts, values): + """Encrypt the (already declared-subset, secret-filtered) per-step `values` to the recipient set and + enqueue them into the tier-2 ledger. Returns the plaintext commitment `values_sha` (to bind into the + tier-1 chain step event) or None when the tier is inert / empty. NEVER raises into the decision path.""" + if not self.value_recipients or not values: + return None + try: + from infra.store import valuecrypt + blob = valuecrypt.encrypt(values, self.value_recipients) + self.mirror.record_values(run_id, step, ts, blob["sha"], blob) + return blob["sha"] + except Exception as e: + sys.stderr.write(f"[govd] value-ledger record (run {run_id} step {step}) skipped: {e}\n") + return None + + def decrypt_values(self, run_id): + """Operator-side plaintext view of one run's tier-2 values (the monitor 'detail of each tool use'). + Decrypts each step's blob with the NODE's recipient key — available only where that key lives (this + node). Returns [{step, ts, values_sha, values|error}] ordered by step; [] when the tier is inert.""" + if not self.value_key_file: + return [] + from infra.store import valuecrypt + be = getattr(self.mirror, "backend", None) + if be is None: + return [] + sk = valuecrypt.load_private(self.value_key_file) + out = [] + for row in be.get_values(run_id): + item = {"step": row["step"], "ts": row["ts"], "values_sha": row["values_sha"]} + try: + item["values"] = valuecrypt.decrypt(row["blob"], sk) + except Exception as e: + item["error"] = f"{type(e).__name__}: {e}"[:200] + out.append(item) + return out def _path(self, run_id): return os.path.join(self.root, run_id, "ledger.json") @@ -943,6 +1006,17 @@ def do_GET(self): if detail: # attach the porters this run runs (public chip code) detail["sources"] = porter_sources(detail.get("skill"), detail.get("perk"), detail.get("seq")) return self._json(200 if detail else 404, detail or {"error": "unknown run_id"}) + if path.startswith("/monitor/values/"): + # tier-2 DETAIL of each tool use: the DECRYPTED per-step var_values for one run. Operator-gated + # (monitor token) AND node-local — decryption needs the node's recipient key, which never leaves + # the node; a mothership decrypts its own replica app-side with the oversight key. The value-free + # /monitor/run is unchanged; this is the deliberate, authenticated plaintext view. + if not self._monitor_authed(cfg): + return self._json(403, {"error": "missing/invalid monitor token"}) + rid = urllib.parse.unquote(path.split("/monitor/values/", 1)[1]) + steps = store.decrypt_values(rid) + return self._json(200, {"run_id": rid, "steps": steps, + "note": "decrypted node-side; values are the declared, non-secret step inputs"}) if path.startswith("/trace/"): # P5-T05: the run's cross-plane trace (claim→grant→step spans under one trace id) by run_id — # value-free, monitor-gated like /monitor/run. @@ -1359,6 +1433,11 @@ def _ws_oversight(self): if k in declared and k not in RESERVED_ENV and not k.startswith("CWS_SECRET_") and not (SECRET_KEY.search(k) and not k.endswith(POINTER_SUFFIX))} + # tier-2 value ledger: record the (declared-subset, secret-filtered) values ENCRYPTED + # at rest — this is the ONE place plaintext values exist server-side — and bind their + # plaintext commitment `values_sha` into the tier-1 chain step event. The wire + the + # value-free chain/monitor stay value-free; only the sha (a hash) crosses into them. + values_sha = store.record_values(bound, step, now(), var_values) d_reply, d_event = delegate.execute_step( rec0, step, psha, exod_socket=sock, grant_key=gk, exod_pub=epub, base=getattr(self.server, "exec_workspace", os.path.join(store.root, "_work")), @@ -1366,8 +1445,11 @@ def _ws_oversight(self): token_proof=msg.get("token_proof"), # ACL M2: agent-relayed token-possession proof var_values=var_values) # caller NON-secret values (declared subset, ACL-gated) if d_event: - store.append(bound, {**d_event, "ts": now(), - "span": tracing.child_span((rec0 or {}).get("traceparent"))}) + ev = {**d_event, "ts": now(), + "span": tracing.child_span((rec0 or {}).get("traceparent"))} + if values_sha: + ev["values_sha"] = values_sha + store.append(bound, ev) ws_send(self.wfile, json.dumps({"type": "executed", "step": msg.get("step"), **d_reply})) finally: store.release_step(bound, step) diff --git a/infra/govern/govd_dashboard.html b/infra/govern/govd_dashboard.html index 8266249..3466832 100644 --- a/infra/govern/govd_dashboard.html +++ b/infra/govern/govd_dashboard.html @@ -335,6 +335,31 @@

cyberware · govd monitor

for(const s of states){ inSeq.add(String(s.n)); list.appendChild(block(s.n, s.tool, s.state, byStep[String(s.n)])); } for(const k of Object.keys(byStep)){ if(!inSeq.has(k)) list.appendChild(block(k==="?"?null:k, null, null, byStep[k])); } ss.b.appendChild(list); panel.appendChild(ss.s); + // Tier-2 tool-use DETAIL: each step's values_sha commits its declared, non-secret inputs into the value-free + // chain; the values live encrypted at rest and are decrypted here on request, node-side (the node holds the + // recipient key). Secrets are never recorded — they stay *_FILE pointers. Only shown when a run actually + // recorded values (a delegated, parameterized run). + const anyV=events.some(e=>e.type==="step_result"&&e.values_sha); + if(anyV){ + const vs=sect("Tool-use detail — decrypted step inputs"); + vs.b.appendChild(el("div","note","Each step's values_sha commits the exact declared, non-secret inputs. Reveal decrypts them here, node-side — secrets stay *_FILE pointers and are never recorded.")); + const btn=el("button","efbtn","reveal tool-use detail ↗"), out=el("div","steps"); + btn.onclick=()=>{ out.textContent="loading…"; + fetch("/monitor/values/"+enc(d.run_id)+"?token="+enc(token),{cache:"no-store"}) + .then(r=>r.json()).then(v=>{ out.textContent=""; + const steps=(v&&v.steps)||[]; if(!steps.length){ out.appendChild(el("div","empty","no recorded values")); return; } + for(const s of steps){ const blk=el("div","stepblk"), hd=el("div","stephd"); + hd.appendChild(el("span","pill","step "+s.step)); hd.appendChild(el("span","t",String(s.values_sha||"").slice(0,16))); blk.appendChild(hd); + const tl=el("div","tl"); + if(s.error){ tl.appendChild(el("div","note","decrypt error: "+s.error)); } + const vv=s.values||{}; const keys=Object.keys(vv).sort(); + if(!keys.length && !s.error) tl.appendChild(el("div","note","no values")); + for(const k of keys){ const row=el("div"); row.appendChild(el("span","ev",k)); row.appendChild(el("span",null,String(vv[k]))); tl.appendChild(row); } + blk.appendChild(tl); out.appendChild(blk); + } + }).catch(e=>{ out.textContent="fetch failed: "+e; }); }; + vs.b.appendChild(btn); vs.b.appendChild(out); panel.appendChild(vs.s); + } } function flowTab(panel, d){ diff --git a/infra/store/backend.py b/infra/store/backend.py index 58b425f..4d95750 100644 --- a/infra/store/backend.py +++ b/infra/store/backend.py @@ -41,6 +41,13 @@ def get_origin(self, run_id): raise NotImplementedError def index_record(self, run_id, seq, prev, link_digest, kind, ts, plan_sha, fields) -> dict: raise NotImplementedError def index_decision(self, summary) -> dict: raise NotImplementedError + # tier-2 value ledger (docs/pg-provenance-ledger.md): the per-step var_values, ENVELOPE-ENCRYPTED at rest + # (infra/store/valuecrypt.py). `blob` is valuecrypt's self-describing dict (ciphertext + per-recipient DEK + # wraps); `values_sha` is the PLAINTEXT commitment also bound into the tier-1 chain. record_values is an + # UPSERT keyed on (run_id, step) — a re-sent step is a no-op duplicate. get_values returns the blobs + # (ciphertext) ONLY; decryption is the caller's, app-side, with a recipient key the store never holds. + def record_values(self, run_id, step, ts, values_sha, blob) -> dict: raise NotImplementedError + def get_values(self, run_id) -> list: raise NotImplementedError # [{step, ts, values_sha, blob}] by step def rows(self, run_id) -> list: raise NotImplementedError # ordered by seq def head(self, run_id): raise NotImplementedError # {seq, link_digest} | None def run_ids(self) -> list: raise NotImplementedError @@ -100,6 +107,10 @@ def open(self): rid INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT, ts TEXT, link_digest TEXT, fields TEXT)""") self.cx.execute("""CREATE TABLE IF NOT EXISTS idx_origin( run_id TEXT PRIMARY KEY, plan_sha TEXT)""") + # tier-2 value ledger: ciphertext blob (JSON) + plaintext commitment. UPSERT on (run_id, step). + self.cx.execute("""CREATE TABLE IF NOT EXISTS run_values( + run_id TEXT, step TEXT, ts TEXT, values_sha TEXT, blob TEXT, + PRIMARY KEY(run_id, step))""") return self def set_origin(self, run_id, plan_sha): @@ -127,6 +138,21 @@ def index_decision(self, summary) -> dict: (summary.get("run_id"), summary.get("ts", ""), ld, _canon(summary))) return {"backend": self.name, "status": "indexed", "run_id": summary.get("run_id")} + def record_values(self, run_id, step, ts, values_sha, blob) -> dict: + with self._lock: + cur = self.cx.execute("INSERT OR IGNORE INTO run_values VALUES(?,?,?,?,?)", + (run_id, str(step), ts, values_sha, _canon(blob))) + rowcount = cur.rowcount + return {"backend": self.name, "run_id": run_id, "step": str(step), + "status": "indexed" if rowcount == 1 else "duplicate"} + + def get_values(self, run_id) -> list: + with self._lock: + cur = self.cx.execute( + "SELECT step, ts, values_sha, blob FROM run_values WHERE run_id=? ORDER BY step", (run_id,)) + fetched = cur.fetchall() + return [{"step": r[0], "ts": r[1], "values_sha": r[2], "blob": json.loads(r[3])} for r in fetched] + def rows(self, run_id) -> list: with self._lock: cur = self.cx.execute( @@ -148,7 +174,7 @@ def run_ids(self) -> list: def reset(self): with self._lock: - for t in ("idx_record", "idx_decision", "idx_origin"): + for t in ("idx_record", "idx_decision", "idx_origin", "run_values"): self.cx.execute(f"DELETE FROM {t}") # ── P5-T04: the single-writer lease (advisory lock) — atomic via BEGIN IMMEDIATE ────────────────────── @@ -328,6 +354,13 @@ def open(self): c.execute("""CREATE TABLE IF NOT EXISTS idx_decision( rid BIGSERIAL PRIMARY KEY, run_id TEXT, ts TEXT, link_digest TEXT, fields JSONB)""") c.execute("""CREATE TABLE IF NOT EXISTS idx_origin(run_id TEXT PRIMARY KEY, plan_sha TEXT)""") + # tier-2 value ledger: raw ciphertext bytes (the hash-able artifact) + a queryable JSONB projection + # of the SAME blob (JSONB normalizes bytes, so it is never the verification input). node_id defaults + # to '' on a standalone node; the fleet publication stamps it at share time. + c.execute("""CREATE TABLE IF NOT EXISTS run_values( + node_id TEXT NOT NULL DEFAULT '', run_id TEXT, step TEXT, ts TEXT, + values_sha TEXT, blob_raw BYTEA, blob JSONB, + PRIMARY KEY(node_id, run_id, step))""") return self def _unconf(self, **kw): @@ -372,6 +405,34 @@ def index_decision(self, summary) -> dict: (summary.get("run_id"), summary.get("ts", ""), ld, _canon(summary))) return {"backend": self.name, "status": "indexed", "run_id": summary.get("run_id")} + def record_values(self, run_id, step, ts, values_sha, blob) -> dict: + if not self.configured(): + return self._unconf(run_id=run_id, step=str(step)) + try: + raw = _canon(blob).encode() + with self.cx.cursor() as c: + c.execute("INSERT INTO run_values(node_id, run_id, step, ts, values_sha, blob_raw, blob) " + "VALUES('', %s,%s,%s,%s,%s,%s) " + "ON CONFLICT (node_id, run_id, step) DO NOTHING", + (run_id, str(step), ts, values_sha, raw, _canon(blob))) + status = "indexed" if c.rowcount == 1 else "duplicate" + return {"backend": self.name, "run_id": run_id, "step": str(step), "status": status} + except Exception as e: + return {"backend": self.name, "status": "error", "detail": f"{type(e).__name__}: {e}"[:300]} + + def get_values(self, run_id) -> list: + if not self.configured(): + return [] + with self.cx.cursor() as c: + c.execute("SELECT step, ts, values_sha, blob_raw FROM run_values WHERE run_id=%s ORDER BY step", + (run_id,)) + out = [] + for r in c.fetchall(): + raw = r[3] + blob = json.loads(raw.tobytes() if hasattr(raw, "tobytes") else raw) + out.append({"step": r[0], "ts": r[1], "values_sha": r[2], "blob": blob}) + return out + def rows(self, run_id) -> list: if not self.configured(): return [] @@ -405,7 +466,7 @@ def reset(self): if not self.configured(): return with self.cx.cursor() as c: - for t in ("idx_record", "idx_decision", "idx_origin"): + for t in ("idx_record", "idx_decision", "idx_origin", "run_values"): c.execute(f"DELETE FROM {t}") # ── P5-T04: the single-writer lease — ONE atomic conditional upsert (no race window) ────────────────── diff --git a/infra/store/mirror.py b/infra/store/mirror.py index 8d7e86c..853524d 100644 --- a/infra/store/mirror.py +++ b/infra/store/mirror.py @@ -27,7 +27,7 @@ _SAFE_RUN_KEYS = ("ts", "skill", "perk", "decision", "destructive", "approved", "plan_sha", "var_keys", "principal", "cost", "snippet_shas", "traceparent") _SAFE_EVENT_KEYS = ("ts", "type", "step", "status", "exit", "reason", "span", "authority", "keyid", - "snippet_shas", "meter", "traceparent") + "snippet_shas", "meter", "traceparent", "values_sha") def value_free_run(rec): @@ -97,6 +97,14 @@ def decision(self, summary): """Enqueue one verdict for the decisions chain/index.""" self._put({"op": "decision", "summary": summary}) + def record_values(self, run_id, step, ts, values_sha, blob): + """Enqueue one ENCRYPTED tier-2 value blob for a step. Value-free-at-rest: the blob is already + ciphertext (infra/store/valuecrypt.py) and only its plaintext commitment `values_sha` — NOT the + values — is what also rides the tier-1 chain (as the step event's values_sha). Goes ONLY to the + derived backend, never the chain (the chain stays the value-free artifact of record).""" + self._put({"op": "values", "run_id": run_id, "step": str(step), "ts": ts, + "values_sha": values_sha, "blob": blob}) + def _put(self, job): if self.chain is None: return @@ -123,6 +131,11 @@ def _drain(self): self.chain.append_decision(job["summary"]) if self.backend is not None: self.backend.index_decision(job["summary"]) + elif job.get("op") == "values": + # tier-2 encrypted value ledger — the DERIVED backend only, NEVER the value-free chain. + if self.backend is not None: + self.backend.record_values(job["run_id"], job["step"], job["ts"], + job["values_sha"], job["blob"]) else: rec = self.chain.append_record(job["run_id"], job["plan_sha"], job["kind"], job["fields"]) if self.backend is not None: diff --git a/infra/store/valuecrypt.py b/infra/store/valuecrypt.py new file mode 100644 index 0000000..9476009 --- /dev/null +++ b/infra/store/valuecrypt.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""infra/store/valuecrypt.py — envelope encryption for the tier-2 value ledger (docs/pg-provenance-ledger.md). + +The tier-1 chain stays value-free; tier 2 records the (already declared-subset, secret-filtered) per-step +`var_values` so a run is INSPECTABLE + REPRODUCIBLE — but never in plaintext at rest. Each value blob gets a +fresh per-blob data key (DEK); the DEK is wrapped to a RECIPIENT SET of X25519 public keys (standalone: the +node's own key; post fleet-handshake: + the mothership oversight key). So: + + * off-node at rest = ciphertext only — a replica / backup / over-granted DB role yields nothing readable; + * fleet join re-wraps the DEK to a new recipient (a few bytes) WITHOUT re-encrypting the data; + * any recipient decrypts OFFLINE from the blob alone — no per-row key fetch. + +The commitment `values_sha` (bound into the tier-1 chain) is sha256 of the canonical PLAINTEXT — NEVER the +ciphertext (a fresh nonce per blob would make a ciphertext hash meaningless). Verify: decrypt -> canon -> hash +-> match the chain. Confidentiality (this module) and integrity (the chain) stay orthogonal. + +Primitives are all from `cryptography` (already required for the Ed25519 grant/exod identities): X25519 ECDH ++ HKDF-SHA256 to derive a per-recipient key-wrapping key, AES-256-GCM to wrap the DEK and to seal the blob. +No new dependency. HONEST LIMIT: govd sees values transiently (it is the writer); a compromised LIVE govd host +reads them regardless — this defends the AT-REST surface (backups, replicas, roles), which is exactly what +federation + backup create. +""" +from __future__ import annotations +import hashlib +import json +import os + +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey, X25519PublicKey +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +_WRAP_INFO = b"cyberware/valuecrypt/dek-wrap/v1" # HKDF domain separation — wrap KEK vs anything else +_BLOB_AAD = b"cyberware/valuecrypt/blob/v1" # AES-GCM AAD binds the ciphertext to this scheme+version + + +def canon(values: dict) -> bytes: + """The canonical plaintext bytes hashed into `values_sha` AND sealed as the blob. Stable: sorted keys, + tight separators — so the same {KEY:value} map always yields the same sha and the same sealed bytes.""" + return json.dumps(values, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + + +def values_sha(values: dict) -> str: + """sha256 of the canonical PLAINTEXT — the commitment bound into the tier-1 chain (never the ciphertext).""" + return hashlib.sha256(canon(values)).hexdigest() + + +def keyid(pub_raw: bytes) -> str: + """A short stable id for a recipient public key — the map key in `dek_wraps` and the audit label.""" + return "x25519:" + hashlib.sha256(pub_raw).hexdigest()[:16] + + +# ── key material ───────────────────────────────────────────────────────────────────────────────────────── +def generate_node_key(path: str) -> bytes: + """Create a fresh X25519 private key at `path` (chmod 600) if absent; return the RAW 32-byte PUBLIC key. + Idempotent — an existing key is loaded, never overwritten (rotation is an explicit operator act, not a + restart side effect). The private key lives beside govd's other key material, NEVER in the DB / a + replicated path.""" + path = os.path.expanduser(path) + if os.path.exists(path): + return load_public(path) + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + sk = X25519PrivateKey.generate() + raw = sk.private_bytes(serialization.Encoding.Raw, serialization.PrivateFormat.Raw, + serialization.NoEncryption()) + # write 0600: create with a restrictive mode from the start (never a world-readable window mid-write) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + os.write(fd, raw) + finally: + os.close(fd) + return sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + +def load_private(path: str) -> X25519PrivateKey: + with open(os.path.expanduser(path), "rb") as f: + return X25519PrivateKey.from_private_bytes(f.read()) + + +def load_public(path: str) -> bytes: + """The RAW 32-byte public key for the private key at `path` (recipient set membership + keyid).""" + return load_private(path).public_key().public_bytes( + serialization.Encoding.Raw, serialization.PublicFormat.Raw) + + +# ── envelope ───────────────────────────────────────────────────────────────────────────────────────────── +def _wrap_kek(recipient_pub_raw: bytes) -> tuple[bytes, bytes]: + """Derive a one-shot key-wrapping key for `recipient` via ephemeral-static X25519 + HKDF. Returns + (kek, ephemeral_pub_raw) — the ephemeral public rides in the wrap so the recipient can re-derive the KEK.""" + eph = X25519PrivateKey.generate() + shared = eph.exchange(X25519PublicKey.from_public_bytes(recipient_pub_raw)) + kek = HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=_WRAP_INFO).derive(shared) + eph_pub = eph.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + return kek, eph_pub + + +def _unwrap_kek(sk: X25519PrivateKey, eph_pub_raw: bytes) -> bytes: + shared = sk.exchange(X25519PublicKey.from_public_bytes(eph_pub_raw)) + return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=_WRAP_INFO).derive(shared) + + +def encrypt(values: dict, recipient_pubs: list[bytes]) -> dict: + """Seal `values` under a fresh DEK, wrap that DEK to EACH recipient public key. Returns a self-describing + blob dict (JSON-friendly, hex fields) plus the plaintext commitment `values_sha`: + + {"v":1, "sha": , "nonce": , "ct": , + "wraps": { : {"eph": , "nonce": , "wrap": } , ... }} + + At least one recipient is required (a blob no one can open is a bug, not a feature).""" + if not recipient_pubs: + raise ValueError("valuecrypt.encrypt: recipient set is empty — a value blob needs >=1 recipient") + pt = canon(values) + dek = AESGCM.generate_key(bit_length=256) + blob_nonce = os.urandom(12) + ct = AESGCM(dek).encrypt(blob_nonce, pt, _BLOB_AAD) + wraps = {} + for pub in recipient_pubs: + kek, eph_pub = _wrap_kek(pub) + wrap_nonce = os.urandom(12) + wrapped = AESGCM(kek).encrypt(wrap_nonce, dek, _WRAP_INFO) + wraps[keyid(pub)] = {"eph": eph_pub.hex(), "nonce": wrap_nonce.hex(), "wrap": wrapped.hex()} + return {"v": 1, "sha": hashlib.sha256(pt).hexdigest(), "nonce": blob_nonce.hex(), + "ct": ct.hex(), "wraps": wraps} + + +def rewrap(blob: dict, sk: X25519PrivateKey, new_recipient_pubs: list[bytes]) -> dict: + """Add recipients to an existing blob WITHOUT re-encrypting the data: recover the DEK with a key we hold, + wrap it to each new recipient, merge into `wraps`. This is the fleet-join backfill — a few bytes per blob. + Returns a new blob dict (the input is not mutated).""" + dek = _recover_dek(blob, sk) + wraps = dict(blob.get("wraps") or {}) + for pub in new_recipient_pubs: + kek, eph_pub = _wrap_kek(pub) + wrap_nonce = os.urandom(12) + wraps[keyid(pub)] = {"eph": eph_pub.hex(), "nonce": wrap_nonce.hex(), + "wrap": AESGCM(kek).encrypt(wrap_nonce, dek, _WRAP_INFO).hex()} + out = dict(blob) + out["wraps"] = wraps + return out + + +def _recover_dek(blob: dict, sk: X25519PrivateKey) -> bytes: + """Find the wrap addressed to our key and unwrap the DEK. Raises if we are not a recipient.""" + my_id = keyid(sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)) + w = (blob.get("wraps") or {}).get(my_id) + if w is None: + raise ValueError("valuecrypt.decrypt: this key is not a recipient of the blob") + kek = _unwrap_kek(sk, bytes.fromhex(w["eph"])) + return AESGCM(kek).decrypt(bytes.fromhex(w["nonce"]), bytes.fromhex(w["wrap"]), _WRAP_INFO) + + +def decrypt(blob: dict, sk: X25519PrivateKey) -> dict: + """Open a blob with a recipient private key -> the original `values` dict. Also re-verifies the plaintext + commitment (`sha`) so a tampered ciphertext is caught here, not silently returned.""" + dek = _recover_dek(blob, sk) + pt = AESGCM(dek).decrypt(bytes.fromhex(blob["nonce"]), bytes.fromhex(blob["ct"]), _BLOB_AAD) + if hashlib.sha256(pt).hexdigest() != blob.get("sha"): + raise ValueError("valuecrypt.decrypt: plaintext commitment mismatch — blob tampered") + return json.loads(pt.decode("utf-8")) diff --git a/infra/tool/fleetdash.py b/infra/tool/fleetdash.py index a675793..d35cff2 100644 --- a/infra/tool/fleetdash.py +++ b/infra/tool/fleetdash.py @@ -143,7 +143,8 @@ def _get_raw(url, token=None, timeout=8): "tlc", "tlc_tla", "tlc_log", "traceparent", "sources", "restored", "failed", "progress", "needs_approve") _EVENT_KEYS = ("type", "step", "status", "exit", "reason", "span", "authority", "keyid", - "snippet_shas", "meter", "ts", "traceparent", "result_nonce", "exod_keyid", "plan_sha") + "snippet_shas", "meter", "ts", "traceparent", "result_nonce", "exod_keyid", "plan_sha", + "values_sha") # tier-2 commitment (a hash) — the value-free per-step link into the value ledger # the compact per-run row the accounting/risk feeds read — MUST include `cost`, or the fleet credit-spend # rollup (_spend_rollup / render_accounting) always reads 0 (the per-run detail carries it, the row dropped it). # Same trap for `needs_approve`/`approved`: the risk queue and the supersession pass read the FEED rows, so a @@ -306,7 +307,7 @@ def load_run_svg(mirror_dir, name, run_id): # the ONLY node sub-paths fleetdash will proxy live (token-injected) — read-only inspection endpoints. The # target host is always the configured node (no SSRF to arbitrary hosts); this bounds it to safe read paths. -_PROXY_PREFIX = ("trace/", "intoto/", "flow/run/", "ledger/", "monitor/run/") +_PROXY_PREFIX = ("trace/", "intoto/", "flow/run/", "ledger/", "monitor/run/", "monitor/values/") _PROXY_EXACT = ("catalog", "oversight") @@ -1128,6 +1129,33 @@ def _run_live(detail): return len(done) < len(detail.get("seq") or []) +def _values_reveal_script(): + """The reveal button's client script: fetch the node-proxied /monitor/values/ and render each + step's decrypted inputs. EVERY node-supplied string is written via textContent (never innerHTML) — the + inline-onclick / raw-interpolation XSS sink is deliberately avoided (see the fleetdash UX pass).""" + return ('') + + def render_run(name, run_id, detail, has_svg=False, refresh=None): """Per-run LEDGER INSPECTION — local-monitor parity from the durable mirror: the full value-free record (claim + approval, the step plan, the event chain, plan + closure pins, the model-check + provenance) + the @@ -1144,14 +1172,18 @@ def render_run(name, run_id, detail, has_svg=False, refresh=None): by_step = {e.get("step"): e for e in detail.get("events", []) if e.get("type") == "step_result"} granted = {e.get("step") for e in detail.get("events", []) if e.get("type") == "granted"} srows = [] + any_values = False for i, tool in enumerate(detail.get("seq", []), 1): e = by_step.get(str(i)) state = ("ok" if e and e.get("status") == "ok" else "error" if e else "granted" if str(i) in granted else "pending") cls = {"ok": "ok", "error": "no", "granted": "warn"}.get(state, "") + vsha = (e or {}).get("values_sha") + any_values = any_values or bool(vsha) srows.append(f'{i}{_esc(tool)}{_esc((e or {}).get("authority", "—"))}' - f'{_esc((e or {}).get("exit", "—"))}{_esc(state)}') - steps = "".join(srows) or 'no steps declared' + f'{_esc((e or {}).get("exit", "—"))}{_esc(state)}' + f'{_cp(vsha, 16) if vsha else "—"}') + steps = "".join(srows) or 'no steps declared' def ev_note(e): note = e.get("reason") or "" @@ -1219,7 +1251,15 @@ def prob_line(p): f'raw ledger ↗

' + callout + _card("steps", f'' - f'{steps}
#toolexec (authority)exitstate
') + f'statetool-use (values_sha){steps}') + + (_card("tool-use detail — decrypted step inputs", + f'

Each step\'s values_sha above commits the ' + 'exact declared, non-secret inputs into the value-free chain. The values themselves are ' + 'encrypted at rest (tier-2 ledger); reveal decrypts them live on the node with its ' + 'recipient key (secrets never recorded — they stay *_FILE pointers).

' + f'' + '
' + + _values_reveal_script()) if any_values else "") + flow + _card("claim & approval", f'

destructive {_esc(detail.get("destructive", False))} · ' diff --git a/tests/test_param_delegated.py b/tests/test_param_delegated.py index c9780f7..6375d1c 100644 --- a/tests/test_param_delegated.py +++ b/tests/test_param_delegated.py @@ -205,3 +205,28 @@ def test_exod_passes_caller_values_for_a_params_granted_actor(): acl = {"skills": ["*"], "max_tier": "community", "secrets": False, "params": True} env = {**_FIXED, "MODEL_HANDLE": "nvidia/Qwen3.6-35B-A3B-NVFP4"} assert _pd_exod()._acl_check(_pd_req(acl, env), _pd_gbody(acl), now=1500) is None + + +def test_value_ledger_records_and_decrypts_the_filtered_values(tmp_path): + """Tier-2 (docs/pg-provenance-ledger.md): the same declared, non-secret values that ride the WS are + ENCRYPTED at rest and their PLAINTEXT commitment (values_sha) is what the tier-1 chain step event carries. + The node decrypts its own ledger with its recipient key; the commitment matches the plaintext.""" + from infra.govern.govd import Store + from infra.store import valuecrypt + + st = Store(str(tmp_path), cfg={}) # value ledger default-on + vals = {"SOURCE": "/repos/curl", "LIMIT": "50"} # the post-filter (declared, non-secret) values + sha = st.record_values("runP", "1", "2026-07-22T00:00:00Z", vals) + assert sha == valuecrypt.values_sha(vals) # commitment is over the canonical PLAINTEXT + st.mirror.flush() + view = st.decrypt_values("runP") + assert len(view) == 1 and view[0]["values"] == vals and view[0]["values_sha"] == sha + + +def test_value_ledger_never_records_empty_or_when_disabled(tmp_path): + from infra.govern.govd import Store + st = Store(str(tmp_path / "on"), cfg={}) + assert st.record_values("r", "1", "t", {}) is None # empty filtered set -> nothing recorded, no sha + off = Store(str(tmp_path / "off"), cfg={"value_ledger": {"enabled": False}}) + assert off.record_values("r", "1", "t", {"A": "1"}) is None + assert off.decrypt_values("r") == [] diff --git a/tests/test_value_ledger.py b/tests/test_value_ledger.py new file mode 100644 index 0000000..a7feb3e --- /dev/null +++ b/tests/test_value_ledger.py @@ -0,0 +1,105 @@ +"""Tier-2 value ledger (docs/pg-provenance-ledger.md): envelope encryption + the backend run_values table + +the mirror op. Proves the load-bearing invariants: values are ciphertext-at-rest, the `values_sha` commitment +is over PLAINTEXT, a non-recipient cannot open a blob, tampering is caught, and a fleet re-wrap adds a +recipient without re-encrypting. Hermetic — no server, no Postgres.""" +import json +import os +import tempfile + +import pytest + +from infra.store import valuecrypt as vc +from infra.store.backend import PsycopgBackend, SqliteWalBackend +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + + +def _node_key(d): + kf = os.path.join(d, ".value-key") + return kf, vc.generate_node_key(kf) + + +def test_keygen_is_0600_and_idempotent(): + d = tempfile.mkdtemp() + kf, pub = _node_key(d) + assert oct(os.stat(kf).st_mode)[-3:] == "600" + assert vc.generate_node_key(kf) == pub # existing key loaded, never overwritten + + +def test_commitment_is_over_plaintext_not_ciphertext(): + d = tempfile.mkdtemp() + _, pub = _node_key(d) + vals = {"SOURCE": "/repos/curl", "LIMIT": "50"} + blob = vc.encrypt(vals, [pub]) + assert blob["sha"] == vc.values_sha(vals) # commitment == sha256(canonical plaintext) + # two encryptions of the same values differ (fresh nonce/DEK) yet share the plaintext commitment + blob2 = vc.encrypt(vals, [pub]) + assert blob["ct"] != blob2["ct"] and blob["sha"] == blob2["sha"] + + +def test_ciphertext_carries_no_plaintext(): + d = tempfile.mkdtemp() + _, pub = _node_key(d) + blob = vc.encrypt({"SOURCE": "/secret/path/xyzzy", "TOKEN_FILE": "/k"}, [pub]) + assert b"xyzzy" not in json.dumps(blob).encode() + + +def test_roundtrip_and_nonrecipient_refused(): + d = tempfile.mkdtemp() + kf, pub = _node_key(d) + vals = {"A": "1", "B": "two"} + blob = vc.encrypt(vals, [pub]) + assert vc.decrypt(blob, vc.load_private(kf)) == vals + stranger = X25519PrivateKey.generate() + with pytest.raises(ValueError): + vc.decrypt(blob, stranger) + + +def test_tamper_is_caught(): + d = tempfile.mkdtemp() + kf, pub = _node_key(d) + blob = vc.encrypt({"A": "1"}, [pub]) + bad = dict(blob) + bad["ct"] = "00" * (len(bytes.fromhex(blob["ct"]))) + with pytest.raises(Exception): + vc.decrypt(bad, vc.load_private(kf)) + + +def test_rewrap_adds_recipient_without_reencrypting(): + d = tempfile.mkdtemp() + kf, pub = _node_key(d) + vals = {"SOURCE": "/r"} + blob = vc.encrypt(vals, [pub]) + sk = vc.load_private(kf) + moth = X25519PrivateKey.generate() + mpub = moth.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + blob2 = vc.rewrap(blob, sk, [mpub]) + assert blob2["ct"] == blob["ct"] # data NOT re-encrypted, only a wrap added + assert vc.decrypt(blob2, sk) == vals # original recipient still opens it + assert vc.decrypt(blob2, moth) == vals # new recipient opens it too + + +def test_empty_recipient_set_rejected(): + with pytest.raises(ValueError): + vc.encrypt({"A": "1"}, []) + + +def test_backend_run_values_upsert_and_isolation(): + d = tempfile.mkdtemp() + be = SqliteWalBackend(os.path.join(d, "idx.sqlite")).open() + kf, pub = _node_key(d) + blob = vc.encrypt({"SOURCE": "/r", "LIMIT": "9"}, [pub]) + assert be.record_values("run1", "1", "t0", blob["sha"], blob)["status"] == "indexed" + assert be.record_values("run1", "1", "t0", blob["sha"], blob)["status"] == "duplicate" + rows = be.get_values("run1") + assert [r["values_sha"] for r in rows] == [blob["sha"]] + assert vc.decrypt(rows[0]["blob"], vc.load_private(kf)) == {"SOURCE": "/r", "LIMIT": "9"} + assert be.get_values("other") == [] + be.reset() + assert be.get_values("run1") == [] # reset drops the tier-2 table too + + +def test_pg_backend_inert_until_configured(): + pg = PsycopgBackend({}) + assert pg.record_values("r", "1", "t", "s", {})["status"] == "unconfigured" + assert pg.get_values("r") == [] From 24bba895eaab3ec1c3f31e82b1a6cc2a0cc614df Mon Sep 17 00:00:00 2001 From: Rui He Date: Wed, 22 Jul 2026 07:30:55 -0400 Subject: [PATCH 2/3] =?UTF-8?q?fix(store):=20adversarial-review=20fixes=20?= =?UTF-8?q?=E2=80=94=20salted=20commitment,=20record-after-run,=20PG-safe?= =?UTF-8?q?=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four-lens adversarial review (crypto / value-free-boundary / web-surface / enforcement-path) found four real defects; this fixes all of them. - [crypto Medium] values_sha was an UNSALTED hash of low-entropy plaintext, living in the value-free chain/index — the same DB as the ciphertext — so a backup/replica/over-granted role could brute-force the values without touching the envelope (contradicting the module's own at-rest claim). Now SALTED: sha256(salt ‖ plaintext) with a fresh salt sealed INSIDE the AEAD blob; an authorized decryptor recovers salt+plaintext and re-verifies. Full-width keyid too (no rewrap truncation-collision/evict). - [enforcement #2/#3] record_values ran BEFORE execute_step with a first-wins upsert, so a refused/retried step orphaned a value row and a retry with corrected values desynced the chain sha from the stored blob. Now records ONLY on a terminal step_result (ok/error), at-most-once, so the stored blob and the chain-bound sha always agree; a refusal records nothing. - [enforcement #1] decrypt_values read the drain worker's shared (unguarded psycopg) connection — a monitor read racing a drain write could drop a tier-1 chain-index row. Now reads through a FRESH backend connection (reconciler discipline). - [key siting] the node key defaulted into the replicated record root (backup+key = broken at-rest); govd now warns loudly and the doc requires value_ledger.node_key_file off the backup path. Residuals documented (accepted): name-based secret filter, plaintext view behind the shared monitor token, values queue pressure. Tests updated for the salted commitment + a retry-desync regression. infra/ ruff clean; store selftest green. Co-Authored-By: Claude Fable 5 --- docs/pg-provenance-ledger.md | 28 +++++++++++++++++----- infra/govern/govd.py | 44 +++++++++++++++++++++++++---------- infra/store/valuecrypt.py | 41 ++++++++++++++++++++------------ tests/test_param_delegated.py | 28 ++++++++++++++++++++-- tests/test_value_ledger.py | 35 +++++++++++++++++++++------- 5 files changed, 133 insertions(+), 43 deletions(-) diff --git a/docs/pg-provenance-ledger.md b/docs/pg-provenance-ledger.md index 671f7f6..5a353ab 100644 --- a/docs/pg-provenance-ledger.md +++ b/docs/pg-provenance-ledger.md @@ -78,12 +78,28 @@ read-time, backed by signature/chain verification, size caps, and the schema bou ## Crypto -Envelope encryption with a **recipient set**: fresh per-blob DEK (AEAD, e.g. XChaCha20-Poly1305 or AES-256-GCM -per what's already vendored); DEK wrapped to each recipient's X25519 public key. Standalone recipient set = -{node}; post-handshake = {node, mothership-oversight}. Fleet join re-wraps *historical DEKs only* (bytes per -blob — never re-encrypts data). Node recipient key: generated at boot if absent, `chmod 600`, beside govd's -existing key material on the config mount — never in the DB, never on a replicated path. Rotation = new -recipient key + lazy re-wrap; revocation = stop wrapping + rotate. +Envelope encryption with a **recipient set**: fresh per-blob DEK (AES-256-GCM, from the already-required +`cryptography`); DEK wrapped to each recipient's X25519 public key via ephemeral-static ECDH + HKDF-SHA256. +Standalone recipient set = {node}; post-handshake = {node, mothership-oversight}. Fleet join re-wraps +*historical DEKs only* (bytes per blob — never re-encrypts data). Node recipient key: generated at boot if +absent, `chmod 600` from creation. + +**The commitment is SALTED.** `values_sha = sha256(salt ‖ canonical-plaintext)` with a fresh per-blob salt +**sealed inside the AEAD blob** (never in the chain). An unsalted hash would be fatal here: the commitment +lands in the value-free chain/index — the *same database* as the ciphertext — and the tier-2 values are +deliberately **low-entropy** (`LIMIT=50`, `SOURCE=/repos/curl`, provider/model names). An unsalted hash there +is a preimage oracle: a backup / replica / over-granted DB role brute-forces the values without touching the +envelope, defeating encryption-at-rest for exactly the values the ledger records. Salting closes it — the +at-rest attacker holds the commitment but not the salt (it needs the recipient key); an authorized decryptor +recovers salt+plaintext and re-verifies. Trade-off: cross-run same-input detection by hash-equality is gone +(identical inputs get distinct salts) — compare decrypted plaintexts instead. Recipient `keyid` is the +**full** sha256 of the pubkey (not truncated) so two recipients can never collide + evict in `rewrap`. + +**KEY SITING (operator requirement):** the decryption key must NOT sit on the same replicated/backed-up path +as the ciphertext — a backup holding both defeats encryption-at-rest. Set `value_ledger.node_key_file` to a +non-replicated path and exclude it from `cws-backup`. The in-record-root default is a convenience fallback +only; govd warns loudly when it is used. Rotation = new recipient key + lazy re-wrap; revocation = stop +wrapping + rotate. **Honest limit (accepted residual):** govd sees values transiently (it is the writer); a fully compromised *live* govd host reads them regardless. Encrypt-at-rest defends at-rest surfaces — replicas, backups, stolen diff --git a/infra/govern/govd.py b/infra/govern/govd.py index 3c86f1e..81223cf 100644 --- a/infra/govern/govd.py +++ b/infra/govern/govd.py @@ -395,13 +395,21 @@ def _init_value_ledger(self, cfg): node with `value_ledger.enabled` false records exactly as before (no values persisted).""" self.value_key_file = None self.value_recipients = [] # raw X25519 pubkeys the blobs are sealed to (>=1 to record) + self._value_cfg = cfg # for a FRESH read-backend in decrypt_values (never the writer's cx) vc = cfg.get("value_ledger") or {} if not vc.get("enabled", True): # default-on locally; the tier stays value-free-at-rest regardless return try: from infra.store import valuecrypt - self.value_key_file = os.path.expanduser( - vc.get("node_key_file") or os.path.join(self.root, ".value-key")) + keyfile = vc.get("node_key_file") + self.value_key_file = os.path.expanduser(keyfile or os.path.join(self.root, ".value-key")) + if not keyfile: + # KEY SITING: the decryption key must NOT live on the same replicated/backed-up path as the + # ciphertext (a backup with both defeats encryption-at-rest). The in-root default is a + # convenience fallback ONLY — the operator should set value_ledger.node_key_file to a + # non-replicated path and exclude it from cws-backup. Warn loudly so it is not missed. + sys.stderr.write(f"[govd] value-ledger: node key defaulting into the record root " + f"({self.value_key_file}); set value_ledger.node_key_file off the backup path.\n") node_pub = valuecrypt.generate_node_key(self.value_key_file) recips = [node_pub] for hx in (vc.get("recipients") or []): # extra recipient pubkeys (hex) — e.g. mothership oversight @@ -433,11 +441,18 @@ def decrypt_values(self, run_id): """Operator-side plaintext view of one run's tier-2 values (the monitor 'detail of each tool use'). Decrypts each step's blob with the NODE's recipient key — available only where that key lives (this node). Returns [{step, ts, values_sha, values|error}] ordered by step; [] when the tier is inert.""" - if not self.value_key_file: + if not self.value_key_file or not self.mirror.enabled(): return [] + from infra.store import backend as _sb from infra.store import valuecrypt - be = getattr(self.mirror, "backend", None) - if be is None: + # Read through a FRESH backend connection, never self.mirror.backend: the drain worker writes the shared + # cx, and PsycopgBackend is NOT thread-guarded — a monitor read racing a drain write on that one psycopg + # connection desyncs the protocol and can drop a tier-1 chain-index write. The reconciler uses the same + # own-connection discipline. sqlite readers are cheap + WAL-concurrent, so a per-call backend is fine. + try: + be = _sb.make_backend(self.root, self._value_cfg or {}) + except Exception as e: + sys.stderr.write(f"[govd] value-ledger read backend unavailable: {e}\n") return [] sk = valuecrypt.load_private(self.value_key_file) out = [] @@ -1433,11 +1448,6 @@ def _ws_oversight(self): if k in declared and k not in RESERVED_ENV and not k.startswith("CWS_SECRET_") and not (SECRET_KEY.search(k) and not k.endswith(POINTER_SUFFIX))} - # tier-2 value ledger: record the (declared-subset, secret-filtered) values ENCRYPTED - # at rest — this is the ONE place plaintext values exist server-side — and bind their - # plaintext commitment `values_sha` into the tier-1 chain step event. The wire + the - # value-free chain/monitor stay value-free; only the sha (a hash) crosses into them. - values_sha = store.record_values(bound, step, now(), var_values) d_reply, d_event = delegate.execute_step( rec0, step, psha, exod_socket=sock, grant_key=gk, exod_pub=epub, base=getattr(self.server, "exec_workspace", os.path.join(store.root, "_work")), @@ -1447,8 +1457,18 @@ def _ws_oversight(self): if d_event: ev = {**d_event, "ts": now(), "span": tracing.child_span((rec0 or {}).get("traceparent"))} - if values_sha: - ev["values_sha"] = values_sha + # tier-2 value ledger: record the (declared-subset, secret-filtered) values + # ENCRYPTED at rest and bind their commitment `values_sha` into THIS chain event — + # but ONLY for a step that actually RAN (a terminal step_result, ok OR error). A + # refused step is retryable, so recording it would (a) orphan a value row for a + # step with no execution event, and (b) let a retry with corrected values desync + # the chain sha from the first-wins stored blob. Recording exactly when the step + # is terminal makes it at-most-once, so the stored blob and the chain sha always + # agree. The wire + value-free chain/monitor stay value-free; only the sha crosses. + if d_event.get("type") == "step_result": + values_sha = store.record_values(bound, step, ev["ts"], var_values) + if values_sha: + ev["values_sha"] = values_sha store.append(bound, ev) ws_send(self.wfile, json.dumps({"type": "executed", "step": msg.get("step"), **d_reply})) finally: diff --git a/infra/store/valuecrypt.py b/infra/store/valuecrypt.py index 9476009..2d65a85 100644 --- a/infra/store/valuecrypt.py +++ b/infra/store/valuecrypt.py @@ -34,20 +34,30 @@ _BLOB_AAD = b"cyberware/valuecrypt/blob/v1" # AES-GCM AAD binds the ciphertext to this scheme+version +_SALT_LEN = 16 # per-blob commitment salt (defeats low-entropy preimage search) + + def canon(values: dict) -> bytes: - """The canonical plaintext bytes hashed into `values_sha` AND sealed as the blob. Stable: sorted keys, - tight separators — so the same {KEY:value} map always yields the same sha and the same sealed bytes.""" + """The canonical plaintext bytes — SEALED as the blob and committed (salted) into the tier-1 chain. Stable: + sorted keys, tight separators — so the same {KEY:value} map always yields the same sealed bytes.""" return json.dumps(values, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") -def values_sha(values: dict) -> str: - """sha256 of the canonical PLAINTEXT — the commitment bound into the tier-1 chain (never the ciphertext).""" - return hashlib.sha256(canon(values)).hexdigest() +def commit(salt: bytes, pt: bytes) -> str: + """The tier-1 commitment = sha256(SALT || canonical-plaintext). SALTED on purpose: the commitment lands in + the value-free chain/index (the SAME db as the ciphertext), and the tier-2 values are deliberately + LOW-ENTROPY (LIMIT=50, SOURCE=/repos/curl, provider/model names). An UNsalted hash there is a preimage + oracle — a backup / replica / over-granted DB role brute-forces the values without ever touching the + envelope. The salt is sealed INSIDE the AEAD blob (never in the chain), so an at-rest attacker holding the + commitment cannot recover it; an authorized decryptor recovers salt+plaintext and re-verifies.""" + return hashlib.sha256(salt + pt).hexdigest() def keyid(pub_raw: bytes) -> str: - """A short stable id for a recipient public key — the map key in `dek_wraps` and the audit label.""" - return "x25519:" + hashlib.sha256(pub_raw).hexdigest()[:16] + """A stable id for a recipient public key — the map key in `dek_wraps` and the audit label. FULL sha256 + (not truncated): the map key is an identity, and a truncated id lets a second recipient collide with and + evict an existing one in `rewrap` (a DoS). Full-width removes the collision entirely.""" + return "x25519:" + hashlib.sha256(pub_raw).hexdigest() # ── key material ───────────────────────────────────────────────────────────────────────────────────────── @@ -110,17 +120,17 @@ def encrypt(values: dict, recipient_pubs: list[bytes]) -> dict: if not recipient_pubs: raise ValueError("valuecrypt.encrypt: recipient set is empty — a value blob needs >=1 recipient") pt = canon(values) + salt = os.urandom(_SALT_LEN) dek = AESGCM.generate_key(bit_length=256) blob_nonce = os.urandom(12) - ct = AESGCM(dek).encrypt(blob_nonce, pt, _BLOB_AAD) + ct = AESGCM(dek).encrypt(blob_nonce, salt + pt, _BLOB_AAD) # seal SALT||plaintext — the salt is at-rest-protected wraps = {} for pub in recipient_pubs: kek, eph_pub = _wrap_kek(pub) wrap_nonce = os.urandom(12) wrapped = AESGCM(kek).encrypt(wrap_nonce, dek, _WRAP_INFO) wraps[keyid(pub)] = {"eph": eph_pub.hex(), "nonce": wrap_nonce.hex(), "wrap": wrapped.hex()} - return {"v": 1, "sha": hashlib.sha256(pt).hexdigest(), "nonce": blob_nonce.hex(), - "ct": ct.hex(), "wraps": wraps} + return {"v": 1, "sha": commit(salt, pt), "nonce": blob_nonce.hex(), "ct": ct.hex(), "wraps": wraps} def rewrap(blob: dict, sk: X25519PrivateKey, new_recipient_pubs: list[bytes]) -> dict: @@ -150,10 +160,11 @@ def _recover_dek(blob: dict, sk: X25519PrivateKey) -> bytes: def decrypt(blob: dict, sk: X25519PrivateKey) -> dict: - """Open a blob with a recipient private key -> the original `values` dict. Also re-verifies the plaintext - commitment (`sha`) so a tampered ciphertext is caught here, not silently returned.""" + """Open a blob with a recipient private key -> the original `values` dict. Recovers the sealed SALT and + re-verifies the salted commitment (`sha`), so a tampered ciphertext is caught here, not silently returned.""" dek = _recover_dek(blob, sk) - pt = AESGCM(dek).decrypt(bytes.fromhex(blob["nonce"]), bytes.fromhex(blob["ct"]), _BLOB_AAD) - if hashlib.sha256(pt).hexdigest() != blob.get("sha"): - raise ValueError("valuecrypt.decrypt: plaintext commitment mismatch — blob tampered") + sealed = AESGCM(dek).decrypt(bytes.fromhex(blob["nonce"]), bytes.fromhex(blob["ct"]), _BLOB_AAD) + salt, pt = sealed[:_SALT_LEN], sealed[_SALT_LEN:] + if commit(salt, pt) != blob.get("sha"): + raise ValueError("valuecrypt.decrypt: salted commitment mismatch — blob tampered") return json.loads(pt.decode("utf-8")) diff --git a/tests/test_param_delegated.py b/tests/test_param_delegated.py index 6375d1c..442feda 100644 --- a/tests/test_param_delegated.py +++ b/tests/test_param_delegated.py @@ -212,12 +212,11 @@ def test_value_ledger_records_and_decrypts_the_filtered_values(tmp_path): ENCRYPTED at rest and their PLAINTEXT commitment (values_sha) is what the tier-1 chain step event carries. The node decrypts its own ledger with its recipient key; the commitment matches the plaintext.""" from infra.govern.govd import Store - from infra.store import valuecrypt st = Store(str(tmp_path), cfg={}) # value ledger default-on vals = {"SOURCE": "/repos/curl", "LIMIT": "50"} # the post-filter (declared, non-secret) values sha = st.record_values("runP", "1", "2026-07-22T00:00:00Z", vals) - assert sha == valuecrypt.values_sha(vals) # commitment is over the canonical PLAINTEXT + assert sha and len(sha) == 64 # a salted commitment over the plaintext st.mirror.flush() view = st.decrypt_values("runP") assert len(view) == 1 and view[0]["values"] == vals and view[0]["values_sha"] == sha @@ -230,3 +229,28 @@ def test_value_ledger_never_records_empty_or_when_disabled(tmp_path): off = Store(str(tmp_path / "off"), cfg={"value_ledger": {"enabled": False}}) assert off.record_values("r", "1", "t", {"A": "1"}) is None assert off.decrypt_values("r") == [] + + +def test_value_ledger_binds_only_on_terminal_step_not_on_refusal(tmp_path, monkeypatch): + """Regression (adversarial review #2/#3): record_values must fire ONLY when a step actually RAN (a terminal + step_result), so a retried/refused step never orphans a value row nor desyncs the chain sha from the stored + blob. We drive the Store directly to mirror the govd WS-handler contract: refusal -> no record; terminal -> + exactly one record whose commitment equals the chain-bound sha, and a retry with new values is a no-op + (at-most-once), so blob and chain never disagree.""" + from infra.govern.govd import Store + + st = Store(str(tmp_path), cfg={}) + # a refused step (govd stamps no values_sha, calls no record_values) leaves the ledger empty + st.mirror.flush() + assert st.decrypt_values("runR") == [] + # the terminal run records exactly once; the returned sha is what the chain event would carry + sha = st.record_values("runR", "1", "t1", {"SOURCE": "/a", "LIMIT": "1"}) + st.mirror.flush() + view = st.decrypt_values("runR") + assert len(view) == 1 and view[0]["values_sha"] == sha and view[0]["values"] == {"SOURCE": "/a", "LIMIT": "1"} + # at-most-once upsert: a (hypothetical) second record for the same step never overwrites the first blob, + # so the stored blob stays consistent with the sha the chain committed on the real terminal run + st.record_values("runR", "1", "t2", {"SOURCE": "/DIFFERENT", "LIMIT": "999"}) + st.mirror.flush() + view2 = st.decrypt_values("runR") + assert len(view2) == 1 and view2[0]["values_sha"] == sha and view2[0]["values"] == {"SOURCE": "/a", "LIMIT": "1"} diff --git a/tests/test_value_ledger.py b/tests/test_value_ledger.py index a7feb3e..dd4d878 100644 --- a/tests/test_value_ledger.py +++ b/tests/test_value_ledger.py @@ -1,7 +1,8 @@ """Tier-2 value ledger (docs/pg-provenance-ledger.md): envelope encryption + the backend run_values table + the mirror op. Proves the load-bearing invariants: values are ciphertext-at-rest, the `values_sha` commitment -is over PLAINTEXT, a non-recipient cannot open a blob, tampering is caught, and a fleet re-wrap adds a -recipient without re-encrypting. Hermetic — no server, no Postgres.""" +is a SALTED hash over the plaintext (verifiable by a decryptor, not a preimage oracle for an at-rest attacker), +a non-recipient cannot open a blob, tampering is caught, and a fleet re-wrap adds a recipient without +re-encrypting. Hermetic — no server, no Postgres.""" import json import os import tempfile @@ -26,15 +27,19 @@ def test_keygen_is_0600_and_idempotent(): assert vc.generate_node_key(kf) == pub # existing key loaded, never overwritten -def test_commitment_is_over_plaintext_not_ciphertext(): +def test_commitment_is_salted_and_verifiable_but_not_a_preimage_oracle(): d = tempfile.mkdtemp() - _, pub = _node_key(d) + kf, pub = _node_key(d) vals = {"SOURCE": "/repos/curl", "LIMIT": "50"} blob = vc.encrypt(vals, [pub]) - assert blob["sha"] == vc.values_sha(vals) # commitment == sha256(canonical plaintext) - # two encryptions of the same values differ (fresh nonce/DEK) yet share the plaintext commitment + # the commitment is SALTED (defeats brute-force from the value-free chain): an at-rest attacker cannot + # recompute it from the plaintext alone — the salt is sealed inside the AEAD blob, not in the commitment. + assert blob["sha"] != vc.commit(b"", vc.canon(vals)) + # two encryptions of the same values now differ in BOTH ciphertext AND commitment (fresh salt each time) blob2 = vc.encrypt(vals, [pub]) - assert blob["ct"] != blob2["ct"] and blob["sha"] == blob2["sha"] + assert blob["ct"] != blob2["ct"] and blob["sha"] != blob2["sha"] + # an authorized decryptor recovers salt+plaintext and the commitment re-verifies (done inside decrypt) + assert vc.decrypt(blob, vc.load_private(kf)) == vals def test_ciphertext_carries_no_plaintext(): @@ -74,9 +79,17 @@ def test_rewrap_adds_recipient_without_reencrypting(): moth = X25519PrivateKey.generate() mpub = moth.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) blob2 = vc.rewrap(blob, sk, [mpub]) - assert blob2["ct"] == blob["ct"] # data NOT re-encrypted, only a wrap added + assert blob2["ct"] == blob["ct"] and blob2["sha"] == blob["sha"] # data + commitment unchanged, wrap added assert vc.decrypt(blob2, sk) == vals # original recipient still opens it assert vc.decrypt(blob2, moth) == vals # new recipient opens it too + assert len(blob2["wraps"]) == 2 + + +def test_keyid_is_full_width_no_truncation_collision(): + # a full-width keyid (not [:16]) so two distinct recipients can never collide + evict in rewrap + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey as _K + p1 = _K.generate().public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) + assert len(vc.keyid(p1)) == len("x25519:") + 64 def test_empty_recipient_set_rejected(): @@ -103,3 +116,9 @@ def test_pg_backend_inert_until_configured(): pg = PsycopgBackend({}) assert pg.record_values("r", "1", "t", "s", {})["status"] == "unconfigured" assert pg.get_values("r") == [] + + +def test_commit_helper_is_salt_sensitive(): + pt = vc.canon({"A": "1"}) + assert vc.commit(b"\x00" * 16, pt) != vc.commit(b"\x01" * 16, pt) # salt changes the commitment + assert vc.commit(b"s" * 16, pt) == vc.commit(b"s" * 16, pt) # deterministic given salt From 0337b594488fd3ce7eba80346e920a074bb23fbd Mon Sep 17 00:00:00 2001 From: Rui He Date: Wed, 22 Jul 2026 08:05:13 -0400 Subject: [PATCH 3/3] =?UTF-8?q?test(store):=20end-to-end=20HTTP=20proof=20?= =?UTF-8?q?of=20GET=20/monitor/values=20=E2=80=94=20the=20fleet-monitor=20?= =?UTF-8?q?plaintext=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fleet monitor reviews plaintext via the LIVE PROXY: fleetdash → /proxy// monitor/values/ → the node decrypts with its own recipient key → plaintext back. This adds the first over-HTTP test of that node endpoint (unit tests covered the Store; this covers the wire): (1) 403 without the monitor token (fail-closed gate), (2) 200 with the decrypted per-step values when authed, (3) the value-free /monitor/run for the SAME run carries only values_sha — no plaintext leaks into the value-free view. Confirms decryption stays node-side (no central key, no central ciphertext) — the live-proxy oversight model. Co-Authored-By: Claude Fable 5 --- tests/test_monitor_values_http.py | 71 +++++++++++++++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 tests/test_monitor_values_http.py diff --git a/tests/test_monitor_values_http.py b/tests/test_monitor_values_http.py new file mode 100644 index 0000000..f0bbb49 --- /dev/null +++ b/tests/test_monitor_values_http.py @@ -0,0 +1,71 @@ +"""End-to-end HTTP test of GET /monitor/values/ — the exact node endpoint the fleet monitor proxies to +for plaintext tool-use review (live-proxy model, decryption node-side). Proves: (1) the monitor-token gate +(403 without), (2) a recorded run's values decrypt and return over HTTP, (3) the value-free /monitor/run for +the same run carries only values_sha, never the values.""" +import json +import threading +import urllib.error +import urllib.request + +import pytest + +from infra.govern import govd + + +@pytest.fixture +def server(tmp_path): + cfg = govd.load_config() + cfg["mode"] = "local" + cfg["local"] = {"host": "127.0.0.1", "ports": [0]} + cfg["record_root"] = str(tmp_path / "rr") + govd.ensure_monitor_token(cfg) + httpd, _ = govd.bind_server("127.0.0.1", [0]) + httpd.daemon_threads = True + httpd.cfg, httpd.store, httpd.rate_buckets = cfg, govd.Store(cfg["record_root"], cfg=cfg), {} + threading.Thread(target=httpd.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start() + base = f"http://127.0.0.1:{httpd.server_address[1]}" + for _ in range(100): + try: + urllib.request.urlopen(base + "/health", timeout=1); break + except OSError: + import time; time.sleep(0.02) + yield base, httpd, cfg["monitor_token"] + httpd.shutdown(); httpd.server_close() + + +def _get(url, token=None): + req = urllib.request.Request(url, headers={"X-Govd-Monitor": token} if token else {}) + with urllib.request.urlopen(req, timeout=3) as r: + return r.status, json.load(r) + + +def test_monitor_values_gate_and_roundtrip(server): + base, httpd, tok = server + # a real run: the allow record exists (create), then a terminal step records values (the WS-handler order) + httpd.store.create("runE2E", {"run_id": "runE2E", "skill": "x:y", "perk": "run", "decision": "allow", + "seq": ["t1"], "events": [], "var_keys": ["SOURCE", "LIMIT", "INTEL_PROVIDER"], + "plan_sha": "p" * 64, "ts": "2026-07-22T00:00:00Z"}) + vals = {"SOURCE": "/repos/curl", "LIMIT": "50", "INTEL_PROVIDER": "nvidia"} + sha = httpd.store.record_values("runE2E", "1", "2026-07-22T00:00:00Z", vals) + httpd.store.append("runE2E", {"type": "step_result", "step": "1", "status": "ok", "exit": 0, + "authority": "exod", "values_sha": sha}) + httpd.store.mirror.flush() + assert sha and len(sha) == 64 + + # (1) unauthenticated -> 403, no plaintext + try: + _get(base + "/monitor/values/runE2E") + assert False, "expected 403 without monitor token" + except urllib.error.HTTPError as e: + assert e.code == 403 + + # (2) authenticated -> decrypted values over HTTP + st, body = _get(base + "/monitor/values/runE2E", token=tok) + assert st == 200 + steps = body["steps"] + assert len(steps) == 1 and steps[0]["values"] == vals and steps[0]["values_sha"] == sha + + # (3) the value-free /monitor/run for the same run must NOT carry the plaintext, only the commitment + _, detail = _get(base + "/monitor/run/runE2E", token=tok) + blob = json.dumps(detail) + assert "/repos/curl" not in blob and "nvidia" not in blob # no plaintext value leaked into the value-free view