Files
verstack/scripts/validate_dmd_native.py

58 lines
4.3 KiB
Python

#!/usr/bin/env python3
"""Compare indexed bitmap decoding to named firmware routines in isolated Unicorn.
No imported firmware is executed as a host program. Requires the decoder venv.
"""
import argparse,hashlib,json,mmap,pathlib,struct,sys,collections
sys.path.insert(0,str(pathlib.Path(__file__).resolve().parents[1]/'plugins'))
import dmd_bitmap
from elftools.elf.elffile import ELFFile
from unicorn import Uc,UC_ARCH_ARM,UC_MODE_ARM,UC_HOOK_CODE
from unicorn.arm_const import UC_ARM_REG_R0,UC_ARM_REG_R1,UC_ARM_REG_R2,UC_ARM_REG_SP,UC_ARM_REG_LR,UC_ARM_REG_PC
p=argparse.ArgumentParser();p.add_argument('game',type=pathlib.Path);p.add_argument('facts',type=pathlib.Path);p.add_argument('image',type=pathlib.Path);p.add_argument('receipt',type=pathlib.Path);p.add_argument('--limit-per-mode',type=int,default=0);a=p.parse_args()
facts=json.loads(a.facts.read_text());binary_hash=hashlib.sha256(a.game.read_bytes()).hexdigest()
assert facts['input_sha256']==binary_hash
names={f['name']:int(f['address'],16) for f in facts['functions']}
functions={1:'image_alloc_and_decompress_column_whitespace',7:'image_alloc_and_decompress_row_whitespace',12:'image_alloc_and_decompress_row_16',3:'image_get_keyframe_and_decompress_column_delta',9:'image_get_keyframe_and_decompress_row_delta'}
mu=Uc(UC_ARCH_ARM,UC_MODE_ARM)
with a.game.open('rb') as stream:
elf=ELFFile(stream);assert elf.elfclass==32 and elf['e_machine']=='EM_ARM';segments=[s for s in elf.iter_segments() if s['p_type']=='PT_LOAD']
lo=min(s['p_vaddr'] for s in segments)&~4095;hi=(max(s['p_vaddr']+s['p_memsz'] for s in segments)+4095)&~4095;assert hi-lo<128*1024**2
mu.mem_map(lo,hi-lo)
for s in segments:mu.mem_write(s['p_vaddr'],s.data())
MEM=0x30000000;INPUT=MEM;OUTPUT=MEM+0x100000;BASE=MEM+0x200000;NODE=MEM+0x300000;STACK=MEM+0x7ff000;STOP=0x40000000
mu.mem_map(MEM,8*1024**2);mu.mem_map(STOP,4096)
def hook(mu,pc,size,user):
if pc==names['sys_image_alloc_pdi']:
length=mu.reg_read(UC_ARM_REG_R0);assert 0<length<=1024**2
mu.mem_write(OUTPUT,bytes(length));mu.reg_write(UC_ARM_REG_R0,OUTPUT);mu.reg_write(UC_ARM_REG_PC,mu.reg_read(UC_ARM_REG_LR))
elif pc==names['memcpy']:
dst=mu.reg_read(UC_ARM_REG_R0);src=mu.reg_read(UC_ARM_REG_R1);length=mu.reg_read(UC_ARM_REG_R2);assert length<=1024**2
mu.mem_write(dst,bytes(mu.mem_read(src,length)));mu.reg_write(UC_ARM_REG_PC,mu.reg_read(UC_ARM_REG_LR))
mu.hook_add(UC_HOOK_CODE,hook)
frames={};checked=[];modes=collections.Counter();parsed=0
with a.image.open('rb') as stream,mmap.mmap(stream.fileno(),0,access=mmap.ACCESS_READ) as data:
h=struct.unpack_from('<7Q',data);count=(h[2]-h[3])//8;assert 0<count<100000
pointers=struct.unpack_from(f'<{count}Q',data,h[3]);ordered=sorted(set(pointers));ends=dict(zip(ordered,ordered[1:]+[len(data)]))
for offset in pointers:
identity=struct.unpack_from('<I',data,offset)[0];prior=frames.get((identity-1)&65535) or frames.get(identity&65535)
decoded=dmd_bitmap.decode(memoryview(data)[offset:ends[offset]],frames);mode=decoded['mode'];parsed+=1
if mode not in functions or (a.limit_per_mode and modes[mode]>=a.limit_per_mode):continue
modes[mode]+=1
record=data[offset:offset+decoded['consumed']]
mu.mem_write(INPUT,record[:decoded['consumed']])
if mode in (3,9):
assert prior
base=struct.pack('<IIHHB',(identity-1)&65535,4,prior[0],prior[1],0)+prior[2]
mu.mem_write(BASE,base);mu.mem_write(NODE,struct.pack('<III',BASE,0,0));mu.reg_write(UC_ARM_REG_R0,NODE);mu.reg_write(UC_ARM_REG_R1,INPUT)
else:mu.reg_write(UC_ARM_REG_R0,INPUT)
mu.reg_write(UC_ARM_REG_SP,STACK);mu.reg_write(UC_ARM_REG_LR,STOP)
mu.emu_start(names[functions[mode]],STOP,count=1000000)
assert mu.reg_read(UC_ARM_REG_PC)==STOP,'firmware instruction bound exceeded'
result=mu.reg_read(UC_ARM_REG_R0);pixels=bytes(mu.mem_read(result+13,len(decoded['pixels'])))
assert pixels==decoded['pixels'],f'pixel mismatch in record {identity}'
checked.append({'record_id':identity,'offset':offset,'format':decoded['format'],'pixel_sha256':hashlib.sha256(pixels).hexdigest()})
with a.image.open('rb') as stream:image_hash=hashlib.file_digest(stream,'sha256').hexdigest()
a.receipt.write_text(json.dumps({'schema':1,'game_sha256':binary_hash,'image_sha256':image_hash,'parsed_records':parsed,'checked':checked,'all_checked_pixels_match_firmware':True},indent=2))
print(f'{len(checked)} compressed records match firmware pixel-for-pixel')