From a2359496cb1668521e21cd17e2735d5eabf14307 Mon Sep 17 00:00:00 2001 From: Thomas Connally Date: Sun, 16 Aug 2026 14:19:00 +0000 Subject: [PATCH] =?UTF-8?q?feat(authz):=20CVA=20contract=20=E2=80=94=20pol?= =?UTF-8?q?icy-satisfaction=20binding=20+=20replay=20resistance=20(#252)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 9 + docs/authorized-action-receipts.md | 9 + docs/cva-contract-spec.md | 154 ++++++++++++ ledger_agent/cva.py | 367 +++++++++++++++++++++++++++++ ledger_agent/keys.py | 14 +- ledger_agent/prebind.py | 38 ++- ledger_agent/receipts.py | 24 +- tests/test_cva.py | 287 ++++++++++++++++++++++ tests/test_prebind_replay_api.py | 111 ++++++++- 9 files changed, 1007 insertions(+), 6 deletions(-) create mode 100644 docs/cva-contract-spec.md create mode 100644 ledger_agent/cva.py create mode 100644 tests/test_cva.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7485558..2d0ebda 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,15 @@ All notable changes to Ledger are documented here. submission contract, and a `ComposedGate` realizing the paper's compositional gating proposition (preventive monitors + evidential gates over disjoint requirement sets). See `docs/runtime-contract.md`. +- **CVA authorization property contract** (#252). Cryptographically + verifiable authorization (arXiv:2607.21325) over the AAR prebind: + statements bind agent principal + request hash + policy + context + + nonce/epoch; `cva_relation_holds` enforces + R_CVA = BindPrincipal ∧ BindRequest ∧ BindContext ∧ SatisfyPolicy; + replay resistance via a trusted consumed-nonce gateway with an inclusive + timestamp window; prebind v2 receipts now carry `request_hash`/`nonce`/ + `epoch` in the hash-covered payload (backward compatible). See + `docs/cva-contract-spec.md`. - **Evidence levels for receipts** (#235). Receipts now state what they prove: a four-level ladder (`structural` → `attested` → `replay` → `inclusion`) under `verification.evidence`, with stable per-level reason codes. A signed diff --git a/docs/authorized-action-receipts.md b/docs/authorized-action-receipts.md index d013d3c..c506ea9 100644 --- a/docs/authorized-action-receipts.md +++ b/docs/authorized-action-receipts.md @@ -102,3 +102,12 @@ belongs in TTL state/leases. Durable action records declare `searchability`: detection for an action status. - Vault later proves manifest validation, approval lifecycle, scope binding, searchability, and lease race safety before Hermes enforcement is enabled. + +## CVA property contract + +The AAR prebind is Ledger's authorization-request boundary: its optional +`request_hash`, `context_hash`, `policy_hash`, `nonce`, and `epoch` fields make +the request, policy, context, and replay window explicit and hash-covered. +See the [CVA contract specification](cva-contract-spec.md) for the formal +`BindPrincipal ∧ BindRequest ∧ BindContext ∧ SatisfyPolicy` mapping and its +runtime-execution/TOCTOU limitation. diff --git a/docs/cva-contract-spec.md b/docs/cva-contract-spec.md new file mode 100644 index 0000000..0c7b12d --- /dev/null +++ b/docs/cva-contract-spec.md @@ -0,0 +1,154 @@ +# CVA Contract Specification + +Status: implementation slice +Date: 2026-08-16 +Resolves: ledger#252 · Consumed by: `ledger_agent/cva.py`, AAR prebind receipts +Related: [Authorized Action Receipts](authorized-action-receipts.md), arXiv [2607.21325](https://arxiv.org/abs/2607.21325), [2605.20704](https://arxiv.org/abs/2605.20704), [2604.07695](https://arxiv.org/abs/2604.07695) + +## Scope and non-ZK interpretation + +The CVA paper defines authorization as request-bound evidence, not merely +authentication or delegation. Ledger implements the formal shape from +[2607.21325](https://arxiv.org/abs/2607.21325) as a deterministic, hash-bound +contract. It does **not** claim that a Python predicate is a SNARK, that private +attributes are hidden, or that authorization proves runtime execution. + +The public statement is `x = (id_i, h_q, h_c, pid_j, n, t)` (paper Eq. 16): + +```json +{ + "schema": "perseus-ledger-cva-statement/v1", + "agent_id": "agent-a", + "request_hash": "", + "context_hash": "", + "policy_id": "policy/v1", + "nonce": "n1", + "timestamp_ms": 100, + "statement_hash": "" +} +``` + +The Ledger witness is supplied to the verifier as `{principal_key_id, +key_registry, request_payload, context_payload, attrs, policy}`. The relation +is the paper's Eq. 22–24, evaluated as: + +```text +R_CVA = BindPrincipal ∧ BindRequest ∧ BindContext ∧ SatisfyPolicy +``` + +* **BindPrincipal (Eq. 23):** the normalized key registry contains the selected + key, its `agent_id`/`agent_binding` equals `statement.agent_id`, and the key + is not revoked. In an AAR, this is `actor_ref` plus the key-registry custody + label; custody discloses provenance and is not a proof of authority. +* **BindRequest (Eq. 19, property Eq. 30):** the SHA-256 of canonical + `request_payload` equals `request_hash`. AAR carries the same `request_hash`. +* **BindContext (Eq. 20, property Eqs. 32–36):** the SHA-256 of canonical + `context_payload` equals `context_hash`; AAR maps this to + `selected_context_digest` and the optional `context_hash`. +* **SatisfyPolicy (Eq. 21):** the caller-supplied deterministic predicate + returns the literal `True` for `(attrs, request_payload, context_payload)`. + `policy_id` is committed in the statement; a predicate may advertise a + matching `policy_id` for cross-policy rejection. `policy_version` and + `policy_hash` are the AAR policy projection. + +`build_cva_statement` and `build_prebind_v2` use canonical JSON +(`sort_keys=True`, `separators=(',', ':')`) and SHA-256. Relation failures are +reported by conjunct (`bind_principal`, `bind_request`, `bind_context`, +`satisfy_policy`) without short-circuiting. + +## Replay contract + +The gateway keeps mutable `consumed_nonces` outside the stateless relation. +`is_fresh(n, t, N)` is `n ∉ N ∧ t_min ≤ t ≤ t_max` (paper Eqs. 26, +37–40). `CvaGateway.accept` checks replay, timestamp, and the relation, then +adds the nonce only after full acceptance. A failed relation therefore cannot +burn a nonce. The gateway nonce set is a trusted component, as in the paper's +partially trusted gateway model; it must be durable/serialized correctly when +multiple gateways share an authorization domain. [2605.20704](https://arxiv.org/abs/2605.20704) +provides a related freshness/revocation perspective for agent credentials. + +## CVA property matrix + +| Property / paper definition | Attack class defeated | Ledger mechanism | Acceptance criterion | Covering test | +|---|---|---|---|---| +| Authorization soundness: no accepted proof without a valid witness, Eqs. 27–28 | Forged or invalid-witness authorization | Hash-bound statement plus all four fail-closed conjuncts | No relation acceptance when any binding/policy check fails | `test_authorization_soundness_rejects_tampered_request_payload` | +| Principal binding: a proof for `id_i` does not verify for `id_k`, Eq. 29 | Cross-principal transfer | Active registry key's agent binding equals `agent_id`; revocation is rejected | Agent B's key cannot satisfy Agent A's statement | `test_principal_binding_rejects_key_bound_to_another_agent` | +| Request binding: `q_i != q'_i` cannot transfer evidence, Eq. 30 | Cross-request transfer | Canonical request hash in CVA statement and AAR `request_hash` | Changed request yields `bind_request` | `test_request_binding_rejects_different_request_with_same_context` | +| Policy binding: `pid_j != pid_k` cannot transfer evidence, Eq. 31 | Cross-policy transfer | `policy_id` is statement-covered; policy predicate/identifier must match | Flipped predicate or advertised policy ID is rejected | `test_policy_binding_rejects_flipped_predicate_and_policy_identifier` | +| Context binding: distinct context commitments do not verify, Eqs. 32–36 | Context substitution at authorization | Canonical context hash plus AAR context projections | Changed context yields `bind_context` | `test_context_binding_rejects_changed_context_payload` | +| Replay resistance: consumed nonce or out-of-window time rejects, Eqs. 37–40 | Proof/nonce reuse and deferred presentation | Gateway nonce set and inclusive timestamp window | First accept succeeds; replay, stale, and future presentations reject | `test_replay_resistance_consumes_nonce_once`; `test_timestamp_window_rejects_stale_and_future_statements` | + +The authority-trace section additionally exercises old-key revocation, stale +context, and a fresh post-rotation witness. This complements continuous +delegation/revocation work such as [2604.07695](https://arxiv.org/abs/2604.07695) +without treating that protocol as Ledger's implementation. + +## Structural separation and explicit limits + +The paper's central open problem is that the following are distinct security +layers (Eq. 52): + +```text +Identity Binding ≢ Authorization-Request Binding ≢ Runtime-Execution Binding +``` + +1. **Identity binding** establishes which principal a key/credential names. + Ledger's registry binding and AAR `actor_ref` cover this narrow seam. +2. **Authorization-request binding** establishes what request and policy were + accepted under what context. The AAR prebind is this authorization-request + boundary: it is proposed/approved evidence before execution, not execution. +3. **Runtime-execution binding** establishes that the exact authorized request + was the request actually executed. Context drift between verification and + execution is the TOCTOU gap (`c_tv != c_te`; paper Eqs. 41–43 and 53–54). + Execution receipts and trajectory evidence must close that seam separately. + +A request commitment is not an agent's deliberative state: `Request +Commitment ≠ Internal Agent Intent` (paper Eq. 25). A valid hash proves only +that the committed bytes were presented; it says nothing about hidden chain of +thought, motivation, or normative policy correctness. The receipt path also +does not provide selective disclosure or post-quantum security. + +## Falsifiable research agenda + +| Hypothesis | Falsification condition | Experiment sketch | Status | +|---|---|---|---| +| **H1 — three-binding separation is implementable.** Distinct Ledger layers can compose prebind → execution receipt → trajectory evidence without conflating claims. | A controlled TOCTOU or substituted-runtime-request case passes despite distinct commitments, or one layer cannot be independently verified. | Generate paired authorized/executed requests, mutate context/action between layers, verify each digest and chain, and measure detection by layer. | Implemented as a contract boundary; empirical end-to-end experiment pending. | +| **H2 — replay resistance survives key rotation without trusted replay sets in the receipt path.** | A rotated/revoked key or previously accepted nonce is accepted, or a failed relation consumes a nonce. | Run concurrent old/new-key windows with duplicate, stale, future, and failed-relation deliveries. Keep the gateway nonce set as the explicit trusted state; receipts carry commitments but do not replace it. | Local sequential tests pass; distributed-state and concurrency falsification pending. | +| **H3 — receipts-first fits interactive latency better than per-request ZK proofs.** | Under a matched workload and threat model, receipts plus deterministic verification miss the latency budget or lose a required confidentiality guarantee relative to ZK. | Benchmark canonical hashing, registry/policy checks, and chain verification against a Groth16 lane at equal request frequency and policy coverage. The paper's framing reports roughly `zkLLM ~180 s/query` versus receipts `<20 ms`; reproduce rather than generalize those numbers. | Position/hypothesis, not a Ledger benchmark. | + +The stance is **receipts-first, proofs-where-required**: use the cheap, +inspectable AAR/chain path for ordinary authorization evidence, then add a +proof lane where confidential attributes or disclosure minimization is a real +requirement. The comparison must preserve the paper's warning that proof +latency depends on circuit complexity and authorization frequency +([2607.21325](https://arxiv.org/abs/2607.21325)). + +## Selective disclosure / zk-PoC feasibility lane + +When attributes must remain confidential, a prover can hold `(sk, attrs, +request_payload, context_payload)` and produce a Groth16 proof over a circuit +that exposes `agent_id`, request/context commitments, `policy_id`, nonce, and +time as public inputs. The gateway verifies `(statement, proof)` and still +performs stateful freshness checks; the verifier does not receive raw `attrs`. +Ledger can retain the statement hash, proof reference/digest, verification-key +identifier, and AAR linkage without storing witness material. + +This lane has material caveats. Groth16 requires a circuit-specific trusted +setup/CRS and is not post-quantum secure; policy changes may require circuit +and key governance, and dynamic procedural policies are difficult to encode. +The gateway remains partially trusted for replay state. Context binding and +runtime-execution binding must still be tested outside the circuit. Those +constraints are consistent with the paper's PoC limits and with adjacent +credential/revocation and delegation work ([2605.20704](https://arxiv.org/abs/2605.20704), +[2604.07695](https://arxiv.org/abs/2604.07695)). + +## Implementation slice + +- `ledger_agent/cva.py` implements the statement, relation, freshness gateway, + and `PROPERTIES` matrix for arXiv:2607.21325. +- `ledger_agent/receipts.py` adds optional `request_hash`, `nonce`, and `epoch` + fields to v2 prebind blocks; `prebind_hash` covers them. +- `ledger_agent/prebind.py` validates those fields while accepting old v1/v2 + blocks that omit them. +- `tests/test_cva.py` and the authority-trace v2 section exercise each attack + class, round trips, tamper rejection, and key rotation. diff --git a/ledger_agent/cva.py b/ledger_agent/cva.py new file mode 100644 index 0000000..bf8e3be --- /dev/null +++ b/ledger_agent/cva.py @@ -0,0 +1,367 @@ +"""A lightweight CVA authorization relation and replay gateway. + +This module implements the non-zero-knowledge Ledger adaptation of the formal +model in arXiv:2607.21325. A statement is public, hash-bound data; request and +context payloads are supplied to the verifier so the four CVA conjuncts can be +checked without pretending that a Python predicate is a ZK proof. +""" +from __future__ import annotations + +import hashlib +import json +import math +from collections.abc import Callable, Mapping +from typing import Any, Optional + +from .keys import normalize_key_registry + +CVA_STATEMENT_SCHEMA = "perseus-ledger-cva-statement/v1" + + +def _canonical(value: Any) -> bytes: + """Serialize a JSON value using Ledger's canonical hash encoding.""" + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + + +def _sha(value: Any) -> str: + return hashlib.sha256(_canonical(value)).hexdigest() + + +def _is_sha256(value: Any) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(char in "0123456789abcdef" for char in value.lower()) + ) + + +def _require_text(value: Any, field: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{field} must be a non-empty string") + return value + + +def _timestamp(value: Any) -> int | float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("timestamp_ms must be a finite number") + if not math.isfinite(value): + raise ValueError("timestamp_ms must be a finite number") + return value + + +def build_cva_statement(*, agent_id: str, request_hash: str, + context_hash: str, policy_id: str, nonce: str, + timestamp_ms: int | float) -> dict[str, Any]: + """Build the public CVA statement ``x`` from paper Equation (16). + + The returned ``statement_hash`` commits to every field except itself. The + hash is deliberately over the same canonical JSON representation used by + the rest of Ledger; it is not a proof of a private witness. + """ + body: dict[str, Any] = { + "schema": CVA_STATEMENT_SCHEMA, + "agent_id": _require_text(agent_id, "agent_id"), + "request_hash": request_hash.lower() if _is_sha256(request_hash) + else (_raise_hash("request_hash")), + "context_hash": context_hash.lower() if _is_sha256(context_hash) + else (_raise_hash("context_hash")), + "policy_id": _require_text(policy_id, "policy_id"), + "nonce": _require_text(nonce, "nonce"), + "timestamp_ms": _timestamp(timestamp_ms), + } + body["statement_hash"] = _sha(body) + return body + + +def _raise_hash(field: str) -> str: + raise ValueError(f"{field} must be a 64-character SHA-256 hex digest") + + +def _policy_parts(policy: Any) -> tuple[Optional[str], Optional[Callable[..., Any]]]: + """Return an optional policy identifier and its deterministic predicate. + + The issue-level API is a callable. A callable may advertise its bound + policy identifier through ``policy_id``; this gives the non-ZK adapter a + way to detect a cross-policy presentation. A mapping with ``predicate`` + and ``policy_id`` is accepted as a convenience for callers that carry + policy metadata separately. A plain callable remains fully supported. + """ + if isinstance(policy, Mapping): + policy_id = policy.get("policy_id") + predicate = policy.get("predicate", policy.get("evaluate")) + return (policy_id if isinstance(policy_id, str) else None, + predicate if callable(predicate) else None) + policy_id = getattr(policy, "policy_id", None) + return (policy_id if isinstance(policy_id, str) else None, + policy if callable(policy) else None) + + +def _statement_hash_valid(statement: Mapping[str, Any]) -> bool: + supplied = statement.get("statement_hash") + if not _is_sha256(supplied): + return False + body = {key: value for key, value in statement.items() + if key != "statement_hash"} + try: + return supplied.lower() == _sha(body) + except (TypeError, ValueError): + return False + + +def _safe_payload_hash(payload: Any) -> Optional[str]: + try: + return _sha(payload) + except (TypeError, ValueError, OverflowError): + return None + + +def cva_relation_holds(statement: Mapping[str, Any], *, + principal_key_id: str, + key_registry: Optional[Mapping[str, Any]], + request_payload: Any, + context_payload: Any, + attrs: Any, + policy: Any) -> tuple[bool, list[str]]: + """Evaluate the four CVA conjuncts from Equations (22)--(24). + + This is intentionally an honest Ledger adaptation: the registry and + payloads are verifier inputs and ``policy`` is a caller-supplied + deterministic predicate, not a zero-knowledge proof verifier. Every + violated conjunct is reported in formal order rather than short-circuiting. + """ + errors: list[str] = [] + if not isinstance(statement, Mapping): + return False, ["statement"] + + if statement.get("schema") != CVA_STATEMENT_SCHEMA: + errors.append("statement_schema") + if not _statement_hash_valid(statement): + errors.append("statement_hash") + + # BindPrincipal (Equation 23). ``normalize_key_registry`` retains the + # Ledger custody shape and the optional agent/revocation metadata. + principal_ok = False + try: + entry = normalize_key_registry(key_registry).get(principal_key_id) + if entry is not None: + binding = entry.get("agent_id", entry.get("agent_binding")) + revoked = bool(entry.get("revoked", False)) + if entry.get("revoked_at") is not None: + revoked = True + if entry.get("status") in {"revoked", "disabled", "inactive"}: + revoked = True + principal_ok = ( + isinstance(statement.get("agent_id"), str) + and binding == statement.get("agent_id") + and not revoked + ) + except (TypeError, ValueError): + principal_ok = False + if not principal_ok: + errors.append("bind_principal") + + # BindRequest (Equation 19) and BindContext (Equation 20). + request_digest = _safe_payload_hash(request_payload) + if not (_is_sha256(statement.get("request_hash")) + and request_digest == statement.get("request_hash", "").lower()): + errors.append("bind_request") + + context_digest = _safe_payload_hash(context_payload) + if not (_is_sha256(statement.get("context_hash")) + and context_digest == statement.get("context_hash", "").lower()): + errors.append("bind_context") + + # SatisfyPolicy (Equation 21). A bound policy identifier is checked when + # the callable exposes one; a plain predicate is still valid and must be + # selected by the caller according to statement["policy_id"]. + policy_id, predicate = _policy_parts(policy) + policy_ok = predicate is not None + if policy_id is not None and policy_id != statement.get("policy_id"): + policy_ok = False + if predicate is not None: + try: + policy_ok = policy_ok and (predicate(attrs, request_payload, context_payload) is True) + except Exception: + # A policy failure is an authorization failure, not an exception + # escape that could accidentally admit the request. + policy_ok = False + if not policy_ok: + errors.append("satisfy_policy") + + return not errors, errors + + +def is_fresh(nonce: str, timestamp_ms: int | float, + consumed_nonces: set[str], t_min: int | float, + t_max: int | float) -> bool: + """Return the inclusive freshness predicate from Equations (26), (38)--(40).""" + try: + return nonce not in consumed_nonces and t_min <= timestamp_ms <= t_max + except (TypeError, ValueError): + return False + + +class CvaGateway: + """Stateful freshness gateway around the stateless CVA relation. + + ``consumed_nonces`` is the trusted replay-control state. A nonce is added + only after the relation and timestamp checks have both accepted, matching + the state transition in Equation (37). + """ + + def __init__(self, consumed_nonces: Optional[set[str]] = None, *, + t_min: int | float | None = None, + t_max: int | float | None = None) -> None: + self.consumed_nonces = consumed_nonces if consumed_nonces is not None else set() + self.t_min = t_min + self.t_max = t_max + + def consume(self, statement: Mapping[str, Any], *, + consumed: Optional[set[str]] = None) -> None: + """Commit an accepted statement's nonce to replay state.""" + nonce = statement.get("nonce") + if not isinstance(nonce, str) or not nonce: + raise ValueError("statement nonce must be a non-empty string") + target = self.consumed_nonces if consumed is None else consumed + target.add(nonce) + + def accept(self, statement: Mapping[str, Any], + witness: Optional[Mapping[str, Any]] = None, *, + principal_key_id: Optional[str] = None, + key_registry: Optional[Mapping[str, Any]] = None, + request_payload: Any = None, + context_payload: Any = None, + attrs: Any = None, + policy: Any = None, + consumed: Optional[set[str]] = None, + t_min: int | float | None = None, + t_max: int | float | None = None) -> dict[str, Any]: + """Accept a CVA statement or return a precise rejection reason. + + ``witness`` is an optional convenience mapping for callers that carry + all relation inputs together. Explicit keyword arguments take + precedence over values in that mapping. + """ + supplied = dict(witness or {}) + values = { + "principal_key_id": principal_key_id, + "key_registry": key_registry, + "request_payload": request_payload, + "context_payload": context_payload, + "attrs": attrs, + "policy": policy, + } + for name, value in list(values.items()): + if value is None and name in supplied: + values[name] = supplied[name] + + target = self.consumed_nonces if consumed is None else consumed + nonce = statement.get("nonce") if isinstance(statement, Mapping) else None + if isinstance(nonce, str) and nonce in target: + return {"accepted": False, "reason": "replay"} + + lower = self.t_min if t_min is None else t_min + upper = self.t_max if t_max is None else t_max + if lower is None: + lower = float("-inf") + if upper is None: + upper = float("inf") + timestamp = statement.get("timestamp_ms") if isinstance(statement, Mapping) else None + try: + if timestamp < lower: + return {"accepted": False, "reason": "stale_timestamp"} + if timestamp > upper: + return {"accepted": False, "reason": "future_timestamp"} + except (TypeError, ValueError): + # Let the relation report malformed statements consistently. + pass + + relation_ok, relation_errors = cva_relation_holds( + statement, + principal_key_id=values["principal_key_id"], + key_registry=values["key_registry"], + request_payload=values["request_payload"], + context_payload=values["context_payload"], + attrs=values["attrs"], + policy=values["policy"], + ) + if not relation_ok: + return { + "accepted": False, + "reason": "relation_not_satisfied", + "relation_errors": relation_errors, + } + + if not is_fresh(nonce, timestamp, target, lower, upper): + # The explicit boundary reasons above handle normal timestamps; + # this branch covers malformed/non-string freshness inputs. + if isinstance(nonce, str) and nonce in target: + reason = "replay" + elif timestamp < lower: + reason = "stale_timestamp" + elif timestamp > upper: + reason = "future_timestamp" + else: + reason = "relation_not_satisfied" + return {"accepted": False, "reason": reason} + + self.consume(statement, consumed=target) + return {"accepted": True, "reason": "accepted"} + + +PROPERTIES: list[dict[str, str]] = [ + { + "name": "authorization_soundness", + "paper_eq": "27-28", + "attack_class": "proof forgery or invalid-witness acceptance", + "ledger_mechanism": "hash-bound statements plus fail-closed evaluation of all four CVA conjuncts", + }, + { + "name": "principal_binding", + "paper_eq": "29", + "attack_class": "cross-principal proof transfer", + "ledger_mechanism": "normalized key-registry agent binding with revocation awareness", + }, + { + "name": "request_binding", + "paper_eq": "30", + "attack_class": "cross-request proof transfer", + "ledger_mechanism": "SHA-256 canonical request commitment in the CVA statement and AAR prebind", + }, + { + "name": "policy_binding", + "paper_eq": "31", + "attack_class": "cross-policy proof transfer", + "ledger_mechanism": "policy identifier committed in statement; supplied predicate must satisfy that binding", + }, + { + "name": "context_binding", + "paper_eq": "32-36", + "attack_class": "context substitution at authorization time", + "ledger_mechanism": "SHA-256 canonical context commitment and selected-context receipt hashes", + }, + { + "name": "replay_resistance", + "paper_eq": "37-40", + "attack_class": "nonce reuse and deferred presentation outside the validity window", + "ledger_mechanism": "trusted gateway nonce set with inclusive timestamp window", + }, +] + +# Descriptive alias for callers that prefer an explicit constant name. +CVA_PROPERTIES = PROPERTIES + +__all__ = [ + "CVA_STATEMENT_SCHEMA", + "PROPERTIES", + "CVA_PROPERTIES", + "build_cva_statement", + "cva_relation_holds", + "is_fresh", + "CvaGateway", +] diff --git a/ledger_agent/keys.py b/ledger_agent/keys.py index 707748e..3786bc5 100644 --- a/ledger_agent/keys.py +++ b/ledger_agent/keys.py @@ -53,12 +53,22 @@ def custody_label(value: Any) -> dict[str, Any]: return {"custody": v, "known": is_known_custody(v)} +def _copy_binding_metadata(source: Mapping[str, Any], target: dict[str, Any]) -> None: + """Preserve optional CVA identity/revocation metadata during normalization.""" + for field in ("agent_id", "agent_binding", "revoked", "revoked_at", "status"): + if field in source: + target[field] = source[field] + + def normalize_key_registry(registry: Optional[Mapping[str, Any]]) -> dict[str, dict[str, Any]]: """Normalize a key registry into labeled entries. Accepts both legacy ``{key_id: bytes}`` registries (entries carry no custody — rendered as labeled uncertainty) and labeled ``{key_id: {key_material: bytes, custody: str, label: str}}`` entries. + Optional ``agent_id``/``agent_binding`` and revocation fields are preserved + for CVA principal binding; existing signature consumers continue to use + only ``key_material``. """ out: dict[str, dict[str, Any]] = {} for key_id, entry in (registry or {}).items(): @@ -73,12 +83,14 @@ def normalize_key_registry(registry: Optional[Mapping[str, Any]]) -> dict[str, d entry.get("key_material"), (bytes, bytearray)): label = custody_label(entry.get("custody")) entry_label = entry.get("label") - out[key_id] = { + normalized = { "key_material": bytes(entry["key_material"]), "custody": label["custody"], "known": label["known"], "label": entry_label if isinstance(entry_label, str) else None, } + _copy_binding_metadata(entry, normalized) + out[key_id] = normalized else: raise ValueError( f"key_registry entry {key_id!r} must be bytes or " diff --git a/ledger_agent/prebind.py b/ledger_agent/prebind.py index 76c6809..2643eab 100644 --- a/ledger_agent/prebind.py +++ b/ledger_agent/prebind.py @@ -22,7 +22,7 @@ "evidence_hashes", "selected_context_digest", "resource_ref", "boundary_outcome", "non_effective_result", "replay_id", ) -_V2_FIELDS = {"stage_trace", "context_hash", "policy_hash", "uncertainty"} +_V2_FIELDS = {"stage_trace", "context_hash", "policy_hash", "uncertainty", "request_hash", "nonce", "epoch"} STAGE_VALUES = {"proposed", "approved", "leased", "executing", "completed", "failed", "cancelled", "interrupted", "recovered"} @@ -34,6 +34,17 @@ def _is_hash(value: Any) -> bool: return isinstance(value, str) and len(value) == 64 and all(char in "0123456789abcdef" for char in value.lower()) +def _optional_request_fields(request_hash: str | None, nonce: str | None, + epoch: int | str | None) -> tuple[str | None, str | None, int | str | None]: + if request_hash is not None and not _is_hash(request_hash): + raise ValueError("request_hash must be a 64-character SHA-256 hex digest") + if nonce is not None and (not isinstance(nonce, str) or not nonce.strip()): + raise ValueError("nonce must be a non-empty string") + if epoch is not None and (isinstance(epoch, bool) or not isinstance(epoch, (int, str))): + raise ValueError("epoch must be an integer or string") + return (request_hash.lower() if request_hash is not None else None, nonce, epoch) + + def _scan(value: Any, errors: list[str]) -> None: if isinstance(value, Mapping): for key, child in value.items(): @@ -101,7 +112,12 @@ def validate_prebind(block: Mapping[str, Any]) -> tuple[bool, list[str]]: if block.get("boundary_outcome") != "allow" and block.get("non_effective_result") == "executed": errors.append("outcome_result_mismatch") stage_refs = block.get("stage_refs") - if not isinstance(stage_refs, list) or any(not isinstance(value, str) or not value for value in stage_refs): + # The receipts.py v2 builder predates stage_refs and intentionally omits + # it; when present, retain the strict validation used by prebind.py. + if stage_refs is not None and ( + not isinstance(stage_refs, list) + or any(not isinstance(value, str) or not value for value in stage_refs) + ): errors.append("stage_refs") # v2-specific validation (#219, #220) @@ -132,6 +148,15 @@ def validate_prebind(block: Mapping[str, Any]) -> tuple[bool, list[str]]: uncertainty = block.get("uncertainty") if uncertainty is not None and not isinstance(uncertainty, str): errors.append("uncertainty_not_string") + request_hash = block.get("request_hash") + if request_hash is not None and not _is_hash(request_hash): + errors.append("prebind_request_hash") + nonce = block.get("nonce") + if nonce is not None and (not isinstance(nonce, str) or not nonce.strip()): + errors.append("prebind_nonce") + epoch = block.get("epoch") + if epoch is not None and (isinstance(epoch, bool) or not isinstance(epoch, (int, str))): + errors.append("prebind_epoch") supplied = block.get("prebind_hash") if not _is_hash(supplied) or supplied != prebind_digest(block): @@ -196,8 +221,12 @@ def build_prebind_v2(*, attempted_action: str, actor_ref: str, authority_ref: st stage_trace: dict[str, Any] | None = None, context_hash: str | None = None, policy_hash: str | None = None, - uncertainty: str | None = None) -> dict[str, Any]: + uncertainty: str | None = None, + request_hash: str | None = None, + nonce: str | None = None, + epoch: int | str | None = None) -> dict[str, Any]: """Build a v2 prebind with stage-aware fields and context/policy hashes.""" + request_hash, nonce, epoch = _optional_request_fields(request_hash, nonce, epoch) block: dict[str, Any] = { "schema_version": PREBIND_V2_SCHEMA, "attempted_action": attempted_action, @@ -217,6 +246,9 @@ def build_prebind_v2(*, attempted_action: str, actor_ref: str, authority_ref: st "context_hash": context_hash, "policy_hash": policy_hash, "uncertainty": uncertainty, + "request_hash": request_hash, + "nonce": nonce, + "epoch": epoch, } block["prebind_hash"] = prebind_digest(block) return block diff --git a/ledger_agent/receipts.py b/ledger_agent/receipts.py index 52bc3d0..6e5f5e1 100644 --- a/ledger_agent/receipts.py +++ b/ledger_agent/receipts.py @@ -52,6 +52,22 @@ def _opt_text(value: Optional[str], field: str, max_len: int = 512) -> Optional[ return value +def _opt_nonce(value: Optional[str]) -> Optional[str]: + if value is None: + return None + if not isinstance(value, str) or not value.strip(): + raise ValueError("nonce must be a non-empty string when supplied") + return value + + +def _opt_epoch(value: Optional[int | str]) -> Optional[int | str]: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, (int, str)): + raise ValueError("epoch must be an integer or string when supplied") + return value + + # ── #219 / #220: stage-aware prebind v2 ───────────────────────────────────── STAGE_VALUES = { @@ -94,7 +110,10 @@ def build_prebind_v2(*, attempted_action: str, actor_ref: str, authority_ref: st stage_trace: Optional[dict[str, Any]] = None, context_hash: Optional[str] = None, policy_hash: Optional[str] = None, - uncertainty: Optional[str] = None) -> dict[str, Any]: + uncertainty: Optional[str] = None, + request_hash: Optional[str] = None, + nonce: Optional[str] = None, + epoch: Optional[int | str] = None) -> dict[str, Any]: """Build a v2 prebind block with stage-aware fields.""" block: dict[str, Any] = { "schema_version": PREBIND_V2_SCHEMA, @@ -114,6 +133,9 @@ def build_prebind_v2(*, attempted_action: str, actor_ref: str, authority_ref: st "context_hash": _opt_hash(context_hash), "policy_hash": _opt_hash(policy_hash), "uncertainty": uncertainty, + "request_hash": _opt_hash(request_hash), + "nonce": _opt_nonce(nonce), + "epoch": _opt_epoch(epoch), } block["prebind_hash"] = _sha({k: v for k, v in block.items() if k != "prebind_hash"}) return block diff --git a/tests/test_cva.py b/tests/test_cva.py new file mode 100644 index 0000000..b7d4bf4 --- /dev/null +++ b/tests/test_cva.py @@ -0,0 +1,287 @@ +from __future__ import annotations + +import hashlib +import json + +from ledger_agent.cva import ( + PROPERTIES, + CvaGateway, + build_cva_statement, + cva_relation_holds, + is_fresh, +) +from ledger_agent.prebind import ( + build_prebind, + build_prebind_v2, + prebind_digest, + validate_prebind, +) +from ledger_agent.receipts import build_prebind_v2 as build_receipt_prebind_v2 + + +REQUEST = {"action": "deploy", "resource": "prod", "revision": 42} +CONTEXT = {"environment": "prod", "risk": "low"} +KEY = b"agent-a-key-material" + + +def digest(value): + return hashlib.sha256( + json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + ).hexdigest() + + +def policy_for(policy_id="policy/v1", allowed=True): + def predicate(attrs, request_payload, context_payload): + return ( + allowed + and attrs.get("role") == "deployer" + and request_payload["resource"] == "prod" + and context_payload["environment"] == "prod" + ) + + predicate.policy_id = policy_id + return predicate + + +def make_statement(*, nonce="n1", timestamp_ms=100, policy_id="policy/v1"): + return build_cva_statement( + agent_id="agent-a", + request_hash=digest(REQUEST), + context_hash=digest(CONTEXT), + policy_id=policy_id, + nonce=nonce, + timestamp_ms=timestamp_ms, + ) + + +def make_registry(*, agent_id="agent-a", revoked=False): + return { + "key-a": { + "key_material": KEY, + "custody": "self_held", + "agent_id": agent_id, + "revoked": revoked, + } + } + + +def relation(statement, *, request_payload=REQUEST, context_payload=CONTEXT, + registry=None, policy=None, principal_key_id="key-a"): + return cva_relation_holds( + statement, + principal_key_id=principal_key_id, + key_registry=registry or make_registry(), + request_payload=request_payload, + context_payload=context_payload, + attrs={"role": "deployer"}, + policy=policy or policy_for(), + ) + + +def gateway_accept(gateway, statement, *, request_payload=REQUEST, + context_payload=CONTEXT, registry=None, policy=None, + consumed=None, principal_key_id="key-a", t_min=100, t_max=200): + return gateway.accept( + statement, + principal_key_id=principal_key_id, + key_registry=registry or make_registry(), + request_payload=request_payload, + context_payload=context_payload, + attrs={"role": "deployer"}, + policy=policy or policy_for(), + consumed=consumed, + t_min=t_min, + t_max=t_max, + ) + + +def test_cva_relation_accepts_bound_request_context_principal_and_policy(): + valid, errors = relation(make_statement()) + assert valid is True + assert errors == [] + + +def test_authorization_soundness_rejects_tampered_request_payload(): + valid, errors = relation(make_statement(), request_payload={**REQUEST, "revision": 43}) + assert valid is False + assert errors == ["bind_request"] + + +def test_principal_binding_rejects_key_bound_to_another_agent(): + valid, errors = relation(make_statement(), registry=make_registry(agent_id="agent-b")) + assert valid is False + assert errors == ["bind_principal"] + + +def test_request_binding_rejects_different_request_with_same_context(): + valid, errors = relation(make_statement(), request_payload={"action": "delete", "resource": "prod"}) + assert valid is False + assert "bind_request" in errors + + +def test_policy_binding_rejects_flipped_predicate_and_policy_identifier(): + statement = make_statement() + valid, errors = relation(statement, policy=policy_for(allowed=False)) + assert valid is False + assert errors == ["satisfy_policy"] + + valid, errors = relation(statement, policy=policy_for(policy_id="policy/v2")) + assert valid is False + assert "satisfy_policy" in errors + + +def test_context_binding_rejects_changed_context_payload(): + valid, errors = relation(make_statement(), context_payload={"environment": "staging", "risk": "low"}) + assert valid is False + assert "bind_context" in errors + + +def test_replay_resistance_consumes_nonce_once(): + gateway = CvaGateway() + statement = make_statement() + first = gateway_accept(gateway, statement) + second = gateway_accept(gateway, statement) + assert first == {"accepted": True, "reason": "accepted"} + assert second == {"accepted": False, "reason": "replay"} + + +def test_timestamp_window_rejects_stale_and_future_statements(): + gateway = CvaGateway() + stale = gateway_accept(gateway, make_statement(timestamp_ms=99)) + future = gateway_accept(gateway, make_statement(nonce="n2", timestamp_ms=201)) + assert stale == {"accepted": False, "reason": "stale_timestamp"} + assert future == {"accepted": False, "reason": "future_timestamp"} + + +def test_failed_relation_does_not_consume_nonce(): + gateway = CvaGateway() + statement = make_statement() + rejected = gateway_accept( + gateway, statement, request_payload={**REQUEST, "revision": 999}) + assert rejected["accepted"] is False + assert rejected["reason"] == "relation_not_satisfied" + assert "n1" not in gateway.consumed_nonces + + accepted = gateway_accept(gateway, statement) + assert accepted == {"accepted": True, "reason": "accepted"} + + +def test_is_fresh_requires_unused_nonce_and_inclusive_timestamp_window(): + assert is_fresh("n1", 100, set(), 100, 200) is True + assert is_fresh("n1", 100, {"n1"}, 100, 200) is False + assert is_fresh("n2", 99, set(), 100, 200) is False + assert is_fresh("n3", 201, set(), 100, 200) is False + + +def test_receipt_prebind_carries_request_nonce_epoch_and_hash_covers_them(): + block = build_receipt_prebind_v2( + attempted_action="deploy", + actor_ref="agent-a", + authority_ref="authority:1", + trusted_scope="repo:ledger", + policy_version="policy/v1", + evidence_hashes=[digest("evidence")], + selected_context_digest=digest("selection"), + resource_ref="resource:prod", + boundary_outcome="hold", + non_effective_result="not_executed", + replay_id="replay:1", + context_hash=digest(CONTEXT), + policy_hash=digest("policy/v1"), + request_hash=digest(REQUEST), + nonce="n1", + epoch="epoch-1", + ) + assert block["request_hash"] == digest(REQUEST) + assert block["nonce"] == "n1" + assert block["epoch"] == "epoch-1" + assert validate_prebind(block) == (True, []) + tampered = dict(block, request_hash=digest({"action": "other"})) + assert prebind_digest(tampered) != block["prebind_hash"] + + +def test_prebind_validator_accepts_valid_new_fields(): + block = build_prebind_v2( + attempted_action="deploy", + actor_ref="agent-a", + authority_ref="authority:1", + trusted_scope="repo:ledger", + policy_version="policy/v1", + evidence_hashes=[digest("evidence")], + selected_context_digest=digest("selection"), + resource_ref="resource:prod", + boundary_outcome="hold", + non_effective_result="not_executed", + replay_id="replay:1", + request_hash=digest(REQUEST), + nonce="n1", + epoch=1, + ) + assert validate_prebind(block) == (True, []) + + +def test_prebind_validator_rejects_bad_request_hash_and_nonce(): + block = build_prebind_v2( + attempted_action="deploy", + actor_ref="agent-a", + authority_ref="authority:1", + trusted_scope="repo:ledger", + policy_version="policy/v1", + evidence_hashes=[digest("evidence")], + selected_context_digest=digest("selection"), + resource_ref="resource:prod", + boundary_outcome="hold", + non_effective_result="not_executed", + replay_id="replay:1", + request_hash=digest(REQUEST), + nonce="n1", + ) + block["request_hash"] = "not-a-hash" + block["nonce"] = "" + block["prebind_hash"] = prebind_digest(block) + valid, errors = validate_prebind(block) + assert valid is False + assert "prebind_request_hash" in errors + assert "prebind_nonce" in errors + + +def test_old_style_prebind_without_cva_fields_remains_valid(): + block = build_prebind( + attempted_action="deploy", + actor_ref="agent-a", + authority_ref="authority:1", + trusted_scope="repo:ledger", + policy_version="policy/v1", + evidence_hashes=[digest("evidence")], + selected_context_digest=digest("selection"), + resource_ref="resource:prod", + boundary_outcome="hold", + non_effective_result="not_executed", + replay_id="replay:1", + ) + assert "request_hash" not in block + assert "nonce" not in block + assert validate_prebind(block) == (True, []) + + +def test_cva_statement_schema_hash_and_determinism(): + first = make_statement() + second = make_statement() + assert first == second + assert first["schema"] == "perseus-ledger-cva-statement/v1" + assert len(first["statement_hash"]) == 64 + changed = make_statement(nonce="n2") + assert changed["statement_hash"] != first["statement_hash"] + + +def test_cva_properties_cover_paper_security_matrix(): + names = {entry["name"] for entry in PROPERTIES} + assert names == { + "authorization_soundness", + "principal_binding", + "request_binding", + "policy_binding", + "context_binding", + "replay_resistance", + } + assert all(entry["paper_eq"] for entry in PROPERTIES) diff --git a/tests/test_prebind_replay_api.py b/tests/test_prebind_replay_api.py index 3994744..5c3a640 100644 --- a/tests/test_prebind_replay_api.py +++ b/tests/test_prebind_replay_api.py @@ -26,4 +26,113 @@ def test_stored_prebind_replay_is_non_mutating(tmp_path): conn.close() -__all__ = ["test_stored_prebind_replay_is_non_mutating"] +# ── authority-trace v2: CVA replay resistance ─────────────────────────────── + +def _cva_digest(value): + import hashlib + import json + + return hashlib.sha256( + json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + ).hexdigest() + + +def test_authority_trace_v2_cva_replay_resistance_and_key_rotation(tmp_path): + from ledger_agent.cva import CvaGateway, build_cva_statement + from ledger_agent.receipts import build_prebind_v2 + + request = {"action": "deploy", "resource": "prod"} + context = {"environment": "prod", "risk": "low"} + policy = lambda attrs, req, ctx: attrs.get("role") == "deployer" and req["resource"] == "prod" + old_key_registry = { + "old": {"key_material": b"old-key", "custody": "self_held", "agent_id": "agent-a"}, + "new": {"key_material": b"new-key", "custody": "self_held", "agent_id": "agent-a"}, + } + block = build_prebind_v2( + attempted_action="deploy", + actor_ref="agent-a", + authority_ref="authority:1", + trusted_scope="repo:ledger", + policy_version="policy/v1", + evidence_hashes=[_cva_digest("evidence")], + selected_context_digest=_cva_digest("selection"), + resource_ref="resource:prod", + boundary_outcome="allow", + non_effective_result="not_executed", + replay_id="replay:n1", + context_hash=_cva_digest(context), + policy_hash=_cva_digest("policy/v1"), + request_hash=_cva_digest(request), + nonce="n1", + epoch=100, + ) + statement = build_cva_statement( + agent_id=block["actor_ref"], + request_hash=block["request_hash"], + context_hash=block["context_hash"], + policy_id=block["policy_version"], + nonce=block["nonce"], + timestamp_ms=block["epoch"], + ) + gateway = CvaGateway() + kwargs = { + "principal_key_id": "old", + "key_registry": old_key_registry, + "request_payload": request, + "context_payload": context, + "attrs": {"role": "deployer"}, + "policy": policy, + "t_min": 0, + "t_max": 200, + } + assert gateway.accept(statement, **kwargs) == {"accepted": True, "reason": "accepted"} + assert gateway.accept(statement, **kwargs) == {"accepted": False, "reason": "replay"} + + rotated_registry = { + "old": {"key_material": b"old-key", "custody": "self_held", "agent_id": "agent-a", "revoked": True}, + "new": {"key_material": b"new-key", "custody": "self_held", "agent_id": "agent-a"}, + } + old_witness_statement = build_cva_statement( + agent_id="agent-a", + request_hash=_cva_digest(request), + context_hash=_cva_digest(context), + policy_id="policy/v1", + nonce="old-after-rotation", + timestamp_ms=100, + ) + old_witness = dict(kwargs, key_registry=rotated_registry, principal_key_id="old") + old_result = gateway.accept(old_witness_statement, **old_witness) + assert old_result["accepted"] is False + assert old_result["reason"] == "relation_not_satisfied" + assert "bind_principal" in old_result["relation_errors"] + + stale_statement = build_cva_statement( + agent_id="agent-a", + request_hash=_cva_digest(request), + context_hash=_cva_digest(context), + policy_id="policy/v1", + nonce="stale-context", + timestamp_ms=100, + ) + stale_witness = dict(old_witness, principal_key_id="new", context_payload={"environment": "staging", "risk": "low"}) + stale_result = gateway.accept(stale_statement, **stale_witness) + assert stale_result["accepted"] is False + assert stale_result["reason"] == "relation_not_satisfied" + assert "bind_context" in stale_result["relation_errors"] + + fresh_statement = build_cva_statement( + agent_id="agent-a", + request_hash=_cva_digest(request), + context_hash=_cva_digest({"environment": "staging", "risk": "low"}), + policy_id="policy/v1", + nonce="fresh-after-rotation", + timestamp_ms=100, + ) + fresh_result = gateway.accept(fresh_statement, **stale_witness) + assert fresh_result == {"accepted": True, "reason": "accepted"} + + +__all__ = [ + "test_stored_prebind_replay_is_non_mutating", + "test_authority_trace_v2_cva_replay_resistance_and_key_rotation", +]