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>

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