feat: refactor authentication methods and add getBanks functionality

Author: thedtvn

mbbank/main.py

@@ -46,7 +46,7 @@ class MBBank:
 
     FPR = "c7a1beebb9400375bb187daa33de9659"
 
-    def __init__(self, *, username: str, password: str, proxy: dict=None, ocr_class=None):
+    def __init__(self, *, username: str, password: str, proxy: dict = None, ocr_class=None):
         self._userid = username
         self._password = password
         self._wasm_cache = None
@@ -113,53 +113,76 @@ def _get_wasm_file(self):
         self._wasm_cache = file_data
         return file_data
 
+    def get_capcha_image(self) -> bytes:
+        """
+        Get capcha image as bytes
+
+        Returns:
+            success (bytes): capcha image as bytes
+        """
+        rid = f"{self._userid}-{self._get_now_time()}"
+        json_data = {
+            'sessionId': "",
+            'refNo': rid,
+            'deviceIdCommon': self.deviceIdCommon,
+        }
+        headers = headers_default.copy()
+        headers["X-Request-Id"] = rid
+        headers["Deviceid"] = self.deviceIdCommon
+        headers["Refno"] = rid
+        with requests.Session() as s:
+            with s.post("https://online.mbbank.com.vn/retail-web-internetbankingms/getCaptchaImage",
+                        headers=headers, json=json_data,
+                        proxies=self.proxy) as r:
+                data_out = r.json()
+                return base64.b64decode(data_out["imageString"])
+
+    def login(self, captcha_text: str):
+        """
+        Login to MBBank account
+
+        Args:
+            captcha_text (str): capcha text from capcha image
+
+        Raises:
+            MBBankError: if login failed
+        """
+        payload = {
+            "userId": self._userid,
+            "password": hashlib.md5(self._password.encode()).hexdigest(),
+            "captcha": captcha_text,
+            'sessionId': "",
+            'refNo': f'{self._userid}-{self._get_now_time()}',
+            'deviceIdCommon': self.deviceIdCommon,
+            "ibAuthen2faString": self.FPR,
+        }
+        wasm_bytes = self._get_wasm_file()
+        data_encrypt = wasm_encrypt(wasm_bytes, payload)
+        with requests.Session() as s:
+            with s.post("https://online.mbbank.com.vn/retail_web/internetbanking/doLogin",
+                        headers=headers_default, json={"dataEnc": data_encrypt},
+                        proxies=self.proxy) as r:
+                data_out = r.json()
+        if data_out["result"]["ok"]:
+            self.sessionId = data_out["sessionId"]
+            self._userinfo = data_out
+            return
+        else:
+            raise MBBankError(data_out["result"])
+
     def _authenticate(self):
         while True:
             self._userinfo = None
             self.sessionId = None
             self._temp = {}
-            rid = f"{self._userid}-{self._get_now_time()}"
-            json_data = {
-                'sessionId': "",
-                'refNo': rid,
-                'deviceIdCommon': self.deviceIdCommon,
-            }
-            headers = headers_default.copy()
-            headers["X-Request-Id"] = rid
-            headers["Deviceid"] = self.deviceIdCommon
-            headers["Refno"] = rid
-            with requests.Session() as s:
-                with s.post("https://online.mbbank.com.vn/retail-web-internetbankingms/getCaptchaImage",
-                            headers=headers, json=json_data,
-                            proxies=self.proxy) as r:
-                    data_out = r.json()
-            img_bytes = base64.b64decode(data_out["imageString"])
-            text = self.ocr_class.process_image(img_bytes)
-            payload = {
-                "userId": self._userid,
-                "password": hashlib.md5(self._password.encode()).hexdigest(),
-                "captcha": text,
-                'sessionId': "",
-                'refNo': f'{self._userid}-{self._get_now_time()}',
-                'deviceIdCommon': self.deviceIdCommon,
-                "ibAuthen2faString": self.FPR,
-            }
-            wasm_bytes = self._get_wasm_file()
-            dataEnc = wasm_encrypt(wasm_bytes, payload)
-            with requests.Session() as s:
-                with s.post("https://online.mbbank.com.vn/retail_web/internetbanking/doLogin",
-                            headers=headers_default, json={"dataEnc": dataEnc},
-                            proxies=self.proxy) as r:
-                    data_out = r.json()
-            if data_out["result"]["ok"]:
-                self.sessionId = data_out["sessionId"]
-                self._userinfo = data_out
-                return
-            elif data_out["result"]["responseCode"] == "GW283":
-                pass
-            else:
-                err_out = data_out["result"]
-                raise Exception(f"{err_out['responseCode']} | {err_out['message']}")
+            img_bytes = self.get_capcha_image()
+            captcha_text = self.ocr_class.process_image(img_bytes)
+            try:
+                return self.login(captcha_text)
+            except MBBankError as e:
+                if e.code == "GW283":
+                    continue # capcha error, try again
+                raise e
 
     def getTransactionAccountHistory(self, *, accountNo: str = None, from_date: datetime.datetime,
                                      to_date: datetime.datetime):
@@ -391,3 +414,9 @@ def userinfo(self):
         else:
             self.getBalance()
         return self._userinfo
+
+    def getBanks(self):
+        data_out = self._req("https://online.mbbank.com.vn/api/retail_web/common/getBankList")
+        return data_out
+
+

mbbank/mbasync.py

@@ -5,7 +5,6 @@
 import hashlib
 import typing
 import aiohttp
-from .capcha_ocr import CapchaProcessing, CapchaOCR
 from .main import MBBankError, MBBank
 from .wasm_helper import wasm_encrypt
 from .main import headers_default
@@ -30,10 +29,10 @@ class MBBankAsync(MBBank):
         username (str): MBBank Account Username
         password (str): MBBank Account Password
         proxy (str, optional): Proxy url. Example: "http://127.0.0.1:8080". Defaults to None.
-        ocr_class (CapchaProcessing, optional): CapchaProcessing class. Defaults to TesseractOCR().
+        ocr_class (CapchaProcessing, optional): CapchaProcessing class. Defaults to CapchaOCR().
     """
 
-    def __init__(self, *, username: str, password: str, proxy: dict=None, ocr_class=None):
+    def __init__(self, *, username: str, password: str, proxy: dict = None, ocr_class=None):
         super().__init__(username=username, password=password, proxy=proxy, ocr_class=ocr_class)
         # convert proxy dict by requests to aiohttp format
         if len(self.proxy.values()):
@@ -50,54 +49,77 @@ async def _get_wasm_file(self):
                 self._wasm_cache = await r.read()
         return self._wasm_cache
 
+    async def get_capcha_image(self) -> bytes:
+        """
+        Get capcha image as bytes
+
+        Returns:
+            success (bytes): capcha image as bytes
+        """
+        rid = f"{self._userid}-{get_now_time()}"
+        json_data = {
+            'sessionId': "",
+            'refNo': rid,
+            'deviceIdCommon': self.deviceIdCommon,
+        }
+        headers = headers_default.copy()
+        headers["X-Request-Id"] = rid
+        headers["Deviceid"] = self.deviceIdCommon
+        headers["Refno"] = rid
+        async with aiohttp.ClientSession() as s:
+            async with s.post("https://online.mbbank.com.vn/retail-web-internetbankingms/getCaptchaImage",
+                              headers=headers, json=json_data, proxy=self.proxy) as r:
+                data_out = await r.json()
+                return base64.b64decode(data_out["imageString"])
+
+    async def login(self, captcha_text: str):
+        """
+        Login to MBBank account
+
+        Args:
+            captcha_text (str): capcha text from capcha image
+
+        Raises:
+            MBBankError: if api response not ok
+        """
+        payload = {
+            "userId": self._userid,
+            "password": hashlib.md5(self._password.encode()).hexdigest(),
+            "captcha": captcha_text,
+            'sessionId': "",
+            'refNo': f'{self._userid}-{get_now_time()}',
+            'deviceIdCommon': self.deviceIdCommon,
+            "ibAuthen2faString": self.FPR,
+        }
+        wasm_bytes = await self._get_wasm_file()
+        loop = asyncio.get_running_loop()
+        data_encrypt = await loop.run_in_executor(pool, wasm_encrypt, wasm_bytes, payload)
+        async with aiohttp.ClientSession() as s:
+            async with s.post("https://online.mbbank.com.vn/retail_web/internetbanking/doLogin",
+                              headers=headers_default, json={"dataEnc": data_encrypt}, proxy=self.proxy) as r:
+                data_out = await r.json()
+        if data_out["result"]["ok"]:
+            self.sessionId = data_out["sessionId"]
+            self._userinfo = data_out
+            return
+        else:
+            raise MBBankError(data_out["result"])
+
     async def _authenticate(self):
         while True:
             self._userinfo = None
             self.sessionId = None
             self._temp = {}
-            rid = f"{self._userid}-{get_now_time()}"
-            json_data = {
-                'sessionId': "",
-                'refNo': rid,
-                'deviceIdCommon': self.deviceIdCommon,
-            }
-            headers = headers_default.copy()
-            headers["X-Request-Id"] = rid
-            headers["Deviceid"] = self.deviceIdCommon
-            headers["Refno"] = rid
-            async with aiohttp.ClientSession() as s:
-                async with s.post("https://online.mbbank.com.vn/retail-web-internetbankingms/getCaptchaImage",
-                                  headers=headers, json=json_data, proxy=self.proxy) as r:
-                    data_out = await r.json()
-            img_bytes = base64.b64decode(data_out["imageString"])
+            img_bytes = await self.get_capcha_image()
             text = await asyncio.get_event_loop().run_in_executor(
                 pool, self.ocr_class.process_image, img_bytes
             )
-            payload = {
-                "userId": self._userid,
-                "password": hashlib.md5(self._password.encode()).hexdigest(),
-                "captcha": text,
-                'sessionId': "",
-                'refNo': f'{self._userid}-{get_now_time()}',
-                'deviceIdCommon': self.deviceIdCommon,
-                "ibAuthen2faString": self.FPR,
-            }
-            wasm_bytes = await self._get_wasm_file()
-            loop = asyncio.get_running_loop()
-            dataEnc = await loop.run_in_executor(pool, wasm_encrypt, wasm_bytes, payload)
-            async with aiohttp.ClientSession() as s:
-                async with s.post("https://online.mbbank.com.vn/retail_web/internetbanking/doLogin",
-                                  headers=headers_default, json={"dataEnc": dataEnc}, proxy=self.proxy) as r:
-                    data_out = await r.json()
-            if data_out["result"]["ok"]:
-                self.sessionId = data_out["sessionId"]
-                self._userinfo = data_out
-                return
-            elif data_out["result"]["responseCode"] == "GW283":
-                pass
-            else:
-                err_out = data_out["result"]
-                raise Exception(f"{err_out['responseCode']} | {err_out['message']}")
+            try:
+                return await self.login(text)
+            except MBBankError as e:
+                if e.code == "GW283":
+                    continue # capcha error, try again
+                raise e
 
     async def _req(self, url, *, json=None, headers=None):
         if headers is None:
@@ -275,7 +297,6 @@ async def getSavingDetail(self, accNo: str, accType: typing.Literal["OSA", "SBA"
         data_out = await self._req("https://online.mbbank.com.vn/api/retail_web/saving/getDetail", json=json_data)
         return data_out
 
-
     async def getLoanList(self):
         """
         Get all loan list from your account
@@ -362,3 +383,8 @@ async def userinfo(self):
         else:
             await self.getBalance()
         return self._userinfo
+
+    async def getBanks(self):
+        data_out = await self._req("https://online.mbbank.com.vn/api/retail_web/common/getBankList")
+        return data_out
+

mbbank/wasm_helper/__init__.py

@@ -14,7 +14,11 @@
 class globalThis:
     def __init__(self):
         self.exports = undefined
-        self.window = dict_warper({"document": dict_warper({})})
+        self.window = dict_warper({
+            "document": {
+                "we_love_mb": True # From CookieGMVN Library :)
+            }
+        })
         self.fs = fs_object()
         self.process = process_object()
         self.location = dict_warper({"origin": "https://online.mbbank.com.vn"})
@@ -137,14 +141,7 @@ def exit_process(cls, exitCode):
         print("exit code:", exitCode)
 
     def importObject(self, imports_type: list[wasmtime.ImportType]):
-        def proxy(name):
-            def fn(*args, **kwargs):
-                call = getattr(self.go_js, name)
-                return call(*args, **kwargs)
-
-            return fn
-
-        return [wasmtime.Func(self.wasm_store, i.type, proxy(i.name)) for i in imports_type]
+        return [wasmtime.Func(self.wasm_store, i.type, getattr(self.go_js, i.name)) for i in imports_type]
 
     # noinspection PyAttributeOutsideInit
     def run(self, inst):
@@ -213,7 +210,7 @@ def _resume(self):
 
     def _makeFuncWrapper(self, ids):
         def wrapper(*args, **kwargs):
-            event = dict_warper({"id": int(ids), "args": hash_list(args), "this": global_this})
+            event = dict_warper({"id": int(ids), "args": args, "this": global_this})
             self._pendingEvent = event
             self._resume()
             return event.result
@@ -372,7 +369,7 @@ def sysjs_copyBytesToJS(self, *args):
     def debug(self, *args):
         pass
 
-
+# All instances shared the same wasm module
 def wasm_encrypt(wasm_files, json_data):
     if getattr(global_this, 'bder', None) is not None:
         return global_this.bder(json.dumps(json_data), "0")

mbbank/wasm_helper/helper.py

@@ -214,6 +214,10 @@ def getegid(self):
 class dict_warper:
     def __init__(self, dict_data):
         for key, value in dict_data.items():
+            if isinstance(value, dict):
+                value = dict_warper(value)
+            elif isinstance(value, list) or isinstance(value, tuple):
+                value = hash_list(value)
             setattr(self, key, value)
 
     def to_dict(self):

tests/async_test.py

@@ -9,6 +9,7 @@ async def main():
     mb = MBBankAsync(username=os.getenv("MBBANK_USERNAME"), password=os.getenv("MBBANK_PASSWORD"))
     end_query_day = datetime.datetime.now()
     start_query_day = end_query_day - datetime.timedelta(days=30)
+    await mb.getBanks()
     await mb.getBalance()
     await mb.userinfo()
     await mb.getInterestRate()

tests/sync_test.py

@@ -5,6 +5,7 @@
 mb = MBBank(username=os.getenv("MBBANK_USERNAME"), password=os.getenv("MBBANK_PASSWORD"))
 end_query_day = datetime.datetime.now()
 start_query_day = end_query_day - datetime.timedelta(days=30)
+mb.getBanks()
 mb.getBalance()
 mb.userinfo()
 mb.getInterestRate()