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 bdeaa42..81223cf 100644
--- a/infra/govern/govd.py
+++ b/infra/govern/govd.py
@@ -386,6 +386,84 @@ 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)
+ 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
+ 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
+ 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 or not self.mirror.enabled():
+ return []
+ from infra.store import backend as _sb
+ from infra.store import valuecrypt
+ # 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 = []
+ 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 +1021,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.
@@ -1366,8 +1455,21 @@ 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"))}
+ # 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:
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..2d65a85
--- /dev/null
+++ b/infra/store/valuecrypt.py
@@ -0,0 +1,170 @@
+#!/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
+
+
+_SALT_LEN = 16 # per-blob commitment salt (defeats low-entropy preimage search)
+
+
+def canon(values: dict) -> 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 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 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 ─────────────────────────────────────────────────────────────────────────────────────────
+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)
+ salt = os.urandom(_SALT_LEN)
+ dek = AESGCM.generate_key(bit_length=256)
+ blob_nonce = os.urandom(12)
+ 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": commit(salt, pt), "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. 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)
+ 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/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'
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_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
diff --git a/tests/test_param_delegated.py b/tests/test_param_delegated.py
index c9780f7..442feda 100644
--- a/tests/test_param_delegated.py
+++ b/tests/test_param_delegated.py
@@ -205,3 +205,52 @@ 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
+
+ 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 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
+
+
+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") == []
+
+
+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
new file mode 100644
index 0000000..dd4d878
--- /dev/null
+++ b/tests/test_value_ledger.py
@@ -0,0 +1,124 @@
+"""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 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
+
+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_salted_and_verifiable_but_not_a_preimage_oracle():
+ d = tempfile.mkdtemp()
+ kf, pub = _node_key(d)
+ vals = {"SOURCE": "/repos/curl", "LIMIT": "50"}
+ blob = vc.encrypt(vals, [pub])
+ # 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"]
+ # 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():
+ 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"] 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():
+ 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") == []
+
+
+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