alexmandClaude
a4eecb4d5b
The oracle had two import names, `tools.oracle` for the tools and `paperspec` for other repositories, which could load the same code twice in one process and needed a gate keyed on the module name. It now lives in the standard src layout under its one name. - git mv tools/oracle -> src/paperspec; the 22 rule modules are unchanged. - The tools import `paperspec`. `tools/__init__.py` now holds the Python 3.14 and cryptography/jsonschema check and puts src/ first on sys.path, so the tools run this checkout's code with or without pixi. The package's own __init__ checks nothing and exposes __version__. - pyproject.toml: version from git tags via setuptools-scm (guess-next-dev, node-and-timestamp, src/paperspec/_version.py, gitignored). The packages stay listed rather than found because schemas/, sql/ and vectors/ stay at the root: the spec links to them and the site serves the schemas at their $id URLs. - pixi.toml: a `dev` feature and environment with setuptools, setuptools-scm and python-build, and `pixi run -e dev wheel`. - check.py scans src/paperspec for rule IDs; README, CONTRIBUTING, the FAQ, the overview, the vectors and tools READMEs and the changelog point at the new path. The FAQ no longer says the oracle is not a library. pixi.lock is not re-solved in this commit: with a dynamic version, pixi builds the package metadata to lock, which needs the conda environment installed. Run `pixi lock` and commit the result before pushing; the site workflow installs with --locked. pixi run check passes; no computed value, vector, fixture, schema or SQL file changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
734 lines
38 KiB
Python
734 lines
38 KiB
Python
#!/usr/bin/env python3
|
||
"""Stress the reference implementation with random inputs. Exit status 0 means every property held.
|
||
|
||
Non-normative tooling; it needs Python 3.14 and the packages pixi installs (`jsonschema`, `cryptography`).
|
||
Usage: python3 tools/stress.py [--seed N] [--rounds N]
|
||
The vectors pin values; this program checks the invariants the chapters state, over inputs drawn from a seeded
|
||
random generator, so a failure is reproducible from the seed it prints. Canonical JSON round-trips and sorts keys
|
||
by UTF-16 units; normalization is idempotent; notation folding never raises; random corrections keep the five
|
||
values equal to a full recomputation, never reuse an ID and apply deterministically; anchors made before those
|
||
corrections always resolve to one of the five outcomes; Merkle roots agree with a second implementation;
|
||
spot-check samples do not depend on input order; error bounds are the least six-decimal value with the stated tail;
|
||
differently stamped pages share a fingerprint; the zip reader and the package validator refuse or accept mutated
|
||
input without raising and never write outside their folders; rights combine the same in any order; and
|
||
spot-check chains hundreds deep settle without recursion.
|
||
"""
|
||
|
||
import argparse
|
||
import copy
|
||
import json
|
||
import os
|
||
import random
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import zipfile
|
||
from fractions import Fraction
|
||
from math import comb
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, ROOT)
|
||
import tools # noqa: E402,F401 checks the runtime and puts src/ first on the path, before paperspec is imported
|
||
|
||
from paperspec import align as AL # noqa: E402
|
||
from paperspec import anchors as AN # noqa: E402
|
||
from paperspec import corrections as CO # noqa: E402
|
||
from paperspec import crypto, fingerprint as FP, hashing, jcs, receipts as RC, rights as RI, targets as TG # noqa: E402
|
||
from paperspec import keys as K, log as LG, mldsa, vrf # noqa: E402
|
||
from paperspec import package as PK # noqa: E402
|
||
from paperspec.model import LABELLED, LEAVES, Doc # noqa: E402
|
||
from paperspec.notation import fold_math # noqa: E402
|
||
from paperspec.text import NODE_ID_RE, RECORD_ID_RE, letter_first, norm, record_id, sha256_hex, is_ws # noqa: E402
|
||
from paperspec.tree import Refusal, validate_structure # noqa: E402
|
||
from tools.build import ROSTER_BY_ID, SCHEMAS, secret, sentence_doc, tiny_package # noqa: E402
|
||
from tools.check import HALE, STORE, TINY_KEY, reference_values # noqa: E402
|
||
|
||
FAILS = []
|
||
COUNTS = {}
|
||
RULES = {r["id"] for r in json.load(open(os.path.join(ROOT, "registry", "rules.json"), encoding="utf-8"))["rules"]}
|
||
|
||
|
||
def ok(section, cond, msg):
|
||
COUNTS[section] = COUNTS.get(section, 0) + 1
|
||
if not cond and len(FAILS) < 200:
|
||
FAILS.append("%s: %s" % (section, msg))
|
||
|
||
|
||
TEXT_POOL = list("abcdefghijklmnopqrstuvwxyzABCDEFXYZ0123456789 .,;:-()[]") + [
|
||
"é", "Å", "fi", "µ", "−", "–", " ", "\t", "\n", "
", "€",
|
||
"\U0001d49c", "\U0001f600", "é", "H₂O", "10³", "cm⁻¹", " "]
|
||
KEY_POOL = list("abcxyz") + ["€", "\U0001f600", "fi", "A", "é", "", "a b", "\u0001"]
|
||
|
||
|
||
def rand_text(rng, lo=0, hi=12, pool=TEXT_POOL):
|
||
return "".join(rng.choice(pool) for _ in range(rng.randint(lo, hi)))
|
||
|
||
|
||
def rand_json(rng, depth=0):
|
||
r = rng.random()
|
||
if depth > 3 or r < 0.4:
|
||
choice = rng.randint(0, 6)
|
||
if choice == 0:
|
||
return None
|
||
if choice == 1:
|
||
return rng.random() < 0.5
|
||
if choice == 2:
|
||
return rng.randint(-2**53, 2**53)
|
||
if choice == 3:
|
||
return rng.uniform(-1, 1) * 10.0 ** rng.randint(-12, 24)
|
||
if choice == 4:
|
||
return rng.choice([0, -0.0, 1e21, 1e-7, 0.000001, 123456789.125, 5e-324, 1.7976931348623157e308])
|
||
return rand_text(rng)
|
||
if r < 0.7:
|
||
return [rand_json(rng, depth + 1) for _ in range(rng.randint(0, 4))]
|
||
return {rand_text(rng, 0, 3, KEY_POOL): rand_json(rng, depth + 1) for _ in range(rng.randint(0, 4))}
|
||
|
||
|
||
def _as_doubles(text):
|
||
"""Parse JSON as an ECMAScript engine does, which is how canonical JSON is read (vectors/README.md)."""
|
||
return json.loads(text, parse_int=lambda x: int(x) if abs(int(x)) <= 2**53 else float(x))
|
||
|
||
|
||
def stress_canonical_json(rng, rounds):
|
||
for _ in range(rounds):
|
||
v = rand_json(rng)
|
||
try:
|
||
s = jcs.dumps(v)
|
||
except ValueError as e:
|
||
ok("canonical-json", False, "dumps refused a plain value: %r (%s)" % (v, e))
|
||
continue
|
||
ok("canonical-json", jcs.dumpb(v) == s.encode("utf-8"), "dumpb is dumps encoded")
|
||
back = _as_doubles(s)
|
||
ok("canonical-json", back == v, "round trip changed the value: %r" % (v,))
|
||
ok("canonical-json", jcs.dumps(back) == s, "canonical form is not idempotent for %r" % (v,))
|
||
pairs = _as_doubles(s) if not s.startswith(("{", "[")) else json.loads(s, object_pairs_hook=lambda kv: kv)
|
||
|
||
def ordered(x):
|
||
if isinstance(x, list) and x and isinstance(x[0], tuple):
|
||
keys = [k for k, _ in x]
|
||
return keys == sorted(keys, key=lambda k: k.encode("utf-16-be")) and all(ordered(v) for _, v in x)
|
||
if isinstance(x, list):
|
||
return all(ordered(y) for y in x)
|
||
return True
|
||
ok("canonical-json", ordered(pairs), "object members are not sorted by UTF-16 code units in %s" % s[:80])
|
||
|
||
|
||
def stress_normalization(rng, rounds):
|
||
for _ in range(rounds):
|
||
s = rand_text(rng, 0, 30)
|
||
n = norm(s)
|
||
ok("normalize", norm(n) == n, "normalization is not idempotent for %r" % s)
|
||
ok("normalize", not n or (not is_ws(n[0]) and not is_ws(n[-1])), "normalized text keeps edge whitespace")
|
||
ok("normalize", " " not in n and all(ch == " " or not is_ws(ch) for ch in n),
|
||
"normalized text keeps a whitespace run or a non-space whitespace: %r" % n)
|
||
|
||
|
||
TEX_TOKENS = list("xyzKrHO0123456789+-=()'") + ["_", "^", "{", "}", "\\alpha", "\\mathrm{H}", "\\frac{a}{b}", " ",
|
||
"\\,", "\\cdot", "H_2O", "^{14}", "{}", "\\left(", "\\right)", "\\\\",
|
||
"µ", "−", "''", "_{", "^{"]
|
||
TYPST_TOKENS = list("xyzKr0123456789+-=()'") + ["_", "^", "(", ")", "alpha", "upright(H)", "\"cm\"", " ", "dot",
|
||
"plus.minus", "<=", "/", "epsilon.alt", "_(", "^(", "\"a\\b\""]
|
||
|
||
|
||
def stress_notation(rng, rounds):
|
||
for _ in range(rounds):
|
||
for notation, pool in (("tex", TEX_TOKENS), ("typst", TYPST_TOKENS)):
|
||
s = "".join(rng.choice(pool) for _ in range(rng.randint(0, 8)))
|
||
try:
|
||
a = fold_math(notation, s)
|
||
b = fold_math(notation, s)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("notation", False, "fold_math raised %r on %s %r" % (e, notation, s))
|
||
continue
|
||
ok("notation", a == b and (a is None or isinstance(a, str)), "fold_math is not deterministic on %r" % s)
|
||
ok("notation", a is None or not any(is_ws(ch) for ch in a), "a canonical form holds whitespace: %r" % a)
|
||
ok("notation", a is None or "$" not in a, "a canonical form holds a dollar sign: %r" % a)
|
||
|
||
|
||
def _leaf_ids(doc, kinds):
|
||
return [nid for nid in doc.preorder() if doc.nodes[nid]["kind"] in kinds]
|
||
|
||
|
||
def random_correction(rng, st, doc):
|
||
"""One random proposal against `st`, or None."""
|
||
sentences = _leaf_ids(doc, {"sentence"})
|
||
cells = _leaf_ids(doc, {"cell"})
|
||
op = rng.choice(["replace", "replace", "split", "merge", "move", "relabel", "set", "delete", "reclassify",
|
||
"insert-text", "insert-node"])
|
||
evidence = [{"kind": "crop", "sha256": "0" * 64}] if rng.random() < 0.8 else [{"kind": "person", "by": "ana"}]
|
||
c = {"op": op.split("-")[0], "evidence": evidence}
|
||
if op == "replace" and (sentences or cells):
|
||
nid = rng.choice(sentences + cells)
|
||
text = hashing.plain(doc, nid)
|
||
if not text:
|
||
return None
|
||
i = rng.randrange(len(text))
|
||
j = min(len(text), i + rng.randint(1, 6))
|
||
c.update(target=nid, find=text[i:j], replace=rand_text(rng, 0, 8))
|
||
elif op == "split" and sentences:
|
||
nid = rng.choice(sentences)
|
||
c.update(target=nid, offset=rng.randint(0, max(1, len(hashing.plain(doc, nid)))))
|
||
elif op == "merge" and len(sentences) > 1:
|
||
nid = rng.choice(sentences)
|
||
sibs = doc.kids(doc.nodes[nid]["parent"])
|
||
k = sibs.index(nid)
|
||
other = sibs[k + 1] if k + 1 < len(sibs) else rng.choice(sentences)
|
||
c.update(target=nid, **{"with": other}, joiner=rng.choice([" ", ""]))
|
||
elif op == "move" and sentences:
|
||
paragraphs = _leaf_ids(doc, {"paragraph"})
|
||
if not paragraphs:
|
||
return None
|
||
c.update(target=rng.choice(sentences), parent=rng.choice(paragraphs), place_after=None)
|
||
elif op == "relabel":
|
||
labelled = [nid for nid in doc.preorder() if doc.nodes[nid]["kind"] in LABELLED]
|
||
if not labelled:
|
||
return None
|
||
c.update(target=rng.choice(labelled), label=rng.choice([None, "2", "Table 7", "a", rand_text(rng, 1, 5)]))
|
||
elif op == "set" and cells:
|
||
c.update(target=rng.choice(cells), attr=rng.choice(["colspan", "rowspan", "role", "column", "level"]),
|
||
value=rng.choice([1, 2, 3, "2", -1, "data", "row-header", "boss", None]))
|
||
elif op == "delete" and len(doc.order) > 8:
|
||
c.update(target=rng.choice(doc.order[1:]))
|
||
elif op == "reclassify":
|
||
c.update(target=rng.choice(doc.order), kind=rng.choice(["footnote", "quote", "paragraph", "caption", "section"]))
|
||
elif op == "insert-text" and sentences:
|
||
nid = rng.choice(sentences)
|
||
c.update(target=nid, run=rng.randint(0, len(doc.nodes[nid]["content"])), offset=rng.randint(0, 20),
|
||
text=rand_text(rng, 1, 6))
|
||
elif op == "insert-node":
|
||
paragraphs = _leaf_ids(doc, {"paragraph"})
|
||
if not paragraphs:
|
||
return None
|
||
parent = rng.choice(paragraphs)
|
||
kids = doc.kids(parent)
|
||
node = {"kind": "sentence", "content": [{"kind": "text", "text": rand_text(rng, 1, 20) or "x"}]}
|
||
if rng.random() < 0.2:
|
||
node["foo"] = 1
|
||
c.update(parent=parent, place_after=rng.choice(kids + [None]), node=node)
|
||
else:
|
||
return None
|
||
return c
|
||
|
||
|
||
def stress_corrections(rng, rounds):
|
||
"""A random walk of corrections from the tiny document; returns the states it passed through."""
|
||
tiny = tiny_package(None, receipts=False)
|
||
st = tiny["final"]
|
||
base = {"author": {"key": "ana"}, "reason": "Stress.", "at": "2026-10-11T12:00:00Z"}
|
||
seen = {n["id"] for n in st.nodes}
|
||
history, accepted = [st], 0
|
||
for _ in range(rounds):
|
||
doc = st.doc()
|
||
c = random_correction(rng, st, doc)
|
||
if c is None:
|
||
continue
|
||
c = dict(base, **c)
|
||
try:
|
||
st1, rec = CO.apply(st, copy.deepcopy(c), store=SCHEMAS)
|
||
except Refusal as e:
|
||
ok("corrections", e.rule in RULES, "a refusal names the unregistered rule %s" % e.rule)
|
||
continue
|
||
except Exception as e: # noqa: BLE001
|
||
ok("corrections", False, "apply raised %r on %s" % (e, json.dumps(c)[:160]))
|
||
continue
|
||
accepted += 1
|
||
try:
|
||
validate_structure(st1.doc())
|
||
ok("corrections", st1.index().values == reference_values(st1.doc()),
|
||
"incremental values differ from the full computation after %s" % c["op"])
|
||
except Refusal as e:
|
||
ok("corrections", False, "an accepted correction left the tree invalid: %s" % e.rule)
|
||
new_ids = {n["id"] for n in st1.nodes} - {n["id"] for n in st.nodes}
|
||
ok("corrections", not (new_ids & seen), "an ID was reused: %s" % sorted(new_ids & seen))
|
||
seen |= new_ids
|
||
table = {e["id"] for e in st1.ids}
|
||
ok("corrections", set(rec.get("retired", [])) <= table, "a retired ID is not in the ID table")
|
||
ok("corrections", st1.ids == sorted(st1.ids, key=CO.id_order), "ids are not in the order of spec/04 §4.4")
|
||
ok("corrections", rec["id"] == record_id(sha256_hex(jcs.dumpb({k: v for k, v in rec.items() if k != "id"}))),
|
||
"the record's id does not derive from its content")
|
||
ok("corrections", rec["index"] == st.rev + 1 == st1.rev, "the record's index is not the next revision")
|
||
st2, rec2 = CO.apply(st, copy.deepcopy(c), store=SCHEMAS)
|
||
ok("corrections", rec2 == rec and st2.nodes == st1.nodes and st2.ids == st1.ids, "apply is not deterministic")
|
||
try:
|
||
_, rec3 = CO.gate(st, copy.deepcopy(c), SCHEMAS)
|
||
ok("corrections", rec3 == rec, "the gate records something else than apply")
|
||
except Refusal as e:
|
||
ok("corrections", e.rule == "correction.schema", "the gate refused an applied correction with %s" % e.rule)
|
||
st = st1
|
||
history.append(st)
|
||
ok("corrections", accepted >= rounds // 10, "too few random corrections were accepted: %d of %d" % (accepted, rounds))
|
||
return history
|
||
|
||
|
||
def stress_anchors(rng, history, rounds):
|
||
st0, final = history[0], history[-1]
|
||
doc0, vals0 = st0.doc(), hashing.compute(st0.doc())
|
||
outcomes = {"current", "changed-below", "moved", "reattached", "stale"}
|
||
seen_outcomes = set()
|
||
anchors = []
|
||
for nid in doc0.preorder():
|
||
anchors.append(AN.make_anchor(doc0, vals0, nid, 0, store="smithlab"))
|
||
if doc0.nodes[nid]["kind"] in LEAVES:
|
||
t = hashing.plain(doc0, nid)
|
||
if len(t) > 2:
|
||
i = rng.randrange(len(t) - 1)
|
||
j = rng.randint(i + 1, len(t))
|
||
anchors.append(AN.make_anchor(doc0, vals0, nid, 0, span=[i, j], store="smithlab"))
|
||
for _ in range(rounds // 4):
|
||
a = copy.deepcopy(rng.choice(anchors))
|
||
a["node"] = letter_first(os.urandom(32), 7) if rng.random() < 0.5 else a["node"]
|
||
a.pop("kind", None)
|
||
anchors.append(a)
|
||
for state in (final, history[len(history) // 2], history[min(1, len(history) - 1)]):
|
||
values = hashing.compute(state.doc())
|
||
for a in anchors:
|
||
try:
|
||
r = AN.resolve(state, a)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("anchors", False, "resolve raised %r on %s" % (e, json.dumps(a)[:120]))
|
||
continue
|
||
ok("anchors", r["result"] in outcomes, "unknown outcome %s" % r.get("result"))
|
||
seen_outcomes.add(r["result"])
|
||
if r["result"] != "stale" or r.get("node"):
|
||
ok("anchors", r.get("node") in state.doc().nodes, "a non-stale result names no live node")
|
||
if r["result"] == "current":
|
||
ok("anchors", values[r["node"]]["node_hash"] == a["node_hash"], "current with another node_hash")
|
||
if "span" in r:
|
||
t = hashing.plain(state.doc(), r["node"])
|
||
ok("anchors", 0 <= r["span"][0] <= r["span"][1] <= len(t), "a span outside the node's text")
|
||
ok("anchors", {"current", "stale"} <= seen_outcomes, "the random walk produced only %s" % sorted(seen_outcomes))
|
||
|
||
|
||
def _merkle_reference(items):
|
||
leaves = [sha256_hex(b"\x00" + jcs.dumpb(x)) for x in items]
|
||
|
||
def mth(ls):
|
||
if len(ls) == 1:
|
||
return ls[0]
|
||
k = 1
|
||
while 2 * k < len(ls):
|
||
k *= 2
|
||
return sha256_hex(b"\x01" + bytes.fromhex(mth(ls[:k])) + bytes.fromhex(mth(ls[k:])))
|
||
return mth(leaves)
|
||
|
||
|
||
def stress_merkle(rng, rounds):
|
||
"""Receipt items are lists, `["node", id, hash]` or `[kind, record id, hash]` (spec/08 §8.5)."""
|
||
for _ in range(rounds):
|
||
n = rng.randint(1, 40)
|
||
items = [[rng.choice(["node", "annotation", "digitization", "checklist"]), letter_first(os.urandom(32), 7),
|
||
sha256_hex(os.urandom(8))] for _ in range(n)]
|
||
if rng.random() < 0.3:
|
||
items.append([rand_text(rng), [rand_text(rng)], rng.randint(0, 2**53)])
|
||
ok("merkle", crypto.merkle_root(items) == _merkle_reference(items), "Merkle roots differ for %d items" % len(items))
|
||
|
||
|
||
def stress_spotcheck(rng, rounds):
|
||
for _ in range(rounds):
|
||
n = rng.randint(1, 60)
|
||
ids = sorted({letter_first(os.urandom(32), 7) for _ in range(n)})
|
||
receipt = {"format": "scienting-receipt/1", "nodes": [[i, sha256_hex(i)] for i in ids]}
|
||
seed = RC.make_seed(receipt, os.urandom(32))
|
||
rate, minimum = rng.choice(["0.05", "0.1", "0.5", "1"]), rng.randint(1, 15)
|
||
sample = RC.sample(receipt, seed, rate, minimum)
|
||
try:
|
||
RC.sample(receipt, os.urandom(64), rate, minimum)
|
||
ok("spot-check", False, "random bytes were accepted as a seed")
|
||
except ValueError:
|
||
pass
|
||
size = RC.sample_size(len(ids), rate, minimum)
|
||
ok("spot-check", len(sample) == size == min(len(ids), max(minimum, -(-Fraction(rate) * len(ids) // 1))),
|
||
"sample size %d is not the rule's for n=%d" % (len(sample), len(ids)))
|
||
ok("spot-check", sample == sorted(sample) and set(sample) <= set(ids), "sample is not sorted or not from the receipt")
|
||
shuffled = {"nodes": rng.sample(receipt["nodes"], len(receipt["nodes"]))}
|
||
ok("spot-check", RC.sample(shuffled, seed, rate, minimum) == sample, "the sample depends on input order")
|
||
|
||
|
||
def stress_bounds(rng, rounds):
|
||
alpha = Fraction(5, 100)
|
||
for _ in range(rounds // 4):
|
||
n = rng.randint(1, 120)
|
||
k = rng.randint(0, n)
|
||
b = RC.bound_string(k, n)
|
||
ok("bounds", len(b.split(".")[1]) == 6, "bound has not six decimals: %s" % b)
|
||
if k >= n:
|
||
ok("bounds", b == "1.000000", "k = n must give 1")
|
||
continue
|
||
p = Fraction(b)
|
||
|
||
def tail(q):
|
||
return sum(comb(n, i) * q**i * (1 - q)**(n - i) for i in range(k + 1))
|
||
ok("bounds", tail(p) <= alpha, "the tail at the bound is above alpha for k=%d n=%d" % (k, n))
|
||
if p > 0:
|
||
ok("bounds", tail(p - Fraction(1, 10**6)) > alpha, "the bound is not the least six-decimal value for k=%d n=%d" % (k, n))
|
||
if k + 1 < n:
|
||
ok("bounds", Fraction(RC.bound_string(k + 1, n)) >= p, "the bound is not monotone in k")
|
||
|
||
|
||
def stress_fingerprints(rng, rounds):
|
||
rules = json.load(open(os.path.join(ROOT, "registry", "stamps", "scienting-stamps-1.json"), encoding="utf-8"))["rules"]
|
||
for _ in range(rounds):
|
||
lines = [{"text": rand_text(rng, 1, 40)} for _ in range(rng.randint(1, 8))]
|
||
page = {"size": [612.0, 792.0], "lines": lines, "images": [sha256_hex(rand_text(rng)) for _ in range(rng.randint(0, 2))]}
|
||
stamped = copy.deepcopy(page)
|
||
stamped["lines"].insert(rng.randint(0, len(lines)),
|
||
{"text": "Downloaded via EXAMPLE UNIV on %s" % rand_text(rng, 1, 10, list("abcXYZ0123456789-:"))})
|
||
fp, removed = FP.fingerprint(page, rules)
|
||
fp2, removed2 = FP.fingerprint(stamped, rules)
|
||
ok("fingerprints", fp == fp2 and len(removed2) == len(removed) + 1, "a stamped download has another fingerprint")
|
||
digit_lines = [i for i, l in enumerate(lines) if any(ch in "0123456789" for ch in l["text"])]
|
||
if digit_lines:
|
||
changed = copy.deepcopy(page)
|
||
i = rng.choice(digit_lines)
|
||
t = changed["lines"][i]["text"]
|
||
j = next(j for j, ch in enumerate(t) if ch in "0123456789")
|
||
changed["lines"][i]["text"] = t[:j] + str((int(t[j]) + 1) % 10) + t[j + 1:]
|
||
ok("fingerprints", FP.fingerprint(changed, rules)[0] != fp, "a changed digit kept the fingerprint")
|
||
spaced = copy.deepcopy(page)
|
||
spaced["lines"] = [{"text": " ".join(l["text"])} for l in spaced["lines"]]
|
||
ok("fingerprints", FP.fingerprint(spaced, rules)[0] == fp, "inserted spaces changed the fingerprint")
|
||
|
||
|
||
def stress_zip(rng, rounds):
|
||
src = os.path.join(ROOT, "fixtures", "packages", TINY_KEY + PK.ZIP_EXTENSION)
|
||
with open(src, "rb") as f:
|
||
good = f.read()
|
||
accepted = 0
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
for i in range(rounds):
|
||
data = bytearray(good)
|
||
mode = rng.random()
|
||
if mode < 0.6:
|
||
for _ in range(rng.randint(1, 3)):
|
||
data[rng.randrange(len(data))] = rng.randrange(256)
|
||
elif mode < 0.8:
|
||
data = data[:rng.randrange(len(data))]
|
||
else:
|
||
data = data[:rng.randrange(len(data))] + os.urandom(rng.randint(1, 64)) + data[rng.randrange(len(data)):]
|
||
path = os.path.join(tmp, "z%d.zip" % i)
|
||
with open(path, "wb") as f:
|
||
f.write(data)
|
||
try:
|
||
rep = PK.validate_zip(path, STORE)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("zip", False, "validate_zip raised %r on mutation %d" % (e, i))
|
||
continue
|
||
accepted += bool(rep)
|
||
dest = os.path.join(tmp, "d%d" % i)
|
||
try:
|
||
PK.read_zip(path, dest)
|
||
except Refusal as e:
|
||
ok("zip", e.rule in RULES, "read_zip names the unregistered rule %s" % e.rule)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("zip", False, "read_zip raised %r on mutation %d" % (e, i))
|
||
stray = [f for f in os.listdir(tmp) if not (f.startswith("z") and f.endswith(".zip")) and not f.startswith("d")]
|
||
ok("zip", not stray, "the zip reader wrote outside its folder: %s" % stray)
|
||
ok("zip", accepted <= rounds // 20 + 1, "%d of %d mutated archives were accepted" % (accepted, rounds))
|
||
|
||
|
||
def stress_packages(rng, rounds):
|
||
src = os.path.join(ROOT, "fixtures", "packages", TINY_KEY)
|
||
files = sorted(os.path.relpath(os.path.join(d, f), src) for d, _, fs in os.walk(src) for f in fs)
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
for i in range(rounds):
|
||
pkg = os.path.join(tmp, "p%d" % i)
|
||
shutil.copytree(src, pkg)
|
||
mode = rng.choice(["flip", "flip", "truncate", "delete", "add", "rename-case", "symlink"])
|
||
target = rng.choice(files)
|
||
full = os.path.join(pkg, target)
|
||
if mode == "flip":
|
||
data = bytearray(open(full, "rb").read())
|
||
if data:
|
||
k = rng.randrange(len(data))
|
||
data[k] = (data[k] + rng.randrange(1, 256)) % 256
|
||
open(full, "wb").write(data)
|
||
elif mode == "truncate":
|
||
data = open(full, "rb").read()
|
||
open(full, "wb").write(data[:rng.randrange(len(data) + 1)])
|
||
elif mode == "delete":
|
||
os.remove(full)
|
||
elif mode == "add":
|
||
open(os.path.join(pkg, rand_text(rng, 1, 8, list("abcxyz")) + ".txt"), "wb").write(b"stray\n")
|
||
elif mode == "rename-case":
|
||
if target.lower() == target.upper():
|
||
continue
|
||
os.rename(full, os.path.join(pkg, target.upper()))
|
||
else:
|
||
os.remove(full)
|
||
os.symlink(os.path.join(src, target), full)
|
||
try:
|
||
rep = PK.validate(pkg, STORE)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("packages", False, "validate raised %r after %s of %s" % (e, mode, target))
|
||
continue
|
||
ok("packages", not rep, "a package mutated by %s of %s validated" % (mode, target))
|
||
ok("packages", all(e["rule"] in RULES for e in rep.errors), "an unregistered rule was reported")
|
||
shutil.rmtree(pkg)
|
||
|
||
|
||
def stress_rights(rng, rounds):
|
||
tri = [True, False, "unknown"]
|
||
keys = list(RI.USES) + ["ai/" + p for p in RI.AI_PAIRS]
|
||
for _ in range(rounds):
|
||
flats = [{k: rng.choice(tri) for k in keys} for _ in range(rng.randint(1, 4))]
|
||
combined = RI.combine(flats)
|
||
shuffled = RI.combine(rng.sample(flats, len(flats)))
|
||
ok("rights", combined == shuffled, "combining depends on order")
|
||
for k in keys:
|
||
vals = [f[k] for f in flats]
|
||
want = False if False in vals else ("unknown" if "unknown" in vals else True)
|
||
ok("rights", combined[k] == want, "combining %s gave %r for %r" % (k, combined[k], vals))
|
||
|
||
|
||
def _align_items(rng, prefix, texts):
|
||
items = []
|
||
for i, t in enumerate(texts):
|
||
runs = [{"kind": "text", "text": t}]
|
||
doc, s = sentence_doc(runs)
|
||
strict, cx = hashing.strict(doc, s)
|
||
items.append({"id": "%s%d" % (prefix, i), "container": "p%d" % (i // 3), "context": "paragraph", "runs": runs,
|
||
"strict": strict, "lossy": hashing.lossy_from_strict(strict), "complex": cx})
|
||
return items
|
||
|
||
|
||
def stress_alignment(rng, rounds):
|
||
for _ in range(rounds // 4):
|
||
texts = [rand_text(rng, 5, 60) or "x" for _ in range(rng.randint(1, 12))]
|
||
a = _align_items(rng, "a", texts)
|
||
try:
|
||
same = AL.align(a, _align_items(rng, "b", texts))
|
||
b_texts = [t for t in texts if rng.random() > 0.3] + [rand_text(rng, 5, 30)]
|
||
diff = AL.align(a, _align_items(rng, "b", b_texts))
|
||
ok("alignment", AL.align(a, _align_items(rng, "b", b_texts)) == diff, "alignment is not deterministic")
|
||
except Exception as e: # noqa: BLE001
|
||
ok("alignment", False, "align raised %r" % (e,))
|
||
continue
|
||
ok("alignment", len(same["pairs"]) == len(a) and all(p["agree"] for p in same["pairs"])
|
||
and not [s for s in same["suspects"] if s["kind"] not in ("structure", "join")],
|
||
"aligning two equal readings left disagreements or unpaired sentences: %s" % same["suspects"][:3])
|
||
ids_a = {x["id"] for x in a}
|
||
ok("alignment", all(p["a"] in ids_a for p in diff["pairs"]), "a pair names an unknown sentence")
|
||
paired = {p["a"] for p in diff["pairs"]}
|
||
dropped = {s["a"] for s in diff["suspects"] if s["kind"] == "dropped"}
|
||
ok("alignment", paired | dropped == ids_a and not (paired & dropped),
|
||
"every sentence of a reading is paired or dropped, not both or neither")
|
||
|
||
|
||
def stress_addresses(rng, rounds):
|
||
for _ in range(rounds):
|
||
store = rng.choice([None, "smithlab", "a1", "x.y-z"])
|
||
key = rng.choice([HALE, "k-0000", "a1-aaaa.si", "x-k3f9qd.v2"])
|
||
node = letter_first(os.urandom(32), 7)
|
||
pin = sha256_hex(rand_text(rng))[:6] if rng.random() < 0.5 else None
|
||
span = sorted([rng.randint(0, 50), rng.randint(0, 50)]) if rng.random() < 0.5 else None
|
||
s = AN.format_address(store, key, node, pin, span)
|
||
parsed = AN.parse_address(s)
|
||
ok("addresses", parsed and AN.format_address(parsed["store"], parsed["key"], parsed["node"], parsed["pin"],
|
||
parsed["span"]) == s, "address does not round trip: %s" % s)
|
||
junk = rand_text(rng, 0, 20)
|
||
try:
|
||
AN.parse_address(junk)
|
||
except Exception as e: # noqa: BLE001
|
||
ok("addresses", False, "parse_address raised %r on %r" % (e, junk))
|
||
for _ in range(rounds):
|
||
d = os.urandom(32)
|
||
ok("ids", NODE_ID_RE.match(letter_first(d, 7)) and RECORD_ID_RE.match(record_id(d.hex())), "ID form")
|
||
|
||
|
||
def stress_targets(rng, rounds):
|
||
st = PK.read_state(os.path.join(ROOT, "fixtures", "packages", HALE))
|
||
doc = st.doc()
|
||
for _ in range(rounds // 4):
|
||
anchored = rng.sample(doc.order, rng.randint(0, 6))
|
||
sets = {}
|
||
for rule in TG.DIGIT_RULES:
|
||
got = TG.target_set(doc, anchored, rule)
|
||
ids = [n for n, _ in got]
|
||
pos = {nid: i for i, nid in enumerate(doc.order)}
|
||
ok("targets", ids == sorted(ids, key=pos.get) and len(set(ids)) == len(ids), "targets are not in document order")
|
||
ok("targets", all(doc.nodes[n]["kind"] in LEAVES for n in ids), "a target is not a leaf")
|
||
sets[rule] = set(ids)
|
||
ok("targets", sets["none"] <= sets["anchored-sections"] <= sets["all"], "the digit rules do not nest")
|
||
try:
|
||
TG.target_set(doc, anchored + ["zzzzzzz"], "all")
|
||
ok("targets", False, "an unknown anchored node was accepted")
|
||
except ValueError:
|
||
ok("targets", True, "")
|
||
|
||
|
||
def stress_receipt_chains(rng, rounds):
|
||
"""Chains of spot-checks hundreds deep settle without recursion, and alternate as spec/08 §8.6 implies."""
|
||
depth = max(50, rounds)
|
||
ids = sorted({letter_first(os.urandom(32), 7) for _ in range(20)})
|
||
rec = RC.seal({"format": "scienting-receipt/1", "doc": HALE, "rev": 0, "level": "F3", "scope": "all",
|
||
"nodes": [[i, sha256_hex(i)] for i in ids], "evidence": [], "by": "ana",
|
||
"at": "2026-10-11T12:00:00Z"}, secret("ana"), "ana")
|
||
records = [rec]
|
||
for i in range(depth):
|
||
orig = records[-1]
|
||
seed = RC.make_seed(orig, secret("ben-spot"))
|
||
sample = RC.sample(orig, seed)
|
||
hashes = dict(orig["nodes"])
|
||
q = RC.seal({"format": "scienting-receipt/1", "doc": HALE, "rev": 0, "level": "F3",
|
||
"scope": "qa:" + orig["id"], "nodes": [[n, hashes[n]] for n in sample], "evidence": [],
|
||
"by": "ben", "at": "2026-10-11T12:%02d:%02dZ" % (i // 60 % 60, i % 60),
|
||
"qa": {"of": orig["id"], "key": "ben-spot", "seed": __import__("base64").b64encode(seed).decode(),
|
||
"rate": "0.05", "minimum": 10, "sample": sample, "errors": [{"node": sample[0]}]}},
|
||
secret("ben"), "ben")
|
||
records.append(q)
|
||
t0 = time.time()
|
||
try:
|
||
status = RC.statuses(records, ROSTER_BY_ID)
|
||
except RecursionError:
|
||
ok("receipts", False, "a chain of %d spot-checks overflowed the stack" % depth)
|
||
return
|
||
ok("receipts", time.time() - t0 < 30, "statuses of a %d-deep chain took too long" % depth)
|
||
# The last spot-check nobody checks is valid; each receipt it checks is suspended; and so on, alternating.
|
||
want = "valid"
|
||
for r in reversed(records):
|
||
ok("receipts", status[r["id"]] == want, "receipt at depth %d is %s, not %s" % (records.index(r), status[r["id"]], want))
|
||
want = "suspended" if want == "valid" else "valid"
|
||
|
||
|
||
def stress_keys(rng, rounds):
|
||
"""Keys (spec/02 §2.4): the pattern always holds, the suffix is a prefix of longer ones, lengthening settles a
|
||
collision without touching the first key, and a key's hash part depends on the identity alone."""
|
||
words = ["Attention", "On", "the", "Origin", "of", "Species", "H\u2082O", "3D", "Self-consistent", "\u00c9tude", "\u5f20", "don't", "a"]
|
||
for _ in range(rounds):
|
||
meta = {"title": " ".join(rng.choice(words) for _ in range(rng.randint(0, 6))),
|
||
"author": [{"family": rng.choice(["M\u00fcller", "O'Brien", "\u5f20\u4f1f", "", "de la Cruz", "Smith Jr."])}] if rng.random() < 0.8 else [],
|
||
"issued": {"date-parts": [[rng.choice([2017, 1859, 99, 20261, None])]]} if rng.random() < 0.8 else {}}
|
||
same_as = [rng.choice(["doi:10.1/X.Y", "doi:10.1/x.y", "arxiv:1706.03762", "url:https://e.org/p", "openalex:W1"])] if rng.random() < 0.6 else []
|
||
key = K.work_key(same_as, meta)
|
||
ok("keys", K.WORK_KEY_RE.match(key) and K.DOCUMENT_KEY_RE.match(key), "a minted key breaks the pattern: %r" % key)
|
||
ident = K.identity(same_as, meta)
|
||
ok("keys", K.suffix(ident, 8).startswith(K.suffix(ident)), "a longer suffix does not extend the shorter")
|
||
longer = K.unique_work_key(same_as, meta, {key})
|
||
ok("keys", longer != key and longer.startswith(key) and len(longer) == len(key) + 2, "lengthening does not extend the key")
|
||
if same_as and same_as[0].lower().startswith("doi:"):
|
||
other = dict(meta, title="Another title entirely")
|
||
ok("keys", K.suffix_of(K.work_key(same_as, other)) == K.suffix_of(key), "the hash part depends on more than the identity")
|
||
for v in ({"type": "version-of-record"}, {"type": "supplement"}, {"type": "preprint", "seq": rng.randint(1, 9)}, {"type": "erratum"}):
|
||
dk = K.document_key(key, v)
|
||
ok("keys", K.DOCUMENT_KEY_RE.match(dk) and K.split_key(dk)[0] == key, "a document key does not split back")
|
||
|
||
|
||
def stress_vrf(rng, rounds):
|
||
"""The VRF proves and verifies for random keys and inputs, and a single flipped byte never verifies."""
|
||
for _ in range(max(5, rounds // 20)):
|
||
sk = os.urandom(32)
|
||
pk = vrf.public_key(sk)
|
||
alpha = os.urandom(rng.randint(0, 40))
|
||
pi = vrf.prove(sk, alpha)
|
||
beta = vrf.verify(pk, alpha, pi)
|
||
ok("vrf", beta is not None and beta == vrf.proof_to_hash(pi) and len(pi) == 80 and len(beta) == 64, "a proof does not verify")
|
||
ok("vrf", vrf.prove(sk, alpha) == pi, "proving is not deterministic")
|
||
bad = bytearray(pi)
|
||
bad[rng.randrange(80)] ^= 1 << rng.randrange(8)
|
||
ok("vrf", vrf.verify(pk, alpha, bytes(bad)) is None, "a tampered proof verifies")
|
||
ok("vrf", vrf.verify(pk, alpha + b"x", pi) is None and vrf.verify(vrf.public_key(os.urandom(32)), alpha, pi) is None,
|
||
"a proof verifies for another input or key")
|
||
ok("vrf", vrf.verify(pk, alpha, os.urandom(80)) is None and vrf.verify(os.urandom(32), alpha, pi) is None, "garbage verifies")
|
||
|
||
|
||
def stress_proofs(rng, rounds):
|
||
"""RFC 9162 inclusion and consistency proofs verify for every index and prefix of random trees, and fail for a
|
||
wrong index, root or size."""
|
||
for _ in range(rounds // 4):
|
||
n = rng.randint(1, 40)
|
||
items = [["node", letter_first(os.urandom(32), 7), sha256_hex(os.urandom(4))] for _ in range(n)]
|
||
root = crypto.history_root(items)
|
||
for i in range(n):
|
||
path = crypto.inclusion_path(items, i)
|
||
leaf = crypto.leaf_hash(items[i])
|
||
ok("proofs", crypto.verify_inclusion(root, leaf, i, n, path), "inclusion fails for %d of %d" % (i, n))
|
||
ok("proofs", n == 1 or not crypto.verify_inclusion(root, leaf, (i + 1) % n, n, path), "inclusion passes for another index")
|
||
ok("proofs", not crypto.verify_inclusion(sha256_hex(b"x"), leaf, i, n, path), "inclusion passes for another root")
|
||
# A wrong size is not always detectable from a path: RFC 9162 commits to the content through the
|
||
# root, and two sizes with one tree shape verify alike. The root is what a verifier must trust.
|
||
for m in range(n + 1):
|
||
proof = crypto.consistency_proof(items, m)
|
||
ok("proofs", crypto.verify_consistency(crypto.history_root(items[:m]), root, m, n, proof), "consistency fails %d to %d" % (m, n))
|
||
if 0 < m < n:
|
||
ok("proofs", not crypto.verify_consistency(sha256_hex(b"x"), root, m, n, proof)
|
||
and not crypto.verify_consistency(crypto.history_root(items[:m]), sha256_hex(b"y"), m, n, proof),
|
||
"a consistency proof passes with a wrong root")
|
||
ok("proofs", not crypto.verify_consistency(crypto.history_root(items[:1]), root, 1, n, ["zz"]), "garbage path verifies")
|
||
|
||
|
||
def stress_checkpoints(rng, rounds):
|
||
"""Checkpoint notes parse back to what was signed and reject edits anywhere."""
|
||
for _ in range(rounds // 10):
|
||
entries = [[rng.choice(["receipt", "revocation"]), letter_first(os.urandom(32), 13), sha256_hex(os.urandom(4))]
|
||
for _ in range(rng.randint(0, 20))]
|
||
note = LG.sign_checkpoint("scient.ing/log/t", entries, [("ana", secret("ana")), ("ben", secret("ben"))])
|
||
parsed, verified = LG.verify_checkpoint(note, ROSTER_BY_ID)
|
||
ok("checkpoints", parsed and parsed["size"] == len(entries) and parsed["root"] == LG.root(entries) and verified == ["ana", "ben"],
|
||
"a checkpoint does not verify")
|
||
k = rng.randrange(len(note))
|
||
bad = note[:k] + chr((ord(note[k]) + 1) % 128 or 1) + note[k + 1:]
|
||
p2, v2 = LG.verify_checkpoint(bad, ROSTER_BY_ID)
|
||
# An edit in the body loses every signer; an edit in one signature line loses that signer.
|
||
ok("checkpoints", p2 is None or len(v2) < len(verified), "an edited note still fully verifies: %r" % bad[max(0, k - 5):k + 5])
|
||
|
||
|
||
def stress_mldsa(rng, rounds):
|
||
"""ML-DSA-65 signs deterministically and verifies; a flipped bit anywhere, another message or another key fails."""
|
||
for _ in range(max(3, rounds // 50)):
|
||
seed = os.urandom(32)
|
||
pk, sk = mldsa.keygen(seed)
|
||
msg = os.urandom(rng.randint(0, 100))
|
||
sig = mldsa.sign(sk, msg)
|
||
ok("ml-dsa", len(pk) == mldsa.PK_SIZE and len(sk) == mldsa.SK_SIZE and len(sig) == mldsa.SIG_SIZE, "sizes")
|
||
ok("ml-dsa", mldsa.verify(pk, msg, sig) and mldsa.sign(sk, msg) == sig, "a signature does not verify or is not deterministic")
|
||
bad = bytearray(sig)
|
||
bad[rng.randrange(len(bad))] ^= 1 << rng.randrange(8)
|
||
ok("ml-dsa", not mldsa.verify(pk, msg, bytes(bad)), "a tampered signature verifies")
|
||
ok("ml-dsa", not mldsa.verify(pk, msg + b"x", sig) and not mldsa.verify(mldsa.keygen(os.urandom(32))[0], msg, sig),
|
||
"a signature verifies for another message or key")
|
||
ok("ml-dsa", not mldsa.verify(pk, msg, os.urandom(mldsa.SIG_SIZE)) and not mldsa.verify(os.urandom(mldsa.PK_SIZE), msg, sig),
|
||
"garbage verifies")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
ap.add_argument("--seed", type=int, default=20261014, help="the random seed (default 20261014)")
|
||
ap.add_argument("--rounds", type=int, default=200, help="cases per property (default 200)")
|
||
a = ap.parse_args()
|
||
rng = random.Random(a.seed)
|
||
print("seed %d, %d rounds" % (a.seed, a.rounds))
|
||
stress_canonical_json(rng, a.rounds)
|
||
stress_normalization(rng, a.rounds)
|
||
stress_notation(rng, a.rounds)
|
||
history = stress_corrections(rng, a.rounds)
|
||
stress_anchors(rng, history, a.rounds)
|
||
stress_merkle(rng, a.rounds)
|
||
stress_spotcheck(rng, a.rounds)
|
||
stress_bounds(rng, a.rounds)
|
||
stress_fingerprints(rng, a.rounds)
|
||
stress_zip(rng, a.rounds)
|
||
stress_packages(rng, a.rounds // 4)
|
||
stress_rights(rng, a.rounds)
|
||
stress_alignment(rng, a.rounds)
|
||
stress_addresses(rng, a.rounds)
|
||
stress_targets(rng, a.rounds)
|
||
stress_receipt_chains(rng, a.rounds)
|
||
stress_keys(rng, a.rounds)
|
||
stress_vrf(rng, a.rounds)
|
||
stress_proofs(rng, a.rounds)
|
||
stress_checkpoints(rng, a.rounds)
|
||
stress_mldsa(rng, a.rounds)
|
||
for k in sorted(COUNTS):
|
||
print("%-17s %6d checks" % (k, COUNTS[k]))
|
||
if FAILS:
|
||
print("\n%d failures (seed %d):" % (len(FAILS), a.seed))
|
||
for f in FAILS[:60]:
|
||
print(" " + f)
|
||
sys.exit(1)
|
||
print("\nEvery property held.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|