feat: add per-listener SOCKS5 server authentication (RFC 1929)

Per-listener username/password auth via `auth:` config key. When set,
clients must negotiate method 0x02 and pass RFC 1929 subnegotiation;
no-auth (0x00) is rejected to prevent downgrade. Listeners without
`auth` keep current no-auth behavior.

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

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

ROADMAP.md

@@ -42,6 +42,6 @@
 ## v1.0.0
 
 - [ ] Stable API and config format
-- [ ] Comprehensive test suite with mock proxies
+- [ ] Comprehensive test suite with mock proxies (integration tests done)
 - [ ] Systemd service unit
 - [ ] Performance benchmarks

TASKS.md

@@ -59,6 +59,12 @@
   - [x] API: merged `/pool` with per-pool breakdown, `/status` pools summary
   - [x] Backward compat: singular `proxy_pool:` registers as `"default"`
 
+- [x] Integration tests with mock SOCKS5 proxy (end-to-end)
+- [x] Per-destination bypass rules (CIDR, suffix, exact match)
+- [x] Weighted multi-candidate pool selection
+- [x] Onion chain-only routing (.onion skips pool hops)
+- [x] Graceful shutdown timeout (fixes cProfile data dump)
+
 ## Next
-- [ ] Integration tests with mock proxy server
+- [x] Integration tests with mock proxy server
 - [ ] SOCKS5 server-side authentication

TODO.md

@@ -4,7 +4,6 @@
 
 - SOCKS5 BIND and UDP ASSOCIATE commands
 - Chain randomization modes (round-robin, sticky-per-destination)
-- Per-destination chain rules (bypass chain for local addresses)
 - Systemd socket activation
 - Per-pool health test chain override (different base chain per pool)
 - Pool-level proxy protocol filter (only socks5 from pool X, only http from pool Y)

config/example.yaml

@@ -97,6 +97,9 @@ chain:
 # listeners:
 #   - listen: 0.0.0.0:1080
 #     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

docs/CHEATSHEET.md

@@ -90,6 +90,25 @@ listeners:
 | `localhost` | Exact host | String equal |
 | `.local` | Suffix | `*.local` and `local` |
 
+## Listener Authentication (config)
+
+```yaml
+listeners:
+  - listen: 0.0.0.0:1080
+    auth:
+      alice: s3cret
+      bob: hunter2
+    chain:
+      - socks5://127.0.0.1:9050
+      - pool
+```
+
+```bash
+curl -x socks5h://alice:s3cret@127.0.0.1:1080 https://example.com
+```
+
+No `auth:` key = no authentication required (default).
+
 ## Multi-Tor Round-Robin (config)
 
 ```yaml

docs/USAGE.md

@@ -242,6 +242,59 @@ 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

src/s5p/api.py

@@ -49,6 +49,19 @@ def _json_response(
     writer.write(header.encode() + payload)
 
 
+# -- helpers -----------------------------------------------------------------
+
+
+def _multi_pool(lc) -> bool:
+    """Check if a listener uses more than one distinct pool."""
+    return len({n for c in lc.pool_seq for n in c}) > 1
+
+
+def _pool_seq_entry(lc) -> dict:
+    """Build pool_seq dict entry for API responses."""
+    return {"pool_seq": lc.pool_seq}
+
+
 # -- route handlers ----------------------------------------------------------
 
 
@@ -88,7 +101,8 @@ def _handle_status(ctx: dict) -> tuple[int, dict]:
                 "chain": [str(h) for h in lc.chain],
                 "pool_hops": lc.pool_hops,
                 **({"pool": lc.pool_name} if lc.pool_name else {}),
-                **({"pool_seq": lc.pool_seq} if len({n for c in lc.pool_seq for n in c}) > 1 else {}),
+                **(_pool_seq_entry(lc) if _multi_pool(lc) else {}),
+                **({"auth": True} if lc.auth else {}),
                 "latency": metrics.get_listener_latency(
                     f"{lc.listen_host}:{lc.listen_port}"
                 ).stats(),
@@ -167,7 +181,8 @@ def _handle_config(ctx: dict) -> tuple[int, dict]:
                 "chain": [str(h) for h in lc.chain],
                 "pool_hops": lc.pool_hops,
                 **({"pool": lc.pool_name} if lc.pool_name else {}),
-                **({"pool_seq": lc.pool_seq} if len({n for c in lc.pool_seq for n in c}) > 1 else {}),
+                **(_pool_seq_entry(lc) if _multi_pool(lc) else {}),
+                **({"auth_users": len(lc.auth)} if lc.auth else {}),
             }
             for lc in config.listeners
         ],

src/s5p/config.py

@@ -88,6 +88,7 @@ class ListenerConfig:
     pool_seq: list[list[str]] = field(default_factory=list)
     pool_name: str = ""
     bypass: list[str] = field(default_factory=list)
+    auth: dict[str, str] = field(default_factory=dict)
 
     @property
     def pool_hops(self) -> int:
@@ -325,6 +326,10 @@ def load_config(path: str | Path) -> Config:
                 lc.listen_port = int(listen)
             if "bypass" in entry:
                 lc.bypass = list(entry["bypass"])
+            if "auth" in entry:
+                auth_raw = entry["auth"]
+                if isinstance(auth_raw, dict):
+                    lc.auth = {str(k): str(v) for k, v in auth_raw.items()}
             if "pool" in entry:
                 lc.pool_name = entry["pool"]
             default_pool = lc.pool_name or "default"

src/s5p/metrics.py

@@ -82,6 +82,7 @@ class Metrics:
         self.retries: int = 0
         self.bytes_in: int = 0
         self.bytes_out: int = 0
+        self.auth_failures: int = 0
         self.active: int = 0
         self.started: float = time.monotonic()
         self.conn_rate: RateTracker = RateTracker()
@@ -103,9 +104,10 @@ class Metrics:
         lat = self.latency.stats()
         p50 = f" p50={lat['p50']:.1f}ms" if lat else ""
         p95 = f" p95={lat['p95']:.1f}ms" if lat else ""
+        auth = f" auth_fail={self.auth_failures}" if self.auth_failures else ""
         return (
             f"conn={self.connections} ok={self.success} fail={self.failed} "
-            f"retries={self.retries} active={self.active} "
+            f"retries={self.retries} active={self.active}{auth} "
             f"in={_human_bytes(self.bytes_in)} out={_human_bytes(self.bytes_out)} "
             f"rate={rate:.2f}/s{p50}{p95} "
             f"up={h}h{m:02d}m{s:02d}s"
@@ -118,6 +120,7 @@ class Metrics:
             "success": self.success,
             "failed": self.failed,
             "retries": self.retries,
+            "auth_failures": self.auth_failures,
             "active": self.active,
             "bytes_in": self.bytes_in,
             "bytes_out": self.bytes_out,

src/s5p/server.py

@@ -131,13 +131,50 @@ async def _handle_client(
             return
 
         methods = await client_reader.readexactly(header[1])
-        if 0x00 not in methods:
-            client_writer.write(b"\x05\xff")
-            await client_writer.drain()
-            return
 
-        client_writer.write(b"\x05\x00")
+        if listener.auth:
-        await client_writer.drain()
+            # 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()
+                return
+
+            client_writer.write(b"\x05\x00")
+            await client_writer.drain()
 
         # -- connect request --
         req = await asyncio.wait_for(client_reader.readexactly(3), timeout=10.0)
@@ -443,7 +480,11 @@ async def serve(config: Config) -> None:
             else:
                 pool_desc = f" + pool [{' -> '.join(hop_labels)}]"
         bypass_desc = f"  bypass: {len(lc.bypass)} rules" if lc.bypass else ""
-        logger.info("listener %s  chain: %s%s%s", addr, chain_desc, pool_desc, bypass_desc)
+        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)
 

tests/test_api.py

@@ -151,6 +151,77 @@ class TestHandleStatus:
         assert body["listeners"][1]["latency"] is None
 
 
+class TestHandleStatusAuth:
+    """Test auth flag in /status listener entries."""
+
+    def test_auth_flag_present(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(
+                    listen_host="0.0.0.0", listen_port=1080,
+                    auth={"alice": "s3cret", "bob": "hunter2"},
+                ),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_status(ctx)
+        assert body["listeners"][0]["auth"] is True
+
+    def test_auth_flag_absent_when_empty(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_status(ctx)
+        assert "auth" not in body["listeners"][0]
+
+
+class TestHandleConfigAuth:
+    """Test auth_users in /config listener entries."""
+
+    def test_auth_users_count(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(
+                    listen_host="0.0.0.0", listen_port=1080,
+                    auth={"alice": "s3cret", "bob": "hunter2"},
+                ),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert body["listeners"][0]["auth_users"] == 2
+
+    def test_auth_users_absent_when_empty(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(listen_host="0.0.0.0", listen_port=1080),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        assert "auth_users" not in body["listeners"][0]
+
+    def test_passwords_not_exposed(self):
+        config = Config(
+            listeners=[
+                ListenerConfig(
+                    listen_host="0.0.0.0", listen_port=1080,
+                    auth={"alice": "s3cret"},
+                ),
+            ],
+        )
+        ctx = _make_ctx(config=config)
+        _, body = _handle_config(ctx)
+        listener = body["listeners"][0]
+        # only count, never passwords
+        assert "auth_users" in listener
+        assert "auth" not in listener
+        assert "s3cret" not in str(body)
+
+
 class TestHandleStatusPools:
     """Test GET /status with multiple named pools."""
 

tests/test_config.py

@@ -593,6 +593,61 @@ class TestListenerPoolCompat:
         assert lc.pool_hops == 0
 
 
+class TestAuthConfig:
+    """Test auth field in listener config."""
+
+    def test_auth_from_yaml(self, tmp_path):
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "listeners:\n"
+            "  - listen: 1080\n"
+            "    auth:\n"
+            "      alice: s3cret\n"
+            "      bob: hunter2\n"
+        )
+        c = load_config(cfg_file)
+        assert c.listeners[0].auth == {"alice": "s3cret", "bob": "hunter2"}
+
+    def test_auth_empty_default(self):
+        lc = ListenerConfig()
+        assert lc.auth == {}
+
+    def test_auth_absent_from_yaml(self, tmp_path):
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "listeners:\n"
+            "  - listen: 1080\n"
+        )
+        c = load_config(cfg_file)
+        assert c.listeners[0].auth == {}
+
+    def test_auth_numeric_password(self, tmp_path):
+        """YAML parses `admin: 12345` as int; must be coerced to str."""
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "listeners:\n"
+            "  - listen: 1080\n"
+            "    auth:\n"
+            "      admin: 12345\n"
+        )
+        c = load_config(cfg_file)
+        assert c.listeners[0].auth == {"admin": "12345"}
+
+    def test_auth_mixed_listeners(self, tmp_path):
+        """One listener with auth, one without."""
+        cfg_file = tmp_path / "test.yaml"
+        cfg_file.write_text(
+            "listeners:\n"
+            "  - listen: 1080\n"
+            "    auth:\n"
+            "      alice: pass\n"
+            "  - listen: 1081\n"
+        )
+        c = load_config(cfg_file)
+        assert c.listeners[0].auth == {"alice": "pass"}
+        assert c.listeners[1].auth == {}
+
+
 class TestBypassConfig:
     """Test bypass rules in listener config."""
 

tests/test_integration.py

@@ -56,6 +56,64 @@ async def _socks5_connect(
     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()
@@ -354,3 +412,167 @@ class TestOnionChainOnly:
                     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())