Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 vincent
|
f61ef22f48 |
No files matched your search
@@ -0,0 +1,4 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.pyo
|
||||
.pytest_cache/
|
||||
@@ -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).
|
||||
@@ -1,58 +0,0 @@
|
||||
# Season Zero — work-derived X recut
|
||||
|
||||
A deterministic 30-second vertical teaser replacing the rejected abstract visualizer with a legible evidence chain:
|
||||
|
||||
1. the live Stackchain Lab issue queue rendered as a work board, plus Season Zero epic #7;
|
||||
2. Vincent's issue #1 contribution receipt and independent `8 TESTS PASS` verification;
|
||||
3. the verified 2026-08-07 Daily Drop artwork and Slop Cannon scene 04 as transformation inputs;
|
||||
4. the four Episode 01 mutation cards extracted from the verified pilot render;
|
||||
5. the audience gate: **WHICH TIMMY GETS TO REMEMBER THIS?**
|
||||
|
||||
The persistent pipeline rail, moving scan, board drift, source pan/zoom, transfer beam, card reveals, and orbiting final forms make one continuous transformation rather than a static screenshot slideshow.
|
||||
|
||||
## Artifacts
|
||||
|
||||
- [`stackchain-season-zero-work-recut.mp4`](stackchain-season-zero-work-recut.mp4) — delivery master
|
||||
- [`contact-sheet.jpg`](contact-sheet.jpg) — six-frame review strip
|
||||
- [`stills/`](stills/) — five representative full-size review frames
|
||||
- [`build_work_recut.py`](build_work_recut.py) — deterministic renderer and original synthesized soundtrack
|
||||
- [`source/`](source/) — checksummed work-derived image inputs
|
||||
|
||||
## Revised X copy — 216 characters
|
||||
|
||||
> Real work enters the Slop Cannon: a live Timmy/Vincent queue, contribution receipts, 8/8 verification, Daily Drop art, and four mutations.
|
||||
>
|
||||
> Wizard. Machine. Creature. Talking Turd.
|
||||
>
|
||||
> Which Timmy gets to remember this?
|
||||
|
||||
**Do not publish without approval in human gate #15.**
|
||||
|
||||
## Verification
|
||||
|
||||
```text
|
||||
container: MP4
|
||||
video: H.264, 720x1280 (9:16), yuv420p, 30 fps
|
||||
audio: AAC, 48 kHz, mono
|
||||
duration: 30.000000 seconds
|
||||
size: 5,325,465 bytes
|
||||
full decode: pass
|
||||
encoded audio: -20.1 dB mean, -2.2 dB max
|
||||
sha256: 2436d193e5d7ee0660318f585d5b84356cbf623a034d63355ed9232258c1f08d
|
||||
```
|
||||
|
||||
Re-run:
|
||||
|
||||
```bash
|
||||
python3 media/x-teaser/build_work_recut.py
|
||||
ffprobe -v error -show_entries format=duration,size:stream=codec_name,codec_type,width,height,pix_fmt,r_frame_rate,sample_rate,channels -of json media/x-teaser/stackchain-season-zero-work-recut.mp4
|
||||
ffmpeg -v error -i media/x-teaser/stackchain-season-zero-work-recut.mp4 -f null -
|
||||
sha256sum media/x-teaser/stackchain-season-zero-work-recut.mp4
|
||||
```
|
||||
|
||||
## Input provenance
|
||||
|
||||
- `daily-drop-door-phone.png` is the verified Daily Drop preview at `/root/daily-drop/drops/2026-08-07/the-door-that-refused-a-master-key-phone.png`; SHA-256 `9a72358821414328de129a96461b1f4265be356987f4009037e5ae2f0cab138a`.
|
||||
- `cannon-scene-04.png` is scene 04 from `/root/tiktok-slop-cannon/drops/2026-08-07-1786114450/`; SHA-256 `3080c4135c1595251097387b1e9b8a6022688c3f3ed3db4a919ed9e347d51a15`, matching its source manifest.
|
||||
- The four mutation images were extracted at 5s, 9s, 13s, and 17s from `episode01-four-timmy-mutations.mp4` on branch `timmy/8-four-timmy-mutations`, commit `afa304dc6b5d9d3cc0ede17d6f4f0dfb67fc6b70`; source MP4 SHA-256 `5ddc9a9da1b4b2456faf0b390e5c1be0d3cf48112b417fc075473c90e335d1e4`.
|
||||
- Board, epic, receipt, and test copy are drawn from Gitea issues #7 and #1 and their complete comments read before rendering. The work-board composition is an editorial rendering of that live queue state, not a claim that Gitea supplied a screenshot endpoint.
|
||||
@@ -1,286 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Deterministically build the work-derived Season Zero X teaser."""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import subprocess
|
||||
import wave
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, ImageDraw, ImageEnhance, ImageFilter, ImageFont
|
||||
|
||||
W, H, FPS, RATE, DURATION = 720, 1280, 30, 48_000, 30.0
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
SOURCE = ROOT / "source"
|
||||
OUT = ROOT / "stackchain-season-zero-work-recut.mp4"
|
||||
TMP_VIDEO = ROOT / ".video.mp4"
|
||||
TMP_AUDIO = ROOT / ".audio.wav"
|
||||
BOLD = "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"
|
||||
REG = "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"
|
||||
COLORS = {"ink": "#070b14", "panel": "#111827", "line": "#334155", "white": "#f8fafc", "muted": "#a9b7cc", "gold": "#ffc857", "cyan": "#5ee7f2", "violet": "#c69cff", "green": "#8ff0a4", "orange": "#ff9c66"}
|
||||
ASSETS = {name: Image.open(SOURCE / filename).convert("RGB") for name, filename in {
|
||||
"drop": "daily-drop-door-phone.png", "cannon": "cannon-scene-04.png",
|
||||
"wizard": "wizard.png", "machine": "machine.png", "creature": "creature.png", "turd": "talking-turd.png",
|
||||
}.items()}
|
||||
|
||||
|
||||
def font(size: int, bold: bool = False):
|
||||
return ImageFont.truetype(BOLD if bold else REG, size)
|
||||
|
||||
|
||||
def ease(x: float) -> float:
|
||||
x = max(0.0, min(1.0, x))
|
||||
return x * x * (3 - 2 * x)
|
||||
|
||||
|
||||
def fit_crop(image: Image.Image, box: tuple[int, int, int, int], zoom: float = 1.0, pan_y: float = 0.5) -> Image.Image:
|
||||
x0, y0, x1, y1 = box
|
||||
bw, bh = x1-x0, y1-y0
|
||||
scale = max(bw/image.width, bh/image.height) * zoom
|
||||
rw, rh = max(1, int(image.width*scale)), max(1, int(image.height*scale))
|
||||
resized = image.resize((rw, rh), Image.Resampling.LANCZOS)
|
||||
left = max(0, (rw-bw)//2)
|
||||
top = max(0, min(rh-bh, int((rh-bh)*pan_y)))
|
||||
return resized.crop((left, top, left+bw, top+bh))
|
||||
|
||||
|
||||
def rounded_image(canvas: Image.Image, image: Image.Image, box: tuple[int, int, int, int], radius: int = 24, zoom: float = 1.0, pan_y: float = 0.5, border: str = COLORS["line"]):
|
||||
crop = fit_crop(image, box, zoom, pan_y)
|
||||
mask = Image.new("L", crop.size, 0)
|
||||
ImageDraw.Draw(mask).rounded_rectangle((0, 0, crop.width-1, crop.height-1), radius, fill=255)
|
||||
canvas.paste(crop, (box[0], box[1]), mask)
|
||||
d = ImageDraw.Draw(canvas)
|
||||
d.rounded_rectangle(box, radius, outline=border, width=3)
|
||||
|
||||
|
||||
def text(d: ImageDraw.ImageDraw, xy, value: str, size: int, fill: str = COLORS["white"], bold: bool = False, anchor: str | None = None):
|
||||
d.text(xy, value, font=font(size, bold), fill=fill, anchor=anchor)
|
||||
|
||||
|
||||
def centered(d, value: str, y: int, size: int, fill: str = COLORS["white"], bold: bool = False):
|
||||
text(d, (W//2, y), value, size, fill, bold, "ma")
|
||||
|
||||
|
||||
def base(t: float) -> Image.Image:
|
||||
y = np.linspace(0, 1, H)[:, None, None]
|
||||
top = np.array((5, 10, 20))[None, None, :]
|
||||
bottom = np.array((22, 15, 40))[None, None, :]
|
||||
grad = top*(1-y) + bottom*y
|
||||
xwave = 5*np.sin(np.linspace(0, 9, W)[None, :, None] + t*.6)
|
||||
arr = np.broadcast_to(grad, (H, W, 3)) + xwave
|
||||
im = Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8), "RGB")
|
||||
d = ImageDraw.Draw(im)
|
||||
for i in range(7):
|
||||
yy = int((t*52 + i*195) % (H+160))-80
|
||||
d.line((0, yy, W, yy-120), fill=(25, 42, 65), width=2)
|
||||
return im
|
||||
|
||||
|
||||
def chrome(d: ImageDraw.ImageDraw, t: float, active: int):
|
||||
d.rectangle((0, 0, W, 76), fill="#080d18")
|
||||
d.line((0, 76, W, 76), fill=COLORS["line"], width=2)
|
||||
text(d, (32, 23), "STACKCHAIN LAB", 25, COLORS["white"], True)
|
||||
text(d, (W-32, 25), "SEASON ZERO", 20, COLORS["gold"], True, "ra")
|
||||
nodes = [(64, "WORK"), (215, "CHALLENGE"), (390, "MUTATE"), (563, "AUDIENCE")]
|
||||
y = 1208
|
||||
d.line((64, y, 656, y), fill=COLORS["line"], width=5)
|
||||
progress = min(1, max(0, t/DURATION))
|
||||
d.line((64, y, 64+592*progress, y), fill=COLORS["cyan"], width=5)
|
||||
for i, (x, label) in enumerate(nodes):
|
||||
c = COLORS["gold"] if i == active else (COLORS["cyan"] if i < active else COLORS["line"])
|
||||
d.ellipse((x-9, y-9, x+9, y+9), fill=c)
|
||||
text(d, (x, y+27), label, 15, c, True, "ma")
|
||||
scan = int((t*155) % (H-160)) + 80
|
||||
d.rectangle((0, scan, W, scan+2), fill="#5ee7f222")
|
||||
|
||||
|
||||
def issue_card(d, box, number, title, owner, state, accent):
|
||||
d.rounded_rectangle(box, 14, fill=COLORS["panel"], outline=accent, width=2)
|
||||
x0, y0, x1, _ = box
|
||||
text(d, (x0+15, y0+13), f"#{number}", 19, accent, True)
|
||||
text(d, (x1-15, y0+15), state.upper(), 16, COLORS["muted"], True, "ra")
|
||||
words = title.split()
|
||||
lines, line = [], ""
|
||||
for word in words:
|
||||
trial = (line+" "+word).strip()
|
||||
if d.textlength(trial, font=font(19, True)) > x1-x0-30:
|
||||
lines.append(line); line = word
|
||||
else: line = trial
|
||||
lines.append(line)
|
||||
for i, ln in enumerate(lines[:2]): text(d, (x0+15, y0+46+i*25), ln, 19, COLORS["white"], True)
|
||||
text(d, (x0+15, y0+104), f"agent:{owner}", 16, accent)
|
||||
|
||||
|
||||
def scene_board(im: Image.Image, t: float):
|
||||
d = ImageDraw.Draw(im)
|
||||
local = t
|
||||
centered(d, "THE WORK IS PUBLIC", 112, 35, COLORS["white"], True)
|
||||
centered(d, "A LIVE QUEUE. A REAL HANDOFF.", 158, 21, COLORS["cyan"], True)
|
||||
sway = int(7*math.sin(local*1.8))
|
||||
d.rounded_rectangle((28+sway, 205, 692+sway, 1015), 24, fill="#0b1220", outline=COLORS["line"], width=3)
|
||||
text(d, (52+sway, 228), "stackchain / stackchain-lab-loop", 20, COLORS["muted"], True)
|
||||
for x, label, c in [(50, "READY", COLORS["cyan"]), (274, "CLAIMED", COLORS["gold"]), (498, "REVIEW", COLORS["green"])]:
|
||||
text(d, (x+sway, 281), label, 17, c, True)
|
||||
issue_card(d, (48+sway, 320, 254+sway, 455), 7, "The Cannon Needs a Pilot", "vincent", "ready", COLORS["violet"])
|
||||
issue_card(d, (48+sway, 475, 254+sway, 610), 13, "Attack the Cannon Chain", "vincent", "ready", COLORS["cyan"])
|
||||
issue_card(d, (274+sway, 320, 480+sway, 455), 16, "Build the X teaser", "timmy", "claimed", COLORS["gold"])
|
||||
issue_card(d, (498+sway, 320, 672+sway, 455), 1, "Prove the Lab Loop", "vincent", "review", COLORS["green"])
|
||||
d.rounded_rectangle((54+sway, 670, 666+sway, 962), 18, fill="#16112a", outline=COLORS["violet"], width=3)
|
||||
text(d, (78+sway, 694), "EPIC #7 // SEASON ZERO", 21, COLORS["violet"], True)
|
||||
text(d, (78+sway, 744), "THE CANNON NEEDS A PILOT", 28, COLORS["white"], True)
|
||||
text(d, (78+sway, 801), "6-episode character evolution chain", 19, COLORS["muted"])
|
||||
for i, line in enumerate(("✓ persistent Timmy", "✓ power / scar / cost / receipt", "✓ audience choice advances canon")):
|
||||
text(d, (90+sway, 846+i*33), line, 19, COLORS["green"] if i < 2 else COLORS["gold"], True)
|
||||
text(d, (50, 1060), "TIMMY → VINCENT → HUMAN REVIEW", 24, COLORS["gold"], True)
|
||||
|
||||
|
||||
def scene_receipt(im: Image.Image, t: float):
|
||||
d = ImageDraw.Draw(im)
|
||||
local = t-5
|
||||
centered(d, "CHALLENGED. THEN VERIFIED.", 116, 31, COLORS["white"], True)
|
||||
x = 38 + int(8*math.sin(local*2))
|
||||
d.rounded_rectangle((x, 205, W-38+x-38, 945), 22, fill="#0c1423", outline=COLORS["cyan"], width=3)
|
||||
text(d, (x+24, 230), "ISSUE #1 / CONTRIBUTION RECEIPT", 19, COLORS["cyan"], True)
|
||||
d.rounded_rectangle((x+22, 278, W-62, 520), 18, fill=COLORS["panel"], outline=COLORS["line"], width=2)
|
||||
text(d, (x+44, 302), "VINCENT", 23, COLORS["violet"], True)
|
||||
text(d, (W-84, 307), "INDEPENDENT PASS", 16, COLORS["muted"], True, "ra")
|
||||
text(d, (x+44, 355), "Fetched PR #4 from Timmy's branch", 20, COLORS["white"])
|
||||
text(d, (x+44, 397), "8 TESTS PASS", 39, COLORS["green"], True)
|
||||
text(d, (x+44, 452), "Label transition regression verified", 20, COLORS["muted"])
|
||||
d.rounded_rectangle((x+22, 555, W-62, 804), 18, fill="#17122b", outline=COLORS["violet"], width=2)
|
||||
text(d, (x+44, 580), "DEFINITION OF DONE", 19, COLORS["violet"], True)
|
||||
checks = ("Both agents read assigned tickets", "Both leave verified evidence", "Timmy → Vincent → review", "Credentials remain host-local")
|
||||
for i, line in enumerate(checks):
|
||||
text(d, (x+48, 633+i*40), "✓", 22, COLORS["green"], True)
|
||||
text(d, (x+82, 635+i*40), line, 19, COLORS["white"])
|
||||
text(d, (x+24, 856), "RECEIPT: ISSUE #1 • PR #4 • 8/8", 21, COLORS["gold"], True)
|
||||
centered(d, "CRITIQUE BECOMES A BUILD INPUT.", 1012, 23, COLORS["gold"], True)
|
||||
centered(d, "NOT A COMMENT. A CONSEQUENCE.", 1055, 19, COLORS["muted"], True)
|
||||
|
||||
|
||||
def scene_sources(im: Image.Image, t: float):
|
||||
d = ImageDraw.Draw(im)
|
||||
local = t-10
|
||||
centered(d, "REAL ARTIFACTS ENTER THE CHAIN", 112, 29, COLORS["white"], True)
|
||||
shift = int(10*math.sin(local*.9))
|
||||
rounded_image(im, ASSETS["drop"], (35+shift, 205, 350+shift, 930), 22, 1.03+local*.008, .46, COLORS["gold"])
|
||||
rounded_image(im, ASSETS["cannon"], (370-shift, 205, 685-shift, 930), 22, 1.05+local*.01, .5, COLORS["orange"])
|
||||
d = ImageDraw.Draw(im)
|
||||
d.rounded_rectangle((48+shift, 835, 337+shift, 913), 12, fill="#070b14dd")
|
||||
text(d, (65+shift, 852), "DAILY DROP", 19, COLORS["gold"], True)
|
||||
text(d, (65+shift, 881), "THE DOOR / VERIFIED", 16, COLORS["white"], True)
|
||||
d.rounded_rectangle((383-shift, 835, 672-shift, 913), 12, fill="#070b14dd")
|
||||
text(d, (400-shift, 852), "CANNON OUTPUT", 19, COLORS["orange"], True)
|
||||
text(d, (400-shift, 881), "SCENE 04 / SOURCE", 16, COLORS["white"], True)
|
||||
# Animated transfer beam unifies the two work artifacts.
|
||||
beam = int(95 + (local % 2.0)/2.0*520)
|
||||
d.line((beam, 965, beam+70, 965), fill=COLORS["cyan"], width=7)
|
||||
d.polygon([(beam+70, 953), (beam+94, 965), (beam+70, 977)], fill=COLORS["cyan"])
|
||||
centered(d, "SOURCE → PRESSURE → MUTATION", 1025, 23, COLORS["cyan"], True)
|
||||
centered(d, "THE ART CHANGES JOBS. IT ISN'T FILLER.", 1070, 18, COLORS["muted"], True)
|
||||
|
||||
|
||||
def scene_mutations(im: Image.Image, t: float):
|
||||
d = ImageDraw.Draw(im)
|
||||
local = t-16
|
||||
centered(d, "ONE PILOT. FOUR CONSEQUENCES.", 108, 29, COLORS["white"], True)
|
||||
specs = [
|
||||
("wizard", "WIZARD", "AMBIGUITY → SPELL", COLORS["violet"]),
|
||||
("machine", "MACHINE", "REPEATABILITY → EXACT", COLORS["cyan"]),
|
||||
("creature", "CREATURE", "HUNGER → SIGNAL", COLORS["green"]),
|
||||
("turd", "TALKING TURD", "RIDICULE → PUNCHLINE", "#ffd18a"),
|
||||
]
|
||||
boxes = [(35, 190, 350, 610), (370, 190, 685, 610), (35, 650, 350, 1070), (370, 650, 685, 1070)]
|
||||
for i, ((key, name, pressure, color), box) in enumerate(zip(specs, boxes)):
|
||||
reveal = ease((local-i*.55)/1.3)
|
||||
cx = (box[0]+box[2])//2
|
||||
cy = (box[1]+box[3])//2
|
||||
bw, bh = box[2]-box[0], box[3]-box[1]
|
||||
sw, sh = max(8, int(bw*reveal)), max(8, int(bh*reveal))
|
||||
live = (cx-sw//2, cy-sh//2, cx+sw//2, cy+sh//2)
|
||||
rounded_image(im, ASSETS[key], live, 20, 1.0+.025*math.sin(local*1.7+i), .45, color)
|
||||
d = ImageDraw.Draw(im)
|
||||
if reveal > .78:
|
||||
alpha_box = (box[0]+8, box[3]-94, box[2]-8, box[3]-10)
|
||||
d.rounded_rectangle(alpha_box, 12, fill="#070b14e8")
|
||||
text(d, (box[0]+20, box[3]-82), name, 20 if name != "TALKING TURD" else 17, color, True)
|
||||
text(d, (box[0]+20, box[3]-48), pressure, 16, COLORS["white"], True)
|
||||
centered(d, "PLANNED CARDS • EPISODE 01", 1120, 18, COLORS["gold"], True)
|
||||
|
||||
|
||||
def scene_gate(im: Image.Image, t: float):
|
||||
d = ImageDraw.Draw(im)
|
||||
local = t-25
|
||||
for i, color in enumerate((COLORS["violet"], COLORS["cyan"], COLORS["green"], "#ffd18a")):
|
||||
a = local*.8+i*math.pi/2
|
||||
x, y = W/2+math.cos(a)*245, 390+math.sin(a)*130
|
||||
r = 50+8*math.sin(local*3+i)
|
||||
d.ellipse((x-r, y-r, x+r, y+r), fill=color, outline=COLORS["white"], width=3)
|
||||
centered(d, "THE AUDIENCE ALTERS CANON", 125, 25, COLORS["gold"], True)
|
||||
centered(d, "WHICH TIMMY", 590, 57, COLORS["white"], True)
|
||||
centered(d, "GETS TO REMEMBER", 675, 49, COLORS["white"], True)
|
||||
centered(d, "THIS?", 765, 76, COLORS["gold"], True)
|
||||
d.rounded_rectangle((55, 892, W-55, 1000), 18, outline=COLORS["cyan"], width=3)
|
||||
centered(d, "WIZARD / MACHINE", 917, 24, COLORS["white"], True)
|
||||
centered(d, "CREATURE / TALKING TURD", 959, 21, COLORS["white"], True)
|
||||
centered(d, "NO POST UNTIL HUMAN APPROVAL", 1072, 19, COLORS["muted"], True)
|
||||
|
||||
|
||||
def frame(t: float) -> Image.Image:
|
||||
im = base(t)
|
||||
if t < 5: scene_board(im, t); active = 0
|
||||
elif t < 10: scene_receipt(im, t); active = 1
|
||||
elif t < 16: scene_sources(im, t); active = 1
|
||||
elif t < 25: scene_mutations(im, t); active = 2
|
||||
else: scene_gate(im, t); active = 3
|
||||
chrome(ImageDraw.Draw(im), t, active)
|
||||
# Soft entrance/exit only; scene motion remains continuous within each stage.
|
||||
if t < .35:
|
||||
im = ImageEnhance.Brightness(im).enhance(ease(t/.35))
|
||||
if t > DURATION-.5:
|
||||
im = ImageEnhance.Brightness(im).enhance(max(0, (DURATION-t)/.5))
|
||||
return im
|
||||
|
||||
|
||||
def build_audio():
|
||||
n = int(DURATION*RATE)
|
||||
audio = np.zeros(n, dtype=np.float64)
|
||||
rng = np.random.default_rng(1608072026)
|
||||
for beat in np.arange(0, DURATION, .5):
|
||||
i = int(beat*RATE); length = min(int(.16*RATE), n-i); x = np.arange(length)/RATE
|
||||
audio[i:i+length] += .42*np.sin(2*np.pi*(74*x-28*x*x))*np.exp(-x*25)
|
||||
for beat in np.arange(.5, DURATION, 1):
|
||||
i = int(beat*RATE); length = min(int(.12*RATE), n-i); x = np.arange(length)/RATE
|
||||
audio[i:i+length] += .12*rng.normal(0, 1, length)*np.exp(-x*34)
|
||||
for beat in np.arange(.25, DURATION, .5):
|
||||
i = int(beat*RATE); length = min(int(.035*RATE), n-i); x = np.arange(length)/RATE
|
||||
audio[i:i+length] += .04*rng.normal(0, 1, length)*np.exp(-x*80)
|
||||
stages = [(0, 5, (110, 165, 220)), (5, 10, (123, 185, 247)), (10, 16, (98, 147, 196)), (16, 25, (131, 196, 262)), (25, 30, (110, 220, 330))]
|
||||
for start, end, chord in stages:
|
||||
i0, i1 = int(start*RATE), int(end*RATE); x = np.arange(i1-i0)/RATE
|
||||
env = np.minimum(1, x*2)*np.minimum(1, (end-start-x)*2)
|
||||
for f in chord: audio[i0:i1] += .035*np.sin(2*np.pi*f*x)*env
|
||||
audio = np.tanh(audio*1.4)
|
||||
pcm = (audio*32767).astype("<i2")
|
||||
with wave.open(str(TMP_AUDIO), "wb") as w:
|
||||
w.setnchannels(1); w.setsampwidth(2); w.setframerate(RATE); w.writeframes(pcm.tobytes())
|
||||
|
||||
|
||||
def main():
|
||||
cmd = ["ffmpeg", "-y", "-loglevel", "error", "-f", "rawvideo", "-pix_fmt", "rgb24", "-s", f"{W}x{H}", "-r", str(FPS), "-i", "-", "-an", "-c:v", "libx264", "-preset", "medium", "-crf", "20", "-pix_fmt", "yuv420p", "-movflags", "+faststart", str(TMP_VIDEO)]
|
||||
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE)
|
||||
assert proc.stdin
|
||||
try:
|
||||
for i in range(int(DURATION*FPS)): proc.stdin.write(frame(i/FPS).tobytes())
|
||||
finally: proc.stdin.close()
|
||||
if proc.wait(): raise SystemExit("video encode failed")
|
||||
build_audio()
|
||||
subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", str(TMP_VIDEO), "-i", str(TMP_AUDIO), "-c:v", "copy", "-c:a", "aac", "-b:a", "160k", "-shortest", "-movflags", "+faststart", str(OUT)], check=True)
|
||||
TMP_VIDEO.unlink(); TMP_AUDIO.unlink()
|
||||
for sec, name in [(2, "01-board"), (7, "02-vincent-receipt"), (13, "03-source-artifacts"), (21, "04-mutations"), (27, "05-audience-gate")]:
|
||||
frame(sec).save(ROOT / "stills" / f"{name}.jpg", quality=88)
|
||||
print(OUT)
|
||||
|
||||
|
||||
if __name__ == "__main__": main()
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 64 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 73 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 205 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 188 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 150 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 174 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 183 KiB |
Binary file not shown.
Binary file not shown.
|
Before Width: | Height: | Size: 133 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 119 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 201 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 140 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 104 KiB |
@@ -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",
|
||||
]
|
||||
@@ -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()
|
||||
@@ -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, [])
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user