From 5633d52a63bff979f245a207f1144cc83b8b3e83 Mon Sep 17 00:00:00 2001 From: yashksaini-coder Date: Tue, 2 Sep 2025 03:02:41 +0530 Subject: [PATCH] test: Add comprehensive tests for address validation utilities and ensure secure binding addresses (127.0.0.1) are used instead of wildcard (0.0.0.0) --- test_address_validation_demo.py | 64 ++++++++ tests/examples/test_examples_bind_address.py | 110 +++++++++++++ tests/utils/test_default_bind_address.py | 161 +++++++++++++++++++ 3 files changed, 335 insertions(+) create mode 100644 test_address_validation_demo.py create mode 100644 tests/examples/test_examples_bind_address.py create mode 100644 tests/utils/test_default_bind_address.py diff --git a/test_address_validation_demo.py b/test_address_validation_demo.py new file mode 100644 index 00000000..dfca5de3 --- /dev/null +++ b/test_address_validation_demo.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +""" +Demonstration script to test address validation utilities +""" + +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) + +def main(): + print("=== Address Validation Utilities Demo ===\n") + + port = 8000 + + # Test available interfaces + print(f"Available interfaces for port {port}:") + interfaces = get_available_interfaces(port) + for i, addr in enumerate(interfaces, 1): + print(f" {i}. {addr}") + + print() + + # Test optimal binding address + print(f"Optimal binding address for port {port}:") + optimal = get_optimal_binding_address(port) + print(f" -> {optimal}") + + print() + + # Check for wildcard addresses + wildcard_found = any("0.0.0.0" in str(addr) for addr in interfaces) + print(f"Wildcard addresses found: {wildcard_found}") + + # Check for loopback addresses + loopback_found = any("127.0.0.1" in str(addr) for addr in interfaces) + print(f"Loopback addresses found: {loopback_found}") + + # Check if optimal is wildcard + optimal_is_wildcard = "0.0.0.0" in str(optimal) + print(f"Optimal address is wildcard: {optimal_is_wildcard}") + + print() + + if not wildcard_found and loopback_found and not optimal_is_wildcard: + print("✅ All checks passed! Address validation is working correctly.") + print(" - No wildcard addresses") + print(" - Loopback always available") + print(" - Optimal address is secure") + else: + print("❌ Some checks failed. Address validation needs attention.") + + print() + + # Test different protocols + print("Testing different protocols:") + for protocol in ["tcp", "udp"]: + addr = get_optimal_binding_address(port, protocol=protocol) + print(f" {protocol.upper()}: {addr}") + if "0.0.0.0" in str(addr): + print(f" ⚠️ Warning: {protocol} returned wildcard address") + +if __name__ == "__main__": + main() diff --git a/tests/examples/test_examples_bind_address.py b/tests/examples/test_examples_bind_address.py new file mode 100644 index 00000000..2f64ea46 --- /dev/null +++ b/tests/examples/test_examples_bind_address.py @@ -0,0 +1,110 @@ +""" +Tests to verify that all examples use 127.0.0.1 instead of 0.0.0.0 +""" + +import ast +import os +from pathlib import Path + + +class TestExamplesBindAddress: + """Test suite to verify all examples use secure bind addresses""" + + def get_example_files(self): + """Get all Python files in the examples directory""" + examples_dir = Path("examples") + return list(examples_dir.rglob("*.py")) + + def check_file_for_wildcard_binding(self, filepath): + """Check if a file contains 0.0.0.0 binding""" + with open(filepath, 'r', encoding='utf-8') as f: + content = f.read() + + # Check for various forms of wildcard binding + wildcard_patterns = [ + '0.0.0.0', + '/ip4/0.0.0.0/', + ] + + found_wildcards = [] + for line_num, line in enumerate(content.splitlines(), 1): + for pattern in wildcard_patterns: + if pattern in line and not line.strip().startswith('#'): + found_wildcards.append((line_num, line.strip())) + + return found_wildcards + + def test_no_wildcard_binding_in_examples(self): + """Test that no example files use 0.0.0.0 for binding""" + 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', + 'chat/chat.py', + 'bootstrap/bootstrap.py', + 'pubsub/pubsub.py', + 'identify/identify.py', + ] + + for filename in files_with_networking: + filepath = None + for example_file in example_files: + if filename in str(example_file): + filepath = example_file + break + + if filepath is None: + continue + + with open(filepath, 'r', 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 + + assert has_loopback or has_multiaddr_loopback, \ + f"{filepath} should use loopback address (127.0.0.1)" + + def test_doc_examples_use_loopback(self): + """Test that documentation examples use secure addresses""" + doc_examples_dir = Path("examples/doc-examples") + if not doc_examples_dir.exists(): + return + + doc_example_files = list(doc_examples_dir.glob("*.py")) + + for filepath in doc_example_files: + wildcards = self.check_file_for_wildcard_binding(filepath) + assert not wildcards, \ + f"Documentation example {filepath} contains wildcard binding" diff --git a/tests/utils/test_default_bind_address.py b/tests/utils/test_default_bind_address.py new file mode 100644 index 00000000..e5cc412d --- /dev/null +++ b/tests/utils/test_default_bind_address.py @@ -0,0 +1,161 @@ +""" +Tests for default bind address changes from 0.0.0.0 to 127.0.0.1 +""" + +import pytest +from multiaddr import Multiaddr + +from libp2p import new_host +from libp2p.utils.address_validation import ( + get_available_interfaces, + get_optimal_binding_address, +) + + +class TestDefaultBindAddress: + """Test suite for verifying default bind addresses use secure addresses (not 0.0.0.0)""" + + def test_default_bind_address_is_not_wildcard(self): + """Test that default bind address is NOT 0.0.0.0 (wildcard)""" + 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 + + def test_available_interfaces_includes_loopback(self): + """Test that available interfaces always includes loopback address""" + port = 8000 + interfaces = get_available_interfaces(port) + + # Should have at least one interface + assert len(interfaces) > 0 + + # Should include loopback address + 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]) + + def test_host_default_listen_address(self): + """Test that new hosts use secure default addresses""" + # Create a host with a specific port + port = 8000 + listen_addr = Multiaddr(f"/ip4/127.0.0.1/tcp/{port}") + host = new_host(listen_addrs=[listen_addr]) + + # Verify the host configuration + assert host is not None + # 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 + 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) + + # Should be a valid IP address + addr_str = str(addr) + assert "/ip4/" in addr_str + assert f"/tcp/{port}" in addr_str + + @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""" + 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) + + # Should be a valid IP address + addr_str = str(addr) + assert "/ip4/" in addr_str + assert f"/{protocol}/{port}" in addr_str + + def test_security_no_public_binding_by_default(self): + """Test that no public interface binding occurs by default""" + port = 8000 + 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" + + # Verify optimal address selection doesn't choose wildcard + 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 + + def test_loopback_is_always_available(self): + """Test that loopback address is always available as an option""" + port = 8000 + interfaces = get_available_interfaces(port) + + # Loopback should always be available + loopback_addrs = [addr for addr in interfaces if "127.0.0.1" in str(addr)] + assert len(loopback_addrs) > 0, "Loopback address should always be available" + + # At least one loopback address should have the correct port + loopback_with_port = [addr for addr in loopback_addrs if f"/tcp/{port}" in str(addr)] + assert len(loopback_with_port) > 0, f"Loopback address with port {port} should be available" + + def test_optimal_address_selection_behavior(self): + """Test that optimal address selection works correctly""" + port = 8000 + 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] + assert optimal_str in interface_strs, f"Optimal address {optimal_str} should be in available interfaces" + + # If non-loopback interfaces are available, should prefer them + non_loopback_interfaces = [addr for addr in interfaces if "127.0.0.1" not in str(addr)] + if non_loopback_interfaces: + # Should prefer non-loopback when available + assert "127.0.0.1" not in str(optimal), "Should prefer non-loopback when available" + else: + # Should use loopback when no other interfaces available + assert "127.0.0.1" in str(optimal), "Should use loopback when no other interfaces available" + + def test_address_validation_utilities_behavior(self): + """Test that address validation utilities behave as expected""" + port = 8000 + + # Test that we get multiple interface options + interfaces = get_available_interfaces(port) + assert len(interfaces) >= 2, "Should have at least loopback + one network 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 optimal selection + optimal = get_optimal_binding_address(port) + assert optimal in interfaces, "Optimal address should be from available interfaces"