#!/usr/bin/env python3 """Offline structural/export checks. No network, wallet execution or signature engine. Usage: python verify-results.py.txt [package-directory] [--report private-report.json] """ import argparse, csv, hashlib, json, re from pathlib import Path COUNT = 0 def check(condition, message): global COUNT COUNT += 1 if not condition: raise ValueError(message) def digest(data): return hashlib.sha256(hashlib.sha256(data).digest()).digest() def compact(n): if n < 253: return bytes([n]) if n <= 0xffff: return b'\xfd' + n.to_bytes(2, 'little') if n <= 0xffffffff: return b'\xfe' + n.to_bytes(4, 'little') return b'\xff' + n.to_bytes(8, 'little') class Reader: def __init__(self, data): self.data, self.pos = data, 0 def take(self, n): check(n >= 0 and self.pos+n <= len(self.data), 'truncated transaction') b=self.data[self.pos:self.pos+n]; self.pos += n; return b def num(self, n): return int.from_bytes(self.take(n), 'little') def varint(self): n=self.num(1) if n < 253: return n width={253:2,254:4,255:8}[n]; value=self.num(width) check(value >= {253:253,254:65536,255:4294967296}[n], 'noncanonical CompactSize') check(value <= 1000000, 'unexpectedly large CompactSize') return value def blob(self): return self.take(self.varint()) def parse_tx(raw_hex): raw=bytes.fromhex(raw_hex); r=Reader(raw) v=r.take(4); n=r.varint(); witness=False if n == 0: flag=r.num(1) check(flag == 1, 'unsupported transaction flags (MWEB outside scope)') witness=True; n=r.varint() check(n > 0, 'no transparent inputs') vin=[]; vin_raw=[] for _ in range(n): start=r.pos; prev=r.take(32)[::-1].hex(); index=r.num(4) script=r.blob(); sequence=r.num(4) vin.append({'txid':prev,'vout':index,'script_sig':script.hex(),'sequence':sequence}) vin_raw.append(raw[start:r.pos]) m=r.varint(); outputs=[]; out_raw=[] for _ in range(m): start=r.pos; value=r.num(8); script=r.blob() outputs.append({'value_litoshis':value,'script_pub_key':script.hex()}) out_raw.append(raw[start:r.pos]) if witness: for item in vin: item['witness']=[r.blob().hex() for _ in range(r.varint())] lock=r.take(4) check(r.pos == len(raw), 'trailing transaction data') stripped=v+compact(n)+b''.join(vin_raw)+compact(m)+b''.join(out_raw)+lock return {'txid':digest(stripped)[::-1].hex(), 'wtxid':digest(raw)[::-1].hex(), 'version':int.from_bytes(v,'little',signed=True),'locktime':int.from_bytes(lock,'little'), 'inputs':vin,'outputs':outputs,'size':len(raw), 'weight':len(stripped)*3+len(raw),'vsize':(len(stripped)*3+len(raw)+3)//4} def script_pushes(script): r=Reader(bytes.fromhex(script)); stack=[] while r.pos < len(r.data): op=r.num(1) if op <= 75: stack.append(r.take(op)) elif op == 76: stack.append(r.take(r.num(1))) elif op == 77: stack.append(r.take(r.num(2))) elif op == 78: stack.append(r.take(r.num(4))) else: stack.append(None) return stack def csv_operand(script): raw=bytes.fromhex(script); r=Reader(raw); op=r.num(1) if 81 <= op <= 96: operand=op-80 elif op == 0: operand=0 elif op <= 75: b=r.take(op); operand=int.from_bytes(b,'little') if b and b[-1] & 128: operand=-(operand & ~(128 << (8*(len(b)-1)))) else: raise ValueError('unsupported CSV operand encoding') check(r.take(3) == b'\xb2\x75\x21', 'expected CSV DROP and compressed public-key push') pub=r.take(33) check(pub[0] in (2,3), 'expected compressed public key') check(r.take(1) == b'\xac' and r.pos == len(raw), 'unexpected CSV script shape') return operand def csv_ok(version, sequence, operand): if operand is None: return True if operand < 0: return False if operand & 0x80000000: return True if (version & 0xffffffff) < 2 or sequence & 0x80000000: return False mask=0x400000 | 0xffff return bool(sequence & 0x400000) == bool(operand & 0x400000) and (sequence & mask) >= (operand & mask) def admission(value): if isinstance(value,str): value=json.loads(value) if isinstance(value,list): check(len(value)==1,'expected single-candidate admission result'); value=value[0] if isinstance(value,dict) and 'stdout' in value: return admission(value['stdout']) check(isinstance(value,dict) and isinstance(value.get('allowed'),bool), 'missing admission verdict') return value def objects(value): if isinstance(value,dict): yield value for child in value.values(): yield from objects(child) elif isinstance(value,list): for child in value: yield from objects(child) def collect_blocks(value): records={} for item in objects(value): if isinstance(item.get('height'),int) and isinstance(item.get('time'),int): h=item['height']; t=item['time'] if h in records: check(records[h]['time']==t, 'inconsistent block timestamp at same height') records[h]=item return records def mtp(blocks, height): heights=list(range(max(0,height-10),height+1)) if all(h in blocks for h in heights): times=sorted(blocks[h]['time'] for h in heights) return times[len(times)//2] return None def flag_privacy(value): expanded=list(objects(value)) for item in list(expanded): if isinstance(item.get('stdout'),str): try: expanded.extend(objects(json.loads(item['stdout']))) except (ValueError,TypeError): pass for item in expanded: for key,val in item.items(): if key.lower() in ('rpcpassword','private_key','privkey','seed','mnemonic','cookie') and val: check(isinstance(val,str) and any(s in val.lower() for s in ('redact','placeholder','generated','not stored')), 'possible secret field: '+key) if item.get('method')=='dumpprivkey': check('redact' in str(item.get('stdout','')).lower(), 'unredacted dumpprivkey output') text=json.dumps(value) check(not re.search(r'-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY',text), 'private-key PEM marker') check(not re.search(r'\b(?:xprv|tprv|Ltpv|ltpv)[1-9A-HJ-NP-Za-km-z]{40,}',text), 'extended private-key marker') alphabet='123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz' for token in re.findall(r'(?= sum(x['value_litoshis'] for x in t['outputs']),cid+': negative transparent fee') redeem=c.get('redeem_script') operands[cid]=csv_operand(redeem) if redeem else None if redeem: check(operands[cid]==5,cid+': paper scope requires mandatory CSV operand5') expected='a914'+hashlib.new('ripemd160',hashlib.sha256(bytes.fromhex(redeem)).digest()).hexdigest()+'87' check(expected==p['script_pub_key'],cid+': redeem script does not match funding P2SH') check(script_pushes(inp['script_sig'])[-1]==bytes.fromhex(redeem),cid+': scriptSig does not carry recorded redeem script') blocks=collect_blocks(results.get('blocks',[])); time_cases=0 byid={} for case in cases: name=case['id']; cid=case['candidate']; c=candidates[cid]; t=parsed[cid] check(name not in byid,'duplicate case ID'); byid[name]=case tip=case['tip_height']; candidate=case['candidate_height']; h=case.get('parent_height') check(candidate==tip+1,name+': candidate height must be next block') confirmations=case['parent_confirmations'] check(confirmations==(tip-h+1 if h is not None else 0),name+': parent confirmation equation') tip_mtp=mtp(blocks,tip) if tip_mtp is not None: check(tip_mtp==case['median_time_past'],name+': tip MTP reconstruction') seq=c['sequence']; age=True if (t['version'] & 0xffffffff)>=2 and not seq & 0x80000000: count=seq & 0xffff if seq & 0x400000: time_cases += 1 origin=case['median_time_past'] if h is None else mtp(blocks,max(h-1,0)) if origin is None: origin=case.get('parent_previous_median_time_past',c['point'].get('parent_previous_median_time_past')) check(isinstance(origin,int),name+': missing source for relative-time origin MTP') age=case['median_time_past'] > origin+(count<<9)-1 else: parent_h=candidate if h is None else h age=candidate > parent_h+count-1 guard=csv_ok(t['version'],seq,operands[cid]) model=age and guard verdict=admission(case['admission']) check(verdict['allowed']==case['expected_allowed'],name+': observed/expected mismatch') check(verdict['allowed']==model,name+': relative-age/CSV model disagrees') if 'txid' in verdict: check(verdict['txid']==t['txid'],name+': admission txid mismatch') with (folder/'cases.csv').open(encoding='utf-8',newline='') as stream: rows=list(csv.DictReader(stream)) check(len(rows)==len(cases),'CSV case count mismatch') for row in rows: case=byid[row['case']]; c=candidates[case['candidate']] for field,expected in [('candidate',case['candidate']),('version',c['version']),('sequence',c['sequence']),('tip_height',case['tip_height']),('candidate_height',case['candidate_height']),('parent_confirmations',case['parent_confirmations']),('median_time_past',case['median_time_past'])]: check(str(row[field])==str(expected),row['case']+': CSV '+field+' mismatch') check(row['allowed'].lower()==str(admission(case['admission'])['allowed']).lower(),row['case']+': CSV verdict mismatch') for field in ('parent_height','mock_time'): expected=case[field] check(row[field]==('' if expected is None else str(expected)),row['case']+': CSV '+field+' mismatch') check(row['reject_reason']==admission(case['admission']).get('reject-reason',''),row['case']+': CSV rejection text mismatch') for item in objects(trace): if item.get('method')=='decoderawtransaction' and item.get('exitCode')==0: try: decoded=json.loads(item['stdout']) except (ValueError,TypeError): continue if isinstance(decoded,dict) and decoded.get('txid') in {t['txid'] for t in parsed.values()}: t=next(t for t in parsed.values() if t['txid']==decoded['txid']) check(decoded['version']==t['version'],'RPC decode version mismatch') check(decoded['locktime']==t['locktime'],'RPC decode locktime mismatch') check(decoded.get('hash',t['wtxid'])==t['wtxid'],'RPC decode wtxid mismatch') bindings=bind_exports(results,trace,folder,parsed,blocks,candidates,byid) flag_privacy(results); flag_privacy(trace) checksum_count=0 if not skip_checksums: for line in (folder/'SHA256SUMS.txt').read_text(encoding='utf-8').splitlines(): if not line.strip(): continue expected,name=line.split(None,1); name=name.lstrip('*') path=Path(name) check(not path.is_absolute() and '..' not in path.parts,'unsafe checksum path') check(hashlib.sha256((folder/path).read_bytes()).hexdigest()==expected,'checksum mismatch: '+name) checksum_count += 1 check(checksum_count>0,'empty checksum list') return {'status':'passed','checks':COUNT,'candidate_count':len(candidates),'case_count':len(cases), 'relative_time_cases':time_cases,'reconstructed_block_records':len(blocks),'checksum_files':checksum_count,'bindings':bindings, 'limits':['No cryptographic signature or public-key validity verification.','No independent Core execution or binary provenance attestation.','No mainnet inference, timing guarantee, wallet compatibility or production-contract audit.','Privacy checks are structural and not proof that every possible secret encoding is absent.']} if __name__=='__main__': ap=argparse.ArgumentParser(description=__doc__) ap.add_argument('directory',nargs='?',default=str(Path(__file__).resolve().parent)) ap.add_argument('--report',type=Path) ap.add_argument('--skip-checksums',action='store_true',help='development-only: report excludes package checksums') args=ap.parse_args() try: report=verify(Path(args.directory),args.skip_checksums) except Exception as exc: report={'status':'failed','checks':COUNT,'error':str(exc)} if args.report: args.report.write_text(json.dumps(report,indent=2)+'\n',encoding='utf-8') print(json.dumps(report,indent=2)) raise SystemExit(0 if report['status']=='passed' else 1)