feat(check): add Layer 1 local backend probe before SSH auth

check_connectivity now runs a two-layer short-circuit probe:
  Layer 1: TCP connect to (local_host, local_port) to confirm the
           backend service (e.g. Gitea, Next.js) is actually listening.
  Layer 2: original SSH auth probe (unchanged).

This fixes the common false-positive where check returned OK while the
tunneled backend was down. Unlike the previously removed _is_port_in_use
bind check (commit 4c9cd00), this uses connect() with correct semantics:
the forward target SHOULD be listening, not free.
This commit is contained in:
2026-08-05 15:27:04 +08:00
parent 4c9cd00c90
commit 6124abbdb2
3 changed files with 75 additions and 14 deletions

View File

@@ -1,8 +1,22 @@
import os import os
import socket
import subprocess import subprocess
from autossh_mgr.config import TunnelConfig 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]: def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
return [ return [
"ssh", "ssh",
@@ -16,9 +30,15 @@ def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]: def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
ok, msg = check_local_backend(tunnel)
if not ok:
return False, msg
result = subprocess.run( result = subprocess.run(
build_ssh_check_cmd(tunnel), build_ssh_check_cmd(tunnel),
capture_output=True, capture_output=True,
text=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"

View File

@@ -1,6 +1,10 @@
import pytest import pytest
from unittest.mock import patch, MagicMock 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 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 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): def test_check_connectivity_success(tunnel):
mock_result = MagicMock() mock_sock = MagicMock()
mock_result.returncode = 0 with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
mock_result.stderr = "" patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=0, stderr="")) as mock_run:
with patch("autossh_mgr.check.subprocess.run", return_value=mock_result) as mock_run:
success, msg = check_connectivity(tunnel) success, msg = check_connectivity(tunnel)
assert success is True assert success is True
assert msg == "" assert msg == "reachable + backend up"
called_cmd = mock_run.call_args[0][0] called_cmd = mock_run.call_args[0][0]
assert called_cmd == build_ssh_check_cmd(tunnel) assert called_cmd == build_ssh_check_cmd(tunnel)
def test_check_connectivity_failure(tunnel): def test_check_connectivity_ssh_failure(tunnel):
mock_result = MagicMock() mock_sock = MagicMock()
mock_result.returncode = 255 with patch("autossh_mgr.check.socket.socket", return_value=mock_sock), \
mock_result.stderr = "Connection refused" patch("autossh_mgr.check.subprocess.run", return_value=MagicMock(returncode=255, stderr="Connection refused")):
with patch("autossh_mgr.check.subprocess.run", return_value=mock_result):
success, msg = check_connectivity(tunnel) success, msg = check_connectivity(tunnel)
assert success is False assert success is False
assert "SSH auth failed" in msg
assert "Connection refused" in msg assert "Connection refused" in msg

View File

@@ -87,7 +87,9 @@ def test_status_single(runner, config_dir, with_tunnel):
def test_check_success(runner, 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="") mock_run.return_value = MagicMock(returncode=0, stderr="")
result = runner.invoke(cli, ["check", "web-service"]) result = runner.invoke(cli, ["check", "web-service"])
assert result.exit_code == 0 assert result.exit_code == 0
@@ -95,7 +97,9 @@ def test_check_success(runner, with_tunnel):
def test_check_failure(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") mock_run.return_value = MagicMock(returncode=255, stderr="Connection refused")
result = runner.invoke(cli, ["check", "web-service"]) result = runner.invoke(cli, ["check", "web-service"])
assert result.exit_code != 0 assert result.exit_code != 0