intercept code loading

frida/theseus_frida/__init__.py

@@ -1,5 +1,7 @@
 import argparse
+import base64
 import os
+import hashlib
 import subprocess
 import time
 import json
@@ -16,7 +18,7 @@ STACK_CONSUMER_B64 = Path(__file__).parent / "StackConsumer.dex.b64"
 
 
 # Define handler to event generated by the scripts
-def on_message(message, data, data_storage: dict):
+def on_message(message, data, data_storage: dict, file_storage: Path):
     if message["type"] == "error":
         print(f"[error] {message['description']}")
         print(message["stack"])
@@ -26,6 +28,8 @@ def on_message(message, data, data_storage: dict):
         handle_class_new_inst_data(message["payload"]["data"], data_storage)
     elif message["type"] == "send" and message["payload"]["type"] == "cnstr-new-isnt":
         handle_cnstr_new_inst_data(message["payload"]["data"], data_storage)
+    elif message["type"] == "send" and message["payload"]["type"] == "load-dex":
+        handle_load_dex(message["payload"]["data"], data_storage, file_storage)
     else:
         print("[on_message] message:", message)
 
@@ -130,6 +134,43 @@ def handle_cnstr_new_inst_data(data, data_storage: dict):
     )
 
 
+def handle_load_dex(data, data_storage: dict, file_storage: Path):
+    dex = data["dex"]
+    classloader_class = data["classloader_class"]
+    classloader = data["classloader"]
+    short_class = classloader_class.split("/")[-1].removesuffix(";")
+    files = []
+    print("DEX file loaded:")
+    print(f"    by:     {classloader_class} ({classloader})")
+    for file in dex:
+        print(f"{file=}")
+        file_bin = base64.b64decode(file)
+        hasher = hashlib.sha1()
+        hasher.update(file_bin)
+        h = hasher.digest().hex()
+        print(f"    hash:   {h}")
+        fname = (
+            file_storage / f"{short_class}_{classloader}_{h[:16]}.bytecode"
+        )  # not .dex, can also be .jar or .apk or .oat or ...
+        i = 1
+        while fname.exists():
+            fname = file_storage / f"{short_class}_{classloader}_{h[:16]}_{i}.bytecode"
+            i += 1
+        fname = fname.absolute().resolve()
+
+        with fname.open("wb") as fp:
+            fp.write(file_bin)
+        print(f"    stored: {str(fname)}")
+        files.append(str(fname))
+    data_storage["dyn_code_load"].append(
+        {
+            "classloader_class": classloader_class,
+            "classloader": classloader,
+            "files": files,
+        }
+    )
+
+
 def main():
     parser = argparse.ArgumentParser(
         prog="Android Theseus project",
@@ -151,9 +192,23 @@ def main():
         help="where to dump the collected data, default is stdout",
         type=Path,
     )
+    parser.add_argument(
+        "-d",
+        "--dex-dir",
+        default=Path("."),
+        help="where to store dynamically loaded bytecode",
+        type=Path,
+    )
     args = parser.parse_args()
     env = dict(os.environ)
 
+    file_storage = args.dex_dir
+    if not file_storage.exists():
+        file_storage.mkdir(parents=True)
+    if not file_storage.is_dir():
+        print("--dex-dir must be a directory")
+        exit()
+
     if args.device != "":
         device = frida.get_device(args.device)
         env["ANDROID_SERIAL"] = args.device
@@ -177,17 +232,26 @@ def main():
 
     pid = device.spawn([app])
     session = device.attach(pid)
-    script = session.create_script(script)
+    try:
+        script = session.create_script(script)
+    except frida.InvalidArgumentError as e:
+        print(
+            "\n".join(
+                map(lambda v: f"{v[0]+1: 3} {v[1]}", enumerate(script.split("\n")))
+            )
+        )
+        raise e
 
     data_storage = {
         "invoke_data": [],
         "class_new_inst_data": [],
         "cnstr_new_inst_data": [],
+        "dyn_code_load": [],
     }
 
     script.on(
         "message",
-        lambda msg, data: on_message(msg, data, data_storage),
+        lambda msg, data: on_message(msg, data, data_storage, file_storage),
     )
 
     # Load script