Now I build a Python script to execute automaticaly a dynamic analysis on apk and I block because I use this library and call objection.

When I run my script, the objection command creates an instruction on my script and I can't execude the remaining code because this CLI command waits for an instruction. But I should generate activity using monkey runner when this command is active.

I tried with subprocess.run, popen and thread and I have the same result each time. Do you have a suggestion to solve this?

That's what my console shows:

My code:

import threading
from numpy import string_
import virtualbox
from androguard.core.bytecodes import apk
from linux_version.StaticForecasting import StaticForecasting
from linux_version.TargetPoint import TargetPoint
from objection import *
import sys
import subprocess
import os

analyse = StaticForecasting()


class DynamicTestControler():

    vbox = virtualbox.VirtualBox()
    session = virtualbox.Session()
    statement = {"VMCloned" : False, "VMClonedRuning" : False, "VMClonedDeleted" : False}
    virtualMachine = {"Name" : 'Android-V9-V2'}
    classTargeted = []
    objectionParseReport = []
    objectionReport = []
    verif = {"InstallAPP" : False, "ConnectFrida" : False}


    def __init__(self) -> None:
        #Validation faites
        #Récupération des classes à interroger
        data = open("/opt/Plateforme/linux_version/analysetmp/classlist_targetable.txt" , "r")
        for i in data.readlines():
            self.classTargeted.append(i)

        pass
    
    def runVM(self):
        #Validation faites
        #Intégration avec la librairie virtualbox
        
        machine = self.vbox.find_machine(self.virtualMachine["Name"])
        progress = machine.launch_vm_process(self.session, "gui", [])
        progress.wait_for_completion()
    
        return print("L'instance a été démarrée")

    def stopVM(self):
        #Validation faites
        #Permet d'éteindre l'instance de virtual de box
        self.session.console.power_down()
        return print("L'instance a été arrêtée")

    def install_app(self, analysestat:StaticForecasting):

        #Validation faites
        cmd1 = ["adb", "connect", "192.168.56.101"]
        subprocess.run([cmd1[0],cmd1[1],cmd1[2]])
        cmd2 = ["adb", "install", analysestat.apkChemin]
        subprocess.run([cmd2[0],cmd2[1],cmd2[2]])
        self.verif["InstallAPP"] = True

        pass

    def fridav2(self):
        cmd1 = ["adb", "connect", "192.168.56.101"]
        subprocess.run([cmd1[0],cmd1[1],cmd1[2]])
        cmd = ["frida-ps" , "-U"]
        subprocess.run([cmd[0],cmd[1]],shell=False)
        self.verif["ConnectFrida"] = True
        pass

    def connect_frida(self):
        
        #Validation faites
        #Attention faire cette installation pour adb sudo apt-get install android-tools-adb android-tools-fastboot adb
        cmd1 = ["adb", "connect", "192.168.1.20"]
        subprocess.run([cmd1[0],cmd1[1],cmd1[2]])
        cmd = ["adb", "push", "/opt/Plateforme/linux_version/tools/fridaserver/frida_server", "/data/local/tmp"]
        subprocess.run([cmd[0],cmd[1],cmd[2],cmd[3]])
        cmd2 = ["adb", 'shell', "chmod 755 /data/local/tmp/frida_server"]
        subprocess.run([cmd2[0],cmd2[1],cmd2[2]]) 
        cmd3 = ["adb", "shell", "su root /data/local/tmp/frida_server &"]
        subprocess.run([cmd3[0],cmd3[1],cmd3[2]],shell=False)
        #cmd5 = ["frida", "-U", "-f", analysestat.apkNamePackage]       #Facultatif car utilisation d'objection à conserver pour la thèse
        #subprocess.run([cmd5[0],cmd5[1],cmd5[2],cmd5[3]])
        self.verif["ConnectFrida"] = True

        pass


    def run_objection(self, analysestat:StaticForecasting,classetargeted):
        
        #Validation faites
        #Passage par le CLI
        #Initialiser la séquence d'écoute avec objection
        apk_packagename = analysestat.apkNamePackage
        #Récupération et clean du nom de classe
        tmp = classetargeted.replace("/",".") 
        class_targeted = tmp[1:]                                
        nomficher = "/opt/Plateforme/linux_version/analysetmp/" + class_targeted + "_result.txt"
        self.objectionReport.append(nomficher)
        filetemp = open(nomficher, "w+")
        cmdobj = "'" + "android hooking watch class " + str(class_targeted) + " --dump-args --dump-backtrace --dump-return" + "'"
        tmpcpm = "objection --gadget " + apk_packagename + " explore --quiet --startup-command " + cmdobj #+ " &"
        cmd = ["su" ,"-c", tmpcpm]
        cmd2= ["objection", "--gadget", apk_packagename,"explore", "--startup-command", cmdobj]
        z = subprocess.Popen([cmd[0],cmd[1],cmd[2]],stdout=filetemp)
        filetemp.close()
        #subprocess.run(["rm opt/Plateforme/linux_version/driven/tmp"])
        return z.pid

    def run_monkeyrunner(self, analysestat:StaticForecasting):
        
        #Validation faites  
        apk_packagename = analysestat.apkNamePackage
        cmd = ["adb", "shell", "monkey", "-p", apk_packagename, "-v", "500" ] #Pist add -pct-syskeys
        subprocess.run([cmd[0],cmd[1],cmd[2],cmd[3],cmd[4],cmd[5],cmd[6]])
        
        return print("L'activité de test a été générée !")

    def convert_objection_report(self,fichier:str):
        
        #Validation Faites
        #Récupération des données
        fichierStockage = fichier.replace("_result.txt", "_converted.txt")
        self.objectionParseReport.append(fichierStockage)
        data = open(fichier, "r")
        contenu = data.readlines()
        data.close()
        data2 = open(fichierStockage, "w+")
        for i in contenu :
            if(i[0:9] == "(agent) ["):
                appeltmp = i[(i.find("d",0)+2):]
                tmp = i.split(" ")
                line_convert = tmp[0] + ";" + tmp[1] + ";" + appeltmp
                data2.write(line_convert)
        data2.close()
        pass    
    pass    

posted about 3 years ago

CC BY-SA 4.0

3y ago by Alexei‭

Link213‭

2 reputation 1 0 -3 2