<reponame>MrT3acher/Android-Malware-Sandbox
import subprocess
import logging


class Device:

    def __init__(self, path_config, configuration, type):
        self.adb_path = path_config.get('adb_path')
        self.emulator_path = path_config.get('emulator_path')

        self.type = type
        self.configuration = configuration
        self.device_id = None

    def list_devices(self):
        '''
        List current device recognized by adb and sort them by type (emulator|physical)
        :return:
        '''
        logging.debug("Device:listing_devices()")

        # Get list of launched devices
        pid = subprocess.Popen([self.adb_path, 'devices'], stdout=subprocess.PIPE)
        pid.wait()

        raw_result = pid.communicate()[0].strip()
        devices = raw_result.split(b'\n')[1:]

        result = {
            'emulator': [],
            'physical': []
        }

        for device in devices:
            name, state = device.split(b'\t')
            if name.startswith(b'emulator'):
                result['emulator'].append(name.decode('utf8'))
            else:
                result['physical'].append(name.decode('utf8'))

        return result

    def check_is_up(self):
        '''
        Return if the phone has booted up.
        It returns "true" when you the lockscreen is unlocked
        :return:
        '''
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'getprop', 'dev.bootcomplete'],
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
        pid.wait()
        return b'1' in pid.communicate()[0]

    def kill_emulators(self):
        '''
        Kill all current emulators
        :return:
        '''
        logging.debug("Device:kill_emulators()")
        devices = self.list_devices()

        # Killing all running devices
        for device in devices['emulator']:
            pid = subprocess.Popen([self.adb_path, '-s', device, 'emu', 'kill'], stdout=subprocess.PIPE)
            pid.wait()

    def check_application_is_running(self, package_name):
        '''
        Verify that an application is launched ( Android > 7 )
        :param package_name:
        :return:
        '''
        logging.debug('Device:check_application_is_running()')

        pid = subprocess.Popen([self.adb_path,'-s', self.device_id, 'shell', 'pidof', package_name], stdout= subprocess.PIPE)
        pid.wait()
        return b'' == pid.communicate()[0]

    def install_application(self, path, grant_permissions=False):
        '''
        Install the given application on the device
        :param path:
        :return:
        '''
        pid = None
        if(grant_permissions):
            logging.debug("Device:install_application(grant_permissions=True)")
            pid = subprocess.Popen([self.adb_path,'-s',self.device_id, 'install', '-g', path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        else:
            logging.debug("Device:install_application()")
            pid = subprocess.Popen([self.adb_path,'-s',self.device_id, 'install', path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        
        try:
            pid.wait(timeout=30)
        except:
            pass

    def install_frida(self):
        '''
        Install Frida using `frida-push`
        :return:
        '''
        logging.debug("Device:install_frida()")
        if(self.type == 'Physical'):
            logging.info("You must launch frida manually")
        else:
            pid = subprocess.Popen(['frida-push', '-d', self.device_id],
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE)
            pid.wait()
            stdout, stderr = pid.communicate()
            if stderr != b'':
                logging.error(f"Error while installing frida : {stderr.decode()}")
            else:
                logging.info(f'Frida Installed')

    def install_certificate(self, certificate_path):
        logging.debug("Device:install_certificate()")
        if(self.type == "Physical"):
            logging.info("You need to install the certificate manually ( go to http://mitm.it ) ")



    def uninstall_application(self,package_name):
        '''
        Uninstal the application given by the package name
        :param package_name:
        :return:
        '''
        import frida
        logging.debug("Device:uninstall_application()")


        '''
        This part is used to uninstall even admin applications
        '''
        device = frida.get_device(self.device_id)
        session_system_server =  device.attach("system_server")

        # see : https://cs.android.com/android/platform/superproject/+/master:frameworks/base/services/core/java/com/android/server/pm/PackageManagerService.java?q=DELETE_FAILED_DEVICE_POLICY_MANAGER&ss=android%2Fplatform%2Fsuperproject
        script_system_server = session_system_server.create_script("""
        Java.perform(function(){
            const PackageManager = Java.use("com.android.server.pm.PackageManagerService")

            PackageManager.isPackageDeviceAdmin.implementation = function(packageName, userId){
                return false;
            }
        })
        """)
        
        def message(message, data):
            logging.debug(message)


        script_system_server.on("message", message)
        script_system_server.load()

        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'uninstall', package_name],
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
        pid.wait()

        script_system_server.unload()
        stdout, stderr = pid.communicate()
        if stderr != b'':
            logging.error(f"Error while installing application {package_name} : {stderr}")
        else:
            logging.info(f'Uninstalled')

    def launch_application(self, package_name):
        '''
        Launch an application using adb monkey
        :param package_name:
        :return:
        '''
        logging.debug("Device:launch_application()")
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'monkey', '-p', package_name, '-v', '1'], stdout=subprocess.PIPE)
        pid.wait()
        stdout, stderr = pid.communicate()
        if stderr:
            logging.error(f"Error while launching application {package_name} : {stderr}")
        else:
            logging.debug(f'Launched !')

    def start(self):
        logging.error("Device:start() should not be called on the superclass")

    def get_device_arch(self):
        """ This function tries to determine the architecture of the device
        :returns either "arch" or None.
        """
        logging.debug("Device:get_device_arch()")

        arch = None
        pid = subprocess.Popen([self.adb_path,'-s', self.device_id, 'shell','getprop','ro.product.cpu.abi'],stdout=subprocess.PIPE)
        pid.wait()
        output = pid.communicate()[0].decode().strip()
        getprop_archs = ["armeabi", "armeabi-v7a", "arm64-v8a", "x86", "x86_64"]
        if output in getprop_archs:
            if output in ["armeabi", "armeabi-v7a","arm64-v8a"]:
                arch = "arm"
            else:
                arch = output

        return arch
    
    def list_third_party(self):
        """
        List all third party applications
        :return list(application): the list of application
        """
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'pm', 'list', 'packages', '-3'],
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
        pid.wait()
        stdout, stderr = pid.communicate()
        if(stderr != b''):
            log.error(stderr)
        else:
            return stdout.decode().replace('package:','').strip().split('\n')

    def list_third_party(self):
        """
        List all third party applications
        :return list(application): the list of application
        """
        logging.debug("Device:list_third_party()")
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'pm', 'list', 'packages', '-3'],
                                stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
        pid.wait()
        stdout, stderr = pid.communicate()
        if(stderr != b''):
            log.error(stderr)
        else:
            return stdout.decode().replace('package:','').strip().split('\n')
    
    def get_root_shell(self):
        """
        Get a root shell on the emulator or physical device
        """
        logging.debug("Device:get_root_shell()")

        pid = subprocess.Popen([self.adb_path, 'root'],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE)
        pid.wait()
        stdout,stderr = pid.communicate()
        if(stderr != b''):
            logging.error(stderr) 

    def pull_application_internal_files(self, package_name, out_directory):
        """
        Pull the internal files from a given application to a given directory
        :param package_name: name of the application to pull files from
        :param out_directory: name of directory where to save files
        """
        logging.debug("Device:pull_application_internal_files()")

        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'pull', f'/data/data/{package_name}', out_directory],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
        pid.wait()
        stdout,stderr = pid.communicate()
        if(stderr != b''):
            logging.error(stderr)
    
    def enable_accessibility_services(self, application):
        """
        Enable accesibility for all services that need services
        :param application: application to enable services for
        """
        logging.debug("Device:enable_accessibility_services()")

        for service in application.get_accessibility_services():
            # shell settings put secure enabled_accessibility_services packagname/servicename
            pid = subprocess.Popen([self.adb_path, "-s", self.device_id, "shell", "settings", "put", "secure", "enabled_accessibility_services", f"{application.package}/{service}"])
            pid.wait()

    def rm_tcpdump_file(self, path):
        """
        Remove tcpdump file on emulator
        """
        logging.debug("Device:rm_tcpdump_file()")
        rm_file_pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'rm', path],
                                          stdout=subprocess.DEVNULL)


    def start_tcpdump(self, path, ports):
        """
        Start tcpdump on emulator
        :param path: path for the output file for tcpdump
        :param ports: ignore the ports
        """
        logging.debug("Device:start_tcpdump()")

        if len(ports) > 0:
            option = "'"
            for i in range(len(ports)):
                if i == (len(ports)-1):
                    option += f" not port {ports[i]}'"
                else:
                    option += f" not port {ports[i]} and"
        else:
            option = ""

        try:
            tcpdump_pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'tcpdump', '-nn', '-w ', path, option, '&'],stdout=subprocess.DEVNULL)
        except Exception as e:
            logging.error(e)


    def get_tcpdump_file(self, path, out_directory):
        """
        Get tcpdump file
        :param path: path of the tcpdump file
        :param out_directory: directory in which the file will be stored
        """
        logging.debug("Device:get_tcpdump_file()")
        pid_kill = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'pkill', 'tcpdump'])
        pid_kill.wait()
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'pull', path, out_directory])
        pid.wait()
        stdout,stderr = pid.communicate()
        if(stderr != b''):
            logging.error(stderr)
    
    def start_screenrecord(self, path):
        """
        Record the screen of the emulator
        :param path: path of the screenrecord file
        """
        logging.debug("Device:start_screenrecord()")
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'screenrecord', '--size 1280x720', path, '&'])
        # pid.wait()
    
    def get_screenrecord_file(self, path, out_directory):
        """
        Get screenrecord file
        :param path: path of the screenrecord file
        :param out_directory: directory in which the file will be stored
        """
        logging.debug("Device:get_screenrecord_file()")
        pid_kill = subprocess.Popen([self.adb_path, '-s', self.device_id, 'shell', 'pkill', '-l 2 ', '-V', 'screenrecord'])
        pid_kill.wait()
        pid = subprocess.Popen([self.adb_path, '-s', self.device_id, 'pull', path, out_directory])
        pid.wait()
        stdout,stderr = pid.communicate()
        if(stderr != b''):
            logging.error(stderr)

    def stop(self):
        pass