File size: 6,356 Bytes
108e227 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | """Tests for the API server bind-address startup guard.
Validates that is_network_accessible() correctly classifies addresses and
that connect() refuses to start without API_SERVER_KEY.
"""
import socket
from unittest.mock import patch
import pytest
from gateway.config import PlatformConfig
from gateway.platforms.api_server import APIServerAdapter
from gateway.platforms.base import is_network_accessible
# ---------------------------------------------------------------------------
# Unit tests: is_network_accessible()
# ---------------------------------------------------------------------------
class TestIsNetworkAccessible:
"""Direct tests for the address classification helper."""
# -- Loopback (safe, should return False) --
def test_ipv4_mapped_loopback(self):
# ::ffff:127.0.0.1 — Python's is_loopback returns False for mapped
# addresses; the helper must unwrap and check ipv4_mapped.
assert is_network_accessible("::ffff:127.0.0.1") is False
# -- Network-accessible (should return True) --
def test_ipv6_wildcard(self):
# This is the bypass vector that the string-based check missed.
assert is_network_accessible("::") is True
def test_private_ipv4(self):
assert is_network_accessible("10.0.0.1") is True
def test_public_ipv4(self):
assert is_network_accessible("8.8.8.8") is True
# -- Hostname resolution --
def test_hostname_mixed_resolution(self):
"""If a hostname resolves to both loopback and non-loopback, it's
network-accessible (any non-loopback address is enough)."""
mixed_result = [
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("127.0.0.1", 0)),
(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("10.0.0.1", 0)),
]
with patch("gateway.platforms.base._socket.getaddrinfo", return_value=mixed_result):
assert is_network_accessible("dual-host.local") is True
# ---------------------------------------------------------------------------
# Integration tests: connect() startup guard
# ---------------------------------------------------------------------------
class TestConnectBindGuard:
"""Verify that connect() refuses dangerous configurations."""
@pytest.mark.asyncio
async def test_refuses_loopback_without_key(self):
"""Loopback binds are still an auth boundary and require API_SERVER_KEY."""
adapter = APIServerAdapter(PlatformConfig(enabled=True, extra={"host": "127.0.0.1"}))
assert adapter._api_key == ""
assert is_network_accessible(adapter._host) is False
result = await adapter.connect()
assert result is False
assert adapter._app is None
assert adapter._background_tasks == set()
@pytest.mark.asyncio
async def test_allows_wildcard_with_key(self):
"""Non-loopback with a key should pass the guard."""
adapter = APIServerAdapter(
PlatformConfig(enabled=True, extra={"host": "0.0.0.0", "key": "sk-test"})
)
# The guard checks: is_network_accessible(host) AND NOT api_key
# With a key set, the guard should not block.
assert adapter._api_key == "sk-test"
assert is_network_accessible("0.0.0.0") is True
# Combined: the guard condition is False (key is set), so it passes
# ---------------------------------------------------------------------------
# Integration tests: bind mechanics (direct bind, no pre-probe — #10297)
# ---------------------------------------------------------------------------
class TestBindMechanics:
"""connect() binds directly instead of pre-probing 127.0.0.1.
The old ``_port_is_available()`` probe connected to 127.0.0.1 only and
reported a lingering TIME_WAIT socket as "in use", failing gateway
restarts for up to ~60s (#10297). The fix removes the probe: bind
directly, keep SO_REUSEADDR default semantics on Linux (rebind past
TIME_WAIT), and surface a real bind conflict as a clean ``False`` with
the runner torn down.
"""
_KEY = "sk-test-strong-key-0123456789"
def _make_adapter(self, port: int) -> APIServerAdapter:
return APIServerAdapter(
PlatformConfig(
enabled=True,
extra={"host": "127.0.0.1", "port": port, "key": self._KEY},
)
)
@staticmethod
def _free_port() -> int:
with socket.socket() as s:
s.bind(("", 0))
return s.getsockname()[1]
@pytest.mark.asyncio
async def test_immediate_rebind_after_disconnect(self):
"""A restarted adapter can rebind the same port immediately.
This is the #10297 symptom: the old pre-probe (and disabled address
reuse) made a quick gateway restart fail while the previous socket
sat in TIME_WAIT.
"""
port = self._free_port()
first = self._make_adapter(port)
assert await first.connect() is True
await first.disconnect()
second = self._make_adapter(port)
try:
assert await second.connect() is True
finally:
await second.disconnect()
@pytest.mark.asyncio
async def test_port_conflict_sets_non_retryable_fatal_error(self):
"""A real port conflict (EADDRINUSE) must set a non-retryable fatal
error so the reconnect watcher drops the platform from the retry
queue instead of looping indefinitely.
Previously connect() returned bare ``False``, which the reconnect
watcher treated as retryable — retrying every 5 minutes forever,
filling errors.log and leaking 2 fds per retry (#52132: 1568+
retries over 5 days in a multi-profile setup).
"""
port = self._free_port()
first = self._make_adapter(port)
assert await first.connect() is True
second = self._make_adapter(port)
try:
result = await second.connect()
assert result is False
assert second.has_fatal_error is True
assert second.fatal_error_retryable is False
assert second.fatal_error_code == "api_server_port_in_use"
assert str(port) in (second.fatal_error_message or "")
finally:
await first.disconnect()
await second.disconnect()
|