saphid--frame-control/ui/frame_host.py

"""What differs between the computers Frame Control runs on: macOS, Linux, Windows.

Everything here runs on your computer, not the Frame. Python stdlib only.

CLI (used by the Electron app, so terminal handling lives in one place):
  python3 ui/frame_host.py terminal -- CMD [ARG...]   # open CMD in a terminal window
"""
import os
import shlex
import shutil
import ssl
import subprocess
import sys
from pathlib import Path

MAC = sys.platform == "darwin"
WINDOWS = os.name == "nt"
LINUX = not MAC and not WINDOWS
NAME = "macOS" if MAC else "Windows" if WINDOWS else "Linux"
FILE_MANAGER = "Finder" if MAC else "File Explorer" if WINDOWS else "your file manager"

# Windows' OpenSSH client can't share one connection between commands
# (no ControlMaster), so there each command opens its own.
MUX = not WINDOWS

# Popen() keyword arguments that detach a child from our console and signals.
DETACHED = ({"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} if WINDOWS
            else {"start_new_session": True})


class HostError(RuntimeError):
    pass


def data_dir(*parts):
    """Per-user app data: ~/Library/Application Support, %APPDATA% or $XDG_DATA_HOME
    (or $FRAME_CONTROL_DATA_DIR, which the tests point at a throwaway directory)."""
    if os.environ.get("FRAME_CONTROL_DATA_DIR"):
        base = Path(os.environ["FRAME_CONTROL_DATA_DIR"])
    elif MAC:
        base = Path.home() / "Library" / "Application Support" / "Frame Control"
    elif WINDOWS:
        base = Path(os.environ.get("APPDATA") or Path.home() / "AppData" / "Roaming") / "Frame Control"
    else:
        base = Path(os.environ.get("XDG_DATA_HOME") or Path.home() / ".local" / "share") / "frame-control"
    return base.joinpath(*parts)


def cache_dir(*parts):
    if MAC:
        base = Path.home() / "Library" / "Caches" / "Frame Control"
    elif WINDOWS:
        base = Path(os.environ.get("LOCALAPPDATA") or Path.home() / "AppData" / "Local") / "Frame Control" / "Cache"
    else:
        base = Path(os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache") / "frame-control"
    return base.joinpath(*parts)


def control_path(tag="x", *, private=None):
    """ssh ControlPath for the shared connection, or None where it isn't supported.

    `tag` names the headset: ssh's %C hashes only the address, user and port, so two
    headsets reached at the same address (one of them moved) would otherwise share a
    connection, and one's commands would run on the other.
    /tmp, not $TMPDIR: macOS's per-user temp path overflows the unix socket path limit.
    """
    # A private server (the MCP adapter's) keeps its own masters: FRAME_PRIVATE_SSH=1.
    if private is None:
        private = os.environ.get("FRAME_PRIVATE_SSH") == "1"
    suffix = f"-{os.getpid()}" if private else ""
    return f"/tmp/frame-ui-{os.getuid()}{suffix}-{tag}-%C" if MUX else None


def which(name, *extra):
    """First executable among PATH and the extra candidate paths."""
    for cand in (shutil.which(name), *extra):
        if cand and os.path.isfile(cand) and os.access(cand, os.X_OK):
            return cand
    return None


def install_hint(tool):
    """How to get a missing command-line tool on this computer."""
    hints = {
        "adb": {"mac": "brew install android-platform-tools",
                "win": "winget install Google.PlatformTools",
                "linux": "install your distribution's adb package (e.g. sudo apt install adb)"},
    }
    return hints[tool]["mac" if MAC else "win" if WINDOWS else "linux"]


def android_sdk_dirs():
    """Where the Android SDK usually lives, for adb."""
    dirs = [os.environ.get("ANDROID_HOME"), os.environ.get("ANDROID_SDK_ROOT")]
    if MAC:
        dirs += ["~/Library/Android/sdk", "/opt/homebrew/share/android-commandlinetools",
                 "~/.homebrew/share/android-commandlinetools"]
    elif WINDOWS:
        dirs += [os.path.join(os.environ.get("LOCALAPPDATA", ""), "Android", "Sdk")]
    else:
        dirs += ["~/Android/Sdk", "/usr/lib/android-sdk"]
    return [os.path.expanduser(d) for d in dirs if d]


def adb():
    exe = "adb.exe" if WINDOWS else "adb"
    extra = [os.path.join(d, "platform-tools", exe) for d in android_sdk_dirs()]
    if MAC:
        extra += ["/opt/homebrew/bin/adb", str(Path.home() / ".homebrew/bin/adb"), "/usr/local/bin/adb"]
    # The app bundles adb as a last resort: an adb you already use goes first, so
    # two different adb versions don't keep restarting each other's server.
    tools = os.environ.get("FRAME_CONTROL_TOOLS")
    if tools:
        extra.append(os.path.join(tools, exe))
    env = os.environ.get("ADB")
    found = (env if env and os.access(env, os.X_OK) else None) or which("adb", *extra)
    if not found:
        raise HostError(f"adb isn't installed on this computer: {install_hint('adb')}")
    return found


def trust_bundled_cas():
    """Trust the app's CA bundle for HTTPS as well as the system's certificates.

    Python on Windows only sees the root certificates already in the Windows
    store, and a fresh install fetches those lazily, so Steam and F-Droid can
    fail with CERTIFICATE_VERIFY_FAILED. The app bundles curl's copy of Mozilla's
    CA list (app/build/fetch-deps.js); outside the app this does nothing. Call it
    before the first urlopen: urllib keeps the HTTPS context it builds then.
    """
    tools = os.environ.get("FRAME_CONTROL_TOOLS")
    cafile = os.path.join(tools, "cacert.pem") if tools else None
    if not cafile or not os.path.isfile(cafile):
        return

    def context(*args, **kwargs):
        ctx = ssl.create_default_context(*args, **kwargs)
        ctx.load_verify_locations(cafile)
        return ctx
    ssl._create_default_https_context = context  # urllib's default for HTTPS


def open_path(path):
    """Show a folder or file in the file manager."""
    path = str(path)
    if WINDOWS:
        os.startfile(path)  # noqa: pylint only on Windows
        return
    opener = "open" if MAC else which("xdg-open")
    if not opener:
        raise HostError("xdg-open isn't installed, so the folder can't be opened")
    subprocess.Popen([opener, path], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, **DETACHED)


def reveal_path(path):
    """Show a file selected in its folder (Linux file managers vary, so there the folder opens)."""
    path = Path(path)
    if MAC:
        cmd = ["open", "-R", str(path)]
    elif WINDOWS:
        cmd = f'explorer /select,"{path}"'  # as one string: Explorer wants the quotes after the comma
    else:
        return open_path(path.parent)
    subprocess.Popen(cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, **DETACHED)


open_url = open_path  # the same openers hand URLs to the default browser


def _spawn(argv):
    subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
                     stderr=subprocess.DEVNULL, **DETACHED)


def open_terminal(argv, title="Frame Control"):
    """Run argv in a new terminal window, for anything that asks for a password.

    The window stays open after the command ends, so its output can be read.
    """
    argv = [str(a) for a in argv]
    if MAC:
        command = shlex.join(argv).replace("\\", "\\\\").replace('"', '\\"')
        r = subprocess.run(["osascript", "-e", 'tell application "Terminal"',
                            "-e", f'do script "{command}"', "-e", "activate", "-e", "end tell"],
                           capture_output=True, text=True)
        if r.returncode != 0:
            # Usually macOS Automation consent for Terminal was denied.
            raise HostError(f"Couldn't open Terminal: {r.stderr.strip()}")
        return "Terminal"
    if WINDOWS:
        # `start` gives the command its own console window; cmd /k keeps it open.
        # One hand-built command line: quoting it twice through list2cmdline would
        # produce backslash-escaped quotes, which cmd doesn't understand.
        # Every argument is quoted, so cmd treats & | < > ^ in them literally. cmd has
        # no escape for a quote inside quotes (and expands %VAR% regardless), so refuse those.
        if any(c in a for a in argv for c in '"%\r\n'):
            raise HostError("Can't pass quotes or % to a Windows terminal")
        inner = " ".join(f'"{a}"' for a in argv)
        subprocess.Popen(f'cmd.exe /c start "{title}" cmd.exe /k "{inner}"', **DETACHED)
        return "a terminal window"
    script = f'{shlex.join(argv)}; echo; read -r -p "Press Enter to close. " _'
    # flags=None: the terminal takes the whole command as one string after -e.
    for name, flags in (("x-terminal-emulator", ["-e"]), ("gnome-terminal", ["--"]), ("ptyxis", ["--"]),
                        ("kgx", ["--"]), ("konsole", ["-e"]), ("xfce4-terminal", ["-x"]),
                        ("tilix", None), ("lxterminal", None), ("kitty", []), ("alacritty", ["-e"]),
                        ("wezterm", ["start", "--"]), ("foot", []), ("xterm", ["-e"])):
        exe = which(name)
        if not exe:
            continue
        if flags is None:
            _spawn([exe, "-e", "bash -c " + shlex.quote(script)])
        elif name == "x-terminal-emulator" and "lxterminal" in os.path.realpath(exe):
            _spawn([exe, "-e", "bash -c " + shlex.quote(script)])  # Debian alternative -> lxterminal
        else:
            _spawn([exe, *flags, "bash", "-c", script])
        return name
    raise HostError("No terminal program found (tried gnome-terminal, konsole, xterm and others)")


def clipboard_text():
    """The text on this computer's clipboard."""
    if MAC:
        cmds = [["pbpaste"]]
    elif WINDOWS:
        cmds = [["powershell.exe", "-NoProfile", "-Command",
                 "[Console]::OutputEncoding=[Text.Encoding]::UTF8; Get-Clipboard -Raw"]]
    else:
        cmds = [["wl-paste", "--no-newline"], ["xclip", "-selection", "clipboard", "-o"],
                ["xsel", "--clipboard", "--output"]]
    for cmd in cmds:
        if not shutil.which(cmd[0]):
            continue
        r = subprocess.run(cmd, capture_output=True, stdin=subprocess.DEVNULL, timeout=10)
        if r.returncode == 0:
            text = r.stdout.decode("utf-8", errors="replace")
            return text[:-2] if WINDOWS and text.endswith("\r\n") else text
    if LINUX:
        raise HostError("Can't read the clipboard: install wl-clipboard (Wayland) or xclip (X11)")
    raise HostError("Can't read the clipboard")


def ssh_hostname(alias):
    """The real host name an ssh alias points at (`ssh -G`), for non-SSH clients like RDP."""
    try:
        out = subprocess.run(["ssh", "-G", alias], capture_output=True, stdin=subprocess.DEVNULL, text=True, timeout=10).stdout
    except (OSError, subprocess.TimeoutExpired):
        return alias
    for line in out.splitlines():
        if line.startswith("hostname "):
            return line.split(None, 1)[1].strip()
    return alias


# Apps the UI can hand off to, per platform: (installed-check, launch argv) pairs,
# and where to get the app when none is installed.
def open_steam_link():
    if MAC:
        if subprocess.run(["open", "-a", "Steam Link"], capture_output=True).returncode == 0:
            return "Opened Steam Link"
    elif WINDOWS:
        for base in (os.environ.get("ProgramFiles(x86)"), os.environ.get("ProgramFiles")):
            exe = base and os.path.join(base, "Steam Link", "SteamLink.exe")
            if exe and os.path.isfile(exe):
                _spawn([exe])
                return "Opened Steam Link"
    else:
        if which("steamlink"):
            _spawn([which("steamlink")])
            return "Opened Steam Link"
        if which("flatpak") and subprocess.run(["flatpak", "info", "com.valvesoftware.SteamLink"],
                                               capture_output=True).returncode == 0:
            _spawn(["flatpak", "run", "com.valvesoftware.SteamLink"])
            return "Opened Steam Link"
    open_url("https://store.steampowered.com/remoteplay")
    return "Steam Link isn't installed; opened its download page"


def open_rdp(alias, host=None):
    """Remote desktop to the Frame's xrdp (user steamos), at `host` or where the alias points."""
    host = host or ssh_hostname(alias)
    if MAC:
        if subprocess.run(["open", "-a", "Windows App"], capture_output=True).returncode == 0:
            return "Opened Windows App"
        open_url("https://apps.apple.com/app/windows-app/id1295203466")
        return "Windows App isn't installed; opened its App Store page"
    if WINDOWS:
        _spawn(["mstsc.exe", f"/v:{host}"])
        return f"Opened Remote Desktop to {host}"
    if which("remmina"):
        _spawn(["remmina", "-c", f"rdp://steamos@{host}"])
        return f"Opened Remmina to {host}"
    for name in ("xfreerdp3", "xfreerdp"):
        if which(name):
            _spawn([name, f"/v:{host}", "/u:steamos", "/dynamic-resolution"])
            return f"Opened FreeRDP to {host}"
    raise HostError("No RDP client found: install Remmina or FreeRDP")


def main(argv):
    if len(argv) >= 3 and argv[0] == "terminal" and argv[1] == "--":
        try:
            print(f"Opened {open_terminal(argv[2:])}")
        except HostError as e:
            sys.exit(str(e))
        return
    sys.exit(__doc__)


if __name__ == "__main__":
    main(sys.argv[1:])