mirror of
https://github.com/varun-r-mallya/py-libp2p.git
synced 2025-12-31 20:36:24 +00:00
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)
This commit is contained in:
64
test_address_validation_demo.py
Normal file
64
test_address_validation_demo.py
Normal file
@ -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()
|
||||
110
tests/examples/test_examples_bind_address.py
Normal file
110
tests/examples/test_examples_bind_address.py
Normal file
@ -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"
|
||||
161
tests/utils/test_default_bind_address.py
Normal file
161
tests/utils/test_default_bind_address.py
Normal file
@ -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"
|
||||
Reference in New Issue
Block a user