Compare commits

..

5 Commits

Author SHA1 Message Date
user fa3621806d feat: add per-listener SOCKS5 server authentication (RFC 1929)
Per-listener username/password auth via `auth:` config key. When set,
clients must negotiate method 0x02 and pass RFC 1929 subnegotiation;
no-auth (0x00) is rejected to prevent downgrade. Listeners without
`auth` keep current no-auth behavior.

Includes auth_failures metric, API integration (/status auth flag,
/config auth_users count without exposing passwords), config parsing
with YAML int coercion, integration tests (success, failure, method
rejection, no-auth unchanged), and documentation updates.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-21 17:03:03 +01:00
user 76dac61eb6 fix: add shutdown timeout so cProfile data is written on SIGTERM
srv.wait_closed() blocked indefinitely on active relay connections,
preventing serve() from returning and prof.dump_stats() from running.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-21 16:32:50 +01:00
user 918d03cc58 feat: skip pool hops for .onion destinations
Onion addresses require Tor to resolve, so pool proxies after Tor
would break connectivity. Detect .onion targets and use the static
chain only (Tor), skipping pool selection and retries.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-21 02:28:34 +01:00
user c191942712 feat: add bypass rules, weighted pool selection, integration tests
Per-listener bypass rules skip the chain for local/private destinations
(CIDR, exact IP/hostname, domain suffix). Weighted multi-candidate pool
selection biases toward pools with more alive proxies. End-to-end
integration tests validate the full client->s5p->hop->target path using
mock SOCKS5 proxies.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-20 19:58:12 +01:00
user ef0d8f347b feat: add per-hop pool references in listener chains
Allow listeners to mix named pools in a single chain using pool:name
syntax. Bare "pool" continues to use the listener's default pool.
Replaces pool_hops field with pool_seq list; pool_hops is now a
backward-compatible property. Each hop draws from its own pool and
failure reporting targets the correct source pool.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-20 17:50:17 +01:00
15 changed files with 1543 additions and 86 deletions
+5 -10
View File
@@ -93,6 +93,8 @@ proxy_pools:
max_fails: 3
# Multi-listener: each port gets a chain depth and pool assignment
# Use "pool" for listener default, "pool:name" for explicit pool per hop,
# or [pool:a, pool:b] for random choice from candidates per connection
listeners:
- listen: 0.0.0.0:1080
pool: clean
@@ -101,19 +103,13 @@ listeners:
- pool # Tor + 2 clean proxies
- pool
- listen: 0.0.0.0:1081
pool: clean
chain:
- socks5://127.0.0.1:9050
- pool # Tor + 1 clean proxy
- [pool:clean, pool:mitm] # random choice per connection
- [pool:clean, pool:mitm] # independent random choice
- listen: 0.0.0.0:1082
chain:
- socks5://127.0.0.1:9050 # Tor only
- listen: 0.0.0.0:1083
pool: mitm
chain:
- socks5://127.0.0.1:9050
- pool # Tor + 2 MITM proxies
- pool
# Singular proxy_pool: still works (becomes pool "default")
@@ -150,9 +146,8 @@ Options:
```
:1080 Client -> s5p -> Tor -> [clean] -> [clean] -> Dest (2 clean hops)
:1081 Client -> s5p -> Tor -> [clean] -> Dest (1 clean hop)
:1081 Client -> s5p -> Tor -> [clean|mitm] -> [clean|mitm] -> Dest (random)
:1082 Client -> s5p -> Tor -> Dest (Tor only)
:1083 Client -> s5p -> Tor -> [mitm] -> [mitm] -> Dest (2 MITM hops)
```
s5p connects to Hop1 via TCP, negotiates the hop protocol (SOCKS5/4/HTTP),
+1 -1
View File
@@ -42,6 +42,6 @@
## v1.0.0
- [ ] Stable API and config format
- [ ] Comprehensive test suite with mock proxies
- [ ] Comprehensive test suite with mock proxies (integration tests done)
- [ ] Systemd service unit
- [ ] Performance benchmarks
+7 -1
View File
@@ -59,6 +59,12 @@
- [x] API: merged `/pool` with per-pool breakdown, `/status` pools summary
- [x] Backward compat: singular `proxy_pool:` registers as `"default"`
- [x] Integration tests with mock SOCKS5 proxy (end-to-end)
- [x] Per-destination bypass rules (CIDR, suffix, exact match)
- [x] Weighted multi-candidate pool selection
- [x] Onion chain-only routing (.onion skips pool hops)
- [x] Graceful shutdown timeout (fixes cProfile data dump)
## Next
- [ ] Integration tests with mock proxy server
- [x] Integration tests with mock proxy server
- [ ] SOCKS5 server-side authentication
-1
View File
@@ -4,7 +4,6 @@
- SOCKS5 BIND and UDP ASSOCIATE commands
- Chain randomization modes (round-robin, sticky-per-destination)
- Per-destination chain rules (bypass chain for local addresses)
- Systemd socket activation
- Per-pool health test chain override (different base chain per pool)
- Pool-level proxy protocol filter (only socks5 from pool X, only http from pool Y)
+26 -7
View File
@@ -87,30 +87,49 @@ chain:
# a random alive proxy from the named pool (or "default" if unnamed).
# Multiple "pool" entries = multiple pool hops (deeper chaining).
#
# Per-hop pool references: use "pool:name" to draw from a specific pool
# at that hop position. Bare "pool" uses the listener's "pool:" default.
# This lets a single listener mix pools in one chain.
#
# Multi-candidate hops: use a YAML list to randomly pick from a set of
# pools at each hop. On each connection, one pool is chosen per hop.
#
# listeners:
# - listen: 0.0.0.0:1080
# pool: clean # draw from "clean" pool
# pool: clean # default for bare "pool"
# auth: # SOCKS5 username/password (RFC 1929)
# alice: s3cret # username: password
# bob: hunter2
# bypass: # skip chain for these destinations
# - 127.0.0.0/8 # loopback
# - 10.0.0.0/8 # RFC 1918
# - 192.168.0.0/16 # RFC 1918
# - 172.16.0.0/12 # RFC 1918
# - fc00::/7 # IPv6 ULA
# - localhost # exact hostname
# - .local # domain suffix
# chain:
# - socks5://127.0.0.1:9050
# - pool # Tor + 2 clean pool proxies
# - pool
# - [pool:clean, pool:mitm] # random choice per connection
# - [pool:clean, pool:mitm] # independent random choice
#
# - listen: 0.0.0.0:1081
# pool: clean
# chain:
# - socks5://127.0.0.1:9050
# - pool # Tor + 1 clean pool proxy
# - pool # bare: uses default "clean"
# - pool
#
# - listen: 0.0.0.0:1082
# chain:
# - socks5://127.0.0.1:9050 # Tor only (no pool hops)
#
# - listen: 0.0.0.0:1083
# pool: mitm # draw from "mitm" pool
# pool: clean
# chain:
# - socks5://127.0.0.1:9050
# - pool # Tor + 2 MITM pool proxies
# - pool
# - pool # bare "pool" = clean
# - pool:mitm # explicit = mitm
#
# When using "listeners:", the top-level "listen" and "chain" keys are ignored.
# If "listeners:" is absent, the old format is used (single listener).
+48 -3
View File
@@ -43,7 +43,7 @@ cp config/example.yaml config/s5p.yaml # create live config (gitignored)
```yaml
listeners:
- listen: 0.0.0.0:1080
pool: clean # named pool assignment
pool: clean # default for bare "pool"
chain:
- socks5://127.0.0.1:9050
- pool # Tor + 2 clean hops
@@ -52,18 +52,63 @@ listeners:
pool: clean
chain:
- socks5://127.0.0.1:9050
- pool # Tor + 1 clean hop
- pool:clean # per-hop: explicit clean
- pool:mitm # per-hop: explicit mitm
- listen: 0.0.0.0:1082
chain:
- socks5://127.0.0.1:9050 # Tor only
- listen: 0.0.0.0:1083
pool: mitm # MITM-capable proxies
chain:
- socks5://127.0.0.1:9050
- [pool:clean, pool:mitm] # random choice per connection
- [pool:clean, pool:mitm] # independent random choice
```
Per-hop pool: `pool` = listener default, `pool:name` = explicit pool,
`[pool:a, pool:b]` = random choice from candidates.
## Bypass Rules (config)
```yaml
listeners:
- listen: 0.0.0.0:1080
bypass:
- 127.0.0.0/8 # CIDR
- 10.0.0.0/8 # CIDR
- 192.168.0.0/16 # CIDR
- localhost # exact hostname
- .local # domain suffix
chain:
- socks5://127.0.0.1:9050
- pool
```
| Pattern | Type | Matches |
|---------|------|---------|
| `10.0.0.0/8` | CIDR | IPs in network |
| `127.0.0.1` | Exact IP | That IP only |
| `localhost` | Exact host | String equal |
| `.local` | Suffix | `*.local` and `local` |
## Listener Authentication (config)
```yaml
listeners:
- listen: 0.0.0.0:1080
auth:
alice: s3cret
bob: hunter2
chain:
- socks5://127.0.0.1:9050
- pool
```
```bash
curl -x socks5h://alice:s3cret@127.0.0.1:1080 https://example.com
```
No `auth:` key = no authentication required (default).
## Multi-Tor Round-Robin (config)
```yaml
+136
View File
@@ -177,9 +177,57 @@ listeners:
- pool
```
### Per-hop pool references
Use `pool:name` to draw from a specific named pool at that hop position.
Bare `pool` uses the listener's `pool:` default. This lets a single listener
mix pools in one chain.
```yaml
listeners:
- listen: 0.0.0.0:1080
pool: clean # default for bare "pool"
chain:
- socks5://10.200.1.13:9050
- pool:clean # explicit: from clean pool
- pool:mitm # explicit: from mitm pool
- listen: 0.0.0.0:1081
pool: clean
chain:
- socks5://10.200.1.13:9050
- pool # bare: uses default "clean"
- pool:mitm # explicit: from mitm pool
```
| Syntax | Resolves to |
|--------|-------------|
| `pool` | Listener's `pool:` value, or `"default"` if unset |
| `pool:name` | Named pool `name` (case-sensitive) |
| `pool:` | Same as bare `pool` (empty name = default) |
| `Pool:name` | Prefix is case-insensitive; name is case-sensitive |
| `[pool:a, pool:b]` | Random choice from candidates `a` or `b` per connection |
The `pool` keyword in a chain means "append a random alive proxy from the
assigned pool". Multiple `pool` entries = multiple pool hops (deeper chaining).
### Multi-candidate pool hops
Use a YAML list to randomly pick from a set of candidate pools at each hop.
On each connection, one candidate is chosen at random per hop (independently).
```yaml
listeners:
- listen: 0.0.0.0:1080
chain:
- socks5://10.200.1.13:9050
- [pool:clean, pool:mitm] # hop 1: random choice
- [pool:clean, pool:mitm] # hop 2: random choice
```
Single-element pool references (`pool`, `pool:name`) and multi-candidate
lists can be mixed freely in the same chain. All existing syntax is unchanged.
When `pool:` is omitted on a listener with pool hops, it defaults to
`"default"`. A listener referencing an unknown pool name causes a fatal
error at startup. Listeners without pool hops ignore the `pool:` key.
@@ -194,6 +242,94 @@ error at startup. Listeners without pool hops ignore the `pool:` key.
| FirstHopPool | per unique first hop | Listeners with same first hop share it |
| Chain + pool_hops | per listener | Each listener has its own chain depth |
## Listener Authentication
Per-listener SOCKS5 username/password authentication (RFC 1929). When `auth`
is configured on a listener, clients must authenticate before connecting.
Listeners without `auth` continue to accept unauthenticated connections.
```yaml
listeners:
- listen: 0.0.0.0:1080
auth:
alice: s3cret
bob: hunter2
chain:
- socks5://127.0.0.1:9050
- pool
```
### Testing with curl
```bash
curl --proxy socks5h://alice:s3cret@127.0.0.1:1080 https://example.com
```
### Behavior
| Client offers | Listener has `auth` | Result |
|---------------|---------------------|--------|
| `0x00` (no-auth) | yes | Rejected (`0xFF`) |
| `0x02` (user/pass) | yes | Subnegotiation, then accept/reject |
| `0x00` (no-auth) | no | Accepted (current behavior) |
| `0x02` (user/pass) | no | Rejected (`0xFF`) |
Authentication failures are logged and counted in the `auth_fail` metric.
The `/status` API endpoint includes `"auth": true` on authenticated listeners.
The `/config` endpoint shows `"auth_users": N` (passwords are never exposed).
### Mixed listeners
Different listeners can have different auth settings:
```yaml
listeners:
- listen: 0.0.0.0:1080 # public, no auth
chain:
- socks5://127.0.0.1:9050
- listen: 0.0.0.0:1081 # authenticated
auth:
alice: s3cret
chain:
- socks5://127.0.0.1:9050
- pool
```
## Bypass Rules
Per-listener rules to skip the chain for specific destinations. When a target
matches a bypass rule, s5p connects directly (no chain, no pool hops).
```yaml
listeners:
- listen: 0.0.0.0:1080
bypass:
- 127.0.0.0/8 # CIDR: loopback
- 10.0.0.0/8 # CIDR: RFC 1918
- 192.168.0.0/16 # CIDR: RFC 1918
- fc00::/7 # CIDR: IPv6 ULA
- localhost # exact hostname
- .local # domain suffix (matches *.local and local)
chain:
- socks5://127.0.0.1:9050
- pool
```
### Rule syntax
| Pattern | Type | Matches |
|---------|------|---------|
| `10.0.0.0/8` | CIDR | Any IP in the network |
| `127.0.0.1` | Exact IP | That IP only |
| `localhost` | Exact hostname | String-equal match |
| `.local` | Domain suffix | `*.local` and `local` itself |
CIDR rules only match IP addresses, not hostnames. Domain suffix rules only
match hostnames, not IPs. Exact rules match both (string compare for hostnames,
parsed compare for IPs).
When bypass is active, retries are disabled (direct connections are not retried).
### Backward compatibility
When no `listeners:` key is present, the old `listen`/`chain` format creates
+17
View File
@@ -49,6 +49,19 @@ def _json_response(
writer.write(header.encode() + payload)
# -- helpers -----------------------------------------------------------------
def _multi_pool(lc) -> bool:
"""Check if a listener uses more than one distinct pool."""
return len({n for c in lc.pool_seq for n in c}) > 1
def _pool_seq_entry(lc) -> dict:
"""Build pool_seq dict entry for API responses."""
return {"pool_seq": lc.pool_seq}
# -- route handlers ----------------------------------------------------------
@@ -88,6 +101,8 @@ def _handle_status(ctx: dict) -> tuple[int, dict]:
"chain": [str(h) for h in lc.chain],
"pool_hops": lc.pool_hops,
**({"pool": lc.pool_name} if lc.pool_name else {}),
**(_pool_seq_entry(lc) if _multi_pool(lc) else {}),
**({"auth": True} if lc.auth else {}),
"latency": metrics.get_listener_latency(
f"{lc.listen_host}:{lc.listen_port}"
).stats(),
@@ -166,6 +181,8 @@ def _handle_config(ctx: dict) -> tuple[int, dict]:
"chain": [str(h) for h in lc.chain],
"pool_hops": lc.pool_hops,
**({"pool": lc.pool_name} if lc.pool_name else {}),
**(_pool_seq_entry(lc) if _multi_pool(lc) else {}),
**({"auth_users": len(lc.auth)} if lc.auth else {}),
}
for lc in config.listeners
],
+46 -5
View File
@@ -85,8 +85,15 @@ class ListenerConfig:
listen_host: str = "127.0.0.1"
listen_port: int = 1080
chain: list[ChainHop] = field(default_factory=list)
pool_hops: int = 0
pool_seq: list[list[str]] = field(default_factory=list)
pool_name: str = ""
bypass: list[str] = field(default_factory=list)
auth: dict[str, str] = field(default_factory=dict)
@property
def pool_hops(self) -> int:
"""Number of pool hops (backward compat)."""
return len(self.pool_seq)
@dataclass
@@ -188,6 +195,21 @@ def _parse_pool_config(pool_raw: dict) -> ProxyPoolConfig:
return ProxyPoolConfig(**kwargs)
def _parse_pool_ref(item: str, default: str) -> str:
"""Resolve a pool reference string to a pool name.
``pool`` or ``pool:`` -> *default*; ``pool:name`` -> ``name``.
The ``pool`` prefix is matched case-insensitively.
"""
lower = item.lower()
if lower == "pool" or lower == "pool:":
return default
if lower.startswith("pool:"):
_, _, name = item.partition(":")
return name if name else default
raise ValueError(f"not a pool reference: {item!r}")
def load_config(path: str | Path) -> Config:
"""Load configuration from a YAML file."""
path = Path(path)
@@ -302,15 +324,30 @@ def load_config(path: str | Path) -> Config:
lc.listen_port = int(port_str)
elif isinstance(listen, (str, int)) and listen:
lc.listen_port = int(listen)
if "bypass" in entry:
lc.bypass = list(entry["bypass"])
if "auth" in entry:
auth_raw = entry["auth"]
if isinstance(auth_raw, dict):
lc.auth = {str(k): str(v) for k, v in auth_raw.items()}
if "pool" in entry:
lc.pool_name = entry["pool"]
default_pool = lc.pool_name or "default"
chain_raw = entry.get("chain", [])
for item in chain_raw:
if isinstance(item, str) and item.lower() == "pool":
lc.pool_hops += 1
elif isinstance(item, str):
if isinstance(item, str):
lower = item.lower()
if lower == "pool" or lower.startswith("pool:"):
lc.pool_seq.append([_parse_pool_ref(item, default_pool)])
else:
lc.chain.append(parse_proxy_url(item))
elif isinstance(item, dict):
# YAML parses "pool:" and "pool: name" as dicts
pool_key = next((k for k in item if k.lower() == "pool"), None)
if pool_key is not None and len(item) == 1:
name = item[pool_key]
lc.pool_seq.append([name if name else default_pool])
else:
lc.chain.append(
ChainHop(
proto=item.get("proto", "socks5"),
@@ -320,6 +357,10 @@ def load_config(path: str | Path) -> Config:
password=item.get("password"),
)
)
elif isinstance(item, list):
# multi-candidate hop: [pool:clean, pool:mitm]
candidates = [_parse_pool_ref(str(el), default_pool) for el in item]
lc.pool_seq.append(candidates)
config.listeners.append(lc)
else:
# backward compat: build single listener from top-level fields
@@ -330,7 +371,7 @@ def load_config(path: str | Path) -> Config:
)
# legacy behavior: if proxy_pool configured, auto-append 1 pool hop
if config.proxy_pool and config.proxy_pool.sources:
lc.pool_hops = 1
lc.pool_seq = [["default"]]
config.listeners.append(lc)
return config
+4 -1
View File
@@ -82,6 +82,7 @@ class Metrics:
self.retries: int = 0
self.bytes_in: int = 0
self.bytes_out: int = 0
self.auth_failures: int = 0
self.active: int = 0
self.started: float = time.monotonic()
self.conn_rate: RateTracker = RateTracker()
@@ -103,9 +104,10 @@ class Metrics:
lat = self.latency.stats()
p50 = f" p50={lat['p50']:.1f}ms" if lat else ""
p95 = f" p95={lat['p95']:.1f}ms" if lat else ""
auth = f" auth_fail={self.auth_failures}" if self.auth_failures else ""
return (
f"conn={self.connections} ok={self.success} fail={self.failed} "
f"retries={self.retries} active={self.active} "
f"retries={self.retries} active={self.active}{auth} "
f"in={_human_bytes(self.bytes_in)} out={_human_bytes(self.bytes_out)} "
f"rate={rate:.2f}/s{p50}{p95} "
f"up={h}h{m:02d}m{s:02d}s"
@@ -118,6 +120,7 @@ class Metrics:
"success": self.success,
"failed": self.failed,
"retries": self.retries,
"auth_failures": self.auth_failures,
"active": self.active,
"bytes_in": self.bytes_in,
"bytes_out": self.bytes_out,
+131 -25
View File
@@ -3,7 +3,9 @@
from __future__ import annotations
import asyncio
import ipaddress
import logging
import random
import signal
import struct
import time
@@ -70,13 +72,44 @@ def _socks5_reply(rep: int) -> bytes:
return struct.pack("!BBB", 0x05, rep, 0x00) + b"\x01\x00\x00\x00\x00\x00\x00"
def _bypass_match(rules: list[str], host: str) -> bool:
"""Check if host matches any bypass rule (CIDR, suffix, or exact)."""
addr = None
try:
addr = ipaddress.ip_address(host)
except ValueError:
pass
for rule in rules:
if "/" in rule:
if addr is not None:
try:
if addr in ipaddress.ip_network(rule, strict=False):
return True
except ValueError:
pass
elif rule.startswith("."):
if addr is None and (host.endswith(rule) or host == rule[1:]):
return True
else:
if addr is not None:
try:
if addr == ipaddress.ip_address(rule):
return True
except ValueError:
pass
if host == rule:
return True
return False
async def _handle_client(
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
listener: ListenerConfig,
timeout: float,
retries: int,
proxy_pool: ProxyPool | None = None,
pool_seq: list[list[ProxyPool]] | None = None,
metrics: Metrics | None = None,
first_hop_pool: FirstHopPool | None = None,
tor_rr: _RoundRobin | None = None,
@@ -98,6 +131,43 @@ async def _handle_client(
return
methods = await client_reader.readexactly(header[1])
if listener.auth:
# require username/password auth (RFC 1929)
if 0x02 not in methods:
client_writer.write(b"\x05\xff")
await client_writer.drain()
return
client_writer.write(b"\x05\x02")
await client_writer.drain()
# subnegotiation: [ver, ulen, uname..., plen, passwd...]
ver = (await asyncio.wait_for(
client_reader.readexactly(1), timeout=10.0,
))[0]
if ver != 0x01:
client_writer.write(b"\x01\x01")
await client_writer.drain()
return
ulen = (await client_reader.readexactly(1))[0]
uname = (await client_reader.readexactly(ulen)).decode("utf-8", errors="replace")
plen = (await client_reader.readexactly(1))[0]
passwd = (await client_reader.readexactly(plen)).decode("utf-8", errors="replace")
if listener.auth.get(uname) != passwd:
logger.warning("[%s] auth failed for user %r", tag, uname)
if metrics:
metrics.auth_failures += 1
client_writer.write(b"\x01\x01")
await client_writer.drain()
return
client_writer.write(b"\x01\x00")
await client_writer.drain()
else:
# no auth required
if 0x00 not in methods:
client_writer.write(b"\x05\xff")
await client_writer.drain()
@@ -118,11 +188,24 @@ async def _handle_client(
target_host, target_port = await read_socks5_address(client_reader)
logger.info("[%s] connect %s:%d", tag, target_host, target_port)
# -- bypass / onion check --
bypass = bool(listener.bypass and _bypass_match(listener.bypass, target_host))
onion = target_host.endswith(".onion")
skip_pool = bypass or onion
if bypass:
logger.debug("[%s] bypass %s:%d", tag, target_host, target_port)
elif onion:
logger.debug("[%s] onion %s:%d (chain only)", tag, target_host, target_port)
# -- build chain (with retry) --
attempts = retries if proxy_pool and listener.pool_hops > 0 else 1
attempts = retries if pool_seq and not skip_pool else 1
last_err: Exception | None = None
for attempt in range(attempts):
if bypass:
effective_chain: list[ChainHop] = []
fhp = None
else:
effective_chain = list(listener.chain)
fhp = first_hop_pool
if tor_rr and effective_chain:
@@ -131,15 +214,17 @@ async def _handle_client(
if hop_pools:
fhp = hop_pools.get((node.host, node.port))
pool_hops: list[ChainHop] = []
if proxy_pool and listener.pool_hops > 0:
for _ in range(listener.pool_hops):
hop = await proxy_pool.get()
pool_hops: list[tuple[ChainHop, ProxyPool]] = []
if pool_seq and not skip_pool:
for candidates in pool_seq:
weights = [max(pp.alive_count, 1) for pp in candidates]
pp = random.choices(candidates, weights=weights)[0]
hop = await pp.get()
if hop:
pool_hops.append(hop)
pool_hops.append((hop, pp))
effective_chain.append(hop)
if pool_hops:
logger.debug("[%s] +pool %s", tag, " ".join(str(h) for h in pool_hops))
logger.debug("[%s] +pool %s", tag, " ".join(str(h) for h, _ in pool_hops))
try:
t0 = time.monotonic()
@@ -157,9 +242,9 @@ async def _handle_client(
break
except (ProtoError, TimeoutError, ConnectionError, OSError) as e:
last_err = e
if pool_hops and proxy_pool:
for hop in pool_hops:
proxy_pool.report_failure(hop)
if pool_hops:
for hop, pp in pool_hops:
pp.report_failure(hop)
if metrics:
metrics.retries += 1
if attempt + 1 < attempts:
@@ -295,17 +380,20 @@ async def serve(config: Config) -> None:
await pool.start()
proxy_pools["default"] = pool
def _pool_for(lc: ListenerConfig) -> ProxyPool | None:
"""Resolve the proxy pool for a listener."""
if lc.pool_hops <= 0:
return None
name = lc.pool_name or "default"
def _pools_for(lc: ListenerConfig) -> list[list[ProxyPool]]:
"""Resolve the ordered list of candidate proxy pools for a listener."""
result: list[list[ProxyPool]] = []
for candidates in lc.pool_seq:
resolved: list[ProxyPool] = []
for name in candidates:
if name not in proxy_pools:
raise RuntimeError(
f"listener {lc.listen_host}:{lc.listen_port} "
f"references unknown pool {name!r}"
)
return proxy_pools[name]
resolved.append(proxy_pools[name])
result.append(resolved)
return result
# -- per-unique first-hop connection pools --------------------------------
hop_pools: dict[tuple[str, int], FirstHopPool] = {}
@@ -361,17 +449,17 @@ async def serve(config: Config) -> None:
servers: list[asyncio.Server] = []
for lc in listeners:
hp = _hop_pool_for(lc)
lc_pool = _pool_for(lc)
lc_pools = _pools_for(lc)
async def on_client(
r: asyncio.StreamReader, w: asyncio.StreamWriter,
_lc: ListenerConfig = lc, _hp: FirstHopPool | None = hp,
_pool: ProxyPool | None = lc_pool,
_pools: list[list[ProxyPool]] = lc_pools,
) -> None:
async with sem:
await _handle_client(
r, w, _lc, config.timeout, config.retries,
_pool, metrics, _hp, tor_rr, hop_pools,
_pools, metrics, _hp, tor_rr, hop_pools,
)
srv = await asyncio.start_server(on_client, lc.listen_host, lc.listen_port)
@@ -380,10 +468,23 @@ async def serve(config: Config) -> None:
addr = f"{lc.listen_host}:{lc.listen_port}"
chain_desc = " -> ".join(str(h) for h in lc.chain) if lc.chain else "direct"
nhops = lc.pool_hops
pool_desc = f" + {nhops} pool hop{'s' if nhops != 1 else ''}" if nhops else ""
if lc_pool and lc_pool.name != "default":
pool_desc += f" [{lc_pool.name}]"
logger.info("listener %s chain: %s%s", addr, chain_desc, pool_desc)
pool_desc = ""
if nhops:
all_names = {n for cands in lc.pool_seq for n in cands}
hop_labels = ["|".join(cands) for cands in lc.pool_seq]
if len(all_names) == 1:
name = next(iter(all_names))
pool_desc = f" + {nhops} pool hop{'s' if nhops != 1 else ''}"
if name != "default":
pool_desc += f" [{name}]"
else:
pool_desc = f" + pool [{' -> '.join(hop_labels)}]"
bypass_desc = f" bypass: {len(lc.bypass)} rules" if lc.bypass else ""
auth_desc = f" auth: {len(lc.auth)} users" if lc.auth else ""
logger.info(
"listener %s chain: %s%s%s%s",
addr, chain_desc, pool_desc, bypass_desc, auth_desc,
)
logger.info("max_connections=%d", config.max_connections)
@@ -468,7 +569,12 @@ async def serve(config: Config) -> None:
for srv in servers:
srv.close()
for srv in servers:
await srv.wait_closed()
try:
await asyncio.wait_for(srv.wait_closed(), timeout=5.0)
except TimeoutError:
pass
if metrics.active:
logger.info("shutdown: %d connections still active", metrics.active)
if api_srv:
api_srv.close()
await api_srv.wait_closed()
+138
View File
@@ -0,0 +1,138 @@
"""Shared helpers for integration tests."""
from __future__ import annotations
import asyncio
import socket
import struct
from s5p.proto import encode_address, read_socks5_address
def free_port() -> int:
"""Return an available TCP port."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
# -- echo server -------------------------------------------------------------
async def _echo_handler(
reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
) -> None:
"""Echo back everything received, then close."""
try:
while True:
data = await reader.read(65536)
if not data:
break
writer.write(data)
await writer.drain()
except (ConnectionError, asyncio.CancelledError):
pass
finally:
writer.close()
await writer.wait_closed()
async def start_echo_server() -> tuple[str, int, asyncio.Server]:
"""Start a TCP echo server. Returns (host, port, server)."""
host = "127.0.0.1"
port = free_port()
srv = await asyncio.start_server(_echo_handler, host, port)
await srv.start_serving()
return host, port, srv
# -- mock SOCKS5 proxy -------------------------------------------------------
async def _mock_socks5_handler(
reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
) -> None:
"""Minimal SOCKS5 proxy: greeting, CONNECT, relay."""
remote_writer = None
try:
# greeting
header = await reader.readexactly(2)
if header[0] != 0x05:
return
await reader.readexactly(header[1]) # skip methods
writer.write(b"\x05\x00")
await writer.drain()
# connect request
req = await reader.readexactly(3)
if req[0] != 0x05 or req[1] != 0x01:
return
target_host, target_port = await read_socks5_address(reader)
# connect to actual target
try:
remote_reader, remote_writer = await asyncio.wait_for(
asyncio.open_connection(target_host, target_port),
timeout=5.0,
)
except (OSError, TimeoutError):
# connection refused reply
reply = struct.pack("!BBB", 0x05, 0x05, 0x00)
reply += b"\x01\x00\x00\x00\x00\x00\x00"
writer.write(reply)
await writer.drain()
return
# success reply
atyp, addr_bytes = encode_address(target_host)
reply = struct.pack("!BBB", 0x05, 0x00, 0x00)
reply += bytes([atyp]) + addr_bytes + struct.pack("!H", target_port)
writer.write(reply)
await writer.drain()
# relay both directions (close dst on EOF so peer sees shutdown)
async def _fwd(src: asyncio.StreamReader, dst: asyncio.StreamWriter) -> None:
try:
while True:
data = await src.read(65536)
if not data:
break
dst.write(data)
await dst.drain()
except (ConnectionError, asyncio.CancelledError):
pass
finally:
try:
dst.close()
await dst.wait_closed()
except OSError:
pass
await asyncio.gather(
_fwd(reader, remote_writer),
_fwd(remote_reader, writer),
)
except (ConnectionError, asyncio.IncompleteReadError, asyncio.CancelledError):
pass
finally:
if remote_writer:
remote_writer.close()
try:
await remote_writer.wait_closed()
except OSError:
pass
writer.close()
try:
await writer.wait_closed()
except OSError:
pass
async def start_mock_socks5() -> tuple[str, int, asyncio.Server]:
"""Start a mock SOCKS5 proxy. Returns (host, port, server)."""
host = "127.0.0.1"
port = free_port()
srv = await asyncio.start_server(_mock_socks5_handler, host, port)
await srv.start_serving()
return host, port, srv
+124 -3
View File
@@ -132,7 +132,7 @@ class TestHandleStatus:
ListenerConfig(
listen_host="0.0.0.0", listen_port=1081,
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_hops=1,
pool_seq=[["default"]],
),
],
)
@@ -151,6 +151,77 @@ class TestHandleStatus:
assert body["listeners"][1]["latency"] is None
class TestHandleStatusAuth:
"""Test auth flag in /status listener entries."""
def test_auth_flag_present(self):
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
auth={"alice": "s3cret", "bob": "hunter2"},
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_status(ctx)
assert body["listeners"][0]["auth"] is True
def test_auth_flag_absent_when_empty(self):
config = Config(
listeners=[
ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_status(ctx)
assert "auth" not in body["listeners"][0]
class TestHandleConfigAuth:
"""Test auth_users in /config listener entries."""
def test_auth_users_count(self):
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
auth={"alice": "s3cret", "bob": "hunter2"},
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_config(ctx)
assert body["listeners"][0]["auth_users"] == 2
def test_auth_users_absent_when_empty(self):
config = Config(
listeners=[
ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_config(ctx)
assert "auth_users" not in body["listeners"][0]
def test_passwords_not_exposed(self):
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
auth={"alice": "s3cret"},
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_config(ctx)
listener = body["listeners"][0]
# only count, never passwords
assert "auth_users" in listener
assert "auth" not in listener
assert "s3cret" not in str(body)
class TestHandleStatusPools:
"""Test GET /status with multiple named pools."""
@@ -170,6 +241,56 @@ class TestHandleStatusPools:
assert body["pools"]["mitm"] == {"alive": 3, "total": 8}
class TestHandleStatusMultiPool:
"""Test pool_seq appears in /status only for multi-pool listeners."""
def test_single_pool_no_pool_seq(self):
"""Single-pool listener: no pool_seq in response."""
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_seq=[["clean"], ["clean"]], pool_name="clean",
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_status(ctx)
assert "pool_seq" not in body["listeners"][0]
def test_multi_pool_has_pool_seq(self):
"""Multi-pool listener: pool_seq appears in response."""
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_seq=[["clean"], ["mitm"]], pool_name="clean",
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_status(ctx)
assert body["listeners"][0]["pool_seq"] == [["clean"], ["mitm"]]
assert body["listeners"][0]["pool_hops"] == 2
def test_multi_pool_in_config(self):
"""Multi-pool listener: pool_seq appears in /config response."""
config = Config(
listeners=[
ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_seq=[["clean"], ["mitm"]], pool_name="clean",
),
],
)
ctx = _make_ctx(config=config)
_, body = _handle_config(ctx)
assert body["listeners"][0]["pool_seq"] == [["clean"], ["mitm"]]
class TestHandleMetrics:
"""Test GET /metrics handler."""
@@ -318,7 +439,7 @@ class TestHandleConfig:
proxy_pool=pp,
listeners=[ListenerConfig(
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_hops=1,
pool_seq=["default"],
)],
)
ctx = _make_ctx(config=config)
@@ -345,7 +466,7 @@ class TestHandleConfig:
listeners=[ListenerConfig(
listen_host="0.0.0.0", listen_port=1080,
chain=[ChainHop("socks5", "127.0.0.1", 9050)],
pool_hops=2, pool_name="clean",
pool_seq=[["clean"], ["clean"]], pool_name="clean",
)],
)
ctx = _make_ctx(config=config)
+253
View File
@@ -10,6 +10,7 @@ from s5p.config import (
parse_api_proxies,
parse_proxy_url,
)
from s5p.server import _bypass_match
class TestParseProxyUrl:
@@ -407,6 +408,127 @@ class TestListenerConfig:
assert c.listeners[0].chain == []
class TestPoolSeq:
"""Test per-hop pool references (pool:name syntax)."""
def test_bare_pool_uses_default_name(self, tmp_path):
"""Bare `pool` + `pool: clean` -> pool_seq=[["clean"]]."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" pool: clean\n"
" chain:\n"
" - pool\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["clean"]]
def test_bare_pool_no_pool_name(self, tmp_path):
"""Bare `pool` with no `pool:` key -> pool_seq=[["default"]]."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" chain:\n"
" - pool\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["default"]]
def test_pool_colon_name(self, tmp_path):
"""`pool:clean, pool:mitm` -> pool_seq=[["clean"], ["mitm"]]."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" chain:\n"
" - pool:clean\n"
" - pool:mitm\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["clean"], ["mitm"]]
def test_mixed_bare_and_named(self, tmp_path):
"""Bare `pool` + `pool:mitm` with `pool: clean` -> [["clean"], ["mitm"]]."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" pool: clean\n"
" chain:\n"
" - pool\n"
" - pool:mitm\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["clean"], ["mitm"]]
def test_pool_colon_case_insensitive_prefix(self, tmp_path):
"""`Pool:MyPool` -> pool_seq=[["MyPool"]] (prefix case-insensitive)."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" chain:\n"
" - Pool:MyPool\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["MyPool"]]
def test_pool_colon_empty_is_bare(self, tmp_path):
"""`pool:` (empty name) -> treated as bare pool."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" pool: clean\n"
" chain:\n"
" - pool:\n"
)
c = load_config(cfg_file)
assert c.listeners[0].pool_seq == [["clean"]]
def test_backward_compat_pool_hops_property(self):
"""pool_hops property returns len(pool_seq)."""
lc = ListenerConfig(pool_seq=[["clean"], ["mitm"]])
assert lc.pool_hops == 2
lc2 = ListenerConfig()
assert lc2.pool_hops == 0
def test_legacy_auto_append(self, tmp_path):
"""Singular `proxy_pool:` -> pool_seq=[["default"]]."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listen: 0.0.0.0:1080\n"
"chain:\n"
" - socks5://127.0.0.1:9050\n"
"proxy_pool:\n"
" sources:\n"
" - url: http://api:8081/proxies\n"
)
c = load_config(cfg_file)
lc = c.listeners[0]
assert lc.pool_seq == [["default"]]
assert lc.pool_hops == 1
def test_list_candidates(self, tmp_path):
"""List in chain -> multi-candidate hop."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" chain:\n"
" - socks5://tor:9050\n"
" - [pool:clean, pool:mitm]\n"
" - [pool:clean, pool:mitm]\n"
)
c = load_config(cfg_file)
lc = c.listeners[0]
assert len(lc.chain) == 1
assert lc.pool_hops == 2
assert lc.pool_seq == [["clean", "mitm"], ["clean", "mitm"]]
class TestListenerBackwardCompat:
"""Test backward-compatible single listener from old format."""
@@ -469,3 +591,134 @@ class TestListenerPoolCompat:
lc = c.listeners[0]
# explicit listeners: no auto pool_hops
assert lc.pool_hops == 0
class TestAuthConfig:
"""Test auth field in listener config."""
def test_auth_from_yaml(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" auth:\n"
" alice: s3cret\n"
" bob: hunter2\n"
)
c = load_config(cfg_file)
assert c.listeners[0].auth == {"alice": "s3cret", "bob": "hunter2"}
def test_auth_empty_default(self):
lc = ListenerConfig()
assert lc.auth == {}
def test_auth_absent_from_yaml(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
)
c = load_config(cfg_file)
assert c.listeners[0].auth == {}
def test_auth_numeric_password(self, tmp_path):
"""YAML parses `admin: 12345` as int; must be coerced to str."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" auth:\n"
" admin: 12345\n"
)
c = load_config(cfg_file)
assert c.listeners[0].auth == {"admin": "12345"}
def test_auth_mixed_listeners(self, tmp_path):
"""One listener with auth, one without."""
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" auth:\n"
" alice: pass\n"
" - listen: 1081\n"
)
c = load_config(cfg_file)
assert c.listeners[0].auth == {"alice": "pass"}
assert c.listeners[1].auth == {}
class TestBypassConfig:
"""Test bypass rules in listener config."""
def test_bypass_from_yaml(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" bypass:\n"
" - 127.0.0.0/8\n"
" - 192.168.0.0/16\n"
" - localhost\n"
" - .local\n"
" chain:\n"
" - socks5://127.0.0.1:9050\n"
)
c = load_config(cfg_file)
lc = c.listeners[0]
assert lc.bypass == ["127.0.0.0/8", "192.168.0.0/16", "localhost", ".local"]
def test_bypass_empty_default(self):
lc = ListenerConfig()
assert lc.bypass == []
def test_bypass_absent_from_yaml(self, tmp_path):
cfg_file = tmp_path / "test.yaml"
cfg_file.write_text(
"listeners:\n"
" - listen: 1080\n"
" chain:\n"
" - socks5://127.0.0.1:9050\n"
)
c = load_config(cfg_file)
assert c.listeners[0].bypass == []
class TestBypassMatch:
"""Test _bypass_match function."""
def test_cidr_ipv4(self):
assert _bypass_match(["10.0.0.0/8"], "10.1.2.3") is True
assert _bypass_match(["10.0.0.0/8"], "11.0.0.1") is False
def test_cidr_ipv6(self):
assert _bypass_match(["fc00::/7"], "fd00::1") is True
assert _bypass_match(["fc00::/7"], "2001:db8::1") is False
def test_exact_ip(self):
assert _bypass_match(["127.0.0.1"], "127.0.0.1") is True
assert _bypass_match(["127.0.0.1"], "127.0.0.2") is False
def test_exact_hostname(self):
assert _bypass_match(["localhost"], "localhost") is True
assert _bypass_match(["localhost"], "otherhost") is False
def test_domain_suffix(self):
assert _bypass_match([".local"], "myhost.local") is True
assert _bypass_match([".local"], "local") is True
assert _bypass_match([".local"], "notlocal") is False
assert _bypass_match([".example.com"], "api.example.com") is True
assert _bypass_match([".example.com"], "example.com") is True
def test_multiple_rules(self):
rules = ["10.0.0.0/8", "192.168.0.0/16", ".local"]
assert _bypass_match(rules, "10.1.2.3") is True
assert _bypass_match(rules, "192.168.1.1") is True
assert _bypass_match(rules, "host.local") is True
assert _bypass_match(rules, "8.8.8.8") is False
def test_empty_rules(self):
assert _bypass_match([], "anything") is False
def test_hostname_not_matched_by_cidr(self):
assert _bypass_match(["10.0.0.0/8"], "example.com") is False
+578
View File
@@ -0,0 +1,578 @@
"""End-to-end integration tests with mock SOCKS5 proxies."""
from __future__ import annotations
import asyncio
import struct
from s5p.config import ChainHop, ListenerConfig
from s5p.proto import encode_address
from s5p.server import _handle_client
from .conftest import free_port, start_echo_server, start_mock_socks5
# -- helpers -----------------------------------------------------------------
async def _socks5_connect(
host: str, port: int, target_host: str, target_port: int,
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
"""Connect as a SOCKS5 client, perform greeting + CONNECT."""
reader, writer = await asyncio.open_connection(host, port)
# greeting: version 5, 1 method (no-auth)
writer.write(b"\x05\x01\x00")
await writer.drain()
resp = await reader.readexactly(2)
assert resp == b"\x05\x00", f"greeting failed: {resp!r}"
# connect request
atyp, addr_bytes = encode_address(target_host)
writer.write(
struct.pack("!BBB", 0x05, 0x01, 0x00)
+ bytes([atyp])
+ addr_bytes
+ struct.pack("!H", target_port)
)
await writer.drain()
# read reply
rep_header = await reader.readexactly(3)
atyp_resp = (await reader.readexactly(1))[0]
if atyp_resp == 0x01:
await reader.readexactly(4)
elif atyp_resp == 0x03:
length = (await reader.readexactly(1))[0]
await reader.readexactly(length)
elif atyp_resp == 0x04:
await reader.readexactly(16)
await reader.readexactly(2) # port
if rep_header[1] != 0x00:
writer.close()
await writer.wait_closed()
raise ConnectionError(f"SOCKS5 reply={rep_header[1]:#x}")
return reader, writer
async def _socks5_connect_auth(
host: str, port: int, target_host: str, target_port: int,
username: str, password: str,
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
"""Connect as a SOCKS5 client with username/password auth (RFC 1929)."""
reader, writer = await asyncio.open_connection(host, port)
# greeting: version 5, 1 method (user/pass)
writer.write(b"\x05\x01\x02")
await writer.drain()
resp = await reader.readexactly(2)
assert resp == b"\x05\x02", f"greeting failed: {resp!r}"
# subnegotiation
uname = username.encode("utf-8")
passwd = password.encode("utf-8")
writer.write(
b"\x01"
+ bytes([len(uname)]) + uname
+ bytes([len(passwd)]) + passwd
)
await writer.drain()
auth_resp = await reader.readexactly(2)
if auth_resp[1] != 0x00:
writer.close()
await writer.wait_closed()
raise ConnectionError(f"auth failed: status={auth_resp[1]:#x}")
# connect request
atyp, addr_bytes = encode_address(target_host)
writer.write(
struct.pack("!BBB", 0x05, 0x01, 0x00)
+ bytes([atyp])
+ addr_bytes
+ struct.pack("!H", target_port)
)
await writer.drain()
# read reply
rep_header = await reader.readexactly(3)
atyp_resp = (await reader.readexactly(1))[0]
if atyp_resp == 0x01:
await reader.readexactly(4)
elif atyp_resp == 0x03:
length = (await reader.readexactly(1))[0]
await reader.readexactly(length)
elif atyp_resp == 0x04:
await reader.readexactly(16)
await reader.readexactly(2) # port
if rep_header[1] != 0x00:
writer.close()
await writer.wait_closed()
raise ConnectionError(f"SOCKS5 reply={rep_header[1]:#x}")
return reader, writer
async def _close_server(srv: asyncio.Server) -> None:
"""Close a server and wait."""
srv.close()
await srv.wait_closed()
# -- tests -------------------------------------------------------------------
class TestDirectNoChain:
"""Client -> s5p -> echo (empty chain)."""
def test_echo(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
listener = ListenerConfig(listen_host="127.0.0.1", listen_port=free_port())
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect(
listener.listen_host, listener.listen_port, echo_host, echo_port,
)
writer.write(b"hello direct")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello direct"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestSingleHop:
"""Client -> s5p -> mock socks5 -> echo."""
def test_echo_through_one_hop(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
mock_host, mock_port, mock_srv = await start_mock_socks5()
servers.append(mock_srv)
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
chain=[ChainHop(proto="socks5", host=mock_host, port=mock_port)],
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect(
listener.listen_host, listener.listen_port, echo_host, echo_port,
)
writer.write(b"hello one hop")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello one hop"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestTwoHops:
"""Client -> s5p -> mock1 -> mock2 -> echo."""
def test_echo_through_two_hops(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
m1_host, m1_port, m1_srv = await start_mock_socks5()
servers.append(m1_srv)
m2_host, m2_port, m2_srv = await start_mock_socks5()
servers.append(m2_srv)
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
chain=[
ChainHop(proto="socks5", host=m1_host, port=m1_port),
ChainHop(proto="socks5", host=m2_host, port=m2_port),
],
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect(
listener.listen_host, listener.listen_port, echo_host, echo_port,
)
writer.write(b"hello two hops")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello two hops"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestConnectionRefused:
"""Dead hop returns SOCKS5 error to client."""
def test_refused(self):
async def _run():
servers = []
try:
# use a port with nothing listening
dead_port = free_port()
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
chain=[ChainHop(proto="socks5", host="127.0.0.1", port=dead_port)],
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=3.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await asyncio.open_connection(
listener.listen_host, listener.listen_port,
)
# greeting
writer.write(b"\x05\x01\x00")
await writer.drain()
resp = await reader.readexactly(2)
assert resp == b"\x05\x00"
# connect to a dummy target
atyp, addr_bytes = encode_address("127.0.0.1")
writer.write(
struct.pack("!BBB", 0x05, 0x01, 0x00)
+ bytes([atyp])
+ addr_bytes
+ struct.pack("!H", 9999)
)
await writer.drain()
# should get error reply (non-zero rep field)
rep = await asyncio.wait_for(reader.read(4096), timeout=5.0)
assert len(rep) >= 3
assert rep[1] != 0x00, "expected non-zero SOCKS5 reply code"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestBypassDirectConnect:
"""Target matches bypass rule -> chain skipped, direct connect to echo."""
def test_bypass_skips_chain(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
# dead hop -- would fail if bypass didn't skip it
dead_port = free_port()
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
chain=[ChainHop(proto="socks5", host="127.0.0.1", port=dead_port)],
bypass=["127.0.0.0/8"],
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect(
listener.listen_host, listener.listen_port, echo_host, echo_port,
)
writer.write(b"hello bypass")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello bypass"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestOnionChainOnly:
"""Onion target uses static chain only, pool hops skipped."""
def test_onion_skips_pool(self):
async def _run():
servers = []
try:
# mock socks5 acts as the "Tor" hop
mock_host, mock_port, mock_srv = await start_mock_socks5()
servers.append(mock_srv)
# fake pool that would add a dead hop if called
from unittest.mock import AsyncMock, MagicMock
dead_port = free_port()
fake_pool = MagicMock()
fake_pool.alive_count = 1
fake_pool.get = AsyncMock(
return_value=ChainHop(
proto="socks5", host="127.0.0.1", port=dead_port,
),
)
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
chain=[ChainHop(proto="socks5", host=mock_host, port=mock_port)],
pool_seq=[["default"]],
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(
r, w, listener, timeout=5.0, retries=1,
pool_seq=[[fake_pool]],
),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
# connect with .onion target -- mock socks5 will fail to
# resolve it, but the key assertion is pool.get NOT called
reader, writer = await asyncio.open_connection(
listener.listen_host, listener.listen_port,
)
writer.write(b"\x05\x01\x00")
await writer.drain()
await reader.readexactly(2)
atyp, addr_bytes = encode_address("fake.onion")
writer.write(
struct.pack("!BBB", 0x05, 0x01, 0x00)
+ bytes([atyp])
+ addr_bytes
+ struct.pack("!H", 80)
)
await writer.drain()
await asyncio.wait_for(reader.read(4096), timeout=3.0)
writer.close()
await writer.wait_closed()
# pool.get must NOT have been called (onion skips pool)
fake_pool.get.assert_not_called()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestAuthSuccess:
"""Authenticate with valid credentials, relay echo data."""
def test_auth_echo(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
auth={"alice": "s3cret"},
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect_auth(
listener.listen_host, listener.listen_port,
echo_host, echo_port, "alice", "s3cret",
)
writer.write(b"hello auth")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello auth"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestAuthFailure:
"""Wrong password returns auth failure response."""
def test_wrong_password(self):
async def _run():
servers = []
try:
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
auth={"alice": "s3cret"},
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await asyncio.open_connection(
listener.listen_host, listener.listen_port,
)
# greeting with auth method
writer.write(b"\x05\x01\x02")
await writer.drain()
resp = await reader.readexactly(2)
assert resp == b"\x05\x02"
# subnegotiation with wrong password
uname = b"alice"
passwd = b"wrong"
writer.write(
b"\x01"
+ bytes([len(uname)]) + uname
+ bytes([len(passwd)]) + passwd
)
await writer.drain()
auth_resp = await reader.readexactly(2)
assert auth_resp == b"\x01\x01", f"expected auth failure, got {auth_resp!r}"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestAuthMethodNotOffered:
"""Client offers only no-auth when auth is required -> 0xFF rejection."""
def test_no_auth_method_rejected(self):
async def _run():
servers = []
try:
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
auth={"alice": "s3cret"},
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await asyncio.open_connection(
listener.listen_host, listener.listen_port,
)
# greeting with only no-auth method (0x00)
writer.write(b"\x05\x01\x00")
await writer.drain()
resp = await reader.readexactly(2)
assert resp == b"\x05\xff", f"expected method rejection, got {resp!r}"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())
class TestNoAuthListenerUnchanged:
"""No auth configured -- 0x00 still works as before."""
def test_no_auth_still_works(self):
async def _run():
servers = []
try:
echo_host, echo_port, echo_srv = await start_echo_server()
servers.append(echo_srv)
listener = ListenerConfig(
listen_host="127.0.0.1",
listen_port=free_port(),
)
s5p_srv = await asyncio.start_server(
lambda r, w: _handle_client(r, w, listener, timeout=5.0, retries=1),
listener.listen_host, listener.listen_port,
)
servers.append(s5p_srv)
await s5p_srv.start_serving()
reader, writer = await _socks5_connect(
listener.listen_host, listener.listen_port, echo_host, echo_port,
)
writer.write(b"hello no auth")
await writer.drain()
data = await asyncio.wait_for(reader.read(4096), timeout=2.0)
assert data == b"hello no auth"
writer.close()
await writer.wait_closed()
finally:
for s in servers:
await _close_server(s)
asyncio.run(_run())