leefer
7229642c62
- 向导开场:像素字符画 Logo(❯_ 提示符,青绿+光标绿)+ 版本/平台/配置信息 + 横线 - 运行面板头部加 ❯_ 标识;连接中状态改为 ◐◓◑◒ 旋转动画 - SSH banner 错误改为可读诊断文案;握手超时 12→25s;重连日志带尝试次数
364 lines
13 KiB
Python
364 lines
13 KiB
Python
"""SSH 隧道引擎:基于 paramiko 的本地端口转发,含自动重连与状态事件。"""
|
|
from __future__ import annotations
|
|
|
|
import select
|
|
import socket
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
|
|
import paramiko
|
|
|
|
_POLL = 0.5 # 监听 socket 的 accept 轮询间隔
|
|
_KEEPALIVE = 15 # SSH 保活秒数
|
|
|
|
|
|
def friendly_error(e: Exception) -> str:
|
|
if isinstance(e, paramiko.BadHostKeyException):
|
|
return "主机密钥校验失败:该主机的主机密钥与已记录的不一致(主机可能被重装)。"
|
|
if isinstance(e, paramiko.AuthenticationException):
|
|
return "SSH 认证失败:请检查用户名、密码或私钥。"
|
|
if isinstance(e, paramiko.SSHException):
|
|
msg = str(e)
|
|
if "banner" in msg.lower():
|
|
return ("SSH 握手失败:已连上端口但对方未回应 SSH 协议"
|
|
"(远程主机宕机/过载,或端口映射网关异常)")
|
|
if "negotiation" in msg or "timed out" in msg or "Timeout" in msg:
|
|
return "SSH 握手超时:远程主机无响应。"
|
|
if "not a valid" in msg.lower() or ("key" in msg.lower() and "invalid" in msg.lower()):
|
|
return f"密钥无效:{msg}"
|
|
return f"SSH 错误:{msg}"
|
|
if isinstance(e, socket.gaierror):
|
|
return f"无法解析主机名:{getattr(e, 'args', [''])[0]}"
|
|
if isinstance(e, ConnectionRefusedError):
|
|
return "连接被拒绝:远程主机的 SSH 端口不可达(服务未运行或防火墙拦截)。"
|
|
if isinstance(e, (socket.timeout, TimeoutError)):
|
|
return "连接超时:远程主机无响应。"
|
|
winerr = getattr(e, "winerror", None)
|
|
if winerr == 10048:
|
|
return "本地端口绑定失败:该端口已被其他程序占用。"
|
|
if winerr == 10013:
|
|
return "本地端口绑定失败:权限不足(可换一个 >1024 的端口)。"
|
|
return str(e) or e.__class__.__name__
|
|
|
|
|
|
class Tunnel:
|
|
"""一条本地转发隧道。生命周期:stopped → connecting → connected ⇄ error。"""
|
|
|
|
def __init__(self, cfg: dict, host: dict | None, emit, known_hosts_path):
|
|
self.cfg = dict(cfg)
|
|
self.host = dict(host) if host else None
|
|
self._emit = emit
|
|
self.known_hosts_path = known_hosts_path
|
|
|
|
self.status = "stopped"
|
|
self.error = ""
|
|
self.started_at: float | None = None
|
|
self.conn_count = 0
|
|
self.logs: deque = deque(maxlen=200)
|
|
|
|
self._desired = False
|
|
self._thread: threading.Thread | None = None
|
|
self._client: paramiko.SSHClient | None = None
|
|
self._listener: socket.socket | None = None
|
|
self._wakeup = threading.Event()
|
|
self.attempts = 0 # 本次断线后的重试次数
|
|
|
|
# ---------- 状态 ----------
|
|
|
|
def snapshot(self) -> dict:
|
|
return {
|
|
"id": self.cfg["id"],
|
|
"status": self.status,
|
|
"error": self.error,
|
|
"started_at": self.started_at,
|
|
"conn_count": self.conn_count,
|
|
}
|
|
|
|
def _log(self, level: str, msg: str) -> None:
|
|
entry = {"t": time.strftime("%H:%M:%S"), "level": level, "msg": msg}
|
|
self.logs.append(entry)
|
|
try:
|
|
self._emit({"type": "log", "tunnel_id": self.cfg["id"], "entry": entry})
|
|
except Exception:
|
|
pass
|
|
|
|
def _set(self, status: str, error: str = "") -> None:
|
|
self.status = status
|
|
self.error = error
|
|
try:
|
|
self._emit({"type": "tunnel", **self.snapshot()})
|
|
except Exception:
|
|
pass
|
|
|
|
# ---------- 控制 ----------
|
|
|
|
def start(self) -> None:
|
|
if self._desired:
|
|
return
|
|
if not self.host:
|
|
self._set("error", "隧道未绑定有效的主机")
|
|
return
|
|
self._desired = True
|
|
self._wakeup.clear()
|
|
self._set("connecting")
|
|
self._thread = threading.Thread(
|
|
target=self._run, daemon=True, name=f"tunnel-{self.cfg['id']}"
|
|
)
|
|
self._thread.start()
|
|
|
|
def stop(self) -> None:
|
|
self._desired = False
|
|
self._wakeup.set()
|
|
listener, self._listener = self._listener, None
|
|
if listener:
|
|
try:
|
|
listener.close()
|
|
except OSError:
|
|
pass
|
|
client, self._client = self._client, None
|
|
if client:
|
|
try:
|
|
client.close()
|
|
except Exception:
|
|
pass
|
|
thread = self._thread
|
|
if thread and thread.is_alive() and thread is not threading.current_thread():
|
|
thread.join(timeout=3)
|
|
self._set("stopped")
|
|
self.started_at = None
|
|
|
|
# ---------- 主循环 ----------
|
|
|
|
def _run(self) -> None:
|
|
backoff = 1.0
|
|
while self._desired:
|
|
try:
|
|
self._serve_once()
|
|
if not self._desired:
|
|
return
|
|
# 连接意外断开
|
|
self._log("warn", "SSH 连接已断开")
|
|
if not self.cfg.get("autoreconnect", True):
|
|
self._set("error", "连接已断开(未开启自动重连)")
|
|
return
|
|
self._set("connecting", "连接断开,正在重连")
|
|
except Exception as e: # noqa: BLE001
|
|
if not self._desired:
|
|
return
|
|
msg = friendly_error(e)
|
|
self._log("error", msg)
|
|
if not self.cfg.get("autoreconnect", True):
|
|
self._set("error", msg)
|
|
return
|
|
self._set("connecting", msg)
|
|
self._wakeup.wait(backoff)
|
|
backoff = min(backoff * 2, 15)
|
|
self._set("stopped")
|
|
|
|
def _serve_once(self) -> None:
|
|
host = self.host
|
|
cfg = self.cfg
|
|
client = paramiko.SSHClient()
|
|
self._client = client
|
|
try:
|
|
if self.known_hosts_path.exists():
|
|
client.load_host_keys(str(self.known_hosts_path))
|
|
else:
|
|
self.known_hosts_path.write_text("", encoding="utf-8")
|
|
client.load_host_keys(str(self.known_hosts_path))
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
port = int(host.get("port") or 22)
|
|
kw = dict(
|
|
hostname=host["hostname"], port=port, username=host.get("username"),
|
|
timeout=15, banner_timeout=25, auth_timeout=15,
|
|
allow_agent=False, look_for_keys=False,
|
|
)
|
|
if host.get("auth") == "key":
|
|
kw["key_filename"] = host.get("key_path") or None
|
|
if host.get("key_passphrase"):
|
|
kw["passphrase"] = host["key_passphrase"]
|
|
else:
|
|
kw["password"] = host.get("password")
|
|
self.attempts += 1
|
|
n = f"(第 {self.attempts} 次尝试)" if self.attempts > 1 else ""
|
|
self._log("info", f"正在连接 {host.get('username')}@{host['hostname']}:{port} ... {n}")
|
|
client.connect(**kw)
|
|
self.attempts = 0
|
|
try:
|
|
client.save_host_keys(str(self.known_hosts_path))
|
|
except Exception:
|
|
pass
|
|
transport = client.get_transport()
|
|
transport.set_keepalive(_KEEPALIVE)
|
|
|
|
local_port = int(cfg["local_port"])
|
|
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
listener.bind(("127.0.0.1", local_port))
|
|
listener.listen(32)
|
|
listener.settimeout(_POLL)
|
|
self._listener = listener
|
|
|
|
self.started_at = time.time()
|
|
self._set("connected")
|
|
self._log("info",
|
|
f"隧道已建立: 127.0.0.1:{local_port} → {cfg['remote_host']}:{cfg['remote_port']}")
|
|
|
|
while self._desired and transport.is_active():
|
|
try:
|
|
sock, _addr = listener.accept()
|
|
except socket.timeout:
|
|
continue
|
|
threading.Thread(target=self._pipe, args=(sock, transport),
|
|
daemon=True).start()
|
|
finally:
|
|
listener = self._listener
|
|
self._listener = None
|
|
if listener:
|
|
try:
|
|
listener.close()
|
|
except OSError:
|
|
pass
|
|
try:
|
|
client.close()
|
|
except Exception:
|
|
pass
|
|
self._client = None
|
|
|
|
def _pipe(self, sock: socket.socket, transport) -> None:
|
|
cfg = self.cfg
|
|
self.conn_count += 1
|
|
self._set("connected") # 顺带刷新连接计数
|
|
dest = (cfg["remote_host"], int(cfg["remote_port"]))
|
|
try:
|
|
try:
|
|
chan = transport.open_channel("direct-tcpip", dest, sock.getpeername(), timeout=10)
|
|
except Exception as e:
|
|
self._log("warn", f"打开远程通道失败 ({dest[0]}:{dest[1]}): {friendly_error(e)}")
|
|
return
|
|
if chan is None:
|
|
self._log("warn", f"远程拒绝了到 {dest[0]}:{dest[1]} 的转发请求")
|
|
return
|
|
self._pump(sock, chan)
|
|
finally:
|
|
for c in (sock,):
|
|
try:
|
|
c.close()
|
|
except OSError:
|
|
pass
|
|
|
|
@staticmethod
|
|
def _pump(sock: socket.socket, chan) -> None:
|
|
try:
|
|
while True:
|
|
r, _, _ = select.select([sock, chan], [], [], 1.0)
|
|
if not r:
|
|
if chan.closed or chan.eof_received:
|
|
break
|
|
continue
|
|
if sock in r:
|
|
data = sock.recv(65536)
|
|
if not data:
|
|
break
|
|
chan.sendall(data)
|
|
if chan in r:
|
|
data = chan.recv(65536)
|
|
if not data:
|
|
break
|
|
sock.sendall(data)
|
|
except Exception:
|
|
pass
|
|
finally:
|
|
try:
|
|
chan.close()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _opsig(cfg: dict) -> tuple:
|
|
return (cfg.get("host_id"), int(cfg.get("local_port") or 0),
|
|
cfg.get("remote_host"), int(cfg.get("remote_port") or 0),
|
|
bool(cfg.get("autoreconnect", True)))
|
|
|
|
|
|
def _hostsig(h: dict | None) -> tuple:
|
|
if not h:
|
|
return ()
|
|
return (h.get("hostname"), h.get("port"), h.get("username"), h.get("auth"),
|
|
h.get("password"), h.get("key_path"), h.get("key_passphrase"))
|
|
|
|
|
|
class TunnelManager:
|
|
def __init__(self, emit, data_dir):
|
|
self._emit = emit
|
|
self.known_hosts_path = data_dir / "known_hosts"
|
|
self.tunnels: dict[str, Tunnel] = {}
|
|
|
|
def apply_config(self, tunnels_cfg: list, hosts_by_id: dict) -> None:
|
|
alive_ids = {t["id"] for t in tunnels_cfg}
|
|
for tid in list(self.tunnels):
|
|
if tid not in alive_ids:
|
|
self.tunnels.pop(tid).stop()
|
|
|
|
for cfg in tunnels_cfg:
|
|
host = hosts_by_id.get(cfg.get("host_id"))
|
|
cur = self.tunnels.get(cfg["id"])
|
|
if cur is None:
|
|
tun = Tunnel(cfg, host, self._emit, self.known_hosts_path)
|
|
self.tunnels[cfg["id"]] = tun
|
|
if cfg.get("autostart"):
|
|
tun.start()
|
|
else:
|
|
changed = (_opsig(cfg) != _opsig(cur.cfg)
|
|
or _hostsig(host) != _hostsig(cur.host))
|
|
cur.cfg = dict(cfg)
|
|
cur.host = dict(host) if host else None
|
|
if changed and cur.status != "stopped":
|
|
cur.stop()
|
|
cur.start()
|
|
|
|
def stop_all(self, *_args) -> None:
|
|
for tun in list(self.tunnels.values()):
|
|
tun.stop()
|
|
|
|
|
|
def probe_host(host: dict, known_hosts_path) -> tuple[bool, str]:
|
|
"""测试一台主机的连通性与认证,不影响隧道。"""
|
|
client = paramiko.SSHClient()
|
|
try:
|
|
if known_hosts_path.exists():
|
|
client.load_host_keys(str(known_hosts_path))
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
port = int(host.get("port") or 22)
|
|
kw = dict(
|
|
hostname=host["hostname"], port=port, username=host.get("username"),
|
|
timeout=8, banner_timeout=8, auth_timeout=8,
|
|
allow_agent=False, look_for_keys=False,
|
|
)
|
|
if host.get("auth") == "key":
|
|
kw["key_filename"] = host.get("key_path") or None
|
|
if host.get("key_passphrase"):
|
|
kw["passphrase"] = host["key_passphrase"]
|
|
else:
|
|
kw["password"] = host.get("password")
|
|
client.connect(**kw)
|
|
try:
|
|
client.save_host_keys(str(known_hosts_path))
|
|
except Exception:
|
|
pass
|
|
version = client.get_transport().remote_version
|
|
return True, f"认证成功 · 服务端 {version}"
|
|
except Exception as e: # noqa: BLE001
|
|
return False, friendly_error(e)
|
|
finally:
|
|
try:
|
|
client.close()
|
|
except Exception:
|
|
pass
|