230 lines
16 KiB
Python
230 lines
16 KiB
Python
"""Verify archived native receipts and bytes; do not rerun the GUI or checker."""
|
|
from pathlib import Path, PurePosixPath
|
|
from datetime import datetime
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import io
|
|
import re
|
|
import stat
|
|
import zipfile
|
|
|
|
BASE = {"prepared.json", "started.json", "finished.json", "generation.json", "generation.stdout.log",
|
|
"generation.stderr.log", "app.stdout.log", "app.stderr.log", "capture-executed.js"}
|
|
OUTER = {"prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip"}
|
|
|
|
|
|
def need(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
|
|
def pin(data):
|
|
return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def safe(name):
|
|
return (isinstance(name, str) and bool(name) and "\\" not in name and ":" not in name
|
|
and not PurePosixPath(name).is_absolute() and all(ord(c) >= 32 for c in name)
|
|
and all(part not in ("", ".", "..") for part in name.split("/")))
|
|
|
|
|
|
def parsed(data):
|
|
def unique(items):
|
|
result = {}
|
|
for key, value in items:
|
|
need(key not in result, "duplicate JSON key: " + key)
|
|
result[key] = value
|
|
return result
|
|
return json.loads(data, object_pairs_hook=unique)
|
|
|
|
|
|
def norm(path):
|
|
return path.replace("\\", "/").rstrip("/")
|
|
|
|
|
|
def relative(path, root):
|
|
prefix = norm(root) + "/"
|
|
need(norm(path).startswith(prefix), "recorded path outside owner: " + path)
|
|
value = norm(path)[len(prefix):]
|
|
need(safe(value), "unsafe recorded relative path")
|
|
return value
|
|
|
|
|
|
def ordered(*values):
|
|
times = [datetime.fromisoformat(value) for value in values]
|
|
return all(t.utcoffset() is not None for t in times) and times == sorted(times)
|
|
|
|
|
|
def zip_inventory(archive, members):
|
|
names = archive.namelist()
|
|
need(len(names) == len(set(names)) and set(names) == set(members), "exact archive members")
|
|
for info in archive.infolist():
|
|
need(safe(info.filename) and not info.is_dir() and not stat.S_ISLNK(info.external_attr >> 16), "unsafe archive member")
|
|
need(info.file_size == members[info.filename]["bytes"], "archive member size")
|
|
|
|
|
|
def verify_data(manifest, read):
|
|
need(manifest["schema"] == 1 and manifest["kind"] == "process-multiselect-native", "manifest kind")
|
|
members, runs = manifest["members"], manifest["runs"]
|
|
need(bool(runs) and all(re.fullmatch(r"run-[0-9]+", run) for run in runs), "run names")
|
|
need(all(safe(name) for name in members), "safe member names")
|
|
for name, expected in members.items():
|
|
need(pin(read(name)) == expected, "member pin: " + name)
|
|
used, summaries = set(), {}
|
|
def raw(name):
|
|
need(name in members, "missing member: " + name)
|
|
used.add(name)
|
|
return read(name)
|
|
def obj(name):
|
|
return parsed(raw(name))
|
|
def tree(prefix):
|
|
return {name[len(prefix):]: raw(name) for name in members if name.startswith(prefix)}
|
|
def pinned_tree(prefix, expected):
|
|
actual = tree(prefix)
|
|
need({name: pin(data) for name, data in actual.items()} == expected, "pinned tree: " + prefix)
|
|
return actual
|
|
for run, spec in runs.items():
|
|
prefix = "runs/" + run + "/raw/"
|
|
prepared, start, end, generation = [obj(prefix + n + ".json") for n in ("prepared", "started", "finished", "generation")]
|
|
need(spec["reviewStatus"] in ("failed", "unreviewed") and isinstance(spec["reviewNote"], str) and spec["reviewNote"], "explicit non-acceptance status")
|
|
if run == "run-1":
|
|
need(spec["reviewStatus"] == "failed", "run-1 must retain observed native failure")
|
|
origin = norm(prepared["cwd"]) + "/.local/process-multiselect-native/" + run
|
|
need(spec["originalRun"] == origin and prepared["run"] == run and start["cwd"] == prepared["cwd"], "original run mapping")
|
|
need(start["prepared"] == pin(raw(prefix + "prepared.json")), "launch preparation pin")
|
|
need(ordered(generation["start"], generation["finish"], prepared["prepared"], start["start"], end["finish"]), "native lifetime")
|
|
need(type(start["pid"]) is int and start["pid"] > 0 and type(end["exitCode"]) is int, "closed native process")
|
|
for key in ("guiUnchanged", "checkerUnchanged", "fixtureUnchanged"):
|
|
need(type(end[key]) is bool, "closure flag")
|
|
need(start["gui"] == prepared["gui"] and start["checker"] == prepared["checker"] == generation["checker"], "binary identity")
|
|
need(start["command"] == [prepared["gui"]["binary"]], "GUI command")
|
|
need(generation["exitCode"] == 0 and generation["command"][:2] == [prepared["checker"]["binary"], "generate"] and len(generation["command"]) == 3 and norm(generation["command"][2]) == origin + "/fixture", "generation command")
|
|
for stream in ("stdout", "stderr"):
|
|
need(end[stream] == pin(raw(prefix + "app." + stream + ".log")), "GUI log pin")
|
|
need(generation[stream] == pin(raw(prefix + "generation." + stream + ".log")), "generation log pin")
|
|
need(set(spec["helpers"]) == set(prepared["helpers"]), "exact prepared helper mapping")
|
|
for name, expected in prepared["helpers"].items():
|
|
need(safe(name) and "/" not in name and pin(raw(spec["helpers"][name])) == expected, "historical helper: " + name)
|
|
need(start["helper"] == prepared["helpers"]["native.py"], "launch helper")
|
|
need(pin(raw(prefix + "capture-executed.js")) == spec["executedCapture"], "actual capture function pin")
|
|
fixture = pinned_tree(prefix + "fixture/", prepared["fixture"])
|
|
fm = parsed(fixture["manifest.json"])
|
|
need(fm["fixture"] == "process-multiselect-native-20260923" and fm["source_entry"] == prepared["entry"] and safe(prepared["entry"]), "fixture identity/entry")
|
|
need(parsed(fixture["baseline.json"])["id"] == fm["ids"]["project"], "complete Project baseline identity")
|
|
fixture_source = {name[7:]: data for name, data in fixture.items() if name.startswith("source/")}
|
|
need(prepared["entry"] in fixture_source and norm(prepared["workingEntry"]) == origin + "/working/" + prepared["entry"], "working entry mapping")
|
|
working = tree(prefix + "working/")
|
|
need(bool(working) and all(name.endswith(".arcadia") for name in working), "saved working tree")
|
|
build_maps = {}
|
|
for kind, target in (("gui", "arcadia-studio"), ("checker", "process_multiselect_native")):
|
|
recorded, b = prepared[kind], spec["builds"][kind]
|
|
result, began, process = [obj(b[n]) for n in ("result.json", "started.json", "process.json")]
|
|
need(pin(raw(b["result.json"])) == recorded["receiptPin"] and relative(recorded["receipt"], prepared["cwd"]) == b["originalRelative"], "build receipt mapping")
|
|
need(all(result.get(k) == v for k, v in began.items()) and process["command"] == result["command"] and type(process["pid"]) is int and process["pid"] > 0, "build start/result/process")
|
|
need(result["exitCode"] == 0 and result["inputsUnchanged"] is True and ordered(result["start"], process["observed"], result["finish"], generation["start"]), "successful preceding build")
|
|
for stream in ("stdout", "stderr"):
|
|
need(pin(raw(b[stream + ".log"])) == result[stream], "build log pin")
|
|
need(pin(raw(b["helper"])) == result["helper"], "build helper pin")
|
|
need(pin(raw(b["inputs.json"])) == recorded["inputs"] == result["inputs"], "build source map pin")
|
|
inputs = obj(b["inputs.json"])
|
|
need(len(inputs) == result["inputCount"] and all(safe(name) for name in inputs), "source map count/paths")
|
|
capsule = raw(b["inputs.zip"])
|
|
need(pin(capsule) == result["archive"], "build source capsule pin")
|
|
with zipfile.ZipFile(io.BytesIO(capsule)) as archive:
|
|
zip_inventory(archive, inputs)
|
|
need(all(pin(archive.read(name)) == expected for name, expected in inputs.items()), "all captured source bytes")
|
|
if kind == "checker":
|
|
path = "crates/oa-core/examples/process_multiselect_native.rs"
|
|
need(raw(spec["checkerSource"]) == archive.read(path), "historical checker source")
|
|
artifacts = [row for row in result["artifacts"] if row["cargo"]["target"]["name"] == target]
|
|
need(len(artifacts) == 1 and artifacts[0]["readback"] == recorded["pin"] and artifacts[0]["cargo"]["executable"] == recorded["binary"] and artifacts[0]["cargo"]["profile"]["test"] is False, "native executable provenance")
|
|
stdout_rows = [parsed(line) for line in raw(b["stdout.log"]).splitlines() if line.strip().startswith(b"{")]
|
|
need(artifacts[0]["cargo"] in stdout_rows, "artifact appears in raw Cargo output")
|
|
build_maps[kind] = inputs
|
|
need(prepared["gui"]["inputs"] == prepared["checker"]["inputs"] and build_maps["gui"] == build_maps["checker"], "same GUI/checker source map")
|
|
top = {name[len(prefix):] for name in members if name.startswith(prefix) and "/" not in name[len(prefix):]}
|
|
base = BASE | (top & {"history-bytes.json", "observations.md"})
|
|
if "observations.md" in top:
|
|
raw(prefix + "observations.md")
|
|
captures = sorted(name for name in top - base if name.endswith(".json"))
|
|
need(captures, "native captures required")
|
|
expected_top = base | set(captures)
|
|
for name in captures:
|
|
capture = obj(prefix + name)
|
|
need(ordered(start["start"], capture["time"], end["finish"]), "capture lifetime")
|
|
window = capture["window"]
|
|
need(norm(window["app"]) == "process:" + norm(prepared["gui"]["binary"]) and type(window["id"]) is int and window["id"] > 0, "captured executable/window")
|
|
need(isinstance(capture["accessibility"], dict) and isinstance(capture["images"], list), "accessibility/capture payload")
|
|
for image in capture["images"]:
|
|
filename = image["file"]
|
|
need(safe(filename) and "/" not in filename and filename.startswith(name[:-5] + "-") and filename.endswith((".jpg", ".png")) and filename not in expected_top, "unique image mapping")
|
|
need(all(type(image[k]) is int and image[k] > 0 for k in ("width", "height")), "image dimensions")
|
|
expected_top.add(filename)
|
|
data = raw(prefix + filename)
|
|
need(data.startswith(b"\xff\xd8") if filename.endswith(".jpg") else data.startswith(b"\x89PNG\r\n\x1a\n"), "image signature")
|
|
need(top == expected_top, "exact top-level run files")
|
|
labels = {name[len(prefix + "snapshots/"):].split("/")[0] for name in members if name.startswith(prefix + "snapshots/")}
|
|
states, results, positive = {}, {}, []
|
|
for label in sorted(labels):
|
|
need(re.fullmatch(r"[a-z0-9-]+", label), "snapshot label")
|
|
folder = prefix + "snapshots/" + label + "/"
|
|
began, result = obj(folder + "started.json"), obj(folder + "result.json")
|
|
need(all(result.get(k) == v for k, v in began.items()), "snapshot start/result")
|
|
need(result["prepared"] == start["prepared"] and result["checker"] == prepared["checker"] and result["helper"] == prepared["helpers"]["snapshot.py"], "snapshot provenance")
|
|
need(ordered(start["start"], result["start"], result["finish"], end["finish"]), "snapshot lifetime")
|
|
mode, command = result["mode"], result["command"]
|
|
need(mode in ("unchanged", "reordered") and len(command) == 5 and command[:3] == [prepared["checker"]["binary"], "check", mode] and norm(command[3]) == origin + "/fixture/baseline.json" and norm(command[4]) == origin + "/snapshots/" + label + "/source/" + prepared["entry"], "snapshot command")
|
|
need(result["baseline"] == pin(fixture["baseline.json"]), "snapshot baseline")
|
|
for stream in ("stdout", "stderr"):
|
|
need(result[stream] == pin(raw(folder + stream + ".log")), "snapshot log pin")
|
|
need(type(result["exitCode"]) is int and all(type(result[k]) is bool for k in ("binaryUnchanged", "workingSourceUnchanged", "snapshotUnchanged", "baselineUnchanged")), "snapshot closure")
|
|
states[label] = pinned_tree(folder + "source/", result["source"])
|
|
need({name for name in members if name.startswith(folder)} == {folder + n for n in ("started.json", "result.json", "stdout.log", "stderr.log")} | {folder + "source/" + n for n in states[label]}, "exact snapshot files")
|
|
if result["exitCode"] == 0:
|
|
report = obj(folder + "stdout.log")
|
|
need(report["result"] == "PASS" and report["comparison"] == "complete Project equality; literal expected order" and report["mode"] == mode and report["baseline"] == command[3] and report["saved"] == command[4], "positive complete Project report")
|
|
positive.append(label)
|
|
results[label] = result
|
|
history_present = "history-bytes.json" in top
|
|
if history_present:
|
|
history = obj(prefix + "history-bytes.json")
|
|
groups = history["equalByteGroups"]
|
|
need(len(groups) == 2 and groups[0][:2] == ["baseline", "undo"] and groups[1] == ["reordered", "redo", "reopened"], "literal Process history groups")
|
|
history_labels = [label for group in groups for label in group]
|
|
need(len(history_labels) == len(set(history_labels)) and set(history_labels) == set(history["snapshots"]) and set(history_labels) <= set(positive), "history labels/positive reports")
|
|
need(history["status"] == "PASS" and history["prepared"] == start["prepared"] and history["helper"] == prepared["helpers"]["history-bytes.py"] and ordered(end["finish"], history["verified"]), "history provenance")
|
|
need(end["exitCode"] == 0 and all(end[k] is True for k in ("guiUnchanged", "checkerUnchanged", "fixtureUnchanged")), "history clean closure")
|
|
for group, mode in zip(groups, ("unchanged", "reordered")):
|
|
for label in group:
|
|
need(states[label] == states[group[0]] and results[label]["mode"] == mode and history["snapshots"][label] == pin(raw(prefix + "snapshots/" + label + "/result.json")), "literal history bytes/receipt")
|
|
need(all(results[label][k] is True for k in ("binaryUnchanged", "workingSourceUnchanged", "snapshotUnchanged", "baselineUnchanged")), "stable history snapshot")
|
|
need(len(states[label]) == history["sourceFilesPerSnapshot"], "history source count")
|
|
need(states["baseline"] == fixture_source and states["baseline"] != states["reordered"], "baseline exact and reorder changed")
|
|
changed = sorted(name for name in set(states["baseline"]) | set(states["reordered"]) if states["baseline"].get(name) != states["reordered"].get(name))
|
|
need(changed == history["changedFilesFromBaseline"] == ["oa/behavior/process_oa_process.arcadia"], "only intended OA source changed")
|
|
sequence = [results[label] for label in ("baseline", "reordered", "undo", "redo", "reopened")]
|
|
need(all(ordered(a["finish"], b["start"]) for a, b in zip(sequence, sequence[1:])), "history chronology")
|
|
summaries[run] = {"reviewStatus": spec["reviewStatus"], "exitCode": end["exitCode"], "captures": len(captures), "snapshots": len(labels), "positiveReports": len(positive), "historyBytes": history_present}
|
|
need(used == set(members), "unreferenced package members: " + repr(set(members) - used))
|
|
return summaries
|
|
|
|
|
|
def verify_package(root):
|
|
files = parsed((root / "files.json").read_bytes())
|
|
need(set(files) == OUTER, "outer files")
|
|
for name, expected in files.items():
|
|
need(pin((root / name).read_bytes()) == expected, "outer pin: " + name)
|
|
manifest = parsed((root / "manifest.json").read_bytes())
|
|
with zipfile.ZipFile(root / "evidence.zip") as archive:
|
|
zip_inventory(archive, manifest["members"])
|
|
return verify_data(manifest, archive.read)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent)
|
|
args = parser.parse_args()
|
|
print("PASS byte/receipt consistency only:", verify_package(args.package))
|
|
print("No GUI/checker execution or native acceptance claim; see independent parent review.")
|