#!/usr/bin/env python3 """Independent verifier for loopgrid.synthetic.v1 ONLY. Not the production LoopGrid verifier.""" import argparse, base64, hashlib, json, sys, zipfile from pathlib import Path from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.exceptions import InvalidSignature SCHEMA = 'loopgrid.synthetic.v1' ZERO = '0' * 64 MAX_SAFE = 2**53 - 1 MAX_BYTES = 2_000_000 def pairs(items): result = {} for key, value in items: if key in result: raise ValueError('Duplicate JSON key: '+key) result[key] = value return result def parse(data): def reject(value): raise ValueError('Floating-point numbers are not supported') return json.loads(data, object_pairs_hook=pairs, parse_float=reject, parse_constant=reject) def canonical(value): def check(v): if v is None or type(v) in (str, bool): return if type(v) == int and abs(v) <= MAX_SAFE: return if type(v) == list: for x in v: check(x) return if type(v) == dict: for k, x in v.items(): if not isinstance(k, str) or not k.isascii(): raise ValueError('Canonical object keys must be ASCII') check(x) return raise ValueError('Unsupported canonical JSON value') check(value) return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=True, allow_nan=False).encode('ascii') def sha(data): return hashlib.sha256(data).hexdigest() def decode64(value): return base64.b64decode(value, validate=True) def content(event): return {k:v for k,v in event.items() if k not in ('content_hash','chain_hash','signature')} def verify(bundle, trusted_pin): errors = [] def fail(code, detail): errors.append({'code':code,'detail':detail}) if not isinstance(bundle, dict) or bundle.get('format') != SCHEMA: raise ValueError('Unsupported evidence format') manifest = bundle['manifest']; events = bundle['events'] cp = bundle['checkpoint']['payload']; cp_sig = bundle['checkpoint']['signature'] if not isinstance(events,list) or not 1 <= len(events) <= 64: raise ValueError('Event count outside supported range') der = decode64(bundle['public_key_spki_base64']) key = serialization.load_der_public_key(der) if not isinstance(key,Ed25519PublicKey): raise ValueError('Expected Ed25519 public key') fp = 'sha256:'+sha(der) trusted_pin = trusted_pin.strip().lower() if not trusted_pin.startswith('sha256:') or len(trusted_pin)!=71: raise ValueError('Supply a SHA-256 SPKI fingerprint as an independent trust pin') if fp != trusted_pin: fail('SIGNER_PIN_MISMATCH','Public key does not match the trusted fingerprint') if manifest.get('schema_version') != SCHEMA or manifest.get('canonicalization') != 'loopgrid-synthetic-json-v1' or manifest.get('hash_algorithm') != 'SHA-256' or manifest.get('signature_algorithm') != 'Ed25519' or manifest.get('genesis_hash') != ZERO: fail('MANIFEST_INVALID','Unsupported manifest algorithms or genesis') workspace=manifest.get('workspace_id');decision=manifest.get('decision_id') if not isinstance(workspace,str) or not isinstance(decision,str):raise ValueError('Invalid decision identity') if manifest.get('signer_fingerprint')!=fp:fail('SIGNER_ID_MISMATCH','Manifest signer differs from public key') if manifest.get('event_count')!=len(events):fail('EVENT_COUNT_MISMATCH','Manifest event count differs from supplied events') prev=ZERO for i,event in enumerate(events,1): label='event '+str(i) if event.get('schema_version')!=SCHEMA or event.get('workspace_id')!=workspace or event.get('decision_id')!=decision: fail('EVENT_IDENTITY_MISMATCH',label+' has inconsistent identity') if event.get('signer_fingerprint')!=fp:fail('SIGNER_ID_MISMATCH',label+' signer differs from public key') if event.get('sequence')!=i or event.get('event_id')!=f'evt_{i:03d}':fail('SEQUENCE_MISMATCH',label+' is not in the expected sequence') if event.get('previous_hash')!=prev:fail('CHAIN_LINK_MISMATCH',label+' previous hash is not the expected link') ch=sha(canonical(content(event))) if event.get('content_hash')!=ch:fail('CONTENT_HASH_MISMATCH',label+' protected content was changed') chain=sha(canonical({'domain':'loopgrid.synthetic.chain.v1','previous_hash':event.get('previous_hash'),'content_hash':ch})) if event.get('chain_hash')!=chain:fail('CHAIN_HASH_MISMATCH',label+' chain commitment differs') try: key.verify(decode64(event['signature']),('loopgrid.synthetic.event.v1\n'+chain).encode('ascii')) except (InvalidSignature,ValueError):fail('SIGNATURE_INVALID',label+' Ed25519 signature does not verify') prev=chain expected={'schema_version':SCHEMA,'workspace_id':workspace,'decision_id':decision,'signer_fingerprint':fp,'event_count':len(events),'final_chain_hash':prev} if cp!=expected:fail('CHECKPOINT_MISMATCH','Signed checkpoint does not match the reconstructed complete history') cp_hash=sha(canonical(cp)) try: key.verify(decode64(cp_sig),('loopgrid.synthetic.checkpoint.v1\n'+cp_hash).encode('ascii')) except (InvalidSignature,ValueError):fail('CHECKPOINT_SIGNATURE_INVALID','Checkpoint signature does not verify') return {'valid':not errors,'errors':errors,'event_count':len(events),'signer_fingerprint':fp,'final_chain_hash':prev,'format':SCHEMA} def load(path): if Path(path).stat().st_size>MAX_BYTES:raise ValueError('Evidence exceeds 2 MB limit') if zipfile.is_zipfile(path): with zipfile.ZipFile(path) as z: infos=z.infolist(); names=[i.filename for i in infos] if len(infos)>12 or len(names)!=len(set(names)): raise ValueError('Duplicate or excessive ZIP entries') required={'bundle.json','manifest.json','events.jsonl','checkpoint.json','public-key.pem'} if not required.issubset(set(names)): raise ValueError('Synthetic ZIP is missing required evidence files') if any(i.file_size>MAX_BYTES for i in infos) or sum(i.file_size for i in infos)>8_000_000: raise ValueError('Uncompressed ZIP contents exceed reference limits') bundle=parse(z.read('bundle.json')) # The split files are inspection conveniences, but they must agree # with the authoritative signed bundle. Otherwise the ZIP cannot # be presented as a consistently verified evidence package. if canonical(parse(z.read('manifest.json')))!=canonical(bundle['manifest']): raise ValueError('ZIP manifest.json differs from bundle.json') if canonical(parse(z.read('checkpoint.json')))!=canonical(bundle['checkpoint']): raise ValueError('ZIP checkpoint.json differs from bundle.json') lines=[line for line in z.read('events.jsonl').splitlines() if line.strip()] if canonical([parse(line) for line in lines])!=canonical(bundle['events']): raise ValueError('ZIP events.jsonl differs from bundle.json') key=serialization.load_pem_public_key(z.read('public-key.pem')) der=key.public_bytes(serialization.Encoding.DER,serialization.PublicFormat.SubjectPublicKeyInfo) if der!=decode64(bundle['public_key_spki_base64']): raise ValueError('ZIP public-key.pem differs from bundle.json') return bundle return parse(Path(path).read_bytes()) def main(): p=argparse.ArgumentParser(description=__doc__) p.add_argument('evidence',help='Synthetic JSON bundle or ZIP containing bundle.json') group=p.add_mutually_exclusive_group(required=True) group.add_argument('--pin',help='Independently obtained SHA-256 SPKI fingerprint') group.add_argument('--pin-file',help='File containing an independently obtained fingerprint') p.add_argument('--json',action='store_true',help='Machine-readable output') args=p.parse_args() try: pin=Path(args.pin_file).read_text().strip() if args.pin_file else args.pin result=verify(load(args.evidence),pin) except Exception as exc: result={'valid':False,'errors':[{'code':'INPUT_ERROR','detail':str(exc)}]} if args.json:print(json.dumps(result,indent=2)) else: print('LOOPGRID SYNTHETIC REFERENCE VERIFIER') print('VERIFIED' if result['valid'] else 'INVALID') for e in result['errors']:print(e['code']+': '+e['detail']) if result['valid']:print('Events:',result['event_count'],'\nFinal hash:',result['final_chain_hash']) return 0 if result['valid'] else 1 if __name__=='__main__':sys.exit(main())