""" Core path-proof receipt primitives: canonical JSON, receipt construction, deterministic verification, and the spent-nonce / idempotency registry. No I/O, no clock, no randomness. All time-sensitive logic takes ``now`` (epoch seconds) as an explicit argument so every test is deterministic from a clean checkout and leaves no generated artifacts. See ``docs/path-proof-receipt.md`` (on ``timmy/3-path-proof-receipt``) for the field rationale. This module implements the *verification gate* for that contract, not the runtime recorder itself. """ from __future__ import annotations import hashlib import json from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple SPEND_KIND_NONCE = "nonce" SPEND_KIND_ACTION_ID = "action_id" # --------------------------------------------------------------------------- # Canonical JSON (RFC 8785 subset, sufficient for the receipt envelope) # --------------------------------------------------------------------------- def canonical_json(obj: Any) -> bytes: """Deterministic JSON bytes: sorted keys, no whitespace, UTF-8. This is the byte form the signature commits to. It is a faithful subset of RFC 8785 (JCS) for the object/array/string/number/bool/null shapes the receipt uses: keys sorted by code point, no insignificant whitespace, floats rendered via ``repr``-stable ``json`` (receipt fields are ints/strs). """ text = json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return text.encode("utf-8") def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest() # --------------------------------------------------------------------------- # Spent-nonce / idempotency registry (deterministic, in-memory) # --------------------------------------------------------------------------- @dataclass(frozen=True) class SpendRecord: """A single consumed identity (nonce or action_id) with its binding.""" kind: str # SPEND_KIND_NONCE | SPEND_KIND_ACTION_ID value: str receipt_action_id: str signer: str accepted_at: int # epoch seconds of first (and only) acceptance class SpentRegistry: """Proves the same valid receipt cannot be accepted twice. Idempotency is enforced on two axes: * ``nonce`` -- the verifier-issued single-use token. * ``action_id`` -- the unique identity of the consequential action. A receipt is *fresh* only if neither its nonce nor its action_id has been previously consumed. Accepting it marks both as spent. """ def __init__(self) -> None: self._spends: List[SpendRecord] = [] def spent_nonces(self) -> set: return {r.value for r in self._spends if r.kind == SPEND_KIND_NONCE} def spent_action_ids(self) -> set: return {r.value for r in self._spends if r.kind == SPEND_KIND_ACTION_ID} def is_fresh(self, nonce: str, action_id: str) -> Tuple[bool, str]: """Return (fresh, reason). Fresh only if neither identity is spent.""" if nonce in self.spent_nonces(): return False, f"nonce {nonce!r} already spent (replay)" if action_id in self.spent_action_ids(): return False, f"action_id {action_id!r} already spent (replay)" return True, "" def mark_spent(self, nonce: str, action_id: str, receipt_action_id: str, signer: str, accepted_at: int) -> None: self._spends.append(SpendRecord(SPEND_KIND_NONCE, nonce, receipt_action_id, signer, accepted_at)) self._spends.append(SpendRecord(SPEND_KIND_ACTION_ID, action_id, receipt_action_id, signer, accepted_at)) def records(self) -> List[SpendRecord]: return list(self._spends) def __len__(self) -> int: return len(self._spends) # --------------------------------------------------------------------------- # Receipt construction (trusted-recorder side, for fixture building) # --------------------------------------------------------------------------- def receipt_payload(action: Dict[str, Any], path: Dict[str, Any], evidence: Dict[str, Any], attestation: Dict[str, Any]) -> Dict[str, Any]: """Assemble the full receipt envelope (``v`` + 4 sections).""" return { "v": 0, "action": action, "path": path, "evidence": evidence, "attestation": attestation, } def build_receipt( *, kind: str, target: str, request_sha256: str, result_sha256: str, model_id: str, model_invocation: str, tool_trace_sha256: str, policy_bundle_sha256: str, code_repo: str, code_commit: str, code_entrypoint: str, evidence_uri: str, issuer: str, key_id: str, sig: str, action_id: str, nonce: str, issued_at: int, expires_at: int, ) -> Dict[str, Any]: """Build a complete v0 receipt with explicit binding + freshness fields. ``action_id`` is the unique identity of the consequential action; ``nonce`` is the verifier-issued single-use token; ``issued_at`` / ``expires_at`` are epoch-seconds validity bounds. These four are the replay/forgery controls layered on top of the base envelope. """ action = { "kind": kind, "target": target, "request_sha256": request_sha256, "result_sha256": result_sha256, } path = { "model": {"id": model_id, "invocation": model_invocation}, "tools": {"trace_sha256": tool_trace_sha256}, "policy": {"bundle_sha256": policy_bundle_sha256}, "code": {"repo": code_repo, "commit": code_commit, "entrypoint": code_entrypoint}, } evidence = {"uri": evidence_uri} attestation = { "issuer": issuer, "key_id": key_id, "alg": "Ed25519", "sig": sig, "action_id": action_id, "nonce": nonce, "issued_at": issued_at, "expires_at": expires_at, } return receipt_payload(action, path, evidence, attestation) # --------------------------------------------------------------------------- # Verification # --------------------------------------------------------------------------- class VerifyResult: """A structured, fail-closed verification outcome. ``ok`` is True only when *every* check passed. ``reasons`` is a list of human-readable failure reasons (empty on success). ``status`` is one of: * ``"verified"`` -- every check passed * ``"unverified"`` -- one or more checks failed (never a partial pass) * ``"malformed"`` -- the structure itself could not be read """ def __init__(self, status: str, ok: bool, reasons: Optional[List[str]] = None): self.status = status self.ok = ok self.reasons: List[str] = reasons or [] def reason(self) -> str: return "; ".join(self.reasons) if self.reasons else "" def __repr__(self) -> str: return f"VerifyResult(status={self.status!r}, ok={self.ok}, reasons={self.reasons!r})" def _is_sha256(value: Any) -> bool: if not isinstance(value, str): return False if len(value) != 64: return False try: int(value, 16) except ValueError: return False return True def _is_epoch_int(value: Any) -> bool: # bool is a subclass of int; reject it explicitly. return isinstance(value, int) and not isinstance(value, bool) and value >= 0 def _check_envelope_shape(payload: Any) -> Optional[VerifyResult]: """Return a malformed result if the top-level shape is unusable.""" if not isinstance(payload, dict): return VerifyResult("malformed", False, ["receipt must be a JSON object"]) if payload.get("v") != 0: return VerifyResult("malformed", False, ["unsupported receipt version (v must be 0)"]) for section in ("action", "path", "evidence", "attestation"): if section not in payload or not isinstance(payload[section], dict): return VerifyResult("malformed", False, [f"missing or non-object section: {section!r}"]) return None def verify_receipt( payload: Any, registry, now: int, *, expected_target: Optional[str] = None, expected_result_sha256: Optional[str] = None, ) -> VerifyResult: """Verify a bound receipt. Pure, deterministic, fail-closed. Checks, in order (all must pass for ``verified``): 1. envelope shape (else ``malformed``) 2. signature present and well-formed 3. attestation identity resolves to a trusted, non-revoked, in-validity key 4. action binding: kind/target/request/result all present; result is sha256 5. request/result substitution guard: ``expected_result_sha256`` (if given) must equal the bound ``result_sha256`` 6. target/resource guard: ``expected_target`` (if given) must equal ``target`` 7. path binding: model id+invocation, tool trace, policy bundle, code identity all present and well-formed (hashes are sha256) 8. evidence uri present and non-empty 9. freshness: issued_at < expires_at; now in [issued_at, expires_at] 10. single-use identity: nonce + action_id present, non-empty, unique-shaped This is a *pure* check: it does not mark anything spent. Use :func:`verify_and_accept` to also enforce idempotency against a registry. ``now`` is the explicit current time (epoch seconds) — never read from the wall clock — so verification is reproducible. """ malformed = _check_envelope_shape(payload) if malformed is not None: return malformed reasons: List[str] = [] action = payload["action"] path = payload["path"] evidence = payload["evidence"] att = payload["attestation"] # 2. signature sig = att.get("sig") if not isinstance(sig, str) or not sig: reasons.append("signature missing or empty") # 3. signer trust issuer = att.get("issuer") key_id = att.get("key_id") if not isinstance(issuer, str) or not issuer: reasons.append("attestation.issuer missing") if not isinstance(key_id, str) or not key_id: reasons.append("attestation.key_id missing") if isinstance(issuer, str) and isinstance(key_id, str): key = registry.resolve(issuer, key_id) if key is None: reasons.append(f"unknown signer: {issuer!r}/{key_id!r} not in the trusted registry") elif key.is_trusted(now) is False: if key.revoked: reasons.append(f"signer revoked: {issuer!r}/{key_id!r}") else: reasons.append(f"signer key out of validity: {issuer!r}/{key_id!r}") # 4. action binding if not isinstance(action.get("kind"), str) or not action.get("kind"): reasons.append("action.kind missing or empty") target = action.get("target") if not isinstance(target, str) or not target: reasons.append("action.target missing or empty") if not _is_sha256(action.get("request_sha256")): reasons.append("action.request_sha256 missing or not sha256 hex") if not _is_sha256(action.get("result_sha256")): reasons.append("action.result_sha256 missing or not sha256 hex") # 5. result substitution guard if expected_result_sha256 is not None and action.get("result_sha256") != expected_result_sha256: reasons.append("result_sha256 does not match the expected observed result (substitution)") # 6. target guard if expected_target is not None and target != expected_target: reasons.append("action.target does not match the expected target/resource (substitution)") # 7. path binding model = path.get("model") if not isinstance(model, dict): reasons.append("path.model missing or non-object") else: if not isinstance(model.get("id"), str) or not model.get("id"): reasons.append("path.model.id missing or empty") if not isinstance(model.get("invocation"), str) or not model.get("invocation"): reasons.append("path.model.invocation missing or empty") tools = path.get("tools") if not isinstance(tools, dict) or not _is_sha256(tools.get("trace_sha256")): reasons.append("path.tools.trace_sha256 missing or not sha256 hex") policy = path.get("policy") if not isinstance(policy, dict) or not _is_sha256(policy.get("bundle_sha256")): reasons.append("path.policy.bundle_sha256 missing or not sha256 hex") code = path.get("code") if not isinstance(code, dict): reasons.append("path.code missing or non-object") else: if not isinstance(code.get("repo"), str) or not code.get("repo"): reasons.append("path.code.repo missing or empty") if not isinstance(code.get("commit"), str) or not code.get("commit"): reasons.append("path.code.commit missing or empty") if not isinstance(code.get("entrypoint"), str) or not code.get("entrypoint"): reasons.append("path.code.entrypoint missing or empty") # 8. evidence uri = evidence.get("uri") if not isinstance(uri, str) or not uri: reasons.append("evidence.uri missing or empty") # 9. freshness issued_at = att.get("issued_at") expires_at = att.get("expires_at") if not _is_epoch_int(issued_at): reasons.append("attestation.issued_at missing or not a non-negative epoch int") if not _is_epoch_int(expires_at): reasons.append("attestation.expires_at missing or not a non-negative epoch int") if _is_epoch_int(issued_at) and _is_epoch_int(expires_at): if issued_at >= expires_at: reasons.append("attestation validity window invalid (issued_at >= expires_at)") else: if now < issued_at: reasons.append("receipt not yet valid (issued_at is in the future)") elif now > expires_at: reasons.append("receipt expired (now > expires_at)") # 10. single-use identity presence action_id = att.get("action_id") nonce = att.get("nonce") if not isinstance(action_id, str) or not action_id: reasons.append("attestation.action_id missing or empty") if not isinstance(nonce, str) or not nonce: reasons.append("attestation.nonce missing or empty") if reasons: return VerifyResult("unverified", False, reasons) return VerifyResult("verified", True, []) def verify_and_accept( payload: Any, registry, spent: SpentRegistry, now: int, *, expected_target: Optional[str] = None, expected_result_sha256: Optional[str] = None, ) -> VerifyResult: """Verify a receipt AND enforce single-use idempotency. On success the receipt's nonce and action_id are marked spent, so an identical replay is rejected with a specific replay reason. On failure nothing is marked spent (no side effects from a rejected receipt). Returns a :class:`VerifyResult`. A replay of an already-accepted receipt is ``unverified`` with a ``replay`` reason, not a silent second success. """ result = verify_receipt(payload, registry, now, expected_target=expected_target, expected_result_sha256=expected_result_sha256) if not result.ok: return result att = payload["attestation"] nonce = att["nonce"] action_id = att["action_id"] issuer = att["issuer"] fresh, why = spent.is_fresh(nonce, action_id) if not fresh: return VerifyResult("unverified", False, [why]) spent.mark_spent(nonce, action_id, action_id, issuer, now) return VerifyResult("verified", True, [])