Initial Commit

.gitignore

@@ -0,0 +1,6 @@
+cache/
+dumps/
+report/
+todo.txt
+__pycache__
+*.class

analyzer.py

@@ -0,0 +1,70 @@
+import os
+import sys
+import json
+import shutil
+
+from utils import Utils
+from modules import packages
+
+class Analyzer:
+    def __init__(self, folder):
+        self.folder = folder
+        self.dumps = Utils.list_files(folder, ".tar.gz")
+        self.internal_path = None
+        self.external_path = None
+        self.app_id = self.app_id_parser()
+
+        self.cache_path = os.path.join(sys.path[0], "cache")
+        self.report_path = os.path.join(sys.path[0], "report")
+        try:
+            Utils.remove_folder(self.cache_path)
+        except:
+            pass
+        Utils.check_and_generate_folder(self.cache_path)
+
+    def app_id_parser(self):
+        app_id = None
+        for name in self.dumps:
+            if '_internal.tar.gz' in name:
+                self.internal_path = name
+                app_id = os.path.basename(name.split('_internal.tar.gz')[0])
+            elif '_external.tar.gz' in name:
+                self.external_path = name
+                app_id = os.path.basename(name.split('_external.tar.gz')[0])
+
+        return app_id
+
+    def generate_report(self):
+        module_file = packages.get(self.app_id)
+        if not module_file:
+            print("[Analyzer] Module not found for {}".format(self.app_id))
+            return None
+
+        #tiktok   
+        m = __import__("modules.{}".format(module_file), fromlist=[None])
+        module = m.Module(self.internal_path, self.external_path)
+        user_id = module.get_user_id()
+        report_header = {
+            "report_date": Utils.get_current_time(),
+            "user": user_id
+        }
+        report = {}
+
+        report["header"] = report_header
+        report["profile"] = module.get_user_profile()
+        report["messages"] = module.get_user_messages()
+        report["users"] = module.get_user_profiles()
+        report["searches"] = module.get_user_searches()
+        report["videos"] = module.get_videos()
+        report["freespace"] = module.get_undark_db()
+        # with open("report/"+ name+"_freespace.txt",'w') as f:
+        #         output = Utils.run_undark("./cache/tiktok/internal/databases/" + name)            
+        #         print(output)
+        f = open(os.path.join(self.report_path, "REPORT_{}.json".format(user_id)), "w")
+        f.write(json.dumps(report, indent=2))
+        f.close()
+        # print(module.get_user_profiles())
+        # print(module.get_user_id())
+        # print(module.get_user_searches())
+        # print(module.get_user_profile())
+        # print(module.get_user_messages())

database.py

@@ -0,0 +1,37 @@
+import sqlite3
+import subprocess
+
+class DatabaseParser:
+    #https://svn.python.org/projects/python/trunk/Lib/sqlite3/dump.py
+    @staticmethod
+    def dump_columns(database):
+        tables = {}
+        #print("Reading {}".format(database))
+        connection = sqlite3.connect(database)
+        cu = connection.cursor()
+
+        #Tables
+        q = """SELECT name, type, sql FROM sqlite_master WHERE sql NOT NULL AND type == 'table'"""
+        schema_res = cu.execute(q)
+        for table_name, type, sql in schema_res.fetchall():
+            tables[table_name] = []
+            if table_name.startswith('sqlite_'):
+                continue
+
+            res = cu.execute("PRAGMA table_info('%s')" % table_name)
+            column_names = [str(table_info[1]) for table_info in res.fetchall()]
+            for col in column_names:
+                tables[table_name].append(col)
+
+        #'index', 'trigger', or 'view'
+        '''
+        q = """SELECT name, type, sql FROM sqlite_master WHERE sql NOT NULL AND type IN ('index', 'trigger', 'view')"""
+        schema_res = cu.execute(q)
+        for name, type, sql in schema_res.fetchall():
+            print('%s;' % sql)
+            #pass
+        '''
+
+        return tables
+
+

dependencies/source.txt

@@ -0,0 +1,4 @@
+Instalar ambos os utilitários como variáveis de sistema para obter sucesso na extracção
+adb -> https://developer.android.com/studio/releases/platform-tools
+base64 -> http://rtner.de/software/base64.html
+undark -> https://github.com/witwall/undark/releases

device.py

@@ -0,0 +1,8 @@
+from ppadb.client import Client as AdbClient
+
+class DeviceCommunication:
+    def __init__(self):
+        self.client = AdbClient(host="127.0.0.1", port=5037)
+
+    def list_devices(self):
+        return self.client.devices()

extract.py

@@ -0,0 +1,105 @@
+import subprocess
+import datetime
+import os
+import sys
+
+from utils import Utils
+from device import DeviceCommunication
+
+class Extract:
+    def __init__(self):
+        self.internal_data_path = "/data/data/{}"
+        self.external_data_path = "/sdcard/Android/data/{}"
+        self.internal_data_dump_name = "{}_internal.tar.gz"
+        self.external_data_dump_name = "{}_external.tar.gz"
+
+        self.dumps_path = os.path.join(sys.path[0], "dumps")
+
+        Utils.check_and_generate_folder(self.dumps_path)
+
+    def dump_from_adb(self, app_package):
+        folders = []
+
+        device_communication = DeviceCommunication()
+        for device in device_communication.list_devices():
+            serial_number = device.get_serial_no()
+
+            current_time = Utils.get_current_time()
+            path_dump_folder = os.path.join(self.dumps_path, current_time)
+            path_dump_internal = os.path.join(path_dump_folder, self.internal_data_dump_name.format(app_package))
+            path_dump_external = os.path.join(path_dump_folder, self.external_data_dump_name.format(app_package))
+
+            adb_location = Utils.get_adb_location()
+            base64_location = Utils.get_base64_location()
+
+            #Check if we have dump folder
+            Utils.check_and_generate_folder(path_dump_folder)
+
+            #Dump internal data https://android.stackexchange.com/questions/85564/need-one-line-adb-shell-su-push-pull-to-access-data-from-windows-batch-file
+            print("[{}] Extracting internal app (root) data!".format(serial_number))
+
+            command = """{} -s {} shell "su -c 'cd {} && tar czf - ./ --exclude='./files'| base64' 2>/dev/null" | {} -d > {}""".format(adb_location, serial_number, self.internal_data_path.format(app_package), base64_location, path_dump_internal)
+            subprocess.Popen(command, shell=True, stdout=subprocess.PIPE).stdout.read()
+
+            #Clean the file if it's empty
+            if os.path.getsize(path_dump_internal) == 0:
+                print("[{}] Nothing extracted!".format(serial_number))
+                try:
+                    os.remove(path_dump_internal)
+                except:
+                    pass
+            else:
+                print("[{}] File generated! {}".format(serial_number, path_dump_internal))
+
+            #Dump external
+            print("[{}] Extracting external app data!".format(serial_number))
+
+            command = """{} -s {} shell "su -c 'cd {} && tar czf - ./ | base64' 2>/dev/null" | {} -d > {}""".format(adb_location, serial_number, self.external_data_path.format(app_package), base64_location, path_dump_external)
+            subprocess.Popen(command, shell=True, stdout=subprocess.PIPE).stdout.read()
+
+            if os.path.getsize(path_dump_external) == 0:
+                print("[{}] Nothing extracted!".format(serial_number))
+                try:
+                    os.remove(path_dump_external)
+                except:
+                    pass
+            else:
+                print("[{}] File generated! {}".format(serial_number, path_dump_external))
+
+            #Generated folders
+            folders.append(path_dump_folder)
+
+        return folders
+
+    def dump_from_path(self, base_path, app_package):
+        base_path = Utils.replace_slash_platform(base_path)
+
+        if not os.path.exists(base_path):
+            print("[Dump] Dump from path failed: {} doesn't exists".format(base_path))
+            return None
+
+        current_time = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+        path_dump_folder = os.path.join(self.dumps_path, current_time)
+        path_dump_internal = os.path.join(path_dump_folder, self.internal_data_dump_name.format(app_package))
+        path_dump_external = os.path.join(path_dump_folder, self.external_data_dump_name.format(app_package))
+
+        #Check if we have dump folder
+        Utils.check_and_generate_folder(path_dump_folder)
+
+        #Extract internal data from mount
+        path_original_internal = Utils.replace_slash_platform(os.path.join(base_path, self.internal_data_path.format(app_package)[1:])) #clean first / to allow concat
+        if os.path.exists(path_original_internal):
+            print("[Dump] Extracting internal app data!")
+            Utils.generate_tar_gz_file(path_original_internal, path_dump_internal)
+        else:
+            print("[Dump] Internal app folder {} doesn't exist".format(path_original_internal))
+
+        #Extract external data from mount
+        path_original_external = Utils.replace_slash_platform(os.path.join(base_path, self.external_data_path.format(app_package)[1:]))
+        if os.path.exists(path_original_external):
+            print("[Dump] Extracting external app data!")
+            Utils.generate_tar_gz_file(path_original_external, path_dump_external)
+        else:
+            print("[Dump] External app folder {} doesn't exist".format(path_original_external))        
+
+        return [path_dump_folder]

modules/__init__.py

@@ -0,0 +1,3 @@
+packages = {
+    "com.zhiliaoapp.musically": "tiktok"
+}