Repository: bitcoin/bitcoin
File: test/functional/test_framework/socks5.py (Python)

Code before:
                    with socket.create_connection((dest["actual_to_addr"], dest["actual_to_port"])) as conn_to:
                        forward_sockets(self.conn, conn_to)
                else:
                    logger.debug(f"Can't serve the connection to {requested_to}: the destinations factory returned None")
            else:
                logger.debug(f"Can't serve the connection to {requested_to}: no destinations factory")

            # Fall through to disconnect
        except Exception as e:
            logger.exception("socks5 request handling failed.")
            self.serv.queue.put(e)
        finally:
            if not self.serv.keep_alive:
                self.conn.close()
            else:
                logger.debug("Keeping client connection alive")

class Socks5Server():
    def __init__(self, conf):
        self.conf = conf
        self.s = socket.socket(conf.af)
        self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        # When using dynamic port allocation (port=0), ensure we don't get a
        # port that conflicts with the test framework's static port range.
        if conf.addr[1] == 0:
            set_freebsd_high_port_range(self.s)
        self.s.bind(conf.addr)
        # When port=0, the OS assigns an available port. Update conf.addr
        # to reflect the actual bound address so callers can use it.
        self.conf.addr = self.s.getsockname()
        self.s.listen(5)
        self.running = False
        self.thread = None
        self.queue = queue.Queue() # report connections and exceptions to client
        self.keep_alive = conf.keep_alive

    def run(self):
        while self.running:
            (sockconn, _) = self.s.accept()
            if self.running:
                conn = Socks5Connection(self, sockconn)
                thread = threading.Thread(None, conn.handle)
                thread.daemon = True
                thread.start()

    def start(self):
        assert not self.running
        self.running = True
        self.thread = threading.Thread(None, self.run)
        self.thread.daemon = True
        self.thread.start()

Review comment (bug):
I checked that this works as expected - without it some ports are below `49152` and with it the selected ports are above (for a couple of dozen tries).

This covers only the port for the Socks5 server itself, but not ports used by the P2P listeners, created with `network_event_loop.create_server()`. Here is an extension to this patch that would cover the later as well:

<details>
<summary>[patch] set port range also for P2P listeners</summary>

```diff
diff --git i/test/functional/test_framework/netutil.py w/test/functional/test_framework/netutil.py
index 792afabff6..8fda0baa31 100644
--- i/test/functional/test_framework/netutil.py
+++ w/test/functional/test_framework/netutil.py
@@ -180,13 +180,13 @@ def format_addr_port(addr, port):
     if ":" in addr:
         return f"[{addr}]:{port}"
     else:
         return f"{addr}:{port}"
 
 
-def set_freebsd_high_port_range(sock):
+def set_ephemeral_port_range(sock):
     '''On FreeBSD, set socket to use the high ephemeral port range (49152-65535).
 
     FreeBSD's default ephemeral port range (10000-65535) overlaps with the test
     framework's static port range (11000-26000). Using IP_PORTRANGE_HIGH avoids
     this overlap when binding to port 0 for dynamic port allocation.
     '''
diff --git i/test/functional/test_framework/p2p.py w/test/functional/test_framework/p2p.py
index 986eaf1e88..b6903a6ba2 100755
--- i/test/functional/test_framework/p2p.py
+++ w/test/functional/test_framework/p2p.py
@@ -19,16 +19,18 @@ P2PDataStore: A p2p interface class that keeps a store of transactions and block
               and can respond correctly to getdata and getheaders messages
 P2PTxInvStore: A p2p interface class that inherits from P2PDataStore, and keeps
               a count of how many times each txid has been announced."""
 
 import asyncio
 from collections import defaultdict
+import ipaddress
 from io import BytesIO
 import logging
 import platform
 import struct
+import socket
 import sys
 import threading
 
 from test_framework.messages import (
     CBlockHeader,
     MAX_HEADERS_RESULTS,
@@ -73,12 +75,15 @@ from test_framework.messages import (
     msg_wtxidrelay,
     NODE_NETWORK,
     NODE_WITNESS,
     MAGIC_BYTES,
     sha256,
 )
+from test_framework.netutil import (
+    set_ephemeral_port_range,
+)
 from test_framework.util import (
     assert_not_equal,
     MAX_NODES,
     p2p_port,
     wait_until_helper_internal,
 )
@@ -744,12 +749,14 @@ class NetworkThread(threading.Thread):
         self.network_event_loop.run_forever()
 
     def close(self, *, timeout=10):
         """Close the connections and network event loop."""
         self.network_event_loop.call_soon_threadsafe(self.network_event_loop.stop)
         wait_until_helper_internal(lambda: not self.network_event_loop.is_running(), timeout=timeout)
+        for listener in NetworkThread.listeners.values():
+            listener.close()
         self.network_event_loop.close()
         self.join(timeout)
         # Safe to remove event loop.
         NetworkThread.network_event_loop = None
 
     @classmethod
@@ -790,14 +797,28 @@ class NetworkThread(threading.Thread):
         if port == 0 or (addr, port) not in cls.listeners:
             # When creating a listener on a given (addr, port) we only need to
             # do it once. If we want different behaviors for different
             # connections, we can accomplish this by providing different
             # `proto` functions
 
-            listener = await cls.network_event_loop.create_server(peer_protocol, addr, port)
-            port = listener.sockets[0].getsockname()[1]
+            if port == 0:
+                # Manually create the socket in order to set the range to be
+                # used for the port before the bind() call.
+                if ipaddress.ip_address(addr).version == 4:
+                    address_family = socket.AF_INET
+                else:
+                    address_family = socket.AF_INET6
+                s = socket.socket(address_family)
+                set_ephemeral_port_range(s)
+                s.bind((addr, 0))
+                s.listen()
+                listener = await cls.network_event_loop.create_server(peer_protocol, sock=s)
+                port = listener.sockets[0].getsockname()[1]
+            else:
+                listener = await cls.network_event_loop.create_server(peer_protocol, addr, port)
+
             logger.debug("Listening server on %s:%d should be started" % (addr, port))
             cls.listeners[(addr, port)] = listener
 
         cls.protos[(addr, port)] = proto
         callback(addr, port)
 
diff --git i/test/functional/test_framework/socks5.py w/test/functional/test_framework/socks5.py
index 4c77a6ff25..085c5a2e32 100644
--- i/test/functional/test_framework/socks5.py
+++ w/test/functional/test_framework/socks5.py
@@ -9,13 +9,13 @@ import socket
 import threading
 import queue
 import logging
 
 from .netutil import (
     format_addr_port,
-    set_freebsd_high_port_range,
+    set_ephemeral_port_range,
 )
 
 logger = logging.getLogger("TestFramework.socks5")
 
 # Protocol constants
 class Command:
@@ -203,13 +203,13 @@ class Socks5Server():
         self.conf = conf
         self.s = socket.socket(conf.af)
         self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         # When using dynamic port allocation (port=0), ensure we don't get a
         # port that conflicts with the test framework's static port range.
         if conf.addr[1] == 0:
-            set_freebsd_high_port_range(self.s)
+            set_ephemeral_port_range(self.s)
         self.s.bind(conf.addr)
         # When port=0, the OS assigns an available port. Update conf.addr
         # to reflect the actual bound address so callers can use it.
         self.conf.addr = self.s.getsockname()
         self.s.listen(5)
         self.running = False
```
</details>

Code after:
                    with socket.create_connection((dest["actual_to_addr"], dest["actual_to_port"])) as conn_to:
                        forward_sockets(self.conn, conn_to)
                else:
                    logger.debug(f"Can't serve the connection to {requested_to}: the destinations factory returned None")
            else:
                logger.debug(f"Can't serve the connection to {requested_to}: no destinations factory")

            # Fall through to disconnect
        except Exception as e:
            logger.exception("socks5 request handling failed.")
            self.serv.queue.put(e)
        finally:
            if not self.serv.keep_alive:
                self.conn.close()
            else:
                logger.debug("Keeping client connection alive")

class Socks5Server():
    def __init__(self, conf):
        self.conf = conf
        self.s = socket.socket(conf.af)
        self.s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        # When using dynamic port allocation (port=0), ensure we don't get a
        # port that conflicts with the test framework's static port range.
        if conf.addr[1] == 0:
            set_ephemeral_port_range(self.s)
        self.s.bind(conf.addr)
        # When port=0, the OS assigns an available port. Update conf.addr
        # to reflect the actual bound address so callers can use it.
        self.conf.addr = self.s.getsockname()
        self.s.listen(5)
        self.running = False
        self.thread = None
        self.queue = queue.Queue() # report connections and exceptions to client
        self.keep_alive = conf.keep_alive

    def run(self):
        while self.running:
            (sockconn, _) = self.s.accept()
            if self.running:
                conn = Socks5Connection(self, sockconn)
                thread = threading.Thread(None, conn.handle)
                thread.daemon = True
                thread.start()

    def start(self):
        assert not self.running
        self.running = True
        self.thread = threading.Thread(None, self.run)
        self.thread.daemon = True
        self.thread.start()