hermes-sshx-link/__init__.py

"""
ttyd-link plugin (replaces sshx)
=================================
Share terminal via ttyd - no curl|sh, auto arch detection, no dashboard restart.
"""

import subprocess
import threading
import logging
import shutil
import os
import platform
import stat
import urllib.request
import socket
import time

logger = logging.getLogger(__name__)

_ttyd_process = None
_ttyd_link = None
_ttyd_port = None

TTYD_BIN = "/tmp/ttyd"
TTYD_DIR = os.path.expanduser("~/.local/bin")
TTYD_BIN_ALT = os.path.join(TTYD_DIR, "ttyd")

# arch mapping for ttyd releases: https://github.com/tsl0922/ttyd/releases
_ARCH_MAP = {
    "x86_64": "x86_64",
    "amd64": "x86_64",
    "aarch64": "aarch64",
    "arm64": "aarch64",
    "armv8l": "aarch64",
    "armv7l": "arm",
    "armv6l": "arm",
    "arm": "arm",
    "armhf": "armhf",
    "i386": "i686",
    "i686": "i686",
    "mips": "mips",
    "mips64": "mips64",
    "mips64el": "mips64el",
    "mipsel": "mipsel",
    "s390x": "s390x",
}

def _detect_arch() -> str:
    machine = platform.machine().lower()
    # handle uname -m variations
    arch = _ARCH_MAP.get(machine, "x86_64")
    logger.info(f"Detected machine={machine} -> ttyd arch={arch}")
    return arch

def _get_ttyd_url() -> str:
    arch = _detect_arch()
    return f"https://github.com/tsl0922/ttyd/releases/latest/download/ttyd.{arch}"

def _find_ttyd_bin() -> str | None:
    for p in [TTYD_BIN, TTYD_BIN_ALT, shutil.which("ttyd")]:
        if p and os.path.exists(p) and os.access(p, os.X_OK):
            return p
    return None

def _ensure_ttyd_installed() -> str | None:
    """Ensure ttyd binary exists, auto-download with arch detection. Returns bin path or None."""
    existing = _find_ttyd_bin()
    if existing:
        logger.info(f"ttyd already installed at {existing}")
        return existing

    url = _get_ttyd_url()
    dest = TTYD_BIN
    # try ~/.local/bin if /tmp not writable or prefer persistence
    try:
        os.makedirs(TTYD_DIR, exist_ok=True)
        # we download to /tmp first then also copy to ~/.local/bin for caching
        logger.info(f"Downloading ttyd from {url} -> {dest}")
        urllib.request.urlretrieve(url, dest)
        os.chmod(dest, os.stat(dest).st_mode | stat.S_IEXEC)

        # cache copy
        try:
            if dest != TTYD_BIN_ALT:
                import shutil as _sh
                _sh.copy2(dest, TTYD_BIN_ALT)
                os.chmod(TTYD_BIN_ALT, os.stat(TTYD_BIN_ALT).st_mode | stat.S_IEXEC)
        except Exception:
            pass
        logger.info(f"ttyd installed successfully to {dest}")
        return dest
    except Exception as e:
        logger.error(f"ttyd install failed from {url}: {e}")
        return None

def _get_free_port() -> int:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("", 0))
    port = s.getsockname()[1]
    s.close()
    return port

def _wait_for_port(port: int, timeout: int = 5) -> bool:
    for _ in range(timeout * 10):
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=1):
                return True
        except OSError:
            time.sleep(0.1)
    return False

def _run_ttyd(tool_ctx):
    """Run ttyd in background, return link."""
    global _ttyd_process, _ttyd_link, _ttyd_port

    if _ttyd_process and _ttyd_process.poll() is None and _ttyd_link:
        tool_ctx.yield_result({"status": "already_running", "link": _ttyd_link, "port": _ttyd_port})
        return

    bin_path = _ensure_ttyd_installed()
    if not bin_path:
        tool_ctx.yield_result({"status": "error", "message": "Failed to install ttyd (auto arch detect failed). Check network to github.com"})
        return

    _ttyd_port = _get_free_port()
    logger.info(f"Starting ttyd on port {_ttyd_port} using {bin_path}...")

    try:
        _ttyd_process = subprocess.Popen(
            [bin_path, "-p", str(_ttyd_port), "-t", "fontSize=14", "bash"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            stdin=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as e:
        logger.error(f"Failed to start ttyd: {e}")
        tool_ctx.yield_result({"status": "error", "message": f"Failed to start ttyd: {e}"})
        return

    if not _wait_for_port(_ttyd_port, timeout=5):
        # process might have died
        if _ttyd_process.poll() is not None:
            tool_ctx.yield_result({"status": "error", "message": "ttyd exited immediately"})
            return
        logger.warning("ttyd port not ready but continuing")

    # construct link - use host detection via socket
    try:
        hostname = socket.gethostname()
        # try to get outward IP, fallback to localhost
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(("8.8.8.8", 80))
            host_ip = s.getsockname()[0]
        except Exception:
            host_ip = "127.0.0.1"
        finally:
            s.close()
    except Exception:
        host_ip = "127.0.0.1"

    _ttyd_link = f"http://{host_ip}:{_ttyd_port}"
    logger.info(f"ttyd running at {_ttyd_link}")
    tool_ctx.yield_result({"status": "running", "link": _ttyd_link, "port": _ttyd_port})


def register(ctx):
    # no background auto-install that blocks - lazy install on demand
    # but we can warm up in thread without requiring restart
    def _warmup():
        try:
            if not _find_ttyd_bin():
                logger.info("Warming up ttyd install (no restart needed)...")
                _ensure_ttyd_installed()
        except Exception as e:
            logger.debug(f"warmup failed: {e}")

    threading.Thread(target=_warmup, daemon=True).start()

    ctx.register_tool(
        name="sshx_start",
        toolset="sshx-link",
        schema={
            "description": "Start a ttyd shared terminal session and return the shareable link. No dashboard restart needed, auto arch detection.",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
        handler=_run_ttyd,
    )
    # also register new name for clarity
    ctx.register_tool(
        name="ttyd_start",
        toolset="sshx-link",
        schema={
            "description": "Start ttyd terminal (alias of sshx_start, auto arch, no restart)",
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
        handler=_run_ttyd,
    )
    logger.info("ttyd-link plugin registered (replaces sshx, no restart)")