diff --git a/README.md b/README.md index c23edf4..8ada012 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,13 @@ codebases that handle hostile input. Stele treats every parser as untrusted: pointer in the sealed bundle. A trusted resolver re-reads every anchor from the archive and refuses any unit whose text doesn't match. `ExtractionAdapter` delivers the units through the Dispatcher. +- **Sign off a new sandbox setup on the whole corpus.** `stele.verification` + pins a corpus by content and runs every document through today's baseline + (bubblewrap) and each candidate backend. Its report re-checks every sealed + run against the ledger, archive and corpus, compares each candidate with + the baseline and applies explicit gates. A named person's sign-off is + anchored in the event log, and goes stale if a record it covers is later + invalidated. ``` source ─▶ staging ─▶ sandboxed parser ─▶ /stele/output ─▶ ledger (pending) @@ -155,6 +162,7 @@ Details and known limits: [docs/containment.md](docs/containment.md). | [Ledger](docs/ledger.md) | State machine, record fields, parser identity, migration, the hash-chained event log | | [Replay](docs/replay.md) | Validation, replay outcomes, invalidation, ledger views | | [Adapter contract](docs/adapter.md) | `SteleAdapter`, `TargetWriter`, the Dispatcher and delivery log, the ChatGPT export adapter | +| [Production verification](docs/verification.md) | Corpus manifests, campaigns, the gated report, sign-off | | [Extraction contract](docs/extraction.md) | `stele.extraction` v1, normalizers, the trusted resolver, `ExtractionAdapter` | | [Identity contract](docs/identity.md) | `stele:` references to sources, snapshots, records, observations, anchors and events, and how they resolve | | [Evidence store](docs/archive.md) | Content-addressed Snapshot and artifact archive | @@ -171,6 +179,7 @@ stele/ ├── replay/ validation, replay engine, invalidation, views ├── contracts/ adapter, dispatcher and target-writer protocols ├── adapters/ adapters (ChatGPT export: every branch, streamed; canonical extraction) +├── verification/ full-corpus production verification and sign-off ├── extraction/ stele.extraction v1 contract, normalizers, trusted resolver ├── extractors/ deterministic Wasm extractors └── parsers/ packaged ML parsers in pinned images diff --git a/docs/verification.md b/docs/verification.md new file mode 100644 index 0000000..d68eeb4 --- /dev/null +++ b/docs/verification.md @@ -0,0 +1,215 @@ +# Full-Corpus Production Verification and Sign-Off + +Before a new sandbox setup replaces the one production runs today, it has to +agree with it on the real corpus, every document of it. `stele.verification` +runs that check and records a named person's sign-off on the result. The +baseline is today's bubblewrap-only setup. + +The procedure has four steps: pin the corpus, run the campaign, build the +report, sign it off. + +``` +corpus ──▶ campaign ──────────────▶ journal (resumable) + │ every document × every lane + ▼ + ledger + archive (sealed runs) + │ + ▼ + report (re-checks everything; gates) ──▶ sign-off event in the hash chain +``` + +## 1. Pin the corpus (`corpus.py`) + +```bash +python -m stele.verification corpus /data/production -o corpus.json +``` + +- `stele.corpus` v1 lists every regular file under the root with its SHA-256 + and size, sorted by path. +- It also lists everything it left out, with the reason: symlinks (never + followed), FIFOs, devices, unreadable files. Coverage is explicit, never + silently partial. +- The manifest is canonical JSON, and its SHA-256 is the **corpus id**. Every + journal, report and sign-off names it. + +## 2. Run the campaign (`campaign.py`) + +```bash +python -m stele.verification run --corpus corpus.json --root /data/production \ + --ledger verify.db --archive verify-archive --journal campaign.jsonl \ + --script parser.py --baseline bubblewrap --candidate oci-runc --candidate oci-runsc +``` + +A **lane** is one way to run the parser under test: a backend plus the spec +that runs the parser on it. Every lane must be the same parser name and +version. + +| Parser | Option | Lanes | +|--------|--------|-------| +| Python script run as `/stele/parser` | `--script parser.py` | `bubblewrap` (host interpreter), `oci-runc`, `oci-runsc`, `oci-runc-gpu` (the image's `python3`) | +| Packaged ML parser | `--parser mineru\|marker\|docling` | `oci-*`, with the parser's pinned image and limits | +| Wasm extractor | `--extractor chatgpt-export-split` | `wasmtime` | + +A script parser's version defaults to `sha256-`, +so a changed script is a different parser. + +For each document, the baseline lane runs first, then each candidate lane. +For each (document, lane) the campaign: + +1. runs the parser through the normal path: `run_parser` → `run_in_sandbox`, + with the ledger's archive and the lane's backend forced, never + auto-selected; +2. requires the input Snapshot the parser read to hash to the corpus + manifest's SHA-256 for that document; +3. records and seals a successful run in the ledger, then deletes the working + output, since the archive holds it; +4. appends one **observation** to the journal: `sealed`, `failed` (with the + run's structured failure reason and telemetry), `not_applicable` (the + parser does not accept the document), `containment_error`, or `error`. + +The journal is JSON lines, and each line is fsynced before the next run +starts. + +- **Resuming:** the header pins the corpus id, parser, config digest and + lanes. A rerun skips every (document, lane) already observed, and a torn + final line from a crash is dropped and run again. `--limit N` runs a batch + and stops. +- **Unavailable backends:** a campaign whose lane backend is unavailable + refuses to start. It never falls back to another backend. + +## 3. Build the report (`report.py`) + +```bash +python -m stele.verification report --corpus corpus.json --root /data/production \ + --ledger verify.db --archive verify-archive --journal campaign.jsonl \ + [--waivers waivers.json] [--max-baseline-failure-rate 0.01] [--max-slowdown 3] \ + -o report.json +``` + +The journal is trusted only for what the ledger cannot know: why a run failed +and how long it took. For every sealed run, the report re-checks: + +- the record exists, is still `SEALED`, and is the run the journal names + (run id, artifact hash, parser name and version); +- the ledger says it was made on the backend of the lane that claims it, with + the campaign's parser configuration, and no other document or lane claims + the same record (so one lane's output can never stand in for another's); +- its `source_hash` is the corpus document's SHA-256; +- its bundle re-verifies in the evidence archive; +- where the parser has an extraction normalizer (Markdown from MinerU, + Marker, Docling; ChatGPT), the canonical extraction normalizes and every + unit resolves against the sealed bytes. + +Each candidate is then compared with the baseline, document by document: + +| Outcome | Meaning | +|---------|---------| +| `identical` | both sealed, same artifact manifest | +| `equivalent` | both sealed, and the parser's comparison policy (the same one replays use) accepts the difference | +| `diverged` | both sealed, outside the policy (or any difference, without one) | +| `regression` | the baseline sealed, the candidate failed | +| `recovered` | the candidate sealed where the baseline fails | +| `consistent_failure` | both failed, for the same reason | +| `failure_mismatch` | both failed for different reasons, or only one lane accepted the document | +| `not_applicable`, `incomplete`, `error` | neither lane applies; a lane has no observation; an error on either lane | + +The report also summarizes each lane: status counts, failure reasons, and +p50/p95/max of wall time, CPU time and peak memory. For each candidate it +gives outcome counts and the median slowdown against the baseline. + +### Gates + +A report passes only if every gate passes. + +| Gate | Passes when | +|------|-------------| +| `corpus_unchanged` | the root still hashes exactly as the manifest (fails if the corpus was not checked) | +| `coverage` | every document has an observation on every lane | +| `baseline_produced_output` | the baseline sealed at least one document | +| `no_errors` | no Stele, host or containment errors | +| `records_verified` | every sealed record passes the re-checks above | +| `ledger_verified` | the event chain and the ledger tables agree (`verify_ledger`) | +| `candidates_agree` | no unwaived `diverged`, `regression` or `failure_mismatch` | +| `waivers_used` | every waiver matches a finding | +| `baseline_failure_rate` | (optional) the baseline fails at most this share of applicable documents | +| `slowdown` | (optional) each candidate's median wall-time ratio is within the limit | + +**Waivers** accept a known comparison finding. Each waiver names one +document, one candidate lane and one outcome, and gives a written reason: + +```json +[{"path": "scans/0042.pdf", "lane": "oci-runsc", "outcome": "regression", + "reason": "gVisor lacks the AVX path; tracked in #123"}] +``` + +Only `diverged`, `regression` and `failure_mismatch` can be waived. +Integrity problems (a record that does not verify, a changed corpus, an +unverifiable ledger, an error) cannot be waived. + +The report is canonical JSON (`stele.verification-report` v1). Its SHA-256 +is what gets signed off. + +## 4. Sign off (`signoff.py`) + +```bash +python -m stele.verification sign-off --corpus corpus.json --root /data/production \ + --ledger verify.db --archive verify-archive --journal campaign.jsonl \ + report.json --by "Name" --note "Q3 production sign-off" +``` + +The sign-off does not take the report's word. It refuses unless all of these +hold: + +- a **rebuild** of the report from the ledger, journal and corpus, with the + report's own thresholds and waivers, gives the same report. An edited + report with its gates flipped to passed is refused, even if it is + re-encoded canonically; +- every gate passed; +- the event chain still contains the head the report was built at; +- every record in the report is still `SEALED` with the artifact hash it + names. + +It then stores the report in the evidence archive and appends a +`verification.signed_off` event to the hash-chained event log. The event +names the report digest, the corpus, the parser, the lanes and the person +signing off. The command prints the event's anchor `seq:hash`; publish it +outside the ledger. + +```bash +python -m stele.verification check --ledger verify.db --archive verify-archive +python -m stele.verification list --ledger verify.db --archive verify-archive +``` + +`check` re-verifies a sign-off later: + +- the event is in the chain; +- the archived report still hashes to its digest; +- every record the report covers is still sealed. + +If a record was invalidated after the sign-off, `check` reports the sign-off +as **stale**. + +## What this does not do + +- **It does not decide which corpus is production.** Pinning it is the + operator's job; the manifest makes the choice auditable. +- **Equivalence is only as strong as the parser's comparison policy.** A + non-deterministic parser without a policy reports every difference as + `diverged`. +- **An orphan record after a crash.** If a crash lands between sealing a + record and journaling it, the resumed campaign runs that document again, + and the earlier sealed record stays in the ledger unreferenced by the + report. +- **Anyone with write access to the SQLite file can rewrite the chain, + sign-off included.** As with the event log, publishing the anchor is what + makes it hold. + +## Tests (`tests/test_verification.py`) + +| Proof | Test | +|-------|------| +| Content-pinned corpus, strict manifest, exclusions listed, drift caught | `TestCorpus` | +| Every document on every lane, sealed or failed with its reason; resume, torn lines, mismatched journals | `TestCampaign` | +| Every comparison outcome, policy equivalence, waivers, coverage, integrity and threshold gates | `TestReport` | +| Anchored sign-off; refused for failed, stale or forged reports; stale after invalidation | `TestSignOff` | +| Bubblewrap baseline end to end, an OCI candidate against it, and the whole CLI | `TestRealSandboxes` | diff --git a/stele/verification/__init__.py b/stele/verification/__init__.py new file mode 100644 index 0000000..081787d --- /dev/null +++ b/stele/verification/__init__.py @@ -0,0 +1,30 @@ +"""Full-corpus production verification and sign-off (roadmap #15). + +corpus + Corpus: the documents under verification, pinned by content (stele.corpus v1). +campaign + Campaign: every document through a baseline lane (today's production + setup) and candidate lanes, sealed in the ledger, journaled resumably. +report + build_report: re-checks the campaign against the ledger, archive and + corpus, compares candidates with the baseline, and decides the gates. +signoff + sign_off / check_sign_off: a named person's acceptance of a passed report, + anchored in the hash-chained event log; stale once a record it covers is + invalidated. +lanes + Lanes for script parsers, packaged ML parsers and Wasm extractors. + +CLI: python -m stele.verification {corpus,run,report,sign-off,check,list}. +""" +from .campaign import Campaign, CampaignError, Journal, Lane, Observation +from .corpus import Corpus, CorpusEntry, CorpusFormatError +from .report import Gates, ReportFormatError, VerificationReport, Waiver, build_report +from .signoff import SIGNOFF_EVENT, SignOff, SignOffRefused, check_sign_off, sign_off, sign_offs + +__all__ = [ + "Campaign", "CampaignError", "Journal", "Lane", "Observation", + "Corpus", "CorpusEntry", "CorpusFormatError", + "Gates", "ReportFormatError", "VerificationReport", "Waiver", "build_report", + "SIGNOFF_EVENT", "SignOff", "SignOffRefused", "check_sign_off", "sign_off", "sign_offs", +] diff --git a/stele/verification/__main__.py b/stele/verification/__main__.py new file mode 100644 index 0000000..eb9b152 --- /dev/null +++ b/stele/verification/__main__.py @@ -0,0 +1,227 @@ +""" +python -m stele.verification corpus ROOT -o corpus.json +python -m stele.verification run --corpus corpus.json --root ROOT --ledger L --archive A \ + --journal J (--script P | --parser NAME | --extractor NAME) \ + --baseline bubblewrap [--candidate oci-runc ...] +python -m stele.verification report --corpus corpus.json --root ROOT --ledger L --archive A \ + --journal J [--waivers W.json] [--max-baseline-failure-rate R] [--max-slowdown X] -o report.json +python -m stele.verification sign-off --corpus corpus.json --root ROOT --ledger L --archive A \ + --journal J report.json --by NAME [--note TEXT] +python -m stele.verification check --ledger L --archive A REPORT_DIGEST +python -m stele.verification list --ledger L --archive A +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from ..archive import BlobStore +from ..ledger.store import LedgerStore +from .campaign import Campaign, Journal, Observation +from .corpus import Corpus +from .report import Gates, VerificationReport, Waiver, build_report +from .signoff import SignOffRefused, check_sign_off, sign_off, sign_offs + + +def _ledger(args: argparse.Namespace) -> LedgerStore: + return LedgerStore(args.ledger, BlobStore(args.archive)) + + +def _store_args(p: argparse.ArgumentParser) -> None: + p.add_argument("--ledger", type=Path, required=True) + p.add_argument("--archive", type=Path, required=True) + + +def _campaign_args(p: argparse.ArgumentParser) -> None: + _store_args(p) + p.add_argument("--corpus", type=Path, required=True, help="stele.corpus manifest") + p.add_argument("--root", type=Path, required=True, help="the corpus directory") + p.add_argument("--journal", type=Path, required=True) + + +def _progress(obs: Observation) -> None: + extra = (obs.failure or {}).get("reason") or obs.detail or obs.record_id or "" + print(f"{obs.status:17} {obs.lane:14} {obs.path} {extra}"[:240], file=sys.stderr) + + +def _cmd_corpus(args: argparse.Namespace) -> int: + corpus = Corpus.build(args.root, name=args.name) + corpus.save(args.output) + print(json.dumps({ + "corpus_id": corpus.corpus_id, "documents": len(corpus), "bytes": corpus.total_bytes, + "excluded": len(corpus.excluded), + }, indent=2)) + return 0 + + +def _cmd_run(args: argparse.Namespace) -> int: + from . import lanes as lane_mod + + backends = [args.baseline, *args.candidate] + config = json.loads(args.config) if args.config else {} + if args.script: + lanes = lane_mod.script_lanes( + args.script, backends, name=args.name, version=args.version, + deterministic=args.deterministic, timeout_seconds=args.timeout, engine=args.engine, + **({"image": args.image} if args.image else {}), + ) + elif args.parser: + lanes, defaults, _ = lane_mod.packaged_lanes(args.parser, backends, engine=args.engine) + config = {**defaults, **config} + else: + lanes = lane_mod.extractor_lanes(args.extractor, backends) + ledger = _ledger(args) + try: + campaign = Campaign( + ledger, Corpus.load(args.corpus), args.root, lanes, baseline=args.baseline, + journal=args.journal, parser_config=config, keep_outputs=args.keep_outputs, + ) + done = campaign.run(limit=args.limit, progress=None if args.quiet else _progress) + remaining = sum(1 for _ in campaign.pending()) + finally: + ledger.close() + print(json.dumps({"observed": len(done), "remaining": remaining}, indent=2)) + return 0 + + +def _policy_for(parser: dict) -> object | None: + from ..parsers.catalog import PARSERS + + image = PARSERS.get(parser["name"]) + if image is not None and image.version == parser["version"]: + return image.comparison + return None + + +def _cmd_report(args: argparse.Namespace) -> int: + ledger = _ledger(args) + try: + journal = Journal.read(args.journal) + report = build_report( + ledger, Corpus.load(args.corpus), journal, root=args.root, + policy=_policy_for(journal.header["parser"]), + gates=Gates( + max_baseline_failure_rate=args.max_baseline_failure_rate, + max_slowdown=args.max_slowdown, check_extractions=not args.no_extractions, + ), + waivers=Waiver.load_all(args.waivers) if args.waivers else (), + ) + finally: + ledger.close() + args.output.write_bytes(report.to_canonical()) + for gate in report.data["gates"]: + mark = "PASS" if gate["passed"] else "FAIL" + print(f"{mark} {gate['name']:26} {gate['detail']}") + for example in gate.get("examples", [])[:5] if not gate["passed"] else []: + print(f" {example}") + print(f"report sha256:{report.digest} -> {args.output} " + f"({'passed' if report.passed else 'NOT passed'})") + return 0 if report.passed else 1 + + +def _cmd_sign_off(args: argparse.Namespace) -> int: + report = VerificationReport.from_canonical(args.report.read_bytes()) + ledger = _ledger(args) + try: + journal = Journal.read(args.journal) + done = sign_off( + ledger, report, corpus=Corpus.load(args.corpus), journal=journal, root=args.root, + policy=_policy_for(journal.header["parser"]), signed_off_by=args.by, note=args.note, + ) + except SignOffRefused as exc: + print("sign-off refused:", file=sys.stderr) + for problem in exc.problems: + print(f" {problem}", file=sys.stderr) + return 1 + finally: + ledger.close() + print(json.dumps({ + "report_digest": done.report_digest, "corpus_id": done.corpus_id, + "signed_off_by": done.signed_off_by, "anchor": f"{done.seq}:{done.event_hash}", + }, indent=2)) + return 0 + + +def _cmd_check(args: argparse.Namespace) -> int: + ledger = _ledger(args) + try: + problems = check_sign_off(ledger, args.digest) + finally: + ledger.close() + print(json.dumps({"ok": not problems, "problems": problems}, indent=2)) + return 0 if not problems else 1 + + +def _cmd_list(args: argparse.Namespace) -> int: + ledger = _ledger(args) + try: + print(json.dumps(sign_offs(ledger), indent=2)) + finally: + ledger.close() + return 0 + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(prog="python -m stele.verification", + description="Full-corpus production verification and sign-off.") + sub = ap.add_subparsers(dest="command", required=True) + + p = sub.add_parser("corpus", help="pin a corpus directory by content") + p.add_argument("root", type=Path) + p.add_argument("-o", "--output", type=Path, required=True) + p.add_argument("--name", default=None) + p.set_defaults(func=_cmd_corpus) + + p = sub.add_parser("run", help="run (or resume) a campaign") + _campaign_args(p) + which = p.add_mutually_exclusive_group(required=True) + which.add_argument("--script", type=Path, help="a Python parser script") + which.add_argument("--parser", help="a packaged parser: mineru, marker, docling") + which.add_argument("--extractor", help="a Wasm extractor, e.g. chatgpt-export-split") + p.add_argument("--baseline", default="bubblewrap") + p.add_argument("--candidate", action="append", default=[]) + p.add_argument("--name", default=None, help="script parser name (default: file stem)") + p.add_argument("--version", default=None, help="script parser version (default: its digest)") + p.add_argument("--deterministic", action="store_true") + p.add_argument("--timeout", type=int, default=300) + p.add_argument("--image", default=None, help="container image for a script parser") + p.add_argument("--engine", default=None, choices=["podman", "docker"]) + p.add_argument("--config", default=None, help="parser config as JSON") + p.add_argument("--limit", type=int, default=None, help="stop after this many runs") + p.add_argument("--keep-outputs", action="store_true") + p.add_argument("--quiet", action="store_true") + p.set_defaults(func=_cmd_run) + + p = sub.add_parser("report", help="build and gate the verification report") + _campaign_args(p) + p.add_argument("--waivers", type=Path, default=None) + p.add_argument("--max-baseline-failure-rate", type=float, default=None) + p.add_argument("--max-slowdown", type=float, default=None) + p.add_argument("--no-extractions", action="store_true") + p.add_argument("-o", "--output", type=Path, required=True) + p.set_defaults(func=_cmd_report) + + p = sub.add_parser("sign-off", help="sign off a passed report") + _campaign_args(p) + p.add_argument("report", type=Path) + p.add_argument("--by", required=True) + p.add_argument("--note", default="") + p.set_defaults(func=_cmd_sign_off) + + p = sub.add_parser("check", help="re-verify a sign-off") + _store_args(p) + p.add_argument("digest") + p.set_defaults(func=_cmd_check) + + p = sub.add_parser("list", help="list sign-offs") + _store_args(p) + p.set_defaults(func=_cmd_list) + + args = ap.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/stele/verification/campaign.py b/stele/verification/campaign.py new file mode 100644 index 0000000..6e89971 --- /dev/null +++ b/stele/verification/campaign.py @@ -0,0 +1,329 @@ +""" +A verification campaign: every corpus document through every lane. + +A lane is one way of running the parser under test: a sandbox backend plus +the ParserSpec that runs the parser there (a script parser needs the host +interpreter under bubblewrap but the image's python3 in a container, so each +lane may bring its own spec, but every lane must be the same parser name and +version). One lane is the baseline, the setup production runs today; the +others are candidates that must agree with it before they replace it. + +For each document and lane the campaign: + +- runs the parser through the normal path (run_parser -> run_in_sandbox with + the ledger's archive), forcing the lane's backend; +- checks the parser read exactly the corpus bytes (the input Snapshot digest + must equal the manifest's SHA-256); +- records a successful run in the ledger and seals it, and drops the working + output (the archive holds it); +- appends one observation to the journal: status, record id, the run's + structured failure and its telemetry. + +The journal (JSON lines, fsynced per line) makes a full-corpus campaign +resumable: its header pins the corpus id, parser and lanes, and a rerun skips +every (document, lane) already observed. Failed runs are not ledger records, +so the journal is where their failure reasons live; the report (report.py) +re-checks everything a journal line claims about sealed records against the +ledger itself. +""" +from __future__ import annotations + +import dataclasses +import json +import os +import shutil +import traceback +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Iterator, Mapping, Sequence + +from ..archive.records import Source, canonical_json +from ..containment.backend import ( + ContainmentCleanupError, + SandboxBackend, + UnsupportedBackendError, +) +from ..ledger.store import LedgerStore +from ..ledger.transaction import record_run +from ..replay.parsers import ParserSpec, run_parser +from .corpus import Corpus + +JOURNAL_SCHEMA = "stele.verification-journal" +JOURNAL_VERSION = 1 + +# Observation statuses. +SEALED = "sealed" # ran, recorded and sealed +FAILED = "failed" # ran and failed (structured failure reason) +NOT_APPLICABLE = "not_applicable" # the parser does not accept this document +CONTAINMENT_ERROR = "containment_error" # the sandbox could not be proven gone +ERROR = "error" # anything else: a Stele or host fault +STATUSES = (SEALED, FAILED, NOT_APPLICABLE, CONTAINMENT_ERROR, ERROR) + + +class CampaignError(RuntimeError): + """The campaign cannot start or continue (mismatched journal, unusable lane).""" + + +@dataclass(frozen=True) +class Lane: + name: str + backend: SandboxBackend + spec: ParserSpec + + def forced_spec(self) -> ParserSpec: + """The lane's spec with its backend forced (never auto-selected).""" + backend = self.backend + return dataclasses.replace(self.spec, backend=lambda _identity: backend) + + +@dataclass(frozen=True) +class Observation: + path: str + lane: str + status: str + backend: str | None = None + run_id: str | None = None + record_id: str | None = None + artifact_hash: str | None = None + input_sha256: str | None = None + failure: dict[str, Any] | None = None + telemetry: dict[str, Any] | None = None + detail: str | None = None + at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def to_json(self) -> dict[str, Any]: + return dataclasses.asdict(self) + + @classmethod + def from_json(cls, obj: Mapping[str, Any]) -> "Observation": + names = {f.name for f in dataclasses.fields(cls)} + if not isinstance(obj, Mapping) or set(obj) != names or obj.get("status") not in STATUSES: + raise CampaignError(f"malformed journal observation: {obj!r}") + return cls(**obj) + + +def config_digest(parser_config: Mapping[str, Any]) -> str: + import hashlib + + return hashlib.sha256(canonical_json(dict(parser_config))).hexdigest() + + +class Journal: + """Append-only JSON-lines journal of one campaign.""" + + def __init__(self, path: Path, header: dict[str, Any]) -> None: + self.path = Path(path) + self.header = header + self.observations: list[Observation] = [] + if self.path.exists() and self.path.stat().st_size > 0: + data = self.path.read_bytes() + if not data.endswith(b"\n"): + # A torn final line from a crash mid-append: that observation + # was never complete, so drop it and observe again. + data = data[:data.rfind(b"\n") + 1] + self._truncate_to(data) + lines = data.split(b"\n")[:-1] + try: + existing = json.loads(lines[0]) if lines else None + except ValueError as exc: + raise CampaignError(f"{self.path}: unreadable journal header: {exc}") from exc + if existing is None: + self._append_line(header) + elif existing != header: + raise CampaignError( + f"{self.path} belongs to another campaign (corpus, parser, config or lanes " + "differ); use a new journal" + ) + for number, line in enumerate(lines[1:], start=2): + try: + self.observations.append(Observation.from_json(json.loads(line))) + except ValueError as exc: + raise CampaignError(f"{self.path} line {number} is malformed: {exc}") from exc + else: + self.path.parent.mkdir(parents=True, exist_ok=True) + self._append_line(header) + + @classmethod + def read(cls, path: Path) -> "Journal": + """Open an existing journal without knowing its header in advance.""" + with open(path, "rb") as f: + header = json.loads(f.readline()) + if header.get("schema") != JOURNAL_SCHEMA or header.get("schema_version") != JOURNAL_VERSION: + raise CampaignError(f"{path} is not a {JOURNAL_SCHEMA} v{JOURNAL_VERSION}") + return cls(path, header) + + def done(self) -> set[tuple[str, str]]: + return {(o.path, o.lane) for o in self.observations} + + def append(self, observation: Observation) -> None: + self._append_line(observation.to_json()) + self.observations.append(observation) + + def _append_line(self, obj: Mapping[str, Any]) -> None: + data = canonical_json(dict(obj)) + b"\n" + fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o644) + try: + os.write(fd, data) + os.fsync(fd) + finally: + os.close(fd) + + def _truncate_to(self, data: bytes) -> None: + with open(self.path, "r+b") as f: + f.truncate(len(data)) + f.flush() + os.fsync(f.fileno()) + + +class Campaign: + """Runs a corpus through a baseline lane and candidate lanes.""" + + def __init__( + self, + ledger: LedgerStore, + corpus: Corpus, + root: Path, + lanes: Sequence[Lane], + *, + baseline: str, + journal: Path, + parser_config: Mapping[str, Any] | None = None, + work_dir: Path | None = None, + keep_outputs: bool = False, + ) -> None: + names = [lane.name for lane in lanes] + if len(set(names)) != len(names) or not names: + raise CampaignError("lanes must have unique names") + if baseline not in names: + raise CampaignError(f"baseline lane {baseline!r} is not one of {names}") + identities = {(lane.spec.name, lane.spec.version) for lane in lanes} + if len(identities) != 1: + raise CampaignError(f"every lane must run the same parser; got {sorted(identities)}") + self.ledger = ledger + self.corpus = corpus + self.root = Path(root) + # The baseline runs first for every document. + self.lanes = sorted(lanes, key=lambda lane: lane.name != baseline) + self.baseline = baseline + self.parser_config = dict(parser_config or {}) + self.work_dir = Path(work_dir) if work_dir else Path(journal).with_suffix(".work") + self.keep_outputs = keep_outputs + self._index = {entry.path: i for i, entry in enumerate(corpus.entries)} + spec = self.lanes[0].spec + self.journal = Journal(journal, header={ + "schema": JOURNAL_SCHEMA, "schema_version": JOURNAL_VERSION, + "corpus_id": corpus.corpus_id, "corpus_name": corpus.name, + "parser": {"name": spec.name, "version": spec.version}, + "parser_config_digest": config_digest(self.parser_config), + "lanes": [{"name": lane.name, "backend": lane.backend.name} for lane in self.lanes], + "baseline": baseline, + }) + + def pending(self) -> Iterator[tuple[str, Lane]]: + done = self.journal.done() + for entry in self.corpus.entries: + for lane in self.lanes: + if (entry.path, lane.name) not in done: + yield entry.path, lane + + def run( + self, + *, + limit: int | None = None, + progress: Callable[[Observation], None] | None = None, + ) -> list[Observation]: + """Observe every pending (document, lane); returns this call's observations.""" + for lane in self.lanes: + if not lane.backend.available(): + raise CampaignError( + f"lane {lane.name}: backend {lane.backend.name} is unavailable here: " + f"{lane.backend.unavailable_reason()}" + ) + observed: list[Observation] = [] + for path, lane in self.pending(): + if limit is not None and len(observed) >= limit: + break + observation = self.observe(path, lane) + self.journal.append(observation) + observed.append(observation) + if progress is not None: + progress(observation) + return observed + + def observe(self, path: str, lane: Lane) -> Observation: + number = self._index[path] + entry = self.corpus.entries[number] + document = self.root / path + out = self.work_dir / lane.name / f"{number:08d}" + if out.exists(): + shutil.rmtree(out) # a leftover from an interrupted attempt + out.parent.mkdir(parents=True, exist_ok=True) + spec = lane.forced_spec() + base = {"path": path, "lane": lane.name} + try: + spec.build_config(document, out, self.parser_config) + except (ValueError, TypeError, KeyError) as exc: + return Observation(**base, status=NOT_APPLICABLE, detail=str(exc)) + try: + result = run_parser( + spec, artifact_dir=out, parser_config=self.parser_config, + input_path=document, store=self.ledger.archive, + ) + except ContainmentCleanupError as exc: + return Observation(**base, status=CONTAINMENT_ERROR, backend=lane.backend.name, + detail=str(exc)) + except UnsupportedBackendError as exc: + return Observation(**base, status=ERROR, backend=lane.backend.name, + detail=f"backend refused the run: {exc}") + except Exception as exc: # noqa: BLE001 - every fault is an observation + return Observation(**base, status=ERROR, backend=lane.backend.name, + detail=_one_line(exc)) + + common = dict( + base, backend=result.backend, run_id=str(result.run_id), + input_sha256=result.input_sha256, + telemetry=result.telemetry.to_dict() if result.telemetry is not None else None, + ) + if result.input_sha256 != entry.sha256: + _discard(out) + return Observation(**common, status=ERROR, detail=( + f"the parser read {result.input_sha256}, not the corpus document " + f"{entry.sha256}: the file changed after the manifest was taken" + )) + if not result.succeeded or not result.produced_artifacts: + failure = result.failure + _discard(out) + return Observation( + **common, status=FAILED, + failure=None if failure is None else { + "reason": failure.reason.value, "detail": failure.detail, + "exit_code": failure.exit_code, "signal": failure.signal, + }, + detail=None if result.produced_artifacts else "the run produced no artifacts", + ) + try: + record = record_run( + self.ledger, result, parser=spec.identity(), + parser_config=self.parser_config, source=Source.from_path(document), + ) + except Exception as exc: # noqa: BLE001 + _discard(out) + return Observation(**common, status=ERROR, detail=f"ledgering failed: {_one_line(exc)}") + if not self.keep_outputs: + _discard(out) + return Observation( + **common, status=SEALED, record_id=record.record_id, + artifact_hash=record.artifact_hash, + ) + + +def _discard(path: Path) -> None: + shutil.rmtree(path, ignore_errors=True) + + +def _one_line(exc: BaseException) -> str: + text = f"{type(exc).__name__}: {exc}".splitlines() + frame = traceback.extract_tb(exc.__traceback__)[-1:] if exc.__traceback__ else [] + where = f" (at {Path(frame[0].filename).name}:{frame[0].lineno})" if frame else "" + return (text[0] if text else type(exc).__name__)[:500] + where diff --git a/stele/verification/corpus.py b/stele/verification/corpus.py new file mode 100644 index 0000000..2a565e8 --- /dev/null +++ b/stele/verification/corpus.py @@ -0,0 +1,227 @@ +""" +The corpus under verification, pinned by content: `stele.corpus`, version 1. + +A production verification is only meaningful if everyone can say which +documents it covered. A Corpus is the sorted list of every regular file under +a root directory with its SHA-256 and size, plus every path that was left out +and why (symlinks, sockets, unreadable files): coverage is explicit, never +silently partial. Its canonical JSON is strict, and its digest is the corpus +id every campaign, report and sign-off names. + +check(root) re-hashes the tree: a verification whose documents changed, +disappeared or gained siblings since the manifest was taken is not a +verification of that corpus. +""" +from __future__ import annotations + +import hashlib +import json +import os +import stat +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any, Iterator + +from ..archive.records import canonical_json, is_digest + +SCHEMA = "stele.corpus" +SCHEMA_VERSION = 1 +_CHUNK = 1 << 20 + + +class CorpusFormatError(ValueError): + """Bytes or values that are not a valid stele.corpus/v1.""" + + +def _check_path(path: Any) -> str: + if not isinstance(path, str) or not path: + raise CorpusFormatError(f"corpus path must be a non-empty string: {path!r}") + pure = PurePosixPath(path) + if pure.is_absolute() or any(p in ("", ".", "..") for p in path.split("/")) or "\\" in path: + raise CorpusFormatError(f"corpus path must be a normalized relative POSIX path: {path!r}") + return path + + +@dataclass(frozen=True) +class CorpusEntry: + path: str # relative POSIX path under the corpus root + sha256: str + size: int + + def __post_init__(self) -> None: + _check_path(self.path) + if not is_digest(self.sha256): + raise CorpusFormatError(f"{self.path}: not a SHA-256 digest: {self.sha256!r}") + if type(self.size) is not int or self.size < 0: + raise CorpusFormatError(f"{self.path}: size must be a non-negative int") + + def to_json(self) -> dict[str, Any]: + return {"path": self.path, "sha256": self.sha256, "size": self.size} + + +@dataclass(frozen=True) +class Excluded: + path: str + why: str # "symlink", "not a regular file", "unreadable: ..." + + def to_json(self) -> dict[str, Any]: + return {"path": self.path, "why": self.why} + + +@dataclass(frozen=True) +class Corpus: + name: str + entries: tuple[CorpusEntry, ...] + excluded: tuple[Excluded, ...] = () + + def __post_init__(self) -> None: + object.__setattr__(self, "entries", tuple(self.entries)) + object.__setattr__(self, "excluded", tuple(self.excluded)) + if not isinstance(self.name, str) or not self.name: + raise CorpusFormatError("corpus name must be a non-empty string") + paths = [e.path for e in self.entries] + if paths != sorted(paths) or len(set(paths)) != len(paths): + raise CorpusFormatError("corpus entries must be unique and sorted by path") + skipped = [x.path for x in self.excluded] + if skipped != sorted(skipped) or set(skipped) & set(paths): + raise CorpusFormatError("excluded paths must be sorted and not also be entries") + + @property + def corpus_id(self) -> str: + return hashlib.sha256(self.to_canonical()).hexdigest() + + @property + def total_bytes(self) -> int: + return sum(e.size for e in self.entries) + + def __len__(self) -> int: + return len(self.entries) + + def by_path(self) -> dict[str, CorpusEntry]: + return {e.path: e for e in self.entries} + + # -- building and checking ------------------------------------------------ + + @classmethod + def build(cls, root: Path, *, name: str | None = None) -> "Corpus": + """Hash every regular file under root; record everything left out.""" + root = Path(root) + if not root.is_dir(): + raise NotADirectoryError(f"corpus root {root} is not a directory") + entries, excluded = [], [] + for rel, full, kind in _walk(root): + if kind != "file": + excluded.append(Excluded(rel, kind)) + continue + try: + digest, size = _hash_file(full) + except OSError as exc: + excluded.append(Excluded(rel, f"unreadable: {exc.strerror or exc}")) + continue + entries.append(CorpusEntry(rel, digest, size)) + return cls( + name=name or root.resolve().name or "corpus", + entries=tuple(sorted(entries, key=lambda e: e.path)), + excluded=tuple(sorted(excluded, key=lambda x: x.path)), + ) + + def check(self, root: Path) -> list[str]: + """Problems between the manifest and the tree at root; [] means it is this corpus.""" + root = Path(root) + problems: list[str] = [] + present = {rel for rel, _, _ in _walk(root)} + wanted = {e.path: e for e in self.entries} + for rel in sorted(present - set(wanted) - {x.path for x in self.excluded}): + problems.append(f"{rel}: not in the corpus manifest") + for path, entry in wanted.items(): + full = root / path + try: + st = os.lstat(full) + except FileNotFoundError: + problems.append(f"{path}: missing") + continue + if not stat.S_ISREG(st.st_mode): + problems.append(f"{path}: no longer a regular file") + continue + try: + digest, size = _hash_file(full) + except OSError as exc: + problems.append(f"{path}: unreadable: {exc.strerror or exc}") + continue + if (digest, size) != (entry.sha256, entry.size): + problems.append(f"{path}: content changed since the manifest was taken") + return problems + + # -- serialization ---------------------------------------------------------- + + def to_canonical(self) -> bytes: + return canonical_json({ + "schema": SCHEMA, "schema_version": SCHEMA_VERSION, "name": self.name, + "entries": [e.to_json() for e in self.entries], + "excluded": [x.to_json() for x in self.excluded], + }) + + @classmethod + def from_canonical(cls, data: bytes) -> "Corpus": + try: + obj = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, ValueError) as exc: + raise CorpusFormatError(f"not UTF-8 JSON: {exc}") from exc + fields = {"schema", "schema_version", "name", "entries", "excluded"} + if not isinstance(obj, dict) or set(obj) != fields: + raise CorpusFormatError(f"corpus fields must be exactly {sorted(fields)}") + if obj["schema"] != SCHEMA or obj["schema_version"] != SCHEMA_VERSION: + raise CorpusFormatError(f"expected {SCHEMA} v{SCHEMA_VERSION}") + try: + corpus = cls( + name=obj["name"], + entries=tuple(CorpusEntry(**e) for e in obj["entries"]), + excluded=tuple(Excluded(**x) for x in obj["excluded"]), + ) + except TypeError as exc: + raise CorpusFormatError(f"malformed entry: {exc}") from exc + if corpus.to_canonical() != data: + raise CorpusFormatError("corpus manifest is not in canonical form") + return corpus + + @classmethod + def load(cls, path: Path) -> "Corpus": + return cls.from_canonical(Path(path).read_bytes()) + + def save(self, path: Path) -> None: + Path(path).write_bytes(self.to_canonical()) + + +def _walk(root: Path) -> Iterator[tuple[str, Path, str]]: + """(relative POSIX path, full path, "file" | why excluded), never following links.""" + for dirpath, dirnames, filenames in os.walk(root, followlinks=False): + base = Path(dirpath) + dirnames.sort() + for d in list(dirnames): + if (base / d).is_symlink(): + dirnames.remove(d) + yield (base / d).relative_to(root).as_posix(), base / d, "symlink" + for f in sorted(filenames): + full = base / f + rel = full.relative_to(root).as_posix() + try: + mode = os.lstat(full).st_mode + except OSError as exc: + yield rel, full, f"unreadable: {exc.strerror or exc}" + continue + if stat.S_ISLNK(mode): + yield rel, full, "symlink" + elif not stat.S_ISREG(mode): + yield rel, full, "not a regular file" + else: + yield rel, full, "file" + + +def _hash_file(path: Path) -> tuple[str, int]: + h, size = hashlib.sha256(), 0 + fd = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)) + with os.fdopen(fd, "rb") as f: + while chunk := f.read(_CHUNK): + h.update(chunk) + size += len(chunk) + return h.hexdigest(), size diff --git a/stele/verification/lanes.py b/stele/verification/lanes.py new file mode 100644 index 0000000..d59a81a --- /dev/null +++ b/stele/verification/lanes.py @@ -0,0 +1,123 @@ +""" +Lanes for the parsers Stele knows how to run, by backend name. + + bubblewrap Linux namespaces + seccomp + Landlock (today's production baseline) + oci-runc Podman/Docker container, runc + oci-runsc Podman/Docker container, gVisor + oci-runc-gpu Podman/Docker container with NVIDIA GPUs + wasmtime WebAssembly/WASI (Wasm extractors only) + +A script parser (a Python file run as /stele/parser) runs on bubblewrap with +the host interpreter and in a container with the image's python3. A packaged +parser (MinerU, Marker, Docling) runs its pinned image on the OCI lanes. A +Wasm extractor runs on wasmtime. +""" +from __future__ import annotations + +import hashlib +import sys +from pathlib import Path +from typing import Any, Mapping + +from ..containment.backend import BubblewrapBackend, ParserRequirements, SandboxBackend +from ..containment.oci import DEFAULT_OCI_IMAGE, OciBackend +from ..containment.sandbox import SandboxConfig +from ..replay.parsers import ParserSpec +from ..replay.policy import ComparisonPolicy +from .campaign import Lane + +BACKEND_NAMES = ("bubblewrap", "oci-runc", "oci-runsc", "oci-runc-gpu", "wasmtime") + + +def _oci(name: str, *, image: str, engine: str | None, **limits: Any) -> OciBackend: + runtime = "runsc" if name == "oci-runsc" else "runc" + return OciBackend(image=image, engine=engine, runtime=runtime, + gpu=name.endswith("-gpu"), **limits) + + +def backend_named(name: str, *, image: str = DEFAULT_OCI_IMAGE, engine: str | None = None, + **limits: Any) -> SandboxBackend: + if name == "bubblewrap": + return BubblewrapBackend() + if name.startswith("oci-"): + if name not in BACKEND_NAMES: + raise ValueError(f"unknown backend {name!r}; expected one of {BACKEND_NAMES}") + return _oci(name, image=image, engine=engine, **limits) + if name == "wasmtime": + from ..containment.wasm import WasmtimeBackend + + return WasmtimeBackend() + raise ValueError(f"unknown backend {name!r}; expected one of {BACKEND_NAMES}") + + +def script_lanes( + script: Path, + backends: list[str], + *, + name: str | None = None, + version: str | None = None, + deterministic: bool = False, + timeout_seconds: int = 300, + image: str = DEFAULT_OCI_IMAGE, + engine: str | None = None, +) -> list[Lane]: + """Lanes for a Python script parser. Its version defaults to the script's digest.""" + script = Path(script).resolve() + digest = hashlib.sha256(script.read_bytes()).hexdigest() + name = name or script.stem + version = version or f"sha256-{digest[:16]}" + host_python = str(Path(sys.executable).resolve()) + lanes = [] + for backend_name in backends: + if backend_name == "wasmtime": + raise ValueError("a script parser cannot run on wasmtime") + interpreter = host_python if backend_name == "bubblewrap" else "python3" + + def build(input_path: Path | None, artifact_dir: Path, config: Mapping[str, Any], + interpreter: str = interpreter) -> SandboxConfig: + return SandboxConfig( + command=[interpreter, "/stele/parser"], artifact_dir=artifact_dir, + input_path=input_path, script_path=script, timeout_seconds=timeout_seconds, + ) + + spec = ParserSpec( + name=name, version=version, + requirements=ParserRequirements(deterministic=deterministic), + build_config=build, + ) + lanes.append(Lane(backend_name, backend_named(backend_name, image=image, engine=engine), spec)) + return lanes + + +def packaged_lanes( + parser_name: str, backends: list[str], *, engine: str | None = None, +) -> tuple[list[Lane], dict[str, Any], ComparisonPolicy | None]: + """Lanes, merged default config and comparison policy of a packaged ML parser.""" + from ..parsers.catalog import get_parser + from ..parsers.replay import replay_spec + + parser = get_parser(parser_name) + lanes = [] + for backend_name in backends: + if not backend_name.startswith("oci-"): + raise ValueError(f"{parser_name} runs only in its container image (oci-* lanes)") + device = "gpu" if backend_name.endswith("-gpu") else "cpu" + backend = backend_named( + backend_name, image=parser.image_for(device), engine=engine, + memory=parser.memory, cpus=parser.cpus, pids_limit=parser.pids_limit, + tmpfs_size=parser.tmpfs_size, + ) + lanes.append(Lane(backend_name, backend, replay_spec(parser, backend=backend))) + return lanes, dict(parser.config), parser.comparison + + +def extractor_lanes(name: str, backends: list[str]) -> list[Lane]: + """Lanes of a Wasm extractor (e.g. chatgpt-export-split).""" + from ..extractors import EXTRACTOR_SPECS + + specs = {spec.name: spec for spec in EXTRACTOR_SPECS} + if name not in specs: + raise ValueError(f"unknown extractor {name!r}") + if any(b != "wasmtime" for b in backends): + raise ValueError(f"{name} is a Wasm module; its only lane is wasmtime") + return [Lane(b, backend_named(b), specs[name]) for b in backends] diff --git a/stele/verification/report.py b/stele/verification/report.py new file mode 100644 index 0000000..599d0fd --- /dev/null +++ b/stele/verification/report.py @@ -0,0 +1,651 @@ +""" +The verification report: what a campaign proved, and whether it may be signed off. + +build_report() trusts the journal only for what the ledger cannot know (why a +run failed, how long it took). Everything a journal line claims about a sealed +run is re-checked against the ledger and the evidence archive: + +- the record exists, is still SEALED, and is the run the journal names + (run id, artifact hash, parser name and version), made on the backend of + the lane that claims it with the campaign's parser configuration, and no + other document or lane claims the same record; +- its source_hash is the corpus manifest's SHA-256 for that document; +- its bundle re-verifies in the archive (validate_artifact); +- where the parser has an extraction normalizer, the canonical extraction + normalizes and every unit resolves against the sealed bytes. + +Then each candidate lane is compared with the baseline, document by document: + + both sealed IDENTICAL (same manifest), EQUIVALENT (the parser's comparison + policy accepts the difference), or DIVERGED + baseline only REGRESSION: the candidate failed where production succeeds + candidate only RECOVERED: the candidate succeeds where production fails + both failed CONSISTENT_FAILURE (same reason) or FAILURE_MISMATCH + otherwise NOT_APPLICABLE, INCOMPLETE (a lane has no observation) or ERROR + +Gates decide sign-off. A comparison finding (DIVERGED, REGRESSION, +FAILURE_MISMATCH) may be waived per document and lane with a written reason, +and every waiver must match something. Integrity problems (a record that does +not verify, an unverifiable ledger, a changed corpus, a containment error) +are never waivable. + +The report is canonical JSON (`stele.verification-report` v1); its SHA-256 is +what a sign-off (signoff.py) anchors in the ledger's event log. +""" +from __future__ import annotations + +import hashlib +import json +import math +import statistics +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Iterable, Mapping, Sequence + +from ..archive.records import canonical_json +from ..contracts.adapter import SealedBundle +from ..ledger.events import EventLog, verify_ledger +from ..ledger.models import ArtifactState +from ..ledger.store import LedgerStore, RecordNotFoundError +from ..replay.policy import ComparisonPolicy +from ..replay.validator import validate_artifact +from .campaign import ( + CONTAINMENT_ERROR, ERROR, FAILED, NOT_APPLICABLE, SEALED, Journal, Observation, config_digest, +) +from .corpus import Corpus + +REPORT_SCHEMA = "stele.verification-report" +REPORT_VERSION = 1 + +IDENTICAL = "identical" +EQUIVALENT = "equivalent" +DIVERGED = "diverged" +REGRESSION = "regression" +RECOVERED = "recovered" +CONSISTENT_FAILURE = "consistent_failure" +FAILURE_MISMATCH = "failure_mismatch" +INCOMPLETE = "incomplete" +# NOT_APPLICABLE and ERROR share the observation status names. + +WAIVABLE = frozenset({DIVERGED, REGRESSION, FAILURE_MISMATCH}) +_MAX_FINDINGS = 20 + + +class ReportFormatError(ValueError): + """Bytes that are not a valid stele.verification-report/v1.""" + + +@dataclass(frozen=True) +class Waiver: + """An accepted comparison finding: one document, one candidate lane, one outcome.""" + + path: str + lane: str + outcome: str + reason: str + + def __post_init__(self) -> None: + if self.outcome not in WAIVABLE: + raise ValueError( + f"only {sorted(WAIVABLE)} can be waived, not {self.outcome!r}; integrity " + "problems must be fixed" + ) + if not isinstance(self.reason, str) or not self.reason.strip(): + raise ValueError("a waiver needs a written reason") + + def to_json(self) -> dict[str, str]: + return {"path": self.path, "lane": self.lane, "outcome": self.outcome, "reason": self.reason} + + @classmethod + def load_all(cls, path: Path) -> list["Waiver"]: + data = json.loads(Path(path).read_text()) + if not isinstance(data, list): + raise ValueError(f"{path}: waivers must be a JSON list") + return [cls(**w) for w in data] + + +@dataclass(frozen=True) +class Gates: + """Thresholds beyond the fixed gates. None disables a gate.""" + + # Share of applicable documents the baseline may fail (0.01 = 1%). + max_baseline_failure_rate: float | None = None + # Median candidate/baseline wall-time ratio over documents both sealed. + max_slowdown: float | None = None + # Verify the canonical extraction of every sealed record that has a normalizer. + check_extractions: bool = True + + def to_json(self) -> dict[str, Any]: + return { + "max_baseline_failure_rate": self.max_baseline_failure_rate, + "max_slowdown": self.max_slowdown, + "check_extractions": self.check_extractions, + } + + +@dataclass(frozen=True) +class VerificationReport: + data: dict[str, Any] + _canonical: bytes = field(repr=False, compare=False, default=b"") + + def __post_init__(self) -> None: + if not self._canonical: + object.__setattr__(self, "_canonical", canonical_json(self.data)) + + def to_canonical(self) -> bytes: + return self._canonical + + @property + def digest(self) -> str: + return hashlib.sha256(self._canonical).hexdigest() + + @property + def passed(self) -> bool: + return bool(self.data["passed"]) + + @property + def corpus_id(self) -> str: + return self.data["corpus"]["id"] + + @property + def ledger_head(self) -> tuple[int, str] | None: + head = self.data["ledger"]["head"] + return None if head is None else (head["seq"], head["hash"]) + + def failed_gates(self) -> list[dict[str, Any]]: + return [g for g in self.data["gates"] if not g["passed"]] + + def record_ids(self) -> list[str]: + ids = [] + for doc in self.data["documents"]: + for lane in doc["lanes"].values(): + if lane.get("record_id"): + ids.append(lane["record_id"]) + return ids + + @classmethod + def from_canonical(cls, data: bytes) -> "VerificationReport": + try: + obj = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, ValueError) as exc: + raise ReportFormatError(f"not UTF-8 JSON: {exc}") from exc + if not isinstance(obj, dict) or obj.get("schema") != REPORT_SCHEMA \ + or obj.get("schema_version") != REPORT_VERSION: + raise ReportFormatError(f"expected {REPORT_SCHEMA} v{REPORT_VERSION}") + for key in ("corpus", "parser", "lanes", "baseline", "ledger", "documents", "summary", + "gates", "waivers", "passed"): + if key not in obj: + raise ReportFormatError(f"report has no {key!r}") + if canonical_json(obj) != data: + raise ReportFormatError("report is not in canonical form") + return cls(obj, data) + + +# --------------------------------------------------------------------------- +# Building +# --------------------------------------------------------------------------- + +def build_report( + ledger: LedgerStore, + corpus: Corpus, + journal: Journal, + *, + root: Path | None, + policy: ComparisonPolicy | None = None, + gates: Gates = Gates(), + waivers: Sequence[Waiver] = (), +) -> VerificationReport: + """Re-check a campaign against the ledger and decide every gate. + + root is the corpus directory, re-hashed against the manifest; without it + the corpus_unchanged gate fails (a report that has not checked the corpus + cannot be signed off). policy is the parser's comparison policy (the same + one replays are judged by); without one, any difference is DIVERGED. + """ + header = journal.header + if header["corpus_id"] != corpus.corpus_id: + raise ValueError("the journal is of another corpus") + lanes = [lane["name"] for lane in header["lanes"]] + baseline = header["baseline"] + candidates = [name for name in lanes if name != baseline] + parser = header["parser"] + + by_doc: dict[str, dict[str, Observation]] = {} + for obs in journal.observations: + by_doc.setdefault(obs.path, {})[obs.lane] = obs # the latest observation wins + + checker = _RecordChecker( + ledger, parser, gates.check_extractions, + backends={lane["name"]: lane["backend"] for lane in header["lanes"]}, + config_digest=header["parser_config_digest"], + ) + documents: list[dict[str, Any]] = [] + waiver_index = {(w.path, w.lane, w.outcome): w for w in waivers} + used: set[tuple[str, str, str]] = set() + outcome_counts: dict[str, Counter] = {c: Counter() for c in candidates} + unwaived: list[str] = [] + integrity: list[str] = [] + errors: list[str] = [] + missing = 0 + slowdowns: dict[str, list[float]] = {c: [] for c in candidates} + + for entry in corpus.entries: + observed = by_doc.get(entry.path, {}) + lane_rows: dict[str, dict[str, Any]] = {} + problems: list[str] = [] + records: dict[str, Any] = {} + for name in lanes: + obs = observed.get(name) + if obs is None: + missing += 1 + lane_rows[name] = {"status": INCOMPLETE} + continue + row = _lane_row(obs) + if obs.status in (ERROR, CONTAINMENT_ERROR): + errors.append(f"{entry.path} [{name}]: {obs.status}: {obs.detail}") + if obs.status == SEALED: + record, lane_problems, units = checker.check(obs, entry.sha256) + problems += [f"[{name}] {p}" for p in lane_problems] + if units is not None: + row["extraction_units"] = units + if record is not None: + records[name] = record + lane_rows[name] = row + integrity += [f"{entry.path} {p}" for p in problems] + + comparisons: dict[str, dict[str, Any]] = {} + base = observed.get(baseline) + for cand in candidates: + outcome, findings = _compare( + base, observed.get(cand), records.get(baseline), records.get(cand), + policy, ledger, + ) + row: dict[str, Any] = {"outcome": outcome} + if findings: + row["findings"] = findings[:_MAX_FINDINGS] + if len(findings) > _MAX_FINDINGS: + row["findings_truncated"] = len(findings) - _MAX_FINDINGS + key = (entry.path, cand, outcome) + if outcome in WAIVABLE: + if key in waiver_index: + used.add(key) + row["waived"] = waiver_index[key].reason + else: + unwaived.append(f"{entry.path} [{cand}]: {outcome}") + outcome_counts[cand][outcome] += 1 + comparisons[cand] = row + if outcome in (IDENTICAL, EQUIVALENT): + ratio = _ratio(base, observed.get(cand)) + if ratio is not None: + slowdowns[cand].append(ratio) + + doc: dict[str, Any] = { + "path": entry.path, "sha256": entry.sha256, "size": entry.size, + "lanes": lane_rows, "comparisons": comparisons, + } + if problems: + doc["problems"] = problems + documents.append(doc) + + ledger_report = verify_ledger(ledger) + log = EventLog(ledger) + try: + head = log.head() + finally: + log.close() + + summary = _summary(corpus, journal, lanes, candidates, outcome_counts, slowdowns) + gate_rows = _gates( + corpus=corpus, root=root, lanes=lanes, baseline=baseline, missing=missing, + errors=errors, integrity=integrity, ledger_report=ledger_report, + unwaived=unwaived, unused=[w for k, w in waiver_index.items() if k not in used], + summary=summary, gates=gates, slowdowns=slowdowns, + ) + data = { + "schema": REPORT_SCHEMA, + "schema_version": REPORT_VERSION, + "generated_at": datetime.now(timezone.utc).isoformat(), + "corpus": { + "id": corpus.corpus_id, "name": corpus.name, "documents": len(corpus), + "bytes": corpus.total_bytes, "excluded": [x.to_json() for x in corpus.excluded], + }, + "parser": parser, + "parser_config_digest": header["parser_config_digest"], + "lanes": header["lanes"], + "baseline": baseline, + "policy": policy.describe() if policy is not None else None, + "ledger": { + "head": None if head is None else {"seq": head[0], "hash": head[1]}, + "verified": ledger_report.ok, + }, + "thresholds": gates.to_json(), + "waivers": [w.to_json() for w in waivers], + "documents": documents, + "summary": summary, + "gates": gate_rows, + "passed": all(g["passed"] for g in gate_rows), + } + return VerificationReport(data) + + +def _lane_row(obs: Observation) -> dict[str, Any]: + row: dict[str, Any] = {"status": obs.status} + for key in ("backend", "record_id", "artifact_hash"): + if getattr(obs, key) is not None: + row[key] = getattr(obs, key) + if obs.failure is not None: + row["failure"] = {"reason": obs.failure.get("reason"), "detail": obs.failure.get("detail")} + if obs.detail: + row["detail"] = obs.detail + t = obs.telemetry or {} + for key in ("wall_time_seconds", "cpu_time_seconds", "peak_memory_bytes"): + if t.get(key) is not None: + row[key] = t[key] + return row + + +class _RecordChecker: + def __init__( + self, + ledger: LedgerStore, + parser: Mapping[str, str], + extractions: bool, + *, + backends: Mapping[str, str], + config_digest: str, + ) -> None: + self.ledger = ledger + self.parser = parser + self.extractions = extractions + self.backends = backends # lane name -> the backend that lane runs on + self.config_digest = config_digest + self.claimed: dict[str, str] = {} # record_id -> " []" that cited it first + + def check(self, obs: Observation, sha256: str) -> tuple[Any, list[str], int | None]: + """(record or None, problems, extraction unit count or None).""" + try: + record = self.ledger.get(obs.record_id) + except RecordNotFoundError: + return None, [f"record {obs.record_id} is not in the ledger"], None + problems = [] + if record.state is not ArtifactState.SEALED: + problems.append(f"record {record.record_id} is {record.state.value}, not sealed") + if record.run_id != obs.run_id or record.artifact_hash != obs.artifact_hash: + problems.append(f"record {record.record_id} is not the run the journal names") + if record.parser is None or (record.parser.name, record.parser.version) != ( + self.parser["name"], self.parser["version"] + ): + problems.append(f"record {record.record_id} was made by another parser") + # The journal says which record a lane produced; the ledger must agree + # that this lane's backend made it, with the campaign's configuration, + # or one lane's output could stand in for another's. + expected = self.backends.get(obs.lane) + if record.backend != expected: + problems.append( + f"record {record.record_id} was made on {record.backend!r}, " + f"not on lane {obs.lane}'s backend {expected!r}" + ) + if record.parser_config is None or config_digest(record.parser_config) != self.config_digest: + problems.append(f"record {record.record_id} was made with another parser configuration") + claimant = f"{obs.path} [{obs.lane}]" + first = self.claimed.setdefault(record.record_id, claimant) + if first != claimant: + problems.append(f"record {record.record_id} is already the result of {first}") + if record.source_hash != sha256: + problems.append( + f"record {record.record_id} read input {record.source_hash}, not the corpus document" + ) + validation = validate_artifact(record, self.ledger.archive) + if validation.status not in ("ok", "archived"): + problems.append( + f"record {record.record_id} does not verify: {validation.status} " + f"{list(validation.drifted_files or validation.missing_files)[:5]}" + ) + return record, problems, None + units = None + if self.extractions: + units, extraction_problems = _check_extraction(record, self.ledger) + problems += extraction_problems + return record, problems, units + + +def _check_extraction(record: Any, ledger: LedgerStore) -> tuple[int | None, list[str]]: + from ..extraction.normalizers import NormalizeError, normalizer_for + from ..extraction.resolver import BundleResolver + + bundle = SealedBundle.from_record(record, ledger.archive) + try: + normalizer = normalizer_for(bundle) + except NormalizeError: + return None, [] # no normalizer for this parser: nothing to check + try: + extraction = normalizer(bundle) + except Exception as exc: # noqa: BLE001 - a normalizer failure is a finding + return None, [f"record {record.record_id}: extraction failed: {exc}"] + problems = BundleResolver(bundle).verify(extraction) + return len(extraction.units), [ + f"record {record.record_id}: extraction: {p}" for p in problems[:_MAX_FINDINGS] + ] + + +def _compare( + base: Observation | None, + cand: Observation | None, + base_record: Any, + cand_record: Any, + policy: ComparisonPolicy | None, + ledger: LedgerStore, +) -> tuple[str, list[str]]: + if base is None or cand is None: + return INCOMPLETE, [] + statuses = {base.status, cand.status} + if statuses & {ERROR, CONTAINMENT_ERROR}: + return ERROR, [] + if base.status == NOT_APPLICABLE and cand.status == NOT_APPLICABLE: + return NOT_APPLICABLE, [] + if NOT_APPLICABLE in statuses: + # One lane accepted the document and the other did not: the lanes + # do not agree on what they run. + return FAILURE_MISMATCH, ["the parser accepted the document on one lane only"] + if base.status == SEALED and cand.status == SEALED: + if base_record is None or cand_record is None: + return ERROR, ["a sealed record is missing from the ledger"] + recorded, candidate = base_record.artifact_manifest, cand_record.artifact_manifest + if recorded == candidate: + return IDENTICAL, [] + if policy is not None: + verdict = policy.compare(recorded, candidate, ledger.archive) + if verdict.accepted: + return EQUIVALENT, [] + return DIVERGED, list(verdict.findings) + differing = sorted( + p for p in set(recorded) | set(candidate) if recorded.get(p) != candidate.get(p) + ) + return DIVERGED, [f"differs: {p}" for p in differing] + if base.status == SEALED: + return REGRESSION, [_failure_text(cand)] + if cand.status == SEALED: + return RECOVERED, [_failure_text(base)] + if _reason(base) == _reason(cand): + return CONSISTENT_FAILURE, [_failure_text(base)] + return FAILURE_MISMATCH, [f"baseline {_failure_text(base)}; candidate {_failure_text(cand)}"] + + +def _reason(obs: Observation) -> str | None: + return (obs.failure or {}).get("reason") + + +def _failure_text(obs: Observation) -> str: + failure = obs.failure or {} + return f"{failure.get('reason', 'failed')}: {failure.get('detail') or obs.detail or ''}".strip() + + +def _wall(obs: Observation | None) -> float | None: + if obs is None or obs.telemetry is None: + return None + return obs.telemetry.get("wall_time_seconds") + + +def _ratio(base: Observation | None, cand: Observation | None) -> float | None: + b, c = _wall(base), _wall(cand) + if b is None or c is None or b <= 0: + return None + return c / b + + +# --------------------------------------------------------------------------- +# Summary and gates +# --------------------------------------------------------------------------- + +def _percentile(values: Sequence[float], q: float) -> float | None: + if not values: + return None + ordered = sorted(values) + index = max(0, math.ceil(q * len(ordered)) - 1) + return round(ordered[index], 6) + + +def _stats(values: Iterable[float | None]) -> dict[str, Any]: + vals = [v for v in values if v is not None] + return { + "n": len(vals), "p50": _percentile(vals, 0.5), "p95": _percentile(vals, 0.95), + "max": round(max(vals), 6) if vals else None, + } + + +def _summary( + corpus: Corpus, + journal: Journal, + lanes: list[str], + candidates: list[str], + outcome_counts: dict[str, Counter], + slowdowns: dict[str, list[float]], +) -> dict[str, Any]: + latest: dict[tuple[str, str], Observation] = {} + wanted = corpus.by_path() + for obs in journal.observations: + if obs.path in wanted: + latest[(obs.path, obs.lane)] = obs + per_lane: dict[str, Any] = {} + for name in lanes: + observations = [o for (p, lane), o in latest.items() if lane == name] + statuses = Counter(o.status for o in observations) + reasons = Counter(_reason(o) or "unknown" for o in observations if o.status == FAILED) + per_lane[name] = { + "observed": len(observations), + "statuses": dict(sorted(statuses.items())), + "failure_reasons": dict(sorted(reasons.items())), + "wall_time_seconds": _stats((o.telemetry or {}).get("wall_time_seconds") + for o in observations), + "cpu_time_seconds": _stats((o.telemetry or {}).get("cpu_time_seconds") + for o in observations), + "peak_memory_bytes": _stats((o.telemetry or {}).get("peak_memory_bytes") + for o in observations), + } + return { + "lanes": per_lane, + "comparisons": { + cand: { + "outcomes": dict(sorted(outcome_counts[cand].items())), + "median_slowdown": ( + round(statistics.median(slowdowns[cand]), 6) if slowdowns[cand] else None + ), + } + for cand in candidates + }, + } + + +def _gate(name: str, passed: bool, detail: str, examples: Sequence[str] = ()) -> dict[str, Any]: + row: dict[str, Any] = {"name": name, "passed": bool(passed), "detail": detail} + if examples: + row["examples"] = list(examples[:_MAX_FINDINGS]) + return row + + +def _gates( + *, + corpus: Corpus, + root: Path | None, + lanes: list[str], + baseline: str, + missing: int, + errors: list[str], + integrity: list[str], + ledger_report: Any, + unwaived: list[str], + unused: list[Waiver], + summary: dict[str, Any], + gates: Gates, + slowdowns: dict[str, list[float]], +) -> list[dict[str, Any]]: + rows = [] + if root is None: + rows.append(_gate("corpus_unchanged", False, "the corpus was not re-hashed (no root given)")) + else: + changed = corpus.check(root) + rows.append(_gate( + "corpus_unchanged", not changed, + "every document still hashes as in the manifest" if not changed + else f"{len(changed)} difference(s) from the manifest", changed, + )) + expected = len(corpus) * len(lanes) + rows.append(_gate( + "coverage", missing == 0 and len(corpus) > 0, + f"{expected - missing}/{expected} document-lane runs observed" + + ("" if len(corpus) else "; the corpus is empty"), + )) + base_sealed = summary["lanes"][baseline]["statuses"].get(SEALED, 0) + rows.append(_gate( + "baseline_produced_output", base_sealed > 0, + f"the baseline sealed {base_sealed} document(s)", + )) + rows.append(_gate( + "no_errors", not errors, + "no Stele, host or containment errors" if not errors + else f"{len(errors)} run(s) ended in an error", errors, + )) + rows.append(_gate( + "records_verified", not integrity, + "every sealed record verifies against the ledger, archive and corpus" if not integrity + else f"{len(integrity)} integrity problem(s)", integrity, + )) + rows.append(_gate( + "ledger_verified", ledger_report.ok, + "the event chain and the ledger tables agree" if ledger_report.ok + else "the ledger does not verify", + [*ledger_report.chain.problems, *ledger_report.problems], + )) + rows.append(_gate( + "candidates_agree", not unwaived, + "every candidate agrees with the baseline or is waived" if not unwaived + else f"{len(unwaived)} unwaived finding(s)", unwaived, + )) + rows.append(_gate( + "waivers_used", not unused, + "every waiver matches a finding" if not unused + else f"{len(unused)} waiver(s) match nothing", + [f"{w.path} [{w.lane}]: {w.outcome}" for w in unused], + )) + if gates.max_baseline_failure_rate is not None: + statuses = summary["lanes"][baseline]["statuses"] + applicable = sum(n for s, n in statuses.items() if s != NOT_APPLICABLE) + rate = statuses.get(FAILED, 0) / applicable if applicable else 0.0 + rows.append(_gate( + "baseline_failure_rate", rate <= gates.max_baseline_failure_rate, + f"{rate:.4%} of applicable documents failed on the baseline " + f"(limit {gates.max_baseline_failure_rate:.4%})", + )) + if gates.max_slowdown is not None: + slow = [ + f"{cand}: median {statistics.median(v):.2f}x" for cand, v in slowdowns.items() + if v and statistics.median(v) > gates.max_slowdown + ] + rows.append(_gate( + "slowdown", not slow, + f"median candidate/baseline wall time within {gates.max_slowdown}x" if not slow + else f"{len(slow)} lane(s) slower than {gates.max_slowdown}x", slow, + )) + return rows diff --git a/stele/verification/signoff.py b/stele/verification/signoff.py new file mode 100644 index 0000000..a3a6b35 --- /dev/null +++ b/stele/verification/signoff.py @@ -0,0 +1,204 @@ +""" +Sign-off: a person accepting a passed verification report, on the record. + +sign_off() does not take the report's word for anything. It rebuilds the +report from the ledger, the journal and the corpus (with the report's own +thresholds and waivers) and refuses unless the result is the same report, so +an edited report, with its gates flipped to passed and re-encoded, is +refused. It also refuses unless every gate passed, the ledger's chain still +contains the head the report was built at (nothing was rewritten since), and +every record the report relies on is still SEALED with the artifact hash it +names. It then stores the report's canonical bytes +in the evidence archive and appends a `verification.signed_off` event to the +hash-chained event log, whose body names the report digest, the corpus id, +the parser, the lanes and who signed off. The event log is what makes the +sign-off tamper-evident; publishing the resulting head (seq, hash) outside +the ledger makes it hold against a rewritten ledger too. + +check_sign_off() re-verifies a sign-off later: the event exists and is +chained, the archived report still hashes to its digest, and every record it +covers is still sealed. A sign-off whose records were since invalidated is +reported as stale, never silently kept. +""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from ..archive.store import ArchiveError +from ..ledger.events import EventLog, append_event, write_transaction +from ..ledger.models import ArtifactState +from ..ledger.store import LedgerStore, RecordNotFoundError, connect +from ..replay.policy import ComparisonPolicy +from .campaign import Journal +from .corpus import Corpus +from .report import Gates, ReportFormatError, VerificationReport, Waiver, build_report + +SIGNOFF_EVENT = "verification.signed_off" + + +class SignOffRefused(Exception): + """The report cannot be signed off; .problems says why.""" + + def __init__(self, problems: list[str]) -> None: + super().__init__("; ".join(problems[:5]) + (" ..." if len(problems) > 5 else "")) + self.problems = problems + + +@dataclass(frozen=True) +class SignOff: + report_digest: str + corpus_id: str + signed_off_by: str + seq: int # the event's position in the chain + event_hash: str # publish (seq, event_hash) to anchor the sign-off + + +def recheck(ledger: LedgerStore, report: VerificationReport) -> list[str]: + """Why the report no longer describes the ledger; [] if it still does.""" + problems: list[str] = [] + log = EventLog(ledger) + try: + chain = log.verify(anchor=report.ledger_head) + finally: + log.close() + problems += [f"event log: {p}" for p in chain.problems] + for doc in report.data["documents"]: + for lane, row in doc["lanes"].items(): + record_id = row.get("record_id") + if not record_id: + continue + try: + record = ledger.get(record_id) + except RecordNotFoundError: + problems.append(f"{doc['path']} [{lane}]: record {record_id} is gone") + continue + if record.state is not ArtifactState.SEALED: + problems.append( + f"{doc['path']} [{lane}]: record {record_id} is now {record.state.value}" + ) + elif record.artifact_hash != row.get("artifact_hash"): + problems.append(f"{doc['path']} [{lane}]: record {record_id} changed") + return problems + + +def _comparable(data: dict[str, Any]) -> dict[str, Any]: + """The report without what legitimately differs between two builds of it.""" + return { + **data, "generated_at": None, + "ledger": {**data["ledger"], "head": None}, + } + + +def rebuild_problems( + ledger: LedgerStore, + report: VerificationReport, + *, + corpus: Corpus, + journal: Journal, + root: Path, + policy: ComparisonPolicy | None = None, +) -> list[str]: + """[] if building the report again from its inputs gives the same report.""" + thresholds = report.data["thresholds"] + try: + again = build_report( + ledger, corpus, journal, root=root, policy=policy, + gates=Gates(**thresholds), + waivers=[Waiver(**w) for w in report.data["waivers"]], + ) + except (TypeError, ValueError) as exc: + return [f"the report cannot be rebuilt from its inputs: {exc}"] + if _comparable(again.data) == _comparable(report.data): + return [] + differing = sorted( + k for k in set(again.data) | set(report.data) + if _comparable(again.data).get(k) != _comparable(report.data).get(k) + ) + return [f"the report does not match a rebuild from the ledger, journal and corpus " + f"(differs in: {', '.join(differing)})"] + + +def sign_off( + ledger: LedgerStore, + report: VerificationReport, + *, + corpus: Corpus, + journal: Journal, + root: Path, + signed_off_by: str, + policy: ComparisonPolicy | None = None, + note: str = "", +) -> SignOff: + if not isinstance(signed_off_by, str) or not signed_off_by.strip(): + raise ValueError("a sign-off names the person who gives it") + problems = rebuild_problems( + ledger, report, corpus=corpus, journal=journal, root=root, policy=policy + ) + problems += [ + f"gate {g['name']} failed: {g['detail']}" for g in report.failed_gates() + ] + if not report.data["ledger"]["verified"]: + problems.append("the report was built on a ledger that did not verify") + problems += recheck(ledger, report) + if problems: + raise SignOffRefused(problems) + + digest = ledger.archive.put_bytes(report.to_canonical()) + if digest != report.digest: + raise RuntimeError("the archive stored the report under another digest") + data = report.data + body = { + "report_digest": digest, + "corpus_id": data["corpus"]["id"], + "corpus_name": data["corpus"]["name"], + "documents": data["corpus"]["documents"], + "parser": data["parser"], + "lanes": data["lanes"], + "baseline": data["baseline"], + "report_ledger_head": data["ledger"]["head"], + "waivers": len(data["waivers"]), + "signed_off_by": signed_off_by.strip(), + "note": note, + } + conn = connect(ledger.db_path) + try: + with write_transaction(conn): + event = append_event(conn, SIGNOFF_EVENT, data["corpus"]["id"], body) + finally: + conn.close() + return SignOff(digest, data["corpus"]["id"], body["signed_off_by"], event.seq, event.hash) + + +def sign_offs(ledger: LedgerStore) -> list[dict]: + """Every sign-off event: [{seq, hash, at, **body}], oldest first.""" + log = EventLog(ledger) + try: + return [ + {"seq": e.seq, "hash": e.hash, "at": e.at, **e.body} + for e in log.events() if e.kind == SIGNOFF_EVENT + ] + finally: + log.close() + + +def check_sign_off(ledger: LedgerStore, report_digest: str) -> list[str]: + """Problems with a recorded sign-off; [] means it still holds.""" + matches = [s for s in sign_offs(ledger) if s.get("report_digest") == report_digest] + if not matches: + return [f"no sign-off of report {report_digest} in the event log"] + log = EventLog(ledger) + try: + chain = log.verify(anchor=(matches[-1]["seq"], matches[-1]["hash"])) + finally: + log.close() + problems = [f"event log: {p}" for p in chain.problems] + try: + report = VerificationReport.from_canonical(ledger.archive.read(report_digest)) + except (ArchiveError, ReportFormatError) as exc: + return problems + [f"the signed-off report is unreadable: {exc}"] + if not report.passed: + problems.append("the signed-off report did not pass") + problems += [f"stale: {p}" for p in recheck(ledger, report)] + return problems diff --git a/tests/test_verification.py b/tests/test_verification.py new file mode 100644 index 0000000..719033d --- /dev/null +++ b/tests/test_verification.py @@ -0,0 +1,682 @@ +""" +Full-corpus production verification and sign-off (roadmap #15). + + 1. A corpus is pinned by content: every regular file hashed, everything left + out listed with its reason; strict canonical manifest; check() catches a + changed, missing or added document. + 2. A campaign runs every document on the baseline and each candidate lane, + seals successful runs, journals every observation (failures with their + structured reason) and resumes where it stopped. + 3. The report re-checks every sealed record against the ledger, archive, + corpus and extraction resolver, compares candidates with the baseline + (IDENTICAL, EQUIVALENT under the parser's policy, DIVERGED, REGRESSION, + RECOVERED, CONSISTENT_FAILURE, FAILURE_MISMATCH) and decides the gates. + Only comparison findings can be waived, each with a reason, and a waiver + that matches nothing fails the gates. + 4. Sign-off is refused unless every gate passes and the report still holds; + it is anchored in the event log and goes stale when a record it covers + is invalidated. + 5. End to end with real sandboxes (bubblewrap baseline; OCI candidate when + an engine is available) and through the CLI. +""" +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Callable + +import pytest + +from stele.containment.backend import ( + Capability, + ExecutionOutcome, + ParserRequirements, + SandboxBackend, +) +from stele.containment.sandbox import SandboxConfig +from stele.ledger.events import EventLog, verify_ledger +from stele.replay.parsers import ParserSpec +from stele.replay.policy import JsonTolerancePolicy +from stele.verification import ( + SIGNOFF_EVENT, + Campaign, + CampaignError, + Corpus, + CorpusFormatError, + Gates, + Journal, + Lane, + SignOffRefused, + VerificationReport, + Waiver, + build_report, + check_sign_off, + sign_off, +) +from stele.verification.report import ( + CONSISTENT_FAILURE, + DIVERGED, + EQUIVALENT, + IDENTICAL, + RECOVERED, + REGRESSION, +) +from stele.verification.campaign import SEALED +from tests.ledger_helpers import open_ledger + +# --------------------------------------------------------------------------- +# Fixtures: a corpus and in-process lanes +# --------------------------------------------------------------------------- + +DOCS = { + "a.txt": b"alpha document\n", + "nested/b.txt": b"bravo document\n", + "nested/deeper/c.txt": "charlie — ünïcode\n".encode(), + "fail.txt": b"FAIL: this document breaks the parser\n", +} + + +def _corpus_dir(tmp_path: Path, docs: dict[str, bytes] = DOCS) -> Path: + root = tmp_path / "corpus" + for rel, data in docs.items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + return root + + +# (document bytes) -> (exit code, {relative path: bytes}) +Behaviour = Callable[[bytes], tuple[int, dict[str, bytes]]] + + +def honest(data: bytes) -> tuple[int, dict[str, bytes]]: + if data.startswith(b"FAIL"): + return 3, {} + words = data.decode().split() + return 0, {"out.json": json.dumps({"words": words, "score": 0.5}).encode()} + + +class InProcessBackend(SandboxBackend): + """A lane whose 'parser' is a Python function (no sandbox: tests only).""" + + def __init__(self, name: str, behaviour: Behaviour) -> None: + self.name = name + self.behaviour = behaviour + + def capabilities(self): + return frozenset(Capability) + + def available(self): + return True + + def unavailable_reason(self): + return "" + + def execute(self, config: SandboxConfig) -> ExecutionOutcome: + code, files = self.behaviour(config.input_path.read_bytes()) + for rel, data in files.items(): + (config.artifact_dir / rel).parent.mkdir(parents=True, exist_ok=True) + (config.artifact_dir / rel).write_bytes(data) + return ExecutionOutcome(exit_code=code, stdout="", stderr="", wall_time_seconds=0.01) + + +def _spec(name: str = "probe", version: str = "1") -> ParserSpec: + def build(input_path, artifact_dir, config): + if input_path.suffix == ".bin": + raise ValueError("probe does not accept .bin") + return SandboxConfig(command=["probe"], artifact_dir=artifact_dir, input_path=input_path) + + return ParserSpec(name=name, version=version, requirements=ParserRequirements(), + build_config=build) + + +def _lane(name: str, behaviour: Behaviour, spec: ParserSpec | None = None) -> Lane: + return Lane(name, InProcessBackend(name, behaviour), spec or _spec()) + + +def _campaign(tmp_path, lanes, *, root=None, ledger=None, journal="journal.jsonl", **kw): + root = root or _corpus_dir(tmp_path) + ledger = ledger or open_ledger(tmp_path / "ledger.db") + corpus = Corpus.build(root) + campaign = Campaign(ledger, corpus, root, lanes, baseline=lanes[0].name, + journal=tmp_path / journal, **kw) + return campaign, ledger, corpus, root + + +def _report(campaign, ledger, corpus, root, **kw) -> VerificationReport: + return build_report(ledger, corpus, campaign.journal, root=root, **kw) + + +def _gate(report: VerificationReport, name: str) -> dict: + return next(g for g in report.data["gates"] if g["name"] == name) + + +def _doc(report: VerificationReport, path: str) -> dict: + return next(d for d in report.data["documents"] if d["path"] == path) + + +# --------------------------------------------------------------------------- +# 1 — the corpus +# --------------------------------------------------------------------------- + +class TestCorpus: + + def test_build_hashes_every_regular_file(self, tmp_path) -> None: + root = _corpus_dir(tmp_path) + corpus = Corpus.build(root, name="prod") + assert corpus.name == "prod" + assert [e.path for e in corpus.entries] == sorted(DOCS) + assert corpus.by_path()["a.txt"].size == len(DOCS["a.txt"]) + assert corpus.total_bytes == sum(len(d) for d in DOCS.values()) + assert corpus.excluded == () + assert corpus.check(root) == [] + + @pytest.mark.skipif(not hasattr(os, "mkfifo"), + reason="Windows feature: no FIFOs or unprivileged symlinks") + def test_everything_left_out_is_listed(self, tmp_path) -> None: + root = _corpus_dir(tmp_path) + os.symlink(root / "a.txt", root / "link.txt") + os.symlink(root / "nested", root / "linked-dir") + os.mkfifo(root / "pipe") + corpus = Corpus.build(root) + assert [e.path for e in corpus.entries] == sorted(DOCS) + assert [(x.path, x.why) for x in corpus.excluded] == [ + ("link.txt", "symlink"), ("linked-dir", "symlink"), ("pipe", "not a regular file"), + ] + assert corpus.check(root) == [] + + def test_canonical_round_trip_and_strictness(self, tmp_path) -> None: + corpus = Corpus.build(_corpus_dir(tmp_path)) + data = corpus.to_canonical() + assert Corpus.from_canonical(data) == corpus + assert Corpus.from_canonical(data).corpus_id == corpus.corpus_id + pretty = json.dumps(json.loads(data), indent=1).encode() + with pytest.raises(CorpusFormatError, match="canonical"): + Corpus.from_canonical(pretty) + obj = json.loads(data) + obj["entries"][0]["path"] = "../escape" + with pytest.raises(CorpusFormatError, match="relative"): + Corpus.from_canonical(json.dumps(obj, sort_keys=True, separators=(",", ":")).encode()) + + def test_check_catches_changed_missing_and_added(self, tmp_path) -> None: + root = _corpus_dir(tmp_path) + corpus = Corpus.build(root) + (root / "a.txt").write_bytes(b"alpha, edited\n") + (root / "nested" / "b.txt").unlink() + (root / "new.txt").write_bytes(b"new") + assert sorted(corpus.check(root)) == [ + "a.txt: content changed since the manifest was taken", + "nested/b.txt: missing", + "new.txt: not in the corpus manifest", + ] + + +# --------------------------------------------------------------------------- +# 2 — the campaign +# --------------------------------------------------------------------------- + +class TestCampaign: + + def test_every_document_on_every_lane(self, tmp_path) -> None: + campaign, ledger, corpus, _ = _campaign( + tmp_path, [_lane("base", honest), _lane("cand", honest)] + ) + observed = campaign.run() + assert len(observed) == 2 * len(DOCS) + by = {(o.path, o.lane): o for o in observed} + sealed = by[("a.txt", "base")] + assert sealed.status == "sealed" + record = ledger.get(sealed.record_id) + assert record.source_hash == corpus.by_path()["a.txt"].sha256 + assert record.parser.name == "probe" and record.backend == "base" + failed = by[("fail.txt", "cand")] + assert failed.status == "failed" and failed.record_id is None + assert failed.failure["reason"] == "exit_status" and failed.failure["exit_code"] == 3 + assert failed.telemetry["wall_time_seconds"] is not None + # Working outputs are dropped once sealed; the archive holds them. + assert not any(p.is_file() for p in campaign.work_dir.rglob("*")) + + def test_resumes_where_it_stopped(self, tmp_path) -> None: + lanes = [_lane("base", honest), _lane("cand", honest)] + campaign, ledger, corpus, root = _campaign(tmp_path, lanes) + assert len(campaign.run(limit=3)) == 3 + again = Campaign(ledger, corpus, root, lanes, baseline="base", + journal=tmp_path / "journal.jsonl") + assert len(again.journal.observations) == 3 + rest = again.run() + assert len(rest) == 2 * len(DOCS) - 3 + assert not list(again.pending()) + assert len(Journal.read(tmp_path / "journal.jsonl").observations) == 2 * len(DOCS) + + def test_torn_last_line_is_dropped(self, tmp_path) -> None: + lanes = [_lane("base", honest)] + campaign, ledger, corpus, root = _campaign(tmp_path, lanes) + campaign.run(limit=2) + with open(tmp_path / "journal.jsonl", "ab") as f: + f.write(b'{"path":"nested/b.txt","la') # a crash mid-append + again = Campaign(ledger, corpus, root, lanes, baseline="base", + journal=tmp_path / "journal.jsonl") + assert len(again.journal.observations) == 2 + again.run() + assert len(Journal.read(tmp_path / "journal.jsonl").observations) == len(DOCS) + + def test_journal_of_another_campaign_is_refused(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest)]) + campaign.run(limit=1) + with pytest.raises(CampaignError, match="another campaign"): + Campaign(ledger, corpus, root, [_lane("base", honest)], baseline="base", + journal=tmp_path / "journal.jsonl", parser_config={"mode": "fast"}) + + def test_lanes_must_run_one_parser(self, tmp_path) -> None: + with pytest.raises(CampaignError, match="same parser"): + _campaign(tmp_path, [_lane("base", honest), _lane("cand", honest, _spec(version="2"))]) + + def test_not_applicable_documents(self, tmp_path) -> None: + docs = {**DOCS, "blob.bin": b"\x00\x01"} + campaign, *_ = _campaign(tmp_path, [_lane("base", honest)], + root=_corpus_dir(tmp_path, docs)) + by = {o.path: o for o in campaign.run()} + assert by["blob.bin"].status == "not_applicable" + assert "does not accept" in by["blob.bin"].detail + + def test_a_document_changed_mid_campaign_is_an_error(self, tmp_path) -> None: + campaign, _, _, root = _campaign(tmp_path, [_lane("base", honest)]) + (root / "a.txt").write_bytes(b"swapped after the manifest\n") + by = {o.path: o for o in campaign.run()} + assert by["a.txt"].status == "error" and "changed after the manifest" in by["a.txt"].detail + + +# --------------------------------------------------------------------------- +# 3 — the report and its gates +# --------------------------------------------------------------------------- + +def noisy(data: bytes): + """Same words, a score that moved by a tolerable amount.""" + code, files = honest(data) + if code: + return code, files + obj = json.loads(files["out.json"]) + obj["score"] = 0.5000001 + return 0, {"out.json": json.dumps(obj).encode()} + + +def wrong(data: bytes): + code, files = honest(data) + if code: + return code, files + return 0, {"out.json": json.dumps({"words": ["something", "else"], "score": 0.5}).encode()} + + +def flaky(data: bytes): + """Fails on 'bravo'; succeeds where the baseline fails.""" + if b"bravo" in data: + return 1, {} + if data.startswith(b"FAIL"): + return 0, {"out.json": b"{}"} + return honest(data) + + +class TestReport: + + def test_clean_campaign_passes_every_gate(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("cand", honest)]) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert report.passed, report.failed_gates() + assert [g["name"] for g in report.data["gates"]] == [ + "corpus_unchanged", "coverage", "baseline_produced_output", "no_errors", + "records_verified", "ledger_verified", "candidates_agree", "waivers_used", + ] + assert _doc(report, "a.txt")["comparisons"]["cand"]["outcome"] == IDENTICAL + assert _doc(report, "fail.txt")["comparisons"]["cand"]["outcome"] == CONSISTENT_FAILURE + lane = report.data["summary"]["lanes"]["base"] + assert lane["statuses"] == {"failed": 1, "sealed": 3} + assert lane["failure_reasons"] == {"exit_status": 1} + assert lane["wall_time_seconds"]["n"] == 4 + again = VerificationReport.from_canonical(report.to_canonical()) + assert again.digest == report.digest and again.passed + + def test_outcomes_against_the_baseline(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [ + _lane("base", honest), _lane("noisy", noisy), _lane("wrong", wrong), + _lane("flaky", flaky), + ]) + campaign.run() + policy = JsonTolerancePolicy(abs_tol=1e-3) + report = _report(campaign, ledger, corpus, root, policy=policy) + a = _doc(report, "a.txt")["comparisons"] + assert a["noisy"]["outcome"] == EQUIVALENT + assert a["wrong"]["outcome"] == DIVERGED and a["wrong"]["findings"] + assert _doc(report, "nested/b.txt")["comparisons"]["flaky"]["outcome"] == REGRESSION + assert _doc(report, "fail.txt")["comparisons"]["flaky"]["outcome"] == RECOVERED + assert not _gate(report, "candidates_agree")["passed"] + assert report.data["policy"] == policy.describe() + # Without the parser's policy, the tolerable difference is a divergence too. + strict = _report(campaign, ledger, corpus, root) + assert _doc(strict, "a.txt")["comparisons"]["noisy"]["outcome"] == DIVERGED + + def test_waivers_accept_findings_and_must_all_match(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("flaky", flaky)]) + campaign.run() + waiver = Waiver("nested/b.txt", "flaky", REGRESSION, "known upstream bug #123") + report = _report(campaign, ledger, corpus, root, waivers=[waiver]) + assert report.passed, report.failed_gates() + assert _doc(report, "nested/b.txt")["comparisons"]["flaky"]["waived"] == waiver.reason + unused = Waiver("a.txt", "flaky", DIVERGED, "no longer needed") + report = _report(campaign, ledger, corpus, root, waivers=[waiver, unused]) + assert not _gate(report, "waivers_used")["passed"] + with pytest.raises(ValueError, match="integrity"): + Waiver("a.txt", "flaky", "error", "hide it") + with pytest.raises(ValueError, match="reason"): + Waiver("a.txt", "flaky", REGRESSION, " ") + + def test_incomplete_campaign_fails_coverage(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("cand", honest)]) + campaign.run(limit=5) + report = _report(campaign, ledger, corpus, root) + assert not _gate(report, "coverage")["passed"] + assert "5/8" in _gate(report, "coverage")["detail"] + + def test_integrity_problems_fail_the_report(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest)]) + campaign.run() + record_id = next(o.record_id for o in campaign.journal.observations if o.path == "a.txt") + ledger.invalidate(record_id, "manual") + report = _report(campaign, ledger, corpus, root) + gate = _gate(report, "records_verified") + assert not gate["passed"] and "invalidated" in gate["examples"][0] + + def _forge_journal(self, campaign, rewrite) -> Journal: + """Rewrite the journal's observations (not its header) and reopen it.""" + path = campaign.journal.path + header, *lines = path.read_text().splitlines() + observations = [json.loads(line) for line in lines] + forged = [header, *(json.dumps(o, sort_keys=True) for o in rewrite(observations))] + path.write_text("\n".join(forged) + "\n") + return Journal.read(path) + + def test_a_lane_cannot_claim_another_lanes_record(self, tmp_path) -> None: + """A candidate that diverges cannot pass by citing the baseline's records.""" + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("cand", wrong)]) + campaign.run() + assert not _gate(_report(campaign, ledger, corpus, root), "candidates_agree")["passed"] + + def cite_baseline(observations): + base = {o["path"]: o for o in observations if o["lane"] == "base"} + return [ + dict(base[o["path"]], lane="cand") + if o["lane"] == "cand" and base[o["path"]]["status"] == SEALED else o + for o in observations + ] + + journal = self._forge_journal(campaign, cite_baseline) + report = build_report(ledger, corpus, journal, root=root) + gate = _gate(report, "records_verified") + assert not report.passed and not gate["passed"] + assert any("not on lane cand's backend" in e for e in gate["examples"]) + assert any("already the result of" in e for e in gate["examples"]) + with pytest.raises(SignOffRefused, match="records_verified"): + sign_off(ledger, report, corpus=corpus, journal=journal, root=root, + signed_off_by="reviewer") + + def test_a_record_made_with_another_config_is_refused(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest)]) + campaign.run() + other, *_ = _campaign(tmp_path, [_lane("base", honest)], root=root, ledger=ledger, + journal="other.jsonl", parser_config={"mode": "lenient"}) + other.run() + swapped = {o.path: o for o in other.journal.observations} + + def use_other_config(observations): + return [ + {**o, **{k: getattr(swapped[o["path"]], k) + for k in ("run_id", "record_id", "artifact_hash")}} + if o["status"] == SEALED else o + for o in observations + ] + + journal = self._forge_journal(campaign, use_other_config) + gate = _gate(build_report(ledger, corpus, journal, root=root), "records_verified") + assert not gate["passed"] + assert any("another parser configuration" in e for e in gate["examples"]) + + def test_tampered_evidence_fails_the_report(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest)]) + campaign.run() + record = ledger.get(campaign.journal.observations[0].record_id) + blob = ledger.archive._blob_path(record.artifact_manifest["out.json"]) + blob.chmod(0o644) + blob.write_bytes(b'{"words":[],"score":9}') + gate = _gate(_report(campaign, ledger, corpus, root), "records_verified") + assert not gate["passed"] and "does not verify" in gate["examples"][0] + + def test_changed_or_unchecked_corpus_fails(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest)]) + campaign.run() + assert not _gate(_report(campaign, ledger, corpus, None), "corpus_unchanged")["passed"] + (root / "extra.txt").write_bytes(b"late arrival") + assert not _gate(_report(campaign, ledger, corpus, root), "corpus_unchanged")["passed"] + + def test_errors_and_thresholds(self, tmp_path) -> None: + def broken(data): + raise RuntimeError("host fault") + + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("cand", broken)]) + campaign.run() + report = _report(campaign, ledger, corpus, root, + gates=Gates(max_baseline_failure_rate=0.2, max_slowdown=100)) + assert not _gate(report, "no_errors")["passed"] + assert "host fault" in _gate(report, "no_errors")["examples"][0] + rate = _gate(report, "baseline_failure_rate") + assert not rate["passed"] and "25.0000%" in rate["detail"] + + def test_extractions_are_verified(self, tmp_path) -> None: + def markdown(data: bytes): + if data.startswith(b"FAIL"): + return 3, {} + return 0, { + "document.md": b"# Title\n\n" + data, + "stele-parser.json": json.dumps({"outputs": {"markdown": "document.md"}}).encode(), + } + + def bad_markdown(data: bytes): + code, files = markdown(data) + if code == 0 and b"bravo" in data: + files["document.md"] = b"caf\xe9\n" + return code, files + + spec = _spec(name="marker", version="test") + campaign, ledger, corpus, root = _campaign( + tmp_path, [_lane("base", markdown, spec), _lane("cand", bad_markdown, spec)] + ) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert _doc(report, "a.txt")["lanes"]["base"]["extraction_units"] == 2 + gate = _gate(report, "records_verified") + assert not gate["passed"] and "extraction failed" in gate["examples"][0] + + +# --------------------------------------------------------------------------- +# 4 — sign-off +# --------------------------------------------------------------------------- + +class TestSignOff: + + def _passed(self, tmp_path): + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("cand", honest)]) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert report.passed + self.inputs = {"corpus": corpus, "journal": campaign.journal, "root": root} + return ledger, report + + def test_sign_off_is_anchored_and_checkable(self, tmp_path) -> None: + ledger, report = self._passed(tmp_path) + done = sign_off(ledger, report, **self.inputs, signed_off_by="C. Snyder", + note="Q3 production") + assert done.report_digest == report.digest + log = EventLog(ledger) + [event] = [e for e in log.events() if e.kind == SIGNOFF_EVENT] + log.close() + assert (event.seq, event.hash) == (done.seq, done.event_hash) + assert event.body["report_digest"] == report.digest + assert event.body["signed_off_by"] == "C. Snyder" + assert event.body["corpus_id"] == report.corpus_id + assert ledger.archive.read(report.digest) == report.to_canonical() + assert check_sign_off(ledger, report.digest) == [] + assert verify_ledger(ledger).ok + + def test_failed_report_is_refused(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("wrong", wrong)]) + campaign.run() + with pytest.raises(SignOffRefused, match="candidates_agree"): + sign_off(ledger, _report(campaign, ledger, corpus, root), corpus=corpus, + journal=campaign.journal, root=root, signed_off_by="x") + + def test_report_that_no_longer_holds_is_refused(self, tmp_path) -> None: + ledger, report = self._passed(tmp_path) + ledger.invalidate(report.record_ids()[0], "found a problem") + with pytest.raises(SignOffRefused, match="does not match a rebuild") as refused: + sign_off(ledger, report, **self.inputs, signed_off_by="x") + assert any("now invalidated" in p for p in refused.value.problems) + + def test_edited_report_is_refused(self, tmp_path) -> None: + campaign, ledger, corpus, root = _campaign(tmp_path, [_lane("base", honest), + _lane("wrong", wrong)]) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert not report.passed + obj = json.loads(report.to_canonical()) + obj["passed"] = True + for gate in obj["gates"]: + gate["passed"] = True + with pytest.raises(Exception, match="canonical"): + VerificationReport.from_canonical(json.dumps(obj, indent=1).encode()) + # Re-encoded canonically it parses, but it is not what the evidence says. + forged = VerificationReport.from_canonical( + json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() + ) + assert forged.passed + with pytest.raises(SignOffRefused, match="does not match a rebuild"): + sign_off(ledger, forged, corpus=corpus, journal=campaign.journal, root=root, + signed_off_by="x") + + def test_invalidation_makes_a_sign_off_stale(self, tmp_path) -> None: + ledger, report = self._passed(tmp_path) + sign_off(ledger, report, **self.inputs, signed_off_by="x") + ledger.invalidate(report.record_ids()[0], "source_changed") + problems = check_sign_off(ledger, report.digest) + assert problems and all(p.startswith("stale:") for p in problems) + assert check_sign_off(ledger, "0" * 64) == [f"no sign-off of report {'0' * 64} in the event log"] + + +# --------------------------------------------------------------------------- +# 5 — real sandboxes and the CLI +# --------------------------------------------------------------------------- + +PARSER_SCRIPT = ''' +import hashlib, json, os, sys +data = open(os.environ["STELE_INPUT_PATH"], "rb").read() +if data.startswith(b"FAIL"): + sys.exit(3) +out = os.environ.get("STELE_OUTPUT_DIR", "/stele/output") +with open(os.path.join(out, "summary.json"), "w") as f: + json.dump({"sha256": hashlib.sha256(data).hexdigest(), "lines": data.count(b"\\n")}, f, + sort_keys=True) +''' + + +def _script(tmp_path: Path) -> Path: + path = tmp_path / "summarize.py" + path.write_text(PARSER_SCRIPT) + return path + + +def _require(backend_name: str): + from stele.verification.lanes import backend_named + + backend = backend_named(backend_name) + if not backend.available(): + pytest.skip(f"{backend_name} unavailable: {backend.unavailable_reason()}") + + +class TestRealSandboxes: + + def test_bubblewrap_baseline_end_to_end(self, tmp_path) -> None: + from stele.verification.lanes import script_lanes + + _require("bubblewrap") + lanes = script_lanes(_script(tmp_path), ["bubblewrap"], deterministic=False) + campaign, ledger, corpus, root = _campaign(tmp_path, lanes) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert report.passed, report.failed_gates() + statuses = report.data["summary"]["lanes"]["bubblewrap"]["statuses"] + assert statuses == {"failed": 1, "sealed": 3} + sealed = ledger.get(_doc(report, "a.txt")["lanes"]["bubblewrap"]["record_id"]) + summary = json.loads(ledger.archive.read(sealed.artifact_manifest["summary.json"])) + assert summary["sha256"] == corpus.by_path()["a.txt"].sha256 + assert sealed.parser.version.startswith("sha256-") + + def test_oci_candidate_against_the_bubblewrap_baseline(self, tmp_path) -> None: + from stele.verification.lanes import script_lanes + + _require("bubblewrap") + _require("oci-runc") + lanes = script_lanes(_script(tmp_path), ["bubblewrap", "oci-runc"]) + campaign, ledger, corpus, root = _campaign(tmp_path, lanes) + campaign.run() + report = _report(campaign, ledger, corpus, root) + assert report.passed, report.failed_gates() + outcomes = report.data["summary"]["comparisons"]["oci-runc"]["outcomes"] + assert outcomes == {IDENTICAL: 3, CONSISTENT_FAILURE: 1} + + def test_cli_corpus_run_report_sign_off_check(self, tmp_path, capsys) -> None: + from stele.verification.__main__ import main + + _require("bubblewrap") + root = _corpus_dir(tmp_path) + store = ["--ledger", str(tmp_path / "ledger.db"), "--archive", str(tmp_path / "archive")] + manifest, journal, out = tmp_path / "corpus.json", tmp_path / "j.jsonl", tmp_path / "r.json" + campaign = [*store, "--corpus", str(manifest), "--root", str(root), "--journal", str(journal)] + assert main(["corpus", str(root), "-o", str(manifest)]) == 0 + assert json.loads(capsys.readouterr().out)["documents"] == len(DOCS) + assert main(["run", *campaign, "--script", str(_script(tmp_path)), "--quiet", + "--limit", "2"]) == 0 + assert json.loads(capsys.readouterr().out) == {"observed": 2, "remaining": 2} + assert main(["report", *campaign, "-o", str(out)]) == 1 # incomplete + assert "FAIL coverage" in capsys.readouterr().out + assert main(["run", *campaign, "--script", str(_script(tmp_path)), "--quiet"]) == 0 + capsys.readouterr() + assert main(["report", *campaign, "-o", str(out)]) == 0 + digest = VerificationReport.from_canonical(out.read_bytes()).digest + capsys.readouterr() + assert main(["sign-off", *campaign, str(out), "--by", "operator"]) == 0 + signed = json.loads(capsys.readouterr().out) + assert signed["report_digest"] == digest and ":" in signed["anchor"] + assert main(["check", *store, digest]) == 0 + assert json.loads(capsys.readouterr().out)["ok"] is True + assert main(["list", *store]) == 0 + assert json.loads(capsys.readouterr().out)[0]["signed_off_by"] == "operator" + + +def test_unavailable_lane_refuses_to_start(tmp_path) -> None: + class Missing(InProcessBackend): + def available(self): + return False + + def unavailable_reason(self): + return "not installed" + + campaign, *_ = _campaign(tmp_path, [_lane("base", honest), + Lane("gone", Missing("gone", honest), _spec())]) + with pytest.raises(CampaignError, match="not installed"): + campaign.run()