| |
| """Generate or verify the redistributable code file inventory, using SHA-256.""" |
| from __future__ import annotations |
| import argparse |
| import hashlib |
| import re |
| import sys |
| from pathlib import Path |
|
|
| IGNORED_DIRS = {'.git', '__pycache__', '.ipynb_checkpoints', '.venv', 'venv', '.pytest_cache', 'outputs', 'logs', 'checkpoints', 'results'} |
| IGNORED_NAMES = {'.DS_Store', '.coverage'} |
|
|
|
|
| def digest(path: Path) -> str: |
| h = hashlib.sha256() |
| with path.open('rb') as f: |
| for block in iter(lambda: f.read(1024 * 1024), b''): |
| h.update(block) |
| return h.hexdigest() |
|
|
|
|
| def inventory(root: Path, manifest: Path) -> dict[str, Path]: |
| files = {} |
| for path in root.rglob('*'): |
| rel = path.relative_to(root) |
| if any(part in IGNORED_DIRS or part.endswith('.egg-info') for part in rel.parts): |
| continue |
| if rel.parts[:2] == ('dataset_metadata', 'manifests_h100'): |
| continue |
| if rel.parts[0] in ('data', 'third_party') and rel.as_posix() != rel.parts[0] + '/README.md': |
| continue |
| if not path.is_file() or path.resolve() in (manifest.resolve(), root / 'SHA256SUMS'): |
| continue |
| if path.name in IGNORED_NAMES or path.suffix in ('.pyc', '.pyo'): |
| continue |
| if path.is_symlink(): |
| raise ValueError(f'Symlinks are not supported in the code inventory: {rel}') |
| name = rel.as_posix() |
| if '\n' in name or '\r' in name: |
| raise ValueError('Newlines are not supported in manifest filenames') |
| files[name] = path |
| return dict(sorted(files.items())) |
|
|
|
|
| def verify(root: Path, manifest: Path) -> list[str]: |
| errors, entries = [], {} |
| for line_no, line in enumerate(manifest.read_text(encoding='utf-8').splitlines(), 1): |
| match = re.fullmatch(r'([0-9a-fA-F]{64}) (.+)', line) |
| if not match: |
| raise ValueError(f'Invalid manifest entry at line {line_no}') |
| expected, name = match.groups() |
| path = root / name |
| if name in entries or not path.resolve().is_relative_to(root): |
| raise ValueError(f'Duplicate or escaping manifest path: {name}') |
| entries[name] = expected.lower() |
| if not path.is_file() or path.is_symlink(): |
| errors.append(f'Missing or unsupported: {name}') |
| elif digest(path) != expected.lower(): |
| errors.append(f'Hash mismatch: {name}') |
| expected_files = set(inventory(root, manifest)) |
| errors.extend(f'Unlisted file: {p}' for p in sorted(expected_files - entries.keys())) |
| errors.extend(f'Unexpected manifest file: {p}' for p in sorted(entries.keys() - expected_files)) |
| if not entries: |
| errors.append('Checksum manifest is empty') |
| return errors |
|
|
|
|
| def main() -> int: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('command', choices=['generate', 'gen', 'g', 'verify', 'check', 'c']) |
| parser.add_argument('root', type=Path, nargs='?', default=Path.cwd()) |
| parser.add_argument('manifest', type=Path, nargs='?') |
| args = parser.parse_args() |
| root = args.root.resolve() |
| if not root.is_dir(): |
| parser.error('Root must be an existing directory') |
| |
| manifest = args.manifest.resolve() if args.manifest else root / 'SHA256SUMS' |
| try: |
| if args.command in ('generate', 'gen', 'g'): |
| files = inventory(root, manifest) |
| content = ''.join(f'{digest(path)} {name}\n' for name, path in files.items()) |
| manifest.parent.mkdir(parents=True, exist_ok=True) |
| manifest.write_text(content, encoding='utf-8') |
| print(f'Generated {len(files)} checksums: {manifest}') |
| else: |
| errors = verify(root, manifest) |
| if errors: |
| print('\n'.join(errors), file=sys.stderr) |
| return 1 |
| print(f'PASS: {len(manifest.read_text().splitlines())} code checksums and file inventory match') |
| except (OSError, ValueError) as exc: |
| print(f'Integrity check failed: {exc}', file=sys.stderr) |
| return 1 |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| raise SystemExit(main()) |
|
|
