update SSL pinner

create utils helper

pyhtools/attackers/Android/mitm/cert_pin.py

@@ -1,10 +1,13 @@
 from ppadb.client import Client
 from ppadb.device import Device
 from os.path import isfile, basename
-from textwrap import dedent
+from os import system
+from . import utils
 
+import asyncio
 import frida
 import logging
+import threading
 logging.basicConfig(level=logging.DEBUG,
                     format='[%(asctime)s] [%(levelname)s] - %(message)s')
 
@@ -21,16 +24,16 @@ class NoDevicesFound(Exception):
 
 
 class PinCertificate:
-    def __init__(self, apk_path: str,  package_name: str, cert_path: str, frida_binary_path: str, frida_script_path: str, device_name: str, host: str = '127.0.0.1', port: int = 5037, apk_installed: bool = False,):
+    def __init__(self, apk_path: str,  package_name: str, cert_path: str, frida_binary_path: str, frida_script_path: str, device_name: str, host: str = '127.0.0.1', port: int = 5037,):
         # check data types
         assert type(apk_path) == str
         assert type(package_name) == str
         assert type(cert_path) == str
         assert type(device_name) == str
         assert type(frida_binary_path) == str
+        assert type(frida_script_path) == str
         assert type(host) == str
         assert type(port) == int
-        assert type(apk_installed) == bool
 
         # check if files are available at their paths
         if not isfile(apk_path):
@@ -50,26 +53,26 @@ def __init__(self, apk_path: str,  package_name: str, cert_path: str, frida_bina
         # assign values
         self.__device_name = device_name
         self.__apk_path = apk_path
-        self.__apk_installed = apk_installed
         self.__package_name = package_name
         self.__cert_path = cert_path
         self.__frida_path = frida_binary_path
-
-        # set initial values
-        self.device = None
+        self.__frida_script_path = frida_script_path
 
         # connect to adb server
         self._adb = Client(
             host=host,
             port=port
         )
 
+        # set initial values
+        self.device = self.get_device()
+
     def get_device(self):
-        self.devices()
+        _ = self.get_adb_devices()
         device: Device = self._adb.device(self.__device_name)
         return device
 
-    def devices(self):
+    def get_adb_devices(self):
         try:
             devices: list[Device] = self._adb.devices()
             if len(devices) == 0:
@@ -89,18 +92,34 @@ def get_frida_devices(self):
 
         return devices
 
+    def install_apk(self, force_install: bool = True):
+        if self.device.is_installed(self.__package_name) and force_install:
+            self.device.uninstall(self.__package_name)
+
+        self.device.install(self.__apk_path)
+
+        if self.device.is_installed(self.__package_name):
+            return True
+        return False
+
+    def start_frida(self):
+        asyncio.run(utils.run(f'adb shell /data/local/tmp/frida-server &'))
+
     def pin_certificate(self):
         logging.info("Starting Certificate Pinning Procedure..")
 
         # get device
         self.device: Device = self.get_device()
         logging.info(f'Connected to {self.__device_name} device')
 
-        # install certificate
-        if not self.__apk_installed:
-            self.device.install(path=self.__apk_path, reinstall=True)
+        # install apk
+        logging.info(f'Installing package')
+        if self.install_apk():
             logging.info(
-                f'{basename(self.__apk_path)} APK installation completed')
+                f'{basename(self.__apk_path)} APK installation completed successfully')
+        else:
+            logging.error(
+                f'{basename(self.__apk_path)} APK installation failed!')
 
         # push certificate to the device
         self.device.push(
@@ -120,22 +139,12 @@ def pin_certificate(self):
         logging.info(
             f'{self.__frida_path} frida binary pushed to /data/local/tmp/frida-server')
 
-        # start frida server
+        # start frida server in different thread
         logging.info("Starting Frida server")
-        self.device.shell('/data/local/tmp/frida-server &')
-
-
-
-if __name__ == '__main__':
-    pinner = PinCertificate(
-        apk_path=r'apk-path',
-        package_name=r'com.app.package',
-        cert_path=r'burp_pro_cert.der',
-        frida_binary_path=r'frida-server-15.1.28-android-x86',
-        frida_script_path=r'bypass-ssl-pinning.js',
-        device_name='emulator-5554',
-        host='127.0.0.1',
-        port=5037,
-    )
+        frida_thread = threading.Thread(target=self.start_frida)
+        frida_thread.start()
+        # self.device.shell('su /data/local/tmp/frida-server &')
 
-    pinner.pin_certificate()
+        # Start SSL pinning
+        system(
+            f'frida -U -l {self.__frida_script_path} --no-paus -f {self.__package_name}')

pyhtools/attackers/Android/mitm/utils.py

@@ -0,0 +1,16 @@
+import asyncio
+
+
+async def run(cmd):
+    proc = await asyncio.create_subprocess_shell(
+        cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE)
+
+    stdout, stderr = await proc.communicate()
+
+    # print(f'[{cmd!r} exited with {proc.returncode}]')
+    if stdout:
+        print(f'[stdout]\n{stdout.decode()}')
+    if stderr:
+        print(f'[stderr]\n{stderr.decode()}')