run.py

#!/usr/bin/env python3
"""
run.py - load the Instagram TLS-unpinning Frida agent.

Spawns the target, loads the agent while suspended, then resumes; detaches
cleanly on Ctrl-C. Targets Frida 16.x (raw scripts need the global Java bridge);
on 17.x use the `frida` CLI, which bundles it.

Usage:
  python3 run.py                                  # spawn com.instagram.android
  python3 run.py --proxy 192.168.1.17:8080        # also set/clear the device proxy
  python3 run.py --attach                         # attach to a running app
  python3 run.py -p com.example -l path/to.js -s SERIAL
"""

import argparse
import os
import signal
import subprocess
import sys
import threading

DEFAULT_PKG = "com.instagram.android"
HERE = os.path.dirname(os.path.abspath(__file__))
DEFAULT_AGENT = os.path.join(HERE, "scripts", "instagram-unpin.js")


def info(*a):
    print(*a, file=sys.stderr, flush=True)


def adb_runner(serial):
    base = ["adb"] + (["-s", serial] if serial else [])

    def adb(*args):
        return subprocess.run(base + list(args), capture_output=True, text=True)

    return adb


def check_versions(frida, dev, serial):
    """Best-effort: warn if the device frida-server doesn't match the host."""
    host = frida.__version__
    try:
        adb = adb_runner(serial)
        out = adb("shell", "/data/local/tmp/frida-server", "--version").stdout.strip()
        if out and out != host:
            info(f"[!] version mismatch: host frida {host} vs device frida-server {out}")
            info("    They must match exactly. Push the matching frida-server and retry.")
    except Exception:
        pass


def main():
    ap = argparse.ArgumentParser(description="Load the Instagram TLS-unpinning Frida agent.")
    ap.add_argument("-p", "--package", default=DEFAULT_PKG, help="target package")
    ap.add_argument("-l", "--script", default=DEFAULT_AGENT, help="agent .js to load")
    ap.add_argument("-s", "--serial", default=None, help="device serial (USB auto-detected otherwise)")
    ap.add_argument("--attach", action="store_true", help="attach to a running process instead of spawning")
    ap.add_argument("--proxy", default=None, metavar="HOST:PORT",
                    help="set device global http_proxy on start, clear it on exit")
    ap.add_argument("--allow-frida17", action="store_true",
                    help="skip the Frida-17 guard (only if your agent bundles its own Java bridge)")
    args = ap.parse_args()

    try:
        import frida
    except ImportError:
        info("[!] frida not installed:  pip install 'frida==16.7.19' 'frida-tools==13.7.1'")
        sys.exit(1)

    major = int(frida.__version__.split(".")[0])
    if major >= 17 and not args.allow_frida17:
        info(f"[!] Frida {frida.__version__}: raw scripts no longer get a global `Java` bridge,")
        info("    which this agent needs. Either:")
        info("      - pip install 'frida==16.7.19' 'frida-tools==13.7.1'  (+ matching frida-server), or")
        info(f"      - load via the CLI, which bundles the bridge on 17.x:")
        info(f"          frida -U -f {args.package} -l {args.script}")
        sys.exit(2)

    if not os.path.isfile(args.script):
        info(f"[!] agent not found: {args.script}")
        sys.exit(1)
    code = open(args.script).read()

    dev = frida.get_device(args.serial, timeout=10) if args.serial else frida.get_usb_device(timeout=10)
    info(f"[*] device: {dev.name}  (host frida {frida.__version__})")
    check_versions(frida, dev, args.serial)

    adb = adb_runner(args.serial)
    proxy_set = False
    if args.proxy:
        adb("shell", "settings", "put", "global", "http_proxy", args.proxy)
        proxy_set = True
        info(f"[*] device proxy set -> {args.proxy}")

    def on_message(msg, data):
        t = msg.get("type")
        if t in ("send", "log"):
            print(msg.get("payload"), flush=True)
        elif t == "error":
            info("[agent error]", msg.get("stack") or msg.get("description"))

    stop = threading.Event()

    def cleanup():
        if proxy_set:
            adb("shell", "settings", "put", "global", "http_proxy", ":0")
            info("[*] device proxy cleared")

    session = None
    try:
        if args.attach:
            session = dev.attach(args.package)
            info(f"[*] attached to {args.package}")
            load_script(session, code, on_message)
        else:
            # Load the agent while suspended, then resume, so its hooks are in
            # place before the app makes any connections.
            pid = dev.spawn([args.package])
            session = dev.attach(pid)
            info(f"[*] spawned {args.package} (pid {pid})")
            load_script(session, code, on_message)
            dev.resume(pid)
            info("[*] resumed")

        session.on("detached", lambda reason, *_: (info(f"[*] session detached: {reason}"), stop.set()))

        info("[*] agent loaded - press Ctrl-C to detach and exit.")
        signal.signal(signal.SIGINT, lambda *_: stop.set())
        signal.signal(signal.SIGTERM, lambda *_: stop.set())
        while not stop.wait(0.5):
            pass
    finally:
        try:
            if session is not None:
                session.detach()
        except Exception:
            pass
        cleanup()
        info("[*] done.")


def load_script(session, code, on_message):
    script = session.create_script(code)
    script.on("message", on_message)
    script.load()
    return script


if __name__ == "__main__":
    main()