hermes-sshx-link/__init__.py

"""
sshx-link plugin
================
Downloads sshx, starts a shared terminal session,
and returns the shareable link.
"""

import subprocess
import threading
import re
import logging
import shutil
import os

logger = logging.getLogger(__name__)

_sshx_process = None
_sshx_link = None


def _find_link_from_output(line: str) -> str | None:
    """Extract sshx link from process output."""
    # sshx outputs links like: https://sshx.io/s/xxxxx
    match = re.search(r"https://sshx\.io/s/[A-Za-z0-9_-]+", line)
    return match.group(0) if match else None


def _ensure_sshx_installed() -> bool:
    """Download and install sshx binary if not present."""
    if shutil.which("sshx"):
        logger.info("sshx already installed at %s", shutil.which("sshx"))
        return True

    logger.info("Downloading sshx...")
    try:
        result = subprocess.run(
            ["sh", "-c", "curl -sSf https://sshx.io/get | sh"],
            capture_output=True,
            text=True,
            timeout=60,
        )
        if result.returncode != 0:
            logger.error("sshx install failed: %s", result.stderr)
            return False
        return True
    except subprocess.TimeoutExpired:
        logger.error("sshx install timed out")
        return False


def _run_sshx(tool_ctx):
    """Run sshx serve in background, capture the link."""
    global _sshx_process, _sshx_link

    if _sshx_process and _sshx_process.poll() is None:
        # Already running
        tool_ctx.yield_result({"status": "already_running", "link": _sshx_link})
        return

    if not _ensure_sshx_installed():
        tool_ctx.yield_result({"status": "error", "message": "Failed to install sshx"})
        return

    logger.info("Starting sshx...")
    _sshx_process = subprocess.Popen(
        ["sshx", "--quiet"],
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
    )

    _sshx_link = None

    def _reader():
        global _sshx_link
        for line in _sshx_process.stdout:
            line = line.strip()
            logger.debug("sshx: %s", line)
            link = _find_link_from_output(line)
            if link:
                _sshx_link = link
                tool_ctx.yield_result({"status": "running", "link": link})
                return
        # Process ended without outputting a link
        tool_ctx.yield_result({"status": "error", "message": "sshx exited without producing a link"})

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


def _auto_install_sshx():
    if shutil.which("sshx"):
        logger.info("sshx already installed at %s", shutil.which("sshx"))
        return
    logger.info("Auto-installing sshx...")
    try:
        result = subprocess.run(
            ["sh", "-c", "curl -sSf https://sshx.io/get | sh"],
            capture_output=True, text=True, timeout=60,
        )
        if result.returncode == 0 and shutil.which("sshx"):
            logger.info("sshx installed successfully")
        else:
            logger.error("sshx install failed: %s", result.stderr)
    except Exception as e:
        logger.error("sshx auto-install error: %s", e)


def register(ctx):
    threading.Thread(target=_auto_install_sshx, daemon=True).start()
    ctx.register_tool(
        name="sshx_start",
        toolset="sshx-link",
        schema={
            "description": "Start a sshx shared terminal session and return the shareable link. The link lets others join your terminal via browser.",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
        handler=_run_sshx,
    )
    logger.info("sshx-link plugin registered")