78 lines
4.5 KiB
Python
78 lines
4.5 KiB
Python
|
"""Verify frozen-source integrity and bilingual split isolation; no inference."""
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
def require(condition, message):
|
||
|
|
if not condition:
|
||
|
|
raise ValueError(message)
|
||
|
|
|
||
|
|
|
||
|
|
def verify(root):
|
||
|
|
root = root.resolve()
|
||
|
|
inventory = json.loads((root / "inventory.json").read_text(encoding="utf-8"))
|
||
|
|
blind = json.loads((root / "blind-review-input.json").read_text(encoding="utf-8"))
|
||
|
|
require(blind["assets"] == inventory["assets"], "Fixture integrity invariant failed")
|
||
|
|
ids, groups, pairs = set(), {}, {}
|
||
|
|
for asset in inventory["assets"]:
|
||
|
|
require(asset["id"] not in ids, "Fixture integrity invariant failed")
|
||
|
|
ids.add(asset["id"])
|
||
|
|
path = (root / asset["path"]).resolve()
|
||
|
|
require(path.is_relative_to(root), "Fixture integrity invariant failed")
|
||
|
|
data = path.read_bytes()
|
||
|
|
require(len(data) == asset["bytes"], "Fixture integrity invariant failed")
|
||
|
|
require(hashlib.sha256(data).hexdigest() == asset["sha256"], "Fixture integrity invariant failed")
|
||
|
|
group = asset.get("repository", asset.get("topic_family"))
|
||
|
|
require(groups.setdefault(group, asset["split"]) == asset["split"], "Fixture integrity invariant failed")
|
||
|
|
for license_file in asset.get("license_files", []):
|
||
|
|
license_path = (root / license_file["path"]).resolve()
|
||
|
|
require(license_path.is_relative_to(root), "Fixture integrity invariant failed")
|
||
|
|
require(hashlib.sha256(license_path.read_bytes()).hexdigest() == license_file["sha256"], "Fixture integrity invariant failed")
|
||
|
|
query_ids = set()
|
||
|
|
for query in blind["queries"]:
|
||
|
|
require(query["id"] not in query_ids, "Fixture integrity invariant failed")
|
||
|
|
query_ids.add(query["id"])
|
||
|
|
require(query["required_artifact"], "Fixture integrity invariant failed")
|
||
|
|
if query["language"] == "de":
|
||
|
|
require(not any(marker in query["query"] for marker in ("\u00c3", "\u00c2", "\ufffd")), "Fixture integrity invariant failed")
|
||
|
|
pairs.setdefault(query["pair_id"], []).append(query)
|
||
|
|
for pair in pairs.values():
|
||
|
|
require({q["language"] for q in pair} == {"en", "de"}, "Fixture integrity invariant failed")
|
||
|
|
require(len(pair) == 2, "Fixture integrity invariant failed")
|
||
|
|
require(len({q["split"] for q in pair}) == 1, "Fixture integrity invariant failed")
|
||
|
|
labels = json.loads((root / "author-labels.json").read_text(encoding="utf-8"))
|
||
|
|
require(len(labels["labels"]) == len(query_ids), "Fixture integrity invariant failed")
|
||
|
|
require({label["query_id"] for label in labels["labels"]} == query_ids, "Fixture integrity invariant failed")
|
||
|
|
for label in labels["labels"]:
|
||
|
|
require(set(label["expected_ids"]) <= ids, "Fixture integrity invariant failed")
|
||
|
|
fixture = json.loads((root / "fixture.json").read_text(encoding="utf-8"))
|
||
|
|
lock = json.loads((root / "fixture.lock.json").read_text(encoding="utf-8"))
|
||
|
|
require(fixture["assets"] == inventory["assets"], "Frozen inventory differs")
|
||
|
|
combined = [{**query, **next(label for label in labels["labels"] if label["query_id"] == query["id"])} for query in blind["queries"]]
|
||
|
|
require(fixture["queries"] == combined, "Frozen queries differ")
|
||
|
|
require(lock["frozen_at_utc"] == fixture["frozen_at_utc"], "Freeze timestamps differ")
|
||
|
|
for name, expected in lock["sha256"].items():
|
||
|
|
path = (root / name).resolve()
|
||
|
|
require(path.is_relative_to(root), "Lock path escapes fixture")
|
||
|
|
require(hashlib.sha256(path.read_bytes()).hexdigest() == expected, "Frozen hash differs: " + name)
|
||
|
|
review = json.loads((root / "blind-review.json").read_text(encoding="utf-8"))
|
||
|
|
require(review["inputSHA256"] == hashlib.sha256((root / "blind-review-input.json").read_bytes()).hexdigest(), "Blind input hash differs")
|
||
|
|
reviewed = {row["id"]: set(row["expectedIDs"]) for row in review["queries"]}
|
||
|
|
require(len(review["queries"]) == len(query_ids) and set(reviewed) == query_ids, "Blind review coverage differs")
|
||
|
|
for query in fixture["queries"]:
|
||
|
|
require(reviewed[query["id"]] == set(query["expected_ids"]), "Blind disagreement unresolved")
|
||
|
|
for expected_id in query["expected_ids"]:
|
||
|
|
asset = next(asset for asset in fixture["assets"] if asset["id"] == expected_id)
|
||
|
|
require(asset["split"] == query["split"], "Expected artifact leaks between splits")
|
||
|
|
return len(ids), len(pairs)
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
files, pairs = verify(Path(__file__).resolve().parent)
|
||
|
|
print(f"Verified {files} files and {pairs} bilingual pairs; no inference performed.")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|