File size: 10,050 Bytes
bdf6c1f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | #!/usr/bin/env python3
"""Native-fixture differential recoding, second-object checks and modeled seed codebook."""
from collections import Counter
import gzip
import hashlib
import json
from pathlib import Path
import struct
from analyze_collision_composition import get_word,put_word
from analyze_engine_control import control_flow
from analyze_frame_timing import read_address
from analyze_huffman_updates import increment,run,state
from analyze_native import elf
from analyze_qvm import write_json
from analyze_vm_extensions import native_functions,verified
from huffman_model import Huffman
from analyze_huffman_encoding import transmit,DATA,CURSOR,TABLE,OUT
def digest(value):return hashlib.sha256(json.dumps(value,sort_keys=True,separators=(',',':')).encode()).hexdigest()
def invariants(h,counts):
s=h.state();nodes={x['address']:x for x in s['nodes']};seen=set()
def walk(addr,parent):
if addr in seen:raise ValueError('tree cycle')
seen.add(addr);n=nodes[addr]
if n['parent']!=parent:raise ValueError('parent mismatch')
if n['symbol']!=257:
expected=counts.get(n['symbol'],0)
if n['weight']!=expected or n['left'] or n['right']:raise ValueError('leaf mismatch')
return expected
weight=walk(n['left'],addr)+walk(n['right'],addr)
if n['weight']!=weight:raise ValueError('subtree mismatch')
return weight
total=walk(s['root'],0)
if total!=sum(counts.values()) or seen!=set(nodes):raise ValueError('tree coverage')
chain=[];cur=s['list_start'];prev=0
while cur:
if cur in chain or nodes[cur]['previous']!=prev:raise ValueError('rank links')
chain.append(cur);prev=cur;cur=nodes[cur]['next']
if set(chain)!=seen or [nodes[a]['weight'] for a in chain]!=sorted(nodes[a]['weight'] for a in chain):raise ValueError('rank order')
heads=dict(s['heads']);pool={0x10000+0xb030+8*i:v for i,v in enumerate(s['pointer_cells'])};free=set();cur=s['free_head']
while cur:
if cur in free or cur not in pool:raise ValueError('free chain')
free.add(cur);cur=pool[cur]
if set(heads)&free or set(heads)|free!=set(pool):raise ValueError('pool partition')
for addr in chain:
node=nodes[addr]
if node['weight'] and heads[node['head_cell']]!=next(a for a in reversed(chain) if nodes[a]['weight']==node['weight']):raise ValueError('block leader')
return {'total_weight':total,'nodes':len(nodes),'rank_nodes':len(chain),'live_cells':len(heads),'free_cells':len(free),'allocated_cells':len(pool)}
def main():
init_path=Path('docs/research/huffman-initialization.json');updates_path=Path('docs/research/huffman-updates.json');init=json.loads(init_path.read_text());updates=json.loads(updates_path.read_text());out={'native':{},'limitations':'Compact recoding is provisional. Differential checks compare complete sampled states to synthetic native instruction compositions, not original executions. Full seed training/codebook is MODEL-DERIVED, not an evaluated million-step native trace or runtime capture. No packet codec/original process/native callee executes. Branch completeness, actual codewords/runtime/ABI and other-platform conformance remain open.'}
fixtures=[[0,1]*8,list(range(16)),[255,0,128,127]*6,list(range(8))+list(reversed(range(8)))]
frequency=None;native_graphs={}
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());data=verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(),r['source']['sha256']);meta,_=elf(data);a=r['artifacts']['disassembly.txt'];fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode());graphs={n:control_flow(fs[n]) for n in ['Huff_addRef','increment','send','Huff_offsetTransmit','Huff_offsetReceive']};native_graphs[label]=graphs;p=init['native'][label];u=updates['native'][label]
if p['source']!=r['source'] or u['source']!=r['source'] or p['disassembly_sha256']!=a['uncompressed_sha256'] or u['disassembly_sha256']!=a['uncompressed_sha256']:raise ValueError('native report provenance')
raw,_=read_address(data,meta['sections'],p['frequency_table']['address'],1024);counts=list(struct.unpack('<256I',raw))
if counts!=p['frequency_table']['counts'] or hashlib.sha256(raw).hexdigest()!=p['frequency_table']['sha256']:raise ValueError('frequency mismatch')
if frequency is not None and frequency!=counts:raise ValueError('paired seeds differ')
frequency=counts;prior=[]
for case in u['sequences']:
h=Huffman();checked=0
for i,symbol in enumerate(case['symbols']):
h.add(symbol)
if case['steps']:
if h.state()!=case['steps'][i]:raise ValueError('prior prefix mismatch')
checked+=1
if h.state()!=case['final_state']:raise ValueError('prior final mismatch')
prior.append({'symbols':case['symbols'],'checked_prefixes':checked,'final_state_sha256':digest(h.state()),'events':h.events,'invariants':invariants(h,Counter(case['symbols']))})
differential=[]
for second in [False,True]:
base=p['initialization_probes'][0]['base']+int(second)*0xc830;seed=p['initialization_probes'][0]['objects'][int(second)]
for sequence in fixtures:
m={base+i:0 for i in range(0xc830)}
for offset,value in seed['nonzero_bytes']:m[base+offset]=value
m[base-1]=m[base+0xc830]=0x7b;h=Huffman(second);trace=[];checks=[]
for i,symbol in enumerate(sequence):
g=graphs['Huff_addRef'];t=run(g,g['instructions'][0]['address'],[graphs['increment']['instructions'][0]['address']],m,{'%rdi':base,'%rsi':symbol});trace+=t['trace'];c=increment(graphs['increment'],t['memory'],base,t['registers']['rsi']);trace+=c['trace'];m=c['memory'];h.add(symbol);actual=state(m,base)
if actual!=h.state(base):raise ValueError('fresh differential mismatch')
checks.append(digest(actual))
differential.append({'second_object':second,'base':base,'symbols':sequence,'matched_prefix_sha256':checks,'events':h.events,'invariants':invariants(h,Counter(sequence)),'canaries':[m[base-1],m[base+0xc830]],'final_state_sha256':digest(h.state(base)),'trace_length':len(trace),'trace_sha256':digest(trace)})
out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,'prior_cases':prior,'fresh_differential_cases':differential}
h=Huffman();checkpoints=[];trained={}
for symbol,count in enumerate(frequency):
for _ in range(count):h.add(symbol)
trained[symbol]=count;checkpoints.append({'through_symbol':symbol,'repetitions':sum(trained.values()),'state_sha256':digest(h.state()),'invariants':invariants(h,trained)})
words=h.codewords();codebook=[{'symbol':s,'bits':bits,'length':len(bits)} for s,bits in sorted(words.items())]
out['model_derived_seed_training']={'frequencies':frequency,'repetitions':sum(frequency),'checkpoints':checkpoints,'final_state':h.state(),'final_state_sha256':digest(h.state()),'codebook':codebook,'codebook_sha256':digest(codebook),'events':h.events,'invariants':invariants(h,dict(enumerate(frequency))),'origin':'Compact native-derived model; not runtime captured or full native-instruction training.'}
roundtrips=[]
for label,graphs in native_graphs.items():
for sequence,start in [(list(range(256)),0),(list(range(16)),7),([0,255,128,1]*8,0),([0,255,128,1]*8,7)]:
st=h.state(TABLE);m={TABLE+i:0 for i in range(0xc830)}
for slot,value in [(0,st['node_count']),(4,st['pointer_cell_count'])]:put_word(m,TABLE+slot,value)
for slot,value in [(8,st['root']),(16,st['list_start']),(24,st['tail_slot']),(0x828,st['free_head'])]:put_word(m,TABLE+slot,value,8)
for sym,ptr in st['lookup']:put_word(m,TABLE+32+8*sym,ptr,8)
for i,value in enumerate(st['pointer_cells']):put_word(m,TABLE+0xb030+8*i,value,8)
for node in st['nodes']:
for slot,key in [(0,'left'),(8,'right'),(16,'parent'),(24,'next'),(32,'previous'),(40,'head_cell')]:put_word(m,node['address']+slot,node[key],8)
put_word(m,node['address']+48,node['weight']);put_word(m,node['address']+52,node['symbol'])
for i in range(4096):m[DATA+i]=0
m[DATA-1]=m[DATA+4096]=0x7b;put_word(m,CURSOR,start);trace=[];end=start
for symbol in sequence:m,part,_=transmit(graphs,m,symbol,None);trace+=part
end=get_word(m,CURSOR);encoded=[m[DATA+i] for i in range((end+7)//8)];put_word(m,CURSOR,start);decoded=[];readcursors=[]
g=graphs['Huff_offsetReceive'];stops=[x['address'] for x in g['instructions'] if x['opcode']=='retq']
for _ in sequence:
t=run(g,g['instructions'][0]['address'],stops,m,{'%rdi':st['root'],'%rsi':OUT,'%rdx':DATA,'%rcx':CURSOR});m=t['memory'];trace+=t['trace'];decoded.append(get_word(m,OUT));readcursors.append(get_word(m,CURSOR))
roundtrips.append({'native_label':label,'symbols':sequence,'initial_bit':start,'encoded_bytes':encoded,'transmit_cursor':end,'decoded_symbols':decoded,'receive_cursors':readcursors,'canaries':[m[DATA-1],m[DATA+4096]],'trace_length':len(trace),'trace_sha256':digest(trace),'tree_origin':'model-derived seed-trained tree, supplied to native instruction compositions'})
out['modeled_tree_native_symbol_roundtrips']=roundtrips
out['model_source_sha256']=hashlib.sha256(Path('scripts/huffman_model.py').read_bytes()).hexdigest()
out['input_report_sha256']={str(path):hashlib.sha256(path.read_bytes()).hexdigest() for path in [init_path,updates_path]};write_json(Path('docs/research/huffman-codebook.json'),out);print('Verified prior/fresh differential states and generated provisional seed-trained model codebook')
if __name__=='__main__':main()
|