these-android-re/android_of_theseus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

wip

Browse source
This commit is contained in:
Jean-Marie Mineau 2025-03-31 17:32:23 +02:00
parent f8c70b576b
commit a374769389
Signed by: histausse
GPG key ID: B66AEEDA9B645AD2
6 changed files with 163 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

152
frida/theseus_frida/__init__.py
View file

@ -6,8 +6,11 @@ import subprocess
import time import time
import json import json
import sys import sys
import tempfile
import shutil
import lzma
from pathlib import Path from pathlib import Path
from typing import TextIO from typing import TextIO, Any
import frida # type: ignore import frida # type: ignore
from androguard.core.apk import get_apkid # type: ignore from androguard.core.apk import get_apkid # type: ignore
@ -23,6 +26,12 @@ STACK_CONSUMER_B64 = Path(__file__).parent / "StackConsumer.dex.b64"
HASH_NB_BYTES = 4 HASH_NB_BYTES = 4
def spinner(symbs: str = "-\\|/"):
while True:
for s in symbs:
yield s
# Define handler to event generated by the scripts # Define handler to event generated by the scripts
def on_message(message, data, data_storage: dict, file_storage: Path): def on_message(message, data, data_storage: dict, file_storage: Path):
if message["type"] == "error": if message["type"] == "error":
@ -36,6 +45,8 @@ def on_message(message, data, data_storage: dict, file_storage: Path):
handle_cnstr_new_inst_data(message["payload"]["data"], data_storage) handle_cnstr_new_inst_data(message["payload"]["data"], data_storage)
elif message["type"] == "send" and message["payload"]["type"] == "load-dex": elif message["type"] == "send" and message["payload"]["type"] == "load-dex":
handle_load_dex(message["payload"]["data"], data_storage, file_storage) handle_load_dex(message["payload"]["data"], data_storage, file_storage)
elif message["type"] == "send" and message["payload"]["type"] == "apk-cl":
handle_classloader_data(message["payload"]["data"], data_storage)
else: else:
print("[-] message:", message) print("[-] message:", message)
@ -48,8 +59,13 @@ def print_stack(stack, prefix: str):
print(f" {prefix}{frame['method']}:{frame['bytecode_index']}{native}") print(f" {prefix}{frame['method']}:{frame['bytecode_index']}{native}")
def handle_classloader_data(data: dict, data_storage: dict):
data_storage["initial_classloaders"].append(data)
def handle_invoke_data(data, data_storage: dict): def handle_invoke_data(data, data_storage: dict):
method = data["method"] method = data["method"]
method_cl_id = data["method_cl_id"]
# TODO: good idea? # TODO: good idea?
if method in [ if method in [
"Landroid/view/View;->getTranslationZ()F", "Landroid/view/View;->getTranslationZ()F",
@ -59,6 +75,7 @@ def handle_invoke_data(data, data_storage: dict):
if len(data["stack"]) == 0: if len(data["stack"]) == 0:
return return
caller_method = data["stack"][0]["method"] caller_method = data["stack"][0]["method"]
caller_cl_id = data["stack"][0]["cl_id"]
addr = data["stack"][0]["bytecode_index"] addr = data["stack"][0]["bytecode_index"]
is_static = data["is_static"] is_static = data["is_static"]
if is_static: if is_static:
@ -66,8 +83,8 @@ def handle_invoke_data(data, data_storage: dict):
else: else:
is_static_str = "" is_static_str = ""
print("[+] Method.Invoke:") print("[+] Method.Invoke:")
print(f" called: {method}{is_static_str}") print(f" called: [{method_cl_id}]{method}{is_static_str}")
print(f" by: {caller_method}") print(f" by: [{caller_cl_id}]{caller_method}")
print(f" at: 0x{addr:08x}") print(f" at: 0x{addr:08x}")
# print(f" stack:") # print(f" stack:")
# print_stack(data["stack"], " ") # print_stack(data["stack"], " ")
@ -76,7 +93,11 @@ def handle_invoke_data(data, data_storage: dict):
data_storage["invoke_data"].append( data_storage["invoke_data"].append(
{ {
"method": method, "method": method,
"method_cl_id": method_cl_id,
"renamed_method": None,
"caller_method": caller_method, "caller_method": caller_method,
"caller_cl_id": caller_cl_id,
"renamed_caller_method": None,
"addr": addr, "addr": addr,
"is_static": is_static, "is_static": is_static,
} }
@ -85,6 +106,7 @@ def handle_invoke_data(data, data_storage: dict):
def handle_class_new_inst_data(data, data_storage: dict): def handle_class_new_inst_data(data, data_storage: dict):
constructor = data["constructor"] constructor = data["constructor"]
constructor_cl_id = data["constructor_cl_id"]
if len(data["stack"]) == 0: if len(data["stack"]) == 0:
return return
if ( if (
@ -97,10 +119,11 @@ def handle_class_new_inst_data(data, data_storage: dict):
else: else:
return return
caller_method = frame["method"] caller_method = frame["method"]
caller_cl_id = frame["cl_id"]
addr = frame["bytecode_index"] addr = frame["bytecode_index"]
print("[+] Class.NewInstance:") print("[+] Class.NewInstance:")
print(f" called: {constructor}") print(f" called: [{constructor_cl_id}]{constructor}")
print(f" by: {caller_method}") print(f" by: [{caller_cl_id}]{caller_method}")
print(f" at: 0x{addr:08x}") print(f" at: 0x{addr:08x}")
# print(f" stack:") # print(f" stack:")
# print_stack(data["stack"], " ") # print_stack(data["stack"], " ")
@ -109,7 +132,11 @@ def handle_class_new_inst_data(data, data_storage: dict):
data_storage["class_new_inst_data"].append( data_storage["class_new_inst_data"].append(
{ {
"constructor": constructor, "constructor": constructor,
"constructor_cl_id": constructor_cl_id,
"renamed_constructor": None,
"caller_method": caller_method, "caller_method": caller_method,
"caller_cl_id": caller_cl_id,
"renamed_caller_method": None,
"addr": addr, "addr": addr,
} }
) )
@ -117,15 +144,17 @@ def handle_class_new_inst_data(data, data_storage: dict):
def handle_cnstr_new_inst_data(data, data_storage: dict): def handle_cnstr_new_inst_data(data, data_storage: dict):
constructor = data["constructor"] constructor = data["constructor"]
constructor_cl_id = data["constructor_cl_id"]
if not constructor.startswith("Lcom/example/theseus"): if not constructor.startswith("Lcom/example/theseus"):
return return
if len(data["stack"]) == 0: if len(data["stack"]) == 0:
return return
caller_method = data["stack"][0]["method"] caller_method = data["stack"][0]["method"]
caller_cl_id = data["stack"][0]["cl_id"]
addr = data["stack"][0]["bytecode_index"] addr = data["stack"][0]["bytecode_index"]
print("[+] Constructor.newInstance:") print("[+] Constructor.newInstance:")
print(f" called: {constructor}") print(f" called: [{constructor_cl_id}]{constructor}")
print(f" by: {caller_method}") print(f" by: [{caller_cl_id}]{caller_method}")
print(f" at: 0x{addr:08x}") print(f" at: 0x{addr:08x}")
# print(f" stack:") # print(f" stack:")
# print_stack(data["stack"], " ") # print_stack(data["stack"], " ")
@ -134,7 +163,11 @@ def handle_cnstr_new_inst_data(data, data_storage: dict):
data_storage["cnstr_new_inst_data"].append( data_storage["cnstr_new_inst_data"].append(
{ {
"constructor": constructor, "constructor": constructor,
"constructor_cl_id": constructor_cl_id,
"renamed_constructor": None,
"caller_method": caller_method, "caller_method": caller_method,
"caller_cl_id": caller_cl_id,
"renamed_caller_method": None,
"addr": addr, "addr": addr,
} }
) )
@ -183,10 +216,10 @@ FRIDA_SERVER_BIN = Path(__file__).parent / "frida-server-16.7.0-android-x86_64.x
FRIDA_SERVER_ANDROID_PATH = "/data/local/tmp/frida-server" FRIDA_SERVER_ANDROID_PATH = "/data/local/tmp/frida-server"
def setup_frida(device: str, env: dict[str, str]) -> frida.core.Device: def setup_frida(device_name: str, env: dict[str, str]) -> frida.core.Device:
if device != "": if device_name != "":
device = frida.get_device(args.device) device = frida.get_device(device_name)
env["ANDROID_SERIAL"] = args.device env["ANDROID_SERIAL"] = device_name
else: else:
device = frida.get_usb_device() device = frida.get_usb_device()
@ -197,14 +230,19 @@ def setup_frida(device: str, env: dict[str, str]) -> frida.core.Device:
except frida.ServerNotRunningError: except frida.ServerNotRunningError:
pass pass
# Start server # Start server
proc = subprocess.run( proc: subprocess.CompletedProcess[str] | subprocess.CompletedProcess[bytes] = (
["adb", "shell", "whoami"], encoding="utf-8", stdout=subprocess.PIPE, env=env subprocess.run(
["adb", "shell", "whoami"],
encoding="utf-8",
stdout=subprocess.PIPE,
env=env,
)
) )
if proc.stdout.strip() != "root": if proc.stdout.strip() != "root":
proc = subprocess.run(["adb", "root"], env=env) proc = subprocess.run(["adb", "root"], env=env)
# Rooting adb will disconnect the device # Rooting adb will disconnect the device
if device != "": if device_name != "":
device = frida.get_device(device) device = frida.get_device(device_name)
else: else:
device = frida.get_usb_device() device = frida.get_usb_device()
perm = subprocess.run( perm = subprocess.run(
@ -221,20 +259,31 @@ def setup_frida(device: str, env: dict[str, str]) -> frida.core.Device:
"7", "7",
] # int(perm[0]) & 1 == 1 ] # int(perm[0]) & 1 == 1
if perm == "": if perm == "":
subprocess.run( with tempfile.TemporaryDirectory() as tmpdname:
[ tmpd = Path(tmpdname)
"adb", with (
"push", lzma.open(str(FRIDA_SERVER_BIN.absolute())) as fin,
str(FRIDA_SERVER_BIN.absolute()), (tmpd / "frida-server").open("wb") as fout,
FRIDA_SERVER_ANDROID_PATH, ):
], shutil.copyfileobj(fin, fout)
env=env,
) subprocess.run(
[
"adb",
"push",
str((tmpd / "frida-server").absolute()),
FRIDA_SERVER_ANDROID_PATH,
],
env=env,
)
if need_perm_resset: if need_perm_resset:
subprocess.run(["adb", "chmod", "755", FRIDA_SERVER_ANDROID_PATH], env=env) subprocess.run(
["adb", "shell", "chmod", "755", FRIDA_SERVER_ANDROID_PATH], env=env
)
subprocess.Popen(["adb", "shell", FRIDA_SERVER_ANDROID_PATH], env=env) subprocess.Popen(["adb", "shell", FRIDA_SERVER_ANDROID_PATH], env=env)
# The server take some time to start # The server take some time to start
# time.sleep(3) # time.sleep(3)
t = spinner()
while True: while True:
try: try:
s = device.attach(0) s = device.attach(0)
@ -242,11 +291,11 @@ def setup_frida(device: str, env: dict[str, str]) -> frida.core.Device:
print("[*] Server started: begin analysis ") print("[*] Server started: begin analysis ")
return device return device
except frida.ServerNotRunningError: except frida.ServerNotRunningError:
print("[-] Waiting for frida server to start", end="\r") print(f"[{t.__next__()}] Waiting for frida server to start", end="\r")
time.sleep(0.3) time.sleep(0.3)
def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO): def collect_runtime(apk: Path, device_name: str, file_storage: Path, output: TextIO):
env = dict(os.environ) env = dict(os.environ)
if not file_storage.exists(): if not file_storage.exists():
@ -255,7 +304,7 @@ def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO):
print("[!] file_storage must be a directory") print("[!] file_storage must be a directory")
exit() exit()
device = setup_frida(device, env) device = setup_frida(device_name, env)
app = get_apkid(apk)[0] app = get_apkid(apk)[0]
@ -265,9 +314,9 @@ def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO):
subprocess.run(["adb", "install", str(apk.absolute())], env=env) subprocess.run(["adb", "install", str(apk.absolute())], env=env)
with FRIDA_SCRIPT.open("r") as file: with FRIDA_SCRIPT.open("r") as file:
script = file.read() jsscript = file.read()
with STACK_CONSUMER_B64.open("r") as file: with STACK_CONSUMER_B64.open("r") as file:
script = script.replace( jsscript = jsscript.replace(
"<PYTHON REPLACE StackConsumer.dex.b64>", "<PYTHON REPLACE StackConsumer.dex.b64>",
file.read().replace("\n", "").strip(), file.read().replace("\n", "").strip(),
) )
@ -275,7 +324,7 @@ def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO):
pid = device.spawn([app]) pid = device.spawn([app])
session = device.attach(pid) session = device.attach(pid)
try: try:
script = session.create_script(script) script = session.create_script(jsscript)
except frida.InvalidArgumentError as e: except frida.InvalidArgumentError as e:
print("[!] Error:") print("[!] Error:")
print( print(
@ -286,11 +335,12 @@ def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO):
) )
raise e raise e
data_storage = { data_storage: dict[str, Any] = {
"invoke_data": [], "invoke_data": [],
"class_new_inst_data": [], "class_new_inst_data": [],
"cnstr_new_inst_data": [], "cnstr_new_inst_data": [],
"dyn_code_load": [], "dyn_code_load": [],
"initial_classloaders": [],
} }
script.on( script.on(
@ -305,6 +355,40 @@ def collect_runtime(apk: Path, device: str, file_storage: Path, output: TextIO):
print("==> Press ENTER to finish the analysis <==") print("==> Press ENTER to finish the analysis <==")
input() input()
main_class_loader: str | None = None
cls = {d["id"]: d for d in data_storage["initial_classloaders"]}
for load_data in data_storage["dyn_code_load"]:
if load_data["classloader"] in cls:
del cls[load_data["classloader"]]
for id_ in cls.keys():
if (
'dalvik.system.PathClassLoader[DexPathList[[directory "."],'
in cls[id_]["str"]
):
del cls[id_]
elif cls[id_]["cname"] == "java.lang.BootClassLoader":
del cls[id_]
if len(cls) == 0:
print("[!] No classloader found for the main APK")
elif len(cls) > 1:
print(
"[!] Multiple classloader found that could be the main APK, try to guess the right one"
)
nb_occ = {k: 0 for k in cls.keys()}
for data in data_storage["class_new_inst_data"]:
if data["caller_cl_id"] in nb_occ:
nb_occ[data["caller_cl_id"]] += 1
for data in data_storage["invoke_data"]:
if data["caller_cl_id"] in nb_occ:
nb_occ[data["caller_cl_id"]] += 1
for data in data_storage["cnstr_new_inst_data"]:
if data["caller_cl_id"] in nb_occ:
nb_occ[data["caller_cl_id"]] += 1
main_class_loader = max(cls.keys(), key=lambda x: nb_occ[x])
else:
main_class_loader = list(cls.keys())[0]
data_storage["apk_cl_id"] = main_class_loader
json.dump(data_storage, output, indent=" ") json.dump(data_storage, output, indent=" ")
@ -340,7 +424,7 @@ def main():
if args.output is None: if args.output is None:
collect_runtime( collect_runtime(
apk=args.apk, apk=args.apk,
device=args.device, device_name=args.device,
file_storage=args.dex_dir, file_storage=args.dex_dir,
output=sys.stdout, output=sys.stdout,
) )
@ -348,7 +432,7 @@ def main():
with args.output.open("w") as fp: with args.output.open("w") as fp:
collect_runtime( collect_runtime(
apk=args.apk, apk=args.apk,
device=args.device, device_name=args.device,
file_storage=args.dex_dir, file_storage=args.dex_dir,
output=fp, output=fp,
) )

21
frida/theseus_frida/hook.js
View file

@ -46,6 +46,7 @@ Java.perform(() => {
"bytecode_index": frame.getByteCodeIndex(), "bytecode_index": frame.getByteCodeIndex(),
"is_native": frame.isNativeMethod(), "is_native": frame.isNativeMethod(),
"method": frame.getDeclaringClass().descriptorString() + "->" + frame.getMethodName() + frame.getDescriptor(), "method": frame.getDeclaringClass().descriptorString() + "->" + frame.getMethodName() + frame.getDescriptor(),
"cl_id": System.identityHashCode(frame.getDeclaringClass().getClassLoader()),
//{ //{
//"descriptor": frame.getDescriptor(), //"descriptor": frame.getDescriptor(),
//"name": frame.getMethodName(), //"name": frame.getMethodName(),
@ -101,6 +102,7 @@ Java.perform(() => {
"type": "invoke", "type": "invoke",
"data": { "data": {
"method": get_method_dsc(this), "method": get_method_dsc(this),
"method_cl_id": System.identityHashCode(this.getDeclaringClass().getClassLoader()),
/*{ /*{
"name": this.getName(), "name": this.getName(),
"class": this.getDeclaringClass().getName(), "class": this.getDeclaringClass().getName(),
@ -123,6 +125,7 @@ Java.perform(() => {
"type": "class-new-inst", "type": "class-new-inst",
"data": { "data": {
"constructor": this.descriptorString() + "-><init>()V", "constructor": this.descriptorString() + "-><init>()V",
"constructor_cl_id": System.identityHashCode(this.getClassLoader()),
/*{ /*{
"name": "<init>", "name": "<init>",
"class": this.getName(), "class": this.getName(),
@ -144,6 +147,7 @@ Java.perform(() => {
"type": "cnstr-new-isnt", "type": "cnstr-new-isnt",
"data": { "data": {
"constructor": get_constr_dsc(this), "constructor": get_constr_dsc(this),
"constructor_cl_id": System.identityHashCode(this.getDeclaringClass().getClassLoader()),
/* /*
{ {
"name": "<init>", "name": "<init>",
@ -275,5 +279,22 @@ Java.perform(() => {
elements, elements,
); );
}; };
// Find the main APK class loader:
// Not so easy, just send all class loader and sort this out later:
var class_loader = Java.enumerateClassLoadersSync();
for (var cl of class_loader) {
//if (cl.toString().includes("dalvik.system.PathClassLoader[DexPathList[[directory \".\"],")) {
// continue;
//}
//if (cl.$className == "java.lang.BootClassLoader") {
// continue;
//}
send({"type": "classloader", "data": {
"id": System.identityHashCode(cl),
"str": cl.toString(),
"cname": cl.$className
}});
}
}); });

5
patcher/src/bin/patcher.rs
View file

@ -43,10 +43,11 @@ fn main() {
.unwrap() .unwrap()
.read_to_string(&mut json) .read_to_string(&mut json)
.unwrap(); .unwrap();
let rt_data: RuntimeData = serde_json::from_str(&json).unwrap(); let mut rt_data: RuntimeData = serde_json::from_str(&json).unwrap();
// Dynamic Loading // Dynamic Loading
insert_code(cli.code_loading_patch_strategy, &mut apk, &rt_data).unwrap(); insert_code(cli.code_loading_patch_strategy, &mut apk, &mut rt_data).unwrap();
let rt_data = rt_data; // not mut anymore
// Reflection // Reflection
let mut test_methods = HashMap::new(); let mut test_methods = HashMap::new();

11
patcher/src/code_loading_patcher.rs
View file

@ -7,6 +7,12 @@ use clap::ValueEnum;
use crate::runtime_data::RuntimeData; use crate::runtime_data::RuntimeData;
// TODO TODO TODO TODO TODO TODO TODO TODO TODO TODO TODO
//
// INSERT EMPTY CLASS LOADERS WHEN ID REFERS TO UNKNOWN CLASS LOADER
//
// TODO TODO TODO TODO TODO TODO TODO TODO TODO TODO TODO
#[derive(ValueEnum, Debug, PartialEq, Clone, Copy, Default)] #[derive(ValueEnum, Debug, PartialEq, Clone, Copy, Default)]
pub enum CodePatchingStrategy { pub enum CodePatchingStrategy {
#[default] #[default]
@ -41,7 +47,10 @@ fn insert_code_model_class_loaders(apk: &mut Apk, runtime_data: &mut RuntimeData
let mut class_defined = apk.list_classes(); let mut class_defined = apk.list_classes();
let mut class_redefined = HashSet::new(); let mut class_redefined = HashSet::new();
let mut class_loaders = HashMap::new(); let mut class_loaders = HashMap::new();
let main_cl_id = runtime_data.apk_cl_id.clone(); let main_cl_id = runtime_data
.apk_cl_id
.clone()
.unwrap_or_else(|| "MAIN".to_string());
class_loaders.insert( class_loaders.insert(
main_cl_id.clone(), main_cl_id.clone(),
ClassLoader { ClassLoader {

2
patcher/src/runtime_data.rs
View file

@ -11,7 +11,7 @@ pub struct RuntimeData {
pub cnstr_new_inst_data: Vec<ReflectionCnstrNewInstData>, pub cnstr_new_inst_data: Vec<ReflectionCnstrNewInstData>,
pub dyn_code_load: Vec<DynamicCodeLoadingData>, pub dyn_code_load: Vec<DynamicCodeLoadingData>,
/// The id of the class loader of the apk (the main classloader) /// The id of the class loader of the apk (the main classloader)
pub apk_cl_id: String, pub apk_cl_id: Option<String>,
} }
impl RuntimeData { impl RuntimeData {

11
theseus_autopatcher/src/theseus_autopatcher/__init__.py
View file

@ -8,6 +8,12 @@ from shutil import which
from theseus_frida import collect_runtime from theseus_frida import collect_runtime
def spinner(symbs: str = "◜◠◝◞◡◟"):
while True:
for s in symbs:
yield s
def get_android_sdk_path() -> Path | None: def get_android_sdk_path() -> Path | None:
if "ANDROID_HOME" in os.environ: if "ANDROID_HOME" in os.environ:
return Path(os.environ["ANDROID_HOME"]) return Path(os.environ["ANDROID_HOME"])
@ -216,7 +222,10 @@ def main():
(tmpd / "dex").mkdir() (tmpd / "dex").mkdir()
with (tmpd / "runtime.json").open("w") as fp: with (tmpd / "runtime.json").open("w") as fp:
collect_runtime( collect_runtime(
apk=args.apk, device=args.device, file_storage=tmpd / "dex", output=fp apk=args.apk,
device_name=args.device,
file_storage=tmpd / "dex",
output=fp,
) )
patch_apk( patch_apk(
runtime_data=tmpd / "runtime.json", runtime_data=tmpd / "runtime.json",