XFEEDToolBox

"""Offline PE compatibility evidence. No attachment, execution or memory writes."""
import argparse, bisect, collections, hashlib, json, re, struct
from pathlib import Path
import pefile
from capstone import Cs, CS_ARCH_X86, CS_MODE_64, CS_GRP_JUMP, CS_GRP_CALL
from capstone.x86 import X86_OP_MEM, X86_REG_RIP

p=argparse.ArgumentParser(description=__doc__)
p.add_argument('old',type=Path);p.add_argument('new',type=Path);p.add_argument('--output',type=Path,required=True)
a=p.parse_args()
md=Cs(CS_ARCH_X86,CS_MODE_64);md.detail=True

class Image:
    def __init__(self,path):
        self.path=path;self.pe=pefile.PE(str(path),fast_load=True)
        self.pe.parse_data_directories(directories=[pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_EXCEPTION']])
        self.base=self.pe.OPTIONAL_HEADER.ImageBase
        self.rows=sorted((x.struct.BeginAddress,x.struct.EndAddress) for x in self.pe.DIRECTORY_ENTRY_EXCEPTION)
        self.starts=[x[0] for x in self.rows];self.cache={}
    def data(self,rva,n):return self.pe.get_data(rva,n)
    def row(self,rva):
        i=bisect.bisect_right(self.starts,rva)-1
        if i>=0 and self.rows[i][0]<=rva<self.rows[i][1]:return self.rows[i]
        data=self.data(rva,512);used=0
        for ins in md.disasm(data,rva):
            used+=ins.size
            if ins.mnemonic in ('ret','jmp','int3'):break
        return rva,rva+used
    def norm(self,rva,end=None):
        key=(rva,end)
        if key in self.cache:return self.cache[key]
        if end is None:rva,end=self.row(rva)
        data=self.data(rva,end-rva);out=bytearray(data);refs=[];used=0
        for ins in md.disasm(data,rva):
            offset=ins.address-rva;used+=ins.size
            for op in ins.operands:
                if op.type==X86_OP_MEM and op.mem.base==X86_REG_RIP:
                    at=offset+ins.disp_offset;n=ins.disp_size
                    out[at:at+n]=b'\0'*n;refs.append((offset,ins.address+ins.size+op.mem.disp,'rip'))
                elif op.type==X86_OP_MEM and 0x100000<=op.mem.disp<self.pe.OPTIONAL_HEADER.SizeOfImage and ins.disp_size==4:
                    at=offset+ins.disp_offset;out[at:at+4]=b'\0'*4
                    refs.append((offset,op.mem.disp,'indexedImage'))
            if (ins.group(CS_GRP_JUMP) or ins.group(CS_GRP_CALL)) and ins.imm_size:
                at=offset+ins.imm_offset;n=ins.imm_size
                target=ins.operands[0].imm
                out[at:at+n]=b'\0'*n;refs.append((offset,target,'branch'))
        result=(bytes(out),refs,used==len(data));self.cache[key]=result;return result

old=Image(a.old);new=Image(a.new)
sources=list(Path('native/ManagedPlugin').glob('*.cs'))+list(Path('XFEEDToolBox.Core/Native').glob('*.cs'))
sources=[f for f in sources if 'Tests' not in f.name]
targets=collections.defaultdict(list)
for f in sources:
    for m in re.finditer(r'0x[0-9a-fA-F]+',f.read_text(encoding='utf-8-sig')):
        n=int(m[0],16)
        if 0x100000<=n<old.pe.OPTIONAL_HEADER.SizeOfImage:targets[n].append(str(f))

mapping={};evidence={};conflicts=[];ambiguous={}
def add(x,y,why):
    if x in mapping:
        if mapping[x]!=y:conflicts.append([hex(x),hex(mapping[x]),hex(y),why])
        return False
    mapping[x]=y;evidence[x]=why;return True

textold=old.pe.sections[0];textnew=new.pe.sections[0]
newbytes=textnew.get_data();textstart=textnew.VirtualAddress
def match_function(x):
    begin,end=old.row(x);norm,refs,valid=old.norm(begin,end)
    if not valid or len(norm)==0:return []
    # Find the longest unrelocated prefix anchor; then verify the entire normalized function.
    raw=old.data(begin,end-begin);chunks=[];start=0
    for i,(b,c) in enumerate(zip(raw,norm)):
        if b!=c:
            if i-start>=4:chunks.append((start,raw[start:i]))
            start=i+1
    if len(raw)-start>=4:chunks.append((start,raw[start:]))
    if not chunks:return []
    offset,needle=max(chunks,key=lambda v:len(v[1]));needle=needle[:96]
    found=[];at=max(0,begin-textstart-0x100000)
    limit=min(len(newbytes),begin-textstart+0x100000+len(raw))
    while True:
        at=newbytes.find(needle,at,limit)
        if at<0:break
        candidate=textstart+at-offset;at+=1
        if candidate<0:continue
        candidate_row=new.row(candidate)
        if candidate_row[0]!=candidate or candidate_row[1]-candidate!=len(raw):continue
        candidate_norm,_,valid=new.norm(candidate,candidate+len(raw))
        if valid and candidate_norm==norm:found.append(candidate+(x-begin))
    return sorted(set(found))

for x in targets:
    if x>=0x4DB6000:continue
    hits=match_function(x)
    if len(hits)==1:add(x,hits[0],'unique full normalized function')
    else:ambiguous[x]=hits

# Resolve globals through original RIP references, accepting only whole matching functions.
oldtext=textold.get_data();referenced_functions=set()
pattern=rb'[\x40-\x4f]?(?:[\x8b\x8d\x89\x3b\x39\x85\x83\x80\xc7\xff]|\x0f[\xb6\xb7\xbe\xbf\x10\x11\x28\x29\x2e\x2f])[\x05\x0d\x15\x1d\x25\x2d\x35\x3d]....'
for m in re.finditer(pattern,oldtext,re.S):
    rva=textold.VirtualAddress+m.start();instructions=list(md.disasm(old.data(rva,15),rva,count=1))
    if not instructions:continue
    ins=instructions[0]
    if not any(op.type==X86_OP_MEM and op.mem.base==X86_REG_RIP and ins.address+ins.size+op.mem.disp in targets for op in ins.operands):continue
    begin,end=old.row(rva)
    if begin==rva and end-begin<=0:continue
    if begin in referenced_functions:continue
    referenced_functions.add(begin)
    hits=match_function(begin)
    if len(hits)==1:add(begin,hits[0],'unique full function containing original RIP reference')
for m in re.finditer(rb'[\xe8\xe9]....',oldtext,re.S):
    at=textold.VirtualAddress+m.start();target=at+5+struct.unpack_from('<i',m[0],1)[0]
    if target not in targets or target in mapping:continue
    begin,end=old.row(at)
    if begin in referenced_functions:continue
    referenced_functions.add(begin);hits=match_function(begin)
    if len(hits)==1:add(begin,hits[0],'unique full original caller function')

# Mapped instruction references also verify static global addresses and leaf functions.
queue=list(mapping);done=set()
while queue:
    x=queue.pop()
    if x in done or x>=0x4DB6000:continue
    done.add(x);y=mapping[x];ob,oe=old.row(x);nb,ne=new.row(y)
    on,orr,ok=old.norm(ob,oe);nn,nrr,nok=new.norm(nb,ne)
    if not(ok and nok and on==nn):continue
    for (off,ot,kind),(nf,nt,nkind) in zip(orr,nrr):
        if off!=nf or kind!=nkind or ot not in targets:continue
        if add(ot,nt,f'{kind} reference in mapped function {ob:#x} at +{off:#x}'):queue.append(ot)

# Vtables must match at least eight consecutive entries, after applying verified code mappings.
for x in targets:
    if x<0x4DB6000 or x>=0x5E66000 or x in mapping:continue
    oldvalues=struct.unpack('<16Q',old.data(x,128));anchors=[]
    for i,value in enumerate(oldvalues):
        rva=value-old.base
        if rva not in mapping and 0x1000<=rva<0x4DB6000:
            hits=match_function(rva)
            if len(hits)==1:add(rva,hits[0],'unique vtable entry full normalized function')
        if rva in mapping:anchors.append((i,new.base+mapping[rva]))
    if len(anchors)<2:continue
    i,value=anchors[0];rdata=new.pe.sections[1];rd=rdata.get_data();at=0;hits=[]
    while True:
        at=rd.find(struct.pack('<Q',value),at)
        if at<0:break
        candidate=rdata.VirtualAddress+at-i*8;at+=1
        if candidate<rdata.VirtualAddress:continue
        if all(struct.unpack('<Q',new.data(candidate+j*8,8))[0]==v for j,v in anchors):hits.append(candidate)
    if len(hits)==1:add(x,hits[0],f'unique vtable match using {len(anchors)} independently mapped entries')

# Short identical leaf routines are distinguished by their original vtable neighbourhood.
rdold=old.pe.sections[1].get_data();rdnew=new.pe.sections[1].get_data()
for x in list(targets):
    if x in mapping or x>=0x4db6000:continue
    pointer=struct.pack('<Q',old.base+x);at=0;candidate_values=set()
    while True:
        at=rdold.find(pointer,at)
        if at<0:break
        here=at;at+=1;anchors=[]
        for off in range(-64,65,8):
            if off==0 or here+off<0 or here+off+8>len(rdold):continue
            val=struct.unpack_from('<Q',rdold,here+off)[0]-old.base
            if val in mapping:anchors.append((off,new.base+mapping[val]))
        if len(anchors)<3:continue
        off,val=anchors[0];pos=0
        while True:
            pos=rdnew.find(struct.pack('<Q',val),pos)
            if pos<0:break
            origin=pos-off;pos+=1
            if origin<64 or origin+72>len(rdnew):continue
            if all(struct.unpack_from('<Q',rdnew,origin+o)[0]==v for o,v in anchors):
                candidate_values.add(struct.unpack_from('<Q',rdnew,origin)[0]-new.base)
    if len(candidate_values)==1:
        y=next(iter(candidate_values));ob,oe=old.row(x);nb,ne=new.row(y)
        on,_,ok=old.norm(ob,oe);nn,_,nok=new.norm(nb,ne)
        if ok and nok and on==nn:add(x,y,'original vtable neighbourhood plus identical complete normalized leaf')
        else:ambiguous[x]=[y]
for table,offset,target in [(0x5165e48,0x30,0xeacb30)]:
    if table not in mapping:continue
    assert struct.unpack('<Q',old.data(table+offset,8))[0]==old.base+target
    y=struct.unpack('<Q',new.data(mapping[table]+offset,8))[0]-new.base
    if old.norm(target)[0]==new.norm(y)[0]:add(target,y,'verified mapped owning vtable slot and full normalized function')
if 0x8ae6f0 not in mapping:
    ins=list(md.disasm(old.data(0x8ae6f0,12),0x8ae6f0))[0]
    global_id=ins.address+ins.size+ins.operands[1].mem.disp
    if global_id in mapping:
        hits=[]
        for y in ambiguous.get(0x8ae6f0,[]):
            newins=list(md.disasm(new.data(y,12),y))[0]
            if newins.address+newins.size+newins.operands[1].mem.disp==mapping[global_id]:hits.append(y)
        if len(hits)==1:add(0x8ae6f0,hits[0],'identical complete leaf and independently verified component identifier global')
if 0x8ae6f0 in mapping:
    add(0x8ae6f6,mapping[0x8ae6f0]+6,'end of verified getter RIP-relative instruction')
if 0x5fdf490 in mapping and 0x5fdf494 in targets:
    add(0x5fdf494,mapping[0x5fdf490]+4,'second uint32 of the verified packed stellar identifier pair')
# The radius getter includes a two-entry image-relative jump table after its RET.
# Verify all code and both table targets rather than disassembling the table as code.
radius=0x3cf04b0
if radius in targets and radius not in mapping:
    body_pointer=struct.pack('<Q',old.base+0x3ceb400);index=rdold.find(body_pointer)
    if index>=0 and rdold.find(body_pointer,index+1)<0:
        oldtable=old.pe.sections[1].VirtualAddress+index-0x60
        newptr=struct.pack('<Q',new.base+mapping[0x3ceb400]);ni=rdnew.find(newptr)
        if ni>=0 and rdnew.find(newptr,ni+1)<0:
            newtable=new.pe.sections[1].VirtualAddress+ni-0x60
            candidate=struct.unpack('<Q',new.data(newtable+0xd0,8))[0]-new.base
            assert struct.unpack('<Q',old.data(oldtable+0xd0,8))[0]-old.base==radius
            same=old.norm(radius,radius+0xbf)[0]==new.norm(candidate,candidate+0xbf)[0]
            ot=struct.unpack('<II',old.data(radius+0xc0,8));nt=struct.unpack('<II',new.data(candidate+0xc0,8))
            same= same and all(oa-radius==na-candidate for oa,na in zip(ot,nt))
            same= same and old.data(radius+0xc8,17)==new.data(candidate+0xc8,17)
            if same:add(radius,candidate,'unique body-service vtable; complete code, relative switch targets and 17 case indices verified')

result={'oldSha256':hashlib.sha256(a.old.read_bytes()).hexdigest(),'newSha256':hashlib.sha256(a.new.read_bytes()).hexdigest(),
 'targets':len(targets),'mappedTargets':sum(x in mapping for x in targets),'conflicts':conflicts,
 'mapping':[{ 'old':hex(x),'new':hex(mapping[x]),'evidence':evidence[x],'sources':targets[x]} for x in sorted(targets) if x in mapping],
 'unresolved':[{'old':hex(x),'candidates':[hex(y) for y in ambiguous.get(x,[])],'sources':targets[x]} for x in sorted(targets) if x not in mapping]}
a.output.parent.mkdir(parents=True,exist_ok=True);a.output.write_text(json.dumps(result,indent=2),encoding='utf-8')
print(json.dumps({k:v for k,v in result.items() if k not in ('mapping','unresolved')},indent=2))
print('Unresolved:',', '.join(x['old'] for x in result['unresolved']))