stackchain/stackchain-lab-loop
3
0
Fork 0
Code Issues 182 Pull Requests 7 Actions Packages Projects 1 Releases Wiki Activity

Compare commits

base: stackchain:vincent/path-proof-replay-gate
Branches Tags
stackchain:main
stackchain:timmy/42-companion-fixes
stackchain:vincent/path-proof-replay-gate
stackchain:vincent/19-canonical-receipts-r2
stackchain:vincent/19-daily-canon-receipts
stackchain:vincent/13-cannon-chain-review
stackchain:timmy/16-work-derived-teaser
stackchain:timmy/8-four-timmy-mutations
stackchain:timmy/5-cannon-chain
stackchain:timmy/3-path-proof-receipt
stackchain:timmy/1-fix-label-transitions
..
compare: stackchain:main
Branches Tags
stackchain:timmy/42-companion-fixes
stackchain:vincent/path-proof-replay-gate
stackchain:vincent/19-canonical-receipts-r2
stackchain:vincent/19-daily-canon-receipts
stackchain:vincent/13-cannon-chain-review
stackchain:timmy/16-work-derived-teaser
stackchain:timmy/8-four-timmy-mutations
stackchain:timmy/5-cannon-chain
stackchain:timmy/3-path-proof-receipt
stackchain:timmy/1-fix-label-transitions
stackchain:main

No commits in common. "vincent/path-proof-replay-gate" and "main" have entirely different histories.
vincent/pa ... main

6 changed files with 0 additions and 932 deletions
Show Stats Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,4 +0,0 @@
__pycache__/
*.pyc
*.pyo
.pytest_cache/

66
docs/path-proof-threat-boundary.md
View File

@ -1,66 +0,0 @@
# 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).

52
path_proof/__init__.py
View File

@ -1,52 +0,0 @@
"""
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",
]

84
path_proof/keys.py
View File

@ -1,84 +0,0 @@
"""
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()

406
path_proof/receipt.py
View File

@ -1,406 +0,0 @@
"""
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, [])

320
tests/test_path_proof.py
View File

@ -1,320 +0,0 @@
"""
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()