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:
@@ -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"
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user