← back to Ticket System
data/codex-yoloforever/cycle-20260908T1322Z.m4gNE5/ops-object-verifier-v1.py
412 lines
#!/usr/bin/env python3
"""Scratch-only SHA-1 Git object validator; receipts are resumable, not git fsck.
No source writes. Exactly two cat-file subprocesses maximum. Stream blobs in
1 MiB chunks; structured objects have a 2 MiB cap. Every successful object has
an fsynced hash-chained receipt. Torn final lines are retained and uncredited.
"""
import argparse
import concurrent.futures
import hashlib
import json
import os
from pathlib import Path
import queue
import re
import select
import stat
import subprocess
import threading
import time
import uuid
VERSION = 'ops-object-verifier-v1'
GUARD = Path('/Users/macstudio3/Projects/ticket-system/config/dtd-cost-mode')
HEX = re.compile(r'^[0-9a-f]{40}$')
CHUNK = 1024 * 1024
STRUCT_LIMIT = 2 * CHUNK
def canon(x):
return json.dumps(x, sort_keys=True, separators=(',', ':')).encode()
def sha(path):
h = hashlib.sha256()
with Path(path).open('rb') as f:
for b in iter(lambda: f.read(CHUNK), b''):
h.update(b)
return h.hexdigest()
def write_json(path, data):
with Path(path).open('x') as f:
json.dump(data, f, indent=2)
f.write('\n')
f.flush()
os.fsync(f.fileno())
def env():
e = {k: v for k, v in os.environ.items() if not k.startswith('GIT_')}
e.update(GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null',
GIT_OPTIONAL_LOCKS='0', GIT_NO_REPLACE_OBJECTS='1')
return e
def gitcmd(repo, args):
return ['git', '--no-pager', '--git-dir=' + str(repo),
'-c', 'core.commitGraph=false', '-c', 'gc.auto=0', *args]
def command(repo, args, deadline, commands):
c = gitcmd(repo, args)
p = subprocess.run(c, env=env(), capture_output=True,
timeout=max(.01, deadline-time.monotonic()))
commands.append({'argv': c, 'exit': p.returncode,
'stdout_sha256': hashlib.sha256(p.stdout).hexdigest(),
'stderr': p.stderr.decode(errors='replace')[:1000]})
if p.returncode:
raise ValueError('git command failed: ' + repr(commands[-1]))
return p.stdout
def inventory(path):
out = {}
for line in Path(path).read_text().splitlines():
oid, kind, size, stored = line.split()
if not HEX.fullmatch(oid) or oid in out or kind not in ('commit', 'tree', 'blob', 'tag'):
raise ValueError('malformed/duplicate inventory row: ' + line)
out[oid] = (kind, int(size), int(stored))
if not out:
raise ValueError('empty inventory')
return out
def inspect_source(repo, manifest_path, require_immutable, deadline):
expected = json.loads(Path(manifest_path).read_text())
actual = {}
immutable = 0
for p in [repo, *repo.rglob('*')]:
if time.monotonic() >= deadline:
raise TimeoutError('deadline during manifest verification')
s = p.lstat()
if stat.S_ISLNK(s.st_mode):
raise ValueError('source symlink forbidden: ' + str(p))
if s.st_flags & stat.UF_IMMUTABLE:
immutable += 1
elif require_immutable:
raise ValueError('source immutable flag missing: ' + str(p))
if p.is_file():
rel = str(p.relative_to(repo))
if rel not in expected:
raise ValueError('unexpected source file: ' + rel)
x = expected[rel]
if s.st_size != x['bytes'] or sha(p) != x['sha256']:
raise ValueError('source manifest mismatch: ' + rel)
actual[rel] = {'bytes': s.st_size, 'sha256': x['sha256']}
if set(actual) != set(expected):
raise ValueError('missing source files: ' + repr(sorted(set(expected)-set(actual))))
for rel in ('objects/info/alternates', 'objects/info/http-alternates', 'shallow', 'info/grafts'):
if (repo/rel).exists():
raise ValueError('external/replaced/shallow object source forbidden: ' + rel)
if (repo/'refs/replace').exists():
raise ValueError('replacement refs forbidden')
return {'files': len(actual), 'bytes': sum(x['bytes'] for x in actual.values()),
'sha256_verified': len(actual), 'immutable_entries': immutable}
def edges(kind, data, inv):
links = []
external = []
if kind == 'tree':
at = 0
while at < len(data):
sp = data.find(b' ', at)
nul = data.find(b'\0', sp+1)
if sp < at or nul < sp or nul+21 > len(data):
raise ValueError('malformed tree')
mode = data[at:sp]
name = data[sp+1:nul]
oid = data[nul+1:nul+21].hex()
if not name or b'/' in name or name in (b'.', b'..'):
raise ValueError('invalid tree name')
if mode == b'160000':
external.append(oid)
elif mode in (b'40000', b'040000'):
links.append((oid, 'tree'))
elif mode in (b'100644', b'100755', b'120000'):
links.append((oid, 'blob'))
else:
raise ValueError('unsupported tree mode ' + repr(mode))
at = nul+21
elif kind in ('commit', 'tag'):
header = data.split(b'\n\n', 1)[0].splitlines()
if kind == 'commit':
trees = [x[5:].decode() for x in header if x.startswith(b'tree ')]
if len(trees) != 1:
raise ValueError('commit must have one tree')
links.append((trees[0], 'tree'))
links.extend((x[7:].decode(), 'commit') for x in header if x.startswith(b'parent '))
else:
objects = [x[7:].decode() for x in header if x.startswith(b'object ')]
types = [x[5:].decode() for x in header if x.startswith(b'type ')]
if len(objects) != 1 or len(types) != 1:
raise ValueError('tag must have one object and type')
links.append((objects[0], types[0]))
for oid, expected in links:
if not HEX.fullmatch(oid) or oid not in inv:
raise ValueError('missing linked object: ' + oid)
if inv[oid][0] != expected:
raise ValueError('linked object type mismatch: ' + oid)
return {'internal_edges': len(links), 'edges_sha256': hashlib.sha256(canon(links)).hexdigest(),
'external_gitlinks': len(external),
'external_gitlinks_sha256': hashlib.sha256(canon(external)).hexdigest()}
class Batch:
def __init__(self, repo, deadline, stderr_path, commands):
self.deadline = deadline
self.buffer = bytearray()
self.err = Path(stderr_path).open('xb')
self.argv = gitcmd(repo, ['cat-file', '--batch'])
self.p = subprocess.Popen(self.argv, env=env(), stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=self.err, bufsize=0)
self.entry = {'argv': self.argv, 'pid': self.p.pid, 'stderr_path': str(stderr_path)}
commands.append(self.entry)
def fill(self):
remaining = self.deadline-time.monotonic()
if remaining <= 0:
raise TimeoutError('data deadline')
if not select.select([self.p.stdout], [], [], remaining)[0]:
raise TimeoutError('cat-file deadline')
b = os.read(self.p.stdout.fileno(), CHUNK)
if not b:
raise ValueError('cat-file unexpectedly ended')
self.buffer.extend(b)
def take(self, n):
while not self.buffer:
self.fill()
n = min(n, len(self.buffer))
b = bytes(self.buffer[:n])
del self.buffer[:n]
return b
def line(self):
while b'\n' not in self.buffer:
if len(self.buffer) > 512:
raise ValueError('oversized cat-file header')
self.fill()
i = self.buffer.index(10)+1
b = bytes(self.buffer[:i])
del self.buffer[:i]
return b
def validate(self, oid, inv):
self.p.stdin.write((oid+'\n').encode())
head = self.line().decode().strip().split()
if len(head) != 3:
raise ValueError('invalid/missing object header: ' + repr(head))
got, kind, n = head
n = int(n)
if (got, kind, n) != (oid, inv[oid][0], inv[oid][1]):
raise ValueError('object type/size identity mismatch: ' + oid)
if kind != 'blob' and n > STRUCT_LIMIT:
raise ValueError('structured object exceeds memory boundary')
h = hashlib.sha1((kind+' '+str(n)+'\0').encode())
structured = bytearray()
left = n
while left:
b = self.take(min(CHUNK, left))
h.update(b)
if kind != 'blob':
structured.extend(b)
left -= len(b)
if self.take(1) != b'\n':
raise ValueError('cat-file framing mismatch')
digest = h.hexdigest()
if digest != oid:
raise ValueError('Git SHA-1 mismatch: expected '+oid+' got '+digest)
linked = edges(kind, structured, inv)
return {'oid': oid, 'type': kind, 'bytes': n, 'git_hash': digest,
'hash_algorithm': 'sha1', 'connectivity': 'PASS', **linked}
def close(self):
# Only this verifier's child is affected. EOF normally exits cat-file;
# deadline interruption closes the pipe and terminates this child only.
self.p.stdin.close()
self.p.stdout.close()
try:
self.p.wait(timeout=.2)
except subprocess.TimeoutExpired:
self.p.terminate()
self.p.wait(timeout=2)
self.entry['exit'] = self.p.returncode
self.err.close()
def read_receipts(receipts, binding, inv):
done = {}
torn = []
for path in sorted(Path(receipts).glob('segment-*.jsonl')):
prev = binding
seq = 0
with path.open('rb') as f:
for raw in f:
if not raw.endswith(b'\n'):
torn.append({'path': str(path), 'uncredited_tail_bytes': len(raw)})
break
row = json.loads(raw)
digest = row.pop('receipt_sha256')
if hashlib.sha256(canon(row)).hexdigest() != digest:
raise ValueError('receipt tamper: ' + str(path))
if row['binding'] != binding or row['prev'] != prev or row['seq'] != seq:
raise ValueError('receipt binding/chain mismatch')
oid = row['oid']
if oid in done or oid not in inv:
raise ValueError('duplicate or foreign receipt')
if (row['type'], row['bytes']) != inv[oid][:2] or row['git_hash'] != oid:
raise ValueError('receipt object identity mismatch')
if row['connectivity'] != 'PASS':
raise ValueError('failed receipt cannot be credited')
done[oid] = row
prev = digest
seq += 1
return done, torn
def run(repo, inventory_path, manifest_path, output, seconds=600, limit=None,
require_immutable=False, audit=False):
started = time.monotonic()
deadline = started + min(seconds, 600)
runid = str(uuid.uuid4())
output = Path(output)
output.mkdir(exist_ok=True, parents=True)
receipts = output/'receipts'
receipts.mkdir(exist_ok=True)
commands = []
report = {'version': VERSION, 'run_id': runid, 'started_epoch': time.time(),
'data_budget_seconds': min(seconds, 600), 'max_workers': 2,
'python_blob_chunk_bytes': CHUNK, 'structured_cap_bytes': STRUCT_LIMIT,
'source': str(repo), 'full_fsck_semantics': False}
try:
if GUARD.read_bytes() != b'ZERO_COST_REQUIRED\n':
raise ValueError('canonical zero-cost guard changed')
inv = inventory(inventory_path)
pins = {'version': VERSION, 'verifier_sha256': sha(__file__),
'source': str(Path(repo).resolve()),
'inventory_sha256': sha(inventory_path), 'manifest_sha256': sha(manifest_path),
'object_count': len(inv), 'logical_bytes': sum(x[1] for x in inv.values()),
'require_immutable': require_immutable}
binding = hashlib.sha256(canon(pins)).hexdigest()
pinpath = output/'pins.json'
if pinpath.exists():
if json.loads(pinpath.read_text()) != pins:
raise ValueError('resume input/verifier pins changed')
else:
write_json(pinpath, pins)
report['binding'] = binding
report['pins'] = pins
report['source_manifest'] = inspect_source(Path(repo), manifest_path, require_immutable, deadline)
raw = command(repo, ['cat-file', '--batch-all-objects', '--batch-check=%(objectname) %(objecttype) %(objectsize) %(objectsize:disk)'], deadline, commands)
actual = sorted(raw.decode().splitlines())
expected = sorted(Path(inventory_path).read_text().splitlines())
if actual != expected:
raise ValueError('pinned inventory is not complete/exact for source')
report['inventory_exact'] = True
refs = command(repo, ['for-each-ref', '--format=%(objectname) %(refname)'], deadline, commands).decode().splitlines()
head = command(repo, ['rev-parse', '--verify', 'HEAD'], deadline, commands).decode().strip()
for root in [head, *[x.split()[0] for x in refs]]:
if root not in inv:
raise ValueError('missing ref/HEAD root: '+root)
report['roots'] = {'HEAD': head, 'refs': refs, 'all_in_inventory': True}
done, torn = read_receipts(receipts, binding, inv)
report['resume_credited'] = len(done)
report['retained_uncredited_tails'] = torn
pending = sorted(set(inv)-set(done), key=lambda oid: (inv[oid][0]=='blob', inv[oid][1], oid))
if limit is not None:
pending = pending[:limit]
q = queue.Queue()
for oid in pending:
q.put(oid)
stop = threading.Event()
errors = []
lock = threading.Lock()
def worker(idx):
batch = None
path = receipts/('segment-'+runid+'-'+str(idx)+'.jsonl')
prev = binding
seq = 0
try:
batch = Batch(repo, deadline, output/('cat-file-'+runid+'-'+str(idx)+'.stderr'), commands)
with path.open('xb', buffering=0) as out:
while not stop.is_set() and time.monotonic() < deadline:
try:
oid = q.get_nowait()
except queue.Empty:
break
row = batch.validate(oid, inv)
row.update(binding=binding, prev=prev, seq=seq, completed_epoch=time.time())
digest = hashlib.sha256(canon(row)).hexdigest()
row['receipt_sha256'] = digest
out.write(canon(row)+b'\n')
os.fsync(out.fileno())
with lock:
done[oid] = row
prev = digest
seq += 1
except Exception as ex:
with lock:
errors.append({'worker': idx, 'error': str(ex), 'class': type(ex).__name__})
stop.set()
finally:
if batch:
batch.close()
if not audit and pending:
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
futures = [pool.submit(worker, i) for i in range(2)]
for f in futures:
f.result()
done, torn = read_receipts(receipts, binding, inv)
complete = len(done) == len(inv)
report.update(completed=len(done), remaining=len(inv)-len(done),
completed_bytes=sum(inv[o][1] for o in done),
remaining_bytes=sum(v[1] for o,v in inv.items() if o not in done),
completed_by_type={t: sum(inv[o][0]==t for o in done) for t in ('blob','tree','commit','tag')},
internal_edges_checked=sum(r['internal_edges'] for r in done.values()),
gitlinks_external=sum(r['external_gitlinks'] for r in done.values()),
structured_objects_remaining=sum(v[0]!='blob' for o,v in inv.items() if o not in done),
roots_hash_verified=all(x in done for x in [head,*[r.split()[0] for r in refs]]),
retained_uncredited_tails=torn, errors=errors)
report['verdict'] = 'PASS' if complete and not errors else ('FAIL' if any(e['class']!='TimeoutError' for e in errors) else 'PARTIAL')
report['completion_claim_allowed'] = complete and not errors
report['receipts'] = {str(p): sha(p) for p in receipts.glob('segment-*.jsonl')}
write_json(output/('remaining-'+runid+'.json'), [o for o in inv if o not in done])
except Exception as ex:
report.update(verdict='FAIL', completion_claim_allowed=False, error=str(ex), error_class=type(ex).__name__)
report['elapsed_seconds'] = time.monotonic()-started
report['finished_epoch'] = time.time()
report['commands'] = commands
write_json(output/('report-'+runid+'.json'), report)
print(json.dumps({k:v for k,v in report.items() if k not in ('commands','receipts')}), flush=True)
return report
if __name__ == '__main__':
a = argparse.ArgumentParser()
a.add_argument('--repo', required=True, type=Path)
a.add_argument('--inventory', required=True, type=Path)
a.add_argument('--manifest', required=True, type=Path)
a.add_argument('--output', required=True, type=Path)
a.add_argument('--seconds', type=float, default=600)
a.add_argument('--limit', type=int)
a.add_argument('--require-immutable', action='store_true')
a.add_argument('--audit', action='store_true')
x = a.parse_args()
r = run(x.repo,x.inventory,x.manifest,x.output,x.seconds,x.limit,x.require_immutable,x.audit)
raise SystemExit(0 if r['verdict']=='PASS' else 2 if r['verdict']=='PARTIAL' else 1)