feat: zipped packages, target set, checklists, audit fixes and a stress test

Features for the next six months (accessible HTML export, paperpack and
Lode on packages, the personal bundle):

- Zipped packages: the oracle writes and reads `.sct.zip` archives under
  the rules of spec/12 §12.2; validate.py and load.py accept them; a
  byte-identical zip of the tiny package joins the fixtures, and two
  broken archives (traversal, symbolic link) join fixtures/invalid.
- The target set of targeted F3 (spec/08 §8.10) is computed by the
  oracle and pinned by vectors/targets.jsonl under each digit rule; the
  rule now covers cells as well as sentences and the leaves of anchored
  containers, and defines a node's section.
- scienting-checklist/1 (provisional): a conformance check of an export,
  covered by a receipt as ["checklist", id, hash] (spec/08 §8.11), with a
  WCAG 2.1 AA fixture and its receipt.
- QA reports are signed like receipts (spec/08 §8.9).
- A `description` annotation body with role alt or long, so an
  accessible export can find a confirmed figure description; the layer
  fixture carries one, and the store's body-kind constraint allows it.
- scienting-spot-check/1, the seed payload, is registered with a schema.
- exercise, submission and grade are provisional until a course has
  used them.

Audit fixes, each with a vector or a broken package where one was
missing:

- Folder packages: symbolic links are refused on validate and load,
  manifest paths must not collide under case folding, listed files are
  matched by exact name, files are hashed in chunks under a size limit.
- Files are compared byte for byte, so CR LF, a byte order mark, NaN,
  a lone surrogate or invalid UTF-8 is refused instead of translated;
  integers beyond 2^53 are read as doubles, as canonical JSON writes them.
- Receipts: signature.key must equal by and the key must be valid when
  signed; a qa receipt's nodes must be the sample at the original's
  hashes, its scope must name qa.of, its errors must be sampled nodes;
  statuses are computed once, iteratively, so chains of spot-checks
  hundreds deep settle; Base64 and manifest shape errors become refusals
  instead of tracebacks; counts and the fidelity of roster-less packages
  are recomputed.
- The gate enforces rule 11 of spec/11 §11.4: changed nodes are
  validated against the node schema and the references between nodes
  are checked, so a set to a bad value, an insert with unknown
  properties or a delete of a referenced header cell is refused; a
  relabel to null drops the label; a reclassify drops foreign
  attributes; a split strips whitespace across runs; ids.jsonl order
  follows spec/04 §4.4 in every state; a move after itself is refused.
- Anchors: the relabel rule requires that nothing else changed below the
  node; an alias chain ending in a deletion no longer crashes resolve.
- Notation: a script after an empty group applies to the previous atom
  ({}^{14}C); letters are defined by Unicode block in spec/05 §5.4 and
  the oracle; \left) is complex.
- Page fingerprints delete whitespace before folding, so "10 -20" and
  "10-20" fold alike (spec/07 §7.1).
- Grades are summed with exact decimals; the reviewer error bound is
  computed with integer arithmetic.
- receipts and revocations are keyed by (doc_key, id) in SQL, since one
  revocation may be carried by several packages.

Checking:

- tools/stress.py: seeded randomized property tests over canonical JSON,
  normalization, notation, random correction walks, anchor resolution,
  Merkle roots, spot-check samples, error bounds, fingerprints, mutated
  archives and packages, rights and spot-check chains; CI runs it.
- check.py verifies every count and version the prose states against
  the repository, the zipped fixture and crafted archives, the checklist
  coverage, and nine more broken packages.
- CONTRIBUTING.md gains a release checklist.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
alexmandClaude committed 2026-10-14 20:07:22 -04:00
1 parent 8c8b51cc3f
commit ea0a7efa78
142 files changed
+2312 -268

No files matched your search

+4 -3
View File
@@ -15,10 +15,11 @@ A fixture is an example file or package under `fixtures/`.
| Program | Does |
| --- | --- |
| `check.py` | Checks everything: schemas, records, packages, the broken packages, every vector, a byte-for-byte rebuild, the SQL (a database query language) round trip and the prose. `--quick` skips the rebuild. |
| `check.py` | Checks everything: schemas, records, packages and the zipped package, the broken packages, every vector, a byte-for-byte rebuild, the SQL (a database query language) round trip, the counts the prose states, and the prose. `--quick` skips the rebuild. |
| `build.py` | Builds `fixtures/` and `vectors/` from the hand-written sources in `fixtures/src/` and the cases it defines. `--out DIR` builds into another folder. |
| `validate.py` | Validates packages: `python3 tools/validate.py PACKAGE [PACKAGE ...]`. With `--correction PROPOSALS` it runs proposed corrections through the gate against one package's current state, and prints the record each would append or the rule it breaks, without writing anything ([spec/11 §11.4](../spec/11-corrections.md#114-the-gate)). |
| `load.py` | Validates packages, then loads them into a SQLite database with the tables of `sql/`: `python3 tools/load.py --sqlite lab.db PACKAGE [PACKAGE ...]`. With `--bundle --built-from REV` it builds a new bundle instead, with the full-text search index ([spec/13 §13.5](../spec/13-database.md#135-the-bundle)). |
| `validate.py` | Validates packages, as folders or as zipped packages (`.sct.zip`, unpacked under the rules of [spec/12 §12.2](../spec/12-package.md#122-paths)): `python3 tools/validate.py PACKAGE [PACKAGE ...]`. With `--correction PROPOSALS` it runs proposed corrections through the gate against one package's current state, and prints the record each would append or the rule it breaks, without writing anything ([spec/11 §11.4](../spec/11-corrections.md#114-the-gate)). |
| `load.py` | Validates packages, folders or zipped, then loads them into a SQLite database with the tables of `sql/`: `python3 tools/load.py --sqlite lab.db PACKAGE [PACKAGE ...]`. With `--bundle --built-from REV` it builds a new bundle instead, with the full-text search index ([spec/13 §13.5](../spec/13-database.md#135-the-bundle)). |
| `stress.py` | Stresses the oracle with random inputs from a seeded generator: `python3 tools/stress.py [--seed N] [--rounds N]`. It checks the invariants the chapters state rather than the vectors' values, such as canonical JSON round trips, corrections keeping the five values consistent with a full recomputation, anchors always resolving to one of the five outcomes, the zip reader never writing outside its folder, and a mutated package always being refused. |
| `oracle/` | A compact implementation of every rule. `build.py` uses it to compute expected values, and `check.py` uses it to recompute them. |
## When to run what
+137 -11
View File
@@ -14,6 +14,7 @@ import os
import shutil
import struct
import sys
import zipfile
import zlib
from fractions import Fraction
@@ -24,14 +25,17 @@ from tools.oracle import align as AL # noqa: E402
from tools.oracle import anchors as AN # noqa: E402
from tools.oracle import corrections as CO # noqa: E402
from tools.oracle import crypto, fingerprint as FP, grade as GR, hashing, jcs, receipts as RC, rights as RI # noqa: E402
from tools.oracle import targets as TG # noqa: E402
from tools.oracle.model import LEAVES, MATH_KINDS, RUN_KINDS, VERBATIM_KINDS, Doc # noqa: E402
from tools.oracle.notation import fold_math # noqa: E402
from tools.oracle.package import jsonl, line, witness_id, write # noqa: E402
from tools.oracle.package import ZIP_EXTENSION, jsonl, line, witness_id, write, write_zip # noqa: E402
from tools.oracle.schema import Store # noqa: E402
from tools.oracle.text import letter_first, mint_node_id, norm, record_id, sha256_hex # noqa: E402
from tools.oracle.tree import Refusal, _flatten, mint_ids, validate_structure # noqa: E402
VERSION = "0.1.0"
STORE = "smithlab"
SCHEMAS = Store(os.path.join(ROOT, "schemas"))
def H(s):
@@ -440,7 +444,7 @@ def build_document(src, used=None, extra_files=None):
if before is not None:
d = state.doc()
c["offset"] = hashing.plain(d, c["target"]).index(before)
state, rec = CO.apply(state, c)
state, rec = CO.apply(state, c, store=SCHEMAS)
if new_name:
names.ids[new_name] = rec["created"][0]
records.append(rec)
@@ -798,7 +802,11 @@ def vectors(out, hale, tiny):
("tex", "a \\cdot b"), ("typst", "a dot b"), ("tex", "a \\times b \\pm c"), ("tex", "\\left(x\\right)^2"),
("typst", "(a+b)^2"), ("tex", "\\frac{a}{b}"), ("typst", "a/b"), ("tex", "\\sqrt{2}"),
("tex", "\\mathbf{F}"), ("tex", "{ab}_2"), ("tex", "x_1_2"), ("tex", "\\Delta G^\\circ"),
("tex", "E_{\\mathrm{bond}} = \\frac{1}{2} K_r (r - r_{\\mathrm{eq}})^2"), ("typst", "E = K_r (r - r_\"eq\")^2")]
("tex", "E_{\\mathrm{bond}} = \\frac{1}{2} K_r (r - r_{\\mathrm{eq}})^2"), ("typst", "E = K_r (r - r_\"eq\")^2"),
# A script after an empty group applies to the atom before it, or to an empty base (the isotope idiom).
("tex", "{}^{14}C"), ("tex", "x{}^2"), ("tex", "{}_2"), ("tex", "\\left)x\\right("),
# Letters are the Latin and Greek blocks: a fullwidth or modifier letter is complex, a Coptic letter too.
("tex", "A"), ("tex", "ʰ"), ("tex", "Ϣ"), ("tex", "ἀω")]
wv(out, "notation.jsonl", [{"id": "%s:%s" % (n, s), "notation": n, "source": s, "canonical": fold_math(n, s)}
for n, s in notes])
# Projections.
@@ -1018,12 +1026,21 @@ def corrections_vectors():
("merge", {"op": "merge", "target": nm["@c"], "with": nm["@d"], "joiner": " ", "evidence": person}),
("merge-not-adjacent", {"op": "merge", "target": nm["@a"], "with": nm["@c"], "evidence": person}),
("move", {"op": "move", "target": nm["@d"], "parent": nm["@p1"], "place_after": None, "evidence": person}),
("reclassify", {"op": "reclassify", "target": nm["@p2"], "kind": "quote", "evidence": person}),
("move-after-itself", {"op": "move", "target": nm["@d"], "parent": nm["@p2"], "place_after": nm["@d"],
"evidence": person}),
("reclassify", {"op": "reclassify", "target": nm["@p2"], "kind": "footnote", "evidence": person}),
("reclassify-blocks", {"op": "reclassify", "target": nm["@p2"], "kind": "quote", "evidence": person}),
("reclassify-bad", {"op": "reclassify", "target": nm["@a"], "kind": "paragraph", "evidence": person}),
("relabel-section", {"op": "relabel", "target": nm["@sec"], "label": "2", "evidence": crop}),
("relabel-remove", {"op": "relabel", "target": nm["@sec"], "label": None, "evidence": crop}),
("relabel-digit-no-crop", {"op": "relabel", "target": nm["@t"], "label": "Table 2", "evidence": person}),
("set-colspan", {"op": "set", "target": nm["@y1"], "attr": "colspan", "value": 2, "evidence": person}),
("set-bad-value", {"op": "set", "target": nm["@y1"], "attr": "colspan", "value": "2", "evidence": person}),
("set-not-settable", {"op": "set", "target": nm["@y1"], "attr": "level", "value": 2, "evidence": person}),
("insert-node-junk", {"op": "insert", "parent": nm["@p2"], "place_after": nm["@c"],
"node": {"kind": "sentence", "content": [{"kind": "text", "text": "Junk."}], "foo": 1},
"evidence": crop}),
("delete-header-cell", {"op": "delete", "target": nm["@hx"], "evidence": person}),
]
out = []
vals0 = hashing.compute(st0.doc())
@@ -1032,7 +1049,7 @@ def corrections_vectors():
c = dict(base, **c)
case = {"id": cid, "doc": st0.key, "before": {"nodes": before, "ids": [], "rev": 0}, "correction": c}
try:
st1, rec = CO.apply(st0, c)
st1, rec = CO.apply(st0, c, store=SCHEMAS)
v1 = hashing.compute(st1.doc())
case["expect"] = {"record": rec, "nodes": [dict(n, **v1[n["id"]]) for n in st1.nodes], "ids": st1.ids}
except Refusal as e:
@@ -1242,6 +1259,10 @@ def records(out, hale):
target(AN.make_anchor(doc, vals, I["@tab1"], rev, store=STORE)), "assessing", "confirmed"),
ann({"type": "TextualBody", "kind": "exercise", "value": check_body["prompt"], "exercise": check_body},
target(anchor("@s1")), "assessing", "confirmed"),
# A person's description of the figure, which an accessible export may use because it is confirmed.
ann({"type": "TextualBody", "kind": "description", "role": "alt", "language": "en",
"value": "Energy along the C–C bond stretch (panel a) and the fit residual (panel b)."},
target(AN.make_anchor(doc, vals, I["@fig1"], rev, store=STORE)), "describing", "confirmed"),
]
anns.append(ann({"type": "TextualBody", "kind": "comment", "value": "Agreed; see the residual panel."},
{"type": "SpecificResource", "source": anns[2]["id"]}, "replying", created="2026-10-11T19:05:00Z"))
@@ -1367,12 +1388,48 @@ def records(out, hale):
q2 = hale["by_name"]["q2"]
n = len(q1["qa"]["sample"]) + len(q2["qa"]["sample"])
k = len(q1["qa"]["errors"]) + len(q2["qa"]["errors"])
wr(out, "qa-report.json", {"format": "scienting-qa-report/1", "id": record_id(H("qa report 1")),
"delivery": "smithlab evidence pilot (fictional)", "docs": ["smithlab/hale2026harmonic"],
"receipts": [hale["by_name"]["r1"]["id"], hale["by_name"]["r4"]["id"]],
"reviewers": [{"key": "ana", "sampled": n, "errors": k, "confidence": "0.95",
"method": "clopper-pearson", "bound": RC.bound_string(k, n)}],
"at": "2026-10-12T10:00:00Z"})
wr(out, "qa-report.json", RC.seal({
"format": "scienting-qa-report/1", "delivery": "smithlab evidence pilot (fictional)",
"docs": ["smithlab/hale2026harmonic"], "receipts": [hale["by_name"]["r1"]["id"], hale["by_name"]["r4"]["id"]],
"reviewers": [{"key": "ana", "sampled": n, "errors": k, "confidence": "0.95", "method": "clopper-pearson",
"bound": RC.bound_string(k, n)}],
"by": "ana", "at": "2026-10-12T10:00:00Z"}, secret("ana"), "ana"))
# A checklist for an accessible HTML export of the paper, and the receipt that covers it (spec/08 §8.11).
html_sha = H("accessible HTML export of hale2026harmonic (synthetic)")
checker = {"name": "example-checker", "version": "1.0"}
checklist = {"format": "scienting-checklist/1", "doc": "smithlab/hale2026harmonic", "rev": rev,
"standard": {"name": "WCAG 2.1", "level": "AA", "url": "https://www.w3.org/TR/WCAG21/"},
"export": {"kind": "html", "sha256": html_sha, "size": 48213},
"items": [
{"criterion": "1.1.1", "result": "pass", "method": "manual",
"note": "Every figure has a confirmed description in the layer."},
{"criterion": "1.2.1", "result": "not-applicable", "method": "manual", "note": "No audio or video."},
{"criterion": "1.3.1", "result": "pass", "method": "manual",
"note": "Headings, lists and table headers are marked up; each cell names its header path."},
{"criterion": "1.3.2", "result": "pass", "method": "automated", "tool": checker},
{"criterion": "1.4.3", "result": "pass", "method": "automated", "tool": checker},
{"criterion": "1.4.5", "result": "pass", "method": "manual",
"note": "Equations are MathML, never images of text."},
{"criterion": "2.4.2", "result": "pass", "method": "automated", "tool": checker},
{"criterion": "3.1.1", "result": "pass", "method": "automated", "tool": checker}],
"by": "ana", "at": "2026-10-12T11:00:00Z"}
checklist_hash = H(jcs.dumpb(checklist))
checklist["id"] = record_id(checklist_hash)
wr(out, "checklist/checklist.json", checklist)
wr(out, "checklist/receipt.json", RC.seal({
"format": "scienting-receipt/1", "doc": "hale2026harmonic", "rev": rev, "level": "F3",
"scope": "targeted:checklist", "nodes": [], "records": [["checklist", checklist["id"], checklist_hash]],
"evidence": [{"kind": "source-file", "sha256": html_sha, "note": "The exported HTML."}],
"by": "ana", "at": "2026-10-12T11:30:00Z"}, secret("ana"), "ana"))
# The target set of targeted F3 (spec/08 §8.10) under each digit rule: the dataset's cells, the sentence a
# confirmed claim link anchors, and the panel a digitization reads from are the anchored nodes.
dataset_anchors = [json.loads(x) for x in
hale["files"]["datasets/ds-hale-bond-params/anchors.jsonl"].decode().splitlines()]
anchored = sorted({a["anchor"]["node"] for a in dataset_anchors} | {I["@s2"], pa})
wv(out, "targets.jsonl", [{"id": "hale-" + rule, "package": "fixtures/packages/hale2026harmonic",
"anchored": anchored, "digit_rule": rule,
"expect": {"nodes": TG.target_set(doc, anchored, rule)}}
for rule in TG.DIGIT_RULES])
# Draft snapshot and sentence IDs.
s5 = I["@s5"]
snap_anchor = AN.make_anchor(doc, vals, s5, rev, store=STORE)
@@ -1462,6 +1519,72 @@ def invalid(out, tiny):
case("style-sup-sub", "doc.run.style", {"nodes.jsonl": edit_nodes(lambda ns: by(ns, "@c").update(
content=[{"kind": "text", "text": "A second paragraph.", "style": ["sub", "sup"]}]))},
"A run cannot be both a subscript and a superscript.")
# Zipped packages that break spec/12 §12.2: the tiny package's files plus one bad entry.
tiny_dir = os.path.join(out, "fixtures", "packages", "tiny2026example")
entries = []
for rel in sorted(os.path.relpath(os.path.join(d, f), tiny_dir).replace(os.sep, "/")
for d, _, fs in os.walk(tiny_dir) for f in fs):
with open(os.path.join(tiny_dir, rel), "rb") as fh:
entries.append(("tiny2026example/" + rel, fh.read(), 0o644))
def zip_case(name, rule, note, extra):
d = os.path.join(base, name)
os.makedirs(d)
with zipfile.ZipFile(os.path.join(d, "package" + ZIP_EXTENSION), "w") as zf:
for ename, data, mode in entries + extra:
zi = zipfile.ZipInfo(ename, date_time=(1980, 1, 1, 0, 0, 0))
zi.compress_type = zipfile.ZIP_STORED
zi.create_system = 3
zi.external_attr = mode << 16
zf.writestr(zi, data)
with open(os.path.join(d, "expect.json"), "wb") as fh:
fh.write((jcs.dumps({"rule": rule, "note": note}) + "\n").encode())
zip_case("zip-traversal", "package.zip.entry", "An archive entry whose path leaves the package folder.",
[("tiny2026example/../escape.txt", b"outside\n", 0o644)])
zip_case("zip-symlink", "package.zip.entry", "An archive entry that is a symbolic link.",
[("tiny2026example/assets/link", b"document.json", 0o120777)])
# Rules the audit found no broken package for.
case("missing-required-file", "package.file.required", {"provenance.json": None}, "A required file is missing.")
case("counts-wrong", "doc.counts", {"document.json": _doc_edit(files, counts={"nodes": 1, "leaves": 1})},
"document.json's counts are not the number of nodes and leaves.")
case("rev-without-corrections", "doc.correction.rev", {"document.json": _doc_edit(files, rev=3)},
"document.json's rev is not the number of corrections.")
case("receipt-unsorted-nodes", "doc.receipt.nodes", {"receipts.jsonl": _receipt_edit(
files, nodes=list(reversed([json.loads(x) for x in files["receipts.jsonl"].decode().splitlines()][0]["nodes"])))},
"A receipt whose nodes are not sorted by ID.")
case("retired-live", "doc.id.retired-live", {"ids.jsonl": jsonl([
{"id": nm["@a"], "kind": "retired", "successor": None, "node_kind": "sentence", "rev": 1}])},
"An ID that is both live and retired.")
case("manifest-key", "package.manifest.key", {}, "The manifest names another key than document.json.",
raw=lambda d: _manifest_edit(d, key="other2026example"))
case("unsafe-path", "package.path.unsafe", {}, "The manifest lists a path that leaves the package.",
raw=lambda d: _manifest_edit(d, files=[{"path": "../escape.txt", "sha256": "0" * 64, "size": 1}]
+ json.load(open(os.path.join(d, "MANIFEST.json")))["files"]))
case("crlf", "package.file.canonical", {}, "A file whose lines end in CR LF, which is not the canonical form.",
raw=lambda d: _crlf(d))
signer = [json.loads(x) for x in files["receipts.jsonl"].decode().splitlines()][0]
case("signer-mismatch", "doc.receipt.signature", {"receipts.jsonl": jsonl([
RC.seal(dict(signer, by="cara"), secret("ana"), "ana")])},
"A receipt whose signature was made with another key than the signer it names.")
def _manifest_edit(d, **kw):
p = os.path.join(d, "MANIFEST.json")
m = json.load(open(p, encoding="utf-8"))
m.update(kw)
open(p, "w", encoding="utf-8").write(jcs.dumps(m) + "\n")
def _crlf(d):
p = os.path.join(d, "provenance.json")
data = open(p, "rb").read().replace(b"\n", b"\r\n")
open(p, "wb").write(data)
m = json.load(open(os.path.join(d, "MANIFEST.json"), encoding="utf-8"))
for f in m["files"]:
if f["path"] == "provenance.json":
f["sha256"], f["size"] = H(data), len(data)
open(os.path.join(d, "MANIFEST.json"), "w", encoding="utf-8").write(jcs.dumps(m) + "\n")
def _swap_first(ns, sec):
@@ -1533,6 +1656,9 @@ def main(out):
build_simple(out, "okafor2025open")
build_simple(out, "smithlab2027charges")
tiny = tiny_package(os.path.join(out, "fixtures", "packages", "tiny2026example"))
# The same package zipped as spec/12 §12.2 says a producer writes a byte-identical archive.
write_zip(os.path.join(out, "fixtures", "packages", "tiny2026example"),
os.path.join(out, "fixtures", "packages", "tiny2026example" + ZIP_EXTENSION))
records(out, hale)
vectors(out, hale, tiny)
invalid(out, tiny)
+214 -6
View File
@@ -15,6 +15,7 @@ import subprocess
import sys
import shutil
import tempfile
import zipfile
from fractions import Fraction
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -25,6 +26,7 @@ from tools.oracle import anchors as AN # noqa: E402
from tools.oracle import corrections as CO # noqa: E402
from tools.oracle import crypto, db, fingerprint as FP, grade as GR, hashing, jcs, receipts as RC, rights as RI # noqa: E402
from tools.oracle import package as PK # noqa: E402
from tools.oracle import targets as TG # noqa: E402
from tools.oracle.model import Doc # noqa: E402
from tools.oracle.notation import fold_math # noqa: E402
from tools.oracle.schema import Store, validate as sv # noqa: E402
@@ -134,6 +136,17 @@ def check_records():
if rec.get("signature"):
ok("records", RC.verify_signature(rec, roster), "%s: signature does not verify" % rel)
ok("records", rec["id"] == record_id(RC.record_hash(rec)), "%s: id does not derive" % rel)
ok("records", rec.get("by") == rec["signature"]["key"], "%s: by differs from the signing key" % rel)
ok("records", rec["format"] in RC.SIGNED_FORMATS, "%s: %s is not a signed format" % (rel, rec["format"]))
# A QA report is signed (spec/08 §8.9); a checklist is covered by its receipt at its hash (spec/08 §8.11).
ok("records", "signature" in js(os.path.join(R, "qa-report.json")), "the QA report is not signed")
cl = js(os.path.join(R, "checklist", "checklist.json"))
clr = js(os.path.join(R, "checklist", "receipt.json"))
cl_hash = sha256_hex(jcs.dumpb({k: v for k, v in cl.items() if k != "id"}))
ok("records", cl["id"] == record_id(cl_hash), "checklist id does not derive from its content hash")
ok("records", ["checklist", cl["id"], cl_hash] in clr["records"], "the checklist is not covered at its hash")
ok("records", RC.receipt_status(clr, [clr], roster) == "valid", "the checklist receipt is not valid")
ok("records", {i["result"] for i in cl["items"]} <= {"pass", "fail", "not-applicable"}, "checklist results")
# A project's settings.stages holds the id of a stage file.
stage_ids = {js(p("registry", "stages", f))["id"] for f in os.listdir(p("registry", "stages")) if f.endswith(".json")}
stages = js(os.path.join(R, "project.json")).get("settings", {}).get("stages")
@@ -157,6 +170,15 @@ def check_records():
ok("records", items.get(("annotation", a["id"].rsplit("/", 1)[1])) == sha256_hex(jcs.dumpb(body)),
"a confirmed annotation is not covered at its hash: " + a["id"])
ok("records", lr["root"] == crypto.merkle_root(crypto.receipt_items(lr)), "layer receipt root")
# Every figure of the main example has a confirmed description, which is what an accessible export needs.
anns = jl(os.path.join(R, "layer", "annotations.jsonl"))
described = {x["anchor"]["node"] for a in anns if a["body"]["kind"] == "description" and a["status"] == "confirmed"
for x in a["target"].get("selector", []) if x.get("type") == "AnchorSelector"}
figures = {n["id"] for n in jl(p("fixtures", "packages", "hale2026harmonic", "nodes.jsonl")) if n["kind"] == "figure"}
ok("records", figures and figures <= described, "a figure of the example has no confirmed description: %s"
% sorted(figures - described))
ok("records", any(a["body"]["kind"] == "generated" and a["status"] == "proposed" for a in anns),
"the layer has no proposed generated annotation to contrast with the confirmed description")
def state_of(pkg):
@@ -167,6 +189,8 @@ def check_packages():
team_key = crypto.team_key(bytes.fromhex(open(p("fixtures", "keys", "smithlab-team-secret.hex")).read().strip()))
pk = p("fixtures", "packages")
for d in sorted(os.listdir(pk)):
if not os.path.isdir(os.path.join(pk, d)):
continue # the zipped form is checked by check_zips
rep = PK.validate(os.path.join(pk, d), STORE, team_key if d.endswith("checks-only") else None)
ok("packages", bool(rep), "%s: %s" % (d, rep.errors[:4]))
# The checks-only export carries exactly the HMAC values of the full document.
@@ -197,6 +221,85 @@ def check_packages():
valid("packages", "mathml.schema.json", m, "derived/mathml.jsonl")
def _zip(path, entries):
"""Write a zip of (name, bytes, mode) entries the way a producer does (spec/12 §12.2)."""
with zipfile.ZipFile(path, "w") as zf:
for name, data, mode in entries:
zi = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
zi.compress_type = zipfile.ZIP_STORED
zi.create_system = 3
zi.external_attr = mode << 16
zf.writestr(zi, data)
def check_zips():
"""Zipped packages (spec/12 §12.2 and §12.8).
The fixture archive unpacks to the folder it was made from byte for byte, validates, keeps the producer's
byte-identical rules, and is reproduced by write_zip; crafted archives are refused with the rule named.
"""
pk = p("fixtures", "packages")
zips = [f for f in sorted(os.listdir(pk)) if f.endswith(PK.ZIP_EXTENSION)]
ok("zips", len(zips) >= 1, "no zipped package among the fixtures")
for z in zips:
key = z[:-len(PK.ZIP_EXTENSION)]
folder = os.path.join(pk, key)
rep = PK.validate_zip(os.path.join(pk, z), STORE)
ok("zips", bool(rep), "%s: %s" % (z, rep.errors[:4]))
with zipfile.ZipFile(os.path.join(pk, z)) as zf:
infos = zf.infolist()
names = [i.filename for i in infos]
ok("zips", names == sorted(names, key=lambda n: n.encode("utf-8")), "%s: entries are sorted by path" % z)
ok("zips", all(i.date_time == (1980, 1, 1, 0, 0, 0) and i.extra == b"" and not i.is_dir()
and i.compress_type in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED) for i in infos),
"%s: entries are dated 1980-01-01 00:00, carry no extra field and are stored or deflated" % z)
ok("zips", {n.partition("/")[0] for n in names} == {key}, "%s: one top-level folder named by the key" % z)
ok("zips", key + "/MANIFEST.json" in names, "%s: the manifest is an entry" % z)
with tempfile.TemporaryDirectory() as tmp:
root = PK.read_zip(os.path.join(pk, z), tmp)
ok("zips", _same_tree(folder, root), "%s: unpacks to its folder byte for byte" % z)
again = os.path.join(tmp, "again.zip")
PK.write_zip(root, again)
ok("zips", filecmp.cmp(again, os.path.join(pk, z), shallow=False), "%s: write_zip reproduces it" % z)
# Crafted archives, built from the tiny package.
tiny = p("fixtures", "packages", "tiny2026example")
entries = []
for rel in sorted(os.path.relpath(os.path.join(d, f), tiny).replace(os.sep, "/")
for d, _, fs in os.walk(tiny) for f in fs):
with open(os.path.join(tiny, rel), "rb") as fh:
entries.append(("tiny2026example/" + rel, fh.read(), 0o644))
cases = [
("two-folders", entries + [("other2026/x.txt", b"x", 0o644)], "package.zip.entry"),
("duplicate", entries + [entries[0]], "package.zip.entry"),
("folder-not-a-key", [("Tiny Example/" + n.partition("/")[2], d, m) for n, d, m in entries], "package.zip.entry"),
("folder-not-the-key", [("other2026example/" + n.partition("/")[2], d, m) for n, d, m in entries],
"package.zip.entry"),
("dot-dot", entries + [("tiny2026example/assets/../../x", b"x", 0o644)], "package.zip.entry"),
("symlink", entries + [("tiny2026example/assets/link", b"document.json", 0o120777)], "package.zip.entry"),
("no-folder", [(n.partition("/")[2], d, m) for n, d, m in entries], "package.zip.entry"),
("directory-entries", entries + [("tiny2026example/assets/", b"", 0o755)], None),
]
with tempfile.TemporaryDirectory() as tmp:
for name, ents, rule in cases:
path = os.path.join(tmp, name + ".zip")
with __import__("warnings").catch_warnings():
__import__("warnings").simplefilter("ignore")
_zip(path, ents)
rules = {e["rule"] for e in PK.validate_zip(path, STORE).errors}
ok("zips", (rule in rules) if rule else not rules, "%s: expected %s, got %s" % (name, rule, sorted(rules)))
ok("zips", {e["rule"] for e in PK.validate_zip(os.path.join(tmp, "directory-entries.zip"), STORE, limit=100).errors}
== {"package.zip.size"}, "an archive past the size limit is refused with package.zip.size")
with open(os.path.join(tmp, "not-a.zip"), "wb") as fh:
fh.write(b"PK\x03\x04 this is not an archive")
ok("zips", {e["rule"] for e in PK.validate_zip(os.path.join(tmp, "not-a.zip"), STORE).errors}
== {"package.zip.entry"}, "bytes that are not an archive are refused")
_zip(os.path.join(tmp, "empty.zip"), [])
ok("zips", {e["rule"] for e in PK.validate_zip(os.path.join(tmp, "empty.zip"), STORE).errors}
== {"package.zip.entry"}, "an empty archive is refused")
outside = [f for f in os.listdir(tmp) if not f.endswith(".zip")]
ok("zips", not outside, "reading archives wrote outside their folders: %s" % outside)
def check_owner_revocation():
"""A key the roster makes an owner may revoke anyone's receipt (spec/08 §8.7); any other key may not.
@@ -321,23 +424,38 @@ def check_gate():
ok("packages", rec["index"] == st.rev + 1 and st1.rev == st.rev + 1, "the gate's record has the next index")
ok("packages", json.dumps(st.nodes, sort_keys=True) == before, "the gate changed the state it was given")
valid("packages", "correction.schema.json", rec, "gate record")
crop = [{"kind": "crop", "sha256": "0" * 64}]
cell = next(n for n in st.nodes if n["kind"] == "cell")
section = next(n for n in st.nodes if n["kind"] == "section" and "label" in n)
header = next(c["header_path"][0] for n in st.nodes if n["kind"] == "table" for c in n["attrs"]["columns"])
for bad, rule in (({"op": "replace", "target": target["id"]}, "correction.schema"),
(dict(base, op="replace", target=target["id"], find="0.01", replace="0.02"), "correction.digits"),
(dict(base, op="insert", target=target["id"], run=9, offset=0, text="x",
evidence=[{"kind": "crop", "sha256": "0" * 64}]), "correction.insert")):
(dict(base, op="insert", target=target["id"], run=9, offset=0, text="x", evidence=crop),
"correction.insert"),
# Rule 11 of spec/11 §11.4: the result keeps to chapter 3, attribute shapes included.
(dict(base, op="set", target=cell["id"], attr="colspan", value="2"), "doc.node.attrs"),
(dict(base, op="set", target=cell["id"], attr="role", value="boss"), "doc.node.attrs"),
(dict(base, op="insert", parent=target["parent"], place_after=target["id"], evidence=crop,
node={"kind": "sentence", "content": [{"kind": "text", "text": "Junk."}], "foo": 1}),
"doc.node.schema"),
(dict(base, op="delete", target=header), "doc.table.header")):
try:
CO.gate(st, bad, STORE)
got = None
except Refusal as e:
got = e.rule
ok("packages", got == rule, "the gate gave %s for a proposal that breaks %s" % (got, rule))
# A relabel to null removes the label instead of writing null (spec/03 §3.1).
st2, _ = CO.gate(st, dict(base, op="relabel", target=section["id"], label=None, evidence=crop), STORE)
ok("packages", "label" not in next(n for n in st2.nodes if n["id"] == section["id"]), "relabel to null drops the label")
def check_invalid():
base = p("fixtures", "invalid")
for d in sorted(os.listdir(base)):
expect = js(os.path.join(base, d, "expect.json"))["rule"]
rep = PK.validate(os.path.join(base, d, "package"), STORE)
zipped = os.path.join(base, d, "package" + PK.ZIP_EXTENSION)
rep = PK.validate_zip(zipped, STORE) if os.path.exists(zipped) else PK.validate(os.path.join(base, d, "package"), STORE)
rules = {e["rule"] for e in rep.errors}
ok("invalid", expect in rules, "%s: expected %s, got %s" % (d, expect, sorted(rules) or "no error"))
@@ -438,7 +556,7 @@ def check_vectors():
st = CO.State(c["doc"], [{k: v for k, v in n.items() if k not in (
"text_hash", "match_hash", "align_key", "own_hash", "node_hash")} for n in c["before"]["nodes"]])
try:
st1, rec = CO.apply(st, c["correction"])
st1, rec = CO.apply(st, c["correction"], store=STORE)
v1 = hashing.compute(st1.doc())
ok("vectors", st1.index().values == reference_values(st1.doc()),
"correction %s: incremental values differ from the full computation" % c["id"])
@@ -467,6 +585,17 @@ def check_vectors():
for c in V("grading.jsonl"):
st = states.setdefault(c["package"], state_of(p(c["package"])))
ok("vectors", GR.grade(c["exercise"], c["submission"], st) == c["expect"], "grading " + c["id"])
sets = {}
for c in V("targets.jsonl"):
st = states.setdefault(c["package"], state_of(p(c["package"])))
got = TG.target_set(st.doc(), c["anchored"], c["digit_rule"])
ok("vectors", got == c["expect"]["nodes"], "targets " + c["id"])
sets[c["digit_rule"]] = {n for n, _ in got}
d = st.doc()
ok("vectors", all(d.nodes[n]["kind"] in TG.MATH_KINDS for n, r in got if r == "equation")
and all(d.nodes[n]["kind"] in TG.LEAVES for n, _ in got), "targets %s: only leaves, equations by kind" % c["id"])
ok("vectors", set(sets) == set(TG.DIGIT_RULES) and sets["none"] <= sets["anchored-sections"] <= sets["all"]
and sets["none"] < sets["all"], "the three digit rules nest")
def check_determinism():
@@ -532,6 +661,8 @@ def check_sql():
assets = os.path.join(tmp, "assets")
pk = p("fixtures", "packages")
for d in sorted(os.listdir(pk)):
if not os.path.isdir(os.path.join(pk, d)):
continue
key = db.load(conn, os.path.join(pk, d), assets) if not d.endswith("checks-only") else None
if key is None:
# A second load of the same key goes to its own database.
@@ -701,11 +832,86 @@ def check_links():
ok("links", frag in anchors.get(dest, ()), "%s:%d: no heading #%s in %s" % (rel, i, frag, file or "this file"))
WORD_NUMBERS = {w: i for i, w in enumerate(
"zero one two three four five six seven eight nine ten eleven twelve thirteen fourteen fifteen sixteen "
"seventeen eighteen nineteen twenty".split())}
WORD_NUMBERS.update({"twenty-" + w: 20 + i for i, w in enumerate("one two three four five six seven eight nine".split(), 1)})
# A count: digits or a number word, not the tail of a section number such as "22.3".
NUM = r"(?<![\d.])\b(\d+|[a-z]+(?:-[a-z]+)?)"
# What the prose may say about the repository, and the count it must equal. The text is lower-cased first.
COUNT_PATTERNS = [
(r"\b(?:of the|lists|defines|has) " + NUM + r" formats\b", "formats"),
(NUM + r" (?:formats )?(?:are |of them |of the \d+ formats are )?`?candidate`?\b", "candidate"),
(NUM + r" (?:formats )?(?:are )?`?provisional`?\b", "provisional"),
(NUM + r" (?:json (?:\([^)]*\) )?)?schema(?:s| files)\b", "schemas"),
(NUM + r" (?:files of test vectors|vector files)\b", "vectors"),
(r"\ball " + NUM + r" files\b", "vectors"),
(NUM + r" (?:are )?(?:deliberately broken|broken packages|invalid ones|broken ones|invalid packages)\b", "invalid"),
(NUM + r" valid (?:example |fixture )?packages\b", "valid"),
(NUM + r" chapters\b", "chapters"),
(NUM + r" cases in all\b", "cases"),
]
# Where the prose names the specification's own version; other standards' versions are not checked.
VERSION_PATTERN = r"\b(?:manuspec|the formats are|the specification is at|this is|it is|is at|at) version (\d+\.\d+\.\d+)\b|\bmanuspec (\d+\.\d+\.\d+)\b"
def check_counts():
"""Every count and version the prose states about the repository equals what the repository holds."""
formats = js(p("registry", "formats.json"))
rules = js(p("registry", "rules.json"))
vec = sorted(f for f in os.listdir(p("vectors")) if f.endswith(".jsonl"))
have = {
"formats": len(formats["formats"]),
"candidate": sum(f["status"] == "candidate" for f in formats["formats"]),
"provisional": sum(f["status"] == "provisional" for f in formats["formats"]),
"schemas": len(SCHEMA_FILES),
"vectors": len(vec),
"cases": sum(len(jl(p("vectors", f))) for f in vec),
"invalid": len(os.listdir(p("fixtures", "invalid"))),
"valid": sum(os.path.isdir(os.path.join(p("fixtures", "packages"), d)) for d in os.listdir(p("fixtures", "packages"))),
"chapters": len([f for f in os.listdir(p("spec")) if re.match(r"\d\d-.*\.md$", f)]),
}
top = re.search(r"^## (\S+)", open(p("CHANGELOG.md"), encoding="utf-8").read(), re.M).group(1)
pixi = re.search(r'^version = "([^"]+)"', open(p("pixi.toml"), encoding="utf-8").read(), re.M).group(1)
from tools.build import VERSION
ok("counts", top == pixi == formats["version"] == rules["version"] == VERSION,
"versions differ: CHANGELOG %s, pixi.toml %s, formats %s, rules %s, build.py %s"
% (top, pixi, formats["version"], rules["version"], VERSION))
mds = []
for d, dirs, fs in os.walk(ROOT):
dirs[:] = [x for x in dirs if x not in SKIP_DIRS]
mds += [os.path.join(d, f) for f in fs if f.endswith(".md")]
for path in sorted(mds):
rel = os.path.relpath(path, ROOT)
fence = False
for i, ln in enumerate(open(path, encoding="utf-8"), 1):
if ln.startswith("```"):
fence = not fence
continue
if fence:
continue
text = ln.lower()
for m in re.finditer(VERSION_PATTERN, text):
v = m.group(1) or m.group(2)
ok("counts", v == top, "%s:%d: version %s is not %s" % (rel, i, v, top))
for pattern, what in COUNT_PATTERNS:
for m in re.finditer(pattern, text):
word = m.group(1)
if word.isdigit():
n = int(word)
elif word in WORD_NUMBERS:
n = WORD_NUMBERS[word]
else:
continue
ok("counts", n == have[what], "%s:%d: says %s %s, the repository has %d"
% (rel, i, word, what, have[what]))
def main():
import argparse
ap = argparse.ArgumentParser(
description="Check the whole repository: schemas, records, packages, broken packages, vectors, a "
"byte-for-byte rebuild, the SQL round trip and the bundle, prose and links. "
description="Check the whole repository: schemas, records, packages, zipped packages, broken packages, "
"vectors, a byte-for-byte rebuild, the SQL round trip and the bundle, counts, prose and links. "
"Exit status 0 means every check passed.")
ap.add_argument("--quick", action="store_true", help="skip the byte-for-byte rebuild, the slowest check")
args = ap.parse_args()
@@ -713,6 +919,7 @@ def main():
check_registry()
check_records()
check_packages()
check_zips()
check_owner_revocation()
check_unsafe_paths()
check_incremental()
@@ -722,6 +929,7 @@ def main():
if not args.quick:
check_determinism()
check_sql()
check_counts()
check_prose()
check_links()
for k in sorted(COUNTS):
+41 -26
View File
@@ -4,6 +4,7 @@
Non-normative tooling.
Usage: python3 tools/load.py --sqlite lab.db PACKAGE [PACKAGE ...]
python3 tools/load.py --sqlite lab.db --bundle --built-from REV PACKAGE [PACKAGE ...]
A package is a folder, or a zipped package (`.sct.zip`), unpacked under the rules of spec/12 §12.2 first.
With --bundle the new file is a scienting-bundle/1: it also gets the tables of sql/bundle-sqlite.sql,
the full-text search index and bundle_meta (spec/13 §13.5).
Each package is validated first, and nothing is loaded if any package fails.
@@ -15,6 +16,7 @@ import argparse
import os
import sqlite3
import sys
import tempfile
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
@@ -29,37 +31,50 @@ def main():
ap.add_argument("--bundle", action="store_true", help="build a new scienting-bundle/1 file with search")
ap.add_argument("--built-from", help="with --bundle: the store revision the packages come from")
ap.add_argument("--store", help="with --bundle: the store's name (default: the packages' common store)")
ap.add_argument("packages", nargs="+", help="package folders")
ap.add_argument("--max-size", metavar="BYTES", type=int, default=package.ZIP_LIMIT,
help="refuse a zipped package that expands past this many bytes (default %d)" % package.ZIP_LIMIT)
ap.add_argument("packages", nargs="+", help="package folders or zipped packages")
a = ap.parse_args()
if a.bundle and (os.path.exists(a.sqlite) or not a.built_from):
ap.error("--bundle needs --built-from and a database file that does not exist yet")
store, bad = Store(os.path.join(ROOT, "schemas")), False
for p in a.packages:
for e in package.validate(p, store).errors:
print("%s: %s: %s: %s" % (p, e["where"], e["rule"], e["message"]), file=sys.stderr)
bad = True
if bad:
sys.exit("Nothing was loaded.")
new = not os.path.exists(a.sqlite)
conn = sqlite3.connect(a.sqlite)
if a.bundle:
db.create_bundle(conn, os.path.join(ROOT, "sql"))
elif new:
db.create(conn, os.path.join(ROOT, "sql"))
for p in a.packages:
key = db.load(conn, p, a.sqlite + ".assets")
n = conn.execute("SELECT COUNT(*) FROM nodes WHERE doc_key = ?", (key,)).fetchone()[0]
with tempfile.TemporaryDirectory() as tmp:
folders = []
for i, p in enumerate(a.packages):
if package.is_zip(p):
try:
folders.append(package.read_zip(p, os.path.join(tmp, str(i)), a.max_size))
except package.Refusal as e:
print("%s: %s: %s" % (p, e.rule, e.message), file=sys.stderr)
bad = True
continue
else:
folders.append(p)
for e in package.validate(folders[-1], store).errors:
print("%s: %s: %s: %s" % (p, e["where"], e["rule"], e["message"]), file=sys.stderr)
bad = True
if bad:
sys.exit("Nothing was loaded.")
new = not os.path.exists(a.sqlite)
conn = sqlite3.connect(a.sqlite)
if a.bundle:
print("%s: %d nodes, %d search rows" % (key, n, db.index(conn, key)))
else:
print("%s: %d nodes" % (key, n))
if a.bundle:
stores = {r[0] for r in conn.execute("SELECT DISTINCT store FROM documents WHERE store IS NOT NULL")}
store = a.store or (stores.pop() if len(stores) == 1 else None)
if store is None:
sys.exit("The packages name no single store; give one with --store.")
db.write_meta(conn, store, a.built_from)
conn.commit()
db.create_bundle(conn, os.path.join(ROOT, "sql"))
elif new:
db.create(conn, os.path.join(ROOT, "sql"))
for p in folders:
key = db.load(conn, p, a.sqlite + ".assets")
n = conn.execute("SELECT COUNT(*) FROM nodes WHERE doc_key = ?", (key,)).fetchone()[0]
if a.bundle:
print("%s: %d nodes, %d search rows" % (key, n, db.index(conn, key)))
else:
print("%s: %d nodes" % (key, n))
if a.bundle:
stores = {r[0] for r in conn.execute("SELECT DISTINCT store FROM documents WHERE store IS NOT NULL")}
store_name = a.store or (stores.pop() if len(stores) == 1 else None)
if store_name is None:
sys.exit("The packages name no single store; give one with --store.")
db.write_meta(conn, store_name, a.built_from)
conn.commit()
if __name__ == "__main__":
+12 -5
View File
@@ -158,9 +158,12 @@ def _follow(state, nid):
def _relabel_reverted_own(state, doc, nid, anchor):
"""The own_hash the node had before the relabel corrections since the anchor, or None."""
rel = [c for c in state.corrections if c["index"] > anchor["rev"] and c.get("target") == nid]
if not rel or any(c["op"] != "relabel" for c in rel):
"""The own_hash the node had before the relabel corrections since the anchor, or None when anything else
changed the node or what is below it since (spec/06 §6.5: its *only* changes were relabel corrections)."""
since = [c for c in state.corrections if c["index"] > anchor["rev"] and (
c.get("target") == nid or nid in (c.get("path") or []) or nid in (c.get("to_path") or []))]
rel = [c for c in since if c.get("target") == nid]
if not rel or any(c["op"] != "relabel" or c.get("target") != nid for c in since):
return None
n = dict(doc.nodes[nid], label=rel[0]["from"])
tmp = Doc(doc.key, [n if x["id"] == nid else x for x in [doc.nodes[i] for i in doc.order]])
@@ -175,7 +178,11 @@ def resolve(state, anchor, threshold=Fraction(1, 4)):
original = anchor["node"]
nid, chain, via = _follow(state, original)
live_kind = doc.nodes[nid]["kind"] if nid else None
kind = kind or live_kind or next((e["node_kind"] for e in state.ids if e["id"] == original), None)
table = {e["id"]: e for e in state.ids}
# The kind falls back to the retired entry the chain reached (spec/06 §6.5); an alias carries no kind.
retired_kind = next((table[x].get("node_kind") for x in reversed(chain)
if table.get(x, {}).get("kind") == "retired"), None)
kind = kind or live_kind or retired_kind
cls = anchor_class(kind) if kind else "leaf"
is_span = "span" in anchor
result = {"node": nid}
@@ -256,7 +263,7 @@ def resolve(state, anchor, threshold=Fraction(1, 4)):
now = set(doc.leaves(nid))
now_match = {values[x]["match_hash"] for x in now}
held = sum(1 for i, m in orig if i in now or m in now_match)
if orig and 2 * held >= len(orig):
if 2 * held >= len(orig):
return out("moved")
return out("stale", reason="the successor holds fewer than half of the original leaves")
if values[nid]["node_hash"] == anchor["node_hash"]:
+59 -21
View File
@@ -6,8 +6,8 @@ Non-normative; see spec/11-corrections.md.
import copy
from . import hashing
from .model import LABELLED, LEAVES, MATH_KINDS, RUN_KINDS, VERBATIM_KINDS, Doc, CHILDREN
from .tree import Refusal, canonical_runs, mint_ids, validate_structure, _flatten
from .model import ATTRS, LABELLED, LEAVES, MATH_KINDS, RUN_KINDS, VERBATIM_KINDS, Doc, CHILDREN
from .tree import Refusal, canonical_runs, mint_ids, validate_references, validate_structure, _flatten
from . import jcs
from .schema import validate as schema_validate
from .text import digits, record_id, sha256_hex
@@ -137,15 +137,33 @@ def gate(state, proposal, store, hasher=hashing.PLAIN):
errors = schema_validate(store, "correction.schema.json", proposal)
if errors:
raise Refusal("correction.schema", "; ".join(errors[:3]))
return apply(state, proposal, hasher)
return apply(state, proposal, hasher, store)
def _is_index(v):
return isinstance(v, int) and not isinstance(v, bool) and v >= 0
def apply(state, c, hasher=hashing.PLAIN):
"""Apply correction `c` to `state` and return the recorded correction. Raises Refusal."""
def id_order(e):
"""The order of ids.jsonl (spec/04 §4.4): aliases first by ID, then retirements by revision and ID."""
return (0 if e["kind"] == "alias" else 1, e.get("rev", 0), e["id"])
def _node_rule(error):
"""The rule a node-schema error breaks, from the property the error names."""
path = error.split(":", 1)[0]
for prefix, rule in (("/attrs", "doc.node.attrs"), ("/label", "doc.node.label"), ("/content", "doc.node.content")):
if path.startswith(prefix):
return rule
return "doc.node.schema"
def apply(state, c, hasher=hashing.PLAIN, store=None):
"""Apply correction `c` to `state` and return the recorded correction. Raises Refusal.
With a schema `store`, every node the correction changed or created is also validated against
node.schema.json, which is how the gate enforces the attribute shapes of spec/03 §3.4 (rule 11 of §11.4).
"""
c = copy.deepcopy(c)
doc = state.doc()
before_ix = state.index(hasher)
@@ -308,7 +326,7 @@ def apply(state, c, hasher=hashing.PLAIN):
elif op == "move":
parent = c["parent"]
if parent not in by_id or parent in set(_subtree(doc, target)):
if parent not in by_id or parent in set(_subtree(doc, target)) or c.get("place_after") == target:
raise Refusal("correction.move", "bad destination", target)
_place(nodes, target, parent, c.get("place_after"))
nodes = _rebuild(nodes)
@@ -326,6 +344,12 @@ def apply(state, c, hasher=hashing.PLAIN):
node["kind"] = new_kind
if new_kind not in LABELLED:
node.pop("label", None)
# Each kind carries only its own attributes (spec/03 §3.4).
kept = {k: v for k, v in node.get("attrs", {}).items() if k in ATTRS.get(new_kind, ())}
if kept:
node["attrs"] = kept
else:
node.pop("attrs", None)
record["from"] = old
touched = [x for x in doc.leaves(target)]
@@ -334,7 +358,10 @@ def apply(state, c, hasher=hashing.PLAIN):
if node["kind"] not in LABELLED:
raise Refusal("correction.relabel", "%s takes no label" % node["kind"], target)
record["from"] = node.get("label")
node["label"] = c["label"]
if c.get("label") is None:
node.pop("label", None) # optional properties are omitted, never written as null (spec/03 §3.1)
else:
node["label"] = c["label"]
touched = [target] if node["kind"] in LEAVES else []
elif op == "set":
@@ -353,8 +380,15 @@ def apply(state, c, hasher=hashing.PLAIN):
after_doc = Doc(state.key, nodes)
validate_structure(after_doc)
validate_references(after_doc)
after_ix = before_ix.update(after_doc)
after_vals = after_ix.values
if store is not None:
changed = {x for x in set(touched) | set(created) | {target} if x in after_doc.nodes}
for nid in sorted(changed):
errors = schema_validate(store, "node.schema.json", dict(after_doc.nodes[nid], **after_vals[nid]))
if errors:
raise Refusal(_node_rule(errors[0]), errors[0], nid)
# Digits: a change to a digit needs crop evidence, and no leaf outside `touched` may change its digits.
needs_crop = False
@@ -393,24 +427,28 @@ def apply(state, c, hasher=hashing.PLAIN):
for rid, succ in retired:
new_ids.append({"id": rid, "kind": "retired", "successor": succ, "node_kind": old_kinds[rid],
"rev": record["index"]})
new_ids.sort(key=id_order)
new_state = State(state.key, [after_doc.nodes[x] for x in after_doc.preorder()], new_ids,
state.corrections + [record], {hasher.key: after_ix})
return new_state, record
def _strip_edge(runs, end):
"""Remove the whitespace at one edge of a split part, across as many text runs as it spans (spec/11 §11.1)."""
from .text import is_ws
if not runs:
return
i = len(runs) - 1 if end else 0
r = runs[i]
if r["kind"] != "text":
return
t = r["text"]
if end:
while t and is_ws(t[-1]):
t = t[:-1]
else:
while t and is_ws(t[0]):
t = t[1:]
runs[i] = dict(r, text=t)
while runs:
i = len(runs) - 1 if end else 0
r = runs[i]
if r["kind"] != "text":
return
t = r["text"]
if end:
while t and is_ws(t[-1]):
t = t[:-1]
else:
while t and is_ws(t[0]):
t = t[1:]
if t:
runs[i] = dict(r, text=t)
return
del runs[i]
+2 -1
View File
@@ -10,7 +10,7 @@ import shutil
from . import hashing, jcs
from .model import Doc
from .package import SHA256_RE, read_json, read_jsonl, safe_path, write
from .package import SHA256_RE, _all_files, read_json, read_jsonl, safe_path, write
from .text import norm, sha256_hex
from .tree import Refusal
@@ -107,6 +107,7 @@ def load(conn, root, asset_store):
Validate the package first (package.validate); this only refuses paths and hashes that are unsafe to use.
"""
_all_files(root) # refuses a symbolic link anywhere in the package before anything is copied
manifest = read_json(os.path.join(root, "MANIFEST.json"))
listed = {f["path"]: f for f in manifest["files"]}
for path, f in listed.items():
+7 -2
View File
@@ -34,11 +34,16 @@ def is_stamp(line, size, rules):
return None
def fp_text(text):
s = unicodedata.normalize("NFKC", fold_plain_text(text))
def _strip(s):
return "".join(c for c in s if not is_ws(c) and unicodedata.category(c) not in ("Cc", "Cf"))
def fp_text(text):
"""Steps 3 and 4 of spec/07 §7.1: whitespace and control characters go first, so that "10 -20" and "10-20"
fold alike; then scripts and dashes fold as in the strict projection, then NFKC."""
return _strip(unicodedata.normalize("NFKC", fold_plain_text(_strip(text))))
def fingerprint(page, rules):
"""`page`: {"size": [w, h], "lines": [{"text", "box"?}], "images": [sha256...]}.
+18 -15
View File
@@ -3,18 +3,21 @@
Non-normative; see spec/19-exercises.md.
"""
from decimal import Decimal
from fractions import Fraction
from . import anchors as A
from .tree import Refusal
def dec(fr):
fr = Fraction(fr)
if fr.denominator == 1:
return str(fr.numerator)
s = "%.4f" % float(fr)
return s.rstrip("0").rstrip(".")
def dec(d):
"""A decimal string, exactly as the sum came out (spec/19 §19.4: points are decimal strings, summed exactly)."""
return format(Decimal(d), "f")
def _zero_like(d):
"""Zero with the scale of `d`, so a missed element worth "0.5" scores "0.0"."""
return Decimal(0).quantize(Decimal(d))
def _resolved(state, anchor):
@@ -57,34 +60,34 @@ def grade(body, submission, state):
items = []
def item(name, pts, got, fb):
items.append({"element": name, "points": dec(pts if got else 0), "max": dec(pts), "feedback": fb})
items.append({"element": name, "points": dec(pts if got else _zero_like(pts)), "max": dec(pts), "feedback": fb})
if kind in ("anchor", "check"):
if kind == "check":
v = key["verdict"]
ok = ans.get("verdict") == v["value"]
item("verdict", Fraction(v["points"]), ok, v.get("feedback", {}).get("hit" if ok else "miss", ""))
item("verdict", v["points"], ok, v.get("feedback", {}).get("hit" if ok else "miss", ""))
for i, el in enumerate(key["elements"]):
ok = _any(state, rule, ans.get("anchors", []), el["anchors"])
item("element:%d" % i, Fraction(el["points"]), ok, el.get("feedback", {}).get("hit" if ok else "miss", ""))
item("element:%d" % i, el["points"], ok, el.get("feedback", {}).get("hit" if ok else "miss", ""))
elif kind == "trace":
chain = ans.get("chain", [])
for i, link in enumerate(key["chain"]):
ok = i < len(chain) and _any(state, rule, chain[i], link["anchors"])
item("link:%d" % i, Fraction(link["points"]), ok, link.get("feedback", {}).get("hit" if ok else "miss", ""))
item("link:%d" % i, link["points"], ok, link.get("feedback", {}).get("hit" if ok else "miss", ""))
elif kind == "reproduce":
ok = abs(Fraction(ans["value"]) - Fraction(key["value"])) <= Fraction(key["tolerance"])
item("value", Fraction(key["points"]), ok, key.get("feedback", {}).get("hit" if ok else "miss", ""))
ok = abs(Decimal(ans["value"]) - Decimal(key["value"])) <= Decimal(key["tolerance"])
item("value", key["points"], ok, key.get("feedback", {}).get("hit" if ok else "miss", ""))
elif kind == "verify":
flags = ans.get("flags", [])
for i, err in enumerate(key["errors"]):
hit = [f for f in flags if _any(state, rule, [f["anchor"]], err["anchors"])]
ok = any(f["verdict"] == err["verdict"] for f in hit)
fb = err.get("feedback", {}).get("hit" if ok else "miss", "")
item("error:%d" % i, Fraction(err["points"]), ok, fb)
item("error:%d" % i, err["points"], ok, fb)
else:
raise Refusal("grade.kind", "unknown exercise kind " + kind)
score = sum(Fraction(x["points"]) for x in items)
total = sum(Fraction(x["max"]) for x in items)
score = sum((Decimal(x["points"]) for x in items), Decimal(0))
total = sum((Decimal(x["max"]) for x in items), Decimal(0))
return {"format": "scienting-grade/1", "submission": submission["id"], "exercise": submission["exercise"],
"score": dec(score), "max": dec(total), "items": items}
+12
View File
@@ -69,6 +69,18 @@ LABELLED = frozenset(
"code", "algorithm"]
)
# The attributes each kind may carry (spec/03 §3.4); kinds absent here carry none.
ATTRS = {
"section": frozenset(["level"]),
"table": frozenset(["columns"]),
"cell": frozenset(["column", "rowspan", "colspan", "role"]),
"row": frozenset(["header_path"]),
"list": frozenset(["ordered"]),
"item": frozenset(["marker"]),
"theorem": frozenset(["variant"]),
"panel": frozenset(["image", "crop", "figure_text"]),
}
RUNS = frozenset(["text", "math", "cite", "xref", "fnmark", "link", "equation"])
STYLES = frozenset(["em", "strong", "sup", "sub", "sc", "mono"])
+11 -6
View File
@@ -79,14 +79,19 @@ class NotSimple(Exception):
pass
# The Latin and Greek letter blocks of spec/05 \u00a75.4: Basic Latin through Latin Extended-B, Greek and Coptic without
# the Coptic letters, Latin Extended Additional, and Greek Extended.
LETTER_RANGES = ((0x0041, 0x024F), (0x0370, 0x03E1), (0x03F0, 0x03FF), (0x1E00, 0x1EFF), (0x1F00, 0x1FFF))
def is_letter(ch):
"""A Latin or Greek letter, or the micro sign."""
"""A letter of General Category L in the Latin or Greek blocks, or the micro sign."""
if ch == "\u00b5":
return True
if not unicodedata.category(ch).startswith("L"):
return False
name = unicodedata.name(ch, "")
return name.startswith("LATIN ") or name.startswith("GREEK ")
cp = ord(ch)
return any(lo <= cp <= hi for lo, hi in LETTER_RANGES)
def fold_char(ch):
@@ -222,9 +227,9 @@ class _TexParser:
elif kind == "ch" and val == "{":
inner = self.seq(True)
nxt = self.peek()
if nxt in (("ch", "_"), ("ch", "^")) and len(inner) != 1:
if nxt in (("ch", "_"), ("ch", "^")) and len(inner) > 1:
raise NotSimple("script on a multi-atom group")
atoms.extend(inner)
atoms.extend(inner) # an empty group leaves a following script to the previous atom or an empty base
else:
atoms.extend(self.command_or_char(kind, val))
@@ -245,7 +250,7 @@ class _TexParser:
return self.arg()
if val in ("left", "right"):
tok = self.take()
if tok in (("ch", "("), ("ch", ")")):
if tok == ("ch", "(" if val == "left" else ")"):
return [Atom(tok[1])]
raise NotSimple("\\left or \\right with another delimiter")
raise NotSimple("command \\" + val)
+260 -62
View File
@@ -4,17 +4,26 @@ Non-normative; see spec/12-package.md and spec/22-conformance.md.
"""
import base64
import binascii
import hashlib
import json
import os
import posixpath
import re
import stat
import tempfile
import unicodedata
import zipfile
from . import crypto, hashing, jcs, receipts as R
from .anchors import ADDRESS_RE
from .model import LEAVES, MATH_KINDS, RUN_KINDS, VERBATIM_KINDS, Doc
from .schema import validate as schema_validate
from .text import letter_first, record_id, sha256_hex
from .tree import Refusal, validate_structure
from .text import KEY_RE, letter_first, record_id, sha256_hex
from .tree import Refusal, validate_references, validate_structure
# The reference tools' own limit on a package's total size (spec/12 §12.2 leaves the limit to the consumer).
SIZE_LIMIT = 1 << 30
JSONL = {
"nodes.jsonl": "node.schema.json",
@@ -85,26 +94,46 @@ def write(root, files, absent=()):
return sha256_hex(mb)
def _refuse_constant(name):
raise ValueError("JSON has no canonical form for " + name)
def _parse_int(text):
"""Read a JSON integer as an ECMAScript engine would: exactly within 2^53, as a double beyond (spec/01 §1.1)."""
v = int(text)
return v if abs(v) <= 2**53 else float(v)
def _canonical_bytes(data, where):
"""Parse one canonical JSON value given as bytes; refuse anything else (spec/01 §1.1, §1.9).
Numbers are read as IEEE 754 doubles, as the canonical form writes them, and the comparison is on bytes, so a
carriage return, a byte order mark, a lone surrogate, NaN or invalid UTF-8 is refused, not silently translated.
"""
try:
rec = json.loads(data.decode("utf-8"), parse_constant=_refuse_constant, parse_int=_parse_int)
canon = jcs.dumps(rec).encode("utf-8")
except (UnicodeDecodeError, ValueError, TypeError, OverflowError, RecursionError) as e:
raise Refusal("package.file.canonical", "%s: not canonical JSON (%s)" % (where, type(e).__name__))
if canon != data:
raise Refusal("package.file.canonical", "%s: not canonical JSON" % where)
return rec
def read_jsonl(path):
out = []
with open(path, encoding="utf-8") as f:
for i, ln in enumerate(f, 1):
if not ln.endswith("\n"):
raise Refusal("package.file.newline", "%s:%d: every line ends with a newline" % (path, i))
rec = json.loads(ln)
if line(rec) != ln:
raise Refusal("package.file.canonical", "%s:%d: line is not canonical JSON" % (path, i))
out.append(rec)
return out
with open(path, "rb") as f:
data = f.read()
if data and not data.endswith(b"\n"):
raise Refusal("package.file.newline", "%s: every line ends with a newline" % path)
return [_canonical_bytes(ln, "%s:%d" % (path, i)) for i, ln in enumerate(data.split(b"\n")[:-1], 1)]
def read_json(path):
with open(path, encoding="utf-8") as f:
text = f.read()
rec = json.loads(text)
if line(rec) != text:
raise Refusal("package.file.canonical", "%s: file is not canonical JSON" % path)
return rec
with open(path, "rb") as f:
data = f.read()
if not data.endswith(b"\n"):
raise Refusal("package.file.canonical", "%s: the file does not end with one newline" % path)
return _canonical_bytes(data[:-1], path)
class Report:
@@ -119,55 +148,84 @@ class Report:
def _all_files(root):
"""Every file under `root` by its exact name, refusing any symbolic link on the way (spec/12 §12.2)."""
out = []
for d, _, fs in os.walk(root):
for d, dirs, fs in os.walk(root):
for name in dirs + fs:
if os.path.islink(os.path.join(d, name)):
rel = os.path.relpath(os.path.join(d, name), root).replace(os.sep, "/")
raise Refusal("package.path.unsafe", "%s: symbolic links are refused" % rel)
for f in fs:
out.append(os.path.relpath(os.path.join(d, f), root).replace(os.sep, "/"))
return sorted(out)
def validate(root, store, key=None):
"""Validate the package at `root`. `store` is a schema store; `key` a hashing key for keyed packages."""
def _hash_file(full, budget):
"""SHA-256 and size of a file read in chunks, refusing a read past `budget` bytes."""
h, n = hashlib.sha256(), 0
with open(full, "rb") as fh:
while True:
chunk = fh.read(1 << 20)
if not chunk:
break
n += len(chunk)
if n > budget:
raise Refusal("package.size", "the package's files exceed the limit of %d bytes" % SIZE_LIMIT)
h.update(chunk)
return h.hexdigest(), n
def validate(root, store, key=None, limit=SIZE_LIMIT):
"""Validate the package at `root`. `store` is a schema store; `key` a hashing key for keyed packages;
`limit` the most bytes the validator reads before refusing."""
rep = Report()
try:
_validate(root, store, key, rep)
_validate(root, store, key, rep, limit)
except Refusal as e:
rep.add(e.rule, e.node or "", e.message)
return rep
def _validate(root, store, key, rep):
def _validate(root, store, key, rep, limit=SIZE_LIMIT):
mpath = os.path.join(root, "MANIFEST.json")
if not os.path.exists(mpath):
if not os.path.lexists(mpath):
raise Refusal("package.manifest.missing", "MANIFEST.json is missing")
actual = set(_all_files(root))
manifest = read_json(mpath)
for e in schema_validate(store, "manifest.schema.json", manifest):
rep.add("package.manifest.schema", "MANIFEST.json", e)
if not rep:
return
listed = {}
paths = [f["path"] for f in manifest.get("files", [])]
paths = [f["path"] for f in manifest["files"]]
if paths != sorted(paths, key=lambda p: p.encode("utf-8")) or len(set(paths)) != len(paths):
rep.add("package.manifest.order", "MANIFEST.json", "files must be sorted by path and unique")
for f in manifest.get("files", []):
folded = {}
budget = limit
for f in manifest["files"]:
p = f["path"]
listed[p] = f
if not safe_path(p):
if not safe_path(p) or p == "MANIFEST.json":
rep.add("package.path.unsafe", p, "unsafe path")
continue
full = os.path.join(root, p)
fold = unicodedata.normalize("NFC", p).casefold()
if fold in folded:
rep.add("package.manifest.order", p, "collides with %s under case folding" % folded[fold])
folded[fold] = p
if f.get("present", True) is False:
if not p.startswith("assets/"):
rep.add("package.manifest.absent", p, "only assets may be listed without their bytes")
if os.path.exists(full):
if p in actual:
rep.add("package.manifest.absent", p, "a file listed as absent is present")
continue
if not os.path.exists(full):
if p not in actual:
rep.add("package.manifest.missing-file", p, "listed file is missing")
continue
with open(full, "rb") as fh:
data = fh.read()
if sha256_hex(data) != f["sha256"] or len(data) != f["size"]:
sha, size = _hash_file(os.path.join(root, p), budget)
budget -= size
if sha != f["sha256"] or size != f["size"]:
rep.add("package.manifest.hash", p, "SHA-256 or size does not match the manifest")
for p in _all_files(root):
for p in sorted(actual):
if p != "MANIFEST.json" and p not in listed:
rep.add("package.manifest.unlisted", p, "file is not listed in the manifest")
for p in REQUIRED:
@@ -223,28 +281,15 @@ def _validate(root, store, key, rep):
if scheme == "hmac-team" and key is not None and document["hashing"]["key_id"] != crypto.key_id(key):
rep.add("doc.hash.key", "document.json", "key_id does not match the key supplied")
# References inside runs.
# References between nodes, and the counts.
kinds = {n["id"]: n["kind"] for n in nodes}
for n in nodes:
for r in n.get("content", []) if n["kind"] in RUN_KINDS else []:
if r["kind"] == "cite":
for ref in r["refs"]:
if kinds.get(ref) != "reference":
rep.add("doc.run.cite", n["id"], "cite names %s, which is not a reference" % ref)
if r["kind"] == "fnmark" and r.get("footnote") is not None and kinds.get(r["footnote"]) != "footnote":
rep.add("doc.run.fnmark", n["id"], "fnmark names a node that is not a footnote")
if r["kind"] == "xref" and r.get("target") and "#" not in r["target"]:
if r["target"] not in kinds:
rep.add("doc.run.xref", n["id"], "xref names no node in this document")
if n["kind"] == "table":
for col in n.get("attrs", {}).get("columns", []):
for h in col["header_path"]:
if kinds.get(h) != "cell":
rep.add("doc.table.header", n["id"], "a header path names %s, not a cell" % h)
if n["kind"] == "row":
for h in n.get("attrs", {}).get("header_path", []):
if kinds.get(h) != "cell":
rep.add("doc.table.header", n["id"], "a row header path names %s, not a cell" % h)
try:
validate_references(doc)
except Refusal as e:
rep.add(e.rule, e.node or "", e.message)
leaves_n = sum(1 for n in nodes if n["kind"] in LEAVES)
if document.get("counts") != {"nodes": len(nodes), "leaves": leaves_n}:
rep.add("doc.counts", "document.json", "counts do not recompute")
# Hashes.
hasher = hashing.PLAIN if scheme == "plain" else (hashing.Hasher(key) if key else None)
@@ -285,7 +330,7 @@ def _validate(root, store, key, rep):
# Corrections.
cors = data["corrections.jsonl"]
for i, c in enumerate(cors, 1):
if c["index"] != i:
if c.get("index") != i:
rep.add("doc.correction.index", "corrections.jsonl:%d" % i, "indexes count from 1 without gaps")
if c["id"] != record_id(sha256_hex(jcs.dumpb({k: v for k, v in c.items() if k != "id"}))):
rep.add("doc.correction.id", "corrections.jsonl:%d" % i, "id does not derive from the record")
@@ -361,16 +406,33 @@ def _validate(root, store, key, rep):
ids_listed = [x[0] for x in r["nodes"]]
if ids_listed != sorted(ids_listed) or len(set(ids_listed)) != len(ids_listed):
rep.add("doc.receipt.nodes", r["id"], "nodes sort by ID without repeats")
if roster and not R.verify_signature(r, roster):
rep.add("doc.receipt.signature", r["id"], "signature does not verify")
if roster:
signer = roster.get(r["by"])
if not R.verify_signature(r, roster):
rep.add("doc.receipt.signature", r["id"], "signature does not verify")
elif r["signature"]["key"] != r["by"] or not signer or not R.key_valid_at(signer, r["at"]):
rep.add("doc.receipt.signature", r["id"], "the signing key is not the signer's key, valid when signed")
qa = r.get("qa")
if qa:
orig = by_id.get(qa["of"])
if not orig:
rep.add("doc.qa.of", r["id"], "the spot-checked receipt is not in this package")
if not orig or orig["format"] != "scienting-receipt/1":
rep.add("doc.qa.of", r["id"], "the spot-checked receipt is not a receipt in this package")
continue
if r["scope"] != "qa:" + qa["of"]:
rep.add("doc.qa.of", r["id"], "the scope names another receipt than qa.of")
if orig["at"] > r["at"]:
rep.add("doc.qa.of", r["id"], "a spot-check is signed after the receipt it checks")
orig_hashes = dict(orig["nodes"])
if [n for n, _ in r["nodes"]] != qa["sample"] or any(orig_hashes.get(n) != h for n, h in r["nodes"]):
rep.add("doc.qa.sample", r["id"], "the receipt's nodes are not the sampled nodes at the original's hashes")
if not {e["node"] for e in qa["errors"]} <= set(qa["sample"]):
rep.add("doc.qa.sample", r["id"], "an error names a node outside the sample")
try:
seed = base64.b64decode(qa["seed"], validate=True)
except (binascii.Error, ValueError):
rep.add("doc.qa.seed", r["id"], "the seed is not Base64")
continue
ck = roster.get(qa["key"])
seed = base64.b64decode(qa["seed"])
if roster:
if not ck or ck["scheme"] != "ed25519" or "spot-check" not in ck["purposes"]:
rep.add("doc.qa.key", r["id"], "the seed key is not a registered Ed25519 spot-check key")
@@ -380,7 +442,9 @@ def _validate(root, store, key, rep):
rep.add("doc.qa.seed", r["id"], "the seed is not the checker's signature over the receipt")
if qa["sample"] != R.sample(orig, seed, qa["rate"], qa["minimum"]):
rep.add("doc.qa.sample", r["id"], "the sample does not follow from the seed")
if roster and not withheld:
if not withheld and (roster or not recs):
# Without a roster no receipt can be judged valid, so a package that carries receipts but no roster is
# checked against its store instead; one with neither reports the leaves' own levels.
if R.distribution(R.effective_levels(nodes, vals, recs, roster)) != document["fidelity"]:
rep.add("doc.fidelity.report", "document.json", "the fidelity distribution does not recompute")
@@ -401,3 +465,137 @@ def read_state(root):
def n_root_hash(nodes, doc):
return {n["id"]: n for n in nodes}[doc.root]["node_hash"]
# --- zipped packages (spec/12 §12.2 and §12.8) -----------------------------------------
ZIP_EXTENSION = ".sct.zip"
# The reference tools' own limit on an archive's expanded size (spec/12 §12.2 leaves the limit to the consumer).
ZIP_LIMIT = 1 << 30
_ZIP_DATE = (1980, 1, 1, 0, 0, 0)
_FLAG_ENCRYPTED = 0x1
_FLAG_UTF8 = 0x800
def is_zip(path):
return os.path.isfile(path) and path.lower().endswith(".zip")
def write_zip(root, path):
"""Zip the package folder at `root` the way spec/12 §12.2 says a producer writes a byte-identical archive.
Entries sit under one top-level folder named by the key, sorted by path, stored, dated 1980-01-01 00:00,
with no extra fields and no directory entries. Returns the SHA-256 of the archive.
"""
key = read_json(os.path.join(root, "document.json"))["key"]
files = _all_files(root)
for p in files:
if p != "MANIFEST.json" and not safe_path(p):
raise Refusal("package.path.unsafe", "unsafe path %r" % p)
with zipfile.ZipFile(path, "w") as zf:
for p in sorted(files, key=lambda x: x.encode("utf-8")):
zi = zipfile.ZipInfo(key + "/" + p, date_time=_ZIP_DATE)
zi.compress_type = zipfile.ZIP_STORED
zi.create_system = 3
zi.external_attr = 0o644 << 16
with open(os.path.join(root, p), "rb") as f:
zf.writestr(zi, f.read())
with open(path, "rb") as f:
return sha256_hex(f.read())
def _zip_entries(zf, limit):
"""Check every entry of an open archive as spec/12 §12.2 requires; return the top-level folder."""
infos = zf.infolist()
if not infos:
raise Refusal("package.zip.entry", "the archive has no entries")
folder, seen, total = None, set(), 0
for zi in infos:
name = zi.filename
if not (zi.flag_bits & _FLAG_UTF8) and any(ord(ch) > 0x7F for ch in name):
raise Refusal("package.zip.entry", "%r: entry name is not marked as UTF-8" % name)
if zi.flag_bits & _FLAG_ENCRYPTED:
raise Refusal("package.zip.entry", "%s: encrypted entries are refused" % name)
top, sep, rest = name.partition("/")
if not sep or not KEY_RE.match(top):
raise Refusal("package.zip.entry", "%s: every entry sits under one folder named by the key" % name)
if folder is None:
folder = top
elif top != folder:
raise Refusal("package.zip.entry", "%s: entries sit under more than one top-level folder" % name)
if name in seen:
raise Refusal("package.zip.entry", "%s: duplicate entry" % name)
seen.add(name)
if zi.is_dir():
if rest != "" and not safe_path(rest.rstrip("/")):
raise Refusal("package.zip.entry", "%s: unsafe path" % name)
continue
if stat.S_ISLNK(zi.external_attr >> 16):
raise Refusal("package.zip.entry", "%s: symbolic links are refused" % name)
if zi.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED):
raise Refusal("package.zip.entry", "%s: entries are stored or deflated" % name)
if rest != "MANIFEST.json" and not safe_path(rest):
raise Refusal("package.zip.entry", "%s: unsafe path" % name)
total += zi.file_size
if total > limit:
raise Refusal("package.zip.size", "the archive expands past the limit of %d bytes" % limit)
return folder
def read_zip(path, dest, limit=ZIP_LIMIT):
"""Unpack a zipped package into the empty or new folder `dest`, refusing what spec/12 §12.2 refuses.
Returns the package folder, `dest/<key>`. Every byte written is counted against each entry's declared size
and against `limit`, so a header that lies about a size is refused too.
"""
try:
zf = zipfile.ZipFile(path)
except (zipfile.BadZipFile, OSError) as e:
raise Refusal("package.zip.entry", "not a zip archive: %s" % e)
with zf:
folder = _zip_entries(zf, limit)
root = os.path.join(dest, folder)
os.makedirs(root)
written = 0
for zi in zf.infolist():
if zi.is_dir():
continue
rest = zi.filename.partition("/")[2]
full = os.path.join(root, rest)
os.makedirs(os.path.dirname(full), exist_ok=True)
try:
with zf.open(zi) as src, open(full, "wb") as dst:
n = 0
while True:
chunk = src.read(1 << 16)
if not chunk:
break
n += len(chunk)
written += len(chunk)
if n > zi.file_size or written > limit:
raise Refusal("package.zip.size", "%s: holds more than its header declares" % zi.filename)
dst.write(chunk)
except (zipfile.BadZipFile, RuntimeError, OSError) as e:
raise Refusal("package.zip.entry", "%s: %s" % (zi.filename, e))
return root
def validate_zip(path, store, key=None, limit=ZIP_LIMIT):
"""Validate a zipped package: unpack it under spec/12 §12.2's rules, then validate the folder."""
with tempfile.TemporaryDirectory() as tmp:
try:
root = read_zip(path, tmp, limit)
except Refusal as e:
rep = Report()
rep.add(e.rule, e.node or "", e.message)
return rep
rep = validate(root, store, key)
mpath = os.path.join(root, "MANIFEST.json")
if os.path.exists(mpath):
try:
mkey = read_json(mpath).get("key")
except (Refusal, ValueError):
mkey = None
if mkey != os.path.basename(root):
rep.add("package.zip.entry", os.path.basename(root), "the top-level folder is not named by the key")
return rep
+97 -26
View File
@@ -4,15 +4,17 @@ Non-normative; see spec/08-fidelity.md.
"""
import base64
import binascii
import hashlib
import math
from fractions import Fraction
from math import comb
from . import crypto, jcs
from .model import LEVELS
from .text import record_id, sha256_hex
SIGNED_FORMATS = ("scienting-receipt/1", "scienting-revocation/1", "scienting-policy/1")
SIGNED_FORMATS = ("scienting-receipt/1", "scienting-revocation/1", "scienting-policy/1", "scienting-qa-report/1")
def record_hash(rec):
@@ -37,13 +39,29 @@ def seal(rec, secret, key_id):
return rec
def _b64(s):
"""Strict Base64 decoding; None for anything that is not Base64."""
try:
return base64.b64decode(s, validate=True)
except (binascii.Error, ValueError, TypeError):
return None
def verify_signature(rec, roster):
"""Whether the record's Ed25519 signature verifies under the roster key it names. Never raises."""
sig = rec.get("signature") or {}
if not isinstance(sig, dict):
return False
key = roster.get(sig.get("key"))
if not key or sig.get("scheme") != "ed25519" or key["scheme"] != "ed25519":
if not key or sig.get("scheme") != "ed25519" or key.get("scheme") != "ed25519":
return False
pub, value = _b64(key.get("public", "")), _b64(sig.get("value", ""))
if pub is None or value is None or len(pub) != 32 or len(value) != 64:
return False
try:
return crypto.ed25519_verify(pub, signing_payload(rec), value)
except (ValueError, TypeError):
return False
pub = base64.b64decode(key["public"])
return crypto.ed25519_verify(pub, signing_payload(rec), base64.b64decode(sig["value"]))
def key_valid_at(key, at):
@@ -103,10 +121,31 @@ def upper_bound(k, n, confidence="0.95"):
def bound_string(k, n, confidence="0.95"):
"""The bound as the spec reports it: six decimal places, rounded up."""
x = upper_bound(k, n, confidence)
v = math.ceil(x * 10**6 - 1e-9)
return "%d.%06d" % (v // 10**6, v % 10**6)
"""The bound as the spec reports it: six decimal places, rounded up, computed exactly.
The reported value is the least multiple of 10^-6 at which the probability of at most k errors in n is
at most 1 - c, found by bisection with integer arithmetic, so no floating point enters the result.
"""
if n <= 0:
raise ValueError("n must be positive")
if k >= n:
return "1.000000"
alpha = 1 - Fraction(confidence)
scale = 10**6
def tail_at_most(m):
# P[X <= k] at p = m / scale, compared with alpha, both scaled by scale**n and alpha's denominator.
total = sum(comb(n, i) * m**i * (scale - m)**(n - i) for i in range(k + 1))
return total * alpha.denominator <= alpha.numerator * scale**n
lo, hi = 0, scale # the tail at 0 is 1 (above alpha); at 1 it is 0 (k < n)
while hi - lo > 1:
mid = (lo + hi) // 2
if tail_at_most(mid):
hi = mid
else:
lo = mid
return "%d.%06d" % (hi // scale, hi % scale)
# --- levels -------------------------------------------------------------------
@@ -141,26 +180,57 @@ def valid_revocations(revocations, receipts_by_id, roster):
return out
def receipt_status(rec, records, roster):
"""`valid`, `revoked`, `suspended`, or `invalid` for one receipt."""
receipts = [x for x in records if x["format"] == "scienting-receipt/1"]
revs = [x for x in records if x["format"] == "scienting-revocation/1"]
def _base_status(rec, roster, valid_revs):
"""`invalid`, `revoked` or `valid` for one receipt, before spot-checks are considered (spec/08 §8.6)."""
try:
if rec.get("root") != crypto.merkle_root(crypto.receipt_items(rec)):
return "invalid"
if rec["id"] != record_id(record_hash(rec)) or not verify_signature(rec, roster):
return "invalid"
key = roster.get(rec["by"])
if not key or not key_valid_at(key, rec["at"]) or rec["signature"]["key"] != rec["by"]:
return "invalid"
except (KeyError, TypeError, ValueError):
return "invalid"
return "revoked" if revoked(rec, valid_revs) else "valid"
def statuses(records, roster):
"""The status of every receipt in `records` (spec/08 §8.6), computed once.
A receipt is suspended when a valid spot-check of it found an error, and a spot-check's own validity may
depend on its own spot-check, so statuses are settled from the end of each chain of spot-checks; a chain
that loops back on itself settles nothing, and its receipts are invalid.
"""
receipts = [x for x in records if x.get("format") == "scienting-receipt/1"]
revs = [x for x in records if x.get("format") == "scienting-revocation/1"]
by_id = {x["id"]: x for x in receipts}
if rec.get("root") != crypto.merkle_root(crypto.receipt_items(rec)):
return "invalid"
if rec["id"] != record_id(record_hash(rec)) or not verify_signature(rec, roster):
return "invalid"
key = roster.get(rec["by"])
if not key or not key_valid_at(key, rec["at"]) or rec["signature"]["key"] != rec["by"]:
return "invalid"
if revoked(rec, valid_revocations(revs, by_id, roster)):
return "revoked"
valid_revs = valid_revocations(revs, by_id, roster)
base = {rec["id"]: _base_status(rec, roster, valid_revs) for rec in receipts}
checkers = {}
for q in receipts:
qa = q.get("qa")
if qa and qa.get("of") == rec["id"] and qa.get("errors"):
if receipt_status(q, records, roster) == "valid":
return "suspended"
return "valid"
if isinstance(qa, dict) and qa.get("errors") and base.get(q["id"]) == "valid" and qa.get("of") in by_id:
checkers.setdefault(qa["of"], []).append(q["id"])
final = {rid: s for rid, s in base.items() if s != "valid" or rid not in checkers}
pending = set(base) - set(final)
progress = True
while pending and progress:
progress = False
for rid in sorted(pending):
qs = checkers[rid]
if all(q in final for q in qs):
final[rid] = "suspended" if any(final[q] == "valid" for q in qs) else "valid"
pending.discard(rid)
progress = True
for rid in pending:
final[rid] = "invalid"
return final
def receipt_status(rec, records, roster):
"""`valid`, `revoked`, `suspended`, or `invalid` for one receipt among `records`."""
return statuses(records, roster).get(rec.get("id"), "invalid")
def effective_levels(nodes, values, records, roster):
@@ -169,10 +239,11 @@ def effective_levels(nodes, values, records, roster):
A spot-check's `qa:` receipt does not raise the nodes it found in error (spec/08 §8.2).
"""
level = {n["id"]: n.get("fidelity") for n in nodes if "fidelity" in n}
status = statuses(records, roster)
for rec in records:
if rec["format"] != "scienting-receipt/1":
continue
if receipt_status(rec, records, roster) != "valid":
if status.get(rec["id"]) != "valid":
continue
wrong = {e["node"] for e in (rec.get("qa") or {}).get("errors", [])}
for nid, h in rec.get("nodes", []):
+69
View File
@@ -0,0 +1,69 @@
"""The target set of targeted F3: the nodes a team's claims, data and quotations rest on.
Non-normative; see spec/08-fidelity.md §8.10.
"""
from .model import LEAVES, MATH_KINDS
from .text import is_digit
DIGIT_RULES = ("all", "anchored-sections", "none")
# The kinds that count as the section a node sits in: a section, or the front or back matter outside any.
SECTION_KINDS = frozenset(["section", "front", "back"])
DIGIT_KINDS = frozenset(["sentence", "cell"])
DIGIT_RUNS = frozenset(["text", "math"])
def has_digit_run(node):
"""Whether a `text` or `math` run of the node holds a digit (General Category Nd)."""
for r in node.get("content") or []:
if not isinstance(r, dict) or r.get("kind") not in DIGIT_RUNS:
continue
s = r.get("text") if r["kind"] == "text" else r.get("source")
if any(is_digit(ch) for ch in s or ""):
return True
return False
def enclosing_section(doc, nid):
"""The nearest ancestor-or-self that is a section, or the front or back matter, else the root."""
if doc.nodes[nid]["kind"] in SECTION_KINDS:
return nid
for a in reversed(doc.ancestors(nid)):
if doc.nodes[a]["kind"] in SECTION_KINDS:
return a
return doc.root
def target_set(doc, anchored, digit_rule="all"):
"""The target set as `[[node, reason], …]` in document order, with `reason` the first that applies of
`anchored`, `equation` and `digits`.
`anchored` lists the live nodes that extracts, dataset values, confirmed claim links, draft quotations or
pinned citations anchor; a container among them contributes its leaves. Every equation and equation line
is a target. Under `all`, so is every sentence or cell with a digit in a text or math run; under
`anchored-sections`, only those in a section that holds an anchored node; under `none`, none.
"""
if digit_rule not in DIGIT_RULES:
raise ValueError("digit rule must be one of %s" % ", ".join(DIGIT_RULES))
anchored = set(anchored)
unknown = sorted(a for a in anchored if a not in doc.nodes)
if unknown:
raise ValueError("anchored nodes are not live: %s" % ", ".join(unknown))
reasons = {}
for nid in sorted(anchored):
if doc.nodes[nid]["kind"] in LEAVES:
reasons[nid] = "anchored"
else:
for leaf in doc.leaves(nid):
reasons.setdefault(leaf, "anchored")
for nid in doc.preorder():
if doc.nodes[nid]["kind"] in MATH_KINDS:
reasons.setdefault(nid, "equation")
if digit_rule != "none":
sections = {enclosing_section(doc, a) for a in anchored}
for nid in doc.preorder():
n = doc.nodes[nid]
if n["kind"] in DIGIT_KINDS and has_digit_run(n):
if digit_rule == "all" or enclosing_section(doc, nid) in sections:
reasons.setdefault(nid, "digits")
return [[nid, reasons[nid]] for nid in doc.preorder() if nid in reasons]
+33
View File
@@ -212,5 +212,38 @@ def _check_content(doc, nid):
raise Refusal("doc.node.attrs", "a section needs a level", nid)
def validate_references(doc):
"""Check the references between nodes: cite, fnmark and xref runs, and the header paths of tables and rows
(spec/03 §3.4–3.5, spec/09 §9.2). Raises Refusal."""
kinds = {nid: doc.nodes[nid]["kind"] for nid in doc.order}
for nid in doc.order:
n = doc.nodes[nid]
if n["kind"] in RUN_KINDS and not n.get("withheld"):
for r in n.get("content") or []:
if r["kind"] == "cite":
for ref in r["refs"]:
if kinds.get(ref) != "reference":
raise Refusal("doc.run.cite", "cite names %s, which is not a reference" % ref, nid)
if r["kind"] == "fnmark" and r.get("footnote") is not None and kinds.get(r["footnote"]) != "footnote":
raise Refusal("doc.run.fnmark", "fnmark names a node that is not a footnote", nid)
if r["kind"] == "xref" and r.get("target") and "#" not in r["target"] and r["target"] not in kinds:
raise Refusal("doc.run.xref", "xref names no node in this document", nid)
if n["kind"] == "table":
for col in n.get("attrs", {}).get("columns", []):
for h in col["header_path"]:
if kinds.get(h) != "cell":
raise Refusal("doc.table.header", "a header path names %s, not a cell" % h, nid)
if n["kind"] == "row":
for h in n.get("attrs", {}).get("header_path", []):
if kinds.get(h) != "cell":
raise Refusal("doc.table.header", "a row header path names %s, not a cell" % h, nid)
if n["kind"] == "cell":
col = n.get("attrs", {}).get("column")
table = next((a for a in reversed(doc.ancestors(nid)) if kinds[a] == "table"), None)
columns = {c["id"] for c in doc.nodes[table].get("attrs", {}).get("columns", [])} if table else set()
if col is not None and columns and col not in columns:
raise Refusal("doc.table.header", "a cell names a column its table does not define", nid)
def leaf_kinds():
return LEAVES
+621
View File
@@ -0,0 +1,621 @@
#!/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)
from tools.oracle import align as AL # noqa: E402
from tools.oracle import anchors as AN # noqa: E402
from tools.oracle import corrections as CO # noqa: E402
from tools.oracle import crypto, fingerprint as FP, hashing, jcs, receipts as RC, rights as RI, targets as TG # noqa: E402
from tools.oracle import package as PK # noqa: E402
from tools.oracle.model import LABELLED, LEAVES, Doc # noqa: E402
from tools.oracle.notation import fold_math # noqa: E402
from tools.oracle.text import NODE_ID_RE, RECORD_ID_RE, letter_first, norm, record_id, sha256_hex, is_ws # noqa: E402
from tools.oracle.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 STORE, 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 = {"nodes": [[i, sha256_hex(i)] for i in ids]}
seed = os.urandom(64)
rate, minimum = rng.choice(["0.05", "0.1", "0.5", "1"]), rng.randint(1, 15)
sample = RC.sample(receipt, seed, rate, minimum)
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", "tiny2026example" + 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", "tiny2026example")
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(["hale2026harmonic", "k", "a.b_c-d"])
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", "hale2026harmonic"))
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": "hale2026harmonic", "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": "hale2026harmonic", "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 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)
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()
+20 -5
View File
@@ -4,7 +4,9 @@
Non-normative tooling.
Usage: python3 tools/validate.py PACKAGE [PACKAGE ...]
python3 tools/validate.py --correction PROPOSALS PACKAGE
A package is checked as spec/22 §22.2 describes, and each failure printed as `package: where: rule: message`.
A package is a folder, or a zipped package (`.sct.zip`), which is unpacked under the rules of spec/12 §12.2
before it is checked. Each is checked as spec/22 §22.2 describes, and each failure printed as
`package: where: rule: message`.
With --correction, PROPOSALS is a JSON file holding one proposed correction, or a JSON Lines file of several
applied in order. Each runs through the gate (spec/11 §11.4) against the package's current state, and the record
it would append to corrections.jsonl is printed, until one is refused. Nothing is written.
@@ -15,6 +17,7 @@ import argparse
import json
import os
import sys
import tempfile
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)
@@ -33,19 +36,28 @@ def read_proposals(path):
return [json.loads(ln) for ln in text.splitlines() if ln.strip()]
def validate_one(path, store, limit):
"""The validation report of a package folder or a zipped package."""
if package.is_zip(path):
return package.validate_zip(path, store, limit=limit)
return package.validate(path, store)
def main():
lines = __doc__.splitlines()
ap = argparse.ArgumentParser(description=lines[0], epilog="\n".join(lines[5:]),
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--correction", metavar="PROPOSALS",
help="a JSON or JSON Lines file of proposed corrections to run through the gate")
ap.add_argument("packages", nargs="+", metavar="PACKAGE", help="package folders")
ap.add_argument("--max-size", metavar="BYTES", type=int, default=package.ZIP_LIMIT,
help="refuse a zipped package that expands past this many bytes (default %d)" % package.ZIP_LIMIT)
ap.add_argument("packages", nargs="+", metavar="PACKAGE", help="package folders or zipped packages")
a = ap.parse_args()
if a.correction and len(a.packages) != 1:
ap.error("--correction checks proposals against exactly one package")
store, bad = Store(os.path.join(ROOT, "schemas")), False
for p in a.packages:
for e in package.validate(p, store).errors:
for e in validate_one(p, store, a.max_size).errors:
print("%s: %s: %s: %s" % (p, e["where"], e["rule"], e["message"]), file=sys.stderr)
bad = True
if bad:
@@ -55,9 +67,12 @@ def main():
print("%s: valid" % p)
return 0
pkg = a.packages[0]
state = package.read_state(pkg)
with tempfile.TemporaryDirectory() as tmp:
if package.is_zip(pkg):
pkg = package.read_zip(pkg, tmp, a.max_size)
state = package.read_state(pkg)
if any(n.get("withheld") for n in state.nodes):
print("%s: doc.policy.withheld: a package that withholds its text takes no corrections" % pkg,
print("%s: doc.policy.withheld: a package that withholds its text takes no corrections" % a.packages[0],
file=sys.stderr)
return 1
for i, proposal in enumerate(read_proposals(a.correction), 1):