fix(interop): handle IPv6-only multiaddrs in _get_ip_value

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."""

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

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."""

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.

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