tests: SOCKS BIND

The proxy accepts the connection expected and joins it with the
client both ways, and refuses one from another address than the one
asked for. socks_bind() and socks_bind_peer() in the harness make the
request and read the second reply.

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

tests/README.md

@@ -45,7 +45,8 @@ Servers are stopped for you when the case ends, whether or not it passed.
 
 `t` offers `http()` (direct, through an HTTP proxy, or over a CONNECT
 tunnel), `socks_http()` and `socks_connect()` for SOCKS4 and SOCKS5,
-`socks_udp_associate()`, `raw()` for bytes a real client would never send,
+`socks_udp_associate()`, `socks_bind()` and `socks_bind_peer()` for a SOCKS5
+BIND, `raw()` for bytes a real client would never send,
 and `run_config()` for configurations that are meant to be rejected.
 Assertions are `eq`, `ne`, `contains`, `not_contains`, `in_range`,
 `not_in_range`, plus `ok`, `fail` and `skip`. `harness.field()` and

tests/cases/socks.py

@@ -1,5 +1,7 @@
 """The SOCKS proxy, reaching the built-in server."""
 
+import socket
+
 
 def run(t):
     srv = t.free_port()
@@ -85,6 +87,39 @@ def run(t):
     t.eq(None, reply,
          "an allow by host name does not let datagrams to an address through")
 
+    # --- BIND: the proxy accepts a connection for the client --------------
+    # The far side connects to where the first reply says, the second reply
+    # names it, and the two are then joined both ways.
+    ctrl, where = t.socks_bind(plain, "127.0.0.1")
+    peer = None
+    try:
+        peer = socket.create_connection(where, t.timeout)
+        peer.settimeout(t.timeout)
+        t.ne(None, t.socks_bind_peer(ctrl), "BIND takes the connection expected")
+        peer.sendall(b"from the far side")
+        t.eq(b"from the far side", t._recvall(ctrl, 17),
+             "what the far side sends reaches the client")
+        ctrl.sendall(b"from the client")
+        t.eq(b"from the client", t._recvall(peer, 15),
+             "what the client sends reaches the far side")
+    finally:
+        ctrl.close()
+        if peer:
+            peer.close()
+
+    # A connection from elsewhere than the address given is not the one
+    # waited for.
+    ctrl, where = t.socks_bind(plain, "192.0.2.1")
+    peer = None
+    try:
+        peer = socket.create_connection(where, t.timeout)
+        t.eq(None, t.socks_bind_peer(ctrl),
+             "BIND refuses a connection from another address")
+    finally:
+        ctrl.close()
+        if peer:
+            peer.close()
+
     # --- authentication ----------------------------------------------------
     t.eq(200, t.socks_http(guarded, origin + "/echo",
                            auth=("alice", "secret")).status,

tests/harness.py

@@ -677,6 +677,44 @@ class Tester:
         finally:
             conn.close()
 
+    def socks_bind(self, socks, host, port=0):
+        """Ask a SOCKS5 proxy to accept a connection from host for us.
+
+        Returns the control connection and the address the proxy listens
+        on, where the far side is to connect; socks_bind_peer() then reads
+        whether that connection was taken. Raises OSError if refused.
+        """
+        shost, sport = self._hostport(socks)
+        sock = socket.create_connection((shost, sport), self.timeout)
+        sock.settimeout(self.timeout)
+        try:
+            sock.sendall(b"\x05\x01\x00")
+            if self._recvall(sock, 2) != b"\x05\x00":
+                raise OSError("SOCKS5 handshake failed")
+            sock.sendall(b"\x05\x02\x00\x01" + socket.inet_aton(host) +
+                         struct.pack("!H", port))
+            reply = self._recvall(sock, 4)
+            if len(reply) < 4 or reply[1] != 0:
+                raise OSError("SOCKS5 BIND refused")
+            addr, bport = self._read_socks_addr(sock, reply[3])
+            if addr == "0.0.0.0":
+                addr = shost
+            return sock, (addr, bport)
+        except Exception:
+            sock.close()
+            raise
+
+    def socks_bind_peer(self, sock):
+        """The second reply to BIND: the address of the connection the
+        proxy took, or None if it refused it or gave up."""
+        try:
+            reply = self._recvall(sock, 4)
+            if len(reply) < 4 or reply[1] != 0:
+                return None
+            return self._read_socks_addr(sock, reply[3])
+        except OSError:
+            return None
+
     def socks_udp_associate(self, port, host="127.0.0.1"):
         """Ask for a UDP association and report the port handed back.