中文交互向导:ssh-tunnel 无参数进入,装完自动进入,填完直接运行

- 向导:选/建主机(实时连通测试,可重试)→ 选/建隧道(端口智能默认)→ 立即前台运行 - install.sh 结尾 exec 进入向导(stdin 从 /dev/tty 读,兼容 curl|bash) - cli.py 抽出 run_foreground 供 start 与向导复用;修复向导新建隧道分支不可达
2026-09-27 20:15:21 +08:00
parent 06c57dfbd1
commit 078b272eb1
7 changed files with 270 additions and 29 deletions
@@ -0,0 +1 @@
+<h1>WIZARD-OK</h1>
@@ -0,0 +1,28 @@
+{
+  "hosts": [
+    {
+      "id": "a6a8a6f5",
+      "name": "本机向导测试",
+      "hostname": "127.0.0.1",
+      "port": 2222,
+      "username": "test",
+      "auth": "password",
+      "password": "dpapi:AQAAANCMnd8BFdERjHoAwE/Cl+sBAAAAzni2ki6QM0W9HV1ipBcLPAAAAAACAAAAAAAQZgAAAAEAACAAAACEzy/56phqkFnZG+3rc7ODinJyOOszrzskmLsWLohEWgAAAAAOgAAAAAIAACAAAACnk3P/9dav98TxOrMZyz8+CyC0aSS0C3sRghY/u+4JZxAAAAAIPCoTBD+yaT1EUVFGwNITQAAAAMxgRjPaPFHuN2v7CwkgmCKMaCAE1oA0m+vDi1dujjEZJd/yT45VkRe9REH1V95/5CJLhDzf2cp2HY04WgxDMlU=",
+      "key_path": "",
+      "key_passphrase": "",
+      "note": ""
+    }
+  ],
+  "tunnels": [
+    {
+      "id": "5005e5b2",
+      "name": "向导隧道",
+      "host_id": "a6a8a6f5",
+      "local_port": 13003,
+      "remote_host": "127.0.0.1",
+      "remote_port": 3210,
+      "autostart": false,
+      "autoreconnect": true
+    }
+  ]
+}
@@ -0,0 +1 @@
+[127.0.0.1]:2222 ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQD0I9YPL8oqfY6L1d0LUrFmvVBtsMI5sxetirysUdgvtZXH/XRFqvLnrrG40NTJxTdfQ1avnj1ZRtXUVlYHVxYmwWJR0s2Rz0Q+UB5E9sDQhiww8lLwraD3hvN37iItMorS0qry2AWurKiHMtWw28GGQtP8Dwk8TBR5QLAS8BBAI54vq6m7dVbEpkJOzNl0lRMYJ64AL0S0ZrPplJgpJLEeXepcWhk8YTAelg7mPiL2hb81J0Bmy8DJIIwzv1h41jSIJecOZslQ9gYYpmCDbPmozooSgN8pCxPcCo0fmsJB2I73GKsPJACYOpcAeLGZA5uAfx8LHrr8sJkwQFrVaorB
@@ -0,0 +1,26 @@
+import builtins, os, sys
+here = os.path.dirname(os.path.abspath(__file__))
+os.environ["APPDATA"] = here                      # 隔离配置目录
+sys.path.insert(0, os.path.dirname(here))
+import getpass
+getpass.getpass = lambda *a, **k: "test123"       # 密码桩
+import cli
+
+ANSWERS = iter([
+    "1",                  # 选择:新增主机
+    "test@127.0.0.1",     # SSH 目标
+    "2222",               # SSH 端口
+    "1",                  # 认证方式:密码
+    "本机向导测试",        # 主机备注名
+    "1",                  # 选择:新建隧道
+    "127.0.0.1",          # 远程服务地址(默认)
+    "3210",               # 远程服务端口
+    "13003",              # 本地端口(避开本机 3210 测试服务)
+    "向导隧道",            # 隧道备注名
+])
+def fake_input(prompt=""):
+    a = next(ANSWERS)
+    print(f"{prompt}[{a}]")
+    return a
+builtins.input = fake_input
+sys.exit(cli.wizard(cli.Store(cli.data_root())))
@@ -16,6 +16,9 @@ powershell -c "irm https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.
 curl -fsSL https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.sh | bash
 ```

+安装完成会**自动进入中文配置向导**:按提示输入 SSH 地址、密码、远程端口等
+(直接回车采用方括号里的默认值),填完隧道立即建立,Ctrl+C 断开。
+
 ## 典型场景

 远程主机上有个服务只监听它自己的 `127.0.0.1:3000`,公网访问不到。
@@ -24,12 +27,12 @@ curl -fsSL https://gitea.xbay.cc/leefer/ssh-e2e/raw/branch/main/install.sh | bas
 ## 命令行用法(Linux/macOS,Windows 亦可 `python cli.py`)

 ```bash
-ssh-tunnel host add 我的服务器 root@1.2.3.4          # 添加主机(交互输入密码;私钥加 --key ~/.ssh/id_rsa)
-ssh-tunnel test 我的服务器                            # 测试连通与认证
-ssh-tunnel tunnel add web --host 我的服务器 --lp 3000 --rp 3000
-ssh-tunnel start web                                 # 前台运行,Ctrl+C 断开
-ssh-tunnel list                                      # 查看全部
+ssh-tunnel                     # 交互向导:选已有主机/隧道或新建,建好直接运行
+ssh-tunnel start web           # 直接启动某条已配置的隧道(前台,Ctrl+C 断开)
+ssh-tunnel list                # 查看全部
+ssh-tunnel test 我的服务器      # 测试主机认证
 ssh-tunnel remove tunnel web
+ssh-tunnel --help              # 完整参数(host add / tunnel add 等)
 ```

 ## 图形界面(Windows)
@@ -164,22 +164,10 @@ def cmd_remove(store, args) -> int:
    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
+def run_foreground(store: Store, chosen: list) -> int:
+    """前台运行选中的隧道,Ctrl+C 退出。"""
    hosts_by_id = {h["id"]: h for h in store.data["hosts"]}
    mgr = TunnelManager(_QuietEmit(), data_root())
-    chosen = []
-    for r in refs:
-        t = find_tunnel(store, r)
-        if not t:
-            _print(f"错误: 找不到隧道「{r}」")
-            return 1
-        chosen.append(t)
    mgr.apply_config(chosen, hosts_by_id)

    if store.warn_plaintext:
@@ -212,7 +200,208 @@ def cmd_start(store, args) -> int:
    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("\n── 第 1 步:选择 SSH 主机 ──")
+    for i, h in enumerate(hosts, 1):
+        _print(f"  {i}. {h['name']}  ({h['username']}@{h['hostname']}:{h['port']})")
+    _print(f"  {len(hosts) + 1}. 新增主机")
+
+    while True:
+        sel = _ask(f"请选择 [1-{len(hosts) + 1}]", str(len(hosts) + 1 if not hosts else "1"))
+        if not sel.isdigit() or not (1 <= int(sel) <= len(hosts) + 1):
+            _print("  无效选择,请重新输入")
+            continue
+        n = int(sel)
+        if n <= len(hosts):
+            return hosts[n - 1]
+        return _wizard_new_host(store)
+
+
+def _wizard_new_host(store: Store):
+    _print("\n新增主机(带 * 的为必填):")
+    while True:
+        target = _ask("  * SSH 登录目标 (格式 用户名@主机,如 root@1.2.3.4): ")
+        if "@" in target and len(target.split("@")) == 2 and target.split("@")[1]:
+            break
+        _print("  格式不对,应类似 root@1.2.3.4")
+    user, hostaddr = target.rsplit("@", 1)
+    port = _ask_port("    SSH 端口 [22]: ", 22)
+    auth = _ask("    认证方式 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("  * 私钥文件路径 (如 ~/.ssh/id_rsa): ")
+        rec["key_passphrase"] = _ask_secret("    私钥口令 (没有则直接回车): ")
+    else:
+        rec["password"] = _ask_secret("  * 登录密码 (输入时不显示): ")
+    rec["name"] = _ask("    主机备注名 [默认取地址]: ", hostaddr)
+
+    # 连通测试,失败可重试输入
+    while True:
+        _print(f"正在测试 {user}@{hostaddr}:{port} ...")
+        ok, msg = probe_host(rec, data_root() / "known_hosts")
+        _print(("  ✔ " if ok else "  ✘ ") + msg)
+        if ok:
+            break
+        retry = _ask("  连接失败。重试请回车,修改密码输入 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(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("\n── 第 2 步:配置隧道 ──")
+    for i, t in enumerate(ts, 1):
+        _print(f"  {i}. {t['name']}  (127.0.0.1:{t['local_port']} → {t['remote_host']}:{t['remote_port']})")
+    _print(f"  {len(ts) + 1}. 新建隧道")
+
+    while True:
+        sel = _ask(f"请选择 [1-{len(ts) + 1}]", str(len(ts) + 1 if not ts else "1"))
+        if not sel.isdigit() or not (1 <= int(sel) <= len(ts) + 1):
+            _print("  无效选择,请重新输入")
+            continue
+        n = int(sel)
+        if n <= len(ts):
+            return ts[n - 1]
+        return _wizard_new_tunnel(store, host)
+
+
+def _wizard_new_tunnel(store: Store, host: dict):
+    _print("\n新建隧道:把远程主机上的服务映射到本机端口")
+    _print("  (远程服务只监听 127.0.0.1 时,远程地址保持默认即可)")
+    rh = _ask("  远程服务地址 [127.0.0.1]: ", "127.0.0.1")
+    rp = _ask_port("  * 远程服务端口 (如 3000): ")
+    lp = _ask_port(f"  本地端口 [{rp}]: ", rp)
+    for t in store.data["tunnels"]:
+        if t["local_port"] == lp:
+            _print(f"  本地端口 {lp} 已被隧道「{t['name']}」占用")
+            lp = _ask_port("  换一个本地端口: ")
+    name = _ask(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(f"已保存隧道「{name}」")
+    return rec
+
+
+def wizard(store: Store) -> int:
+    _ensure_tty_stdin()
+    _print("╭─────────────────────────────╮")
+    _print("│   SSH 隧道配置向导           │")
+    _print("╰─────────────────────────────╯")
+    _print("按中文提示输入,直接回车采用 [方括号] 里的默认值;Ctrl+C 随时退出。")
+
+    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)

@@ -135,14 +135,7 @@ case ":$PATH:" in
    ;;
 esac

-say "安装完成!"
+say "安装完成!正在进入配置向导 ..."
 echo
-echo "  配置隧道(示例):"
-echo "    ssh-tunnel host add 我的服务器 root@1.2.3.4 22"
-echo "    ssh-tunnel tunnel add web --host 我的服务器 --lp 3000 --rp 3000"
-echo "    ssh-tunnel start web"
-echo
-echo "  常用命令: ssh-tunnel list | test <主机> | remove host|tunnel <名称>"
-echo "  图形界面(Windows): powershell -c \"irm $BASE_URL/raw/branch/main/install.ps1 | iex\""
-echo
-[ -n "${ZSH_VERSION:-}${BASH_VERSION:-}" ] || echo "(注:新开终端或 source ~/.bashrc 后 ssh-tunnel 生效)"
+# curl|bash 时 stdin 是管道,交互输入从终端(/dev/tty)读
+exec "$BIN_DIR/ssh-tunnel" < /dev/tty
				`@@ -0,0 +1 @@`
				`[127.0.0.1]:2222 ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQD0I9YPL8oqfY6L1d0LUrFmvVBtsMI5sxetirysUdgvtZXH/XRFqvLnrrG40NTJxTdfQ1avnj1ZRtXUVlYHVxYmwWJR0s2Rz0Q+UB5E9sDQhiww8lLwraD3hvN37iItMorS0qry2AWurKiHMtWw28GGQtP8Dwk8TBR5QLAS8BBAI54vq6m7dVbEpkJOzNl0lRMYJ64AL0S0ZrPplJgpJLEeXepcWhk8YTAelg7mPiL2hb81J0Bmy8DJIIwzv1h41jSIJecOZslQ9gYYpmCDbPmozooSgN8pCxPcCo0fmsJB2I73GKsPJACYOpcAeLGZA5uAfx8LHrr8sJkwQFrVaorB`