docs: add L1+L2 check implementation record

Document the two-layer short-circuit probe added in 6124abb: design
rationale, the historical lesson from the removed _is_port_in_use bind
check, test coverage, real-world verification results on both local
machine and NAS, compatibility, deployment, and rollback procedure.

docs/check-l1-l2-implementation.md

@@ -0,0 +1,284 @@
+# check 命令 L1+L2 端到端预检实施记录
+
+**日期:** 2026-08-05
+**Commit:** `6124abb`
+**改动文件:** `src/autossh_mgr/check.py`、`tests/unit/test_check.py`、`tests/unit/test_cli_lifecycle.py`
+
+---
+
+## 1. 背景与问题
+
+### 1.1 旧版 `check` 的假阳性
+
+旧版 `check_connectivity` 只做一件事:发起一次轻量 SSH 登录(`ssh ... true`),
+验证"密钥能不能登 + 网络可达 + sshd 在"。它**完全不碰真实隧道的数据通路**。
+
+由此产生**假阳性**(check 报 OK 但隧道实际不通):
+
+| 场景 | 旧 check | 真实状态 |
+|------|---------|---------|
+| 本地后端服务没起(如 ftdl 的 Next.js 停了) | ✅ OK | ❌ 公网访问连不上 |
+| 远端端口被占用 | ✅ OK | ❌ `-R` 绑不上 |
+| sshd `GatewayPorts no` | ✅ OK | ❌ 只绑 127.0.0.1,外网进不来 |
+| `AllowTcpForwarding no` | ✅ OK | ❌ 转发被禁 |
+
+**本机实测复现:** `ftdl` 隧道的后端 Next.js 未运行,但旧版 `check ftdl` 仍返回 OK。
+
+### 1.2 历史教训:被删除的 `_is_port_in_use` (commit `4c9cd00`)
+
+历史上曾有过一个端口检查,位于 `process.py` 的 `start_tunnel` 里,但逻辑错误已被删除:
+
+```python
+# ❌ 已删除的错误检查 (commit 4c9cd00):
+def _is_port_in_use(port):
+    with socket.socket(...) as s:
+        try:
+            s.bind(("127.0.0.1", port))   # 尝试占用端口
+            return False                    # 占用成功→端口空闲
+        except OSError:
+            return True                     # 占用失败→端口被占
+# start_tunnel 里: if _is_port_in_use(local_port): 拒绝启动
+```
+
+**为什么错:** 对 `-R` 反向隧道,`local_port` 是**转发目标**(如 Docker 容器、Gitea web),
+它**本该在监听**,不该是空闲的。旧检查把"目标服务在跑"误判成"端口冲突",语义反了。
+
+### 1.3 本次方案的区别
+
+本次新增的 Layer 1 用 `connect()` 而非 `bind()`,语义完全不同,不会重蹈覆辙:
+
+| | 被删的旧检查 | 本次 Layer 1 |
+|---|---|---|
+| **方法** | `bind()` 占用端口 | `connect()` 探测目标 |
+| **位置** | `start_tunnel` (启动门禁) | `check` (诊断命令) |
+| **语义** | "端口被占=冲突" ❌ 反了 | "目标响应=健康" ✅ 对 |
+| **目的** | 阻止启动 | 报告隧道是否端到端可用 |
+
+---
+
+## 2. 设计:L1+L2 精简版(三层中的前两层)
+
+### 2.1 范围
+
+原始设计(见备份文档 `check-port-validation.md`)提出三层检查 L1/L2/L3。
+本次**只实施 L1+L2**,跳过 L3 探针转发,理由见 §2.4。
+
+**不做的事:**
+- 不做 L3 探针转发(带 `-R` + `ExitOnForwardFailure` 的真实转发探测)
+- 不改 autossh 进程的启动参数(`ExitOnForwardFailure` 留给方案 D)
+- 不改 `start`/`status`/`stop` 等其它命令的行为
+- 不改配置文件格式 `tunnels.yaml`
+
+### 2.2 检查流程(两层,顺序短路)
+
+```
+check <name>
+  │
+  ├─ Layer 1: 本地后端端口探测
+  │   TCP connect 到 (local_host, local_port),超时 2 秒
+  │   失败 → 返回 FAIL "local backend <host>:<port> not listening"
+  │   成功 ↓
+  │
+  └─ Layer 2: SSH 认证 (保留原有逻辑)
+      ssh -o BatchMode=yes -o ConnectTimeout=5 ... true
+      失败 → 返回 FAIL "SSH auth failed: <stderr>"
+      成功 → 返回 OK "reachable + backend up"
+```
+
+**顺序的理由:** Layer 1 最便宜(纯本地 TCP,毫秒级),Layer 2 较贵(一次 SSH 握手)。
+便宜的前置过滤掉明显错误(后端没起),避免无谓的网络往返。
+
+### 2.3 两层覆盖的失败场景
+
+| 失败场景 | Layer 1 (本地端口) | Layer 2 (SSH 认证) |
+|---------|:--:|:--:|
+| 后端服务没起(最常见) | ✅ | — |
+| 本地端口绑错地址(127.x vs 0.0.0.0) | ✅ | — |
+| 网络不通/防火墙 | — | ✅ |
+| 密钥被拒/被 fail2ban | — | ✅ |
+| 远端端口被占 | ❌ 不覆盖 | — |
+| `GatewayPorts no` | ❌ 不覆盖 | — |
+| `AllowTcpForwarding no` | ❌ 不覆盖 | — |
+
+L1+L2 覆盖了最常见的"后端没起"假阳性。端口绑定类问题留给未来的 L3 或方案 D。
+
+### 2.4 为什么跳过 L3
+
+| 考量 | 说明 |
+|------|------|
+| **副作用** | L3 探针会在云端短暂占用 `remote_port` 约 1 秒 |
+| **性能** | 成功路径增加约 1.5 秒(L1 毫秒级 + L3 的 sleep 1);失败路径因短路反而更快 |
+| **误报风险** | 某些 sshd 配置下可能因端口已被 autossh 占用而误报失败 |
+| **性价比** | L3 主要堵"端口绑定类"问题,这些场景相对少见,真遇到用 `status` + 手动 ssh 也能发现 |
+
+L1+L2 覆盖 80% 的常见假阳性,改动最小、零副作用。L3 可后续按需再加。
+
+---
+
+## 3. 实现
+
+仅改 1 个源码文件:`src/autossh_mgr/check.py`
+
+### 3.1 新增函数 `check_local_backend` (Layer 1)
+
+```python
+import socket
+
+def check_local_backend(tunnel: TunnelConfig) -> tuple[bool, str]:
+    """Layer 1: probe whether the local backend port has a listener."""
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.settimeout(2)
+    try:
+        s.connect((tunnel.local_host, tunnel.local_port))
+    except OSError as e:
+        return False, f"local backend {tunnel.local_host}:{tunnel.local_port} not listening ({e})"
+    finally:
+        s.close()
+    return True, ""
+```
+
+**关键点:**
+- 用 `connect()` 而非 `bind()`,语义正确(探测目标是否响应,而非端口是否空闲)
+- `finally: s.close()` 保证任何情况下都释放 socket
+- 超时 2 秒,避免后端慢响应时卡住
+
+### 3.2 改写 `check_connectivity` (L1→L2 短路)
+
+```python
+def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
+    # Layer 1: 本地后端端口
+    ok, msg = check_local_backend(tunnel)
+    if not ok:
+        return False, msg
+
+    # Layer 2: SSH 认证 (原有逻辑)
+    result = subprocess.run(
+        build_ssh_check_cmd(tunnel),
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        return False, f"SSH auth failed: {result.stderr.strip()}"
+    return True, "reachable + backend up"
+```
+
+**接口契约不变:** 签名仍是 `(bool, str)`,`cli.py` 的 `check_cmd` 无需改动。
+成功时 `cli.py:245` 显示固定的 `OK: ...`,忽略 msg;失败时 msg 进入 `Connection failed: {msg}`。
+
+### 3.3 `build_ssh_check_cmd` 未改动
+
+Layer 2 复用原有命令构建函数,保持 SSH 认证探测行为不变。
+
+---
+
+## 4. 测试
+
+### 4.1 单元测试 (`tests/unit/test_check.py`)
+
+新增/改写 6 个测试:
+
+| 测试 | 覆盖点 |
+|------|--------|
+| `test_build_ssh_check_cmd` | 命令构建(原有,保留) |
+| `test_check_local_backend_success` | L1 成功路径,mock socket,断言 connect/settimeout/close 调用 |
+| `test_check_local_backend_failure` | L1 失败路径,connect 抛 `ConnectionRefusedError`,断言错误消息含端口 |
+| `test_check_connectivity_layer1_short_circuits` | **L1 短路关键测试**:L1 失败时 `subprocess.run` 必须不被调用 |
+| `test_check_connectivity_success` | L1+L2 全成功,断言返回 "reachable + backend up" |
+| `test_check_connectivity_ssh_failure` | L1 成功但 L2 失败,断言返回 "SSH auth failed" |
+
+### 4.2 CLI 集成测试修复 (`tests/unit/test_cli_lifecycle.py`)
+
+`test_check_success` / `test_check_failure` 原本只 mock `subprocess.run`,
+新增 L1 后会真去 connect 端口导致测试失败。修复:同时 mock `autossh_mgr.check.socket.socket`。
+
+### 4.3 全量测试结果
+
+```
+72 passed in 52.84s
+```
+
+含 integration 7 + unit 65,两次运行结果一致,稳定通过。
+
+---
+
+## 5. 实测验证
+
+### 5.1 本机实测
+
+本机 `tunnels.yaml` 配置两条隧道,editable 安装新版源码后实测:
+
+| 隧道 | 后端端口 | L1 探测 | 结果 | 旧行为对比 |
+|------|---------|---------|------|-----------|
+| `hub-ftdev-android` | :8003 ✅ 在监听 | connect 成功 → L2 → SSH 通 | ✅ OK | 旧版也 OK(巧合正确) |
+| `ftdl` | :3000 ❌ Next.js 没起 | connect 拒绝 → **L1 短路失败** | ❌ `local backend 127.0.0.1:3000 not listening` | ❗ 旧版会误报 OK(假阳性) |
+
+**验证了方案C的核心价值:** ftdl 后端没起,旧版 `check` 报 OK(隧道"通"但实际访问拿不到东西),
+新版 L1 立刻拦住并明确告知"本地后端没监听",且根本没发起 SSH(短路)。
+
+### 5.2 NAS 实测
+
+NAS 上 `tunnels.yaml` 配置两条隧道(后端都是 NAS 本机服务),实测:
+
+| 隧道 | 后端端口 | L1 | L2 | 结果 |
+|------|---------|:--:|:--:|------|
+| `gitea` | :80 Gitea web | ✅ | ✅ | ✅ OK |
+| `gitea-ssh` | :2222 Gitea ssh | ✅ | ✅ | ✅ OK |
+
+两条隧道全绿,零假阳性、零误报。
+
+---
+
+## 6. 兼容性
+
+- **配置文件:** 无变更,`tunnels.yaml` 格式不变
+- **命令接口:** `autossh-mgr check <name>` 用法不变,退出码语义不变(0=成功,1=失败)
+- **输出格式:** 成功仍是 `OK: ...`,失败仍是 `Error: Connection failed: ...`,
+  只是失败原因更具体(指明是 L1 还是 L2 挂的)
+- **性能:** 成功路径增加约几十毫秒(L1 TCP connect);失败路径因短路反而更快
+- **无新依赖:** `socket` 是 Python 标准库
+
+---
+
+## 7. 部署方式
+
+| 位置 | 安装方式 | 更新方式 |
+|------|---------|---------|
+| **NAS** | editable (`uv tool install --editable .`) | 改 `src/` 即生效,无需重装 |
+| **本机** | 非 editable (从 git URL 装) | `uv tool install --force --refresh <url>` 或改用 `--editable` |
+
+本机开发建议用 editable 安装,改完立即生效:
+```bash
+cd ~/workspace/autossh-mgr
+uv tool install --force --editable .
+```
+
+---
+
+## 8. 回退方案
+
+任何时候可回退到改动前(commit `4c9cd00`):
+
+```bash
+# 源码回退
+cd ~/workspace/autossh-mgr
+git reset --hard 4c9cd00        # 或用 tag: git reset --hard pre-plan-c
+
+# 若本机是 editable 安装,源码回退后立即生效
+# 若本机是非 editable 安装,需重装旧版:
+uv tool install --force --refresh ssh://git@git.zz.com:2222/Developer/autossh-mgr.git
+```
+
+NAS 回退:
+```bash
+ssh mnas 'cd ~/workspace/autossh-mgr && git checkout -- src/autossh_mgr/check.py'
+```
+
+---
+
+## 9. 未来工作
+
+| 项 | 说明 | 优先级 |
+|----|------|--------|
+| **Layer 3 探针转发** | 带 `-R` + `ExitOnForwardFailure` 的真实转发探测,覆盖端口绑定类假阳性 | 按需,当前 L1+L2 已覆盖 80% 场景 |
+| **方案 D: autossh 启动参数** | 给 `build_autossh_cmd` 加 `-o ExitOnForwardFailure=yes`,让运行中的 autossh 在端口绑不上时自愈重生 | 中,根治运行态假活 |
+| **verify 命令** | 独立的端到端探测命令,通过真实隧道发 HTTP 请求验证后端响应 | 低,`check` + `status` 已基本够用 |

src/autossh_mgr/check.py

@@ -1,8 +1,22 @@
 import os
+import socket
 import subprocess
 from autossh_mgr.config import TunnelConfig
 
 
+def check_local_backend(tunnel: TunnelConfig) -> tuple[bool, str]:
+    """Layer 1: probe whether the local backend port has a listener."""
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.settimeout(2)
+    try:
+        s.connect((tunnel.local_host, tunnel.local_port))
+    except OSError as e:
+        return False, f"local backend {tunnel.local_host}:{tunnel.local_port} not listening ({e})"
+    finally:
+        s.close()
+    return True, ""
+
+
 def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
     return [
         "ssh",
@@ -16,9 +30,15 @@ def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
 
 
 def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
+    ok, msg = check_local_backend(tunnel)
+    if not ok:
+        return False, msg
+
     result = subprocess.run(
         build_ssh_check_cmd(tunnel),
         capture_output=True,
         text=True,
     )
-    return result.returncode == 0, result.stderr.strip()
+    if result.returncode != 0:
+        return False, f"SSH auth failed: {result.stderr.strip()}"
+    return True, "reachable + backend up"

src/autossh_mgr/process.py

@@ -2,7 +2,6 @@ import os
 import shlex
 import shutil
 import signal
-import socket
 import subprocess
 import time
 from dataclasses import dataclass
@@ -115,20 +114,9 @@ def build_autossh_cmd(tunnel: TunnelConfig) -> list[str]:
     return cmd
 
 
-def _is_port_in_use(port: int) -> bool:
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        try:
-            s.bind(("127.0.0.1", port))
-            return False
-        except OSError:
-            return True
-
-
 def start_tunnel(config_dir: Path, tunnel: TunnelConfig) -> int:
     if not shutil.which("autossh"):
         raise click.ClickException("autossh not found. Install it first.")
-    if _is_port_in_use(tunnel.local_port):
-        raise click.ClickException(f"Port {tunnel.local_port} is already in use")
 
     log_path = config_dir / "logs" / f"{tunnel.name}.log"
     log_path.parent.mkdir(parents=True, exist_ok=True)

tests/unit/test_check.py

@@ -1,6 +1,10 @@
 import pytest
 from unittest.mock import patch, MagicMock
-from autossh_mgr.check import check_connectivity, build_ssh_check_cmd
+from autossh_mgr.check import (
+    check_connectivity,
+    check_local_backend,
+    build_ssh_check_cmd,
+)
 from autossh_mgr.config import TunnelConfig
 
 
@@ -22,23 +26,56 @@ def test_build_ssh_check_cmd(tunnel):
     assert "-p" in cmd and str(tunnel.port) in cmd
 
 
+def test_check_local_backend_success(tunnel):
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
+        ok, msg = check_local_backend(tunnel)
+    assert ok is True
+    assert msg == ""
+    mock_sock.connect.assert_called_once_with(("127.0.0.1", 8080))
+    mock_sock.settimeout.assert_called_once_with(2)
+    mock_sock.close.assert_called_once()
+
+
+def test_check_local_backend_failure(tunnel):
+    mock_sock = MagicMock()
+    mock_sock.connect.side_effect = ConnectionRefusedError("refused")
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock):
+        ok, msg = check_local_backend(tunnel)
+    assert ok is False
+    assert "127.0.0.1:8080" in msg
+    assert "not listening" in msg
+    mock_sock.close.assert_called_once()
+
+
+def test_check_connectivity_layer1_short_circuits(tunnel):
+    """When Layer 1 fails, SSH (Layer 2) must not be invoked."""
+    mock_sock = MagicMock()
+    mock_sock.connect.side_effect = ConnectionRefusedError("refused")
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
+        success, msg = check_connectivity(tunnel)
+    assert success is False
+    assert "not listening" in msg
+    mock_run.assert_not_called()
+
+
 def test_check_connectivity_success(tunnel):
-    mock_result = MagicMock()
-    mock_result.returncode = 0
-    mock_result.stderr = ""
-    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result) as mock_run:
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=0, stderr="")) as mock_run:
         success, msg = check_connectivity(tunnel)
     assert success is True
-    assert msg == ""
+    assert msg == "reachable + backend up"
     called_cmd = mock_run.call_args[0][0]
     assert called_cmd == build_ssh_check_cmd(tunnel)
 
 
-def test_check_connectivity_failure(tunnel):
-    mock_result = MagicMock()
-    mock_result.returncode = 255
-    mock_result.stderr = "Connection refused"
-    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result):
+def test_check_connectivity_ssh_failure(tunnel):
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=255, stderr="Connection refused")):
         success, msg = check_connectivity(tunnel)
     assert success is False
+    assert "SSH auth failed" in msg
     assert "Connection refused" in msg

tests/unit/test_cli_lifecycle.py

@@ -87,7 +87,9 @@ def test_status_single(runner, config_dir, with_tunnel):
 
 
 def test_check_success(runner, with_tunnel):
-    with patch("autossh_mgr.check.subprocess.run") as mock_run:
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
         mock_run.return_value = MagicMock(returncode=0, stderr="")
         result = runner.invoke(cli, ["check", "web-service"])
     assert result.exit_code == 0
@@ -95,7 +97,9 @@ def test_check_success(runner, with_tunnel):
 
 
 def test_check_failure(runner, with_tunnel):
-    with patch("autossh_mgr.check.subprocess.run") as mock_run:
+    mock_sock = MagicMock()
+    with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
+         patch("autossh_mgr.check.subprocess.run") as mock_run:
         mock_run.return_value = MagicMock(returncode=255, stderr="Connection refused")
         result = runner.invoke(cli, ["check", "web-service"])
     assert result.exit_code != 0