leefer
7229642c62
- 向导开场:像素字符画 Logo(❯_ 提示符,青绿+光标绿)+ 版本/平台/配置信息 + 横线 - 运行面板头部加 ❯_ 标识;连接中状态改为 ◐◓◑◒ 旋转动画 - SSH banner 错误改为可读诊断文案;握手超时 12→25s;重连日志带尝试次数
507 lines
19 KiB
Python
507 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""SSH 隧道管理器 — 命令行模式(跨平台,无界面)。
|
||
|
||
用法:
|
||
ssh-tunnel host add <名称> <user@host> [端口=22] [--password] [--key 路径]
|
||
ssh-tunnel tunnel add <名称> --host <主机名或ID> --lp 本地端口 --rp 远程端口 [--rh 远程地址=127.0.0.1]
|
||
ssh-tunnel start <隧道名或ID> [隧道...] # 前台运行,Ctrl+C 退出
|
||
ssh-tunnel stop-all # (与 start 配套,信号处理自动调用)
|
||
ssh-tunnel list # 列出主机与隧道
|
||
ssh-tunnel remove host|tunnel <名称或ID>
|
||
ssh-tunnel test <主机名或ID>
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import getpass
|
||
import signal
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
from engine import TunnelManager, probe_host # noqa: E402
|
||
from storage import Store, data_root # noqa: E402
|
||
import tui # noqa: E402
|
||
|
||
VERSION = "1.4"
|
||
|
||
|
||
def _print_banner(store: Store) -> None:
|
||
"""开场:像素 Logo + 产品信息(类 Claude Code 风格)。"""
|
||
from platform import python_version, system
|
||
logo = tui.logo_lines()
|
||
info = [
|
||
tui.bold("ssh-e2e · SSH 隧道管理器"),
|
||
tui.dim(f"v{VERSION} · 中文交互向导"),
|
||
tui.faint(f"{system()} · Python {python_version()}"),
|
||
tui.faint(f"配置 {store.path}"),
|
||
tui.faint("回车采用 [默认值] · Ctrl+C 退出"),
|
||
]
|
||
print()
|
||
for row, line in zip(logo, info):
|
||
print(f" {row} {line}")
|
||
print(" " + tui.accent("─" * 46))
|
||
|
||
|
||
def _print(msg: str) -> None:
|
||
print(msg, flush=True)
|
||
|
||
|
||
class _QuietEmit:
|
||
"""引擎事件 → 运行面板(无面板时打印到终端)。"""
|
||
|
||
def __init__(self, dash: "tui.Dashboard | None" = None):
|
||
self._last: dict[str, str] = {}
|
||
self._dash = dash
|
||
|
||
def __call__(self, ev: dict) -> None:
|
||
if ev.get("type") == "tunnel":
|
||
if self._last.get(ev["id"]) == ev["status"]:
|
||
return # 状态未变(如连接计数刷新)
|
||
self._last[ev["id"]] = ev["status"]
|
||
if ev["status"] == "error" and self._dash:
|
||
self._dash.log(tui.err(f"✘ {ev.get('error', '')}"))
|
||
elif not self._dash:
|
||
if ev["status"] == "connected":
|
||
_print(tui.ok("[隧道] 已连接"))
|
||
elif ev["status"] == "error":
|
||
_print(tui.err(f"[隧道] 连接失败: {ev.get('error', '')}"))
|
||
elif ev.get("type") == "log":
|
||
e = ev["entry"]
|
||
mark = {"info": "·", "success": "✔", "warn": "!", "error": "×"}.get(e["level"], "·")
|
||
line = f"{e['t']} {mark} {e['msg']}"
|
||
if self._dash:
|
||
self._dash.log(line)
|
||
else:
|
||
_print(line)
|
||
|
||
|
||
def find_host(store: Store, ref: str):
|
||
hosts = store.data["hosts"]
|
||
return next((h for h in hosts if h["id"] == ref or h["name"] == ref), None)
|
||
|
||
|
||
def find_tunnel(store: Store, ref: str):
|
||
ts = store.data["tunnels"]
|
||
return next((t for t in ts if t["id"] == ref or t["name"] == ref), None)
|
||
|
||
|
||
def cmd_host_add(store, args) -> int:
|
||
if "@" not in args.target:
|
||
_print("错误: 目标格式应为 user@host")
|
||
return 1
|
||
user, host = args.target.rsplit("@", 1)
|
||
auth = "key" if args.key else "password"
|
||
if find_host(store, args.name):
|
||
_print(f"错误: 主机「{args.name}」已存在")
|
||
return 1
|
||
password = passphrase = ""
|
||
if auth == "password" and args.password:
|
||
password = getpass.getpass("登录密码: ")
|
||
if auth == "key" and args.key:
|
||
if not Path(args.key).expanduser().exists():
|
||
_print(f"错误: 私钥文件不存在: {args.key}")
|
||
return 1
|
||
passphrase = getpass.getpass("私钥口令(无则直接回车): ")
|
||
store.data["hosts"].append({
|
||
"id": Store.new_id(), "name": args.name, "hostname": host,
|
||
"port": args.port, "username": user, "auth": auth,
|
||
"password": password, "key_path": str(Path(args.key).expanduser()) if args.key else "",
|
||
"key_passphrase": passphrase, "note": "",
|
||
})
|
||
store.save()
|
||
_print(f"已添加主机 {user}@{host}:{args.port}(认证:{'私钥' if auth == 'key' else '密码'})")
|
||
return 0
|
||
|
||
|
||
def cmd_tunnel_add(store, args) -> int:
|
||
host = find_host(store, args.host)
|
||
if not host:
|
||
_print(f"错误: 找不到主机「{args.host}」,先执行 host add")
|
||
return 1
|
||
for t in store.data["tunnels"]:
|
||
if t["local_port"] == args.lp:
|
||
_print(f"错误: 本地端口 {args.lp} 已被隧道「{t['name']}」使用")
|
||
return 1
|
||
store.data["tunnels"].append({
|
||
"id": Store.new_id(), "name": args.name, "host_id": host["id"],
|
||
"local_port": args.lp, "remote_host": args.rh or "127.0.0.1",
|
||
"remote_port": args.rp, "autostart": False, "autoreconnect": True,
|
||
})
|
||
store.save()
|
||
_print(f"已添加隧道 {args.name}: 127.0.0.1:{args.lp} → {args.rh or '127.0.0.1'}:{args.rp}(经 {host['name']})")
|
||
_print(f"启动: ssh-tunnel start {args.name}")
|
||
return 0
|
||
|
||
|
||
def cmd_list(store, _args) -> int:
|
||
if store.warn_plaintext:
|
||
_print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。")
|
||
print(tui.bold("主机"))
|
||
if not store.data["hosts"]:
|
||
print(tui.faint(" (无) ssh-tunnel 进入向导添加"))
|
||
for h in store.data["hosts"]:
|
||
auth = "私钥" if h.get("auth") == "key" else "密码"
|
||
print(f" {tui.accent('▣')} {h['name']} "
|
||
f"{tui.dim(f'{h["username"]}@{h["hostname"]}:{h["port"]}')} {tui.faint(auth)}")
|
||
print(tui.bold("隧道"))
|
||
if not store.data["tunnels"]:
|
||
print(tui.faint(" (无) ssh-tunnel 进入向导添加"))
|
||
for t in store.data["tunnels"]:
|
||
h = next((x for x in store.data["hosts"] if x["id"] == t["host_id"]), None)
|
||
hn = h["name"] if h else "(主机缺失)"
|
||
print(f" {tui.accent('⇄')} {t['name']} "
|
||
f"{tui.dim(f'127.0.0.1:{t["local_port"]} → {t["remote_host"]}:{t["remote_port"]}')} "
|
||
f"{tui.faint(f'经 {hn}')}")
|
||
return 0
|
||
|
||
|
||
def cmd_test(store, args) -> int:
|
||
host = find_host(store, args.host)
|
||
if not host:
|
||
_print(f"错误: 找不到主机「{args.host}」")
|
||
return 1
|
||
_print(f"正在测试 {host['username']}@{host['hostname']}:{host['port']} ...")
|
||
ok, msg = probe_host(host, data_root() / "known_hosts")
|
||
_print(("✔ " if ok else "× ") + msg)
|
||
return 0 if ok else 1
|
||
|
||
|
||
def cmd_remove(store, args) -> int:
|
||
if args.kind == "host":
|
||
h = find_host(store, args.ref)
|
||
if not h:
|
||
_print(f"错误: 找不到主机「{args.ref}」")
|
||
return 1
|
||
used = [t["name"] for t in store.data["tunnels"] if t["host_id"] == h["id"]]
|
||
if used:
|
||
_print(f"错误: 主机被隧道使用: {', '.join(used)}")
|
||
return 1
|
||
store.data["hosts"].remove(h)
|
||
store.save()
|
||
_print(f"已删除主机 {h['name']}")
|
||
else:
|
||
t = find_tunnel(store, args.ref)
|
||
if not t:
|
||
_print(f"错误: 找不到隧道「{args.ref}」")
|
||
return 1
|
||
store.data["tunnels"].remove(t)
|
||
store.save()
|
||
_print(f"已删除隧道 {t['name']}")
|
||
return 0
|
||
|
||
|
||
def run_foreground(store: Store, chosen: list) -> int:
|
||
"""前台运行选中的隧道,Ctrl+C 退出。tty 下显示实时状态面板。"""
|
||
hosts_by_id = {h["id"]: h for h in store.data["hosts"]}
|
||
dash = tui.Dashboard()
|
||
mgr = TunnelManager(_QuietEmit(dash), data_root())
|
||
mgr.apply_config(chosen, hosts_by_id)
|
||
|
||
if store.warn_plaintext:
|
||
_print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。")
|
||
print()
|
||
print(tui.faint(" " + ", ".join(t["name"] for t in chosen) + " — Ctrl+C 退出"))
|
||
|
||
stopping = False
|
||
|
||
def _stats():
|
||
return [{
|
||
"name": t.cfg.get("name", t.cfg["id"]),
|
||
"local_port": t.cfg["local_port"],
|
||
"remote": f"{t.cfg['remote_host']}:{t.cfg['remote_port']}",
|
||
"status": t.status, "started_at": t.started_at,
|
||
"conns": t.conn_count, "error": t.error,
|
||
} for t in mgr.tunnels.values()]
|
||
|
||
def _stop(*_a):
|
||
nonlocal stopping
|
||
if not stopping:
|
||
stopping = True
|
||
print() # 光标移到面板下方行首
|
||
dash.close()
|
||
_print("正在断开所有隧道 ...")
|
||
sys.stdout.flush()
|
||
mgr.stop_all()
|
||
|
||
signal.signal(signal.SIGINT, _stop)
|
||
signal.signal(signal.SIGTERM, _stop)
|
||
for t in mgr.tunnels.values():
|
||
t.start()
|
||
try:
|
||
while not stopping:
|
||
time.sleep(0.5)
|
||
dash.update(_stats())
|
||
if not [t for t in mgr.tunnels.values() if t.status != "stopped"]:
|
||
dash.close()
|
||
_print(tui.err("所有隧道已停止。"))
|
||
return 1
|
||
except KeyboardInterrupt:
|
||
_stop()
|
||
dash.close()
|
||
return 0
|
||
|
||
|
||
def cmd_start(store, args) -> int:
|
||
refs = args.tunnels
|
||
if not refs: # 未指定则启动全部
|
||
refs = [t["name"] for t in store.data["tunnels"]]
|
||
if not refs:
|
||
_print("没有已配置的隧道。先用 tunnel add 添加。")
|
||
return 1
|
||
chosen = []
|
||
for r in refs:
|
||
t = find_tunnel(store, r)
|
||
if not t:
|
||
_print(f"错误: 找不到隧道「{r}」")
|
||
return 1
|
||
chosen.append(t)
|
||
return run_foreground(store, chosen)
|
||
|
||
|
||
# ---------- 交互向导 ----------
|
||
|
||
def _ensure_tty_stdin() -> None:
|
||
"""curl|bash 场景下 stdin 是管道,交互输入改从终端读。"""
|
||
if sys.stdin.isatty():
|
||
return
|
||
try:
|
||
sys.stdin = open("/dev/tty", "r") # noqa: SIM115
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _ask(prompt: str, default: str = "", allow_empty: bool = False) -> str:
|
||
while True:
|
||
try:
|
||
v = input(prompt).strip()
|
||
except EOFError:
|
||
if default or allow_empty:
|
||
return default
|
||
raise
|
||
if v:
|
||
return v
|
||
if default:
|
||
return default
|
||
if allow_empty:
|
||
return ""
|
||
_print(" (不能为空,请重新输入)")
|
||
|
||
|
||
def _ask_port(prompt: str, default: int | None = None) -> int:
|
||
while True:
|
||
raw = _ask(prompt, str(default) if default else "")
|
||
try:
|
||
p = int(raw)
|
||
except ValueError:
|
||
_print(" 端口必须是数字,请重新输入")
|
||
continue
|
||
if 1 <= p <= 65535:
|
||
return p
|
||
_print(" 端口范围 1-65535,请重新输入")
|
||
|
||
|
||
def _ask_secret(prompt: str) -> str:
|
||
try:
|
||
return getpass.getpass(prompt)
|
||
except EOFError:
|
||
return ""
|
||
|
||
|
||
def _wizard_pick_host(store: Store):
|
||
"""选择已有主机或新建;返回主机 dict 或 None(取消)。"""
|
||
hosts = store.data["hosts"]
|
||
print()
|
||
print(tui.bold("◇ SSH 主机"))
|
||
opts = [f"{h['name']} ({h['username']}@{h['hostname']}:{h['port']})" for h in hosts]
|
||
opts.append(tui.dim("＋ 新增主机") if len(hosts) else "＋ 新增主机")
|
||
n = tui.select(" 选择要使用的主机:", opts, default=len(hosts) if not hosts else 0)
|
||
if n is None:
|
||
return None
|
||
if n < len(hosts):
|
||
return hosts[n]
|
||
return _wizard_new_host(store)
|
||
|
||
|
||
def _wizard_new_host(store: Store):
|
||
print()
|
||
print(tui.bold("◇ 新增主机") + tui.faint(" (带 * 为必填)"))
|
||
while True:
|
||
target = _ask(" " + tui.accent("SSH 登录目标") + tui.faint(" (用户名@主机,如 root@1.2.3.4)") + ": ")
|
||
if "@" in target and len(target.split("@")) == 2 and target.split("@")[1]:
|
||
break
|
||
print(tui.warn(" 格式不对,应类似 root@1.2.3.4"))
|
||
user, hostaddr = target.rsplit("@", 1)
|
||
port = _ask_port(" " + tui.accent("SSH 端口") + tui.faint(" [22]"), 22)
|
||
auth = _ask(" " + tui.accent("认证方式") + tui.faint(" 1)密码 2)私钥 [1]"), "1")
|
||
while auth not in ("1", "2"):
|
||
auth = _ask(" 请输入 1 或 2: ", "1")
|
||
|
||
rec = {
|
||
"id": Store.new_id(), "name": "", "hostname": hostaddr, "port": port,
|
||
"username": user, "auth": "key" if auth == "2" else "password",
|
||
"password": "", "key_path": "", "key_passphrase": "", "note": "",
|
||
}
|
||
if rec["auth"] == "key":
|
||
rec["key_path"] = _ask(" " + tui.accent("* 私钥文件路径") + tui.faint(" (如 ~/.ssh/id_rsa)") + ": ")
|
||
rec["key_passphrase"] = _ask_secret(" 私钥口令 (没有则直接回车): ")
|
||
else:
|
||
rec["password"] = _ask_secret(" " + tui.accent("* 登录密码") + tui.faint(" (输入时不显示)") + ": ")
|
||
rec["name"] = _ask(" " + tui.accent("主机备注名") + tui.faint(f" [{hostaddr}]") + ": ", hostaddr)
|
||
|
||
# 连通测试,失败可重试输入
|
||
while True:
|
||
print(tui.faint(f" 正在测试 {user}@{hostaddr}:{port} ..."))
|
||
ok, msg = probe_host(rec, data_root() / "known_hosts")
|
||
print((" " + tui.ok("✔ ") + tui.ok(msg)) if ok else (" " + tui.err("✘ ") + tui.err(msg)))
|
||
if ok:
|
||
break
|
||
retry = _ask(" " + tui.warn("连接失败") + tui.faint(" — 回车重试 / p 改密码 / q 放弃") + ": ", "")
|
||
if retry.lower() == "q":
|
||
return None
|
||
if retry.lower() == "p" and rec["auth"] == "password":
|
||
rec["password"] = _ask_secret(" 重新输入登录密码: ")
|
||
|
||
store.data["hosts"].append(rec)
|
||
store.save()
|
||
print(" " + tui.ok(f"已保存主机「{rec['name']}」"))
|
||
return rec
|
||
|
||
|
||
def _wizard_pick_tunnel(store: Store, host: dict):
|
||
"""选择该主机的已有隧道或新建;返回隧道 dict 或 None。"""
|
||
ts = [t for t in store.data["tunnels"] if t["host_id"] == host["id"]]
|
||
print()
|
||
print(tui.bold("◇ 隧道"))
|
||
opts = [f"{t['name']} (127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']})"
|
||
for t in ts]
|
||
opts.append("＋ 新建隧道")
|
||
n = tui.select(" 选择要启动的隧道:", opts, default=len(ts) if not ts else 0)
|
||
if n is None:
|
||
return None
|
||
if n < len(ts):
|
||
return ts[n]
|
||
return _wizard_new_tunnel(store, host)
|
||
|
||
|
||
def _wizard_new_tunnel(store: Store, host: dict):
|
||
print()
|
||
print(tui.bold("◇ 新建隧道") + tui.faint(" 把远程主机上的服务映射到本机端口"))
|
||
print(tui.faint(" (远程服务只监听 127.0.0.1 时,远程地址保持默认即可)"))
|
||
rh = _ask(" " + tui.accent("远程服务地址") + tui.faint(" [127.0.0.1]") + ": ", "127.0.0.1")
|
||
rp = _ask_port(" " + tui.accent("* 远程服务端口") + tui.faint(" (如 3000)") + ": ")
|
||
lp = _ask_port(" " + tui.accent("本地端口") + tui.faint(f" [{rp}]") + ": ", rp)
|
||
for t in store.data["tunnels"]:
|
||
if t["local_port"] == lp:
|
||
print(tui.warn(f" 本地端口 {lp} 已被隧道「{t['name']}」占用"))
|
||
lp = _ask_port(" 换一个本地端口: ")
|
||
name = _ask(" " + tui.accent("隧道备注名") + tui.faint(f" [{rh}:{rp}]") + ": ", f"{rh}:{rp}")
|
||
rec = {
|
||
"id": Store.new_id(), "name": name, "host_id": host["id"],
|
||
"local_port": lp, "remote_host": rh, "remote_port": rp,
|
||
"autostart": False, "autoreconnect": True,
|
||
}
|
||
store.data["tunnels"].append(rec)
|
||
store.save()
|
||
print(" " + tui.ok(f"已保存隧道「{name}」"))
|
||
return rec
|
||
|
||
|
||
def wizard(store: Store) -> int:
|
||
_ensure_tty_stdin()
|
||
_print_banner(store)
|
||
|
||
if store.warn_plaintext:
|
||
_print("提示: 当前平台密码以明文保存于配置文件(权限已设为仅本用户可读)。")
|
||
|
||
host = _wizard_pick_host(store)
|
||
if not host:
|
||
_print("已取消。")
|
||
return 1
|
||
tunnel = _wizard_pick_tunnel(store, host)
|
||
if not tunnel:
|
||
_print("已取消。")
|
||
return 1
|
||
|
||
_print(f"\n隧道即将建立: 本机 127.0.0.1:{tunnel['local_port']}"
|
||
f" ←→ 远程 {tunnel['remote_host']}:{tunnel['remote_port']}(经 {host['name']})")
|
||
code = run_foreground(store, [tunnel])
|
||
if code == 0:
|
||
_print(f"隧道已断开。下次直接运行: ssh-tunnel start {tunnel['name']}")
|
||
return code
|
||
|
||
|
||
def main() -> int:
|
||
if len(sys.argv) == 1:
|
||
# 无参数:直接进入交互向导
|
||
return wizard(Store(data_root()))
|
||
|
||
p = argparse.ArgumentParser(prog="ssh-tunnel", description="SSH 隧道管理器(命令行)")
|
||
sub = p.add_subparsers(dest="cmd", required=True)
|
||
|
||
sp = sub.add_parser("host", help="主机管理")
|
||
hsub = sp.add_subparsers(dest="sub", required=True)
|
||
ha = hsub.add_parser("add", help="添加主机")
|
||
ha.add_argument("name", help="备注名称")
|
||
ha.add_argument("target", help="user@host")
|
||
ha.add_argument("port", nargs="?", type=int, default=22)
|
||
ha.add_argument("--password", action="store_true", help="交互输入密码(默认)")
|
||
ha.add_argument("--key", help="私钥文件路径")
|
||
|
||
sp = sub.add_parser("tunnel", help="隧道管理")
|
||
tsub = sp.add_subparsers(dest="sub", required=True)
|
||
ta = tsub.add_parser("add", help="添加隧道")
|
||
ta.add_argument("name")
|
||
ta.add_argument("--host", required=True, help="主机名称或 ID")
|
||
ta.add_argument("--lp", type=int, required=True, help="本地端口")
|
||
ta.add_argument("--rp", type=int, required=True, help="远程端口")
|
||
ta.add_argument("--rh", default="127.0.0.1", help="远程地址(默认 127.0.0.1)")
|
||
|
||
sub.add_parser("list", help="列出主机与隧道")
|
||
sub.add_parser("stop-all", help="预留(前台模式退出时自动断开)")
|
||
|
||
st = sub.add_parser("start", help="前台启动隧道")
|
||
st.add_argument("tunnels", nargs="*", help="隧道名称或 ID(空=全部)")
|
||
|
||
ts = sub.add_parser("test", help="测试主机连通")
|
||
ts.add_argument("host")
|
||
|
||
rm = sub.add_parser("remove", help="删除主机或隧道")
|
||
rm.add_argument("kind", choices=["host", "tunnel"])
|
||
rm.add_argument("ref")
|
||
|
||
args = p.parse_args()
|
||
store = Store(data_root())
|
||
|
||
if args.cmd == "host":
|
||
return cmd_host_add(store, args)
|
||
if args.cmd == "tunnel":
|
||
return cmd_tunnel_add(store, args)
|
||
if args.cmd == "list":
|
||
return cmd_list(store, args)
|
||
if args.cmd == "test":
|
||
return cmd_test(store, args)
|
||
if args.cmd == "remove":
|
||
return cmd_remove(store, args)
|
||
if args.cmd == "start":
|
||
return cmd_start(store, args)
|
||
if args.cmd == "stop-all":
|
||
_print("隧道随前台进程退出自动断开,无需单独停止。")
|
||
return 0
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
sys.exit(main())
|
||
except KeyboardInterrupt:
|
||
sys.exit(130)
|
||
except SystemExit:
|
||
raise
|
||
except Exception: # noqa: BLE001 出错必须可见,禁止无声退出
|
||
import traceback
|
||
sys.stderr.write("\n发生错误,已退出:\n" + traceback.format_exc())
|
||
sys.exit(1)
|