File size: 4,221 Bytes
4b8c78e
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
#!/usr/bin/env python3
"""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')
    # Explicit relative manifest paths are resolved against the caller's cwd.
    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())