XFEEDToolBox

"""Compare captured executable bytes with the exact local PE; no live process access."""
import argparse
import json
import struct
import faulthandler
from pathlib import Path
import pefile
from capstone import Cs, CS_ARCH_X86, CS_MODE_64

p = argparse.ArgumentParser()
p.add_argument('dump', type=Path)
p.add_argument('image', type=Path)
p.add_argument('--output', type=Path, required=True)
a = p.parse_args()
faulthandler.dump_traceback_later(30)
b = a.dump.read_bytes()
n, directory = struct.unpack_from('<II', b, 8)
streams = {struct.unpack_from('<III', b, directory + i * 12)[0]: struct.unpack_from('<III', b, directory + i * 12)[1:] for i in range(n)}
_, m = streams[4]
n, = struct.unpack_from('<I', b, m)
base = None
for i in range(n):
    at = m + 4 + i * 108
    candidate, size, checksum, stamp, name = struct.unpack_from('<QIIII', b, at)
    length, = struct.unpack_from('<I', b, name)
    path = b[name+4:name+4+length].decode('utf-16-le')
    if Path(path).name.lower() == a.image.name.lower():
        base = candidate
        break
if base is None:
    raise ValueError('Image not in dump')
memory = []
if 5 in streams:
    _, at = streams[5]
    n, = struct.unpack_from('<I', b, at)
    memory.extend(struct.unpack_from('<QII', b, at + 4 + i * 16) for i in range(n))
if 9 in streams:
    _, at = streams[9]
    n, rva = struct.unpack_from('<QQ', b, at)
    for i in range(n):
        start, size = struct.unpack_from('<QQ', b, at + 16 + i * 16)
        memory.append((start, size, rva))
        rva += size
pe = pefile.PE(str(a.image), fast_load=True)
pe.parse_data_directories(directories=[5])
image = bytearray(pe.get_memory_mapped_image())
delta = base - pe.OPTIONAL_HEADER.ImageBase
for block in getattr(pe, 'DIRECTORY_ENTRY_BASERELOC', []):
    for entry in block.entries:
        if entry.type == 10:
            value, = struct.unpack_from('<Q', image, entry.rva)
            struct.pack_into('<Q', image, entry.rva, (value + delta) & 0xffffffffffffffff)
        elif entry.type == 3:
            value, = struct.unpack_from('<I', image, entry.rva)
            struct.pack_into('<I', image, entry.rva, (value + delta) & 0xffffffff)
sections = [(s.VirtualAddress, s.VirtualAddress + s.Misc_VirtualSize) for s in pe.sections if s.Characteristics & 0x20000000]
decoder = Cs(CS_ARCH_X86, CS_MODE_64)
diffs, covered = [], 0
for start, size, rva in memory:
    for lower, upper in sections:
        first, last = max(start, base + lower), min(start + size, base + upper)
        if last <= first:
            continue
        covered += last - first
        expected = image[first-base:last-base]
        actual = b[rva+first-start:rva+last-start]
        i = 0
        while i < len(actual):
            if expected[i] == actual[i]:
                i += 1
                continue
            begin = i
            while i < len(actual) and (expected[i] != actual[i] or (i + 4 < len(actual) and any(expected[k] != actual[k] for k in range(i, i + 4)))):
                i += 1
            end = i
            diffs.append({'rva': hex(first-base+begin), 'length': end-begin,
                          'disk': expected[begin:end].hex(), 'dump': actual[begin:end].hex(),
                          'instructions': [d.mnemonic+' '+d.op_str for d in decoder.disasm(actual[begin:min(len(actual), end+16)], first+begin)]})
report = {'dump': str(a.dump), 'image': str(a.image), 'base': hex(base), 'capturedExecutableBytes': covered, 'changes': diffs}
a.output.write_text(json.dumps(report, indent=2), encoding='utf-8')
print(json.dumps({'dump': a.dump.name, 'capturedExecutableBytes': covered, 'changes': diffs[:40]}))