automagical frida setup

This commit is contained in:
Jean-Marie 'Histausse' Mineau 2025-03-14 13:22:47 +01:00
parent 5b4dd74a5e
commit 6380ce9917
Signed by: histausse
GPG key ID: B66AEEDA9B645AD2
4 changed files with 123 additions and 57 deletions

View file

@ -24,8 +24,8 @@ HASH_NB_BYTES = 4
# Define handler to event generated by the scripts
def on_message(message, data, data_storage: dict, file_storage: Path):
if message["type"] == "error":
print(f"[error] {message['description']}")
print(message["stack"])
print(f"[!] {message['description']}")
print(" " + message["stack"].replace("\n", "\n "))
elif message["type"] == "send" and message["payload"]["type"] == "invoke":
handle_invoke_data(message["payload"]["data"], data_storage)
elif message["type"] == "send" and message["payload"]["type"] == "class-new-inst":
@ -35,7 +35,7 @@ def on_message(message, data, data_storage: dict, file_storage: Path):
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)
print("[-] message:", message)
def print_stack(stack, prefix: str):
@ -43,7 +43,7 @@ def print_stack(stack, prefix: str):
native = ""
if frame["is_native"]:
native = " (native)"
print(f"{prefix}{frame['method']}:{frame['bytecode_index']}{native}")
print(f" {prefix}{frame['method']}:{frame['bytecode_index']}{native}")
def handle_invoke_data(data, data_storage: dict):
@ -63,7 +63,7 @@ def handle_invoke_data(data, data_storage: dict):
is_static_str = " (static)"
else:
is_static_str = ""
print("Method.Invoke:")
print("[+] Method.Invoke:")
print(f" called: {method}{is_static_str}")
print(f" by: {caller_method}")
print(f" at: 0x{addr:08x}")
@ -96,7 +96,7 @@ def handle_class_new_inst_data(data, data_storage: dict):
return
caller_method = frame["method"]
addr = frame["bytecode_index"]
print("Class.NewInstance:")
print("[+] Class.NewInstance:")
print(f" called: {constructor}")
print(f" by: {caller_method}")
print(f" at: 0x{addr:08x}")
@ -121,7 +121,7 @@ def handle_cnstr_new_inst_data(data, data_storage: dict):
return
caller_method = data["stack"][0]["method"]
addr = data["stack"][0]["bytecode_index"]
print("Constructor.newInstance:")
print("[+] Constructor.newInstance:")
print(f" called: {constructor}")
print(f" by: {caller_method}")
print(f" at: 0x{addr:08x}")
@ -147,7 +147,7 @@ def handle_load_dex(data, data_storage: dict, file_storage: Path):
classloader = classloader.to_bytes(HASH_NB_BYTES).hex()
short_class = classloader_class.split("/")[-1].removesuffix(";")
files = []
print("DEX file loaded:")
print("[+] DEX file loaded:")
print(f" by: {classloader_class} ({classloader})")
for file in dex:
file_bin = base64.b64decode(file)
@ -177,6 +177,73 @@ def handle_load_dex(data, data_storage: dict, file_storage: Path):
)
FRIDA_SERVER_BIN = Path(__file__).parent / "frida-server-16.7.0-android-x86_64.xz"
FRIDA_SERVER_ANDROID_PATH = "/data/local/tmp/frida-server"
def setup_frida(device: str, env: dict[str, str]) -> frida.core.Device:
if device != "":
device = frida.get_device(args.device)
env["ANDROID_SERIAL"] = args.device
else:
device = frida.get_usb_device()
try:
s = device.attach(0)
s.detach()
return device
except frida.ServerNotRunningError:
pass
# Start server
proc = subprocess.run(
["adb", "shell", "whoami"], encoding="utf-8", stdout=subprocess.PIPE, env=env
)
if proc.stdout.strip() != "root":
proc = subprocess.run(["adb", "root"], env=env)
# Rooting adb will disconnect the device
if device != "":
device = frida.get_device(device)
else:
device = frida.get_usb_device()
perm = subprocess.run(
["adb", "shell", "stat", "-c", "%a", FRIDA_SERVER_ANDROID_PATH],
encoding="utf-8",
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env,
).stdout.strip()
need_perm_resset = (perm == "") or perm[0] not in [
"1",
"3",
"5",
"7",
] # int(perm[0]) & 1 == 1
if perm == "":
subprocess.run(
[
"adb",
"push",
str(FRIDA_SERVER_BIN.absolute()),
FRIDA_SERVER_ANDROID_PATH,
],
env=env,
)
if need_perm_resset:
subprocess.run(["adb", "chmod", "755", FRIDA_SERVER_ANDROID_PATH], env=env)
subprocess.Popen(["adb", "shell", FRIDA_SERVER_ANDROID_PATH], env=env)
# The server take some time to start
# time.sleep(3)
while True:
try:
s = device.attach(0)
s.detach()
print("[*] Server started: begin analysis ")
return device
except frida.ServerNotRunningError:
print("[-] Waiting for frida server to start", end="\r")
time.sleep(0.3)
def main():
parser = argparse.ArgumentParser(
prog="Android Theseus project",
@ -212,14 +279,10 @@ def main():
if not file_storage.exists():
file_storage.mkdir(parents=True)
if not file_storage.is_dir():
print("--dex-dir must be a directory")
print("[!] --dex-dir must be a directory")
exit()
if args.device != "":
device = frida.get_device(args.device)
env["ANDROID_SERIAL"] = args.device
else:
device = frida.get_usb_device()
device = setup_frida(args.device, env)
app = get_apkid(args.apk)[0]
@ -241,8 +304,10 @@ def main():
try:
script = session.create_script(script)
except frida.InvalidArgumentError as e:
print("[!] Error:")
print(
"\n".join(
" "
+ "\n ".join(
map(lambda v: f"{v[0]+1: 3} {v[1]}", enumerate(script.split("\n")))
)
)
@ -265,7 +330,7 @@ def main():
# Resume the execution of the APK
device.resume(pid)
print("Press ENTER to finish the analysis")
print("==> Press ENTER to finish the analysis <==")
input()
if args.output is None:
print(json.dumps(data_storage, indent=" "))