hermes-sshx-link/dashboard/plugin_api.py

import subprocess
import os
import logging
import time
import platform
import stat
import urllib.request
import re
import tarfile
import tempfile
import shutil
import select
from typing import Optional
from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)
router = APIRouter()

_sshx_link = None
_sshx_process: Optional[subprocess.Popen] = None
_sshx_installing = False
_sshx_bin = "/tmp/sshx"
_sshx_bin_alt = os.path.expanduser("~/.local/bin/sshx")
_sshx_tar = "/tmp/sshx.tar.gz"
_sshx_last_output = ""
_sshx_last_error = ""

LINK_RE = re.compile(r"https://sshx\.io/s/[A-Za-z0-9\-_]+(?:#[^\s\"']*)?")


def _detect_sshx_arch():
    machine = platform.machine().lower()
    system = platform.system().lower()
    if system == "darwin":
        suffix = "-apple-darwin"
        if machine in ("aarch64", "arm64", "armv8l", "armv8b"):
            arch = "aarch64"
        elif machine in ("x86_64", "x64", "amd64"):
            arch = "x86_64"
        else:
            arch = "aarch64" if "arm" in machine else "x86_64"
        return arch, suffix
    suffix = "-unknown-linux-musl"
    if machine in ("aarch64", "aarch64_be", "arm64", "armv8b", "armv8l"):
        arch = "aarch64"
    elif machine in ("x86_64", "x64", "amd64"):
        arch = "x86_64"
    elif machine == "armv6l":
        arch = "arm"
        suffix += "eabihf"
    elif machine == "armv7l":
        arch = "armv7"
        suffix += "eabihf"
    else:
        arch = "x86_64"
    return arch, suffix


def _get_sshx_url() -> str:
    arch, suffix = _detect_sshx_arch()
    url = f"https://s3.amazonaws.com/sshx/sshx-{arch}{suffix}.tar.gz"
    return url


def _find_sshx_bin() -> Optional[str]:
    for p in [_sshx_bin, _sshx_bin_alt, shutil.which("sshx")]:
        if p and os.path.exists(p) and os.access(p, os.X_OK):
            return p
    return None


def _is_alive() -> bool:
    global _sshx_process
    if not _sshx_process:
        return False
    return _sshx_process.poll() is None


@router.get("/status")
async def sshx_status():
    global _sshx_link, _sshx_installing, _sshx_process
    # passive mode: only report current link, no auto-start
    # also try to detect externally running sshx (not managed by plugin) via ps
    if _is_alive() and _sshx_link:
        return {"status": "running", "link": _sshx_link, "pid": _sshx_process.pid}
    if _sshx_installing:
        return {"status": "installing", "link": None}
    if _is_alive() and not _sshx_link:
        return {"status": "starting", "link": None, "pid": _sshx_process.pid}
    # no auto background process - just report stopped, frontend will only show link if exists
    # also check if link was persisted to /tmp/sshx_link.txt by external runner
    try:
        if os.path.exists("/tmp/sshx_link.txt"):
            with open("/tmp/sshx_link.txt", "r") as f:
                txt = f.read().strip()
                m = LINK_RE.search(txt)
                if m:
                    return {"status": "running", "link": m.group(0), "pid": None, "source": "file"}
    except Exception:
        pass
    return {"status": "stopped", "link": None}


@router.get("/debug")
async def debug_info():
    return {
        "bin": _find_sshx_bin(),
        "is_alive": _is_alive(),
        "pid": _sshx_process.pid if _sshx_process and _is_alive() else None,
        "link": _sshx_link,
        "last_error": _sshx_last_error,
        "last_output_tail": _sshx_last_output[-600:] if _sshx_last_output else "",
        "platform": platform.platform(),
    }


# keep /start for manual use but do NOT auto-call it; hermes kills background shell so keep it minimal
@router.get("/start")
async def start_sshx(request: Request):
    return JSONResponse(status_code=410, content={"status": "disabled", "message": "Auto background sshx disabled — hermes kills shell. Run sshx manually on host (e.g. nohup sshx --quiet > /tmp/sshx_link.txt 2>&1 &) and dashboard will display the link from /tmp/sshx_link.txt"})

@router.post("/start")
async def start_sshx_post(request: Request):
    return await start_sshx(request)

@router.post("/stop")
async def stop_sshx():
    global _sshx_process, _sshx_link
    _sshx_process = None
    _sshx_link = None
    return {"status": "stopped"}

@router.get("/stop")
async def stop_sshx_get():
    global _sshx_process, _sshx_link
    _sshx_process = None
    _sshx_link = None
    return {"status": "stopped"}

@router.get("/restart-dashboard")
async def restart_dashboard():
    return {"status": "deprecated", "message": "No restart needed."}

@router.post("/restart-dashboard")
async def restart_dashboard_post():
    return {"status": "deprecated", "message": "No restart needed."}