Static check for in-place edits of hash-pinned bytes (flags #6318's 42 stale pins; no κ)

docs/research/pr-intake.md

@@ -19,6 +19,12 @@ or changes the cache. Use the canonical integration checkout; preserve unrelated
 user work. `--live-json` supports offline diagnostics but is explicitly not fresh
 GitHub evidence. Recheck the live head before publication or closure.
 
+Before reviewing a PR that modifies existing files, run
+`python3 -B scripts/check_pinned_edits.py --base main --head <PR head>`. It lists
+every file whose old bytes another file pins by SHA-256 and exits 1 when a frozen
+`research/` or `upstream/` file is changed in place while still pinned. It is a
+static byte check of one diff: it executes no contribution code and replays nothing.
+
 If the fresh plan has zero actionable entries, stop routine intake and report
 the remaining revision/deferred counts. Do not fill an empty batch with dormant
 reviews or poll indefinitely unless the user explicitly requested monitoring.

scripts/check_pinned_edits.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+"""Report pinned bytes that a branch changes, before review or a cold replay.
+
+For every file a branch modifies or deletes relative to its merge base, this
+computes the SHA-256 of the merge-base bytes and searches the merge-base tree
+for that full 64-hex digest. A file containing the digest pins the old bytes.
+Each pin is classified at the branch head:
+
+- ``stale``: the pinning file still contains the old digest, so its check will
+  fail (or silently describe bytes that no longer exist);
+- ``repinned``: the pinning file was rewritten so the old digest is gone. This
+  passes byte checks but may rewrite frozen evidence; review it as such;
+- ``pinner_removed``: the pinning file was deleted on the branch.
+
+Living records (the selected record, the task catalogue, docs and shared
+scripts) are routinely updated while older packages keep their digests as
+context, so their stale pins are advisory. Files under ``research/`` and
+``upstream/`` are frozen by repository policy: exit status is 1 when one of
+them has a stale pin, or for any stale pin with ``--strict``. Standard library
+only.
+
+Only full lowercase 64-hex digests in text files are found. Abbreviated
+digests, other hash functions and pins computed over decompressed payloads or
+canonical JSON are outside this search; an empty report is therefore not proof
+that no frozen evidence changed. A trigger and drift guard, not a replay or a
+mathematical check.
+
+Prepared for apoursh with substantial Anthropic Claude assistance. Apache-2.0.
+"""
+import argparse
+from hashlib import sha256
+import json
+from pathlib import Path
+import subprocess
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+CANONICAL = 'https://github.com/CrocSwap/integer-mult-bounds.git'
+FROZEN_PREFIXES = ('research/', 'upstream/')
+
+
+def git(root, *args, check=True, data=None):
+    result = subprocess.run(['git', '--no-replace-objects', *args], cwd=root, input=data,
+                            capture_output=True, check=False)
+    if check and result.returncode != 0:
+        raise SystemExit('git %s failed: %s' % (' '.join(args[:3]), result.stderr.decode(errors='replace').strip()))
+    return result
+
+
+def blobs(root, specs):
+    """Return bytes for each ``rev:path`` spec, or None when it is absent."""
+    if not specs:
+        return {}
+    out = git(root, 'cat-file', '--batch', data=''.join(s + '\n' for s in specs).encode()).stdout
+    found, pos = {}, 0
+    for spec in specs:
+        end = out.index(b'\n', pos)
+        header = out[pos:end].split()
+        pos = end + 1
+        if header[-1] == b'missing' or header[1] != b'blob':
+            found[spec] = None
+            continue
+        size = int(header[2])
+        found[spec] = out[pos:pos + size]
+        pos += size + 1
+    return found
+
+
+def changed_paths(root, base, head):
+    rows = git(root, 'diff', '--name-status', '-z', '--no-renames', base, head).stdout.split(b'\0')
+    changed = {}
+    for status, name in zip(rows[0::2], rows[1::2]):
+        status = status.decode()
+        if status[:1] in 'MDT':
+            changed[name.decode()] = 'deleted' if status[:1] == 'D' else 'modified'
+    return changed
+
+
+def check(root, base, head='HEAD'):
+    root = Path(root).resolve()
+    merge_base = git(root, 'merge-base', base, head).stdout.decode().strip()
+    head_commit = git(root, 'rev-parse', head + '^{commit}').stdout.decode().strip()
+    changed = changed_paths(root, merge_base, head_commit)
+    old = blobs(root, [merge_base + ':' + p for p in changed])
+    digests = {}
+    for path in changed:
+        data = old[merge_base + ':' + path]
+        if data is not None:
+            digests.setdefault(sha256(data).hexdigest(), []).append(path)
+    pins = []
+    if digests:
+        patterns = [arg for d in sorted(digests) for arg in ('-e', d)]
+        grep = git(root, 'grep', '-I', '-l', '-z', '-F', *patterns, merge_base, '--', check=False)
+        if grep.returncode not in (0, 1):
+            raise SystemExit('git grep failed: ' + grep.stderr.decode(errors='replace').strip())
+        pinners = [entry.decode().split(':', 1)[1] for entry in grep.stdout.split(b'\0') if entry]
+        before = blobs(root, [merge_base + ':' + p for p in pinners])
+        after = blobs(root, [head_commit + ':' + p for p in pinners])
+        for pinner in pinners:
+            text_before = before[merge_base + ':' + pinner]
+            text_after = after[head_commit + ':' + pinner]
+            for digest, targets in digests.items():
+                if digest.encode() not in text_before:
+                    continue
+                if text_after is None:
+                    state = 'pinner_removed'
+                elif digest.encode() in text_after:
+                    state = 'stale'
+                else:
+                    state = 'repinned'
+                for target in targets:
+                    pins.append({'pinned': target, 'change': changed[target], 'pinner': pinner,
+                                 'sha256': digest, 'state': state,
+                                 'frozen': target.startswith(FROZEN_PREFIXES)})
+    pins.sort(key=lambda row: (not row['frozen'], row['state'] != 'stale', row['pinned'], row['pinner']))
+    return {'merge_base': merge_base, 'head': head_commit, 'changed_files': len(changed),
+            'pinned_changed_files': len({row['pinned'] for row in pins}),
+            'stale': sum(row['state'] == 'stale' for row in pins),
+            'frozen_stale': sum(row['state'] == 'stale' and row['frozen'] for row in pins), 'pins': pins,
+            'scope': 'Full lowercase SHA-256 hex digests in merge-base text files only; '
+                     'not a replay, and an empty report does not prove frozen evidence is unchanged.'}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[0])
+    parser.add_argument('--root', type=Path, default=ROOT)
+    parser.add_argument('--base', default='main',
+                        help='base revision; for a fork, fetch %s main and pass FETCH_HEAD or its commit' % CANONICAL)
+    parser.add_argument('--head', default='HEAD')
+    parser.add_argument('--json', action='store_true', help='print the full machine-readable report')
+    parser.add_argument('--strict', action='store_true', help='also fail on stale pins of living records')
+    args = parser.parse_args(argv)
+    report = check(args.root, args.base, args.head)
+    if args.json:
+        print(json.dumps(report, indent=2))
+    else:
+        print('merge base %s, head %s: %d modified or deleted files, %d with pinned old bytes, '
+              '%d stale pins (%d on frozen research/upstream files)'
+              % (report['merge_base'][:12], report['head'][:12], report['changed_files'],
+                 report['pinned_changed_files'], report['stale'], report['frozen_stale']))
+        for row in report['pins']:
+            if row['frozen'] or args.strict:
+                print('%-14s %s  pinned by  %s' % (row['state'].upper(), row['pinned'], row['pinner']))
+        advisory = sorted({row['pinned'] for row in report['pins'] if not row['frozen']})
+        if advisory and not args.strict:
+            print('Advisory: living records with old-byte pins (list them with --strict): ' + ', '.join(advisory))
+        print(report['scope'])
+    failing = report['stale'] if args.strict else report['frozen_stale']
+    return 1 if failing else 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tests/test_check_pinned_edits.py

@@ -0,0 +1,126 @@
+"""A branch that changes hash-pinned bytes must report each stale or rewritten pin."""
+from contextlib import redirect_stdout
+from hashlib import sha256
+import importlib.util
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location('check_pinned_edits', ROOT / 'scripts/check_pinned_edits.py')
+guard = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(guard)
+ENV = {**os.environ, 'GIT_AUTHOR_NAME': 't', 'GIT_AUTHOR_EMAIL': 't@example.invalid',
+       'GIT_COMMITTER_NAME': 't', 'GIT_COMMITTER_EMAIL': 't@example.invalid', 'GIT_CONFIG_GLOBAL': os.devnull}
+
+
+def digest(data):
+    return sha256(data).hexdigest()
+
+
+class Fixture(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+        self.run_git('init', '-q', '-b', 'main')
+        self.write('research/pkg/RESULT.json', b'{"kappa": "1/3"}\n')
+        self.write('research/pkg/notes.md', b'unpinned\n')
+        self.write('research/pkg/data.bin', b'\0binary\0')
+        old = digest(b'{"kappa": "1/3"}\n')
+        selected = json.dumps({'source_sha256': old}).encode()
+        self.write('certificates/selected.json', selected)
+        self.write('research/other/SOURCES.json',
+                   json.dumps({'files': [{'path': 'research/pkg/RESULT.json', 'sha256': old}]}).encode())
+        # A package records the selected record it was written against: context, not a frozen input.
+        self.write('research/other/README.md',
+                   ('Pinned at `%s`; selected record `%s`.\n' % (old, digest(selected))).encode())
+        self.commit('base')
+        self.run_git('switch', '-q', '-c', 'branch')
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def run_git(self, *args):
+        subprocess.run(['git', *args], cwd=self.root, env=ENV, check=True, capture_output=True)
+
+    def write(self, rel, data):
+        path = self.root / rel
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_bytes(data)
+
+    def commit(self, message):
+        self.run_git('add', '-A')
+        self.run_git('commit', '-q', '-m', message)
+
+    def report(self):
+        return guard.check(self.root, 'main', 'HEAD')
+
+    def test_unpinned_edit_is_clean(self):
+        self.write('research/pkg/notes.md', b'edited\n')
+        self.commit('edit unpinned note')
+        report = self.report()
+        self.assertEqual((report['changed_files'], report['pins'], report['stale']), (1, [], 0))
+        with redirect_stdout(io.StringIO()):
+            self.assertEqual(guard.main(['--root', str(self.root), '--base', 'main']), 0)
+
+    def test_in_place_edit_reports_every_stale_pin(self):
+        self.write('research/pkg/RESULT.json', b'{"kappa": "1/2"}\n')
+        self.commit('retune in place')
+        report = self.report()
+        self.assertEqual(report['stale'], 3)
+        self.assertEqual({row['pinner'] for row in report['pins']},
+                         {'certificates/selected.json', 'research/other/SOURCES.json', 'research/other/README.md'})
+        self.assertTrue(all(row['pinned'] == 'research/pkg/RESULT.json' and row['change'] == 'modified'
+                            for row in report['pins']))
+        with redirect_stdout(io.StringIO()):
+            self.assertEqual(guard.main(['--root', str(self.root), '--base', 'main']), 1)
+
+    def test_repinned_and_removed_pinners_are_reported_but_not_stale(self):
+        new = digest(b'{"kappa": "1/2"}\n')
+        self.write('research/pkg/RESULT.json', b'{"kappa": "1/2"}\n')
+        self.write('certificates/selected.json', json.dumps({'source_sha256': new}).encode())
+        self.write('research/other/SOURCES.json', json.dumps({'files': [{'path': 'research/pkg/RESULT.json', 'sha256': new}]}).encode())
+        (self.root / 'research/other/README.md').unlink()
+        self.commit('rewrite pins')
+        states = {row['pinner']: row['state'] for row in self.report()['pins']}
+        self.assertEqual(states, {'certificates/selected.json': 'repinned', 'research/other/SOURCES.json': 'repinned',
+                                  'research/other/README.md': 'pinner_removed'})
+        self.assertEqual(self.report()['stale'], 0)
+
+    def test_deletion_and_binary_files(self):
+        (self.root / 'research/pkg/RESULT.json').unlink()
+        self.write('research/pkg/data.bin', b'\0changed\0')
+        self.commit('delete pinned file')
+        report = self.report()
+        self.assertEqual(report['changed_files'], 2)
+        self.assertEqual({row['change'] for row in report['pins']}, {'deleted'})
+        self.assertEqual(report['stale'], 3)
+
+    def test_living_record_update_is_advisory_unless_strict(self):
+        self.write('certificates/selected.json', b'{"source_sha256": "next record"}')
+        self.commit('maintainer updates the selected record')
+        report = self.report()
+        self.assertEqual((report['stale'], report['frozen_stale']), (1, 0))
+        self.assertEqual(report['pins'][0]['pinner'], 'research/other/README.md')
+        argv = ['--root', str(self.root), '--base', 'main']
+        with redirect_stdout(io.StringIO()):
+            self.assertEqual(guard.main(argv), 0)
+            self.assertEqual(guard.main(argv + ['--strict']), 1)
+
+    def test_uses_merge_base_not_moved_main(self):
+        self.write('research/pkg/notes.md', b'branch edit\n')
+        self.commit('branch')
+        self.run_git('switch', '-q', 'main')
+        self.write('research/pkg/RESULT.json', b'{"kappa": "2/5"}\n')
+        self.commit('main moved on')
+        self.run_git('switch', '-q', 'branch')
+        report = self.report()
+        self.assertEqual((report['changed_files'], report['stale']), (1, 0))
+
+
+if __name__ == '__main__':
+    unittest.main()