adb_handler.py

"""Push files into a running BlueStacks instance via its bundled ADB.

Used by the "Sideload Magisk Module" feature: BlueStacks' in-app file picker
hands Kitsune/Magisk a Windows-style URI that its module installer can't open
("Invalid Uri"). Getting the module .zip into the guest's own storage
(`/sdcard/Download/`) sidesteps that -- the user then flashes it from Magisk's
own picker, which reads guest storage fine.

This is the tool's one *online* operation: the target instance must be RUNNING
so its ADB port is open. Everything else in the app works on shut-down disks.
"""
from __future__ import annotations

import logging
import os
import re
import subprocess
import time
from typing import Callable, Optional

logger = logging.getLogger(__name__)

# BlueStacks ships its own adb as HD-Adb.exe next to HD-Player.exe. Plain adb.exe
# is a fallback for unusual layouts.
_ADB_NAMES = ("HD-Adb.exe", "adb.exe")

# bluestacks.conf: bst.instance.<name>.status.adb_port="5555"
_ADB_PORT_KEY = ".status.adb_port"

# The Magisk/Kitsune manager's package (applicationId). The full manager is a
# normal user app installed via `adb install` after first boot -- it can't be
# placed offline (see magisk_system.install_to_system). Used to uninstall for a
# clean reinstall.
MANAGER_PACKAGE = "io.github.robthepcguy.kyubi"

# Hide the console window adb would otherwise flash on Windows.
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)

Runner = Callable[[list], "subprocess.CompletedProcess"]


def _run(cmd: list, timeout: int = 60) -> subprocess.CompletedProcess:
    # Decode adb/magisk output as UTF-8 and never crash on odd bytes. A module's
    # install log can contain bytes that Windows' default cp1252 can't decode
    # (e.g. box-drawing/emoji), which would otherwise raise UnicodeDecodeError in
    # subprocess's reader thread and dump a traceback mid-install.
    return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout,
                          encoding="utf-8", errors="replace",
                          creationflags=_NO_WINDOW)


def _run_install(cmd: list) -> subprocess.CompletedProcess:
    # APK install streams ~13 MB then runs dexopt; give it more than the 60s the
    # push/shell calls use.
    return _run(cmd, timeout=180)


def find_adb(install_dirs) -> Optional[str]:
    """First HD-Adb.exe / adb.exe found in any of ``install_dirs``, else None."""
    for d in install_dirs:
        if not d:
            continue
        for name in _ADB_NAMES:
            cand = os.path.join(d, name)
            if os.path.isfile(cand):
                return cand
    return None


def instance_adb_port(config_path: str, instance_name: str) -> Optional[int]:
    """The ADB port BlueStacks assigned this instance, from bluestacks.conf.

    Returns None if the key isn't present (e.g. the instance has never been
    started, so BlueStacks hasn't recorded a port).
    """
    if not config_path or not os.path.isfile(config_path):
        return None
    key = re.compile(
        r"^bst\.instance\." + re.escape(instance_name) + re.escape(_ADB_PORT_KEY)
        + r'\s*=\s*"(\d+)"', re.IGNORECASE)
    try:
        with open(config_path, encoding="utf-8") as fh:
            for line in fh:
                m = key.match(line.strip())
                if m:
                    return int(m.group(1))
    except OSError:
        logger.debug("Could not read %s for adb port", config_path, exc_info=True)
    return None


def _parse_devices(stdout: str) -> list:
    """Serials from `adb devices` output that are in the 'device' state."""
    serials = []
    for line in stdout.splitlines()[1:]:          # skip "List of devices attached"
        parts = line.split()
        if len(parts) >= 2 and parts[1] == "device":
            serials.append(parts[0])
    return serials


def _resolve_serial(adb_exe: str, port: Optional[int], runner: Runner) -> str:
    """The ADB serial of the target instance, connecting by ``port`` if given,
    else falling back to the sole attached device. Raises with a user-facing
    message when nothing usable is reachable."""
    if port:
        serial = "127.0.0.1:%d" % port
        cp = runner([adb_exe, "connect", serial])
        out = (cp.stdout or "") + (cp.stderr or "")
        if "connected" in out.lower():
            return serial
        logger.debug("adb connect output: %s", out.strip())  # fall through

    cp = runner([adb_exe, "devices"])
    devices = _parse_devices(cp.stdout or "")
    if not devices:
        raise RuntimeError(
            "No running BlueStacks instance was reachable over ADB. Start the "
            "instance, let it reach the home screen, then try again.")
    if len(devices) > 1:
        raise RuntimeError(
            "Multiple instances are running and the target's ADB port could not "
            "be identified. Close the others and retry, or start only the target "
            "instance.")
    return devices[0]


def wait_until_ready(adb_exe: str, port: Optional[int], timeout: int = 240,
                     progress=None, runner: Runner = _run,
                     sleep=time.sleep) -> Optional[str]:
    """Wait for a just-launched instance to finish booting; its serial, or None.

    ADB accepts a connection well before Android reaches the home screen, and
    installing an app into a half-booted system fails, so readiness means both
    a connectable device *and* ``sys.boot_completed=1``. Polls instead of
    blocking so a caller can keep reporting progress, and gives up at ``timeout``
    rather than hanging a background job forever.
    """
    target = "127.0.0.1:%d" % port if port else None
    interval = 5
    waited = 0
    while waited < timeout:
        if target:
            runner([adb_exe, "connect", target])
        serials = _parse_devices(runner([adb_exe, "devices"]).stdout or "")
        # Prefer the instance's own port; a second transport (emulator-5554) is
        # usually the same guest, so it is an acceptable fallback.
        for serial in ([target] if target in serials else serials):
            cp = runner([adb_exe, "-s", serial, "shell", "getprop", "sys.boot_completed"])
            if (cp.stdout or "").strip() == "1":
                return serial
        sleep(interval)
        waited += interval
        if progress:
            progress("Waiting for the instance to finish booting (%ds)..." % waited)
    return None


def _ensure_su_policy(adb_exe: str, serial: str, runner: Runner) -> None:
    """Re-affirm the permanent 'allow' Superuser policy for the shell uid (2000)
    so magiskd never prompts -- and, on a missed prompt, auto-DENIES -- ``su``
    during a flash.

    The real fix lives offline: the installer plants a Magisk ``service.d``
    script (magisk_system) that runs as root at every boot and sets this same
    policy, so a fresh instance grants the shell from first boot with no tap and
    a missed prompt can never lock it out. This online call is a silent
    belt-and-suspenders re-affirm for instances that already have su; it can't
    bootstrap a shell that's currently denied (it would need the very su it's
    being denied), which is exactly why the grant is planted offline.

    Best-effort: any failure is swallowed so it never aborts the flash itself.
    """
    sql = ("REPLACE INTO policies (uid,policy,until,logging,notification) "
           "VALUES(2000,2,0,1,1)")  # policy 2 = allow, until 0 = forever
    try:
        runner([adb_exe, "-s", serial, "shell", "su", "-c",
                'magisk --sqlite "%s"' % sql])
    except Exception:  # noqa: BLE001 - a pre-grant failure must not stop the flash
        logger.debug("ensure_su_policy failed (non-fatal)", exc_info=True)


def _shell_single_quote(s: str) -> str:
    """Escape ``s`` for embedding inside a single-quoted POSIX shell argument
    (the standard technique: close the quote, emit an escaped literal quote,
    reopen the quote). ``su -c`` hands its argument to a shell on the guest, so
    a filename containing a literal ``'`` would otherwise break out of the
    ``'%s'`` it's substituted into below and inject arbitrary shell commands."""
    return s.replace("'", "'\\''")


def magisk_version_code(adb_exe: str, serial: str, runner: Runner = _run) -> Optional[int]:
    """The running daemon's MAGISK_VER_CODE, or None if it can't be read.

    ``magisk -V`` prints the numeric version code and nothing else. None means
    "couldn't tell" (no root shell, Magisk absent, unexpected output) and must
    never be treated as "too old" -- a version gate that fires on an unreadable
    version would block flashing on any instance whose shell is momentarily
    unavailable.
    """
    try:
        cp = runner([adb_exe, "-s", serial, "shell", "su", "-c", "magisk -V"])
    except Exception:  # noqa: BLE001 - unreadable version is not a failure
        logger.debug("magisk -V failed", exc_info=True)
        return None
    out = ((cp.stdout or "") + (cp.stderr or "")).strip()
    m = re.search(r"\b(\d{4,6})\b", out)
    return int(m.group(1)) if m else None


def install_module(adb_exe: str, port: Optional[int], local_zip: str,
                   progress: Optional[Callable[[str], None]] = None,
                   runner: Runner = _run,
                   min_magisk_ver_code: Optional[int] = None) -> str:
    """Push ``local_zip`` to a running instance and flash it via Magisk directly.

    Runs ``magisk --install-module`` over an ADB root shell (the same command we
    flash by hand). On success the module is installed and only needs a reboot.
    If the root shell / Magisk isn't reachable, the zip is left in the guest's
    Download folder and a RuntimeError explains how to flash it manually.

    ``min_magisk_ver_code`` is the module's own MAGISK_VER_CODE requirement (its
    ``customize.sh`` enforces one and aborts mid-flash otherwise). Checking it
    here turns that into a clear refusal before anything is pushed. Callers that
    have no requirement pass nothing and no check runs; an *unreadable* version
    also proceeds, since a gate that fires on "couldn't tell" would block flashes
    on a healthy instance.
    """
    def _p(msg):
        logger.info(msg)
        if progress:
            progress(msg)

    if not os.path.isfile(local_zip):
        raise RuntimeError("Module file not found: %s" % local_zip)

    name = os.path.basename(local_zip)
    _p("Connecting to the instance...")
    serial = _resolve_serial(adb_exe, port, runner)

    # Re-affirm shell root so the flash isn't auto-denied. On a tool-installed
    # instance the offline service.d grant already did this at boot, so no prompt
    # appears; this is a silent belt-and-suspenders confirm.
    _p("Confirming ADB root access...")
    _ensure_su_policy(adb_exe, serial, runner)

    if min_magisk_ver_code is not None:
        have = magisk_version_code(adb_exe, serial, runner)
        if have is not None and have < min_magisk_ver_code:
            raise RuntimeError(
                "%s needs Magisk %d or newer, but this instance is running %d. "
                "Update the root payload first -- flashing now would fail partway "
                "through the module's own install script."
                % (name, min_magisk_ver_code, have))

    tmp = "/data/local/tmp/" + name
    _p("Pushing %s..." % name)
    cp = runner([adb_exe, "-s", serial, "push", local_zip, tmp])
    if cp.returncode != 0:
        raise RuntimeError("ADB push failed: %s"
                           % ((cp.stdout or "") + (cp.stderr or "")).strip())

    _p("Installing %s via Magisk..." % name)
    cp = runner([adb_exe, "-s", serial, "shell", "su", "-c",
                 "magisk --install-module '%s'" % _shell_single_quote(tmp)])
    out = ((cp.stdout or "") + (cp.stderr or "")).strip()
    runner([adb_exe, "-s", serial, "shell", "rm", "-f", tmp])  # tidy up

    if cp.returncode == 0:
        return "Installed \"%s\". Close and reopen the instance to activate it." % name

    # Couldn't auto-install (no root shell, Magisk not on PATH, module rejected):
    # leave the zip where the user can flash it by hand and say so.
    _p("Direct install failed; copying to Download for manual flashing...")
    runner([adb_exe, "-s", serial, "push", local_zip, "/sdcard/Download/"])
    raise RuntimeError(
        "Couldn't install automatically (%s). If that's a root-permission "
        "rejection, set Magisk/Kitsune -> Settings -> Superuser access -> "
        "\"Apps and ADB\" and try again. The zip was also copied to the "
        "instance's Download folder -- or flash it there: Modules -> Install "
        "from storage -> Download." % (out or "unknown error"))


def install_manager(adb_exe: str, port: Optional[int], apk_path: str,
                    progress: Optional[Callable[[str], None]] = None,
                    runner: Runner = _run_install) -> str:
    """Install the Magisk/Kitsune manager APK into a *running* instance as a
    normal user app (``adb install -r``).

    The full manager can't be placed offline: its stub self-downloads its UI
    from a dead URL (greyed app), and a manager under /system/app trips Magisk's
    "Abnormal State: system app not supported". So the offline root install
    writes only the genuine stub footprint, and the manager is ``pm install``'d
    here after first boot. Offline root works without it; this just gives the
    user the app. Returns a status line; raises with a user-facing message on
    failure.
    """
    def _p(msg):
        logger.info(msg)
        if progress:
            progress(msg)

    if not os.path.isfile(apk_path):
        raise RuntimeError("Manager APK not found: %s" % apk_path)

    _p("Connecting to the instance...")
    serial = _resolve_serial(adb_exe, port, runner)

    _p("Installing the Magisk manager (%s)..." % os.path.basename(apk_path))
    cp = runner([adb_exe, "-s", serial, "install", "-r", apk_path])
    out = ((cp.stdout or "") + (cp.stderr or "")).strip()
    if cp.returncode == 0 and "Success" in out:
        # Flush the install to the vhdx. BlueStacks caches /data host-side, so a
        # hard-kill restart (the app's kill-and-resume) drops an unflushed
        # pm-install -- the manager then "vanishes" after the reboot. A sync makes
        # it durable now. (Magisk's --install-module already syncs, which is why
        # modules survive the restart and the manager did not.) Verified live:
        # without this the manager is gone after restart; with it, it persists.
        runner([adb_exe, "-s", serial, "shell", "sync"])
        return "Installed the Magisk manager. Open it from the app drawer."

    low = out.lower()
    if "signatures do not match" in low or "update_incompatible" in low:
        raise RuntimeError(
            "A different-signed Magisk manager is already installed. Remove it "
            "first -- \"Remove manager\" on the Magisk tab if this app installed "
            "it, otherwise `adb uninstall %s` or Android Settings -> Apps -- then "
            "retry. Details: %s" % (MANAGER_PACKAGE, out))
    raise RuntimeError("Manager install failed: %s" % (out or "unknown error"))


def uninstall_manager(adb_exe: str, port: Optional[int],
                      progress: Optional[Callable[[str], None]] = None,
                      runner: Runner = _run) -> str:
    """Remove the Magisk/Kitsune manager from a running instance. Idempotent:
    "not installed" is reported as success, not an error."""
    def _p(msg):
        logger.info(msg)
        if progress:
            progress(msg)

    _p("Connecting to the instance...")
    serial = _resolve_serial(adb_exe, port, runner)
    _p("Removing the Magisk manager...")
    cp = runner([adb_exe, "-s", serial, "uninstall", MANAGER_PACKAGE])
    out = ((cp.stdout or "") + (cp.stderr or "")).strip()
    if cp.returncode == 0 and "Success" in out:
        return "Removed the Magisk manager."
    if "not installed" in out.lower() or "unknown package" in out.lower():
        return "Magisk manager was not installed."
    raise RuntimeError("Manager uninstall failed: %s" % (out or "unknown error"))


def list_running_instances(adb_exe: str, instances, runner: Runner = _run) -> dict:
    """Which of ``instances`` are currently reachable over ADB.

    ``instances`` is an iterable of (unique_id, config_path, original_name).
    Returns {unique_id: port} for every instance whose configured ADB port
    is currently connected. Instances with no recorded port (never booted)
    are skipped without spawning a process.
    """
    running = {}
    for unique_id, config_path, name in instances:
        port = instance_adb_port(config_path, name)
        if port is None:
            continue
        try:
            cp = runner([adb_exe, "connect", "127.0.0.1:%d" % port])
            out = (cp.stdout or "") + (cp.stderr or "")
            if "connected" in out.lower():
                running[unique_id] = port
        except Exception as exc:  # noqa: BLE001 - one bad instance mustn't abort the rest
            logger.warning("ADB probe failed for %s: %s", unique_id, exc)
    return running