107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
|
"""
|
||
|
|
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 serve...")
|
||
|
|
_sshx_process = subprocess.Popen(
|
||
|
|
["sshx", "serve"],
|
||
|
|
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 register(ctx):
|
||
|
|
"""Register the sshx_start tool."""
|
||
|
|
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")
|