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,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