#!/usr/bin/env python3 """Recover bounded Linux entity-clipping scheduling, without executing callees.""" import gzip import hashlib import itertools import json from pathlib import Path from analyze_collision_composition import evaluate, get_word, lane_sites, put_word from analyze_engine_control import control_flow from analyze_qvm import write_json from analyze_vm_extensions import native_functions, verified def sites(graph): rows=graph['instructions'] def index(op,args): found=[i for i,x in enumerate(rows) if x['opcode']==op and x['operands'].split('#')[0].strip()==args] if len(found)!=1:raise ValueError('ambiguous loop site '+args) return found[0] def addr(op,args):return rows[index(op,args)]['address'] gate=index('testl','%r15d, %r15d') if rows[gate+1]['opcode']!='jle':raise ValueError('count branch changed') exit_addr=int(rows[gate+1]['operands'].split()[0],16) query=[i for i,x in enumerate(rows) if '' in x['operands']] if len(query)!=1:raise ValueError('query call changed') def candidate_entry(args): found=[x['address'] for i,x in enumerate(rows[:-1]) if x['opcode']=='movl' and x['operands']==args and '' in rows[i+1]['operands']] if len(found)!=1:raise ValueError('ambiguous candidate entry') return found[0] entry0=candidate_entry('(%r12), %edi') entry1=candidate_entry('(%r14), %edi') entry2=candidate_entry('(%rbp), %edi') for entry in [entry0,entry1,entry2]: i=next(i for i,x in enumerate(rows) if x['address']==entry) if '' not in rows[i+1]['operands']:raise ValueError('candidate call changed') pass_read=index('movl','0x200(%rax), %edx') if '' not in rows[pass_read-1]['operands']:raise ValueError('pass lookup changed') return {'after_query':rows[query[0]+1]['address'], 'count_gate':rows[gate]['address'], 'pass_lookup_call':rows[pass_read-1]['address'], 'after_pass_lookup':addr('movl','0x200(%rax), %edx'), 'exit':exit_addr,'entries':[entry0,entry1,entry2], 'continuations':[x['stop'] for x in lane_sites(graph)]} def run(graph,start,stops,m,regs): return evaluate(graph,start,stops,m,regs,capture_registers=True) def main(): out={'native':{},'limitations':'Synthetic bounded instruction slices only. Area query, entity lookup, ' 'candidate filters and collision calls are not executed. Scheduling compositions inject status0 ' 'after each skipped candidate body. Normal count ownership/bounds, collision-produced statuses, ' 'geometry, original runtime and other platforms remain open.'} clip,stack,array,obj=0x10000,0x40000,0x40030,0x20000 for label in ['linux--Quake3-UrT-Ded.x86_64','linux--Quake3-UrT.x86_64']: r=json.loads(Path(f'docs/research/native/{label}.json').read_text()) verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(),r['source']['sha256']) a=r['artifacts']['disassembly.txt'] fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode()) g=control_flow(fs['SV_ClipMoveToEntities']);s=sites(g) entries=s['entries'];stops=entries+[s['exit']] entry=[] for count,status in itertools.product([-1,0,1,2,3,4,1023,1024,0x80000000],[0,1,4,0xffffffff]): m={};put_word(m,clip+0x3c,status) t=run(g,s['count_gate'],stops,m,{'%rbx':clip,'%rsp':stack,'%rbp':array,'%r15':count,'%r13':99}) entry.append({'count_bits':count&0xffffffff,'status0':status,'lane':entries.index(t['stop']) if t['stop'] in entries else None, 'trace':t['trace'],'stop':t['stop'],'registers':t['registers'],'writes':t['writes']}) normalization=[] for pass_id in [0,7,1022,1023,0xffffffff]: m={};put_word(m,clip+0x74,pass_id);put_word(m,stack+0x1088,3) t=run(g,s['after_query'],[s['count_gate'],s['pass_lookup_call']],m,{'%rbx':clip,'%rsp':stack}) normalization.append({'kind':'pass','input':pass_id,'lookup_requested':t['stop']==s['pass_lookup_call'], 'normalized_word':get_word(t['memory'],stack+0x2c) if t['stop']==s['count_gate'] else None, 'trace':t['trace'],'registers':t['registers']}) for related in [0,7,1022,1023,0xffffffff]: m={};put_word(m,obj+0x200,related) t=run(g,s['after_pass_lookup'],[s['count_gate']],m,{'%rax':obj,'%rsp':stack}) normalization.append({'kind':'related','input':related,'normalized_word':get_word(t['memory'],stack+0x2c), 'trace':t['trace'],'writes':t['writes']}) continuations=[] for lane,count,counter,status in itertools.product(range(3),[1,2,3,4],range(5),[0,1,4]): m={};put_word(m,clip+0x3c,status) regs={'%rbx':clip,'%rsp':stack,'%r12':array,'%r14':array+4,'%r13':counter,'%r15':count} t=run(g,s['continuations'][lane],stops,m,regs) continuations.append({'lane':lane,'count':count,'counter_before':counter,'status0':status, 'next_lane':entries.index(t['stop']) if t['stop'] in entries else None, 'counter_after':t['registers']['r13'],'registers':t['registers'],'trace':t['trace']}) schedules=[] for count in list(range(1,9))+[31,32,1023,1024]: for stop_after in sorted({-1,0,count-1}): # -1 means no supplied status stop. m={};put_word(m,clip+0x3c,0) t=run(g,s['count_gate'],stops,m,{'%rbx':clip,'%rsp':stack,'%rbp':array,'%r15':count,'%r13':99}) regs=t['registers'];m=t['memory'];pc=t['stop'];visited=[];trace=list(t['trace']);segments=[] while pc!=s['exit']: if len(visited)>=count:raise ValueError('schedule exceeded supplied count') lane=entries.index(pc);ptr=regs['r14' if lane==1 else 'r12'] visited.append({'lane':lane,'array_byte_offset':ptr-array}) # Explicitly skip candidate body/callees and supply its resulting current status0. put_word(m,clip+0x3c,int(len(visited)-1==stop_after)) t=run(g,s['continuations'][lane],stops,m,regs) regs=t['registers'];m=t['memory'];pc=t['stop'];trace.extend(t['trace']);segments.append(t['trace']) schedules.append({'count':count,'status_stop_after_index':stop_after,'visited':visited, 'final_counter':regs['r13'],'trace_instruction_count':len(trace), 'trace_sha256':hashlib.sha256(json.dumps(trace,separators=(',',':')).encode()).hexdigest(), 'trace':trace if count<=32 else None,'last_continuation_trace':segments[-1]}) out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'], 'graph':g,'sites':s,'entry_traces':entry,'normalization_traces':normalization, 'continuation_traces':continuations,'scheduling_compositions':schedules} write_json(Path('docs/research/collision-loop.json'),out) print('Recovered 72 entry, 20 normalization, 360 continuation slices and 70 scheduling compositions') if __name__=='__main__':main()