frida_patcher.py

IN_LLDB = False # DO NOT MOVE THIS

import sys
import struct
import subprocess
import os
import os.path

# usage: python3 frida_patcher.py TARGET_BINARY_PATH [args...]
# spawns the specified target in a "blocked but not suspended" state that allows frida to attach.
# for more details see: https://github.com/frida/frida/issues/1992
# dependencies: clang, lldb
# for Intel Macs only, tested with lldb-1500.0.22.8 on Sonoma 14.3
# warning: hacky code ahead

C_HELPER = """
#include <unistd.h>
#include <stdlib.h>
#include <signal.h>
#include <stdio.h>
#include <fcntl.h>

#define TMP_HELPER_IPC_FP "/tmp/frida_patcher_helper_ipc"
#define ERR() fprintf(stderr, "err\\n"); exit(1)

// usage: ./helper PROG [ARGS...]
// opens an fd to TMP_HELPER_IPC_FP, then executes PROG in a suspended state.
// the fd is still open in the new process.
// outputs the fd to stdout.

int main(int argc, char** argv) {
    if (argc < 2) { ERR(); }

    int fd = open(TMP_HELPER_IPC_FP, O_RDONLY);
    if (fd < 0) { ERR(); }
    printf("%d\\n", fd); fflush(stdout);

    if (kill(getpid(), SIGTSTP) != 0) { ERR(); } // suspend

    argv++; // increment to the "PROG" argument
    if (execv(argv[0], argv) == -1 ) { ERR(); }
}
"""

HELPER_BIN_FP = os.getcwd() + "/frida_patcher_helper"

# mustn't contain quotes or spaces
TMP_IPC_FP = "/tmp/frida_patcher_ipc"
TMP_HELPER_IPC_FP = "/tmp/frida_patcher_helper_ipc"
TMP_PY_FP = "/tmp/frida_patcher_stage_2.py"
TMP_C_FP = "/tmp/frida_patcher_helper.c"
TMP_PAYLOAD_FP = "/tmp/frida_patcher_payload.bin"

STAGE_2_MODULE = TMP_PY_FP.split("/")[-1][:-3]

# payload pseudocode:
# 
#     // NOTE: this probably isn't required because we try to inject before any user code
#     backupRegisters();
#     
#     char b;
#     while (1) {
#         syscall(SYS_read, i32, &b, 1);
#         if (b == '1') {
#             break;
#         }
#     }
#     
#     restoreRegisters();
#     returnToUserCode();
# 
# "i32" is the file descriptor passed to read() and is inserted dynamically into the payload

BASE_PAYLOAD_P1 = bytes.fromhex("9c505756525141534883ec08c6042400b803000002")
# in between: "bfxxxxxxxx" (mov edi, i32)
BASE_PAYLOAD_P2 = bytes.fromhex("4889e6ba010000000f058a04243c3175e54883c408415b595a5e5f589d")
# after: "e9xxxxxxxx" (jmp rel32)

# cmd can't contain untrusted input (not prod ready)
def shell(cmd):
    out = subprocess.run(cmd, shell=True, capture_output=True)
    return out.stdout.decode("utf-8")
def sh_esc_quote(s):
    return s.replace("'", "'''")

if not IN_LLDB:
    # stage 1

    if len(sys.argv) < 2:
        print("usage: python3 frida_patcher.py TARGET_BINARY_PATH [args...]"); sys.exit(1)

    shell(f"{{ echo 'IN_LLDB = True'; tail +2 '{sh_esc_quote(__file__)}'; }} > {TMP_PY_FP}")

    target_bin = sys.argv[1:]
    bin_fp = target_bin[0]
    bin_name = bin_fp.split("/")[-1]
    if "'" in bin_name:
        print("err: illegal char in bin name"); sys.exit(1)

    print("spawning helper...")
    shell(f"rm -f {TMP_HELPER_IPC_FP}; touch {TMP_HELPER_IPC_FP}")

    # compile the helper binary if it doesn't already exist
    if not os.path.exists(HELPER_BIN_FP):
        with open(TMP_C_FP, "w") as f:
            f.write(C_HELPER)
        print(f"compiling helper binary to {HELPER_BIN_FP}...")
        print(shell(f"cc {TMP_C_FP} -o '{sh_esc_quote(HELPER_BIN_FP)}'"))

    popen_args = [HELPER_BIN_FP, *target_bin]
    helper_p = subprocess.Popen(popen_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    print(f"spawned helper with pid {helper_p.pid}")

    # read fd from helper's stdout
    line = ""
    while 1:
        c = helper_p.stdout.read1(1)
        c = c.decode("utf-8")
        if c:
            line += c
        if c == "\n":
            break
    fd = int(line[:-1])

    # pass data to stage 2
    with open(TMP_IPC_FP, "w") as f:
        f.write(f"{bin_name}\n{fd}\n")

    print("spawning lldb...")
    lldb_p = subprocess.Popen(["/usr/bin/lldb", "--no-use-colors", "-b", "-o", f"command script import {TMP_PY_FP}", "-p", f"{helper_p.pid}"])
    lldb_p.wait()

    print(f"done!\n")
    print(f"add the following line to your ~/.zprofile:")
    print(f"alias frida2=\"frida -e new\\ File\\(\\'{TMP_HELPER_IPC_FP}\\',\\'w\\'\\).write\\(\\'1\\'\\)\"", end="\n\n")
    print("then run your frida script like so:")
    print(f"frida2 -p {helper_p.pid} -l <path to script> [args...]", end="\n\n")
    print("the target process will automatically resume after the script runs\n")
    print("waiting for target to exit...")

    helper_p.wait()
    print(helper_p.stdout.read().decode("utf-8"), end="")

else:
    # stage 2 (runs inside lldb)
    import lldb

    # captures and returns the output
    def run_lldb_cmd(ci, cmd, ec=None, print_c=True):
        if print_c:
            print(cmd)
        res = lldb.SBCommandReturnObject()
        if ec:
            ci.HandleCommand(cmd, ec, res)
        else:
            ci.HandleCommand(cmd, res)
        if res.Succeeded():
            return res.GetOutput()
        else:
            return res.GetError() # TODO: signal error

    # absolute func addr -> rel32 (bytes)
    def to_rel_32(curr_addr, f_addr):
        diff = f_addr - curr_addr
        return struct.pack("<i", diff)

    g_stop_hook_enabled = True
    class StopHook():
        def __init__(self, target, extra_args, internal_dict):
            pass
        def handle_stop(self, exe_ctx, stream):
            global g_stop_hook_enabled
            if g_stop_hook_enabled:
                main_2(exe_ctx)
            g_stop_hook_enabled = False
            return True

    def main_1(debugger, internal_dict):
        print("hello from stage 2")

        # get bin_name from stage 1
        with open(TMP_IPC_FP, "r") as f:
            bin_name = f.readline()[:-1]

        ci = debugger.GetCommandInterpreter()

        # continue until "stop reason = exec", which indicates the helper executed the target binary
        print(run_lldb_cmd(ci, "c"))
        # we'll now be in a dyld function, so we continue to the target binary's code
        print(run_lldb_cmd(ci, f"break set -s '{bin_name}' -r '.*'"), end="")
        print(run_lldb_cmd(ci, f"target stop-hook add -P {STAGE_2_MODULE}.StopHook"), end="")

        # after a breakpoint is hit, StopHook() is called, which calls main_2()
        print(run_lldb_cmd(ci, "c"), end="")

    def main_2(exe_ctx):
        # get bin_name and fd from stage 1
        with open(TMP_IPC_FP, "r") as f:
            bin_name = f.readline()[:-1]
            fd = int(f.readline()[:-1])

        ec = exe_ctx
        ci = ec.target.debugger.GetCommandInterpreter()

        out = run_lldb_cmd(ci, f"register read pc", ec=ec); print(out, end="")
        pc_s = out.strip().split()[2]
        if not pc_s.startswith("0x"):
            print("err: unexpected output from command (1)"); return 1
        curr_pc = int(pc_s, 16)

        # there's usually quite a bit of unused space in between the __TEXT segmentand __TEXT.__text
        # (where the code begins), so we'll place our payload here

        sections = run_lldb_cmd(ci, f"target modules dump sections '{bin_name}'", ec=ec); print("\n" + sections, end="\n")
        sections_l = sections.split("\n")[3:-1]
        text_text_addr = None
        for line in sections_l:
            line_l = line.strip().split()
            if line_l[-1].endswith(".__TEXT.__text"):
                addr_s = line_l[2][1:line_l[2].index("-")]
                text_text_addr = int(addr_s, 16)
                break

        if not text_text_addr:
            print("err: unexpected output from command (2)"); return 1

        print("program code starts at: " + hex(text_text_addr))


        # construct the payload

        payload = BASE_PAYLOAD_P1

        # add the fd to the payload
        fd_i32_bs = struct.pack("<i", fd)
        payload += b"\xbf" + fd_i32_bs

        payload += BASE_PAYLOAD_P2

        payload_len = len(payload) + 5

        # add a jmp back to original PC to payload

        # address at which to write payload (squeezes right up against the program code without
        # overwriting it)
        payload_w_addr = text_text_addr - payload_len
        payload_end_addr = text_text_addr
        print(f"will write at {hex(payload_w_addr)}")

        payload += b"\xe9" + to_rel_32(payload_end_addr, curr_pc)

        print(payload.hex(" "))

        # start writing payload
        print("before:")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr)}", ec=ec, print_c=False), end="")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr + 32)}", ec=ec, print_c=False), end="")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr + 64)}", ec=ec, print_c=False), end="")

        with open(TMP_PAYLOAD_FP, "wb") as f:
            f.write(payload)
        print(run_lldb_cmd(ci, f"memory write {hex(payload_w_addr)} -i {TMP_PAYLOAD_FP}", ec=ec), end="")

        print("after:")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr)}", ec=ec, print_c=False), end="")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr + 32)}", ec=ec, print_c=False), end="")
        print(run_lldb_cmd(ci, f"memory read {hex(payload_w_addr + 64)}", ec=ec, print_c=False), end="")

        # payload is written, now we just need to jump to it
        print(run_lldb_cmd(ci, f"thread jump --force -a {hex(payload_w_addr)}", ec=ec), end="")

        print(run_lldb_cmd(ci, f"break del -f", ec=ec), end="")

        # done!

        # at this point, the target is blocked and will resume when the character "1" is written to
        # TMP_HELPER_IPC_FP, e.g. like so:
        # 
        #     echo 1 > /tmp/frida_patcher_helper_ipc

    def __lldb_init_module(debugger, internal_dict):
        main_1(debugger, internal_dict)