"""Verify fixed Phase 4AK Git blobs, not working-tree files or scientific gates. The publication receipt SHA-256 must come from an external trusted record. Only the index or one resolved commit is read. No publish, checkout, index mutation, network fetch, or scientific execution is performed. """ from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path, PurePosixPath import re import subprocess import sys PROJECT = '/data/digital-drosophila' MODEL = 'models/male-cns-lif/apl-timing-validation-v1' EXPERIMENT = 'experiments/phase4ak-apl-timing-validation' REGISTRY = 'registry/phase4ak-apl-timing-validation' CONTROLS = frozenset(('manifest.json', 'checksums.sha256')) SHA256 = re.compile(r'[0-9a-f]{64}\Z') OID = re.compile(r'(?:[0-9a-f]{40}|[0-9a-f]{64})\Z') MAX_EVIDENCE = 4 * 1024 ** 2 MAX_BLOB = 64 * 1024 ** 2 SUFFIXES = frozenset(('.json', '.csv', '.md', '.log', '.sha256')) def require(condition, message): if not condition: raise ValueError(message) def sha(data): return hashlib.sha256(data).hexdigest() def relative(value): require(isinstance(value, str) and bool(value) and not any(char in value for char in '\\\x00\n\r\t'), 'Invalid relative path') path = PurePosixPath(value) require(not path.is_absolute() and str(path) == value and all(part not in ('', '.', '..') and ':' not in part for part in path.parts), 'Noncanonical or escaping path: ' + value) return value def unique_object(pairs): value = {} for key, item in pairs: require(key not in value, 'Duplicate JSON key: ' + key) value[key] = item return value def json_bytes(data): def nonfinite(value): raise ValueError('Nonfinite JSON value: ' + value) return json.loads(data.decode('utf-8'), object_pairs_hook=unique_object, parse_constant=nonfinite) def rows_by_path(rows, label): require(isinstance(rows, list) and bool(rows), 'Missing/empty ' + label) result = {} for row in rows: require(isinstance(row, dict) and set(row) == {'path', 'bytes', 'sha256'}, 'Invalid ' + label + ' row schema') path = relative(row['path']) require(path not in result and '__pycache__' not in PurePosixPath(path).parts, 'Duplicate/cache ' + label + ' path: ' + path) require(type(row['bytes']) is int and row['bytes'] >= 0 and isinstance(row['sha256'], str) and SHA256.fullmatch(row['sha256']), 'Invalid bytes/SHA-256 in ' + label + ': ' + path) result[path] = row return result class GitBlobs: """Snapshot inventory plus immutable object reads; no work-tree fallback.""" def __init__(self, repo, *, staged, commit): require(type(staged) is bool and (staged != (commit is not None)), 'Select exactly one of staged or commit') self.repo = Path(repo).resolve(strict=True) require(self.repo.is_dir(), 'Repository must be a directory') self.env = os.environ.copy() # The explicitly supplied repo/index must not be redirected by caller env. for key in ('GIT_DIR', 'GIT_WORK_TREE', 'GIT_INDEX_FILE', 'GIT_COMMON_DIR', 'GIT_OBJECT_DIRECTORY', 'GIT_ALTERNATE_OBJECT_DIRECTORIES'): self.env.pop(key, None) self.env.update(GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1', GIT_NO_LAZY_FETCH='1') self.base = ['git', '--no-optional-locks', '-c', 'core.fsmonitor=false', '-c', 'gc.auto=0', '-C', str(self.repo)] top = self.command('rev-parse', '--show-toplevel').decode('utf-8').strip() require(Path(top).resolve(strict=True) == self.repo, 'Supply the Git repository top-level directory') if staged: self.commit = None else: require(isinstance(commit, str) and bool(commit) and not commit.startswith('-') and not any(c in commit for c in '\x00\n\r'), 'Invalid commit selector') self.commit = self.command('rev-parse', '--verify', '--end-of-options', commit + '^{commit}').decode('ascii').strip() require(OID.fullmatch(self.commit), 'Commit did not resolve to a full object ID') self.staged = staged self.inventory_bytes = self.inventory() self.entries = self.parse_inventory(self.inventory_bytes) self.verified = {} def command(self, *args): result = subprocess.run(self.base + list(args), env=self.env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) require(result.returncode == 0, 'Git read failed: ' + result.stderr.decode('utf-8', 'replace').strip()) return result.stdout def inventory(self): if self.staged: return self.command('ls-files', '--stage', '--full-name', '-z', '--', MODEL, REGISTRY) return self.command('ls-tree', '-r', '-z', '--full-tree', self.commit, '--', MODEL, REGISTRY) def parse_inventory(self, data): result = {} require(data.endswith(b'\x00'), 'Empty/malformed scoped Git inventory') for entry in data[:-1].split(b'\x00'): metadata, raw_path = entry.split(b'\t', 1) fields = metadata.decode('ascii').split(' ') require(len(fields) == 3, 'Malformed Git entry') if self.staged: mode, oid, stage = fields require(stage == '0', 'Unmerged index entry is forbidden') else: mode, kind, oid = fields require(kind == 'blob', 'Non-blob Git entry is forbidden') path = relative(raw_path.decode('utf-8')) require(path.startswith(MODEL + '/') or path.startswith(REGISTRY + '/'), 'Git entry escapes the fixed AK scopes: ' + path) require(mode in ('100644', '100755') and OID.fullmatch(oid), 'Symlink/gitlink/nonregular Git entry is forbidden: ' + path) require(path not in result, 'Duplicate Git path: ' + path) result[path] = dict(mode=mode, oid=oid) return result def read(self, path): relative(path) require(path in self.entries, 'Missing Git blob: ' + path) oid = self.entries[path]['oid'] size = int(self.command('cat-file', '-s', oid).decode('ascii').strip()) require(0 <= size <= MAX_BLOB, 'Git blob exceeds publication verifier limit: ' + path) # cat-file blob addresses the object, so attributes/filters/autocrlf do # not turn a corrupt staged blob into a good working-tree substitute. data = self.command('cat-file', 'blob', oid) require(len(data) == size, 'Git object size changed: ' + path) self.verified[path] = dict(path=path, bytes=len(data), sha256=sha(data), git_oid=oid, git_mode=self.entries[path]['mode']) return data def match(self, path, row): data = self.read(path) require(len(data) == row['bytes'], 'Git blob byte count mismatch: ' + path) require(sha(data) == row['sha256'], 'Git blob SHA-256 mismatch: ' + path) return data def finish(self): require(self.inventory() == self.inventory_bytes, 'Scoped Git inventory changed during verification') def seal_controls(manifest_data, checksum_data, expected_relative, expected_sha): """Validate control syntax/inventory; does not claim absent bulk was read.""" require(sha(manifest_data) == expected_sha, 'Receipt-bound manifest SHA-256 mismatch') manifest = json_bytes(manifest_data) require(isinstance(manifest, dict) and set(manifest) == {'root', 'files'} and manifest['root'] == PROJECT + '/' + expected_relative, 'Wrong seal manifest schema/root') rows = rows_by_path(manifest['files'], 'manifest') require(not (set(rows) & CONTROLS), 'Seal controls listed as payload') checksums = {} for line in checksum_data.decode('utf-8').splitlines(): require(len(line) >= 67 and line[64:66] == ' ' and SHA256.fullmatch(line[:64]), 'Malformed checksum line') path = relative(line[66:]) require(path not in checksums, 'Duplicate checksum path: ' + path) checksums[path] = line[:64] require(checksums == {**{name: row['sha256'] for name, row in rows.items()}, 'manifest.json': sha(manifest_data)}, 'Checksum inventory or manifest self-checksum mismatch') return rows def control_rows(manifest_data, checksum_data): return {name: dict(path=name, bytes=len(data), sha256=sha(data)) for name, data in (('manifest.json', manifest_data), ('checksums.sha256', checksum_data))} def verify_git_payload(repo, receipt_sha256, *, staged=False, commit=None): require(isinstance(receipt_sha256, str) and SHA256.fullmatch(receipt_sha256), 'External publication receipt SHA-256 is required') blobs = GitBlobs(repo, staged=staged, commit=commit) receipt_data = blobs.read(REGISTRY + '/publication-receipt.json') require(len(receipt_data) <= MAX_EVIDENCE and sha(receipt_data) == receipt_sha256, 'External publication receipt SHA-256 mismatch') receipt = json_bytes(receipt_data) require(isinstance(receipt, dict) and receipt.get('publication_complete') is True and receipt.get('registry') == REGISTRY and receipt.get('project_root') == PROJECT and receipt.get('model_payload_remains_at') == MODEL and receipt.get('bulk_remains_at') == EXPERIMENT, 'Wrong/incomplete AK publication receipt or fixed scope') for key in ('model_manifest_sha256', 'experiment_manifest_sha256'): require(isinstance(receipt.get(key), str) and SHA256.fullmatch(receipt[key]), 'Invalid receipt manifest anchor: ' + key) registry_rows = rows_by_path(receipt.get('files'), 'receipt files') require('publication-receipt.json' not in registry_rows, 'Receipt must not list itself') require({'evidence-index.json', 'seal-verification.json', 'model/manifest.json', 'model/checksums.sha256', 'evidence/manifest.json', 'evidence/checksums.sha256'} <= set(registry_rows), 'Required registry evidence/control missing') observed_registry = {path[len(REGISTRY) + 1:] for path in blobs.entries if path.startswith(REGISTRY + '/')} require(observed_registry == set(registry_rows) | {'publication-receipt.json'}, 'Registry Git inventory has unlisted/missing files: ' + repr(sorted( observed_registry ^ (set(registry_rows) | {'publication-receipt.json'})))) controls_data = {} for path, row in registry_rows.items(): require(row['bytes'] <= MAX_EVIDENCE, 'Registry receipt file exceeds 4 MiB') data = blobs.match(REGISTRY + '/' + path, row) if path in ('evidence-index.json', 'seal-verification.json', 'model/manifest.json', 'model/checksums.sha256', 'evidence/manifest.json', 'evidence/checksums.sha256'): controls_data[path] = data model_manifest, model_checksums = (controls_data['model/' + name] for name in ('manifest.json', 'checksums.sha256')) model_rows = seal_controls(model_manifest, model_checksums, MODEL, receipt['model_manifest_sha256']) model_controls = control_rows(model_manifest, model_checksums) observed_model = {path[len(MODEL) + 1:] for path in blobs.entries if path.startswith(MODEL + '/')} require(observed_model == set(model_rows) | CONTROLS, 'Model Git inventory has unlisted/missing files: ' + repr(sorted(observed_model ^ (set(model_rows) | CONTROLS)))) for path, row in {**model_rows, **model_controls}.items(): blobs.match(MODEL + '/' + path, row) exp_manifest, exp_checksums = (controls_data['evidence/' + name] for name in ('manifest.json', 'checksums.sha256')) exp_rows = seal_controls(exp_manifest, exp_checksums, EXPERIMENT, receipt['experiment_manifest_sha256']) exp_controls = control_rows(exp_manifest, exp_checksums) expected_index = {} omitted = [] for path, row in {**exp_rows, **exp_controls}.items(): if PurePosixPath(path).suffix.lower() in SUFFIXES and row['bytes'] <= MAX_EVIDENCE: expected_index['evidence/' + path] = dict(source=EXPERIMENT + '/' + path, copy='evidence/' + path, bytes=row['bytes'], sha256=row['sha256']) else: omitted.append(dict(row, reason='exceeds_4_MiB' if row['bytes'] > MAX_EVIDENCE else 'bulk_or_non_evidence_extension')) for path, row in model_controls.items(): expected_index['model/' + path] = dict(source=MODEL + '/' + path, copy='model/' + path, bytes=row['bytes'], sha256=row['sha256']) index = json_bytes(controls_data['evidence-index.json']) require(isinstance(index, list), 'Evidence index must be a list') actual_index = {} for row in index: require(isinstance(row, dict) and set(row) == {'source', 'copy', 'bytes', 'sha256'}, 'Invalid evidence-index row schema') path = relative(row['copy']); relative(row['source']) require(path not in actual_index and type(row['bytes']) is int and row['bytes'] >= 0 and isinstance(row['sha256'], str) and SHA256.fullmatch(row['sha256']), 'Invalid/duplicate evidence-index row') actual_index[path] = row require(actual_index == expected_index, 'Evidence index differs from receipt-bound seals/selection policy') require(set(registry_rows) == set(expected_index) | {'evidence-index.json', 'seal-verification.json'}, 'Receipt files disagree with copied evidence index') for path, row in expected_index.items(): require(registry_rows[path] == dict(path=path, bytes=row['bytes'], sha256=row['sha256']), 'Evidence-index SHA/bytes disagree with receipt files: ' + path) require(type(receipt.get('copied_evidence_files')) is int and receipt['copied_evidence_files'] == len(expected_index) and type(receipt.get('copied_evidence_bytes')) is int and receipt['copied_evidence_bytes'] == sum(row['bytes'] for row in expected_index.values()), 'Receipt copied-evidence totals mismatch') omitted_receipt = receipt.get('experiment_payload_not_copied') require(isinstance(omitted_receipt, list) and all(isinstance(row, dict) for row in omitted_receipt) and sorted(omitted_receipt, key=lambda row: row.get('path', '')) == sorted(omitted, key=lambda row: row['path']), 'Receipt omitted-bulk inventory mismatch') proof = json_bytes(controls_data['seal-verification.json']) require(isinstance(proof, dict) and set(proof) == {'model', 'experiment'}, 'Invalid seal-verification evidence') for name, prefix, payload, controls in (('model', MODEL, model_rows, model_controls), ('experiment', EXPERIMENT, exp_rows, exp_controls)): record = proof[name] require(isinstance(record, dict) and record.get('pass_all') is True and record.get('root') == PROJECT + '/' + prefix and record.get('payload_file_count') == len(payload) and record.get('payload_bytes') == sum(row['bytes'] for row in payload.values()) and record.get('manifest_sha256') == controls['manifest.json']['sha256'] and record.get('checksums_sha256') == controls['checksums.sha256']['sha256'], 'Seal-verification evidence contradicts receipt-bound ' + name + ' controls') blobs.finish() return dict(pass_all=True, verification='Phase 4AK actual Git blob publication integrity', source='staged-index' if staged else 'commit', commit=blobs.commit, fixed_scopes=[MODEL, REGISTRY], receipt_sha256=receipt_sha256, model_manifest_sha256=receipt['model_manifest_sha256'], experiment_manifest_sha256=receipt['experiment_manifest_sha256'], declared_audit_status=receipt.get('declared_audit_status'), model_payload_files=len(model_rows), registry_receipt_files=len(registry_rows), verified_blob_count=len(blobs.verified), verified_bytes=sum(row['bytes'] for row in blobs.verified.values()), scoped_inventory_sha256=sha(blobs.inventory_bytes), verified_blobs=[blobs.verified[path] for path in sorted(blobs.verified)], working_tree_payload_used=False, git_index_modified=False, experiment_bulk_verified_by_this_tool=False, scientific_gate_executed=False, publication_integrity_is_not_scientific_pass=True) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--repo', required=True, help='Git repository top-level directory') parser.add_argument('--receipt-sha256', required=True, help='Trusted external publication-receipt.json SHA-256') choice = parser.add_mutually_exclusive_group(required=True) choice.add_argument('--staged', action='store_true', help='Verify the actual stage-0 index blobs') choice.add_argument('--commit', help='Resolve and verify this commit without checking it out') args = parser.parse_args(argv) try: result = verify_git_payload(args.repo, args.receipt_sha256, staged=args.staged, commit=args.commit) except (ValueError, OSError, UnicodeError) as error: result = dict(pass_all=False, verification='Phase 4AK actual Git blob publication integrity', error_type=type(error).__name__, error=str(error), scientific_gate_executed=False, publication_integrity_is_not_scientific_pass=True) print(json.dumps(result, ensure_ascii=False, allow_nan=False)) return 1 print(json.dumps(result, ensure_ascii=False, allow_nan=False)) return 0 if __name__ == '__main__': sys.exit(main())