From 6124abbdb2cde58fe5f21ba2b41a213d41ad3c70 Mon Sep 17 00:00:00 2001 From: tech Date: Wed, 5 Aug 2026 15:27:04 +0800 Subject: [PATCH] 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. --- src/autossh_mgr/check.py | 22 +++++++++++- tests/unit/test_check.py | 59 ++++++++++++++++++++++++++------ tests/unit/test_cli_lifecycle.py | 8 +++-- 3 files changed, 75 insertions(+), 14 deletions(-) diff --git a/src/autossh_mgr/check.py b/src/autossh_mgr/check.py index 03d9c7d..a893b87 100644 --- a/src/autossh_mgr/check.py +++ b/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" diff --git a/tests/unit/test_check.py b/tests/unit/test_check.py index b7f1f28..91b85d8 100644 --- a/tests/unit/test_check.py +++ b/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 diff --git a/tests/unit/test_cli_lifecycle.py b/tests/unit/test_cli_lifecycle.py index 813dc70..e38de9c 100644 --- a/tests/unit/test_cli_lifecycle.py +++ b/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