feat: add bios_integrity.py for bit flips

Rehashes every file database.json names. A confirmed mismatch on a committed file git sees as unmodified is restored from its HEAD blob after the blob is checked, then hashed again. Gitignored files are flagged only.

scripts/bios_integrity.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""Rehash every file database.json names and repair a confirmed bit flip.
+
+git status never shows a flipped bit: the index compares size and mtime, and
+a flip changes neither. Only reading the bytes again finds it. Every file
+the database names is hashed and compared with its recorded SHA-1. A
+mismatch is read a second time: two reads that disagree are a host misread
+and nothing on disk changes.
+
+A confirmed mismatch is handled by where the file comes from:
+
+- committed and unmodified for git: restored from its blob in HEAD, once
+  the blob's object id and its content SHA-1 both check out, then hashed
+  again. Only with --apply.
+- modified for git (git status lists it): a working change some session
+  made, left alone and reported.
+- gitignored (a large-files release asset) or untracked: flagged, never
+  repaired.
+- a blob in HEAD whose content is not the recorded SHA-1: the database and
+  git disagree, flagged.
+
+Run it on an idle host: a busy CPU here computes wrong hashes, which is the
+fault this looks for. autorun.py's load guard holds the weekly job until the
+host is idle.
+
+Usage:
+    python scripts/bios_integrity.py
+    python scripts/bios_integrity.py --apply
+    python scripts/bios_integrity.py --apply --json
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from largefiles import load_registered_paths
+
+MISREAD = "misread"
+RESTORED = "restored"
+FLAGGED = "flagged"
+SKIPPED = "skipped"
+
+
+@dataclass(frozen=True)
+class Finding:
+    path: str
+    verdict: str  # misread, restored, flagged, skipped
+    reason: str
+    expected: str
+    found: str = ""
+
+
+@dataclass
+class Report:
+    checked: int = 0
+    findings: list[Finding] = field(default_factory=list)
+
+    def counts(self) -> dict[str, int]:
+        counts: dict[str, int] = {}
+        for f in self.findings:
+            counts[f.verdict] = counts.get(f.verdict, 0) + 1
+        return counts
+
+    def as_dict(self) -> dict:
+        return {
+            "checked": self.checked,
+            "counts": self.counts(),
+            "findings": [f.__dict__ for f in self.findings],
+        }
+
+    @property
+    def clean(self) -> bool:
+        return all(f.verdict in (MISREAD, RESTORED) for f in self.findings)
+
+
+def sha1_of(path: Path) -> str:
+    digest = hashlib.sha1()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def git(root: Path, *args: str) -> subprocess.CompletedProcess:
+    return subprocess.run(["git", "-C", str(root), *args], capture_output=True, check=False)
+
+
+def head_blob(root: Path, path: str) -> str:
+    """The blob HEAD records for path, empty when HEAD has none."""
+    out = git(root, "ls-tree", "HEAD", "--", path).stdout.decode()
+    parts = out.split()
+    return parts[2] if len(parts) >= 3 and parts[1] == "blob" else ""
+
+
+def modified_for_git(root: Path, path: str) -> bool:
+    return bool(git(root, "status", "--porcelain", "--", path).stdout.strip())
+
+
+def restore_from_blob(root: Path, path: str, blob: str, expected: str) -> str:
+    """Write the blob's bytes over path, after checking them. Empty on success."""
+    data = git(root, "cat-file", "blob", blob).stdout
+    object_id = hashlib.sha1(b"blob %d\0" % len(data) + data).hexdigest()
+    if object_id != blob:
+        return f"blob read back as {object_id[:12]}, not {blob[:12]}"
+    if hashlib.sha1(data).hexdigest() != expected:
+        return "HEAD holds other bytes than database.json records"
+    target = root / path
+    handle, scratch = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
+    try:
+        with os.fdopen(handle, "wb") as out:
+            out.write(data)
+        os.chmod(scratch, target.stat().st_mode & 0o777)
+        os.replace(scratch, target)
+    except OSError:
+        if os.path.exists(scratch):
+            os.unlink(scratch)
+        raise
+    return ""
+
+
+def check(
+    root: Path,
+    db: dict,
+    apply: bool,
+    hasher: Callable[[Path], str] = sha1_of,
+) -> Report:
+    """Hash every database path, confirm each mismatch, repair what may be."""
+    ignored = set(load_registered_paths(root / ".gitignore"))
+    report = Report()
+    for key, entry in sorted(db.get("files", {}).items(), key=lambda kv: kv[1]["path"]):
+        path = entry["path"]
+        expected = str(entry.get("sha1") or key).lower()
+        local = root / path
+        if not local.is_file():
+            continue  # a release asset not fetched yet, verify reports it
+        report.checked += 1
+        first = hasher(local)
+        if first == expected:
+            continue
+        second = hasher(local)
+        if second != first:
+            verdict = MISREAD if second == expected else FLAGGED
+            report.findings.append(Finding(
+                path, verdict, f"two reads disagree ({first[:12]}, {second[:12]})",
+                expected, second))
+            continue
+        if path in ignored:
+            report.findings.append(Finding(
+                path, FLAGGED, "gitignored large file, never repaired here", expected, first))
+            continue
+        blob = head_blob(root, path)
+        if not blob:
+            report.findings.append(Finding(path, FLAGGED, "not in HEAD", expected, first))
+            continue
+        if modified_for_git(root, path):
+            report.findings.append(Finding(
+                path, SKIPPED, "modified for git: a working change, left alone",
+                expected, first))
+            continue
+        if not apply:
+            report.findings.append(Finding(
+                path, FLAGGED, "bit flip, restorable from HEAD with --apply", expected, first))
+            continue
+        problem = restore_from_blob(root, path, blob, expected)
+        if problem:
+            report.findings.append(Finding(path, FLAGGED, problem, expected, first))
+            continue
+        after = hasher(local)
+        if after == expected:
+            report.findings.append(Finding(
+                path, RESTORED, f"restored from blob {blob[:12]}", expected, first))
+        else:
+            report.findings.append(Finding(
+                path, FLAGGED, f"restored but reads {after[:12]}", expected, after))
+    return report
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.split("\n\n")[0])
+    parser.add_argument("--apply", action="store_true",
+                        help="restore confirmed flips of committed files from HEAD")
+    parser.add_argument("--db", default="database.json",
+                        help="database file to read (default: %(default)s)")
+    parser.add_argument("--json", action="store_true", help="machine-readable report")
+    args = parser.parse_args()
+
+    from common import load_database
+
+    root = Path.cwd()
+    report = check(root, load_database(args.db), args.apply)
+    if args.json:
+        print(json.dumps(report.as_dict(), indent=2))
+    else:
+        for f in report.findings:
+            print(f"{f.verdict.upper():9} {f.path}: {f.reason}")
+        counts = ", ".join(f"{n} {v}" for v, n in sorted(report.counts().items()))
+        print(f"{report.checked} files hashed" + (f", {counts}" if counts else ", all match"))
+    return 0 if report.clean else 1
+
+
+if __name__ == "__main__":
+    from common import run_cli
+
+    run_cli(main)