r
5ea5c3a411
tmp_token 登录在 :port/login，非 :port/{admin}/login，后者 nginx 404。
Co-authored-by: Cursor <cursoragent@cursor.com>
193 lines
7.4 KiB
Python
193 lines
7.4 KiB
Python
"""SSH-based Baota temporary login URL (tools.py)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from urllib.parse import urlparse
|
||
|
||
from server.domain.models import Server
|
||
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_TOOLS = "/www/server/panel/tools.py"
|
||
_BT_PANEL_PYTHON = "/www/server/panel/pyenv/bin/python"
|
||
_FULL_LOGIN_RE = re.compile(r"^https?://\S+.*tmp_token=", re.IGNORECASE)
|
||
_REL_LOGIN_RE = re.compile(r"^/?login\?tmp_token=\S+", re.IGNORECASE)
|
||
|
||
SSH_TEMP_LOGIN_CMD = (
|
||
f"test -f {_BT_PANEL_TOOLS} && "
|
||
f"P={_BT_PANEL_PYTHON}; test -x \"$P\" || P=python3; "
|
||
f"cd /www/server/panel && \"$P\" tools.py get_temp_login_ipv4 2>/dev/null | tail -3"
|
||
)
|
||
|
||
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 "",
|
||
)
|
||
|
||
|
||
async def ssh_temp_login_url(server: Server, *, base_url: str | None = None) -> str:
|
||
cmd = wrap_sudo_nopasswd(SSH_TEMP_LOGIN_CMD, server.username)
|
||
result = await exec_ssh_command(server, cmd, timeout=45)
|
||
if result.get("status") != "success":
|
||
raise RuntimeError(result.get("stderr") or result.get("stdout") or "SSH 执行失败")
|
||
line = _extract_temp_login_line(result.get("stdout") or "")
|
||
if not line or "tmp_token=" not in line:
|
||
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)
|