feat: replace HTTP health check with TLS handshake

Replace _http_check (HTTP GET to httpbin.org) with _tls_check that
performs a TLS handshake through the proxy chain. Multiple targets
(google, cloudflare, amazon) rotated round-robin eliminate the single
point of failure. Lighter, faster, harder to block than HTTP.

- Add test_targets config field (replaces test_url)
- Backward compat: legacy test_url extracts hostname automatically
- Add ssl.create_default_context() and round-robin index to ProxyPool
- Update docs (example.yaml, USAGE.md, CHEATSHEET.md)
This commit is contained in:
user
2026-02-17 18:26:21 +01:00
parent 3638c607da
commit e78fc8dc3c
7 changed files with 230 additions and 38 deletions

View File

@@ -175,3 +175,40 @@ class TestConfig:
cfg_file.write_text("listen: 1080\n")
c = load_config(cfg_file)
assert c.tor is None
def test_proxy_pool_test_targets(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"proxy_pool:\n"
" sources: []\n"
" test_targets:\n"
" - host-a.example.com\n"
" - host-b.example.com\n"
)
c = load_config(cfg_file)
assert c.proxy_pool is not None
assert c.proxy_pool.test_targets == ["host-a.example.com", "host-b.example.com"]
assert c.proxy_pool.test_url == ""
def test_proxy_pool_legacy_test_url(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"proxy_pool:\n"
" sources: []\n"
" test_url: http://httpbin.org/ip\n"
)
c = load_config(cfg_file)
assert c.proxy_pool is not None
assert c.proxy_pool.test_targets == ["httpbin.org"]
def test_proxy_pool_defaults(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"proxy_pool:\n"
" sources: []\n"
)
c = load_config(cfg_file)
assert c.proxy_pool is not None
assert c.proxy_pool.test_targets == [
"www.google.com", "www.cloudflare.com", "www.amazon.com",
]

View File

@@ -1,6 +1,9 @@
"""Tests for the managed proxy pool."""
import asyncio
import ssl
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -468,3 +471,118 @@ class TestProxyPoolPersistence:
entry = pool2._proxies["socks5://1.2.3.4:1080"]
assert entry.hop.username == "user"
assert entry.hop.password == "pass"
class TestTlsCheck:
"""Test TLS handshake health check."""
def _make_pool(self, **kwargs):
cfg = ProxyPoolConfig(sources=[], **kwargs)
return ProxyPool(cfg, [], timeout=10.0)
def test_success(self):
pool = self._make_pool(test_targets=["www.example.com"])
mock_writer = MagicMock()
mock_writer.is_closing.return_value = False
mock_transport = MagicMock()
mock_protocol = MagicMock()
mock_transport.get_protocol.return_value = mock_protocol
mock_writer.transport = mock_transport
new_transport = MagicMock()
chain_ret = (MagicMock(), mock_writer)
with (
patch("s5p.pool.build_chain", new_callable=AsyncMock, return_value=chain_ret),
patch("asyncio.get_running_loop") as mock_loop_fn,
):
mock_loop = MagicMock()
mock_loop.start_tls = AsyncMock(return_value=new_transport)
mock_loop_fn.return_value = mock_loop
result = asyncio.run(pool._tls_check([]))
assert result is True
mock_loop.start_tls.assert_called_once_with(
mock_transport, mock_protocol, pool._ssl_ctx,
server_hostname="www.example.com",
)
new_transport.close.assert_called_once()
def test_build_chain_failure(self):
pool = self._make_pool(test_targets=["www.example.com"])
with patch(
"s5p.pool.build_chain", new_callable=AsyncMock,
side_effect=ConnectionError("refused"),
):
result = asyncio.run(pool._tls_check([]))
assert result is False
def test_handshake_failure(self):
pool = self._make_pool(test_targets=["www.example.com"])
mock_writer = MagicMock()
mock_writer.is_closing.return_value = False
mock_transport = MagicMock()
mock_transport.get_protocol.return_value = MagicMock()
mock_writer.transport = mock_transport
chain_ret = (MagicMock(), mock_writer)
with (
patch("s5p.pool.build_chain", new_callable=AsyncMock, return_value=chain_ret),
patch("asyncio.get_running_loop") as mock_loop_fn,
):
mock_loop = MagicMock()
mock_loop.start_tls = AsyncMock(
side_effect=ssl.SSLError("handshake failed"),
)
mock_loop_fn.return_value = mock_loop
result = asyncio.run(pool._tls_check([]))
assert result is False
def test_round_robin_rotation(self):
targets = ["host-a.example.com", "host-b.example.com", "host-c.example.com"]
pool = self._make_pool(test_targets=targets)
selected: list[str] = []
async def fake_build_chain(chain, host, port, timeout=None):
selected.append(host)
raise ConnectionError("test")
with patch("s5p.pool.build_chain", side_effect=fake_build_chain):
for _ in range(6):
asyncio.run(pool._tls_check([]))
assert selected == ["host-a.example.com", "host-b.example.com", "host-c.example.com",
"host-a.example.com", "host-b.example.com", "host-c.example.com"]
def test_empty_targets(self):
pool = self._make_pool(test_targets=[])
result = asyncio.run(pool._tls_check([]))
assert result is False
class TestProxyPoolConfigCompat:
"""Test backward compatibility for test_url -> test_targets."""
def test_legacy_test_url_converts(self):
cfg = ProxyPoolConfig(test_url="http://httpbin.org/ip")
assert cfg.test_targets == ["httpbin.org"]
def test_explicit_test_targets_wins(self):
cfg = ProxyPoolConfig(
test_url="http://httpbin.org/ip",
test_targets=["custom.example.com"],
)
assert cfg.test_targets == ["custom.example.com"]
def test_defaults_when_neither_set(self):
cfg = ProxyPoolConfig()
assert cfg.test_targets == ["www.google.com", "www.cloudflare.com", "www.amazon.com"]
assert cfg.test_url == ""