Refactor, improve xml + xflash / da handling. Add da sla support. Bro…

…m sla untested.

mtkclient/Library/Auth/sla.py

@@ -0,0 +1,43 @@
+#!/usr/bin/python3
+# -*- coding: utf-8 -*-
+# (c) B.Kerler 2018-2024 GPLv3 License
+
+from Cryptodome.Hash import SHA256
+from Cryptodome.Util.number import bytes_to_long, ceil_div, size, long_to_bytes
+from Cryptodome.Cipher import PKCS1_OAEP
+from Cryptodome.PublicKey import RSA
+
+
+def customized_sign(n, e, msg):
+    mod_bits = size(n)
+    k = ceil_div(mod_bits, 8)
+
+    ps = b'\xFF' * (k - len(msg) - 3)
+    em = b'\x00\x01' + ps + b'\x00' + msg
+
+    em_int = bytes_to_long(em)
+    m_int = pow(em_int, e, n)
+    signature = long_to_bytes(m_int, k)
+
+    return signature
+
+
+def generate_brom_sla_challenge(d, e, data):
+    d = bytes_to_long(bytes.fromhex(d))
+    e = bytes_to_long(bytes.fromhex(e))
+    for i in range(0, len(data), 2):
+        data[i], data[i + 1] = data[i + 1], data[i]
+    msg = bytearray(customized_sign(d, e, data))
+    for i in range(0, len(msg), 2):
+        msg[i], msg[i + 1] = msg[i + 1], msg[i]
+    return msg
+
+
+def generate_da_sla_signature(data, d, n, e):
+    d_da = bytes_to_long(bytes.fromhex(d))
+    n_da = bytes_to_long(bytes.fromhex(n))
+    e_da = bytes_to_long(bytes.fromhex(e))
+    pprivate_key = RSA.construct((n_da, d_da, e_da))
+    cipher = PKCS1_OAEP.new(pprivate_key, SHA256, mgfunc=lambda x, y: PKCS1_OAEP.MGF1(x, y, SHA256))
+    ciphertext = cipher.encrypt(data)
+    return ciphertext

mtkclient/Library/Connection/devicehandler.py

@@ -1,6 +1,6 @@
 #!/usr/bin/python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2023
+# (c) B.Kerler 2018-2024
 import inspect
 import traceback
 import logging
@@ -45,7 +45,7 @@ def get_read_packetsize(self):
     def get_write_packetsize(self):
         raise NotImplementedError()
 
-    def connect(self, EP_IN=-1, EP_OUT=-1):
+    def connect(self, ep_in=-1, ep_out=-1):
         raise NotImplementedError()
 
     def setportname(self, portname: str):
@@ -63,16 +63,16 @@ def flush(self):
     def detectdevices(self):
         raise NotImplementedError()
 
-    def getInterfaceCount(self):
+    def get_interface_count(self):
         raise NotImplementedError()
 
-    def setLineCoding(self, baudrate=None, parity=0, databits=8, stopbits=1):
+    def set_line_coding(self, baudrate=None, parity=0, databits=8, stopbits=1):
         raise NotImplementedError()
 
     def setbreak(self):
         raise NotImplementedError()
 
-    def setcontrollinestate(self, RTS=None, DTR=None, isFTDI=False):
+    def setcontrollinestate(self, rts=None, dtr=None, is_ftdi=False):
         raise NotImplementedError()
 
     def write(self, command, pktsize=None):
@@ -87,7 +87,7 @@ def usbread(self, resplen=None, timeout=0):
     def usbxmlread(self, maxtimeout=100):
         raise NotImplementedError()
 
-    def ctrl_transfer(self, bmRequestType, bRequest, wValue, wIndex, data_or_wLength):
+    def ctrl_transfer(self, bm_request_type, b_request, w_value, w_index, data_or_w_length):
         raise NotImplementedError()
 
     def usbreadwrite(self, data, resplen):

mtkclient/Library/Connection/seriallib.py

@@ -1,6 +1,6 @@
 #!/usr/bin/python3
 # -*- coding: utf-8 -*-
-# (c) B.Kerler 2018-2022
+# (c) B.Kerler 2018-2024
 import time
 import sys
 import logging
@@ -22,21 +22,20 @@ def _reset_input_buffer_org(self):
         return termios.tcflush(self.fd, termios.TCIFLUSH)
 
 
-class serial_class(DeviceClass):
+class SerialClass(DeviceClass):
 
     def __init__(self, loglevel=logging.INFO, portconfig=None, devclass=-1):
         super().__init__(loglevel, portconfig, devclass)
         self.is_serial = True
         self.device = None
 
-    def connect(self, EP_IN=-1, EP_OUT=-1):
+    def connect(self, ep_in=-1, ep_out=-1):
         if self.connected:
             self.close()
             self.connected = False
 
         ports = self.detectdevices()
         if ports:
-            port = None
             if self.portname != "DETECT":
                 if self.portname not in ports:
                     self.debug("{} not in detected ports: {}".format(self.portname, ports))
@@ -71,7 +70,7 @@ def setportname(self, portname: str):
     def set_fast_mode(self, enabled):
         pass
 
-    def changeBaud(self):
+    def change_baud(self):
         print("Changing Baudrate")
         self.write(b'\xD2' + b'\x02' + b'\x01')
         self.read(1)
@@ -102,7 +101,7 @@ def detectdevices(self):
                     ids.append(port.device)
         return sorted(ids)
 
-    def setLineCoding(self, baudrate=None, parity=0, databits=8, stopbits=1):
+    def set_line_coding(self, baudrate=None, parity=0, databits=8, stopbits=1):
         self.device.baudrate = baudrate
         self.device.parity = parity
         self.device.stopbbits = stopbits
@@ -113,11 +112,11 @@ def setbreak(self):
         self.device.send_break()
         self.debug("Break set")
 
-    def setcontrollinestate(self, RTS=None, DTR=None, isFTDI=False):
-        if RTS == 1:
-            self.device.setRTS(RTS)
-        if DTR == 1:
-            self.device.setDTR(DTR)
+    def setcontrollinestate(self, rts=None, dtr=None, is_ftdi=False):
+        if rts == 1:
+            self.device.setRTS(rts)
+        if dtr == 1:
+            self.device.setDTR(dtr)
         self.debug("Linecoding set")
 
     def write(self, command, pktsize=None):
@@ -130,7 +129,7 @@ def write(self, command, pktsize=None):
             try:
                 self.device.write(b'')
             except Exception as err:
-                error = str(err.strerror)
+                error = str(err)
                 if "timeout" in error:
                     # time.sleep(0.01)
                     try:
@@ -180,7 +179,7 @@ def read(self, length=None, timeout=-1):
                 length = self.device.in_waiting
         return self.usbread(resplen=length, maxtimeout=timeout)
 
-    def getDevice(self):
+    def get_device(self):
         return self.device
 
     def get_read_packetsize(self):
@@ -190,7 +189,7 @@ def get_write_packetsize(self):
         return 0x200
 
     def flush(self):
-        if self.getDevice() is not None:
+        if self.get_device() is not None:
             self.device.flushOutput()
         return self.device.flush()