prj-canon-federation/tools/validate_fleet_references.py
tegwick 8a1f1b9fb2 Complete G8 fleet source-reference sweep with provenance exceptions
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a070b5-4994-7271-bd8b-7c3dbcedec4b
2026-09-06 02:42:55 +02:00

86 lines
3.8 KiB
Python

"""Validate reviewed legacy-reference exceptions across adjacent Git repositories.
Scans tracked files plus nonignored untracked files, including hidden files and symlink target paths.
Matches are fingerprints, not embedded file content. Git internals and ignored untracked
dependency/cache trees are outside the source scan; live registry/cache
and service checks are separate evidence.
"""
from __future__ import annotations
import argparse
from collections import Counter
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
PROJECT = Path(__file__).resolve().parents[1]
PATTERN = re.compile('identity' + r'[-_ ]?canon(?![a-z])', re.I)
AUDIT_FILES = {
'ledger/reference-sweep.json',
'docs/evidence/2026-09-06-reference-sweep.json',
'docs/evidence/2026-09-06-reference-sweep.md',
'docs/evidence/2026-09-06-reuse-registration.json',
'docs/evidence/2026-09-06-reuse-composed.json',
'docs/evidence/2026-09-06-hub-coordinates.json',
}
def scan(workspace):
repos = sorted(p for p in workspace.iterdir() if (p / '.git').exists())
matches = {}
total_files = 0
for repo in repos:
paths = subprocess.check_output(['git', 'ls-files', '-z', '--cached', '--others', '--exclude-standard'], cwd=repo)
for relative in sorted(set(p.decode() for p in paths.split(b'\0') if p)):
if repo.name == PROJECT.name and relative in AUDIT_FILES:
continue
path = repo / relative
if path.is_symlink():
# Audit the routing pointer even if its destination is absent.
data = os.readlink(path).encode()
elif path.is_file():
data = path.read_bytes()
else:
continue
total_files += 1
if b'\0' in data:
continue
lines = data.decode('utf-8', errors='replace').splitlines()
found = [{'line': i, 'sha256': hashlib.sha256(line.encode()).hexdigest()}
for i, line in enumerate(lines, 1) if PATTERN.search(line)]
if found:
matches[f'{repo.name}/{relative}'] = found
return repos, total_files, matches
def validate(workspace, ledger_path):
ledger = json.loads(ledger_path.read_text())
repos, file_count, found = scan(workspace)
accepted = {e['file']: e for e in ledger['exceptions']}
errors = []
for name, rows in found.items():
entry = accepted.get(name)
if not entry:
errors.append({'file': name, 'reason': 'unreviewed reference', 'lines': [r['line'] for r in rows]})
elif Counter(r['sha256'] for r in rows) != Counter(entry['line_sha256']):
errors.append({'file': name, 'reason': 'reference content changed; review required'})
for name in accepted.keys() - found.keys():
errors.append({'file': name, 'reason': 'obsolete exception; remove or review'})
missing = set(ledger['repositories']) - {p.name for p in repos}
errors.extend({'repo': r, 'reason': 'reviewed checkout missing'} for r in sorted(missing))
return {'gate': 'G8-source-scan', 'result': 'pass' if not errors else 'fail',
'repositories_scanned': len(repos), 'files_scanned': file_count,
'reviewed_exception_files': len(accepted), 'matching_lines': sum(map(len, found.values())),
'errors': errors}
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--workspace', type=Path, default=PROJECT.parent)
parser.add_argument('--ledger', type=Path, default=PROJECT / 'ledger/reference-sweep.json')
args = parser.parse_args()
result = validate(args.workspace.resolve(), args.ledger)
print(json.dumps(result, indent=2))
raise SystemExit(0 if result['result'] == 'pass' else 1)