diff --git a/examples/advanced/network_discover.py b/examples/advanced/network_discover.py index 71edd209..13f7d03a 100644 --- a/examples/advanced/network_discover.py +++ b/examples/advanced/network_discover.py @@ -14,6 +14,7 @@ try: expand_wildcard_address, get_available_interfaces, get_optimal_binding_address, + get_wildcard_address, ) except ImportError: # Fallbacks if utilities are missing @@ -29,6 +30,9 @@ except ImportError: def get_optimal_binding_address(port: int, protocol: str = "tcp"): return Multiaddr(f"/ip4/127.0.0.1/{protocol}/{port}") + def get_wildcard_address(port: int, protocol: str = "tcp"): + return Multiaddr(f"/ip4/0.0.0.0/{protocol}/{port}") + def main() -> None: port = 8080 @@ -37,7 +41,10 @@ def main() -> None: for a in interfaces: print(f" - {a}") - wildcard_v4 = Multiaddr(f"/ip4/0.0.0.0/tcp/{port}") + # Demonstrate wildcard address as a feature + wildcard_v4 = get_wildcard_address(port) + print(f"\nWildcard address (feature): {wildcard_v4}") + expanded_v4 = expand_wildcard_address(wildcard_v4) print("\nExpanded IPv4 wildcard:") for a in expanded_v4: diff --git a/examples/bootstrap/bootstrap.py b/examples/bootstrap/bootstrap.py index 825f3a08..b4fa9234 100644 --- a/examples/bootstrap/bootstrap.py +++ b/examples/bootstrap/bootstrap.py @@ -53,7 +53,11 @@ BOOTSTRAP_PEERS = [ async def run(port: int, bootstrap_addrs: list[str]) -> None: """Run the bootstrap discovery example.""" - from libp2p.utils.address_validation import find_free_port, get_available_interfaces + from libp2p.utils.address_validation import ( + find_free_port, + get_available_interfaces, + get_optimal_binding_address, + ) if port <= 0: port = find_free_port() @@ -93,6 +97,12 @@ async def run(port: int, bootstrap_addrs: list[str]) -> None: logger.info(f"{addr}") print(f"{addr}") + # Display optimal address for reference + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host.get_id().to_string()}" + logger.info(f"Optimal address: {optimal_addr_with_peer}") + print(f"Optimal address: {optimal_addr_with_peer}") + # Keep running and log peer discovery events await trio.sleep_forever() except KeyboardInterrupt: diff --git a/examples/chat/chat.py b/examples/chat/chat.py index 35f98d25..ee133af1 100755 --- a/examples/chat/chat.py +++ b/examples/chat/chat.py @@ -46,7 +46,11 @@ async def write_data(stream: INetStream) -> None: async def run(port: int, destination: str) -> None: - from libp2p.utils.address_validation import find_free_port, get_available_interfaces + from libp2p.utils.address_validation import ( + find_free_port, + get_available_interfaces, + get_optimal_binding_address, + ) if port <= 0: port = find_free_port() @@ -72,11 +76,12 @@ async def run(port: int, destination: str) -> None: for addr in all_addrs: print(f"{addr}") - # Use the first address as the default for the client command - default_addr = all_addrs[0] + # Use optimal address for the client command + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host.get_id().to_string()}" print( f"\nRun this from the same folder in another console:\n\n" - f"chat-demo -d {default_addr}\n" + f"chat-demo -d {optimal_addr_with_peer}\n" ) print("Waiting for incoming connection...") diff --git a/examples/doc-examples/example_multiplexer.py b/examples/doc-examples/example_multiplexer.py index 6963ace0..63a29fc5 100644 --- a/examples/doc-examples/example_multiplexer.py +++ b/examples/doc-examples/example_multiplexer.py @@ -1,6 +1,5 @@ import secrets -import multiaddr import trio from libp2p import ( @@ -13,6 +12,10 @@ from libp2p.security.noise.transport import ( PROTOCOL_ID as NOISE_PROTOCOL_ID, Transport as NoiseTransport, ) +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) async def main(): @@ -39,14 +42,16 @@ async def main(): # Create a host with the key pair, Noise security, and mplex multiplexer host = new_host(key_pair=key_pair, sec_opt=security_options) - # Configure the listening address + # Configure the listening address using the new paradigm port = 8000 - listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}") + listen_addrs = get_available_interfaces(port) + optimal_addr = get_optimal_binding_address(port) # Start the host - async with host.run(listen_addrs=[listen_addr]): + async with host.run(listen_addrs=listen_addrs): print("libp2p has started with Noise encryption and mplex multiplexing") print("libp2p is listening on:", host.get_addrs()) + print(f"Optimal address: {optimal_addr}") # Keep the host running await trio.sleep_forever() diff --git a/examples/doc-examples/example_net_stream.py b/examples/doc-examples/example_net_stream.py index a77a7509..6f7eb4b0 100644 --- a/examples/doc-examples/example_net_stream.py +++ b/examples/doc-examples/example_net_stream.py @@ -38,6 +38,10 @@ from libp2p.network.stream.net_stream import ( from libp2p.peer.peerinfo import ( info_from_p2p_addr, ) +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) PROTOCOL_ID = TProtocol("/echo/1.0.0") @@ -173,7 +177,9 @@ async def run_enhanced_demo( """ Run enhanced echo demo with NetStream state management. """ - listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}") + # Use the new address paradigm + listen_addrs = get_available_interfaces(port) + optimal_addr = get_optimal_binding_address(port) # Generate or use provided key if seed: @@ -185,7 +191,7 @@ async def run_enhanced_demo( host = new_host(key_pair=create_new_key_pair(secret)) - async with host.run(listen_addrs=[listen_addr]): + async with host.run(listen_addrs=listen_addrs): print(f"Host ID: {host.get_id().to_string()}") print("=" * 60) @@ -196,10 +202,12 @@ async def run_enhanced_demo( # type: ignore: Stream is type of NetStream host.set_stream_handler(PROTOCOL_ID, enhanced_echo_handler) + # Use optimal address for client command + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host.get_id().to_string()}" print( "Run client from another console:\n" f"python3 example_net_stream.py " - f"-d {host.get_addrs()[0]}\n" + f"-d {optimal_addr_with_peer}\n" ) print("Waiting for connections...") print("Press Ctrl+C to stop server") diff --git a/examples/doc-examples/example_running.py b/examples/doc-examples/example_running.py index 7f3ade32..2f495979 100644 --- a/examples/doc-examples/example_running.py +++ b/examples/doc-examples/example_running.py @@ -1,6 +1,5 @@ import secrets -import multiaddr import trio from libp2p import ( @@ -13,6 +12,10 @@ from libp2p.security.noise.transport import ( PROTOCOL_ID as NOISE_PROTOCOL_ID, Transport as NoiseTransport, ) +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) async def main(): @@ -39,14 +42,16 @@ async def main(): # Create a host with the key pair, Noise security, and mplex multiplexer host = new_host(key_pair=key_pair, sec_opt=security_options) - # Configure the listening address + # Configure the listening address using the new paradigm port = 8000 - listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}") + listen_addrs = get_available_interfaces(port) + optimal_addr = get_optimal_binding_address(port) # Start the host - async with host.run(listen_addrs=[listen_addr]): + async with host.run(listen_addrs=listen_addrs): print("libp2p has started") print("libp2p is listening on:", host.get_addrs()) + print(f"Optimal address: {optimal_addr}") # Keep the host running await trio.sleep_forever() diff --git a/examples/doc-examples/example_transport.py b/examples/doc-examples/example_transport.py index 8f4c9fa1..9d29d457 100644 --- a/examples/doc-examples/example_transport.py +++ b/examples/doc-examples/example_transport.py @@ -1,6 +1,5 @@ import secrets -import multiaddr import trio from libp2p import ( @@ -9,6 +8,10 @@ from libp2p import ( from libp2p.crypto.secp256k1 import ( create_new_key_pair, ) +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) async def main(): @@ -19,14 +22,16 @@ async def main(): # Create a host with the key pair host = new_host(key_pair=key_pair) - # Configure the listening address + # Configure the listening address using the new paradigm port = 8000 - listen_addr = multiaddr.Multiaddr(f"/ip4/127.0.0.1/tcp/{port}") + listen_addrs = get_available_interfaces(port) + optimal_addr = get_optimal_binding_address(port) # Start the host - async with host.run(listen_addrs=[listen_addr]): + async with host.run(listen_addrs=listen_addrs): print("libp2p has started with TCP transport") print("libp2p is listening on:", host.get_addrs()) + print(f"Optimal address: {optimal_addr}") # Keep the host running await trio.sleep_forever() diff --git a/examples/echo/echo.py b/examples/echo/echo.py index 42e3ff0c..d998f6e8 100644 --- a/examples/echo/echo.py +++ b/examples/echo/echo.py @@ -27,6 +27,7 @@ from libp2p.peer.peerinfo import ( from libp2p.utils.address_validation import ( find_free_port, get_available_interfaces, + get_optimal_binding_address, ) # Configure minimal logging @@ -82,9 +83,13 @@ async def run(port: int, destination: str, seed: int | None = None) -> None: for addr in listen_addr: print(f"{addr}/p2p/{peer_id}") + # Get optimal address for display + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{peer_id}" + print( "\nRun this from the same folder in another console:\n\n" - f"echo-demo -d {host.get_addrs()[0]}\n" + f"echo-demo -d {optimal_addr_with_peer}\n" ) print("Waiting for incoming connections...") await trio.sleep_forever() diff --git a/examples/identify/identify.py b/examples/identify/identify.py index bd973a3e..addfff89 100644 --- a/examples/identify/identify.py +++ b/examples/identify/identify.py @@ -63,7 +63,10 @@ def print_identify_response(identify_response: Identify): async def run(port: int, destination: str, use_varint_format: bool = True) -> None: - from libp2p.utils.address_validation import get_available_interfaces + from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, + ) if not destination: # Create first host (listener) @@ -100,11 +103,12 @@ async def run(port: int, destination: str, use_varint_format: bool = True) -> No for addr in all_addrs: print(f"{addr}") - # Use the first address as the default for the client command - default_addr = all_addrs[0] + # Use optimal address for the client command + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host_a.get_id().to_string()}" print( f"\nRun this from the same folder in another console:\n\n" - f"identify-demo {format_flag} -d {default_addr}\n" + f"identify-demo {format_flag} -d {optimal_addr_with_peer}\n" ) print("Waiting for incoming identify request...") @@ -152,6 +156,7 @@ async def run(port: int, destination: str, use_varint_format: bool = True) -> No from libp2p.utils.address_validation import ( find_free_port, get_available_interfaces, + get_optimal_binding_address, ) if port <= 0: diff --git a/examples/kademlia/kademlia.py b/examples/kademlia/kademlia.py index 80bbc995..cf4b2988 100644 --- a/examples/kademlia/kademlia.py +++ b/examples/kademlia/kademlia.py @@ -151,7 +151,10 @@ async def run_node( key_pair = create_new_key_pair(secrets.token_bytes(32)) host = new_host(key_pair=key_pair) - from libp2p.utils.address_validation import get_available_interfaces + from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, + ) listen_addrs = get_available_interfaces(port) @@ -168,9 +171,10 @@ async def run_node( for addr in all_addrs: logger.info(f"{addr}") - # Use the first address as the default for the bootstrap command - default_addr = all_addrs[0] - bootstrap_cmd = f"--bootstrap {default_addr}" + # Use optimal address for the bootstrap command + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host.get_id().to_string()}" + bootstrap_cmd = f"--bootstrap {optimal_addr_with_peer}" logger.info("To connect to this node, use: %s", bootstrap_cmd) await connect_to_bootstrap_nodes(host, bootstrap_nodes) @@ -182,7 +186,7 @@ async def run_node( # Save server address in server mode if dht_mode == DHTMode.SERVER: - save_server_addr(str(default_addr)) + save_server_addr(str(optimal_addr_with_peer)) # Start the DHT service async with background_trio_service(dht): diff --git a/examples/ping/ping.py b/examples/ping/ping.py index 52bb759a..5c7f54e4 100644 --- a/examples/ping/ping.py +++ b/examples/ping/ping.py @@ -61,7 +61,11 @@ async def send_ping(stream: INetStream) -> None: async def run(port: int, destination: str) -> None: - from libp2p.utils.address_validation import find_free_port, get_available_interfaces + from libp2p.utils.address_validation import ( + find_free_port, + get_available_interfaces, + get_optimal_binding_address, + ) if port <= 0: port = find_free_port() @@ -83,11 +87,12 @@ async def run(port: int, destination: str) -> None: for addr in all_addrs: print(f"{addr}") - # Use the first address as the default for the client command - default_addr = all_addrs[0] + # Use optimal address for the client command + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = f"{optimal_addr}/p2p/{host.get_id().to_string()}" print( f"\nRun this from the same folder in another console:\n\n" - f"ping-demo -d {default_addr}\n" + f"ping-demo -d {optimal_addr_with_peer}\n" ) print("Waiting for incoming connection...") diff --git a/examples/pubsub/pubsub.py b/examples/pubsub/pubsub.py index 6e8495c1..adb3a1d0 100644 --- a/examples/pubsub/pubsub.py +++ b/examples/pubsub/pubsub.py @@ -102,7 +102,10 @@ async def monitor_peer_topics(pubsub, nursery, termination_event): async def run(topic: str, destination: str | None, port: int | None) -> None: - from libp2p.utils.address_validation import get_available_interfaces + from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, + ) if port is None or port == 0: port = find_free_port() @@ -162,11 +165,14 @@ async def run(topic: str, destination: str | None, port: int | None) -> None: for addr in all_addrs: logger.info(f"{addr}") - # Use the first address as the default for the client command - default_addr = all_addrs[0] + # Use optimal address for the client command + optimal_addr = get_optimal_binding_address(port) + optimal_addr_with_peer = ( + f"{optimal_addr}/p2p/{host.get_id().to_string()}" + ) logger.info( f"\nRun this from the same folder in another console:\n\n" - f"pubsub-demo -d {default_addr}\n" + f"pubsub-demo -d {optimal_addr_with_peer}\n" ) logger.info("Waiting for peers...") diff --git a/libp2p/utils/address_validation.py b/libp2p/utils/address_validation.py index a470ad24..5ce58671 100644 --- a/libp2p/utils/address_validation.py +++ b/libp2p/utils/address_validation.py @@ -3,38 +3,24 @@ from __future__ import annotations import socket from multiaddr import Multiaddr - -try: - from multiaddr.utils import ( # type: ignore - get_network_addrs, - get_thin_waist_addresses, - ) - - _HAS_THIN_WAIST = True -except ImportError: # pragma: no cover - only executed in older environments - _HAS_THIN_WAIST = False - get_thin_waist_addresses = None # type: ignore - get_network_addrs = None # type: ignore +from multiaddr.utils import get_network_addrs, get_thin_waist_addresses def _safe_get_network_addrs(ip_version: int) -> list[str]: """ Internal safe wrapper. Returns a list of IP addresses for the requested IP version. - Falls back to minimal defaults when Thin Waist helpers are missing. :param ip_version: 4 or 6 """ - if _HAS_THIN_WAIST and get_network_addrs: - try: - return get_network_addrs(ip_version) or [] - except Exception: # pragma: no cover - defensive - return [] - # Fallback behavior (very conservative) - if ip_version == 4: - return ["127.0.0.1"] - if ip_version == 6: - return ["::1"] - return [] + try: + return get_network_addrs(ip_version) or [] + except Exception: # pragma: no cover - defensive + # Fallback behavior (very conservative) + if ip_version == 4: + return ["127.0.0.1"] + if ip_version == 6: + return ["::1"] + return [] def find_free_port() -> int: @@ -47,16 +33,13 @@ def find_free_port() -> int: def _safe_expand(addr: Multiaddr, port: int | None = None) -> list[Multiaddr]: """ Internal safe expansion wrapper. Returns a list of Multiaddr objects. - If Thin Waist isn't available, returns [addr] (identity). """ - if _HAS_THIN_WAIST and get_thin_waist_addresses: - try: - if port is not None: - return get_thin_waist_addresses(addr, port=port) or [] - return get_thin_waist_addresses(addr) or [] - except Exception: # pragma: no cover - defensive - return [addr] - return [addr] + try: + if port is not None: + return get_thin_waist_addresses(addr, port=port) or [] + return get_thin_waist_addresses(addr) or [] + except Exception: # pragma: no cover - defensive + return [addr] def get_available_interfaces(port: int, protocol: str = "tcp") -> list[Multiaddr]: @@ -122,6 +105,20 @@ def expand_wildcard_address( return expanded +def get_wildcard_address(port: int, protocol: str = "tcp") -> Multiaddr: + """ + Get wildcard address (0.0.0.0) when explicitly needed. + + This function provides access to wildcard binding as a feature when + explicitly required, preserving the ability to bind to all interfaces. + + :param port: Port number. + :param protocol: Transport protocol. + :return: A Multiaddr with wildcard binding (0.0.0.0). + """ + return Multiaddr(f"/ip4/0.0.0.0/{protocol}/{port}") + + def get_optimal_binding_address(port: int, protocol: str = "tcp") -> Multiaddr: """ Choose an optimal address for an example to bind to: @@ -157,6 +154,7 @@ def get_optimal_binding_address(port: int, protocol: str = "tcp") -> Multiaddr: __all__ = [ "get_available_interfaces", "get_optimal_binding_address", + "get_wildcard_address", "expand_wildcard_address", "find_free_port", ] diff --git a/tests/examples/test_examples_bind_address.py b/tests/examples/test_examples_bind_address.py index 1045c90b..c0dd9de3 100644 --- a/tests/examples/test_examples_bind_address.py +++ b/tests/examples/test_examples_bind_address.py @@ -1,12 +1,12 @@ """ -Tests to verify that all examples use 127.0.0.1 instead of 0.0.0.0 +Tests to verify that all examples use the new address paradigm consistently """ from pathlib import Path -class TestExamplesBindAddress: - """Test suite to verify all examples use secure bind addresses""" +class TestExamplesAddressParadigm: + """Test suite to verify all examples use the new address paradigm consistently""" def get_example_files(self): """Get all Python files in the examples directory""" @@ -32,49 +32,26 @@ class TestExamplesBindAddress: return found_wildcards - def test_no_wildcard_binding_in_examples(self): - """Test that no example files use 0.0.0.0 for binding""" + def test_examples_use_address_paradigm(self): + """Test that examples use the new address paradigm functions""" example_files = self.get_example_files() - # Skip certain files that might legitimately discuss wildcards - skip_files = [ - "network_discover.py", # This demonstrates wildcard expansion - ] - - files_with_wildcards = {} - - for filepath in example_files: - if any(skip in str(filepath) for skip in skip_files): - continue - - wildcards = self.check_file_for_wildcard_binding(filepath) - if wildcards: - files_with_wildcards[str(filepath)] = wildcards - - # Assert no wildcards found - if files_with_wildcards: - error_msg = "Found wildcard bindings in example files:\n" - for filepath, occurrences in files_with_wildcards.items(): - error_msg += f"\n{filepath}:\n" - for line_num, line in occurrences: - error_msg += f" Line {line_num}: {line}\n" - - assert False, error_msg - - def test_examples_use_loopback_address(self): - """Test that examples use 127.0.0.1 for local binding""" - example_files = self.get_example_files() - - # Files that should contain listen addresses - files_with_networking = [ - "ping/ping.py", + # Files that should use the new paradigm + networking_examples = [ + "echo/echo.py", "chat/chat.py", + "ping/ping.py", "bootstrap/bootstrap.py", "pubsub/pubsub.py", "identify/identify.py", ] - for filename in files_with_networking: + paradigm_functions = [ + "get_available_interfaces", + "get_optimal_binding_address", + ] + + for filename in networking_examples: filepath = None for example_file in example_files: if filename in str(example_file): @@ -87,24 +64,54 @@ class TestExamplesBindAddress: with open(filepath, encoding="utf-8") as f: content = f.read() - # Check for proper loopback usage - has_loopback = "127.0.0.1" in content or "localhost" in content - has_multiaddr_loopback = "/ip4/127.0.0.1/" in content + # Check that the file uses the new paradigm functions + for func in paradigm_functions: + assert func in content, ( + f"{filepath} should use {func} from the new address paradigm" + ) - assert has_loopback or has_multiaddr_loopback, ( - f"{filepath} should use loopback address (127.0.0.1)" + def test_wildcard_available_as_feature(self): + """Test that wildcard is available as a feature when needed""" + example_files = self.get_example_files() + + # Check that network_discover.py demonstrates wildcard usage + network_discover_file = None + for example_file in example_files: + if "network_discover.py" in str(example_file): + network_discover_file = example_file + break + + if network_discover_file: + with open(network_discover_file, encoding="utf-8") as f: + content = f.read() + + # Should demonstrate wildcard expansion + assert "0.0.0.0" in content, ( + f"{network_discover_file} should demonstrate wildcard usage" + ) + assert "expand_wildcard_address" in content, ( + f"{network_discover_file} should use expand_wildcard_address" ) - def test_doc_examples_use_loopback(self): - """Test that documentation examples use secure addresses""" + def test_doc_examples_use_paradigm(self): + """Test that documentation examples use the new address paradigm""" doc_examples_dir = Path("examples/doc-examples") if not doc_examples_dir.exists(): return doc_example_files = list(doc_examples_dir.glob("*.py")) + paradigm_functions = [ + "get_available_interfaces", + "get_optimal_binding_address", + ] + for filepath in doc_example_files: - wildcards = self.check_file_for_wildcard_binding(filepath) - assert not wildcards, ( - f"Documentation example {filepath} contains wildcard binding" - ) + with open(filepath, encoding="utf-8") as f: + content = f.read() + + # Check that doc examples use the new paradigm + for func in paradigm_functions: + assert func in content, ( + f"Documentation example {filepath} should use {func}" + ) diff --git a/tests/utils/test_default_bind_address.py b/tests/utils/test_default_bind_address.py index b8a501d2..b0598b5a 100644 --- a/tests/utils/test_default_bind_address.py +++ b/tests/utils/test_default_bind_address.py @@ -1,5 +1,5 @@ """ -Tests for default bind address changes from 0.0.0.0 to 127.0.0.1 +Tests for the new address paradigm with wildcard support as a feature """ import pytest @@ -9,28 +9,43 @@ from libp2p import new_host from libp2p.utils.address_validation import ( get_available_interfaces, get_optimal_binding_address, + get_wildcard_address, ) -class TestDefaultBindAddress: +class TestAddressParadigm: """ - Test suite for verifying default bind addresses use - secure addresses (not 0.0.0.0) + Test suite for verifying the new address paradigm: + - get_available_interfaces() returns all available interfaces + - get_optimal_binding_address() returns optimal address for examples + - get_wildcard_address() provides wildcard as a feature when needed """ - def test_default_bind_address_is_not_wildcard(self): - """Test that default bind address is NOT 0.0.0.0 (wildcard)""" + def test_wildcard_address_function(self): + """Test that get_wildcard_address() provides wildcard as a feature""" + port = 8000 + addr = get_wildcard_address(port) + + # Should return wildcard address when explicitly requested + assert "0.0.0.0" in str(addr) + addr_str = str(addr) + assert "/ip4/" in addr_str + assert f"/tcp/{port}" in addr_str + + def test_optimal_binding_address_selection(self): + """Test that optimal binding address uses good heuristics""" port = 8000 addr = get_optimal_binding_address(port) - # Should NOT return wildcard address - assert "0.0.0.0" not in str(addr) - # Should return a valid IP address (could be loopback or local network) addr_str = str(addr) assert "/ip4/" in addr_str assert f"/tcp/{port}" in addr_str + # Should be from available interfaces + available_interfaces = get_available_interfaces(port) + assert addr in available_interfaces + def test_available_interfaces_includes_loopback(self): """Test that available interfaces always includes loopback address""" port = 8000 @@ -43,9 +58,12 @@ class TestDefaultBindAddress: loopback_found = any("127.0.0.1" in str(addr) for addr in interfaces) assert loopback_found, "Loopback address not found in available interfaces" - # Should not have wildcard as the only option - if len(interfaces) == 1: - assert "0.0.0.0" not in str(interfaces[0]) + # Available interfaces should not include wildcard by default + # (wildcard is available as a feature through get_wildcard_address()) + wildcard_found = any("0.0.0.0" in str(addr) for addr in interfaces) + assert not wildcard_found, ( + "Wildcard should not be in default available interfaces" + ) def test_host_default_listen_address(self): """Test that new hosts use secure default addresses""" @@ -59,55 +77,66 @@ class TestDefaultBindAddress: # Note: We can't test actual binding without running the host, # but we've verified the address format is correct - def test_no_wildcard_in_fallback(self): - """Test that fallback addresses don't use wildcard binding""" - # When no interfaces are discovered, fallback should be loopback + def test_paradigm_consistency(self): + """Test that the address paradigm is consistent""" port = 8000 - # Even if we can't discover interfaces, we should get loopback - addr = get_optimal_binding_address(port) - # Should NOT be wildcard - assert "0.0.0.0" not in str(addr) + # get_optimal_binding_address should return a valid address + optimal_addr = get_optimal_binding_address(port) + assert "/ip4/" in str(optimal_addr) + assert f"/tcp/{port}" in str(optimal_addr) - # Should be a valid IP address - addr_str = str(addr) - assert "/ip4/" in addr_str - assert f"/tcp/{port}" in addr_str + # get_wildcard_address should return wildcard when explicitly needed + wildcard_addr = get_wildcard_address(port) + assert "0.0.0.0" in str(wildcard_addr) + assert f"/tcp/{port}" in str(wildcard_addr) + + # Both should be valid Multiaddr objects + assert isinstance(optimal_addr, Multiaddr) + assert isinstance(wildcard_addr, Multiaddr) @pytest.mark.parametrize("protocol", ["tcp", "udp"]) - def test_different_protocols_use_secure_addresses(self, protocol): - """Test that different protocols still use secure addresses by default""" + def test_different_protocols_support(self, protocol): + """Test that different protocols are supported by the paradigm""" port = 8000 - addr = get_optimal_binding_address(port, protocol=protocol) - # Should NOT be wildcard - assert "0.0.0.0" not in str(addr) - assert protocol in str(addr) + # Test optimal address with different protocols + optimal_addr = get_optimal_binding_address(port, protocol=protocol) + assert protocol in str(optimal_addr) + assert f"/{protocol}/{port}" in str(optimal_addr) - # Should be a valid IP address - addr_str = str(addr) - assert "/ip4/" in addr_str - assert f"/{protocol}/{port}" in addr_str + # Test wildcard address with different protocols + wildcard_addr = get_wildcard_address(port, protocol=protocol) + assert "0.0.0.0" in str(wildcard_addr) + assert protocol in str(wildcard_addr) + assert f"/{protocol}/{port}" in str(wildcard_addr) - def test_security_no_public_binding_by_default(self): - """Test that no public interface binding occurs by default""" + # Test available interfaces with different protocols + interfaces = get_available_interfaces(port, protocol=protocol) + assert len(interfaces) > 0 + for addr in interfaces: + assert protocol in str(addr) + + def test_wildcard_available_as_feature(self): + """Test that wildcard binding is available as a feature when needed""" port = 8000 + + # Wildcard should be available through get_wildcard_address() + wildcard_addr = get_wildcard_address(port) + assert "0.0.0.0" in str(wildcard_addr) + + # But should not be in default available interfaces interfaces = get_available_interfaces(port) - - # Check that we don't expose on all interfaces by default - wildcard_addrs = [addr for addr in interfaces if "0.0.0.0" in str(addr)] - assert len(wildcard_addrs) == 0, ( - "Found wildcard addresses in default interfaces" + wildcard_in_interfaces = any("0.0.0.0" in str(addr) for addr in interfaces) + assert not wildcard_in_interfaces, ( + "Wildcard should not be in default interfaces" ) - # Verify optimal address selection doesn't choose wildcard + # Optimal address should not be wildcard by default optimal = get_optimal_binding_address(port) - assert "0.0.0.0" not in str(optimal), "Optimal address should not be wildcard" - - # Should be a valid IP address (could be loopback or local network) - addr_str = str(optimal) - assert "/ip4/" in addr_str - assert f"/tcp/{port}" in addr_str + assert "0.0.0.0" not in str(optimal), ( + "Optimal address should not be wildcard by default" + ) def test_loopback_is_always_available(self): """Test that loopback address is always available as an option""" @@ -132,9 +161,6 @@ class TestDefaultBindAddress: interfaces = get_available_interfaces(port) optimal = get_optimal_binding_address(port) - # Should never return wildcard - assert "0.0.0.0" not in str(optimal) - # Should return one of the available interfaces optimal_str = str(optimal) interface_strs = [str(addr) for addr in interfaces] @@ -142,7 +168,7 @@ class TestDefaultBindAddress: f"Optimal address {optimal_str} should be in available interfaces" ) - # If non-loopback interfaces are available, should prefer them + # Should prefer non-loopback when available, fallback to loopback non_loopback_interfaces = [ addr for addr in interfaces if "127.0.0.1" not in str(addr) ] @@ -157,23 +183,21 @@ class TestDefaultBindAddress: "Should use loopback when no other interfaces available" ) - def test_address_validation_utilities_behavior(self): - """Test that address validation utilities behave as expected""" + def test_address_paradigm_completeness(self): + """Test that the address paradigm provides all necessary functionality""" port = 8000 - # Test that we get multiple interface options + # Test that we get interface options interfaces = get_available_interfaces(port) - assert len(interfaces) >= 2, ( - "Should have at least loopback + one network interface" - ) + assert len(interfaces) >= 1, "Should have at least one interface" # Test that loopback is always included has_loopback = any("127.0.0.1" in str(addr) for addr in interfaces) assert has_loopback, "Loopback should always be available" - # Test that no wildcards are included - has_wildcard = any("0.0.0.0" in str(addr) for addr in interfaces) - assert not has_wildcard, "Wildcard addresses should never be included" + # Test that wildcard is available as a feature + wildcard_addr = get_wildcard_address(port) + assert "0.0.0.0" in str(wildcard_addr) # Test optimal selection optimal = get_optimal_binding_address(port)