diff --git a/src/akkudoktoreos/server/server.py b/src/akkudoktoreos/server/server.py index db8fe625..5abfc4eb 100644 --- a/src/akkudoktoreos/server/server.py +++ b/src/akkudoktoreos/server/server.py @@ -1,5 +1,6 @@ """Server Module.""" +import errno import ipaddress import os import re @@ -94,79 +95,116 @@ def validate_ip_or_hostname(value: str) -> str: return value -def wait_for_port_free(port: int, timeout: int = 0, waiting_app_name: str = "App") -> bool: - """Wait for a network port to become free, with timeout. +def _is_port_available(port: int) -> bool: + """Check TCP binding on local IPv4 and IPv6 addresses without inspecting PIDs.""" + addresses = [(socket.AF_INET, "")] + if socket.has_ipv6: + addresses.append((socket.AF_INET6, "::")) - Checks if the port is currently in use and logs warnings with process details. - Retries every 3 seconds until timeout is reached. + # Reuse avoids waiting for TIME_WAIT connections. On macOS, a wildcard bind + # with reuse can coexist with a listener on a specific interface, so probe + # each local address too. Interface enumeration does not inspect processes. + try: + interface_addresses = [ + (address.family, address.address) + for interface in psutil.net_if_addrs().values() + for address in interface + if address.family == socket.AF_INET + or (socket.has_ipv6 and address.family == socket.AF_INET6) + ] + except (psutil.Error, OSError): + interface_addresses = [] + addresses.extend(interface_addresses) + + for family, address in dict.fromkeys(addresses): + try: + with socket.socket(family, socket.SOCK_STREAM) as probe: + # Fall back to exclusive probes if interface details are unavailable. + if os.name != "nt" and interface_addresses: + probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + exclusive = getattr(socket, "SO_EXCLUSIVEADDRUSE", None) + if os.name == "nt" and exclusive is not None: + probe.setsockopt(socket.SOL_SOCKET, exclusive, 1) + if family == socket.AF_INET6: + probe.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) + # Preserve the scope ID required for link-local IPv6 binds. + sockaddr = socket.getaddrinfo( + address, port, family, socket.SOCK_STREAM, 0, socket.AI_NUMERICHOST + )[0][4] + probe.bind(sockaddr) + else: + probe.bind((address, port)) + except OSError as error: + if error.errno in (errno.EADDRINUSE, errno.EACCES): + return False + if address not in ("", "::") and error.errno == errno.EADDRNOTAVAIL: + # A local interface may disappear after enumeration. + continue + if family == socket.AF_INET6 and error.errno in ( + errno.EAFNOSUPPORT, + errno.EPROTONOSUPPORT, + errno.EADDRNOTAVAIL, + ): + # Python may support IPv6 even when it is disabled on this host. + continue + raise + return True + + +def wait_for_port_free(port: int, timeout: int = 0, waiting_app_name: str = "App") -> bool: + """Wait for a TCP port to become available for binding, with a bounded timeout. + + Probe IPv4 and supported IPv6 sockets, retrying at most every three seconds. + Process details are optional diagnostics when the port remains unavailable. Args: - port: The network port number to check - timeout: Maximum seconds to wait (0 means check once without waiting) - waiting_app_name: Name of the application waiting for the port + port: The network port number to check. + timeout: Maximum seconds to wait (0 means check once without waiting). + waiting_app_name: Name of the application waiting for the port. Returns: - bool: True if port is free, False if port is still in use after timeout + True if the port can be bound, False if it remains unavailable. Raises: - ValueError: If port number or timeout is invalid - psutil.Error: If there are problems accessing process information + ValueError: If the port number or timeout is invalid. + OSError: If socket probing fails for reasons other than an unavailable port + or unsupported IPv6. """ if not 0 <= port <= 65535: raise ValueError(f"Invalid port number: {port}") if timeout < 0: raise ValueError(f"Invalid timeout: {timeout}") - def get_processes_using_port() -> list[dict]: - """Get info about processes using the specified port.""" - processes: list[dict] = [] - seen_pids: set[int] = set() - - try: - for conn in psutil.net_connections(kind="inet"): - if ( - conn.laddr - and conn.laddr.port == port - and conn.pid is not None - and conn.pid not in seen_pids - ): - try: - process = psutil.Process(conn.pid) - seen_pids.add(conn.pid) - processes.append(process.as_dict(attrs=["pid", "cmdline"])) - except psutil.NoSuchProcess: - continue - except psutil.Error as e: - logger.error(f"Error checking port {port}: {e}") - raise - - return processes - - retries = max(int(timeout / 3), 1) if timeout > 0 else 1 - - for _ in range(retries): - process_info = get_processes_using_port() - - if not process_info: + deadline = time.monotonic() + timeout + while True: + if _is_port_available(port): return True - - if timeout <= 0: + remaining = deadline - time.monotonic() + if remaining <= 0: break - logger.info(f"{waiting_app_name} waiting for port {port} to become free...") - time.sleep(3) + time.sleep(min(3.0, remaining)) - if process_info: - logger.warning( - f"{waiting_app_name} port {port} still in use after waiting {timeout} seconds." - ) - for info in process_info: - logger.warning( - f"Process using port - PID: {info['pid']}, Command: {' '.join(info['cmdline'])}" - ) + logger.warning(f"{waiting_app_name} port {port} still in use after waiting {timeout} seconds.") + try: + connections = psutil.net_connections(kind="inet") + except (psutil.Error, OSError) as error: + logger.debug(f"Process details unavailable for port {port}: {error}") return False - return True + seen_pids: set[int] = set() + for conn in connections: + if not conn.laddr or conn.laddr.port != port or conn.pid is None or conn.pid in seen_pids: + continue + seen_pids.add(conn.pid) + try: + process = psutil.Process(conn.pid) + cmdline = process.cmdline() + except (psutil.Error, OSError): + # Protected or disappearing processes do not change the port result. + continue + logger.warning(f"Process using port - PID: {conn.pid}, Command: {' '.join(cmdline)}") + return False def drop_root_privileges(run_as_user: Optional[str] = None) -> bool: diff --git a/tests/test_serverports.py b/tests/test_serverports.py new file mode 100644 index 00000000..5c276277 --- /dev/null +++ b/tests/test_serverports.py @@ -0,0 +1,284 @@ +import errno +import socket +from collections.abc import Generator +from types import SimpleNamespace +from unittest.mock import Mock + +import psutil +import pytest + +from akkudoktoreos.server import server + + +@pytest.fixture +def port_environment(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + """Provide socket probes, optional process diagnostics, and a deterministic clock.""" + probe = Mock() + probe.__enter__ = Mock(return_value=probe) + probe.__exit__ = Mock(return_value=False) + factory = Mock(return_value=probe) + connections = Mock(side_effect=psutil.AccessDenied(1)) + clock = [0.0] + + def sleep(seconds: float) -> None: + clock[0] += seconds + + sleep_mock = Mock(side_effect=sleep) + monkeypatch.setattr(server.socket, "socket", factory) + monkeypatch.setattr(server.socket, "has_ipv6", False) + monkeypatch.setattr(server.psutil, "net_if_addrs", Mock(return_value={})) + monkeypatch.setattr(server.psutil, "net_connections", connections) + monkeypatch.setattr(server.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr(server.time, "sleep", sleep_mock) + return SimpleNamespace( + probe=probe, factory=factory, connections=connections, clock=clock, sleep=sleep_mock + ) + + +def test_free_port_does_not_require_system_process_inspection( + port_environment: SimpleNamespace, +) -> None: + env = port_environment + + assert server.wait_for_port_free(8503) + + env.connections.assert_not_called() + env.sleep.assert_not_called() + env.probe.__exit__.assert_called_once() + + +@pytest.mark.parametrize("error", [psutil.AccessDenied(1), PermissionError(1, "Denied")]) +def test_occupied_port_remains_occupied_when_diagnostics_are_denied( + port_environment: SimpleNamespace, error: Exception +) -> None: + env = port_environment + env.probe.bind.side_effect = OSError(errno.EADDRINUSE, "Already bound") + env.connections.side_effect = error + + assert not server.wait_for_port_free(8503) + + env.connections.assert_called_once_with(kind="inet") + env.sleep.assert_not_called() + + +@pytest.mark.parametrize("timeout", [1, 5, 6]) +def test_port_wait_respects_timeout_with_denied_diagnostics( + port_environment: SimpleNamespace, timeout: int +) -> None: + env = port_environment + env.probe.bind.side_effect = OSError(errno.EADDRINUSE, "Already bound") + + assert not server.wait_for_port_free(8503, timeout=timeout) + + assert env.clock[0] == timeout + assert all(call.args[0] <= 3 for call in env.sleep.call_args_list) + env.connections.assert_called_once_with(kind="inet") + + +def test_port_becomes_available_during_wait(port_environment: SimpleNamespace) -> None: + env = port_environment + env.probe.bind.side_effect = [OSError(errno.EADDRINUSE, "Already bound"), None] + + assert server.wait_for_port_free(8503, timeout=5) + + assert env.clock[0] == 3 + env.connections.assert_not_called() + + +@pytest.mark.parametrize("error", [psutil.AccessDenied(101), psutil.NoSuchProcess(101)]) +def test_diagnostic_process_errors_do_not_change_occupied_result( + port_environment: SimpleNamespace, monkeypatch: pytest.MonkeyPatch, error: Exception +) -> None: + env = port_environment + env.probe.bind.side_effect = OSError(errno.EADDRINUSE, "Already bound") + env.connections.side_effect = None + env.connections.return_value = [ + SimpleNamespace(pid=101, laddr=SimpleNamespace(port=8503)), + SimpleNamespace(pid=102, laddr=SimpleNamespace(port=8503)), + SimpleNamespace(pid=102, laddr=SimpleNamespace(port=8503)), + SimpleNamespace(pid=103, laddr=SimpleNamespace(port=9000)), + SimpleNamespace(pid=None, laddr=SimpleNamespace(port=8503)), + ] + inaccessible = Mock() + inaccessible.cmdline.side_effect = error + accessible = Mock() + accessible.cmdline.return_value = ["python", "-m", "akkudoktoreos.server.eos"] + process = Mock(side_effect=[inaccessible, accessible]) + monkeypatch.setattr(server.psutil, "Process", process) + + assert not server.wait_for_port_free(8503) + + assert [call.args[0] for call in process.call_args_list] == [101, 102] + accessible.cmdline.assert_called_once_with() + + +def test_port_with_unknown_owner_is_still_occupied(port_environment: SimpleNamespace) -> None: + env = port_environment + env.probe.bind.side_effect = OSError(errno.EADDRINUSE, "Already bound") + env.connections.side_effect = None + env.connections.return_value = [SimpleNamespace(pid=None, laddr=SimpleNamespace(port=8503))] + + assert not server.wait_for_port_free(8503) + + +def test_ipv6_listener_is_detected_after_free_ipv4_probe( + port_environment: SimpleNamespace, monkeypatch: pytest.MonkeyPatch +) -> None: + env = port_environment + monkeypatch.setattr(server.socket, "has_ipv6", True) + env.probe.bind.side_effect = [None, OSError(errno.EADDRINUSE, "IPv6 listener")] + + assert not server.wait_for_port_free(8503) + + assert [call.args[0] for call in env.factory.call_args_list] == [ + socket.AF_INET, + socket.AF_INET6, + ] + + +@pytest.mark.parametrize("scope", ["eth0", "3"]) +@pytest.mark.parametrize("occupied", [False, True]) +def test_scoped_ipv6_port_probe_preserves_interface_id( + port_environment: SimpleNamespace, + monkeypatch: pytest.MonkeyPatch, + scope: str, + occupied: bool, +) -> None: + """Linux requires an explicit scope ID when binding link-local IPv6 addresses.""" + env = port_environment + address = f"fe80::1%{scope}" + scoped_sockaddr = ("fe80::1", 8503, 0, 3) + monkeypatch.setattr(server.socket, "has_ipv6", True) + monkeypatch.setattr( + server.psutil, + "net_if_addrs", + Mock(return_value={"eth0": [SimpleNamespace(family=socket.AF_INET6, address=address)]}), + ) + resolver = Mock( + side_effect=[ + [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("::", 8503, 0, 0))], + [(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", scoped_sockaddr)], + ] + ) + monkeypatch.setattr(server.socket, "getaddrinfo", resolver) + + def bind(sockaddr: tuple[str, int] | tuple[str, int, int, int]) -> None: + if sockaddr[0].startswith("fe80:"): + if len(sockaddr) != 4 or sockaddr[3] != 3: + raise OSError(errno.EINVAL, "Missing IPv6 scope ID") + if occupied: + raise OSError(errno.EADDRINUSE, "IPv6 listener") + + env.probe.bind.side_effect = bind + + assert server.wait_for_port_free(8503) is not occupied + + env.probe.bind.assert_any_call(scoped_sockaddr) + resolver.assert_called_with( + address, 8503, socket.AF_INET6, socket.SOCK_STREAM, 0, socket.AI_NUMERICHOST + ) + + +@pytest.mark.parametrize( + "error_number", [errno.EAFNOSUPPORT, errno.EPROTONOSUPPORT, errno.EADDRNOTAVAIL] +) +def test_disabled_ipv6_does_not_block_free_ipv4_port( + port_environment: SimpleNamespace, monkeypatch: pytest.MonkeyPatch, error_number: int +) -> None: + env = port_environment + monkeypatch.setattr(server.socket, "has_ipv6", True) + env.factory.side_effect = [env.probe, OSError(error_number, "IPv6 unavailable")] + + assert server.wait_for_port_free(8503) + + +def test_socket_resource_error_is_not_reported_as_free(port_environment: SimpleNamespace) -> None: + port_environment.factory.side_effect = OSError(errno.EMFILE, "No file descriptors") + + with pytest.raises(OSError) as error: + server.wait_for_port_free(8503) + + assert error.value.errno == errno.EMFILE + + +def test_bind_permission_denied_is_not_reported_as_free(port_environment: SimpleNamespace) -> None: + port_environment.probe.bind.side_effect = OSError(errno.EACCES, "Cannot bind") + + assert not server.wait_for_port_free(8503) + + +@pytest.mark.parametrize("port,timeout", [(-1, 0), (65536, 0), (8503, -1)]) +def test_port_wait_validates_arguments( + port_environment: SimpleNamespace, port: int, timeout: int +) -> None: + with pytest.raises(ValueError): + server.wait_for_port_free(port, timeout=timeout) + + port_environment.factory.assert_not_called() + + +@pytest.fixture(params=[socket.AF_INET, socket.AF_INET6]) +def listener(request: pytest.FixtureRequest) -> Generator[socket.socket, None, None]: + """Use real sockets to verify both protocol families as an ordinary user.""" + family = request.param + try: + sock = socket.socket(family, socket.SOCK_STREAM) + except OSError as error: + if family == socket.AF_INET6 and error.errno in (errno.EAFNOSUPPORT, errno.EPROTONOSUPPORT): + pytest.skip("IPv6 unavailable") + raise + with sock: + if family == socket.AF_INET6: + sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind(("::1" if family == socket.AF_INET6 else "127.0.0.1", 0)) + except OSError as error: + if family == socket.AF_INET6 and error.errno == errno.EADDRNOTAVAIL: + pytest.skip("IPv6 loopback unavailable") + raise + sock.listen() + yield sock + + +def test_real_port_availability_with_denied_enumeration( + listener: socket.socket, monkeypatch: pytest.MonkeyPatch +) -> None: + port = listener.getsockname()[1] + monkeypatch.setattr(server.psutil, "net_connections", Mock(side_effect=psutil.AccessDenied(1))) + + assert not server.wait_for_port_free(port) + # The probe must leave the existing listener open and untouched. + assert listener.getsockname()[1] == port + listener.close() + assert server.wait_for_port_free(port) + + +def test_closed_tcp_connection_does_not_delay_port_reuse( + listener: socket.socket, monkeypatch: pytest.MonkeyPatch +) -> None: + """A server can restart while its previous connections remain in TIME_WAIT.""" + if server.os.name == "nt": + pytest.skip("Windows uses exclusive server sockets") + port = listener.getsockname()[1] + monkeypatch.setattr(server.psutil, "net_connections", Mock(side_effect=psutil.AccessDenied(1))) + with socket.socket(listener.family, socket.SOCK_STREAM) as client: + client.connect(listener.getsockname()) + accepted, _ = listener.accept() + accepted.close() + assert client.recv(1) == b"" + listener.close() + + assert server.wait_for_port_free(port) + + +def test_missing_interface_details_use_exclusive_probe( + port_environment: SimpleNamespace, monkeypatch: pytest.MonkeyPatch +) -> None: + env = port_environment + monkeypatch.setattr(server.psutil, "net_if_addrs", Mock(side_effect=psutil.AccessDenied())) + env.probe.bind.side_effect = OSError(errno.EADDRINUSE, "Already bound") + + assert not server.wait_for_port_free(8503) + + assert all(call.args[1] != socket.SO_REUSEADDR for call in env.probe.setsockopt.call_args_list)