06_verify.py

1"""Phase 6: verify every archive and write CHECKSUMS.sha256 and checkpoints/verification.json.""" 2import glob 3import hashlib 4import json 5import os 6import re 7import subprocess 8import sys 9import tempfile 10import zipfile 11 12sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) 13from nlr_common import ROOT, log, now, sha256_file, write_json 14 15HTMLISH = re.compile(rb"^\s*(<!doctype html|<html)", re.I) 16 17 18def unzip_test(p): 19 r = subprocess.run(["unzip", "-tqq", p], capture_output=True, text=True) 20 return r.returncode == 0, (r.stdout + r.stderr).strip()[-300:] 21 22 23def safe_entry(n): 24 return not (n.startswith("/") or ".." in n.split("/") or "\x00" in n) 25 26 27def verify_deploy_zip(zp, man): 28 want = {e["local_path"]: e for e in man["files"] if not e.get("error")} 29 res = {"expected_entries": len(want), "matched_sha1": 0, "mismatched": [], "missing_in_zip": [], "extra_in_zip": [], 30 "unsafe_names": [], "html_where_asset_expected": []} 31 with zipfile.ZipFile(zp) as z: 32 names = [i.filename for i in z.infolist() if not i.is_dir()] 33 res["unsafe_names"] = [n for n in names if not safe_entry(n)] 34 res["extra_in_zip"] = sorted(set(names) - set(want)) 35 res["missing_in_zip"] = sorted(set(want) - set(names)) 36 for n in names: 37 if n not in want: 38 continue 39 h = hashlib.sha1() 40 first = b"" 41 size = 0 42 with z.open(n) as f: 43 for chunk in iter(lambda: f.read(1 << 20), b""): 44 if not first: 45 first = chunk[:200] 46 h.update(chunk) 47 size += len(chunk) 48 e = want[n] 49 if h.hexdigest() == e["sha1_manifest"] and (e.get("size_manifest") in (None, size)): 50 res["matched_sha1"] += 1 51 else: 52 res["mismatched"].append(n) 53 mt = (e.get("mime_type") or "").lower() 54 if HTMLISH.match(first) and "html" not in mt and not n.lower().endswith((".html", ".htm")): 55 res["html_where_asset_expected"].append(n) 56 res["passed"] = (res["matched_sha1"] == len(want) and not res["mismatched"] and not res["missing_in_zip"] 57 and not res["extra_in_zip"] and not res["unsafe_names"]) 58 return res 59 60 61def verify_source_zip(zp, mirror, commit): 62 r = subprocess.run(["git", "ls-tree", "-r", "-z", commit], cwd=mirror, capture_output=True) 63 blobs = [x for x in r.stdout.split(b"\0") if x and b" blob " in x.split(b"\t")[0]] 64 with zipfile.ZipFile(zp) as z: 65 files = [i for i in z.infolist() if not i.is_dir()] 66 unsafe = [i.filename for i in files if not safe_entry(i.filename)] 67 return {"git_tree_blobs": len(blobs), "zip_file_entries": len(files), "unsafe_names": unsafe, 68 "passed": len(blobs) == len(files) and not unsafe} 69 70 71def verify_history_zip(zp): 72 with tempfile.TemporaryDirectory(dir=os.path.join(ROOT, "tmp")) as td: 73 with zipfile.ZipFile(zp) as z: 74 z.extract("repository_all_refs.bundle", td) 75 b = os.path.join(td, "repository_all_refs.bundle") 76 r = subprocess.run(["git", "bundle", "list-heads", b], capture_output=True, text=True) 77 heads = r.stdout.strip().splitlines() 78 r2 = subprocess.run(["git", "init", "-q", os.path.join(td, "v")], capture_output=True) 79 r3 = subprocess.run(["git", "-C", os.path.join(td, "v"), "bundle", "verify", b], capture_output=True, text=True) 80 return {"bundle_heads": len(heads), "bundle_verify_ok": r3.returncode == 0, "passed": r.returncode == 0 and r3.returncode == 0 and bool(heads)} 81 82 83def main(): 84 desktop = os.path.realpath(os.path.expanduser("~/Desktop")) 85 out = {"verified_at": now(), "root": ROOT, "root_is_on_users_desktop": os.path.realpath(ROOT).startswith(desktop + os.sep), 86 "archives": {}} 87 manifests = {} 88 for mp in glob.glob(os.path.join(ROOT, "sites", "*", "manifests", "deploy_*_recovery.json")): 89 m = json.load(open(mp)) 90 if m.get("zip"): 91 manifests[os.path.basename(m["zip"]["path"])] = m 92 lines = [] 93 for zp in sorted(glob.glob(os.path.join(ROOT, "zips", "*.zip"))): 94 bn = os.path.basename(zp) 95 rec = {"bytes": os.path.getsize(zp), "sha256": sha256_file(zp)} 96 rec["unzip_test_ok"], rec["unzip_test_output"] = unzip_test(zp) 97 try: 98 if "__DEPLOY__" in bn: 99 m = manifests.get(bn) 100 rec["kind"] = "DEPLOY" 101 rec["content"] = verify_deploy_zip(zp, m) if m else {"passed": False, "error": "no recovery manifest"} 102 if m: 103 rec["deploy_status"] = m["status"] 104 rec["manifest_sha256_matches"] = (m["zip"]["sha256"] == rec["sha256"]) 105 elif "__SOURCE__" in bn: 106 rec["kind"] = "SOURCE" 107 commit = bn.rsplit("__SOURCE__", 1)[1][:-4] 108 site_dir = glob.glob(os.path.join(ROOT, "sites", "*__" + bn.split("__")[1]))[0] 109 mirror = glob.glob(os.path.join(site_dir, "source", "*.git"))[0] 110 rec["content"] = verify_source_zip(zp, mirror, commit) 111 elif "__REPOSITORY_HISTORY" in bn: 112 rec["kind"] = "REPOSITORY_HISTORY" 113 rec["content"] = verify_history_zip(zp) 114 elif "__PUBLIC_MIRROR_PARTIAL__" in bn: 115 rec["kind"] = "PUBLIC_MIRROR_PARTIAL" 116 with zipfile.ZipFile(zp) as z: 117 names = z.namelist() 118 rec["content"] = {"entries": len(names), "passed": all(safe_entry(n) for n in names)} 119 else: 120 rec["kind"] = "OTHER" 121 rec["content"] = {"passed": rec["unzip_test_ok"]} 122 except Exception as e: 123 rec["content"] = {"passed": False, "error": f"{type(e).__name__}: {e}"} 124 rec["passed"] = bool(rec["unzip_test_ok"] and rec["content"].get("passed") and rec["bytes"] > 22) 125 out["archives"][bn] = rec 126 lines.append(f"{rec['sha256']} zips/{bn}") 127 log(f"verify {bn}: {'PASS' if rec['passed'] else 'FAIL'} ({rec['bytes']} bytes)", echo=False) 128 # Recovered individual files (from manifests; recomputed only for the content store spot check below). 129 for mp in sorted(glob.glob(os.path.join(ROOT, "sites", "*", "manifests", "deploy_*_recovery.json"))): 130 m = json.load(open(mp)) 131 site_rel = os.path.relpath(os.path.dirname(os.path.dirname(mp)), ROOT) 132 for e in m["files"]: 133 if e.get("sha256"): 134 lines.append(f"{e['sha256']} {site_rel}/originals/deploy_{m['deploy']['id']}/{e['local_path']}") 135 with open(os.path.join(ROOT, "CHECKSUMS.sha256.tmp"), "w") as f: 136 f.write("\n".join(lines) + "\n") 137 os.replace(os.path.join(ROOT, "CHECKSUMS.sha256.tmp"), os.path.join(ROOT, "CHECKSUMS.sha256")) 138 # Re-hash every recovered original on disk against the checksum list. 139 bad = 0 140 for ln in lines: 141 h, rel = ln.split(" ", 1) 142 if not rel.startswith("zips/") and sha256_file(os.path.join(ROOT, rel)) != h: 143 bad += 1 144 out["original_files_checked"] = sum(1 for ln in lines if not ln.split(" ", 1)[1].startswith("zips/")) 145 out["original_files_hash_mismatch"] = bad 146 out["archives_total"] = len(out["archives"]) 147 out["archives_passed"] = sum(1 for r in out["archives"].values() if r["passed"]) 148 write_json(os.path.join(ROOT, "checkpoints", "verification.json"), out) 149 log(f"verification: {out['archives_passed']}/{out['archives_total']} archives passed; " 150 f"{out['original_files_checked']} original files rechecked, {bad} mismatches; on Desktop={out['root_is_on_users_desktop']}") 151 152 153if __name__ == "__main__": 154 main() 155