docs: clarify tor_nodes overrides listener first hop

The first hop in each listener's chain is replaced at connection time
by round-robin selection from tor_nodes. Added comments in both the
example config and the server code to make this precedence explicit.

.containerignore

@@ -0,0 +1,9 @@
+.venv/
+.git/
+tests/
+docs/
+*.prof
+*.egg-info/
+__pycache__/
+.gitea/
+.pytest_cache/

.gitea/workflows/ci.yaml

@@ -0,0 +1,48 @@
+name: ci
+
+on:
+  push:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: linux
+    container: python:3.13-alpine
+    steps:
+      - run: apk add --no-cache git
+      - run: |
+          git clone --depth 1 \
+            -c "http.extraHeader=Authorization: token ${{ github.token }}" \
+            "${{ github.server_url }}/${{ github.repository }}.git" .
+      - run: pip install --no-cache-dir -r requirements.txt ruff pytest
+      - run: ruff check src/ tests/
+      - run: PYTHONPATH=src pytest tests/ -v
+
+  secrets:
+    runs-on: linux
+    container: ghcr.io/gitleaks/gitleaks:latest
+    steps:
+      - run: |
+          git clone \
+            -c "http.extraHeader=Authorization: token ${{ github.token }}" \
+            "${{ github.server_url }}/${{ github.repository }}.git" .
+      - run: gitleaks detect --source . -v
+
+  build:
+    needs: [test, secrets]
+    runs-on: linux
+    container: quay.io/podman/stable
+    env:
+      CONTAINER_HOST: unix:///var/run/docker.sock
+    steps:
+      - run: dnf install -y git
+      - run: |
+          git clone --depth 1 \
+            -c "http.extraHeader=Authorization: token ${{ github.token }}" \
+            "${{ github.server_url }}/${{ github.repository }}.git" .
+      - run: echo "$HARBOR_PASS" | podman --remote login -u "$HARBOR_USER" --password-stdin harbor.mymx.me
+        env:
+          HARBOR_USER: ${{ secrets.HARBOR_USER }}
+          HARBOR_PASS: ${{ secrets.HARBOR_PASS }}
+      - run: podman --remote build -t harbor.mymx.me/s5p/s5p:latest -f Containerfile .
+      - run: podman --remote push harbor.mymx.me/s5p/s5p:latest

.gitleaks.toml

@@ -0,0 +1,5 @@
+[allowlist]
+  paths = [
+    '''tests/''',
+    '''docs/''',
+  ]

Containerfile

@@ -1,13 +1,17 @@
 FROM python:3.13-alpine
 
-RUN pip install --no-cache-dir pyyaml>=6.0
-
 WORKDIR /app
 
+COPY requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
 ENV PYTHONUNBUFFERED=1 \
     PYTHONDONTWRITEBYTECODE=1 \
     PYTHONPATH=/app/src
 
+COPY src/ /app/src/
+
 EXPOSE 1080
 STOPSIGNAL SIGTERM
 

Makefile

@@ -1,10 +1,17 @@
 APP_NAME := s5p
 
-.PHONY: install test lint clean build up down logs
+.PHONY: install install-service test lint clean build up down logs
 
 install:
 	pip install -e .
 
+install-service:
+	sudo mkdir -p /etc/s5p
+	sudo cp config/s5p.service /etc/systemd/system/s5p.service
+	sudo systemctl daemon-reload
+	@echo "Unit installed. Configure /etc/s5p/s5p.yaml, then:"
+	@echo "  sudo systemctl enable --now s5p"
+
 test:
 	pytest tests/ -v
 

PROJECT.md

@@ -19,15 +19,16 @@ Client -------> s5p -------> Hop 1 -------> Hop 2 -------> Target
         SOCKS5       proto1         proto2         protoN
 ```
 
-- **server.py** -- asyncio SOCKS5 server, bidirectional relay, signal handling
+- **server.py** -- asyncio SOCKS5 server, bidirectional relay, signal handling, multi-pool orchestration
 - **proto.py** -- protocol handshakes (SOCKS5, SOCKS4/4a, HTTP CONNECT), chain builder
-- **config.py** -- YAML config loading, proxy URL parsing, API response parsing, pool config
+- **config.py** -- YAML config loading, proxy URL parsing, API response parsing, pool/listener config
-- **pool.py** -- managed proxy pool (multi-source, health-tested, persistent)
+- **pool.py** -- named proxy pool (multi-source, health-tested, persistent, MITM filtering)
 - **http.py** -- minimal async HTTP/1.1 client (GET/POST JSON, no external deps)
 - **connpool.py** -- pre-warmed TCP connection pool to first chain hop
-- **api.py** -- built-in HTTP control API (runtime metrics, pool state, config reload)
+- **api.py** -- built-in HTTP control API (runtime metrics, multi-pool state, config reload)
+- **tor.py** -- Tor control port integration (NEWNYM signaling, periodic circuit rotation)
 - **cli.py** -- argparse CLI, logging setup, cProfile support
-- **metrics.py** -- connection counters and human-readable summary (lock-free, asyncio-only)
+- **metrics.py** -- connection counters, per-listener latency, rate tracking (lock-free, asyncio-only)
 
 ## Deployment
 
@@ -36,9 +37,13 @@ Client -------> s5p -------> Hop 1 -------> Hop 2 -------> Target
 | Local venv | `pip install -e .` then `s5p -c config/s5p.yaml` |
 | Container | `make build && make up` (Alpine, ~59MB) |
 
-Container mounts `./src` and `./config/s5p.yaml` read-only, plus
+Production images bake source into the image via `COPY src/ /app/src/`.
+Config and data are mounted at runtime: `./config/s5p.yaml` (ro) and
 `~/.cache/s5p` as `/data` for pool state and profile output.
-No application code is baked into the image.
+The compose.yaml volume mount overrides source for local dev.
+
+CI pushes `harbor.mymx.me/s5p/s5p:latest` on every push to `main`
+(lint + tests must pass first).
 
 ## Dependencies
 
@@ -68,3 +73,9 @@ All other functionality uses Python stdlib (`asyncio`, `socket`, `struct`).
 - **Async HTTP** -- native asyncio HTTP client replaces blocking urllib, parallel fetches
 - **First-hop pool** -- pre-warmed TCP connections to chain[0], stale-evicted, auto-refilled
 - **Control API** -- built-in asyncio HTTP server, no Flask/external deps, disabled by default
+- **Tor integration** -- control port NEWNYM signaling, periodic circuit rotation
+- **Multi-Tor** -- round-robin traffic across multiple Tor nodes (`tor_nodes`)
+- **Multi-listener** -- per-port chain depth and pool assignment
+- **Named pools** -- independent proxy pools with per-listener binding (`proxy_pools:`)
+- **MITM filtering** -- `mitm: true/false` source filter, `?mitm=0/1` API query param
+- **Per-listener latency** -- independent latency tracking per listener in `/status`

README.md

@@ -11,14 +11,16 @@ through configurable chains of SOCKS4, SOCKS5, and HTTP CONNECT proxies.
 - Per-hop authentication (username/password)
 - DNS leak prevention (domain names forwarded to proxies, never resolved locally)
 - Tor integration (SOCKS5 hop + control port NEWNYM for circuit rotation)
-- Multi-listener: different ports with different chain depths (Tor-only, Tor+1, Tor+2)
+- Multi-Tor round-robin (`tor_nodes` distributes traffic across Tor instances)
-- Managed proxy pool: multiple sources (API + file), health-tested, weighted selection
+- Multi-listener: different ports with different chain depths and pool assignments
+- Named proxy pools: independent sources, health testing, and state per pool
+- MITM source filter (`mitm: true/false` adds `?mitm=0/1` to API requests)
 - Per-proxy failure backoff (60s cooldown), stale proxy expiry, chain pre-flight
 - Fast warm start (seconds on restart vs minutes on cold start)
 - Connection retry with proxy rotation (configurable attempts)
 - Dead proxy reporting to upstream API (optional `report_url`)
 - SIGHUP hot reload (timeout, retries, log_level, pool config)
-- Connection metrics with pool stats (logged periodically and on shutdown)
+- Connection metrics with per-listener latency and pool stats
 - Concurrent connection limit with backpressure (`max_connections`)
 - Async HTTP client for proxy source fetching (parallel, no threads)
 - First-hop TCP connection pool (pre-warmed, stale-evicted)
@@ -55,8 +57,13 @@ make logs    # podman-compose logs -f
 make down    # podman-compose down
 ```
 
-Source, config, and data are bind-mounted, not baked into the image.
+Production images bake source into the image. Config and data are mounted
-Pool state and profile output persist in `~/.cache/s5p/` (`/data` inside container).
+at runtime. Pool state and profile output persist in `~/.cache/s5p/`
+(`/data` inside container). The compose.yaml volume mount overrides
+source for local dev.
+
+CI (Gitea Actions) runs lint + tests on push to `main`, then builds and
+pushes `harbor.mymx.me/s5p/s5p:latest`.
 
 ## Configuration
 
@@ -73,39 +80,48 @@ max_connections: 256                  # concurrent connection limit
 pool_size: 8                          # pre-warmed connections to first hop
 api_listen: 127.0.0.1:1081           # control API (disabled by default)
 
-# Multi-listener: each port gets a different chain depth
+# Named proxy pools (each with independent sources and health testing)
+proxy_pools:
+  clean:
+    sources:
+      - url: http://10.200.1.250:8081/proxies/all
+        mitm: false                   # filter: ?mitm=0
+    refresh: 300
+    test_interval: 120
+    max_fails: 3
+  mitm:
+    sources:
+      - url: http://10.200.1.250:8081/proxies/all
+        mitm: true                    # filter: ?mitm=1
+    refresh: 300
+    test_interval: 120
+    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
     chain:
       - socks5://127.0.0.1:9050
-      - pool                           # Tor + 2 pool proxies
+      - pool                           # Tor + 2 clean proxies
       - pool
   - listen: 0.0.0.0:1081
     chain:
       - socks5://127.0.0.1:9050
-      - pool                           # Tor + 1 pool 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
 
-# Old single-listener format still works:
+# Singular proxy_pool: still works (becomes pool "default")
-# listen: 127.0.0.1:1080
-# chain:
-#   - socks5://127.0.0.1:9050
 
 tor:
   control_port: 9051                  # Tor control port (NEWNYM)
   password: ""                        # or cookie_file for auth
   newnym_interval: 0                  # periodic circuit rotation (0 = manual)
-
-proxy_pool:
-  sources:
-    - url: http://10.200.1.250:8081/proxies
-      proto: socks5
-    - file: /etc/s5p/proxies.txt     # one proxy URL per line
-  refresh: 300                        # re-fetch interval (seconds)
-  test_interval: 120                  # health test cycle (seconds)
-  max_fails: 3                        # evict after N consecutive failures
 ```
 
 `config/s5p.yaml` is gitignored; `config/example.yaml` is the tracked template.
@@ -134,13 +150,13 @@ Options:
 ## How Chaining Works
 
 ```
-:1080  Client -> s5p -> Tor -> [pool] -> [pool] -> Destination  (2 pool hops)
+:1080  Client -> s5p -> Tor -> [clean] -> [clean]       -> Dest  (2 clean hops)
-:1081  Client -> s5p -> Tor -> [pool] -> Destination           (1 pool hop)
+:1081  Client -> s5p -> Tor -> [clean|mitm] -> [clean|mitm] -> Dest  (random)
-:1082  Client -> s5p -> Tor -> Destination                     (0 pool hops)
+:1082  Client -> s5p -> Tor -> Dest                              (Tor only)
 ```
 
 s5p connects to Hop1 via TCP, negotiates the hop protocol (SOCKS5/4/HTTP),
-then over that tunnel negotiates with Hop2, and so on. If a proxy pool is
+then over that tunnel negotiates with Hop2, and so on. Each listener draws
-configured, alive proxies are appended per-connection (one per `pool` entry),
+from its assigned named pool -- alive proxies are appended per-connection
-weighted toward those with the most recent successful health test. Each hop
+(one per `pool` entry), weighted toward those with the most recent successful
-only sees its immediate neighbors.
+health test. Each hop only sees its immediate neighbors.

ROADMAP.md

@@ -1,6 +1,6 @@
 # s5p -- Roadmap
 
-## v0.1.0 (current)
+## v0.1.0
 
 - [x] SOCKS5 server (CONNECT command)
 - [x] Proxy chaining (SOCKS5, SOCKS4/4a, HTTP CONNECT)
@@ -23,12 +23,26 @@
 ## v0.2.0
 
 - [x] Built-in control API (runtime metrics, pool state, config reload)
-- [ ] SOCKS5 server authentication (username/password)
 - [x] Tor control port integration (circuit renewal via NEWNYM)
 - [x] Metrics (connections/sec, bytes relayed, hop latency)
+- [x] Multi-listener with per-port chain depth
+- [x] Per-listener latency tracking
+- [x] Dynamic health test concurrency (auto-scales to ~10% of pool)
+- [x] Multi-Tor round-robin (`tor_nodes`)
+- [x] Named proxy pools with per-listener assignment (`proxy_pools:`)
+- [x] MITM source filter (`mitm: true/false` on pool sources)
 
-## v0.3.0
+## v0.3.0 (current)
 
+- [x] SOCKS5 server authentication (username/password)
+- [x] Systemd service unit
+- [x] CLI test coverage
+- [x] Protocol test coverage (SOCKS5/4/HTTP handshakes)
+- [x] API documentation (full response schemas)
+- [x] Prometheus metrics endpoint (`/metrics` OpenMetrics format)
+- [x] Listener-level retry override
+- [x] Pool-level proxy protocol filter (`allowed_protos`)
+- [x] Connection pooling documentation
 - [ ] UDP ASSOCIATE support (SOCKS5 UDP relay)
 - [ ] BIND support
 - [ ] Chain randomization (random order, random subset)
@@ -36,6 +50,4 @@
 ## v1.0.0
 
 - [ ] Stable API and config format
-- [ ] Comprehensive test suite with mock proxies
-- [ ] Systemd service unit
 - [ ] Performance benchmarks

TASKS.md

@@ -49,7 +49,38 @@
 
 - [x] Multi-listener with configurable proxy chaining (per-port chain depth)
 - [x] Connection rate and chain latency metrics (rate/s, p50/p95/p99)
+- [x] Per-listener latency tracking
+- [x] Dynamic health test concurrency
+- [x] Multi-Tor round-robin via `tor_nodes` config
+- [x] Named proxy pools with per-listener assignment (`proxy_pools:`, `pool:`)
+  - [x] `mitm` source filter (`?mitm=0` / `?mitm=1` API query param)
+  - [x] Per-pool state files (`pool-{name}.json`)
+  - [x] Per-pool log prefixes (`pool[name]: ...`)
+  - [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)
+
+- [x] Gitea CI workflow (lint + test + Harbor image push)
+
+## v0.3.0 Stabilization
+- [x] Version bump to 0.3.0
+- [x] Systemd service unit (`config/s5p.service`, `make install-service`)
+- [x] CLI argument parsing tests (`tests/test_cli.py`)
+- [x] Protocol handshake tests (`tests/test_proto.py` -- SOCKS5/4/HTTP)
+- [x] API reference documentation (`docs/USAGE.md`)
+- [x] Prometheus `/metrics` endpoint (OpenMetrics format)
+
+## Quick Wins
+- [x] Listener-level retry override (`retries` per listener)
+- [x] Pool-level proxy protocol filter (`allowed_protos` on proxy pool)
+- [x] Document connection pooling (`pool_size`/`pool_max_idle` in CHEATSHEET.md)
 
 ## Next
-- [ ] Integration tests with mock proxy server
+- [ ] UDP ASSOCIATE support
-- [ ] SOCKS5 server-side authentication
+- [ ] BIND support
+- [ ] Chain randomization

TODO.md

@@ -4,21 +4,29 @@
 
 - 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)~~ (done)
+- ~~Listener-level retry override (different retry count per listener)~~ (done)
 
 ## Performance
 
 - Benchmark relay throughput vs direct connection
 - Tune buffer sizes for different workloads
-- Connection pooling for frequently-used chains
+- ~~Connection pooling for frequently-used chains~~ (done: `pool_size`/`pool_max_idle`)
 
 ## Security
 
-- Optional SOCKS5 server authentication
+- ~~Optional SOCKS5 server authentication~~ (done: fa36218)
 - Rate limiting per source IP
 - Access control lists
 
+## Observability
+
+- ~~Prometheus metrics endpoint (`/metrics` in OpenMetrics format)~~ (done)
+- Per-pool health test success rate tracking
+- Per-pool latency breakdown in `/status`
+
 ## Docs
 
 - Man page

config/example.yaml

@@ -28,11 +28,12 @@ chain:
 #     sources:
 #       - url: http://10.200.1.250:8081/proxies/all
 #         mitm: false                 # filter: mitm=0 query param
+#     allowed_protos: [socks5]        # only accept socks5 from sources
 #     state_file: /data/pool-clean.json
 #     refresh: 300
 #     test_interval: 120
-#     test_timeout: 8
+#     test_timeout: 12
-#     max_fails: 3
+#     max_fails: 5
 #   mitm:                             # MITM-capable proxies
 #     sources:
 #       - url: http://10.200.1.250:8081/proxies/all
@@ -40,8 +41,8 @@ chain:
 #     state_file: /data/pool-mitm.json
 #     refresh: 300
 #     test_interval: 120
-#     test_timeout: 8
+#     test_timeout: 12
-#     max_fails: 3
+#     max_fails: 5
 
 # Single proxy pool (legacy, still supported -- becomes pool "default"):
 # proxy_pool:
@@ -74,8 +75,11 @@ chain:
 #   newnym_interval: 0             # periodic NEWNYM (seconds, 0 = manual only)
 
 # Multi-Tor round-robin -- distribute traffic across multiple Tor nodes.
-# When present, the first hop in each listener's chain is replaced at
+# When present, the first hop in each listener's chain is REPLACED at
-# connection time by round-robin selection from this list.
+# connection time by round-robin selection from this list. The first hop
+# specified in each listener's chain acts as a fallback only; tor_nodes
+# takes precedence for both client traffic and pool health tests.
+# Connection pools are pre-warmed for every node listed here.
 # tor_nodes:
 #   - socks5://10.200.1.1:9050
 #   - socks5://10.200.1.254:9050
@@ -87,30 +91,50 @@ 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
+#       - socks5://127.0.0.1:9050        # first hop (overridden by tor_nodes)
-#       - pool                           # Tor + 2 clean pool proxies
+#       - [pool:clean, pool:mitm]        # random choice per connection
-#       - pool
+#       - [pool:clean, pool:mitm]        # independent random choice
 #
 #   - listen: 0.0.0.0:1081
 #     pool: clean
+#     retries: 5                         # override global retries for this listener
 #     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                           # bare "pool" = clean
-#       - pool
+#       - 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).

config/s5p.service

@@ -0,0 +1,15 @@
+[Unit]
+Description=s5p SOCKS5 proxy
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=/usr/local/bin/s5p -c /etc/s5p/s5p.yaml
+Restart=on-failure
+RestartSec=5
+KillSignal=SIGTERM
+TimeoutStopSec=10
+
+[Install]
+WantedBy=multi-user.target

docs/CHEATSHEET.md

@@ -21,6 +21,18 @@ s5p --tracemalloc                           # memory profile (top 10)
 s5p --tracemalloc 20                        # memory profile (top 20)
 ```
 
+## Systemd
+
+```
+make install-service                            # install unit + reload
+sudo systemctl enable --now s5p                 # enable + start
+sudo systemctl status s5p                       # check status
+sudo systemctl restart s5p                      # restart
+sudo systemctl stop s5p                         # stop
+journalctl -u s5p -f                            # follow logs
+journalctl -u s5p --since "5 min ago"           # recent logs
+```
+
 ## Container
 
 ```
@@ -30,7 +42,18 @@ make logs    # podman-compose logs -f
 make down    # podman-compose down
 ```
 
-Volumes: `./src` (ro), `./config/s5p.yaml` (ro), `~/.cache/s5p` → `/data` (pool state + profiles)
+Volumes: `./config/s5p.yaml` (ro), `~/.cache/s5p` → `/data` (pool state + profiles)
+Dev override: compose.yaml mounts `./src` (ro) over the baked-in source.
+
+## CI
+
+Gitea Actions runs on push to `main`:
+
+1. `ruff check` + `pytest` (test)
+2. `gitleaks detect` (secrets scan)
+3. Build + push `harbor.mymx.me/s5p/s5p:latest`
+
+Secrets: `HARBOR_USER` / `HARBOR_PASS` (configured in Gitea repo settings).
 
 ## Config
 
@@ -43,7 +66,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 +75,94 @@ 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).
+
+## Listener Retry Override (config)
+
+```yaml
+listeners:
+  - listen: 0.0.0.0:1080
+    retries: 5                               # override global retries
+    chain:
+      - socks5://127.0.0.1:9050
+      - pool
+  - listen: 0.0.0.0:1082
+    chain:
+      - socks5://127.0.0.1:9050             # 0 = use global default
+```
+
+Per-listener `retries` overrides the global `retries` setting. Set to 0 (or
+omit) to inherit the global value.
+
+## Pool Protocol Filter (config)
+
+```yaml
+proxy_pools:
+  socks_only:
+    allowed_protos: [socks5]                 # reject http proxies
+    sources:
+      - url: http://api:8081/proxies/all
+```
+
+When set, proxies not matching `allowed_protos` are silently dropped during
+merge. Useful when a source returns mixed protocols but the pool should
+only serve a specific type.
+
 ## Multi-Tor Round-Robin (config)
 
 ```yaml
@@ -82,6 +181,23 @@ pool_size: 8           # pre-warmed TCP conns to first hop (0 = off)
 pool_max_idle: 30      # evict idle pooled conns (seconds)
 ```
 
+## Connection Pool (config)
+
+```yaml
+pool_size: 8           # pre-warmed TCP connections per first hop (0 = off)
+pool_max_idle: 30      # evict idle connections after N seconds
+```
+
+Pre-warms TCP connections to the first hop in the chain. Only the raw TCP
+connection is pooled -- SOCKS/HTTP negotiation consumes it. One pool is
+created per unique first hop (shared across listeners). Requires at least
+one hop in `chain`.
+
+| Setting | Default | Notes |
+|---------|---------|-------|
+| `pool_size` | 0 (off) | Connections per first hop |
+| `pool_max_idle` | 30 | Idle eviction in seconds |
+
 ## Named Proxy Pools (config)
 
 ```yaml
@@ -161,7 +277,7 @@ http://user:pass@host:port
 s5p --api 127.0.0.1:1081 -c config/s5p.yaml    # enable API
 
 curl -s http://127.0.0.1:1081/status | jq .     # runtime status
-curl -s http://127.0.0.1:1081/metrics | jq .    # full metrics
+curl -s http://127.0.0.1:1081/metrics           # prometheus metrics
 curl -s http://127.0.0.1:1081/pool | jq .       # all proxies
 curl -s http://127.0.0.1:1081/pool/alive | jq . # alive only
 curl -s http://127.0.0.1:1081/config | jq .     # current config
@@ -199,27 +315,26 @@ python -m pstats ~/.cache/s5p/s5p.prof      # container profile output
 metrics: conn=1842 ok=1790 fail=52 retries=67 active=3 in=50.0M out=1.0G rate=4.72/s p50=198.3ms p95=890.1ms up=1h01m01s pool=42/65
 ```
 
-## Metrics JSON (`/metrics`)
+## Prometheus Metrics (`/metrics`)
 
 ```bash
-curl -s http://127.0.0.1:1081/metrics | jq .
+curl -s http://127.0.0.1:1081/metrics
 ```
 
-```json
+```
-{
+# TYPE s5p_connections counter
-  "connections": 1842,
+s5p_connections_total 1842
-  "success": 1790,
+# TYPE s5p_active_connections gauge
-  "rate": 4.72,
+s5p_active_connections 3
-  "latency": {"count": 1000, "min": 45.2, "max": 2841.7, "avg": 312.4, "p50": 198.3, "p95": 890.1, "p99": 1523.6},
+# TYPE s5p_pool_proxies_alive gauge
-  "listener_latency": {
+s5p_pool_proxies_alive{pool="clean"} 30
-    "0.0.0.0:1080": {"count": 500, "p50": 1800.2, "p95": 8200.1, "...": "..."},
+# TYPE s5p_chain_latency_seconds summary
-    "0.0.0.0:1081": {"count": 300, "p50": 1000.1, "p95": 3500.2, "...": "..."},
+s5p_chain_latency_seconds{quantile="0.5"} 0.198300
-    "0.0.0.0:1082": {"count": 200, "p50": 400.1, "p95": 1200.5, "...": "..."}
+s5p_chain_latency_seconds{quantile="0.95"} 0.890100
-  }
+# EOF
-}
 ```
 
-Per-listener latency also appears in `/status` under each listener entry.
+OpenMetrics format. Use `/status` for JSON equivalent.
 
 ## Troubleshooting
 

docs/INSTALL.md

@@ -38,8 +38,40 @@ make build                                # podman-compose build
 make up                                   # podman-compose up -d
 ```
 
-The Alpine-based image (~59MB) contains only Python and PyYAML.
+The Alpine-based image (~59MB) contains Python, PyYAML, and baked-in
-Application source and config are bind-mounted at runtime.
+source. Config is mounted at runtime. The compose.yaml volume mount
+overrides source for local dev.
+
+## Systemd Service
+
+Install s5p as a systemd service for automatic startup and restart.
+
+```bash
+# Install the binary
+cd ~/git/s5p
+source .venv/bin/activate
+pip install -e .
+
+# Copy config
+sudo mkdir -p /etc/s5p
+sudo cp config/example.yaml /etc/s5p/s5p.yaml
+sudo nano /etc/s5p/s5p.yaml                    # edit with your settings
+
+# Install the unit (copies service file + daemon-reload)
+make install-service
+
+# Enable and start
+sudo systemctl enable --now s5p
+
+# Check status
+sudo systemctl status s5p
+journalctl -u s5p -f
+```
+
+The service unit expects:
+- Binary at `/usr/local/bin/s5p`
+- Config at `/etc/s5p/s5p.yaml`
+- Restarts on failure with 5-second delay
 
 ## Install Tor (optional)
 

docs/USAGE.md

@@ -139,6 +139,26 @@ Each pool has independent health testing, state persistence, and source
 refresh cycles. The `mitm` source filter adds `?mitm=0` or `?mitm=1` to
 API requests.
 
+### Pool protocol filter
+
+Use `allowed_protos` to restrict a pool to specific proxy protocols.
+Proxies not matching the list are silently dropped during merge, regardless
+of source type (API or file).
+
+```yaml
+proxy_pools:
+  socks_only:
+    allowed_protos: [socks5]     # reject http/socks4 proxies
+    sources:
+      - url: http://api:8081/proxies/all
+  any_proto:
+    sources:
+      - url: http://api:8081/proxies/all   # no filter, accept all
+```
+
+Valid values: `socks5`, `socks4`, `http`. Visible in `/config` API response
+when set.
+
 ### Backward compatibility
 
 The singular `proxy_pool:` key still works -- it registers as pool `"default"`.
@@ -177,9 +197,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 +262,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
@@ -387,56 +543,344 @@ api_listen: 127.0.0.1:1081
 s5p --api 127.0.0.1:1081 -c config/s5p.yaml
 ```
 
-### Read endpoints
+Responses are `application/json` unless noted otherwise. Errors return
+`{"error": "message"}` with appropriate status code (400, 404, 405, 500).
 
-| Method | Path | Description |
+Settings that require a restart: `listen`, `chain`, `pool_size`, `pool_max_idle`, `api_listen`.
-|--------|------|-------------|
-| `GET` | `/status` | Combined summary: uptime, metrics, pool stats, chain |
-| `GET` | `/metrics` | Full metrics counters (connections, bytes, rate, latency) |
-| `GET` | `/pool` | All proxies with per-entry state |
-| `GET` | `/pool/alive` | Alive proxies only |
-| `GET` | `/config` | Current runtime config (sanitized) |
 
-### Write endpoints
+### API Reference
 
-| Method | Path | Description |
+#### `GET /status`
-|--------|------|-------------|
-| `POST` | `/reload` | Re-read config file (replaces SIGHUP) |
-| `POST` | `/pool/test` | Trigger immediate health test cycle |
-| `POST` | `/pool/refresh` | Trigger immediate source re-fetch |
 
-All responses are `application/json`. Errors return `{"error": "message"}` with
+Combined runtime summary: uptime, metrics, pool stats, listeners.
-appropriate status code (400, 404, 405, 500).
-
-### Examples
 
 ```bash
-# Runtime status
 curl -s http://127.0.0.1:1081/status | jq .
+```
 
-# Full metrics
+```json
-curl -s http://127.0.0.1:1081/metrics | jq .
+{
+  "uptime": 3661.2,
+  "connections": 1842,
+  "success": 1790,
+  "failed": 52,
+  "active": 3,
+  "bytes_in": 52428800,
+  "bytes_out": 1073741824,
+  "rate": 4.72,
+  "latency": {"count": 1000, "min": 45.2, "max": 2841.7, "avg": 312.4, "p50": 198.3, "p95": 890.1, "p99": 1523.6},
+  "pool": {"alive": 42, "total": 65},
+  "pools": {
+    "clean": {"alive": 30, "total": 45},
+    "mitm": {"alive": 12, "total": 20}
+  },
+  "tor_nodes": ["socks5://10.200.1.1:9050", "socks5://10.200.1.254:9050"],
+  "listeners": [
+    {
+      "listen": "0.0.0.0:1080",
+      "chain": ["socks5://10.200.1.13:9050"],
+      "pool_hops": 2,
+      "pool": "clean",
+      "auth": true,
+      "latency": {"count": 500, "p50": 1800.2, "p95": 8200.1, "p99": 10500.3, "...": "..."}
+    }
+  ]
+}
+```
 
-# Pool state (all proxies)
+| Field | Type | Description |
+|-------|------|-------------|
+| `uptime` | float | Seconds since server start |
+| `connections` | int | Total incoming connections |
+| `success` | int | Successfully relayed |
+| `failed` | int | All retries exhausted |
+| `active` | int | Currently relaying |
+| `bytes_in` | int | Bytes client -> remote |
+| `bytes_out` | int | Bytes remote -> client |
+| `rate` | float | Connections/sec (rolling window) |
+| `latency` | object/null | Aggregate chain setup latency (ms), null if no samples |
+| `pool` | object | Aggregate pool counts (present when pool active) |
+| `pools` | object | Per-pool counts (present when multiple pools) |
+| `tor_nodes` | array | Tor node URLs (present when configured) |
+| `listeners` | array | Per-listener state with chain, pool, latency |
+| `listeners[].auth` | bool | Present and `true` when auth is enabled |
+
+#### `GET /metrics`
+
+Prometheus/OpenMetrics exposition format. Content-Type:
+`application/openmetrics-text; version=1.0.0; charset=utf-8`.
+
+```bash
+curl -s http://127.0.0.1:1081/metrics
+```
+
+```
+# HELP s5p_connections Total connection attempts.
+# TYPE s5p_connections counter
+s5p_connections_total 1842
+# HELP s5p_connections_success Connections successfully relayed.
+# TYPE s5p_connections_success counter
+s5p_connections_success_total 1790
+# HELP s5p_connections_failed Connection failures.
+# TYPE s5p_connections_failed counter
+s5p_connections_failed_total 52
+# HELP s5p_retries Connection retry attempts.
+# TYPE s5p_retries counter
+s5p_retries_total 67
+# HELP s5p_auth_failures SOCKS5 authentication failures.
+# TYPE s5p_auth_failures counter
+s5p_auth_failures_total 0
+# HELP s5p_bytes_in Bytes received from clients.
+# TYPE s5p_bytes_in counter
+s5p_bytes_in_total 52428800
+# HELP s5p_bytes_out Bytes sent to clients.
+# TYPE s5p_bytes_out counter
+s5p_bytes_out_total 1073741824
+# HELP s5p_active_connections Currently open connections.
+# TYPE s5p_active_connections gauge
+s5p_active_connections 3
+# HELP s5p_uptime_seconds Seconds since server start.
+# TYPE s5p_uptime_seconds gauge
+s5p_uptime_seconds 3661.2
+# HELP s5p_connection_rate Connections per second (rolling window).
+# TYPE s5p_connection_rate gauge
+s5p_connection_rate 4.72
+# HELP s5p_pool_proxies_alive Alive proxies in pool.
+# TYPE s5p_pool_proxies_alive gauge
+s5p_pool_proxies_alive{pool="clean"} 30
+s5p_pool_proxies_alive{pool="mitm"} 12
+# HELP s5p_pool_proxies_total Total proxies in pool.
+# TYPE s5p_pool_proxies_total gauge
+s5p_pool_proxies_total{pool="clean"} 45
+s5p_pool_proxies_total{pool="mitm"} 20
+# HELP s5p_chain_latency_seconds Chain build latency in seconds.
+# TYPE s5p_chain_latency_seconds summary
+s5p_chain_latency_seconds{quantile="0.5"} 0.198300
+s5p_chain_latency_seconds{quantile="0.95"} 0.890100
+s5p_chain_latency_seconds{quantile="0.99"} 1.523600
+s5p_chain_latency_seconds_count 1000
+s5p_chain_latency_seconds_sum 312.400000
+# EOF
+```
+
+**Metrics exposed:**
+
+| Metric | Type | Labels | Description |
+|--------|------|--------|-------------|
+| `s5p_connections` | counter | -- | Total connection attempts |
+| `s5p_connections_success` | counter | -- | Successfully relayed |
+| `s5p_connections_failed` | counter | -- | Connection failures |
+| `s5p_retries` | counter | -- | Retry attempts |
+| `s5p_auth_failures` | counter | -- | SOCKS5 auth failures |
+| `s5p_bytes_in` | counter | -- | Bytes received from clients |
+| `s5p_bytes_out` | counter | -- | Bytes sent to clients |
+| `s5p_active_connections` | gauge | -- | Currently open connections |
+| `s5p_uptime_seconds` | gauge | -- | Seconds since server start |
+| `s5p_connection_rate` | gauge | -- | Connections/sec (rolling window) |
+| `s5p_pool_proxies_alive` | gauge | `pool` | Alive proxies per pool |
+| `s5p_pool_proxies_total` | gauge | `pool` | Total proxies per pool |
+| `s5p_chain_latency_seconds` | summary | `quantile` | Chain build latency (p50/p95/p99) |
+| `s5p_listener_chain_latency_seconds` | summary | `listener`, `quantile` | Per-listener chain latency |
+
+**Prometheus scrape config:**
+
+```yaml
+scrape_configs:
+  - job_name: s5p
+    metrics_path: /metrics
+    static_configs:
+      - targets: ["127.0.0.1:1081"]
+```
+
+#### `GET /pool`
+
+All proxies with per-entry state.
+
+```bash
 curl -s http://127.0.0.1:1081/pool | jq .
+```
 
-# Alive proxies only
+```json
+{
+  "alive": 42,
+  "total": 65,
+  "pools": {
+    "clean": {"alive": 30, "total": 45},
+    "mitm": {"alive": 12, "total": 20}
+  },
+  "proxies": {
+    "socks5://1.2.3.4:1080": {
+      "alive": true,
+      "fails": 0,
+      "tests": 12,
+      "last_ok": 1708012345.6,
+      "last_test": 1708012345.6,
+      "last_seen": 1708012300.0,
+      "pool": "clean"
+    }
+  }
+}
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `alive` | int | Total alive proxies across all pools |
+| `total` | int | Total proxies across all pools |
+| `pools` | object | Per-pool counts (present when multiple pools) |
+| `proxies` | object | Keyed by proxy URL |
+| `proxies[].alive` | bool | Currently passing health tests |
+| `proxies[].fails` | int | Consecutive failures |
+| `proxies[].tests` | int | Total health tests performed |
+| `proxies[].last_ok` | float | Unix timestamp of last successful test |
+| `proxies[].last_test` | float | Unix timestamp of last test (pass or fail) |
+| `proxies[].last_seen` | float | Unix timestamp of last source refresh that included this proxy |
+| `proxies[].pool` | string | Pool name (present when multiple pools) |
+
+#### `GET /pool/alive`
+
+Same schema as `/pool`, filtered to alive proxies only.
+
+```bash
 curl -s http://127.0.0.1:1081/pool/alive | jq '.proxies | length'
+```
 
-# Current config
+#### `GET /config`
+
+Current runtime config (sanitized -- passwords are never exposed).
+
+```bash
 curl -s http://127.0.0.1:1081/config | jq .
+```
 
-# Reload config (like SIGHUP)
+```json
+{
+  "timeout": 10,
+  "retries": 3,
+  "log_level": "info",
+  "max_connections": 256,
+  "pool_size": 0,
+  "listeners": [
+    {
+      "listen": "0.0.0.0:1080",
+      "chain": ["socks5://10.200.1.13:9050"],
+      "pool_hops": 2,
+      "pool": "clean",
+      "auth_users": 2
+    }
+  ],
+  "tor_nodes": ["socks5://10.200.1.1:9050"],
+  "proxy_pools": {
+    "clean": {
+      "sources": [{"url": "http://10.200.1.250:8081/proxies/all", "mitm": false}],
+      "refresh": 300,
+      "test_interval": 120,
+      "max_fails": 3
+    }
+  }
+}
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `timeout` | float | Per-hop connection timeout (seconds) |
+| `retries` | int | Max connection attempts per request |
+| `log_level` | string | Current log level |
+| `max_connections` | int | Concurrent connection cap |
+| `pool_size` | int | Pre-warmed TCP connections to first hop |
+| `listeners` | array | Listener configs |
+| `listeners[].auth_users` | int | Number of auth users (present when auth enabled) |
+| `tor_nodes` | array | Tor node URLs (present when configured) |
+| `proxy_pools` | object | Pool configs (present when pools configured) |
+
+#### `GET /tor`
+
+Tor controller status.
+
+```bash
+curl -s http://127.0.0.1:1081/tor | jq .
+```
+
+```json
+{"enabled": true, "connected": true, "last_newnym": 45.2, "newnym_interval": 60}
+```
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `enabled` | bool | Whether Tor control is configured |
+| `connected` | bool | Whether connected to Tor control port |
+| `last_newnym` | float/null | Seconds since last NEWNYM signal |
+| `newnym_interval` | int | Auto-rotation interval (0 = manual) |
+
+Returns `{"enabled": false}` when Tor control is not configured.
+
+#### `POST /reload`
+
+Re-read config file (equivalent to SIGHUP).
+
+```bash
 curl -s -X POST http://127.0.0.1:1081/reload | jq .
+```
 
-# Trigger health tests now
+```json
+{"ok": true}
+```
+
+Returns `{"error": "..."}` (500) on failure.
+
+#### `POST /pool/test`
+
+Trigger immediate health test cycle for all pools.
+
+```bash
 curl -s -X POST http://127.0.0.1:1081/pool/test | jq .
+```
 
-# Re-fetch proxy sources now
+```json
+{"ok": true}
+```
+
+Returns `{"error": "no proxy pool configured"}` (400) when no pool is active.
+
+#### `POST /pool/refresh`
+
+Trigger immediate source re-fetch for all pools.
+
+```bash
 curl -s -X POST http://127.0.0.1:1081/pool/refresh | jq .
 ```
 
-Settings that require a restart: `listen`, `chain`, `pool_size`, `pool_max_idle`, `api_listen`.
+```json
+{"ok": true}
+```
+
+Returns `{"error": "no proxy pool configured"}` (400) when no pool is active.
+
+#### `POST /tor/newnym`
+
+Request new Tor circuit (NEWNYM signal).
+
+```bash
+curl -s -X POST http://127.0.0.1:1081/tor/newnym | jq .
+```
+
+```json
+{"ok": true}
+```
+
+Returns `{"ok": false, "reason": "rate-limited or not connected"}` when the
+signal cannot be sent. Returns `{"error": "tor control not configured"}` (400)
+when Tor control is not configured.
+
+#### Error responses
+
+All endpoints return errors as JSON with appropriate HTTP status codes:
+
+| Status | Meaning | Example |
+|--------|---------|---------|
+| 400 | Bad request | `{"error": "no proxy pool configured"}` |
+| 404 | Unknown path | `{"error": "not found"}` |
+| 405 | Wrong method | `{"error": "use GET for /status"}` |
+| 500 | Server error | `{"error": "reload not available"}` |
 
 ## Tor Control Port
 
@@ -515,6 +959,31 @@ retries: 5   # try up to 5 different proxies per connection
 s5p -r 5 -C socks5://127.0.0.1:9050 -S http://api:8081/proxies
 ```
 
+### Per-listener retry override
+
+Each listener can override the global `retries` setting. Set `retries` on
+a listener to use a different retry count for that port. A value of 0 (or
+omitting the key) inherits the global setting.
+
+```yaml
+retries: 3                      # global default
+
+listeners:
+  - listen: 0.0.0.0:1080
+    retries: 5                  # deep chain: more retries
+    chain:
+      - socks5://127.0.0.1:9050
+      - pool
+      - pool
+  - listen: 0.0.0.0:1082
+    chain:
+      - socks5://127.0.0.1:9050  # Tor only: uses global retries=3
+```
+
+The effective retry count for a listener is `listener.retries` if set,
+otherwise `config.retries`. Visible in `/config` and `/status` API responses
+when overridden.
+
 ## Hot Reload
 
 Send `SIGHUP` to reload the config file without restarting:
@@ -564,48 +1033,11 @@ metrics: conn=1842 ok=1790 fail=52 retries=67 active=3 in=50.0M out=1.0G rate=4.
 | `up` | Server uptime |
 | `pool` | Alive/total proxies (only when pool is active) |
 
-### `/metrics` JSON response
+### `/metrics` OpenMetrics endpoint
 
-`GET /metrics` returns all counters plus rate, latency percentiles, and
+`GET /metrics` returns all counters, gauges, pool stats, and latency summaries
-per-listener latency breakdowns:
+in OpenMetrics format (see [API Reference](#get-metrics) above). Use `/status`
-
+for the JSON equivalent with aggregate data.
-```json
-{
-  "connections": 1842,
-  "success": 1790,
-  "failed": 52,
-  "retries": 67,
-  "active": 3,
-  "bytes_in": 52428800,
-  "bytes_out": 1073741824,
-  "uptime": 3661.2,
-  "rate": 4.72,
-  "latency": {
-    "count": 1000,
-    "min": 45.2,
-    "max": 2841.7,
-    "avg": 312.4,
-    "p50": 198.3,
-    "p95": 890.1,
-    "p99": 1523.6
-  },
-  "listener_latency": {
-    "0.0.0.0:1080": {"count": 500, "min": 800.1, "max": 12400.3, "avg": 2100.5, "p50": 1800.2, "p95": 8200.1, "p99": 10500.3},
-    "0.0.0.0:1081": {"count": 300, "min": 400.5, "max": 5200.1, "avg": 1200.3, "p50": 1000.1, "p95": 3500.2, "p99": 4800.7},
-    "0.0.0.0:1082": {"count": 200, "min": 150.2, "max": 2000.1, "avg": 500.3, "p50": 400.1, "p95": 1200.5, "p99": 1800.2}
-  }
-}
-```
-
-| Field | Type | Description |
-|-------|------|-------------|
-| `rate` | float | Connections/sec (rolling window of last 256 events) |
-| `latency` | object/null | Aggregate chain setup latency in ms (null if no samples) |
-| `latency.count` | int | Number of samples in buffer (max 1000) |
-| `latency.p50` | float | Median latency (ms) |
-| `latency.p95` | float | 95th percentile (ms) |
-| `latency.p99` | float | 99th percentile (ms) |
-| `listener_latency` | object | Per-listener latency, keyed by `host:port` |
 
 ### Per-listener latency
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "s5p"
-version = "0.1.0"
+version = "0.3.0"
 description = "SOCKS5 proxy with Tor and proxy-chain support"
 requires-python = ">=3.11"
 dependencies = ["pyyaml>=6.0"]

requirements.txt

@@ -0,0 +1 @@
+pyyaml>=6.0

src/s5p/__init__.py

@@ -1,3 +1,3 @@
 """s5p -- SOCKS5 proxy with chain support."""
 
-__version__ = "0.1.0"
+__version__ = "0.3.0"

src/s5p/api.py

@@ -30,23 +30,46 @@ def _parse_request(data: bytes) -> tuple[str, str]:
     return parts[0].upper(), parts[1].split("?", 1)[0]
 
 
+def _http_response(
+    writer: asyncio.StreamWriter,
+    status: int,
+    payload: bytes,
+    content_type: str = "application/json",
+) -> None:
+    """Write an HTTP response and close."""
+    phrases = {200: "OK", 400: "Bad Request", 404: "Not Found",
+               405: "Method Not Allowed", 500: "Internal Server Error"}
+    header = (
+        f"HTTP/1.1 {status} {phrases.get(status, 'Error')}\r\n"
+        f"Content-Type: {content_type}\r\n"
+        f"Content-Length: {len(payload)}\r\n"
+        f"Connection: close\r\n"
+        f"\r\n"
+    )
+    writer.write(header.encode() + payload)
+
+
 def _json_response(
     writer: asyncio.StreamWriter,
     status: int,
     body: dict | list,
 ) -> None:
     """Write an HTTP response with JSON body and close."""
-    phrases = {200: "OK", 400: "Bad Request", 404: "Not Found",
-               405: "Method Not Allowed", 500: "Internal Server Error"}
     payload = json.dumps(body, separators=(",", ":")).encode()
-    header = (
+    _http_response(writer, status, payload, "application/json")
-        f"HTTP/1.1 {status} {phrases.get(status, 'Error')}\r\n"
+
-        f"Content-Type: application/json\r\n"
+
-        f"Content-Length: {len(payload)}\r\n"
+# -- helpers -----------------------------------------------------------------
-        f"Connection: close\r\n"
+
-        f"\r\n"
+
-    )
+def _multi_pool(lc) -> bool:
-    writer.write(header.encode() + payload)
+    """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 +111,9 @@ 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 {}),
+                **({"retries": lc.retries} if lc.retries else {}),
                 "latency": metrics.get_listener_latency(
                     f"{lc.listen_host}:{lc.listen_port}"
                 ).stats(),
@@ -97,9 +123,111 @@ def _handle_status(ctx: dict) -> tuple[int, dict]:
     return 200, data
 
 
-def _handle_metrics(ctx: dict) -> tuple[int, dict]:
+def _render_openmetrics(ctx: dict) -> str:
-    """GET /metrics -- full metrics counters."""
+    """Render all metrics in OpenMetrics text format."""
-    return 200, ctx["metrics"].to_dict()
+    m: Metrics = ctx["metrics"]
+    lines: list[str] = []
+
+    def _counter(name: str, help_text: str, value: int) -> None:
+        lines.append(f"# HELP {name} {help_text}")
+        lines.append(f"# TYPE {name} counter")
+        lines.append(f"{name}_total {value}")
+
+    def _gauge(name: str, help_text: str, value: float) -> None:
+        lines.append(f"# HELP {name} {help_text}")
+        lines.append(f"# TYPE {name} gauge")
+        lines.append(f"{name} {value}")
+
+    def _summary(name: str, help_text: str, q: dict,
+                 labels: str = "") -> None:
+        lines.append(f"# HELP {name} {help_text}")
+        lines.append(f"# TYPE {name} summary")
+        lb = f"{{{labels}," if labels else "{"
+        for quantile in ("0.5", "0.95", "0.99"):
+            lines.append(f'{name}{lb}quantile="{quantile}"}} {q[quantile]:.6f}')
+        lw = f"{{{labels}}}" if labels else ""
+        lines.append(f"{name}_count{lw} {q['count']}")
+        lines.append(f"{name}_sum{lw} {q['sum']:.6f}")
+
+    # -- counters
+    _counter("s5p_connections", "Total connection attempts.", m.connections)
+    _counter("s5p_connections_success",
+             "Connections successfully relayed.", m.success)
+    _counter("s5p_connections_failed", "Connection failures.", m.failed)
+    _counter("s5p_retries", "Connection retry attempts.", m.retries)
+    _counter("s5p_auth_failures",
+             "SOCKS5 authentication failures.", m.auth_failures)
+    _counter("s5p_bytes_in",
+             "Bytes received from clients.", m.bytes_in)
+    _counter("s5p_bytes_out",
+             "Bytes sent to clients.", m.bytes_out)
+
+    # -- gauges
+    _gauge("s5p_active_connections",
+           "Currently open connections.", m.active)
+    _gauge("s5p_uptime_seconds",
+           "Seconds since server start.",
+           round(time.monotonic() - m.started, 1))
+    _gauge("s5p_connection_rate",
+           "Connections per second (rolling window).",
+           round(m.conn_rate.rate(), 4))
+
+    # -- pool gauges
+    pools: dict = ctx.get("pools") or {}
+    if pools:
+        lines.append("# HELP s5p_pool_proxies_alive Alive proxies in pool.")
+        lines.append("# TYPE s5p_pool_proxies_alive gauge")
+        for name, p in pools.items():
+            lines.append(f's5p_pool_proxies_alive{{pool="{name}"}} {p.alive_count}')
+        lines.append("# HELP s5p_pool_proxies_total Total proxies in pool.")
+        lines.append("# TYPE s5p_pool_proxies_total gauge")
+        for name, p in pools.items():
+            lines.append(f's5p_pool_proxies_total{{pool="{name}"}} {p.count}')
+    elif ctx.get("pool"):
+        p = ctx["pool"]
+        _gauge("s5p_pool_proxies_alive", "Alive proxies in pool.", p.alive_count)
+        _gauge("s5p_pool_proxies_total", "Total proxies in pool.", p.count)
+
+    # -- latency summary (global)
+    q = m.latency.quantiles()
+    if q:
+        _summary("s5p_chain_latency_seconds",
+                 "Chain build latency in seconds.", q)
+
+    # -- per-listener latency summaries
+    if m.listener_latency:
+        lines.append(
+            "# HELP s5p_listener_chain_latency_seconds "
+            "Per-listener chain build latency in seconds.")
+        lines.append("# TYPE s5p_listener_chain_latency_seconds summary")
+        for key, tracker in m.listener_latency.items():
+            lq = tracker.quantiles()
+            if not lq:
+                continue
+            for quantile in ("0.5", "0.95", "0.99"):
+                lines.append(
+                    f's5p_listener_chain_latency_seconds'
+                    f'{{listener="{key}",quantile="{quantile}"}} '
+                    f'{lq[quantile]:.6f}')
+            lines.append(
+                f's5p_listener_chain_latency_seconds_count'
+                f'{{listener="{key}"}} {lq["count"]}')
+            lines.append(
+                f's5p_listener_chain_latency_seconds_sum'
+                f'{{listener="{key}"}} {lq["sum"]:.6f}')
+
+    lines.append("# EOF")
+    return "\n".join(lines) + "\n"
+
+
+_OPENMETRICS_CT = (
+    "application/openmetrics-text; version=1.0.0; charset=utf-8"
+)
+
+
+def _handle_metrics(ctx: dict) -> tuple[int, str]:
+    """GET /metrics -- OpenMetrics exposition."""
+    return 200, _render_openmetrics(ctx)
 
 
 def _handle_pool(ctx: dict, alive_only: bool = False) -> tuple[int, dict]:
@@ -166,6 +294,9 @@ 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 {}),
+                **({"retries": lc.retries} if lc.retries else {}),
             }
             for lc in config.listeners
         ],
@@ -185,12 +316,15 @@ def _handle_config(ctx: dict) -> tuple[int, dict]:
                 if src.mitm is not None:
                     s["mitm"] = src.mitm
                 sources.append(s)
-            pools_data[name] = {
+            pool_entry: dict = {
                 "sources": sources,
                 "refresh": pp.refresh,
                 "test_interval": pp.test_interval,
                 "max_fails": pp.max_fails,
             }
+            if pp.allowed_protos:
+                pool_entry["allowed_protos"] = pp.allowed_protos
+            pools_data[name] = pool_entry
         data["proxy_pools"] = pools_data
     elif config.proxy_pool:
         pp = config.proxy_pool
@@ -293,8 +427,13 @@ _POST_ROUTES: dict[str, str] = {
 }
 
 
-async def _route(method: str, path: str, ctx: dict) -> tuple[int, dict]:
+async def _route(
-    """Dispatch request to the appropriate handler."""
+    method: str, path: str, ctx: dict,
+) -> tuple[int, dict | str]:
+    """Dispatch request to the appropriate handler.
+
+    Returns (status, body) where body is a dict (JSON) or str (text).
+    """
     if method == "GET" and path in _GET_ROUTES:
         name = _GET_ROUTES[path]
         if name == "status":
@@ -349,6 +488,10 @@ async def _handle_connection(
             return
 
         status, body = await _route(method, path, ctx)
+        if isinstance(body, str):
+            _http_response(writer, status, body.encode(),
+                           _OPENMETRICS_CT)
+        else:
             _json_response(writer, status, body)
         await writer.drain()
     except (TimeoutError, ConnectionError, OSError):

src/s5p/config.py

@@ -57,6 +57,7 @@ class ProxyPoolConfig:
     max_fails: int = 3
     state_file: str = ""
     report_url: str = ""
+    allowed_protos: list[str] = field(default_factory=list)
 
     def __post_init__(self) -> None:
         """Backward compat: extract hostname from legacy test_url."""
@@ -85,8 +86,16 @@ 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)
+    retries: int = 0  # 0 = use global default
+
+    @property
+    def pool_hops(self) -> int:
+        """Number of pool hops (backward compat)."""
+        return len(self.pool_seq)
 
 
 @dataclass
@@ -185,9 +194,26 @@ def _parse_pool_config(pool_raw: dict) -> ProxyPoolConfig:
     }
     if "test_targets" in pool_raw:
         kwargs["test_targets"] = list(pool_raw["test_targets"])
+    if "allowed_protos" in pool_raw:
+        kwargs["allowed_protos"] = list(pool_raw["allowed_protos"])
     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 +328,32 @@ 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 "retries" in entry:
+                lc.retries = int(entry["retries"])
             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":
+                if isinstance(item, str):
-                    lc.pool_hops += 1
+                    lower = item.lower()
-                elif isinstance(item, str):
+                    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 +363,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 +377,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

src/s5p/metrics.py

@@ -67,6 +67,23 @@ class LatencyTracker:
             "p99": round(ms[min(int(n * 0.99), n - 1)], 1),
         }
 
+    def quantiles(self) -> dict | None:
+        """Return ``{count, sum, 0.5, 0.95, 0.99}`` in seconds, or None.
+
+        Designed for Prometheus/OpenMetrics summary exposition.
+        """
+        n = len(self._samples)
+        if n == 0:
+            return None
+        s = sorted(self._samples)
+        return {
+            "count": n,
+            "sum": sum(s),
+            "0.5": s[int(n * 0.50)],
+            "0.95": s[min(int(n * 0.95), n - 1)],
+            "0.99": s[min(int(n * 0.99), n - 1)],
+        }
+
 
 class Metrics:
     """Connection metrics.
@@ -82,6 +99,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 +121,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 +137,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,

src/s5p/pool.py

@@ -241,6 +241,8 @@ class ProxyPool:
         now = time.time()
         seen: set[str] = set()
         for hop in proxies:
+            if self._cfg.allowed_protos and hop.proto not in self._cfg.allowed_protos:
+                continue
             key = f"{hop.proto}://{hop.host}:{hop.port}"
             seen.add(key)
             if key in self._proxies:

src/s5p/server.py

@@ -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,28 +188,44 @@ 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
+                # tor_nodes round-robin overrides the listener's first hop
                 if tor_rr and effective_chain:
                     node = tor_rr.next()
                     effective_chain[0] = node
                     if hop_pools:
                         fhp = hop_pools.get((node.host, node.port))
 
-            pool_hops: list[ChainHop] = []
+            pool_hops: list[tuple[ChainHop, ProxyPool]] = []
-            if proxy_pool and listener.pool_hops > 0:
+            if pool_seq and not skip_pool:
-                for _ in range(listener.pool_hops):
+                for candidates in pool_seq:
-                    hop = await proxy_pool.get()
+                    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 +243,9 @@ async def _handle_client(
                 break
             except (ProtoError, TimeoutError, ConnectionError, OSError) as e:
                 last_err = e
-                if pool_hops and proxy_pool:
+                if pool_hops:
-                    for hop in pool_hops:
+                    for hop, pp in pool_hops:
-                        proxy_pool.report_failure(hop)
+                        pp.report_failure(hop)
                 if metrics:
                     metrics.retries += 1
                 if attempt + 1 < attempts:
@@ -295,17 +381,20 @@ async def serve(config: Config) -> None:
         await pool.start()
         proxy_pools["default"] = pool
 
-    def _pool_for(lc: ListenerConfig) -> ProxyPool | None:
+    def _pools_for(lc: ListenerConfig) -> list[list[ProxyPool]]:
-        """Resolve the proxy pool for a listener."""
+        """Resolve the ordered list of candidate proxy pools for a listener."""
-        if lc.pool_hops <= 0:
+        result: list[list[ProxyPool]] = []
-            return None
+        for candidates in lc.pool_seq:
-        name = lc.pool_name or "default"
+            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 +450,18 @@ 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,
+                    r, w, _lc, config.timeout,
-                    _pool, metrics, _hp, tor_rr, hop_pools,
+                    _lc.retries or config.retries,
+                    _pools, metrics, _hp, tor_rr, hop_pools,
                 )
 
         srv = await asyncio.start_server(on_client, lc.listen_host, lc.listen_port)
@@ -380,10 +470,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 ""
+        pool_desc = ""
-        if lc_pool and lc_pool.name != "default":
+        if nhops:
-            pool_desc += f" [{lc_pool.name}]"
+            all_names = {n for cands in lc.pool_seq for n in cands}
-        logger.info("listener %s  chain: %s%s", addr, chain_desc, pool_desc)
+            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 +571,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()

tests/conftest.py

@@ -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

tests/test_api.py

@@ -13,6 +13,7 @@ from s5p.api import (
     _handle_tor_newnym,
     _json_response,
     _parse_request,
+    _render_openmetrics,
     _route,
 )
 from s5p.config import ChainHop, Config, ListenerConfig, PoolSourceConfig, ProxyPoolConfig
@@ -132,7 +133,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 +152,158 @@ 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 TestHandleStatusRetries:
+    """Test retries in /status listener entries."""
+
+    def test_retries_present_when_set(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(
+                    listen_host="0.0.0.0", listen_port=1080,
+                    retries=5,
+                ),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_status(ctx)
+        assert body["listeners"][0]["retries"] == 5
+
+    def test_retries_absent_when_zero(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_status(ctx)
+        assert "retries" not in body["listeners"][0]
+
+
+class TestHandleConfigRetries:
+    """Test retries in /config listener entries."""
+
+    def test_retries_present_when_set(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(
+                    listen_host="0.0.0.0", listen_port=1080,
+                    retries=7,
+                ),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert body["listeners"][0]["retries"] == 7
+
+    def test_retries_absent_when_zero(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert "retries" not in body["listeners"][0]
+
+
+class TestHandleConfigAllowedProtos:
+    """Test allowed_protos in /config pool entries."""
+
+    def test_allowed_protos_present(self):
+        pp = ProxyPoolConfig(
+            sources=[],
+            allowed_protos=["socks5"],
+        )
+        config = Config(
+            proxy_pools={"socks_only": pp},
+            listeners=[ListenerConfig()],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert body["proxy_pools"]["socks_only"]["allowed_protos"] == ["socks5"]
+
+    def test_allowed_protos_absent_when_empty(self):
+        pp = ProxyPoolConfig(sources=[])
+        config = Config(
+            proxy_pools={"default": pp},
+            listeners=[ListenerConfig()],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert "allowed_protos" not in body["proxy_pools"]["default"]
+
+
 class TestHandleStatusPools:
     """Test GET /status with multiple named pools."""
 
@@ -170,21 +323,187 @@ class TestHandleStatusPools:
         assert body["pools"]["mitm"] == {"alive": 3, "total": 8}
 
 
-class TestHandleMetrics:
+class TestHandleStatusMultiPool:
-    """Test GET /metrics handler."""
+    """Test pool_seq appears in /status only for multi-pool listeners."""
 
-    def test_returns_dict(self):
+    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 (OpenMetrics format)."""
+
+    def test_returns_openmetrics_string(self):
         ctx = _make_ctx()
         ctx["metrics"].connections = 42
         ctx["metrics"].bytes_in = 1024
         status, body = _handle_metrics(ctx)
         assert status == 200
-        assert body["connections"] == 42
+        assert isinstance(body, str)
-        assert body["bytes_in"] == 1024
+        assert body.rstrip().endswith("# EOF")
-        assert "uptime" in body
+
-        assert "rate" in body
+    def test_counter_values(self):
-        assert "latency" in body
+        ctx = _make_ctx()
-        assert "listener_latency" in body
+        ctx["metrics"].connections = 42
+        ctx["metrics"].bytes_in = 1024
+        _, body = _handle_metrics(ctx)
+        assert "s5p_connections_total 42" in body
+        assert "s5p_bytes_in_total 1024" in body
+
+
+class TestRenderOpenMetrics:
+    """Test OpenMetrics text rendering."""
+
+    def test_eof_terminator(self):
+        ctx = _make_ctx()
+        text = _render_openmetrics(ctx)
+        assert text.rstrip().endswith("# EOF")
+        assert text.endswith("\n")
+
+    def test_type_declarations(self):
+        ctx = _make_ctx()
+        text = _render_openmetrics(ctx)
+        assert "# TYPE s5p_connections counter" in text
+        assert "# TYPE s5p_active_connections gauge" in text
+        assert "# TYPE s5p_uptime_seconds gauge" in text
+
+    def test_help_lines(self):
+        ctx = _make_ctx()
+        text = _render_openmetrics(ctx)
+        assert "# HELP s5p_connections Total connection attempts." in text
+        assert "# HELP s5p_active_connections Currently open connections." in text
+
+    def test_counter_values(self):
+        ctx = _make_ctx()
+        ctx["metrics"].connections = 100
+        ctx["metrics"].success = 95
+        ctx["metrics"].failed = 5
+        ctx["metrics"].retries = 10
+        ctx["metrics"].auth_failures = 2
+        ctx["metrics"].bytes_in = 4096
+        ctx["metrics"].bytes_out = 8192
+        text = _render_openmetrics(ctx)
+        assert "s5p_connections_total 100" in text
+        assert "s5p_connections_success_total 95" in text
+        assert "s5p_connections_failed_total 5" in text
+        assert "s5p_retries_total 10" in text
+        assert "s5p_auth_failures_total 2" in text
+        assert "s5p_bytes_in_total 4096" in text
+        assert "s5p_bytes_out_total 8192" in text
+
+    def test_gauge_values(self):
+        ctx = _make_ctx()
+        ctx["metrics"].active = 7
+        text = _render_openmetrics(ctx)
+        assert "s5p_active_connections 7" in text
+        assert "s5p_uptime_seconds " in text
+        assert "s5p_connection_rate " in text
+
+    def test_no_latency_when_empty(self):
+        ctx = _make_ctx()
+        text = _render_openmetrics(ctx)
+        assert "s5p_chain_latency_seconds" not in text
+
+    def test_latency_summary(self):
+        ctx = _make_ctx()
+        for i in range(1, 101):
+            ctx["metrics"].latency.record(i / 1000)
+        text = _render_openmetrics(ctx)
+        assert "# TYPE s5p_chain_latency_seconds summary" in text
+        assert 's5p_chain_latency_seconds{quantile="0.5"}' in text
+        assert 's5p_chain_latency_seconds{quantile="0.95"}' in text
+        assert 's5p_chain_latency_seconds{quantile="0.99"}' in text
+        assert "s5p_chain_latency_seconds_count 100" in text
+        assert "s5p_chain_latency_seconds_sum " in text
+
+    def test_listener_latency_summary(self):
+        ctx = _make_ctx()
+        tracker = ctx["metrics"].get_listener_latency("0.0.0.0:1080")
+        for i in range(1, 51):
+            tracker.record(i / 1000)
+        text = _render_openmetrics(ctx)
+        assert "# TYPE s5p_listener_chain_latency_seconds summary" in text
+        assert (
+            's5p_listener_chain_latency_seconds{listener="0.0.0.0:1080",'
+            'quantile="0.5"}'
+        ) in text
+        assert (
+            's5p_listener_chain_latency_seconds_count{listener="0.0.0.0:1080"} 50'
+        ) in text
+
+    def test_pool_gauges_multi(self):
+        pool_a = MagicMock()
+        pool_a.alive_count = 5
+        pool_a.count = 10
+        pool_a.name = "clean"
+        pool_b = MagicMock()
+        pool_b.alive_count = 3
+        pool_b.count = 8
+        pool_b.name = "mitm"
+        ctx = _make_ctx(pools={"clean": pool_a, "mitm": pool_b})
+        text = _render_openmetrics(ctx)
+        assert '# TYPE s5p_pool_proxies_alive gauge' in text
+        assert 's5p_pool_proxies_alive{pool="clean"} 5' in text
+        assert 's5p_pool_proxies_alive{pool="mitm"} 3' in text
+        assert 's5p_pool_proxies_total{pool="clean"} 10' in text
+        assert 's5p_pool_proxies_total{pool="mitm"} 8' in text
+
+    def test_pool_gauges_single(self):
+        pool = MagicMock()
+        pool.alive_count = 12
+        pool.count = 20
+        ctx = _make_ctx(pool=pool)
+        text = _render_openmetrics(ctx)
+        assert "s5p_pool_proxies_alive 12" in text
+        assert "s5p_pool_proxies_total 20" in text
+
+    def test_no_pool_metrics_when_unconfigured(self):
+        ctx = _make_ctx()
+        text = _render_openmetrics(ctx)
+        assert "s5p_pool_proxies" not in text
 
 
 class TestHandlePool:
@@ -318,7 +637,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 +664,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)

tests/test_cli.py

@@ -0,0 +1,153 @@
+"""Tests for CLI argument parsing."""
+
+import pytest
+
+from s5p import __version__
+from s5p.cli import _parse_args
+
+
+class TestDefaults:
+    """Default argument values."""
+
+    def test_no_args(self):
+        args = _parse_args([])
+        assert args.config is None
+        assert args.listen is None
+        assert args.chain is None
+        assert args.timeout is None
+        assert args.retries is None
+        assert args.max_connections is None
+        assert args.verbose is False
+        assert args.quiet is False
+        assert args.proxy_source is None
+        assert args.api is None
+        assert args.cprofile is None
+        assert args.tracemalloc is None
+
+
+class TestFlags:
+    """Flag parsing."""
+
+    def test_verbose(self):
+        args = _parse_args(["-v"])
+        assert args.verbose is True
+
+    def test_quiet(self):
+        args = _parse_args(["-q"])
+        assert args.quiet is True
+
+    def test_config(self):
+        args = _parse_args(["-c", "s5p.yaml"])
+        assert args.config == "s5p.yaml"
+
+    def test_config_long(self):
+        args = _parse_args(["--config", "s5p.yaml"])
+        assert args.config == "s5p.yaml"
+
+    def test_listen(self):
+        args = _parse_args(["-l", "0.0.0.0:9999"])
+        assert args.listen == "0.0.0.0:9999"
+
+    def test_chain(self):
+        args = _parse_args(["-C", "socks5://127.0.0.1:9050"])
+        assert args.chain == "socks5://127.0.0.1:9050"
+
+    def test_chain_multi(self):
+        args = _parse_args(["-C", "socks5://a:1080,http://b:8080"])
+        assert args.chain == "socks5://a:1080,http://b:8080"
+
+    def test_timeout(self):
+        args = _parse_args(["-t", "30"])
+        assert args.timeout == 30.0
+
+    def test_retries(self):
+        args = _parse_args(["-r", "5"])
+        assert args.retries == 5
+
+    def test_max_connections(self):
+        args = _parse_args(["-m", "512"])
+        assert args.max_connections == 512
+
+    def test_proxy_source(self):
+        args = _parse_args(["-S", "http://api:8081/proxies"])
+        assert args.proxy_source == "http://api:8081/proxies"
+
+    def test_api(self):
+        args = _parse_args(["--api", "127.0.0.1:1081"])
+        assert args.api == "127.0.0.1:1081"
+
+    def test_cprofile_default(self):
+        args = _parse_args(["--cprofile"])
+        assert args.cprofile == "s5p.prof"
+
+    def test_cprofile_custom(self):
+        args = _parse_args(["--cprofile", "out.prof"])
+        assert args.cprofile == "out.prof"
+
+    def test_tracemalloc_default(self):
+        args = _parse_args(["--tracemalloc"])
+        assert args.tracemalloc == 10
+
+    def test_tracemalloc_custom(self):
+        args = _parse_args(["--tracemalloc", "20"])
+        assert args.tracemalloc == 20
+
+
+class TestVersion:
+    """--version flag."""
+
+    def test_version_output(self, capsys):
+        with pytest.raises(SystemExit, match="0"):
+            _parse_args(["--version"])
+        captured = capsys.readouterr()
+        assert captured.out.strip() == f"s5p {__version__}"
+
+    def test_version_short(self, capsys):
+        with pytest.raises(SystemExit, match="0"):
+            _parse_args(["-V"])
+        captured = capsys.readouterr()
+        assert "0.3.0" in captured.out
+
+
+class TestCombinations:
+    """Multiple flags together."""
+
+    def test_verbose_with_chain(self):
+        args = _parse_args(["-v", "-C", "socks5://tor:9050"])
+        assert args.verbose is True
+        assert args.chain == "socks5://tor:9050"
+
+    def test_config_with_api(self):
+        args = _parse_args(["-c", "s5p.yaml", "--api", "0.0.0.0:1090"])
+        assert args.config == "s5p.yaml"
+        assert args.api == "0.0.0.0:1090"
+
+    def test_listen_with_timeout_and_retries(self):
+        args = _parse_args(["-l", ":8080", "-t", "15", "-r", "3"])
+        assert args.listen == ":8080"
+        assert args.timeout == 15.0
+        assert args.retries == 3
+
+
+class TestInvalid:
+    """Invalid argument handling."""
+
+    def test_unknown_flag(self):
+        with pytest.raises(SystemExit, match="2"):
+            _parse_args(["--nonexistent"])
+
+    def test_timeout_non_numeric(self):
+        with pytest.raises(SystemExit, match="2"):
+            _parse_args(["-t", "abc"])
+
+    def test_retries_non_numeric(self):
+        with pytest.raises(SystemExit, match="2"):
+            _parse_args(["-r", "abc"])
+
+    def test_max_connections_non_numeric(self):
+        with pytest.raises(SystemExit, match="2"):
+            _parse_args(["-m", "abc"])
+
+    def test_tracemalloc_non_numeric(self):
+        with pytest.raises(SystemExit, match="2"):
+            _parse_args(["--tracemalloc", "abc"])

tests/test_config.py

@@ -10,6 +10,7 @@ from s5p.config import (
     parse_api_proxies,
     parse_proxy_url,
 )
+from s5p.server import _bypass_match
 
 
 class TestParseProxyUrl:
@@ -306,6 +307,39 @@ class TestProxyPools:
         assert c.listeners[1].pool_hops == 0
 
 
+class TestAllowedProtos:
+    """Test pool-level allowed_protos config."""
+
+    def test_allowed_protos_from_yaml(self, tmp_path):
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "proxy_pools:\n"
+            "  socks_only:\n"
+            "    sources: []\n"
+            "    allowed_protos: [socks5]\n"
+            "  any:\n"
+            "    sources: []\n"
+        )
+        c = load_config(cfg_file)
+        assert c.proxy_pools["socks_only"].allowed_protos == ["socks5"]
+        assert c.proxy_pools["any"].allowed_protos == []
+
+    def test_allowed_protos_multiple(self, tmp_path):
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "proxy_pool:\n"
+            "  sources: []\n"
+            "  allowed_protos: [socks5, http]\n"
+        )
+        c = load_config(cfg_file)
+        assert c.proxy_pool.allowed_protos == ["socks5", "http"]
+
+    def test_allowed_protos_default_empty(self):
+        from s5p.config import ProxyPoolConfig
+        cfg = ProxyPoolConfig()
+        assert cfg.allowed_protos == []
+
+
 class TestTorNodes:
     """Test tor_nodes config parsing."""
 
@@ -407,6 +441,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 +624,168 @@ class TestListenerPoolCompat:
         lc = c.listeners[0]
         # explicit listeners: no auto pool_hops
         assert lc.pool_hops == 0
+
+
+class TestListenerRetries:
+    """Test per-listener retry override config."""
+
+    def test_retries_default(self):
+        lc = ListenerConfig()
+        assert lc.retries == 0
+
+    def test_retries_from_yaml(self, tmp_path):
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "listeners:\n"
+            "  - listen: 1080\n"
+            "    retries: 5\n"
+            "    chain:\n"
+            "      - socks5://127.0.0.1:9050\n"
+            "      - pool\n"
+            "  - listen: 1081\n"
+            "    chain:\n"
+            "      - socks5://127.0.0.1:9050\n"
+        )
+        c = load_config(cfg_file)
+        assert c.listeners[0].retries == 5
+        assert c.listeners[1].retries == 0
+
+    def test_retries_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].retries == 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

tests/test_integration.py

@@ -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())

tests/test_metrics.py

@@ -52,6 +52,23 @@ class TestLatencyTracker:
         assert s["min"] == 500.0
         assert s["max"] == 500.0
 
+    def test_quantiles_empty(self):
+        lt = LatencyTracker()
+        assert lt.quantiles() is None
+
+    def test_quantiles_seconds(self):
+        lt = LatencyTracker()
+        for i in range(1, 101):
+            lt.record(i / 1000)
+        q = lt.quantiles()
+        assert q is not None
+        assert q["count"] == 100
+        assert 0.050 <= q["0.5"] <= 0.052
+        assert 0.095 <= q["0.95"] <= 0.097
+        assert 0.099 <= q["0.99"] <= 0.101
+        assert "sum" in q
+        assert q["sum"] > 0
+
 
 # -- RateTracker -------------------------------------------------------------
 

tests/test_pool.py

@@ -171,6 +171,46 @@ class TestProxyPoolMerge:
         assert pool.count == 1
 
 
+class TestProxyPoolAllowedProtos:
+    """Test pool-level proxy protocol filter."""
+
+    def test_allowed_protos_filters_merge(self):
+        cfg = ProxyPoolConfig(sources=[], allowed_protos=["socks5"])
+        pool = ProxyPool(cfg, [], timeout=10.0)
+        proxies = [
+            ChainHop(proto="socks5", host="1.2.3.4", port=1080),
+            ChainHop(proto="http", host="5.6.7.8", port=8080),
+            ChainHop(proto="socks4", host="9.9.9.9", port=1080),
+        ]
+        pool._merge(proxies)
+        assert pool.count == 1
+        assert "socks5://1.2.3.4:1080" in pool._proxies
+
+    def test_allowed_protos_multiple(self):
+        cfg = ProxyPoolConfig(sources=[], allowed_protos=["socks5", "http"])
+        pool = ProxyPool(cfg, [], timeout=10.0)
+        proxies = [
+            ChainHop(proto="socks5", host="1.2.3.4", port=1080),
+            ChainHop(proto="http", host="5.6.7.8", port=8080),
+            ChainHop(proto="socks4", host="9.9.9.9", port=1080),
+        ]
+        pool._merge(proxies)
+        assert pool.count == 2
+        assert "socks5://1.2.3.4:1080" in pool._proxies
+        assert "http://5.6.7.8:8080" in pool._proxies
+        assert "socks4://9.9.9.9:1080" not in pool._proxies
+
+    def test_empty_allowed_protos_accepts_all(self):
+        cfg = ProxyPoolConfig(sources=[], allowed_protos=[])
+        pool = ProxyPool(cfg, [], timeout=10.0)
+        proxies = [
+            ChainHop(proto="socks5", host="1.2.3.4", port=1080),
+            ChainHop(proto="http", host="5.6.7.8", port=8080),
+        ]
+        pool._merge(proxies)
+        assert pool.count == 2
+
+
 class TestProxyPoolGet:
     """Test proxy selection."""
 

tests/test_proto.py

@@ -1,6 +1,66 @@
 """Tests for protocol helpers."""
 
-from s5p.proto import Socks5AddrType, encode_address
+import asyncio
+
+import pytest
+
+from s5p.proto import (
+    ProtoError,
+    Socks5AddrType,
+    Socks5Reply,
+    encode_address,
+    http_connect,
+    socks4_connect,
+    socks5_connect,
+)
+
+# -- helpers -----------------------------------------------------------------
+
+
+class _MockTransport(asyncio.Transport):
+    """Minimal transport that captures writes and supports drain."""
+
+    def __init__(self):
+        super().__init__()
+        self.written = bytearray()
+        self._closing = False
+
+    def write(self, data):
+        self.written.extend(data)
+
+    def is_closing(self):
+        return self._closing
+
+    def close(self):
+        self._closing = True
+
+    def get_extra_info(self, name, default=None):
+        return default
+
+
+def _make_streams(response_data: bytes):
+    """Create mock reader/writer for protocol tests.
+
+    Must be called from within a running event loop.
+    """
+    reader = asyncio.StreamReader()
+    reader.feed_data(response_data)
+    reader.feed_eof()
+
+    protocol = asyncio.StreamReaderProtocol(reader)
+    transport = _MockTransport()
+    protocol.connection_made(transport)
+
+    writer = asyncio.StreamWriter(transport, protocol, reader, asyncio.get_running_loop())
+    return reader, writer
+
+
+def _run(coro):
+    """Run a coroutine in a fresh event loop."""
+    asyncio.run(coro)
+
+
+# -- encode_address ----------------------------------------------------------
 
 
 class TestEncodeAddress:
@@ -11,12 +71,265 @@ class TestEncodeAddress:
         assert atyp == Socks5AddrType.IPV4
         assert data == b"\x7f\x00\x00\x01"
 
+    def test_ipv4_zeros(self):
+        atyp, data = encode_address("0.0.0.0")
+        assert atyp == Socks5AddrType.IPV4
+        assert data == b"\x00\x00\x00\x00"
+
     def test_ipv6(self):
         atyp, data = encode_address("::1")
         assert atyp == Socks5AddrType.IPV6
         assert len(data) == 16
+        assert data[-1] == 1
+
+    def test_ipv6_full(self):
+        atyp, data = encode_address("2001:db8::1")
+        assert atyp == Socks5AddrType.IPV6
+        assert len(data) == 16
 
     def test_domain(self):
         atyp, data = encode_address("example.com")
         assert atyp == Socks5AddrType.DOMAIN
         assert data == bytes([11]) + b"example.com"
+
+    def test_domain_short(self):
+        atyp, data = encode_address("a.co")
+        assert atyp == Socks5AddrType.DOMAIN
+        assert data == bytes([4]) + b"a.co"
+
+    def test_domain_long(self):
+        host = "sub.domain.example.com"
+        atyp, data = encode_address(host)
+        assert atyp == Socks5AddrType.DOMAIN
+        assert data[0] == len(host)
+        assert data[1:] == host.encode("ascii")
+
+
+# -- socks5_connect ----------------------------------------------------------
+
+
+class TestSocks5Connect:
+    """Test SOCKS5 handshake building."""
+
+    def test_no_auth_success(self):
+        """Successful SOCKS5 connect without auth."""
+        bind_addr = b"\x01\x00\x00\x00\x00\x00\x00"  # IPv4 0.0.0.0:0
+        response = b"\x05\x00" + b"\x05\x00\x00" + bind_addr
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await socks5_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+    def test_auth_success(self):
+        """Successful SOCKS5 connect with username/password auth."""
+        bind_addr = b"\x01\x00\x00\x00\x00\x00\x00"
+        response = b"\x05\x02" + b"\x01\x00" + b"\x05\x00\x00" + bind_addr
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await socks5_connect(reader, writer, "example.com", 80, "user", "pass")
+
+        _run(run())
+
+    def test_auth_failure(self):
+        """SOCKS5 auth rejected by server."""
+        response = b"\x05\x02" + b"\x01\x01"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="authentication failed"):
+                await socks5_connect(reader, writer, "example.com", 80, "user", "bad")
+
+        _run(run())
+
+    def test_no_acceptable_methods(self):
+        """Server rejects all auth methods (0xFF)."""
+        response = b"\x05\xff"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="no acceptable"):
+                await socks5_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+    def test_connect_refused(self):
+        """SOCKS5 connect reply with connection refused."""
+        bind_addr = b"\x01\x00\x00\x00\x00\x00\x00"
+        response = b"\x05\x00" + b"\x05\x05\x00" + bind_addr
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="connect failed"):
+                await socks5_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+    def test_wrong_version(self):
+        """Server responds with wrong SOCKS version."""
+        response = b"\x04\x00"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="unexpected version"):
+                await socks5_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+    def test_server_requires_auth_no_creds(self):
+        """Server demands auth but no credentials provided."""
+        response = b"\x05\x02"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="requires auth"):
+                await socks5_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+
+# -- socks4_connect ----------------------------------------------------------
+
+
+class TestSocks4Connect:
+    """Test SOCKS4/4a request building."""
+
+    def test_ip_success(self):
+        """SOCKS4 connect with IP address."""
+        response = b"\x00\x5a" + b"\x00" * 6
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await socks4_connect(reader, writer, "1.2.3.4", 80)
+
+        _run(run())
+
+    def test_domain_success(self):
+        """SOCKS4a connect with domain name."""
+        response = b"\x00\x5a" + b"\x00" * 6
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await socks4_connect(reader, writer, "example.com", 80)
+
+        _run(run())
+
+    def test_rejected(self):
+        """SOCKS4 request rejected."""
+        response = b"\x00\x5b" + b"\x00" * 6
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="rejected"):
+                await socks4_connect(reader, writer, "1.2.3.4", 80)
+
+        _run(run())
+
+
+# -- http_connect ------------------------------------------------------------
+
+
+class TestHttpConnect:
+    """Test HTTP CONNECT request building."""
+
+    def test_success_200(self):
+        """HTTP CONNECT with 200 response."""
+        response = b"HTTP/1.1 200 Connection Established\r\n\r\n"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await http_connect(reader, writer, "example.com", 443)
+
+        _run(run())
+
+    def test_success_with_headers(self):
+        """HTTP CONNECT with extra headers in response."""
+        response = b"HTTP/1.1 200 OK\r\nX-Proxy: test\r\n\r\n"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await http_connect(reader, writer, "example.com", 443)
+
+        _run(run())
+
+    def test_auth_success(self):
+        """HTTP CONNECT with proxy authentication."""
+        response = b"HTTP/1.1 200 OK\r\n\r\n"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            await http_connect(reader, writer, "example.com", 443, "user", "pass")
+
+        _run(run())
+
+    def test_forbidden(self):
+        """HTTP CONNECT with 403 response."""
+        response = b"HTTP/1.1 403 Forbidden\r\n\r\n"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="connect failed"):
+                await http_connect(reader, writer, "example.com", 443)
+
+        _run(run())
+
+    def test_proxy_auth_required(self):
+        """HTTP CONNECT with 407 response."""
+        response = b"HTTP/1.1 407 Proxy Authentication Required\r\n\r\n"
+
+        async def run():
+            reader, writer = _make_streams(response)
+            with pytest.raises(ProtoError, match="connect failed"):
+                await http_connect(reader, writer, "example.com", 443)
+
+        _run(run())
+
+    def test_empty_response(self):
+        """HTTP CONNECT with empty response."""
+
+        async def run():
+            reader, writer = _make_streams(b"")
+            with pytest.raises(ProtoError, match="empty response"):
+                await http_connect(reader, writer, "example.com", 443)
+
+        _run(run())
+
+
+# -- Socks5Reply enum -------------------------------------------------------
+
+
+class TestSocks5Reply:
+    """Test SOCKS5 reply code values."""
+
+    def test_succeeded(self):
+        assert Socks5Reply.SUCCEEDED == 0x00
+
+    def test_general_failure(self):
+        assert Socks5Reply.GENERAL_FAILURE == 0x01
+
+    def test_connection_refused(self):
+        assert Socks5Reply.CONNECTION_REFUSED == 0x05
+
+    def test_command_not_supported(self):
+        assert Socks5Reply.COMMAND_NOT_SUPPORTED == 0x07
+
+    def test_address_type_not_supported(self):
+        assert Socks5Reply.ADDRESS_TYPE_NOT_SUPPORTED == 0x08
+
+
+# -- ProtoError --------------------------------------------------------------
+
+
+class TestProtoError:
+    """Test ProtoError exception."""
+
+    def test_default_reply(self):
+        err = ProtoError("test error")
+        assert str(err) == "test error"
+        assert err.reply == Socks5Reply.GENERAL_FAILURE
+
+    def test_custom_reply(self):
+        err = ProtoError("refused", Socks5Reply.CONNECTION_REFUSED)
+        assert err.reply == Socks5Reply.CONNECTION_REFUSED