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

[Vincent] #28: executable replay/forgery gate for path-proof receipts

Browse Source 
Dependency-free path_proof package under a dedicated module. Bounded to this
issue only: no changes to existing lab_loop tests or scripts.

Deliverables:
1. path_proof/ package (receipt.py, keys.py, __init__.py) — verifier with
   canonical JSON, deterministic build/verify/verify_and_accept, in-memory
   KeyRegistry + SpentRegistry. No I/O, no wall clock, no randomness.
2. Each receipt binds action kind/target/request/result, model id+invocation,
   tool trace, policy bundle, code identity, evidence uri, plus action_id,
   verifier nonce, issued_at/expires_at, and signer.
3. Deterministic spent-nonce/idempotency registry: the same valid receipt is
   accepted exactly once; a replay is rejected on the nonce OR action_id axis.
4. Positive + negative tests: valid passes; replay fails; request/result
   substitution fails; wrong target fails; expired fails; unknown/revoked
   signer fails; malformed structures fail closed with no exception. Every
   negative case asserts rejection AND a specific reason.
5. docs/path-proof-threat-boundary.md documents the exact threat boundary: a
   verified receipt proves a bound, fresh, single-use, trusted-attested
   commitment — NOT runtime execution unless runtime evidence is supplied.

Clean-checkout verification (commands actually run):
- /home/vincent/seedvault-inventory/venv/bin/python3 -m pytest tests/ -q
  -> 36 passed (31 path_proof + 5 pre-existing lab_loop)
- git diff --cached --name-only origin/main -> 6 files, zero .pyc
- __pycache__ gitignored; clean checkout stays clean after tests

Closes #28
Refs: #3 (Vincent path-proof critique), #19/PR #42 (canon validator r2)
[HANDOFF] from=vincent to=timmy
This commit is contained in:
vincent committed 2026-08-15 16:44:58 -04:00
1 parent fdb454414d
commit f61ef22f48
6 files changed
+932
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

path_proof/__init__.py
+52
View File
@@ -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",
]
path_proof/keys.py
+84
View File
@@ -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()
path_proof/receipt.py
+406
View File
@@ -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, [])