Prepare V6 flash commands, improve shellcode (speed for raw read flas…

…h/rpmb)

mtkclient/Library/Connection/devicehandler.py

@@ -75,6 +75,9 @@ def usbwrite(self, data, pktsize=None):
     def usbread(self, resplen=None, timeout=0):
         raise NotImplementedError()
 
+    def usbxmlread(self, maxtimeout=100):
+        raise NotImplementedError()
+
     def ctrl_transfer(self, bmRequestType, bRequest, wValue, wIndex, data_or_wLength):
         raise NotImplementedError()
 

mtkclient/Library/Connection/seriallib.py

@@ -3,6 +3,9 @@
 # (c) B.Kerler 2018-2022
 import time
 import sys
+
+from mtkclient.Library.DA.xml.xml_param import max_xml_data_length
+
 if sys.platform != "win32":
     import termios
 
@@ -161,14 +164,6 @@ def usbread(self, resplen=None, timeout=0):
         self.device.timeout = timeout
         epr = self.device.read
         extend = res.extend
-        if self.xmlread:
-            info=self.device.read(6)
-            bytestoread=resplen-len(info)
-            extend(info)
-            if b"<?xml " in info:
-                while not b"response " in res or res[-7:]!=b"</data>":
-                    extend(epr(1))
-                return res
         bytestoread = resplen
         while len(res) < bytestoread:
             try:
@@ -199,6 +194,45 @@ def usbread(self, resplen=None, timeout=0):
                 self.verify_data(res[:resplen], "RX:")
         return res[:resplen]
 
+    def usbxmlread(self, timeout=0):
+        resplen = self.device.in_waiting
+        res = bytearray()
+        loglevel = self.loglevel
+        self.device.timeout = timeout
+        epr = self.device.read
+        extend = res.extend
+        bytestoread = max_xml_data_length
+        while len(res) < bytestoread:
+            try:
+                val=epr(bytestoread)
+                if len(val)==0:
+                    break
+                extend(val)
+                if res[-1]==b"\x00":
+                    break
+            except Exception as e:
+                error = str(e)
+                if "timed out" in error:
+                    if timeout is None:
+                        return b""
+                    self.debug("Timed out")
+                    if timeout == 10:
+                        return b""
+                    timeout += 1
+                    pass
+                elif "Overflow" in error:
+                    self.error("USB Overflow")
+                    return b""
+                else:
+                    self.info(repr(e))
+                    return b""
+
+        if loglevel == logging.DEBUG:
+            self.debug(inspect.currentframe().f_back.f_code.co_name + ":" + hex(resplen))
+            if self.loglevel == logging.DEBUG:
+                self.verify_data(res[:resplen], "RX:")
+        return res[:resplen]
+
     def usbwrite(self, data, pktsize=None):
         if pktsize is None:
             pktsize = len(data)

mtkclient/Library/Connection/usblib.py

@@ -15,6 +15,8 @@
 from enum import Enum
 from binascii import hexlify
 from ctypes import c_void_p, c_int
+
+from mtkclient.Library.DA.xml.xml_param import max_xml_data_length
 from mtkclient.Library.utils import *
 from mtkclient.Library.Connection.devicehandler import DeviceClass
 USB_DIR_OUT = 0  # to device
@@ -453,6 +455,46 @@ def usbread(self, resplen=None, maxtimeout=100):
         return res[:resplen]
 
 
+    def usbxmlread(self, maxtimeout=100):
+        res = bytearray()
+        timeout = 0
+        loglevel = self.loglevel
+        epr = self.EP_IN.read
+        wMaxPacketSize = self.EP_IN.wMaxPacketSize
+        extend = res.extend
+        buffer = None
+        if self.fast:
+            buffer = self.buffer[:wMaxPacketSize]
+        while len(res) < max_xml_data_length:
+            try:
+                if self.fast:
+                    rlen = epr(buffer, timeout)
+                    extend(buffer[:rlen])
+                else:
+                    extend(epr(wMaxPacketSize))
+            except usb.core.USBError as e:
+                error = str(e.strerror)
+                if "timed out" in error:
+                    self.debug("Timed out")
+                    if timeout == maxtimeout:
+                        return b""
+                    timeout += 1
+                    pass
+                elif "Overflow" in error:
+                    self.error("USB Overflow")
+                    return b""
+                else:
+                    self.info(repr(e))
+                    return b""
+            if res[-1] == b"\x00":
+                break
+
+        if loglevel == logging.DEBUG:
+            self.debug(inspect.currentframe().f_back.f_code.co_name + ":" + hex(len(res)))
+            if self.loglevel == logging.DEBUG:
+                self.verify_data(res, "RX:")
+        return res
+
     def ctrl_transfer(self, bmRequestType, bRequest, wValue, wIndex, data_or_wLength):
         ret = self.device.ctrl_transfer(bmRequestType=bmRequestType, bRequest=bRequest, wValue=wValue, wIndex=wIndex,
                                         data_or_wLength=data_or_wLength)

mtkclient/Library/DA/legacy/dalegacy_lib.py

@@ -463,7 +463,7 @@ def read_flash_info(self):
             return True
         return False
 
-    def upload(self):
+    def upload_da1(self):
         if not self.config.iot:
             if self.daconfig.da_loader is None:
                 self.error("No valid da loader found... aborting.")

mtkclient/Library/DA/mtk_da_handler.py

@@ -8,7 +8,7 @@
 from mtkclient.config.payloads import pathconfig
 from mtkclient.Library.error import ErrorHandler
 from mtkclient.Library.utils import progress
-from mtkclient.config.brom_config import efuse
+from mtkclient.config.brom_config import efuse, damodes
 
 
 class DA_handler(metaclass=LogBase):
@@ -552,7 +552,7 @@ def da_peek(self, addr: int, length: int, filename: str):
         bytestoread = length
         pos = 0
         pagesize = 0x200
-        if self.mtk.daloader.xflash:
+        if self.mtk.daloader.flashmode == damodes.XFLASH:
             pagesize = self.mtk.daloader.get_packet_length()
         pg = progress(pagesize)
         bytesread = 0

mtkclient/Library/DA/mtk_daloader.py

@@ -6,12 +6,14 @@
 import os
 import hashlib
 from binascii import hexlify
+
+from mtkclient.Library.DA.xml.xml_lib import DAXML
 from mtkclient.Library.utils import LogBase, logsetup, progress
 from mtkclient.Library.error import ErrorHandler
 from mtkclient.Library.DA.daconfig import DAconfig
-from mtkclient.Library.DA.legacy.dalegacy import DALegacy
+from mtkclient.Library.DA.legacy.dalegacy_lib import DALegacy
 from mtkclient.Library.DA.legacy.dalegacy_flash_param import norinfo, emmcinfo, sdcinfo, nandinfo64
-from mtkclient.Library.DA.xflash.xflash import DAXFlash
+from mtkclient.Library.DA.xflash.xflash_lib import DAXFlash
 from mtkclient.config.brom_config import damodes
 from mtkclient.Library.DA.xflash.extension.xflash import xflashext
 from mtkclient.Library.DA.legacy.extension.legacy import legacyext
@@ -39,11 +41,16 @@ def __init__(self, mtk, loglevel=logging.INFO):
         self.xft = None
         self.lft = None
         self.da = None
-        self.xflash = None
+        self.flashmode = None
 
     def writestate(self):
         config = {}
-        config["xflash"] = self.mtk.config.chipconfig.damode == damodes.XFLASH
+        if self.mtk.config.chipconfig.damode == damodes.LEGACY:
+            config["flashmode"] = "LEGACY"
+        elif self.mtk.config.chipconfig.damode == damodes.XFLASH:
+            config["flashmode"] = "XFLASH"
+        elif self.mtk.config.chipconfig.damode == damodes.XML:
+            config["flashmode"] = "XML"
         config["hwcode"] = self.config.hwcode
         if self.config.meid is not None:
             config["meid"] = hexlify(self.config.meid).decode('utf-8')
@@ -105,16 +112,34 @@ def reinit(self):
                 self.config.meid = bytes.fromhex(config["meid"])
             if "socid" in config:
                 self.config.socid = bytes.fromhex(config["socid"])
-            self.xflash = config["xflash"]
+            if config["flashmode"] == "LEGACY":
+                self.mtk.config.chipconfig.damode = damodes.LEGACY
+                self.flashmode = damodes.LEGACY
+            elif config["flashmode"] == "XFLASH":
+                self.mtk.config.chipconfig.damode = damodes.XFLASH
+                self.flashmode = damodes.XFLASH
+            elif config["flashmode"] == "XML":
+                self.mtk.config.chipconfig.damode = damodes.XML
+                self.flashmode = damodes.XML
+
             self.config.init_hwcode(self.config.hwcode)
-            if self.xflash:
+            if self.flashmode == damodes.XML:
+                self.da = DAXML(self.mtk, self.daconfig, self.loglevel)
+                self.daconfig.flashtype = config["flashtype"]
+                self.daconfig.flashsize = config["flashsize"]
+                self.da.reinit()
+                self.xmlft = None
+                self.xft = None
+                self.lft = None
+            if self.flashmode == damodes.XFLASH:
                 self.da = DAXFlash(self.mtk, self.daconfig, self.loglevel)
                 self.daconfig.flashtype = config["flashtype"]
                 self.daconfig.flashsize = config["flashsize"]
                 self.da.reinit()
                 self.xft = xflashext(self.mtk, self.da, self.loglevel)
                 self.lft = None
-            else:
+                self.xmlft = None
+            elif self.flashmode == damodes.LEGACY:
                 self.da = DALegacy(self.mtk, self.daconfig, self.loglevel)
                 self.daconfig.flashtype = config["flashtype"]
                 self.daconfig.flashsize = config["flashsize"]
@@ -132,30 +157,33 @@ def reinit(self):
                     self.da.sdc.m_sdmmc_ua_size = config["m_sdmmc_ua_size"]
                 self.da.nor.m_nor_flash_size = config["m_nor_flash_size"]
                 self.xft = None
+                self.xmlft = None
             return True
         return False
 
     def set_da(self):
-        self.xflash = False
+        self.flashmode = damodes.LEGACY
         if self.mtk.config.plcap is not None:
             PL_CAP0_XFLASH_SUPPORT = (0x1 << 0)
             if self.mtk.config.plcap[
                 0] & PL_CAP0_XFLASH_SUPPORT == PL_CAP0_XFLASH_SUPPORT and self.mtk.config.blver > 1:
-                self.xflash = True
-        if self.mtk.config.chipconfig.damode == 1:
-            self.xflash = True
-        if self.xflash:
+                self.flashmode = damodes.XFLASH
+        if self.mtk.config.chipconfig.damode == damodes.XFLASH:
+            self.flashmode = damodes.XFLASH
+        elif self.mtk.config.chipconfig.damode == damodes.XML:
+            self.flashmode = damodes.XML
+        if self.flashmode == damodes.XFLASH:
             self.da = DAXFlash(self.mtk, self.daconfig, self.loglevel)
             self.xft = xflashext(self.mtk, self.da, self.loglevel)
-        else:
+        elif self.flashmode == damodes.LEGACY:
             self.da = DALegacy(self.mtk, self.daconfig, self.loglevel)
             self.lft = legacyext(self.mtk, self.da, self.loglevel)
+        elif self.flashmode == damodes.XML:
+            self.da = DAXML(self.mtk,self.daconfig,self.loglevel)
+            self.xmlft = None
 
     def setmetamode(self, porttype: str):
-        self.xflash = False
-        if self.mtk.config.chipconfig.damode == 1:
-            self.xflash = True
-        if self.xflash:
+        if self.mtk.config.chipconfig.damode == damodes.XFLASH:
             self.da = DAXFlash(self.mtk, self.daconfig, self.loglevel)
             if porttype not in ["off", "usb", "uart"]:
                 self.error("Only \"off\",\"usb\" or \"uart\" are allowed.")
@@ -195,7 +223,7 @@ def get_gpt(self, parttype=None) -> tuple:
         return data, guid_gpt
 
     def upload(self):
-        return self.da.upload()
+        return self.da.upload_da1()
 
     class ShutDownModes:
         NORMAL = 0
@@ -222,59 +250,59 @@ def readflash(self, addr, length, filename, parttype, display=True):
         return self.da.readflash(addr=addr, length=length, filename=filename, parttype=parttype, display=display)
 
     def get_packet_length(self):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             pt = self.da.get_packet_length()
             return pt.read_packet_length
         else:
             return 512
 
     def peek(self, addr: int, length: int):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.custom_read(addr=addr, length=length)
-        else:
+        if self.flashmode == damodes.LEGACY:
             return self.lft.custom_read(addr=addr, length=length)
 
     def poke(self, addr: int, data: bytes or bytearray):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.custom_write(addr=addr, data=data)
-        else:
+        elif self.flashmode == damodes.LEGACY:
             return self.lft.custom_write(addr=addr, data=data)
 
     def keys(self):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.generate_keys()
-        else:
+        elif self.flashmode == damodes.LEGACY:
             return self.lft.generate_keys()
 
     def readfuses(self):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.readfuses()
-        else:
+        elif self.flashmode == damodes.LEGACY:
             return self.lft.readfuses()
 
     def is_patched(self):
         return self.da.patch
 
     def seccfg(self, lockflag):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.seccfg(lockflag)
-        else:
+        elif self.flashmode == damodes.LEGACY:
             return self.lft.seccfg(lockflag)
 
     def read_rpmb(self, filename=None):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.read_rpmb(filename)
         self.error("Device is not in xflash mode, cannot run read rpmb cmd.")
         return False
 
     def write_rpmb(self, filename=None):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.write_rpmb(filename)
         self.error("Device is not in xflash mode, cannot run write rpmb cmd.")
         return False
 
     def erase_rpmb(self):
-        if self.xflash:
+        if self.flashmode == damodes.XFLASH:
             return self.xft.erase_rpmb()
         self.error("Device is not in xflash mode, cannot run erase rpmb cmd.")
         return False

mtkclient/Library/DA/xflash/xflash_lib.py

@@ -20,8 +20,6 @@
 rq = Queue()
 
 
-
-
 class DAXFlash(metaclass=LogBase):
     def __init__(self, mtk, daconfig, loglevel=logging.INFO):
         self.__logger = logsetup(self, self.__logger, loglevel, mtk.config.gui)
@@ -952,7 +950,7 @@ def setup_hw_init(self):
                 return True
         return False
 
-    def upload(self):
+    def upload_da1(self):
         if self.daconfig.da_loader is None:
             self.error("No valid da loader found... aborting.")
             return False
@@ -1057,7 +1055,7 @@ def reinit(self, display=False):
             self.config.set_gui_status(self.config.tr("Connected to stage2 with higher speed"))
 
     def upload_da(self):
-        if self.upload():
+        if self.upload_da1():
             self.get_expire_date()
             self.set_reset_key(0x68)
             # self.set_battery_opt(0x2)