import hashlib import logging import os import sys import json from struct import unpack, pack from Cryptodome.Cipher import AES # from keystone.keystone import Ks # from keystone.keystone_const import KS_ARCH_ARM64, KS_MODE_LITTLE_ENDIAN from mtkclient.Library.Hardware.hwcrypto_sej import sej_cryptmode from mtkclient.Library.mtk_crypto import verify_checksum, nvram_keys, SST_Get_NVRAM_SW_Key from mtkclient.config.payloads import PathConfig from mtkclient.config.brom_config import Efuse from mtkclient.Library.error import ErrorHandler from mtkclient.Library.Hardware.hwcrypto import CryptoSetup, HwCrypto from mtkclient.Library.utils import find_binary, do_tcp_keyserver from mtkclient.Library.gui_utils import LogBase, progress, logsetup from mtkclient.Library.Hardware.seccfg import SecCfgV3, SecCfgV4 from mtkclient.Library.utils import MTKTee from mtkclient.Library.Exploit.exptools.aarch_tools import Aarch64Tools rpmb_error = [ "", "General failure", "Authentication failure", "Counter failure", "Address failure", "Write failure", "Read failure", "Authentication key not yet programmed" ] class XmlFlashExt(metaclass=LogBase): def __init__(self, _mtk, _xmlflash, loglevel): self.pathconfig = PathConfig() self.__logger, self.info, self.debug, self.warning, self.error = logsetup(self, self.__logger, loglevel, _mtk.config.gui) self.mtk = _mtk self.loglevel = loglevel self.__logger = self.__logger self.eh = ErrorHandler() self.config = self.mtk.config self.usbwrite = self.mtk.port.usbwrite self.usbread = self.mtk.port.usbread self.echo = self.mtk.port.echo self.rbyte = self.mtk.port.rbyte self.rdword = self.mtk.port.rdword self.rword = self.mtk.port.rword self.xflash = _xmlflash self.xsend = self.xflash.xsend self.xread = self.xflash.xread self.da2 = None self.da2address = None def patch_custom_command(self, _da2): self.da2address = self.xflash.daconfig.da_loader.region[2].m_start_addr # at_address data = bytearray(_da2) idx = data.find(b"\x00CMD:SET-HOST-INFO\x00") if idx == -1: return None base = self.da2address entry = int.from_bytes(_da2[4:8], byteorder='little') if entry & 0xF0 == 0xC0: self.is_arm64 = True else: self.is_arm64 = False if self.is_arm64: at = Aarch64Tools(_da2, self.da2address) ref_offset = at.find_string_xref("CMD:SET-HOST-INFO") if ref_offset is None: print("Error finding CMD:SET-HOST-INFO") return None bl_offset = at.get_next_bl_from_off(ref_offset) if bl_offset is None: print("Error finding xml_register_cmd") return None cmd_set_host = at.resolve_register_value_back(bl_offset, reg=2, lookback=5) if cmd_set_host is None: print("Error finding cmd_set_host_info") return None cmd_set_host_offset = cmd_set_host - self.da2address # Patch CMD:SET-HOST-INFO to point to our loader newcmd = b"CMD:CUSTOM\x00" data[idx + 1: idx + 1 + len(newcmd)] = newcmd # Now patch the existing SET-HOST-INFO command # ks = Ks(KS_ARCH_ARM64, KS_MODE_LITTLE_ENDIAN) content = """ STP X29, X30, [SP,#-0x30]! STR X21, [SP,#0x10] MOV X29, SP STP X20, X19, [SP,#0x20] MOV X19, X0 LDR X21, [X19] # Receive length 4 bytes ADRP X0, #0x68000000 ADD X0, X0, #0xF000 MOV X1, #4 BLR X21 # Receive data ADRP X0, #0x68000000 MOV X1, X0 ADD X1, X1, #0xF000 LDR X1, [X1] BLR X21 # Jump to payload ADRP X0, #0x68000000 BLR X0 MOV W0, WZR LDP X20, X19, [SP,#0x20] LDR X21, [SP,#0x10] LDP X29, X30, [SP],#0x30 RET """ _ = content # encoding, length = ks.asm(content, addr=cmd_set_host) # newdata = b"".join(int.to_bytes(val, 1, 'little') for val in encoding) # print(newdata.hex()) newdata = bytes.fromhex( "fd7bbda9f50b00f9fd030091f44f02a9f30300aa750240f9e0ff13b0003c4091810080d2a0023fd6e0ff13b0e10300aa213c4091210040f9a0023fd6e0ff13b000003fd6e0031f2af44f42a9f50b40f9fd7bc3a8c0035fd6") sys.stdout.flush() data[cmd_set_host_offset:cmd_set_host_offset + len(newdata)] = newdata return data else: first_op, second_op = offset_to_op_mov(idx + 1, 0, base) first_op = int.to_bytes(first_op, 4, 'little') second_op = int.to_bytes(second_op, 4, 'little') midx = data.find(first_op) midx2 = data.find(second_op, midx) if midx + 8 == midx2: instr1 = int.from_bytes(data[midx + 4:midx + 8], 'little') instr2 = int.from_bytes(data[midx2 + 4:midx2 + 8], 'little') addr = op_mov_to_offset(instr1, instr2, 2) - base # rw_primitive = bytes.fromhex("FF412DE90040A0E30460A0E30C708DE20050A0E10710A0E1003090E508008DE200408" + # "DE506808DE004408DE508408DE50C608DE533FF2FE108309DE50710A0E10D00A0E10C608DE5040053E1003095E50A00001" + # "A33FF2FE100309DE50610A0E10C608DE50800A0E1003093E504308DE5043095E533FF2FE110D08DE2F081BDE833FF2FE10" + # "03095E50710A0E10800A0E10C608DE533FF2FE100309DE50400A0E104209DE5002083E5F2FFFFEA") # ks = Ks(KS_ARCH_ARM, KS_MODE_ARM + KS_MODE_LITTLE_ENDIAN) # content = """ PUSH {R4-R6,R10,R11,LR} ADD R11, SP, #0x10 MOV R8, R0 MOVW R0, #0xF000 MOVT R0, #0x6800 MOV R1, #4 LDR R2, [R8] BLX R2 MOVW R0, #0xF000 MOVT R0, #0x6800 MOV R1, [R0] MOVW R0, #0x0000 MOVT R0, #0x6800 LDR R2, [R8] BLX R2 MOVW R0, #0x0000 MOVT R0, #0x6800 BLX R0 POP {R4-R6,R10,R11,PC} """ # encoding, length = ks.asm(content, addr=addr) # newdata = b"".join(int.to_bytes(val, 1, 'little') for val in encoding) newdata = bytes.fromhex( "704c2de910b08de20080a0e100000fe3000846e30410a0e3002098e532ff2fe100000fe3000846e3000000e3000846e3" + "002098e532ff2fe1000000e3000846e330ff2fe1708cbde8") sys.stdout.flush() data[addr:addr + len(newdata)] = newdata newcmd = b"CMD:CUSTOM\x00" data[idx + 1:idx + 1 + len(newcmd)] = newcmd return data return _da2 def ack(self): xmlcmd = self.xflash.cmd.create_cmd("CUSTOMACK") if self.xsend(xmlcmd): # result = result = self.xflash.get_response() _ = result # DATA data = data = self.xflash.get_response(raw=True) # CMD:END result = result2 = self.xflash.get_response() _ = result2 self.xflash.ack() # CMD:START result = resp = self.xflash.get_response() _ = resp self.xflash.ack() if data == b"\xA4\xA3\xA2\xA1": return True return False def patch(self): self.da2 = self.xflash.daconfig.da2 self.da2address = self.xflash.daconfig.da_loader.region[2].m_start_addr # at_address base = self.da2address entry = int.from_bytes(self.da2[4:8], byteorder='little') if entry & 0xF0 == 0xC0: self.is_arm64 = True else: self.is_arm64 = False if self.is_arm64: daextensions = os.path.join(self.pathconfig.get_payloads_path(), "da_xml_64.bin") else: daextensions = os.path.join(self.pathconfig.get_payloads_path(), "da_xml.bin") if os.path.exists(daextensions): daextdata = bytearray(open(daextensions, "rb").read()) if self.is_arm64: register_ptr = daextdata.find(b"\x11\x11\x11\x11") mmc_get_card_ptr = daextdata.find(b"\x22\x22\x22\x22\x22\x22\x22\x22") mmc_set_part_config_ptr = daextdata.find(b"\x33\x33\x33\x33\x33\x33\x33\x33") mmc_rpmb_send_command_ptr = daextdata.find(b"\x44\x44\x44\x44\x44\x44\x44\x44") ufshcd_queuecommand_ptr = daextdata.find(b"\x55\x55\x55\x55\x55\x55\x55\x55") ufshcd_get_free_tag_ptr = daextdata.find(b"\x66\x66\x66\x66\x66\x66\x66\x66") ptr_g_ufs_hba_ptr = daextdata.find(b"\x77\x77\x77\x77\x77\x77\x77\x77") efuse_addr_ptr = daextdata.find(b"\x88\x88\x88\x88") register_xml_cmd = find_binary(self.da2, b"\xFD\x7B\xBD\xA9\xF5\x0B\x00\xF9\xFD\x03\x00\x91\xF4\x4F\x02\xA9\xF5\x03\x00\xAA\x00\x05\x80\x52") # UFS ar = Aarch64Tools(self.da2, self.da2address) ufs_controller_enable_offset = ar.find_function_from_string("Controller enable failed\n") instr = int.from_bytes(self.da2[ufs_controller_enable_offset + 0xC:ufs_controller_enable_offset + 0x10], 'little') g_ufs_hba = ar.decode_adrp(instr, ufs_controller_enable_offset + 0xC + self.da2address)[0] ufshcd_queuecommand = 0 ufshcd_get_free_tag = 0 ufshcd_queuecommand = find_binary(self.da2, b"\xFD\x7B\xBC\xA9\xF7\x0B\x00\xF9\xFD\x03\x00\x91\xF6\x57\x02\xA9\xF4\x4F\x03\xA9\x14\x34\x40\xF9\xF5\x03\x01\xAA") if ufshcd_queuecommand is None: ufshcd_queuecommand = 0 else: ufshcd_queuecommand = ufshcd_queuecommand + self.da2address ufshcd_get_free_tag = ar.find_function_from_string("[UFS] ufshcd_get_free_tag fail\n") if ufshcd_get_free_tag is None: ufshcd_get_free_tag = 0 else: ufshcd_get_free_tag = ufshcd_get_free_tag + self.da2address # EMMC mmc_switch_part_offset = ar.find_function_from_string("mmc_switch_part") mmc_get_card_ptr2 = ar.get_next_bl_from_off(mmc_switch_part_offset) mmc_get_card = ar.get_bl_target(mmc_get_card_ptr2) mmc_set_part_config = find_binary(self.da2, b"\x08\x80\x40\x39\x88\x00\x08\x37\x08\xB4\x49\x39") if mmc_set_part_config is None: mmc_set_part_config = 0 mmc_rpmb_offs = ar.find_function_from_string("%sreq/1 fail %d\n") logl = ar.get_next_bl_from_off(mmc_rpmb_offs) mmc_rpmb_send_command_ptr2 = ar.get_next_bl_from_off(logl + 0x8) mmc_rpmb_send_command = ar.get_bl_target(mmc_rpmb_send_command_ptr2) if mmc_rpmb_send_command is None: mmc_rpmb_send_command = 0 else: register_ptr = daextdata.find(b"\x11\x11\x11\x11") mmc_get_card_ptr = daextdata.find(b"\x22\x22\x22\x22") mmc_set_part_config_ptr = daextdata.find(b"\x33\x33\x33\x33") mmc_rpmb_send_command_ptr = daextdata.find(b"\x44\x44\x44\x44") ufshcd_queuecommand_ptr = daextdata.find(b"\x55\x55\x55\x55") ufshcd_get_free_tag_ptr = daextdata.find(b"\x66\x66\x66\x66") ptr_g_ufs_hba_ptr = daextdata.find(b"\x77\x77\x77\x77") efuse_addr_ptr = daextdata.find(b"\x88\x88\x88\x88") # 32bit # register_xml_cmd("CMD:GET-SYS-PROPERTY", & a1, cmd_get_sys_property); # open("out" + hex(self.da2address) + ".da", "wb").write(da2) register_xml_cmd = find_binary(self.da2, b"\x70\x4C\x2D\xE9\x10\xB0\x8D\xE2\x00\x50\xA0\xE1\x14\x00\xA0\xE3") # UFS idx = self.da2.find(b"\x00\x00\x94\xE5\x34\x10\x90\xE5\x01\x00\x11\xE3\x03\x00\x00\x0A") g_ufs_hba = 0 ufshcd_queuecommand = 0 ufshcd_get_free_tag = 0 if idx != -1: instr1 = int.from_bytes(self.da2[idx - 0x8:idx - 0x4], 'little') instr2 = int.from_bytes(self.da2[idx - 0x4:idx], 'little') g_ufs_hba = op_mov_to_offset(instr1, instr2, 4) ufshcd_queuecommand = find_binary(self.da2, b"\xF0\x4D\x2D\xE9\x18\xB0\x8D\xE2\x08\xD0\x4D\xE2\x48\x40\x90\xE5") if ufshcd_queuecommand is None: ufshcd_queuecommand = find_binary(self.da2, b"\xF0\x4F\x2D\xE9\x1C\xB0\x8D\xE2\x0C\xD0\x4D\xE2\x48\xA0\x90\xE5\x00\x80\xA0\xE1") if ufshcd_queuecommand is None: ufshcd_queuecommand = 0 else: ufshcd_queuecommand = ufshcd_queuecommand + self.da2address else: ufshcd_queuecommand = ufshcd_queuecommand + self.da2address ufshcd_get_free_tag = find_binary(self.da2, b"\x10\x4C\x2D\xE9\x08\xB0\x8D\xE2\x00\x40\xA0\xE3\x00\x00\x51\xE3") if ufshcd_get_free_tag is None: ufshcd_get_free_tag = 0 else: ufshcd_get_free_tag = ufshcd_get_free_tag + self.da2address # EMMC mmc_get_card = find_binary(self.da2, b"\x90\x12\x20\xE0\x1E\xFF\x2F\xE1") if mmc_get_card is not None: mmc_get_card -= 0xC else: mmc_get_card = 0 mmc_set_part_config = find_binary(self.da2, b"\xF0\x4B\x2D\xE9\x18\xB0\x8D\xE2\x23\xDE\x4D\xE2") if mmc_set_part_config is None: mmc_set_part_config = find_binary(self.da2, b"\xF0\x4B\x2D\xE9\x18\xB0\x8D\xE2\x8E\xDF\x4D\xE2") if mmc_set_part_config is None: mmc_set_part_config = 0 mmc_rpmb_send_command = find_binary(self.da2, b"\xF0\x48\x2D\xE9\x10\xB0\x8D\xE2\x08\x70\x9B\xE5") if mmc_rpmb_send_command is None: mmc_rpmb_send_command = 0 efuse_addr = self.config.chipconfig.efuse_addr if register_ptr != -1: if register_xml_cmd: register_xml_cmd = register_xml_cmd + self.da2address else: register_xml_cmd = 0 length = 4 if self.is_arm64: # Patch the addr length = 8 daextdata[register_ptr:register_ptr + 4] = pack(" 0: with open(filename, "wb") as wf: pos = 0 toread = sectors while toread > 0: sz = min(sectors - pos, 0x10) data = self.custom_rpmb_read(sector=sector + pos, sectors=sz) if data == b"": if display: progressbar.done() self.error("Couldn't read rpmb.") return False if display: progressbar.update(sz * 256) wf.write(data) pos += sz toread -= sz if display: progressbar.done() self.info(f"Done reading rpmb to {filename}") return True return False def write_rpmb(self, filename=None, sector: int = None, sectors: int = None, display=True): if filename is None: self.error("Filename has to be given for writing to rpmb") return False if not os.path.exists(filename): self.error(f"Couldn't find {filename} for writing to rpmb.") return False if sectors == 0: max_sector_size = (512 * 256) if self.xflash.daconfig.storage.emmc is not None: max_sector_size = self.xflash.daconfig.storage.emmc.rpmb_size // 0x100 else: max_sector_size = sectors filesize = os.path.getsize(filename) sectors = min(filesize // 256, max_sector_size) progressbar = progress(total=sectors * 256, pagesize=1, guiprogress=self.mtk.config.guiprogress, prefix="RPMB write:") if self.custom_rpmb_init(): if sectors > 0: with open(filename, "rb") as rf: pos = 0 towrite = sectors while towrite > 0: sz = min(sectors - pos, 0x10) if not self.custom_rpmb_write(sector=sector + pos, sectors=sz, data=rf.read(0x100 * sz)): if display: progressbar.done() self.error(f"Couldn't write rpmb at sector {sector + pos}.") return False if display: progressbar.update(sz * 256) pos += sz towrite -= sz if display: progressbar.done() self.info(f"Done writing {filename} to rpmb") return True return False def auth_rpmb(self, rpmbkey: bytes = None): if self.custom_rpmb_init(rpmbkey): return True return False def erase_rpmb(self, sector: int = None, sectors: int = None, display=True): if sector is None: sector = 0 if sectors is None: if self.xflash.daconfig.storage.emmc is not None: sectors = self.xflash.daconfig.storage.emmc.rpmb_size // 0x100 elif self.xflash.daconfig.storage.ufs.block_size != 0: sectors = (512 * 256) progressbar = progress(total=sectors * 256, pagesize=1, guiprogress=self.mtk.config.guiprogress, prefix="RPMB erase:") if self.custom_rpmb_init(): if sectors > 0: pos = 0 towrite = sectors while towrite > 0: sz = min(sectors - pos, 0x10) if not self.custom_rpmb_write(sector=sector + pos, sectors=sz, data=b"\x00" * 0x100 * sz): if display: progressbar.done() self.error(f"Couldn't erase rpmb at sector {sector + pos}.") return False if display: progressbar.update(sz * 256) pos += sz towrite -= sz if display: progressbar.done() self.info("Done erasing rpmb") return True return False def custom_read(self, addr, length, registers=True) -> bytes: xmlcmd = self.xflash.cmd.create_cmd("CUSTOMMEMR") if self.xsend(xmlcmd): result = self.xflash.get_response() if result == "OK": # DATA self.xsend(data=addr, is64bit=True) self.xsend(length) data = self.xflash.get_response(raw=True) # CMD:END result = self.xflash.get_response() self.xflash.ack() # CMD:START result = self.xflash.get_response() self.xflash.ack() return data return b"" def custom_read_reg(self, addr: int, length: int) -> bytes: tmp = self.custom_readregister(addr, length // 4) if isinstance(tmp, int): return int.to_bytes(tmp, 4, 'little') else: data = bytearray(b"".join([tmp[i].to_bytes(4, 'little') for i in range(len(tmp))])) return data def custom_readregister(self, addr, dwords=1) -> (int, None): xmlcmd = self.xflash.cmd.create_cmd("CUSTOMREGR") if self.xsend(xmlcmd): result = self.xflash.get_response() if result == "OK": # DATA self.xsend(data=addr) self.xsend(data=dwords) data = self.xflash.get_response(raw=True) # CMD:END result = self.xflash.get_response() self.xflash.ack() # CMD:START result = self.xflash.get_response() self.xflash.ack() if dwords == 1: return int.from_bytes(data, 'little') else: return [int.from_bytes(data[pos:pos + 4], 'little') for pos in range(0, len(data), 4)] return None def custom_write(self, addr, data) -> bool: xmlcmd = self.xflash.cmd.create_cmd("CUSTOMMEMR") if self.xsend(xmlcmd): result = self.xflash.get_response() if result == "OK": # DATA self.xsend(data=addr, is64bit=True) self.xsend(len(data)) self.xsend(data) # CMD:END result = self.xflash.get_response() self.xflash.ack() # CMD:START result = self.xflash.get_response() self.xflash.ack() return True return False def custom_writeregister(self, addr: int, data: int): xmlcmd = self.xflash.cmd.create_cmd("CUSTOMREGW") if self.xsend(xmlcmd): result = self.xflash.get_response() if result == "OK": # DATA self.xsend(addr) self.xsend(data) # CMD:END result = self.xflash.get_response() self.xflash.ack() # CMD:START result = self.xflash.get_response() self.xflash.ack() return True return False def readmem(self, addr, dwords=1): if dwords < 0x20: res = self.custom_readregister(addr, dwords) else: res = self.custom_read(addr, dwords * 4) res = [unpack(" " + bytearray(b"".join(pack(" {hex(res)}") return res def writeregister(self, addr, dwords): if isinstance(dwords, int): dwords = [dwords] pos = 0 if len(dwords) < 0x20: for val in dwords: self.debug(f"TX: {hex(addr + pos)} -> " + hex(val)) if not self.custom_writeregister(addr + pos, val): return False pos += 4 else: dat = b"".join([pack("> 12) & 0xF) << 16 | (register << 14) | offset & 0xFFF offset = (offset >> 16) shift = 4 high = (((offset & 0xFFFF) >> 12) & 0xF) << 16 | (register << 14) | offset & 0xFFF | (shift << 20) first_op = (0xE3 << 24) + low second_op = (0xE3 << 24) + high return first_op, second_op def op_mov_to_offset(first_op, second_op, register): reglo = (first_op & 0xF000) >> 12 reghi = (second_op & 0xF000) >> 12 shiftlo = (first_op & 0xF00000) >> 20 shifthi = (second_op & 0xF00000) >> 20 hi = ((second_op & 0xF0000) >> 4 | second_op & 0xFFF) << shifthi * 4 lo = ((first_op & 0xF0000) >> 4 | first_op & 0xFFF) << shiftlo * 4 if reglo == reghi == register: return hi | lo return None