Digital Drosophila Archive
325 lines
18 KiB
Python
325 lines
18 KiB
Python
"""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())
|