prj-canon-federation/tools/validate_fleet_references.py

87 lines
3.8 KiB
Python
Raw Normal View History

"""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)