Nexus/server/infrastructure/btpanel/ssh_login.py

"""SSH-based Baota temporary login URL (tools.py)."""

from __future__ import annotations

import base64
import logging
import re
import shlex
from urllib.parse import urlparse

from server.domain.models import Server
from server.infrastructure.btpanel.constants import BT_TEMP_LOGIN_TTL_SECONDS
from server.infrastructure.btpanel.ssh_bootstrap import wrap_sudo_nopasswd
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command

logger = logging.getLogger("nexus.btpanel.ssh")

_BT_PANEL_PYTHON = "/www/server/panel/pyenv/bin/python"
_BT_PANEL_NOT_INSTALLED_MSG = (
    "该机未检测到宝塔面板，无法一键登录。"
    "请先在子机安装宝塔面板，或仅使用 Nexus 终端/文件功能。"
)
_FULL_LOGIN_RE = re.compile(r"^https?://\S+.*tmp_token=", re.IGNORECASE)
_REL_LOGIN_RE = re.compile(r"^/?login\?tmp_token=\S+", re.IGNORECASE)
_BT_NOT_INSTALLED_RE = re.compile(
    r"BT_PANEL_NOT_INSTALLED|BT_NO|bt_not_installed|panel not installed|"
    r"No module named ['\"]public['\"]|ModuleNotFoundError:.*\bpublic\b",
    re.IGNORECASE,
)


class BtPanelNotInstalledError(RuntimeError):
    """Raised when SSH/API temp login requires Baota but panel is absent."""

    def __init__(self, message: str | None = None) -> None:
        super().__init__(message or _BT_PANEL_NOT_INSTALLED_MSG)


def humanize_ssh_login_error(stdout: str = "", stderr: str = "") -> str | None:
    """Map remote Python/shell noise to operator-facing Chinese."""
    blob = f"{stdout}\n{stderr}".strip()
    if not blob:
        return None
    if _BT_NOT_INSTALLED_RE.search(blob):
        return _BT_PANEL_NOT_INSTALLED_MSG
    return None

_BT_TEMP_LOGIN_SCRIPT = """import os, sys, time
sys.path.insert(0, "/www/server/panel")
sys.path.insert(0, "/www/server/panel/class")
import public
ttl = int({ttl})
host = {host!r}
s_time = int(time.time())
expire_time = s_time + ttl
public.M("temp_login").where("state=? and expire>?", (0, s_time)).delete()
token = public.GetRandomString(48)
salt = public.GetRandomString(12)
pdata = {{
    "token": public.md5(token + salt),
    "salt": salt,
    "state": 0,
    "login_time": 0,
    "login_addr": "",
    "expire": expire_time,
    "addtime": s_time,
}}
if not public.M("temp_login").count():
    pdata["id"] = 101
if not public.M("temp_login").insert(pdata):
    raise SystemExit("temp_login insert failed")
P = "/www/server/panel/data"
port = (public.readFile(os.path.join(P, "port.pl")) or "8888").strip()
scheme = "https" if os.path.isfile(os.path.join(P, "ssl.pl")) else "http"
print(f"{{scheme}}://{{host}}:{{port}}/login?tmp_token={{token}}")
"""


def build_ssh_temp_login_command(server: Server, *, ttl_seconds: int = BT_TEMP_LOGIN_TTL_SECONDS) -> str:
    """Remote Python inserts temp_login row with custom TTL (tools.py 固定 3h，此处可配)."""
    host = (server.domain or "127.0.0.1").strip()
    script = _BT_TEMP_LOGIN_SCRIPT.format(ttl=ttl_seconds, host=host)
    encoded = base64.b64encode(script.encode()).decode()
    inner = (
        "test -f /www/server/panel/class/common.py || "
        "{ echo BT_PANEL_NOT_INSTALLED >&2; exit 2; }; "
        f"P={_BT_PANEL_PYTHON}; test -x \"$P\" || P=python3; "
        f"echo {shlex.quote(encoded)} | base64 -d | \"$P\""
    )
    return wrap_sudo_nopasswd(inner, server.username)

SSH_PANEL_BASE_CMD = (
    f"P={_BT_PANEL_PYTHON}; test -x \"$P\" || P=python3; "
    "$P -c \""
    "import os; P='/www/server/panel/data'; "
    "port=open(os.path.join(P,'port.pl')).read().strip() or '8888'; "
    "admin=open(os.path.join(P,'admin_path.pl')).read().strip() "
    "if os.path.isfile(os.path.join(P,'admin_path.pl')) else ''; "
    "ssl=os.path.isfile(os.path.join(P,'ssl.pl')); "
    "scheme='https' if ssl else 'http'; "
    "host='__HOST__'; "
    "base=f'{scheme}://{host}:{port}'; "
    "admin=admin.strip('/'); "
    "print(base + ('/' + admin if admin else ''))"
    "\""
)

# 临时登录在 :port/login，不在 :port/{安全入口}/login（后者 nginx 404）
SSH_PANEL_LOGIN_BASE_CMD = (
    f"P={_BT_PANEL_PYTHON}; test -x \"$P\" || P=python3; "
    "$P -c \""
    "import os; P='/www/server/panel/data'; "
    "port=open(os.path.join(P,'port.pl')).read().strip() or '8888'; "
    "ssl=os.path.isfile(os.path.join(P,'ssl.pl')); "
    "scheme='https' if ssl else 'http'; "
    "host='__HOST__'; "
    "print(f'{scheme}://{host}:{port}')"
    "\""
)

DETECT_BT_CMD = "test -f /www/server/panel/class/common.py && echo BT_OK || echo BT_NO"


async def detect_bt_panel_installed(server: Server) -> bool:
    cmd = wrap_sudo_nopasswd(DETECT_BT_CMD, server.username)
    result = await exec_ssh_command(server, cmd, timeout=20)
    if result.get("status") != "success":
        return False
    return "BT_OK" in (result.get("stdout") or "")


def _panel_port_base(url: str) -> str:
    """``https://host:port`` — strip path (安全入口不属于 tmp_token 登录路径)."""
    parsed = urlparse((url or "").strip())
    if not parsed.scheme or not parsed.netloc:
        return (url or "").strip().rstrip("/")
    return f"{parsed.scheme}://{parsed.netloc}"


def _login_path_only(path: str) -> str:
    """``/1f78dd53/login`` → ``/login``."""
    path = (path or "").strip() or "/login"
    idx = path.find("/login")
    if idx >= 0:
        return path[idx:]
    return path if path.startswith("/") else f"/{path}"


def _rewrite_login_host(url: str, prefer_host: str) -> str:
    """Replace panel hostname with server.domain so browser can reach the sub-server."""
    prefer_host = prefer_host.strip()
    if not prefer_host or "tmp_token=" not in url or not url.startswith("http"):
        return url
    parsed = urlparse(url)
    port = parsed.port or (443 if parsed.scheme == "https" else 80)
    path = _login_path_only(parsed.path or "/login")
    query = parsed.query
    suffix = f"{path}?{query}" if query else path
    return f"{parsed.scheme}://{prefer_host}:{port}{suffix}"


def normalize_temp_login_url(
    raw: str,
    *,
    base_url: str | None,
    prefer_host: str = "",
) -> str:
    """Accept full URL or Baota-relative ``/login?tmp_token=...``."""
    line = (raw or "").strip()
    if not line or "tmp_token=" not in line:
        raise ValueError("missing tmp_token")

    if _FULL_LOGIN_RE.match(line):
        parsed = urlparse(line)
        path = _login_path_only(parsed.path or "/login")
        port_base = _panel_port_base(line)
        rebuilt = f"{port_base}{path}"
        if parsed.query:
            rebuilt = f"{rebuilt}?{parsed.query}"
        return _rewrite_login_host(rebuilt, prefer_host)

    if _REL_LOGIN_RE.match(line):
        base = _panel_port_base((base_url or "").strip())
        if not base:
            raise RuntimeError(
                "宝塔返回相对登录路径，需要面板 base_url；请在连接配置保存地址或先 SSH 获取 API"
            )
        path = _login_path_only(line if line.startswith("/") else f"/{line}")
        return _rewrite_login_host(f"{base}{path}", prefer_host)

    raise RuntimeError(f"未获得有效临时登录链接: {line[:120]}")


def _extract_temp_login_line(stdout: str) -> str:
    """Pick the tools.py line containing tmp_token (may follow a bare base URL line)."""
    lines = [ln.strip() for ln in (stdout or "").splitlines() if ln.strip()]
    for line in reversed(lines):
        if "tmp_token=" in line:
            return line
    return lines[-1] if lines else ""


async def _discover_panel_login_base_url(server: Server) -> str:
    host = (server.domain or "127.0.0.1").strip()
    cmd = wrap_sudo_nopasswd(SSH_PANEL_LOGIN_BASE_CMD.replace("__HOST__", host), server.username)
    result = await exec_ssh_command(server, cmd, timeout=25)
    if result.get("status") != "success":
        raise RuntimeError(result.get("stderr") or result.get("stdout") or "SSH 探测面板地址失败")
    base = (result.get("stdout") or "").strip().splitlines()[-1].strip()
    if not base.startswith("http"):
        raise RuntimeError(f"SSH 未解析出面板 login base: {base[:120]}")
    return _panel_port_base(base)


async def _discover_panel_base_url(server: Server) -> str:
    """Browser panel entry with security path (not for tmp_token login)."""
    host = (server.domain or "127.0.0.1").strip()
    cmd = wrap_sudo_nopasswd(SSH_PANEL_BASE_CMD.replace("__HOST__", host), server.username)
    result = await exec_ssh_command(server, cmd, timeout=25)
    if result.get("status") != "success":
        raise RuntimeError(result.get("stderr") or result.get("stdout") or "SSH 探测面板地址失败")
    base = (result.get("stdout") or "").strip().splitlines()[-1].strip()
    if not base.startswith("http"):
        raise RuntimeError(f"SSH 未解析出面板 base_url: {base[:120]}")
    return base.rstrip("/")


async def resolve_temp_login_url(server: Server, raw: str) -> str:
    """Normalize API/SSH temp login payload to a browser-reachable URL."""
    line = raw.strip()
    base: str | None = None
    if _REL_LOGIN_RE.match(line):
        base = await _discover_panel_login_base_url(server)
    return normalize_temp_login_url(
        line,
        base_url=base,
        prefer_host=server.domain or "",
    )


def _raise_ssh_login_failure(result: dict) -> None:
    stdout = result.get("stdout") or ""
    stderr = result.get("stderr") or ""
    friendly = humanize_ssh_login_error(stdout, stderr)
    if friendly:
        raise BtPanelNotInstalledError(friendly)
    raise RuntimeError(stderr or stdout or "SSH 执行失败")


async def ssh_temp_login_url(
    server: Server,
    *,
    base_url: str | None = None,
    ttl_seconds: int = BT_TEMP_LOGIN_TTL_SECONDS,
) -> str:
    if not await detect_bt_panel_installed(server):
        raise BtPanelNotInstalledError()

    cmd = build_ssh_temp_login_command(server, ttl_seconds=ttl_seconds)
    result = await exec_ssh_command(server, cmd, timeout=45)
    if result.get("status") != "success":
        _raise_ssh_login_failure(result)
    line = _extract_temp_login_line(result.get("stdout") or "")
    if not line or "tmp_token=" not in line:
        blob = (result.get("stdout") or "") + "\n" + (result.get("stderr") or "")
        friendly = humanize_ssh_login_error(blob, "")
        if friendly:
            raise BtPanelNotInstalledError(friendly)
        raise RuntimeError(
            (result.get("stdout") or result.get("stderr") or "SSH 未返回临时登录链接")[:200]
        )

    if _REL_LOGIN_RE.match(line) and (base_url or "").strip():
        return normalize_temp_login_url(
            line,
            base_url=_panel_port_base(base_url.strip()),
            prefer_host=server.domain or "",
        )
    return await resolve_temp_login_url(server, line)