174 lines
9.3 KiB
Python
174 lines
9.3 KiB
Python
"""Offline consistency checks only; never execute or extract archived inputs."""
|
|
from pathlib import Path, PurePosixPath
|
|
from datetime import datetime
|
|
import argparse
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import math
|
|
import re
|
|
import stat
|
|
import zipfile
|
|
|
|
RAW_FILES = {"started.json", "process.json", "result.json", "stdout.log", "stderr.log", "inputs.json", "inputs.zip"}
|
|
OUTER_FILES = {"prepare.py", "verify.py", "README.md", "manifest.json", "evidence.zip"}
|
|
|
|
|
|
def need(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def pin(data):
|
|
return {"bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()}
|
|
|
|
|
|
def valid_pin(value):
|
|
return (isinstance(value, dict) and set(value) == {"bytes", "sha256"}
|
|
and type(value["bytes"]) is int and value["bytes"] >= 0
|
|
and isinstance(value["sha256"], str)
|
|
and re.fullmatch(r"[0-9a-f]{64}", value["sha256"]) is not None)
|
|
|
|
|
|
def safe(name):
|
|
return (isinstance(name, str) and bool(name) and "\\" not in name and ":" not in name
|
|
and all(ord(char) >= 32 for char in name)
|
|
and not PurePosixPath(name).is_absolute()
|
|
and all(part not in ("", ".", "..") for part in name.split("/")))
|
|
|
|
|
|
def run_name(name):
|
|
return isinstance(name, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*", name) is not None
|
|
|
|
|
|
def parsed(data):
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
need(key not in result, "duplicate JSON key: " + key)
|
|
result[key] = value
|
|
return result
|
|
def constant(value):
|
|
raise ValueError("nonfinite JSON constant: " + value)
|
|
return json.loads(data, object_pairs_hook=pairs, parse_constant=constant)
|
|
|
|
|
|
def inventory(archive, expected):
|
|
names = archive.namelist()
|
|
need(len(names) == len(set(names)) and set(names) == set(expected), "exact ZIP membership")
|
|
for entry in archive.infolist():
|
|
need(safe(entry.filename) and not entry.is_dir(), "unsafe ZIP member: " + entry.filename)
|
|
need(not stat.S_ISLNK(entry.external_attr >> 16), "ZIP symlink: " + entry.filename)
|
|
need(valid_pin(expected[entry.filename]), "invalid member pin: " + entry.filename)
|
|
need(entry.file_size == expected[entry.filename]["bytes"], "ZIP recorded size: " + entry.filename)
|
|
|
|
|
|
def ordered_times(*values):
|
|
times = [datetime.fromisoformat(value) for value in values]
|
|
return all(value.utcoffset() is not None for value in times) and times == sorted(times)
|
|
|
|
|
|
def verify_data(manifest, read):
|
|
"""Shared read-only verifier for packaged bytes or prepare's in-memory payload."""
|
|
need(manifest["schema"] == 1 and manifest["kind"] == "multiselect-cargo-gates", "manifest schema")
|
|
runs, logical, stored = manifest["runs"], manifest["logicalFiles"], manifest["members"]
|
|
helpers = manifest["captureHelpers"]
|
|
need(isinstance(helpers, dict) and set(helpers) == {"gate.py", "gate-feedback.py"}, "capture helper names")
|
|
need(all(isinstance(name, str) and re.fullmatch(r"helpers/[0-9a-f]{64}\.py", name)
|
|
for name in helpers.values()), "capture helper mapping")
|
|
need(isinstance(runs, dict) and bool(runs) and all(run_name(run) for run in runs), "explicit run inventory")
|
|
need(all(safe(name) for name in logical) and all(safe(name) for name in stored), "unsafe manifest path")
|
|
need(set(logical.values()) == set(stored), "unreferenced or absent stored members")
|
|
for name, expected in stored.items():
|
|
need(valid_pin(expected) and pin(read(name)) == expected, "stored member hash: " + name)
|
|
expected_logical = {"cargo/" + run + "/" + name for run in runs for name in RAW_FILES}
|
|
expected_logical.update(spec["helper"] for spec in runs.values())
|
|
expected_logical.update(helpers.values())
|
|
need(set(logical) == expected_logical, "exact logical raw/helper inventory")
|
|
for name, destination in logical.items():
|
|
if name.startswith("cargo/") and name.rsplit("/", 1)[1] in {"inputs.json", "inputs.zip"}:
|
|
expected = "capsules/" + stored[destination]["sha256"] + PurePosixPath(name).suffix
|
|
need(destination == expected, "input capsule mapping: " + name)
|
|
else:
|
|
need(destination == name, "only input maps/archives may be deduplicated: " + name)
|
|
def raw(name):
|
|
return read(logical[name])
|
|
for name, helper in helpers.items():
|
|
need(helper == "helpers/" + pin(raw(helper))["sha256"] + ".py", "named helper hash: " + name)
|
|
seen_capsules = set()
|
|
for run, spec in runs.items():
|
|
prefix = "cargo/" + run + "/"
|
|
start = parsed(raw(prefix + "started.json"))
|
|
process = parsed(raw(prefix + "process.json"))
|
|
result = parsed(raw(prefix + "result.json"))
|
|
required_start = {"start", "command", "cwd", "head", "scope", "inputs", "archive",
|
|
"inputCount", "environment", "helper", "cargo", "rustc"}
|
|
need(required_start <= start.keys(), run + ": incomplete start receipt")
|
|
need(all(result.get(key) == value for key, value in start.items()), run + ": start/result relationship")
|
|
command = result["command"]
|
|
need(isinstance(command, list) and command and command[0] == "cargo"
|
|
and all(isinstance(arg, str) for arg in command), run + ": Cargo command")
|
|
need(command == process["command"] == spec["command"], run + ": command relationship")
|
|
need(result["head"] == spec["head"] and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", result["head"]), run + ": HEAD")
|
|
need(type(result["exitCode"]) is int and result["exitCode"] == spec["exitCode"], run + ": recorded exit")
|
|
need(result["inputsUnchanged"] is True, run + ": changed inputs refused")
|
|
need(type(result["seconds"]) in (int, float) and math.isfinite(result["seconds"])
|
|
and result["seconds"] >= 0, run + ": duration")
|
|
need(type(process["pid"]) is int and process["pid"] > 0, run + ": process identity")
|
|
need(ordered_times(start["start"], process["observed"], result["finish"]), run + ": completed timestamp order")
|
|
for field, filename in [("inputs", "inputs.json"), ("archive", "inputs.zip"), ("stdout", "stdout.log"), ("stderr", "stderr.log")]:
|
|
need(result[field] == pin(raw(prefix + filename)), run + ": " + field + " pin")
|
|
need(valid_pin(result["helper"]) and result["helper"] == pin(raw(spec["helper"])), run + ": helper pin")
|
|
need(spec["helper"] == "helpers/" + result["helper"]["sha256"] + ".py", run + ": helper mapping")
|
|
inputs = parsed(raw(prefix + "inputs.json"))
|
|
need(isinstance(inputs, dict) and all(safe(name) for name in inputs), run + ": safe input inventory")
|
|
need(type(result["inputCount"]) is int and len(inputs) == result["inputCount"] == spec["inputCount"], run + ": input counts")
|
|
need(sum(name.startswith("crates/vendor/iced_widget/") for name in inputs) == 71, run + ": vendor count")
|
|
capsule = (logical[prefix + "inputs.json"], logical[prefix + "inputs.zip"])
|
|
if capsule not in seen_capsules:
|
|
with zipfile.ZipFile(io.BytesIO(raw(prefix + "inputs.zip"))) as source:
|
|
inventory(source, inputs)
|
|
for name, expected in inputs.items():
|
|
need(pin(source.read(name)) == expected, run + ": source hash " + name)
|
|
seen_capsules.add(capsule)
|
|
artifacts = []
|
|
for line in raw(prefix + "stdout.log").decode("utf-8", errors="replace").splitlines():
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict) and value.get("reason") == "compiler-artifact" and value.get("executable"):
|
|
artifacts.append(value)
|
|
need(isinstance(result["artifacts"], list), run + ": artifact records")
|
|
need(artifacts == [record["cargo"] for record in result["artifacts"]], run + ": Cargo JSON artifact relationship")
|
|
for record in result["artifacts"]:
|
|
need(set(record) == {"cargo", "readback"}
|
|
and (record["readback"] is None or valid_pin(record["readback"])), run + ": artifact readback pin")
|
|
print(run, "recorded exit", result["exitCode"], "inputs", len(inputs), "executable records", len(artifacts))
|
|
return {"runs": len(runs), "recordedFailures": sum(spec["exitCode"] != 0 for spec in runs.values()),
|
|
"logicalFiles": len(logical), "storedMembers": len(stored), "sourceCapsulePairs": len(seen_capsules),
|
|
"captureHelpers": helpers}
|
|
|
|
|
|
def verify_package(root):
|
|
files = parsed((root / "files.json").read_bytes())
|
|
need(set(files) == OUTER_FILES, "outer file inventory")
|
|
for name, expected in files.items():
|
|
need(valid_pin(expected) and pin((root / name).read_bytes()) == expected, "outer hash: " + name)
|
|
manifest = parsed((root / "manifest.json").read_bytes())
|
|
with zipfile.ZipFile(root / "evidence.zip") as archive:
|
|
inventory(archive, manifest["members"])
|
|
return verify_data(manifest, archive.read)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("package", nargs="?", type=Path, default=Path(__file__).resolve().parent)
|
|
args = parser.parse_args()
|
|
print("PASS offline consistency:", verify_package(args.package))
|
|
print("Historical receipts only; no source/binary execution or native acceptance claim.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|