#!/usr/bin/env python3 """Reproduce the v3.0 source-auditor adversarial regressions. Usage: python test_omega_auditor.py audit_omega_sources.py LEAN.docx JULIA.zip Requires Python 3.11+ and SymPy 1.14.0. Tests normal and -O modes separately. All changes are disposable; archive manifests are updated after semantic edits. The fixed original hashes identify the release from which mutations are made. """ import argparse from concurrent.futures import ThreadPoolExecutor from hashlib import sha256 import json from pathlib import Path import subprocess import sys from tempfile import TemporaryDirectory import tomllib import xml.etree.ElementTree as ET from zipfile import ZipFile, ZIP_DEFLATED ROOT='OMEGA_NAUGHT_Julia_Proof_v0.1/' def need(condition,message): if not condition: raise ValueError(message) def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('auditor',type=Path) parser.add_argument('docx',type=Path) parser.add_argument('archive',type=Path) args=parser.parse_args() auditor,docx,archive=[p.resolve() for p in (args.auditor,args.docx,args.archive)] need(sha256(docx.read_bytes()).hexdigest()=='275da4ea14130d21aa1b797b643ec78192e195780bfd8352bac699a75b73542e','original DOCX identity') need(sha256(archive.read_bytes()).hexdigest()=='ef8f1eb75b83f1e5763d280374eb8b73d9ab00b8dcb76b1824aee771869cd27a','original ZIP identity') with ZipFile(archive) as z: original={n:z.read(n) for n in z.namelist() if not n.endswith('/')} with TemporaryDirectory(prefix='omega_v3_regression_') as tmp: folder=Path(tmp) cases=[('original',docx,archive,None)] def write_zip(label,members,error,refresh=True): if refresh: members[ROOT+'SHA256SUMS']=(''.join(sha256(members[ROOT+name]).hexdigest()+' '+name+'\n' for _,name in (line.split(maxsplit=1) for line in original[ROOT+'SHA256SUMS'].decode().splitlines()))).encode() path=folder/(label+'.zip') with ZipFile(path,'w',ZIP_DEFLATED) as z: for name,data in members.items():z.writestr(name,data) cases.append((label,docx,path,error)) def change_source(label,old,new,error): members=dict(original);name=ROOT+'src/PolynomialCertificates.jl' text=members[name].decode();need(text.count(old)==1,'unique mutation '+label) members[name]=text.replace(old,new,1).encode();write_zip(label,members,error) def change_table(label,names,operation,error): members=dict(original) for opt in ('O0','O3'): for name in names: path=ROOT+'evidence/'+opt+'/'+name members[path]=operation(members[path].decode()).encode() write_zip(label,members,error) with ZipFile(docx) as z: doc={n:z.read(n) for n in z.namelist()} ns={'w':'http://schemas.openxmlformats.org/wordprocessingml/2006/main'} for label,duplicate in [('no_source_blocks',False),('duplicate_source_block',True)]: tree=ET.fromstring(doc['word/document.xml']) if duplicate: import copy match=next(p for p in tree.findall('.//w:p',ns) if p.find('./w:pPr/w:pStyle',ns) is not None and p.find('./w:pPr/w:pStyle',ns).get('{'+ns['w']+'}val')=='ArchivePath' and ''.join(t.text or '' for t in p.findall('.//w:t',ns)).startswith('OMEGA_NAUGHT_Lean_Proof_v0.1/')) tree.find('w:body',ns).append(copy.deepcopy(match)) else: for parent in tree.iter(): for el in list(parent): st=el.find('./w:pPr/w:pStyle',ns) if st is not None and st.get('{'+ns['w']+'}val') in ('ArchivePath','ArchiveText'):parent.remove(el) path=folder/(label+'.docx') with ZipFile(path,'w',ZIP_DEFLATED) as z: for name,data in doc.items():z.writestr(name,ET.tostring(tree) if name=='word/document.xml' else data) cases.append((label,path,archive,'DOCX source block inventory' if not duplicate else 'duplicate DOCX source block')) change_source('bad_moment_11','sum([(a*c)^2 for _ in heights])-6*(a*c)^2)', 'sum([(a*c)^2 for _ in heights])-6*(a*c)^2+1)','nonzero source polynomial: moment_11') change_table('empty_evidence_tables',['rational_enclosures.tsv','precision_sweep.csv','gap_blocks.csv'], lambda t:t.splitlines()[0]+'\n','table coverage/order/duplicates: rational_enclosures.tsv') for name in ('precision_sweep.csv','gap_blocks.csv'): change_table('empty_'+name.split('.')[0],[name],lambda t:t.splitlines()[0]+'\n', 'table coverage/order/duplicates: '+name) members=dict(original);path=ROOT+'THEOREMS.toml';text=members[path].decode() entries={e['name']:e for e in tomllib.loads(text)['theorem']} start=text.index('name = "Omega.G_eq"');end=text.index('[[theorem]]',start) chunk=text[start:end] old='statement = '+json.dumps(entries['Omega.G_eq']['statement'],ensure_ascii=False) new='statement = '+json.dumps(entries['Omega.C_eq']['statement'],ensure_ascii=False) need(chunk.count(old)==1,'unique theorem binding mutation') members[path]=(text[:start]+chunk.replace(old,new,1)+text[end:]).encode() write_zip('wrong_theorem_statement',members,'theorem statement/name/source binding: Omega.G_eq') change_table('false_recorded_polynomial',['polynomial_certificates.tsv'], lambda t:t.replace('phi_sq\t0\t','phi_sq\t1\t',1), 'recorded polynomial normal form: phi_sq') change_source('bad_moment_12','sum([a*c*a*zi for zi in heights]))', 'sum([a*c*a*zi for zi in heights])+1)','nonzero source polynomial: moment_12') change_source('bad_moment_22','-2*a^2*(s^2+(1+s)^2+(phi^2+s)^2))', '-2*a^2*(s^2+(1+s)^2+(phi^2+s)^2)+1)','nonzero source polynomial: moment_22') change_source('bad_gram','-a^2*(c^2+heights[i]*heights[j]))', '-a^2*(c^2+heights[i]*heights[j])+1)','nonzero source polynomial: gram_entry_1_1') change_source('bad_heights','heights=[-(phi^2+s),','heights=[1-(phi^2+s),', 'nonzero source polynomial: moment_12') change_source('bad_helper','Tk(k,x)=1+k*(x-1)','Tk(k,x)=2+k*(x-1)', 'nonzero source polynomial: displacement_conjugacy') change_table('duplicate_enclosure',['rational_enclosures.tsv'], lambda t:t+t.splitlines()[1]+'\n','table coverage/order/duplicates: rational_enclosures.tsv') members=dict(original);path=ROOT+'SHA256SUMS' members[path]='\n'.join(members[path].decode().splitlines()[1:]).encode()+b'\n' write_zip('missing_manifest_entry',members,'manifest closed inventory',refresh=False) members=dict(original);path=ROOT+'THEOREMS.toml' members[path]=members[path].replace(b'section_context = "variable {V : Type*} [AddCommGroup V] [Module \xe2\x84\x9d V]"',b'section_context = "false context"',1) need(members[path]!=original[path],'context mutation') write_zip('wrong_section_context',members,'theorem statement/name/source binding: Omega.homothety_comp') change_source('unsupported_expression','put("phi_sq",phi^2-phi-1)', 'put("phi_sq",eval(phi))','unsupported function') change_source('unsupported_source_line',' eps=pi/648000; delta=eps/2', ' unknown_dependency=1\n eps=pi/648000; delta=eps/2','unsupported polynomial source line') print('OMEGA V3 SOURCE AUDITOR REGRESSION',flush=True) print('Python',sys.version.split()[0],flush=True) print('Auditor SHA256',sha256(auditor.read_bytes()).hexdigest(),flush=True) print('Cases:',len(cases),'(one original and',len(cases)-1,'mutations); both normal and -O',flush=True) jobs=[(case,mode) for case in cases for mode in ('normal','optimized')] def run(job): (label,doc,arc,error),mode=job command=[sys.executable]+(['-O'] if mode=='optimized' else [])+[str(auditor),str(doc),str(arc)] result=subprocess.run(command,capture_output=True,text=True,timeout=180) output=result.stdout+result.stderr if error is None: need(result.returncode==0,'original rejected: '+output) else: need(result.returncode==1 and 'REJECT: '+error in output, 'wrong acceptance/rejection for '+label+' '+mode+': '+output) return label,mode,output,error results=[] with ThreadPoolExecutor(max_workers=4) as pool: for label,mode,output,error in pool.map(run,jobs): results.append((label,mode,output)) print('PASS',mode,label,':','accepted original' if error is None else 'rejected: '+error,flush=True) need(results[0][2]==results[1][2],'original output differs under -O') print('\nORIGINAL AUDIT OUTPUT\n'+results[0][2],flush=True) print('TOTAL:',len(jobs),'successful subprocess checks;',2*(len(cases)-1),'correct mutation rejections.',flush=True) print('No Lean or Julia runtime was executed.',flush=True) if __name__=='__main__':main()