hr-proxy

#!/usr/bin/env python3
"""
hr-proxy — Tiny Headroom proxy session manager (macOS / Linux / Windows).

Cross-platform rewrite of the original bash hr-proxy. Tracks sessions in
~/.headroom-proxy.tsv  (pid TAB port TAB dir TAB started).

Commands:
  hr-proxy start [PORT]     Start a proxy (default 8787), record pid+port+dir
  hr-proxy stop  PORT|PID   Stop a tracked (or port-matched) proxy
  hr-proxy list             Show tracked sessions still alive
  hr-proxy log   [PORT]     Print the log file path / tail for a session
"""

import os
import signal
import socket
import subprocess
import sys
import tempfile
from datetime import datetime

STORE = os.path.expanduser("~/.headroom-proxy.tsv")


def which(name: str) -> str:
    import shutil
    return shutil.which(name) or name


def port_in_use(port: int) -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(0.3)
        return s.connect_ex(("127.0.0.1", port)) == 0


def pid_alive(pid: int) -> bool:
    if os.name == "nt":
        import ctypes
        h = ctypes.windll.kernel32.OpenProcess(0x1000, False, pid)  # QUERY_LIMITED
        if h:
            ctypes.windll.kernel32.CloseHandle(h)
            return True
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # exists, just not ours to signal
    return True


def log_path(port: int) -> str:
    return os.path.join(tempfile.gettempdir(), f"headroom-proxy-{port}.log")


def read_store() -> list[list[str]]:
    if not os.path.exists(STORE):
        return []
    with open(STORE) as f:
        return [ln.rstrip("\n").split("\t") for ln in f if ln.strip()]


def write_store(rows: list[list[str]]) -> None:
    with open(STORE, "w") as f:
        for r in rows:
            f.write("\t".join(r) + "\n")


def clean_store() -> list[list[str]]:
    rows = [r for r in read_store() if len(r) == 4 and pid_alive(int(r[0]))]
    write_store(rows)
    return rows


def kill(pid: int) -> None:
    if os.name == "nt":
        subprocess.run(["taskkill", "/PID", str(pid), "/F"],
                       capture_output=True)
    else:
        os.kill(pid, signal.SIGTERM)


def cmd_start(argv: list[str]) -> int:
    port = int(argv[0]) if argv else 8787
    if port_in_use(port):
        print(f"port {port} already in use", file=sys.stderr)
        return 1
    log = open(log_path(port), "ab")
    kwargs: dict = {"stdout": log, "stderr": log, "stdin": subprocess.DEVNULL}
    if os.name == "nt":
        kwargs["creationflags"] = (subprocess.CREATE_NEW_PROCESS_GROUP
                                   | subprocess.DETACHED_PROCESS)
    else:
        kwargs["start_new_session"] = True
    p = subprocess.Popen([which("headroom"), "proxy", "--port", str(port)],
                         **kwargs)
    rows = clean_store()
    rows.append([str(p.pid), str(port), os.getcwd(),
                 datetime.now().strftime("%Y-%m-%d %H:%M:%S")])
    write_store(rows)
    print(f"started  pid={p.pid}  port={port}  dir={os.getcwd()}")
    print(f"log: {log_path(port)}")
    return 0


def cmd_stop(argv: list[str]) -> int:
    if not argv:
        print("usage: hr-proxy stop PORT|PID", file=sys.stderr)
        return 2
    target = argv[0]
    rows = clean_store()
    pid = next((r[0] for r in rows if r[0] == target or r[1] == target), target)
    try:
        kill(int(pid))
        print(f"stopped pid={pid}")
        write_store([r for r in rows if r[0] != pid])
        return 0
    except (ProcessLookupError, ValueError):
        print(f"not running (pid={pid})", file=sys.stderr)
        return 1


def cmd_list() -> int:
    rows = clean_store()
    if not rows:
        print("no headroom proxy sessions running")
        return 0
    print(f"{'PID':<7}  {'PORT':<5}  {'STARTED':<19}  DIR")
    print("-------  -----  -------------------  ---")
    for pid, port, dir_, started in rows:
        print(f"{pid:<7}  {port:<5}  {started:<19}  {dir_}")
    return 0


def cmd_log(argv: list[str]) -> int:
    port = int(argv[0]) if argv else 8787
    p = log_path(port)
    if not os.path.exists(p):
        print(f"no log at {p}")
        return 1
    print(p)
    with open(p) as f:
        lines = f.readlines()
    sys.stdout.write("".join(lines[-40:]))
    return 0


def main() -> int:
    args = sys.argv[1:]
    cmd = args[0] if args else "list"
    rest = args[1:]
    if cmd == "start":
        return cmd_start(rest)
    if cmd == "stop":
        return cmd_stop(rest)
    if cmd in ("list", "ls"):
        return cmd_list()
    if cmd == "log":
        return cmd_log(rest)
    print(__doc__)
    return 0


if __name__ == "__main__":
    sys.exit(main())