diff --git a/examples/interop/local_ping_test.py b/examples/interop/local_ping_test.py index 767755967..3aa6e2bfb 100755 --- a/examples/interop/local_ping_test.py +++ b/examples/interop/local_ping_test.py @@ -35,6 +35,7 @@ Transport as NoiseTransport, ) from libp2p.utils.address_validation import get_available_interfaces +from libp2p.utils.multiaddr_utils import extract_ip_from_multiaddr PING_PROTOCOL_ID = TProtocol("/ipfs/ping/1.0.0") PING_LENGTH = 32 @@ -174,9 +175,9 @@ def create_muxer_options(self): else: raise ValueError(f"Unsupported muxer: {self.muxer}") - def _get_ip_value(self, addr) -> str | None: + def _get_ip_value(self, addr: multiaddr.Multiaddr) -> str | None: """Extract IP value from multiaddr (IPv4 or IPv6).""" - return addr.value_for_protocol("ip4") or addr.value_for_protocol("ip6") + return extract_ip_from_multiaddr(addr) def _get_protocol_names(self, addr) -> list: """Get protocol names from multiaddr.""" diff --git a/interop/perf/perf_test.py b/interop/perf/perf_test.py index 91b6c3697..08c4b70ee 100644 --- a/interop/perf/perf_test.py +++ b/interop/perf/perf_test.py @@ -53,6 +53,7 @@ ) from libp2p.transport.quic.config import QUICTransportConfig from libp2p.utils.address_validation import get_available_interfaces +from libp2p.utils.multiaddr_utils import extract_ip_from_multiaddr MAX_TEST_TIMEOUT = 300 logger = logging.getLogger("libp2p.perf_test") @@ -620,14 +621,8 @@ def create_tls_server_config(self) -> ssl.SSLContext | None: return None def _get_ip_value(self, addr: multiaddr.Multiaddr) -> str | None: - for protocol in ("ip4", "ip6"): - try: - value = addr.value_for_protocol(protocol) - except Exception: - value = None - if value: - return value - return None + """Extract IP value from multiaddr (IPv4 or IPv6).""" + return extract_ip_from_multiaddr(addr) def _safe_value_for_protocol( self, addr: multiaddr.Multiaddr, protocol: str diff --git a/interop/transport/ping_test.py b/interop/transport/ping_test.py index 10d3ba70e..8cb8dbd27 100644 --- a/interop/transport/ping_test.py +++ b/interop/transport/ping_test.py @@ -60,6 +60,7 @@ TLSTransport, ) from libp2p.utils.address_validation import get_available_interfaces +from libp2p.utils.multiaddr_utils import extract_ip_from_multiaddr _orig_mk_identify_protobuf = _identify_mod._mk_identify_protobuf @@ -436,7 +437,7 @@ def create_tls_server_config(self) -> ssl.SSLContext | None: def _get_ip_value(self, addr: multiaddr.Multiaddr) -> str | None: """Extract IP value from multiaddr (IPv4 or IPv6).""" - return addr.value_for_protocol("ip4") or addr.value_for_protocol("ip6") + return extract_ip_from_multiaddr(addr) def _get_protocol_names(self, addr: multiaddr.Multiaddr) -> list[str]: """Get protocol names from multiaddr.""" diff --git a/newsfragments/1371.bugfix.rst b/newsfragments/1371.bugfix.rst new file mode 100644 index 000000000..89742142d --- /dev/null +++ b/newsfragments/1371.bugfix.rst @@ -0,0 +1 @@ +Fixed transport interop listener crashes when publishing IPv6-only multiaddrs. ``_get_ip_value`` now catches ``ProtocolLookupError`` instead of assuming ``value_for_protocol("ip4")`` returns falsy on IPv6 addresses. diff --git a/tests/interop/transport/__init__.py b/tests/interop/transport/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/interop/transport/test_ping_test_helpers.py b/tests/interop/transport/test_ping_test_helpers.py new file mode 100644 index 000000000..34f101fc6 --- /dev/null +++ b/tests/interop/transport/test_ping_test_helpers.py @@ -0,0 +1,59 @@ +"""Unit tests for interop/transport/ping_test.py helper logic.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +import sys + +import pytest +import multiaddr + +pytest.importorskip("redis") + +_REPO_ROOT = Path(__file__).resolve().parents[3] +_PING_TEST_PATH = _REPO_ROOT / "interop" / "transport" / "ping_test.py" + + +def _load_ping_test_module(): + spec = importlib.util.spec_from_file_location( + "interop_transport_ping_test", _PING_TEST_PATH + ) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules["interop_transport_ping_test"] = module + spec.loader.exec_module(module) + return module + + +ping_test = _load_ping_test_module() +PingTest = ping_test.PingTest + + +def _ping_test_env(monkeypatch: pytest.MonkeyPatch) -> PingTest: + monkeypatch.setenv("TRANSPORT", "tcp") + monkeypatch.setenv("MUXER", "mplex") + monkeypatch.setenv("SECURE_CHANNEL", "noise") + monkeypatch.setenv("IS_DIALER", "false") + monkeypatch.setenv("LISTENER_IP", "0.0.0.0") + monkeypatch.setenv("TEST_KEY", "test-key") + monkeypatch.setenv("REDIS_ADDR", "redis:6379") + return PingTest() + + +@pytest.mark.parametrize( + ("addr_str", "expected"), + [ + ("/ip6/::1/tcp/46499", "::1"), + ("/ip4/127.0.0.1/tcp/8080", "127.0.0.1"), + ("/tcp/8080", None), + ], +) +def test_get_ip_value( + monkeypatch: pytest.MonkeyPatch, + addr_str: str, + expected: str | None, +) -> None: + test = _ping_test_env(monkeypatch) + addr = multiaddr.Multiaddr(addr_str) + assert test._get_ip_value(addr) == expected