diff --git a/contrib/pyln-testing/pyln/testing/btcproxy.py b/contrib/pyln-testing/pyln/testing/btcproxy.py index 5a24d7bacbff..fc641ace9662 100644 --- a/contrib/pyln-testing/pyln/testing/btcproxy.py +++ b/contrib/pyln-testing/pyln/testing/btcproxy.py @@ -10,7 +10,9 @@ import flask # type: ignore import json import logging +import socket import threading +import time class DecimalEncoder(json.JSONEncoder): @@ -81,17 +83,28 @@ def proxy(self): def start(self): d = PathInfoDispatcher({'/': self.app}) - self.server = Server(('0.0.0.0', self.rpcport), d) + self.server = Server(('127.0.0.1', self.rpcport), d) self.proxy_thread = threading.Thread(target=self.server.start) self.proxy_thread.daemon = True self.proxy_thread.start() - # Now that bitcoind is running on the real rpcport, let's tell all - # future callers to talk to the proxyport. We use the bind_addr as a - # signal that the port is bound and accepting connections. - while self.server.bind_addr[1] == 0: - pass - self.rpcport = self.server.bind_addr[1] + if self.rpcport == 0: + while self.server.bind_addr[1] == 0: + time.sleep(0.01) + self.rpcport = self.server.bind_addr[1] + + deadline = time.time() + 10 + while time.time() < deadline: + try: + s = socket.create_connection(('127.0.0.1', self.rpcport), + timeout=0.5) + s.close() + break + except OSError: + time.sleep(0.05) + else: + raise RuntimeError("BitcoinRpcProxy failed to bind on " + "127.0.0.1:{}".format(self.rpcport)) logging.debug("BitcoinRpcProxy proxying incoming port {} to {}".format(self.rpcport, self.bitcoind.rpcport)) def stop(self): diff --git a/contrib/pyln-testing/pyln/testing/fixtures.py b/contrib/pyln-testing/pyln/testing/fixtures.py index a7eeb0df51b1..e0c181933d88 100644 --- a/contrib/pyln-testing/pyln/testing/fixtures.py +++ b/contrib/pyln-testing/pyln/testing/fixtures.py @@ -193,17 +193,7 @@ def bitcoind(request, directory, teardown_checks): yield bitcoind - try: - bitcoind.stop() - except Exception: - if bitcoind.proc is not None: - bitcoind.proc.kill() - # A test that took the fixture unstarted, or skipped before starting it, - # has nothing to reap here. - if bitcoind.proc is not None: - bitcoind.proc.wait() - - bitcoind.cleanup_files() + bitcoind.kill() class TeardownErrors(object): diff --git a/contrib/pyln-testing/pyln/testing/utils.py b/contrib/pyln-testing/pyln/testing/utils.py index e1e94572f9fe..6ff90eaae858 100644 --- a/contrib/pyln-testing/pyln/testing/utils.py +++ b/contrib/pyln-testing/pyln/testing/utils.py @@ -14,7 +14,8 @@ from pyln.client import Plugin from pyln.client.plugin import PluginLogHandler -import ephemeral_port_reserve # type: ignore +import tempfile +import errno import json import logging import lzma @@ -29,7 +30,6 @@ import string import struct import subprocess -import tempfile import sys import threading import time @@ -173,25 +173,152 @@ def get_tx_p2wsh_outnum(bitcoind, tx, amount): return None -unused_port_lock = threading.Lock() -unused_port_set = set() +_PORT_LOCK_DIR = Path(tempfile.gettempdir()) / "pyln-testing-ports" +_PORT_LOCK_DIR.mkdir(exist_ok=True) + +# If we never hand out this many ports, something is wrong with our +# ephemeral-range detection, and allocating from the OS range is a risky +# fallback we'd rather fail loudly about. +_MIN_RESERVED_PORTS = 2048 + + +def ephemeral_port_range(): + """Return the (lo, hi) of the OS ephemeral source-port range, or None. + + The kernel only ever assigns *source* ports from this range, so a listen + socket bound to a port *outside* it can never collide with an active + OS-assigned connection. Linux/BSD expose it via /proc; macOS via sysctl. + """ + try: + with open('/proc/sys/net/ipv4/ip_local_port_range') as f: + lo, hi = (int(x) for x in f.read().split()) + return lo, hi + except OSError: + pass + try: + import subprocess + lo = int(subprocess.check_output( + ['sysctl', '-n', 'net.inet.ip.portrange.first']).strip()) + hi = int(subprocess.check_output( + ['sysctl', '-n', 'net.inet.ip.portrange.last']).strip()) + return lo, hi + except Exception: + # Unknown platform: assume the Linux default. + return None + + +def _reserved_port_pool(): + """Return (lo, hi) of the port range we may bind. + + We prefer the region strictly below the OS ephemeral floor (which is never + used for source ports, so a test's listen port cannot be stolen by a + *binding* foreign process either, except a deliberate one). If a machine + runs with a very low ephemeral floor we fall back to above the ceiling, + and finally to the ephemeral range itself as a best effort. + """ + rng = ephemeral_port_range() + if rng is None: + lo, hi = 32768, 60999 + else: + lo, hi = rng + + if lo > 1024 and lo - 1024 >= _MIN_RESERVED_PORTS: + return 1024, lo - 1 + if hi < 65535 and 65535 - hi >= _MIN_RESERVED_PORTS: + return hi + 1, 65535 + # No room anywhere else: e.g. an ephemeral range covering everything. + return lo, hi + + +def _port_is_free(port): + """Return True if 127.0.0.1:port can be bound right now. + + This is an actual bind() test (like the old ephemeral_port_reserve), so we + never hand out a port that is genuinely claimed *right now*, as opposed to + merely reserved via lockfile. We use SO_REUSEADDR to match the daemons, so + a lingering TIME_WAIT socket doesn't count as "in use". + """ + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + s.bind(('127.0.0.1', port)) + return True + except OSError as e: + if e.errno != errno.EADDRINUSE: + raise + return False + finally: + s.close() def reserve_unused_port(): - """Get an unused port: avoids handing out the same port unless it's been - returned""" - with unused_port_lock: - while True: - port = ephemeral_port_reserve.reserve() - if port not in unused_port_set: - break - unused_port_set.add(port) + """Get an unused port from the non-ephemeral pool. + + We test-bind the candidate so a port claimed by anything at all right now + is skipped, then take the filesystem lock so our *own* workers never pick + the same genuinely free port concurrently. Note the lockfile only + coordinates our own allocations: the real protection against the OS + stealing a port via source-port assignment is that the pool lies outside + the ephemeral source-port range. + """ + pool_lo, pool_hi = _reserved_port_pool() + while True: + port = random.randint(pool_lo, pool_hi) + if not _port_is_free(port): + continue - return port + lock_path = _PORT_LOCK_DIR / f"{port}.lock" + try: + fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(fd, str(os.getpid()).encode()) + os.close(fd) + return port + except FileExistsError: + continue def drop_unused_port(port): - unused_port_set.remove(port) + if port: + lock_path = _PORT_LOCK_DIR / f"{port}.lock" + lock_path.unlink(missing_ok=True) + + +def cleanup_stale_port_locks(): + """Remove lockfiles whose owning process no longer exists.""" + try: + for lock_path in _PORT_LOCK_DIR.glob("*.lock"): + try: + pid = int(lock_path.read_text()) + try: + os.kill(pid, 0) # signal 0 = existence check, no actual signal + except ProcessLookupError: + lock_path.unlink(missing_ok=True) + except (ValueError, PermissionError, FileNotFoundError): + pass + except Exception: + pass # best-effort, never crash the test run over cleanup + + +def wait_for_port_released(port, timeout=TIMEOUT): + """Wait until 127.0.0.1:port can be bound again. + + A stopped node's connectd holds the listen socket until it exits, + which can be several seconds after lightningd itself is gone + (subdaemons are separate processes, and die slowly under valgrind). + Restarting the node before the port is released makes the new + connectd fail with 'Address already in use'. + """ + start_time = time.time() + while not _port_is_free(port): + if time.time() - start_time > timeout: + raise TimeoutError( + "Port {} was not released within {} seconds" + .format(port, timeout)) + time.sleep(0.1) + + waited = time.time() - start_time + if waited >= 1: + logging.info("Port %d took %.1fs to be released", port, waited) class TailableProc(object): @@ -292,6 +419,8 @@ def kill(self): def cleanup_files(self): """Ensure files are closed.""" + cleanup_stale_port_locks() + for f in ["stdout_write", "stderr_write", "stdout_read", "stderr_read"]: try: getattr(self, f).close() @@ -461,10 +590,7 @@ def __init__(self, bitcoin_dir="/tmp/bitcoind-test", rpcport=None): TailableProc.__init__(self, bitcoin_dir, verbose=False) if rpcport is None: - self.reserved_rpcport = reserve_unused_port() - rpcport = self.reserved_rpcport - else: - self.reserved_rpcport = None + rpcport = reserve_unused_port() self.bitcoin_dir = bitcoin_dir self.rpcport = rpcport @@ -505,9 +631,22 @@ def __init__(self, bitcoin_dir="/tmp/bitcoind-test", rpcport=None): self.rpc = SimpleBitcoinProxy(btc_conf_file=self.conf_file) self.proxies = [] - def __del__(self): - if self.reserved_rpcport is not None: - drop_unused_port(self.reserved_rpcport) + def kill(self): + # Never started: parametrize('bitcoind', [False], indirect=True). + if self.proc is None: + self.cleanup_files() + return + + try: + self.stop() + except Exception: + self.proc.kill() + self.proc.wait() + + self.cleanup_files() + drop_unused_port(self.rpcport) + for p in self.proxies: + drop_unused_port(p.rpcport) def blake2b_activation_args(self): """The argument which activates this chain's rules, if any. @@ -550,7 +689,7 @@ def stop(self): return TailableProc.stop(self) def get_proxy(self): - proxy = BitcoinRpcProxy(self) + proxy = BitcoinRpcProxy(self, rpcport=reserve_unused_port()) self.proxies.append(proxy) proxy.start() return proxy @@ -840,6 +979,9 @@ def cmd_line(self): def start(self, stdin=None, wait_for_initialized=True, stderr_redir=False): self.opts['bitcoin-rpcport'] = self.rpcproxy.rpcport + # On restart, the previous incarnation's connectd may still be + # dying and holding our listen port: don't launch until it's free. + wait_for_port_released(self.port) TailableProc.start(self, stdin, stdout_redir=False, stderr_redir=stderr_redir) if wait_for_initialized: self.wait_for_log("Server started with public key")