diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4315742 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +__pycache__/ +*.pyc +*.pyo +.pytest_cache/ diff --git a/docs/path-proof-threat-boundary.md b/docs/path-proof-threat-boundary.md new file mode 100644 index 0000000..a0b32d6 --- /dev/null +++ b/docs/path-proof-threat-boundary.md @@ -0,0 +1,66 @@ +# Path Proof Receipt — Threat Boundary + +This document states, precisely, what the `path_proof` verification gate proves +and does not prove. It is the contract the verifier enforces and the limit a +consumer must respect. The field rationale lives in +`docs/path-proof-receipt.md` (branch `timmy/3-path-proof-receipt`); this file +is the executable-side boundary statement required by issue #28. + +## What a verified, freshly-accepted receipt proves + +A receipt that passes `verify_receipt` and is accepted by `verify_and_accept` +proves exactly the following, and nothing more: + +1. **Binding.** The receipt commits, as a single signed unit, to: + - the consequential action (`action.kind`, `action.target`), + - the exact request and observed result it claims + (`action.request_sha256`, `action.result_sha256`), + - the model invocation identity (`path.model.id`, `path.model.invocation`), + - the ordered tool path and effective policy + (`path.tools.trace_sha256`, `path.policy.bundle_sha256`), + - the immutable code identity (`path.code.repo`, `path.code.commit`, + `path.code.entrypoint`), and + - the retrievable evidence bundle (`evidence.uri`). +2. **Trust.** The attestation resolves to a signer that is present in the + caller's trusted key registry, is not revoked, and is within its validity + window at the time of verification. +3. **Freshness.** The receipt is inside its `[issued_at, expires_at]` window. +4. **Single-use.** The receipt's `nonce` and `action_id` have not been + previously consumed, so an identical receipt cannot be accepted a second + time (idempotency / replay protection). + +## What a verified receipt does NOT prove + +- **It does not prove runtime execution of the consequential effect.** + `result_sha256` is the hash of an *observed* result the recorder claims it + saw. Proving the side effect actually took place requires the verifier to + independently re-observe it (for example, read the Gitea label state back and + compare it to `result_sha256`). Until that happens the effect is *attested*, + not *confirmed*. +- **It does not prove the provider ran the advertised weights.** `path.model.id` + is an identity string, not a cryptographic attestation of model provenance. + Only a separately trusted provider model attestation could close that gap. +- **It does not survive compromise of both the recorder and its signing key.** + The trust anchor is the key registry. If both are compromised, a forged + receipt with a valid signature is indistinguishable from a genuine one. +- **It does not prove the evidence bundle was not selectively redacted + ambiguously.** The redaction manifest is part of the bundle; verifying the + bundle digest proves integrity of what is present, not that nothing material + was omitted. + +## Failure semantics + +- Any single failed check yields status **`unverified`** with specific reasons + — never a partial pass. +- A structurally unreadable receipt yields status **`malformed`**. +- A rejected receipt has **no side effects**: it does not consume a nonce or + action_id. Only a verified acceptance marks identities as spent. + +## Determinism guarantees + +- Verification takes `now` (epoch seconds) as an explicit argument and never + reads the wall clock. +- The key registry and spent-identity registry are in-memory objects the + caller controls; the module performs no I/O, no randomness, and no network. +- Therefore the gate is fully reproducible from a clean checkout and leaves no + generated artifacts (test caches are gitignored). diff --git a/path_proof/__init__.py b/path_proof/__init__.py new file mode 100644 index 0000000..66962b2 --- /dev/null +++ b/path_proof/__init__.py @@ -0,0 +1,52 @@ +""" +Path Proof receipt replay/forgery gate. + +A dependency-free verifier for runtime-attested consequential-action receipts. +See ``docs/path-proof-receipt.md`` for the threat model and field rationale. + +Threat boundary +--------------- +A valid, freshly-accepted receipt proves exactly one thing: that a trusted +signer attested a specific bound action (request + result + target/resource + +model invocation + code identity) within its validity window, and that the +nonce/action-id has not been spent. It does NOT prove the provider executed +the advertised weights, and it does NOT prove the consequential side effect +actually took place at runtime. Runtime proof requires the verifier to +independently re-observe the effect (e.g. read Gitea state back and compare +``result_sha256``). Anything outside those bounds is reported ``unverified``, +never partial-success. + +Determinism +----------- +All verification takes ``now`` as an explicit parameter (no hidden clock), and +the spent-nonce registry is a plain in-memory object the caller controls. +There is no I/O, no randomness, and no network in this module. +""" + +from . import keys +from .keys import SignerKey +from .receipt import ( + SPEND_KIND_NONCE, + SPEND_KIND_ACTION_ID, + SpendRecord, + SpentRegistry, + canonical_json, + build_receipt, + receipt_payload, + verify_receipt, + verify_and_accept, +) + +__all__ = [ + "SPEND_KIND_NONCE", + "SPEND_KIND_ACTION_ID", + "SignerKey", + "SpendRecord", + "SpentRegistry", + "canonical_json", + "build_receipt", + "receipt_payload", + "verify_receipt", + "verify_and_accept", + "keys", +] diff --git a/path_proof/keys.py b/path_proof/keys.py new file mode 100644 index 0000000..b4f3fc0 --- /dev/null +++ b/path_proof/keys.py @@ -0,0 +1,84 @@ +""" +Trusted-signer key registry for path-proof receipts. + +Dependency-free. Keys are Ed25519 public keys (raw 32-byte bytes) or their +hex encoding. Verification here is a *simulated* cryptographic check: the +signature must equal the canonical commitment bound to the key id. In a +production recorder the signature would be a real Ed25519 signature and the +key a verified public key from a hardware/service-isolated key store; the +shape and the trust decision (known vs unknown vs revoked) are identical. + +The trust decision is what the gate enforces: + - ``known``: (issuer, key_id) is present and not revoked + - ``unknown``: not present + - ``revoked``: present but revoked + +Everything is deterministic and in-memory so tests are reproducible from a +clean checkout with no generated artifacts. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, Optional, Tuple + + +@dataclass(frozen=True) +class SignerKey: + """A trusted runtime-recorder signing identity. + + ``public_key`` is the raw Ed25519 public key bytes (32 bytes) or hex text. + ``valid_until`` is an explicit epoch-seconds bound (``None`` = no bound). + """ + + issuer: str + key_id: str + public_key: str + valid_until: Optional[int] = None + revoked: bool = False + + def is_trusted(self, now: int) -> bool: + if self.revoked: + return False + if self.valid_until is not None and now > self.valid_until: + return False + return True + + +class KeyRegistry: + """In-memory registry of trusted signer keys, keyed by (issuer, key_id).""" + + def __init__(self) -> None: + self._keys: Dict[Tuple[str, str], SignerKey] = {} + + def add(self, key: SignerKey) -> None: + self._keys[(key.issuer, key.key_id)] = key + + def add_hex(self, issuer: str, key_id: str, public_key_hex: str, + valid_until: Optional[int] = None, revoked: bool = False) -> SignerKey: + key = SignerKey(issuer=issuer, key_id=key_id, public_key=public_key_hex, + valid_until=valid_until, revoked=revoked) + self.add(key) + return key + + def resolve(self, issuer: str, key_id: str) -> Optional[SignerKey]: + return self._keys.get((issuer, key_id)) + + def __len__(self) -> int: + return len(self._keys) + + def __contains__(self, item: Tuple[str, str]) -> bool: + return item in self._keys + + +def make_demo_registry() -> KeyRegistry: + """A deterministic demo registry: two trusted recorders, one revoked.""" + reg = KeyRegistry() + reg.add_hex("recorder:forge", "recorder:forge/key-1", "aa" * 32) + reg.add_hex("recorder:gateway", "recorder:gateway/key-1", "bb" * 32) + reg.add_hex("recorder:old", "recorder:old/key-1", "cc" * 32, revoked=True) + return reg + + +# Backwards-compatible alias used by the public package surface. +keys = make_demo_registry() diff --git a/path_proof/receipt.py b/path_proof/receipt.py new file mode 100644 index 0000000..1a852ff --- /dev/null +++ b/path_proof/receipt.py @@ -0,0 +1,406 @@ +""" +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, []) diff --git a/tests/test_path_proof.py b/tests/test_path_proof.py new file mode 100644 index 0000000..4208e21 --- /dev/null +++ b/tests/test_path_proof.py @@ -0,0 +1,320 @@ +""" +Executable replay/forgery gate tests for path-proof receipts (#28). + +Every negative case asserts BOTH rejection and a specific reason. The suite is +deterministic: time is passed explicitly, the key registry and spent-nonce +registry are in-memory, and nothing is written to disk. It runs from a clean +checkout and leaves no generated artifacts. + +Import shim: ``path_proof`` lives at the repo root (a sibling of ``tests/``), +so we add the repo root to ``sys.path``. ``path_proof`` is a normal underscore +package, so a plain import works (no importlib-by-path gymnastics needed). +""" + +import copy +import sys +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +from path_proof import ( # noqa: E402 + SpentRegistry, + build_receipt, + verify_and_accept, + verify_receipt, +) +from path_proof.keys import make_demo_registry # noqa: E402 + +NOW = 1_750_000_000 # a fixed "now" in epoch seconds +ISSUED = NOW - 3600 # issued an hour ago +EXPIRES = NOW + 3600 # expires an hour from now + +# Fixed sha256-shaped identifiers (64 hex chars) for deterministic fixtures. +REQ_SHA = "11" * 32 +RESULT_SHA = "22" * 32 +TOOL_TRACE_SHA = "33" * 32 +POLICY_SHA = "44" * 32 +EVIDENCE_URI = "sha256:" + "55" * 32 +TARGET = "stackchain/stackchain-lab-loop#3" + + +def _registry(): + """Fresh demo registry: two trusted recorders, one revoked.""" + return make_demo_registry() + + +def _receipt(**overrides): + """Build a valid, fresh, trusted v0 receipt (defaults = pass case).""" + base = dict( + kind="gitea.issue.labels.replace", + target=TARGET, + request_sha256=REQ_SHA, + result_sha256=RESULT_SHA, + model_id="custom/qwen-3.8-27b", + model_invocation="recorder:forge/inv-1", + tool_trace_sha256=TOOL_TRACE_SHA, + policy_bundle_sha256=POLICY_SHA, + code_repo="stackchain/stackchain-lab-loop", + code_commit="a" * 40, + code_entrypoint="scripts/lab_loop.py", + evidence_uri=EVIDENCE_URI, + issuer="recorder:forge", + key_id="recorder:forge/key-1", + sig="c2lnbmF0dXJlLWJvdW5kLXRvLWNhbm9uaWNhbC1ieXRlcy0x", + action_id="act-0001", + nonce="nonce-0001", + issued_at=ISSUED, + expires_at=EXPIRES, + ) + base.update(overrides) + return build_receipt(**base) + + +def _assert_rejected(result, *needles): + """Assert the result was rejected AND each needle reason is present.""" + self_ok = result.ok is False + assert self_ok, f"expected rejection but got ok={result.ok} reasons={result.reasons}" + for needle in needles: + assert any(needle in r for r in result.reasons), ( + f"expected a reason containing {needle!r}, got {result.reasons}" + ) + + +# --------------------------------------------------------------------------- +# Positive cases +# --------------------------------------------------------------------------- + +class PositiveTests(unittest.TestCase): + def test_valid_receipt_passes(self): + result = verify_receipt(_receipt(), _registry(), NOW) + self.assertEqual(result.status, "verified") + self.assertTrue(result.ok) + self.assertEqual(result.reasons, []) + + def test_valid_receipt_accepts_and_binds(self): + spent = SpentRegistry() + result = verify_and_accept(_receipt(), _registry(), spent, NOW) + self.assertTrue(result.ok) + self.assertIn("nonce-0001", spent.spent_nonces()) + self.assertIn("act-0001", spent.spent_action_ids()) + + def test_expected_target_match_passes(self): + result = verify_receipt(_receipt(), _registry(), NOW, expected_target=TARGET) + self.assertTrue(result.ok, result.reasons) + + def test_expected_result_match_passes(self): + result = verify_receipt(_receipt(), _registry(), NOW, + expected_result_sha256=RESULT_SHA) + self.assertTrue(result.ok, result.reasons) + + +# --------------------------------------------------------------------------- +# Replay / idempotency +# --------------------------------------------------------------------------- + +class ReplayTests(unittest.TestCase): + def test_replayed_receipt_fails(self): + spent = SpentRegistry() + first = verify_and_accept(_receipt(), _registry(), spent, NOW) + self.assertTrue(first.ok, "first accept should pass") + + replay = verify_and_accept(_receipt(), _registry(), spent, NOW) + _assert_rejected(replay, "replay") + self.assertFalse(replay.ok) + + def test_replay_by_nonce_is_specific(self): + spent = SpentRegistry() + verify_and_accept(_receipt(nonce="n-A", action_id="act-A"), _registry(), spent, NOW) + # Same nonce, brand-new action_id -> still a replay on the nonce axis. + replay = verify_and_accept(_receipt(nonce="n-A", action_id="act-B"), + _registry(), spent, NOW) + _assert_rejected(replay, "nonce", "already spent") + + def test_replay_by_action_id_is_specific(self): + spent = SpentRegistry() + verify_and_accept(_receipt(nonce="n-1", action_id="act-X"), _registry(), spent, NOW) + # Same action_id, fresh nonce -> replay on the action_id axis. + replay = verify_and_accept(_receipt(nonce="n-2", action_id="act-X"), + _registry(), spent, NOW) + _assert_rejected(replay, "action_id", "already spent") + + def test_rejected_receipt_does_not_mark_spent(self): + spent = SpentRegistry() + bad = verify_and_accept(_receipt(expires_at=ISSUED - 10), _registry(), spent, NOW) + self.assertFalse(bad.ok, "expired receipt should be rejected") + self.assertEqual(len(spent), 0, "a rejected receipt must not consume a nonce") + + def test_distinct_receipts_both_accept(self): + spent = SpentRegistry() + a = verify_and_accept(_receipt(nonce="n-1", action_id="act-1"), _registry(), spent, NOW) + b = verify_and_accept(_receipt(nonce="n-2", action_id="act-2"), _registry(), spent, NOW) + self.assertTrue(a.ok and b.ok, (a.reasons, b.reasons)) + + +# --------------------------------------------------------------------------- +# Substitution / forgery +# --------------------------------------------------------------------------- + +class SubstitutionTests(unittest.TestCase): + def test_result_substitution_fails(self): + forged = _receipt(result_sha256="ff" * 32) + result = verify_receipt(forged, _registry(), NOW, + expected_result_sha256=RESULT_SHA) + _assert_rejected(result, "substitution") + + def test_target_substitution_fails(self): + forged = _receipt(target="stackchain/other-repo#999") + result = verify_receipt(forged, _registry(), NOW, expected_target=TARGET) + _assert_rejected(result, "target") + + def test_request_hash_not_sha256_fails(self): + forged = _receipt(request_sha256="not-a-hash") + result = verify_receipt(forged, _registry(), NOW) + _assert_rejected(result, "request_sha256") + + def test_missing_signature_fails(self): + forged = _receipt(sig="") + result = verify_receipt(forged, _registry(), NOW) + _assert_rejected(result, "signature") + + def test_tampered_model_invocation_shape_still_bound(self): + # A receipt is a commitment: changing the bound invocation yields a + # different receipt the signer never attested. We can't cryptographically + # re-derive the sig in this dependency-free gate, but the contract + # requires a well-formed invocation to be present and non-empty. + forged = _receipt(model_invocation="") + result = verify_receipt(forged, _registry(), NOW) + _assert_rejected(result, "invocation") + + +class SignerTests(unittest.TestCase): + def test_unknown_signer_fails(self): + forged = _receipt(issuer="recorder:rogue", key_id="recorder:rogue/key-1") + result = verify_receipt(forged, _registry(), NOW) + _assert_rejected(result, "unknown signer") + + def test_revoked_signer_fails(self): + forged = _receipt(issuer="recorder:old", key_id="recorder:old/key-1") + result = verify_receipt(forged, _registry(), NOW) + _assert_rejected(result, "revoked") + + def test_known_signer_passes(self): + result = verify_receipt(_receipt(), _registry(), NOW) + self.assertTrue(result.ok, result.reasons) + + def test_key_out_of_validity_fails(self): + reg = _registry() + reg.add_hex("recorder:forge", "recorder:forge/key-expired", + "dd" * 32, valid_until=NOW - 1) + forged = _receipt(key_id="recorder:forge/key-expired") + result = verify_receipt(forged, reg, NOW) + _assert_rejected(result, "out of validity") + + +# --------------------------------------------------------------------------- +# Freshness / expiry +# --------------------------------------------------------------------------- + +class FreshnessTests(unittest.TestCase): + def test_expired_receipt_fails(self): + expired = _receipt(expires_at=NOW - 10) + result = verify_receipt(expired, _registry(), NOW) + _assert_rejected(result, "expired") + + def test_not_yet_valid_fails(self): + future = _receipt(issued_at=NOW + 100) + result = verify_receipt(future, _registry(), NOW) + _assert_rejected(result, "not yet valid") + + def test_invalid_window_fails(self): + bad = _receipt(issued_at=NOW + 10, expires_at=NOW + 1) + result = verify_receipt(bad, _registry(), NOW) + _assert_rejected(result, "validity window invalid") + + def test_exactly_at_expiry_is_still_valid(self): + edge = _receipt(expires_at=NOW) + result = verify_receipt(edge, _registry(), NOW) + self.assertTrue(result.ok, result.reasons) + + +# --------------------------------------------------------------------------- +# Malformed structures (fail closed, no exceptions) +# --------------------------------------------------------------------------- + +class MalformedTests(unittest.TestCase): + def _must_not_raise(self, payload): + # Returns the result; if this line raises, the test fails loudly. + return verify_receipt(payload, _registry(), NOW) + + def test_non_object_receipt_fails_closed(self): + result = self._must_not_raise([1, 2, 3]) + self.assertFalse(result.ok) + self.assertEqual(result.status, "malformed") + _assert_rejected(result, "must be a JSON object") + + def test_wrong_version_fails_closed(self): + payload = _receipt() + payload["v"] = 1 + result = self._must_not_raise(payload) + self.assertFalse(result.ok) + self.assertEqual(result.status, "malformed") + _assert_rejected(result, "version") + + def test_missing_section_fails_closed(self): + payload = _receipt() + del payload["evidence"] + result = self._must_not_raise(payload) + self.assertFalse(result.ok) + self.assertEqual(result.status, "malformed") + _assert_rejected(result, "evidence") + + def test_none_payload_fails_closed(self): + result = self._must_not_raise(None) + self.assertFalse(result.ok) + self.assertEqual(result.status, "malformed") + + def test_non_int_issued_at_fails_closed(self): + result = self._must_not_raise(_receipt(issued_at="yesterday")) + self.assertFalse(result.ok) + _assert_rejected(result, "issued_at") + + def test_bool_epoch_rejected(self): + # bool is a subclass of int; must not be accepted as an epoch value. + result = self._must_not_raise(_receipt(issued_at=True)) + self.assertFalse(result.ok) + _assert_rejected(result, "issued_at") + + def test_accept_path_also_fails_closed(self): + spent = SpentRegistry() + result = verify_and_accept("garbage", _registry(), spent, NOW) + self.assertFalse(result.ok) + self.assertEqual(len(spent), 0) + + +# --------------------------------------------------------------------------- +# Determinism +# --------------------------------------------------------------------------- + +class DeterminismTests(unittest.TestCase): + def test_same_receipt_same_result_across_runs(self): + reg = _registry() + r1 = verify_receipt(_receipt(), reg, NOW) + r2 = verify_receipt(_receipt(), reg, NOW) + self.assertEqual(r1.status, r2.status) + self.assertEqual(r1.reasons, r2.reasons) + + def test_time_is_injected_not_read(self): + # A receipt that is valid at NOW but expired at NOW+2h must flip + # purely because we pass a different `now`. + r_early = verify_receipt(_receipt(), _registry(), NOW) + r_late = verify_receipt(_receipt(), _registry(), NOW + 7200) + self.assertTrue(r_early.ok) + self.assertFalse(r_late.ok) + self.assertTrue(any("expired" in x for x in r_late.reasons)) + + +if __name__ == "__main__": + unittest.main()