From b278e112c55fd6bcef7c5a99dad7424ad2bf54c8 Mon Sep 17 00:00:00 2001 From: parth-soni07 Date: Thu, 26 Feb 2026 17:23:31 +0530 Subject: [PATCH 1/7] hole-punch: add Python impl (v0.5) --- hole-punch/images.yaml | 20 + hole-punch/images/python/v0.5/Dockerfile.peer | 23 + .../images/python/v0.5/Dockerfile.relay | 23 + hole-punch/images/python/v0.5/peer.py | 757 ++++++++++++++++++ hole-punch/images/python/v0.5/relay.py | 687 ++++++++++++++++ .../images/python/v0.5/requirements.txt | 14 + hole-punch/images/python/v0.5/run-peer.sh | 10 + hole-punch/images/python/v0.5/run-relay.sh | 4 + hole-punch/images/rust/v0.56/src/bin/peer.rs | 13 +- hole-punch/images/rust/v0.56/src/bin/relay.rs | 7 + hole-punch/lib/generate-tests.sh | 56 +- hole-punch/lib/run-single-test.sh | 2 + 12 files changed, 1584 insertions(+), 32 deletions(-) create mode 100644 hole-punch/images/python/v0.5/Dockerfile.peer create mode 100644 hole-punch/images/python/v0.5/Dockerfile.relay create mode 100644 hole-punch/images/python/v0.5/peer.py create mode 100644 hole-punch/images/python/v0.5/relay.py create mode 100644 hole-punch/images/python/v0.5/requirements.txt create mode 100644 hole-punch/images/python/v0.5/run-peer.sh create mode 100644 hole-punch/images/python/v0.5/run-relay.sh diff --git a/hole-punch/images.yaml b/hole-punch/images.yaml index b72b4438..c9099ac9 100644 --- a/hole-punch/images.yaml +++ b/hole-punch/images.yaml @@ -6,6 +6,8 @@ test-aliases: - alias: "failing" value: "rust-v0.56 x rust-v0.56" + - alias: "python" + value: "py-v0.5" # Router configurations - Multiple router types for testing different NAT implementations routers: @@ -26,6 +28,15 @@ relays: secureChannels: [noise, tls] muxers: [yamux, mplex] + - id: py-v0.5 + source: + type: local + path: images/python/v0.5 + dockerfile: Dockerfile.relay + transports: [tcp, ws] + secureChannels: [noise] + muxers: [yamux] + # Implementation definitions implementations: - id: rust-v0.56 @@ -36,3 +47,12 @@ implementations: transports: [quic-v1, tcp, webrtc-direct, ws] secureChannels: [noise, tls] muxers: [yamux, mplex] + + - id: py-v0.5 + source: + type: local + path: images/python/v0.5 + dockerfile: Dockerfile.peer + transports: [tcp, ws] + secureChannels: [noise] + muxers: [yamux] diff --git a/hole-punch/images/python/v0.5/Dockerfile.peer b/hole-punch/images/python/v0.5/Dockerfile.peer new file mode 100644 index 00000000..68d2ae84 --- /dev/null +++ b/hole-punch/images/python/v0.5/Dockerfile.peer @@ -0,0 +1,23 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Install system dependencies (iproute2 for network route setup, build deps for fastecdsa) +RUN apt-get update && apt-get install -y \ + iproute2 \ + git \ + build-essential \ + libgmp-dev \ + && rm -rf /var/lib/apt/lists/* + +# Install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application +COPY peer.py . +COPY run-peer.sh /usr/local/bin/ + +RUN chmod +x /usr/local/bin/run-peer.sh + +ENTRYPOINT ["/usr/local/bin/run-peer.sh"] diff --git a/hole-punch/images/python/v0.5/Dockerfile.relay b/hole-punch/images/python/v0.5/Dockerfile.relay new file mode 100644 index 00000000..db750cb4 --- /dev/null +++ b/hole-punch/images/python/v0.5/Dockerfile.relay @@ -0,0 +1,23 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Install system dependencies (including build deps for fastecdsa) +RUN apt-get update && apt-get install -y \ + iproute2 \ + git \ + build-essential \ + libgmp-dev \ + && rm -rf /var/lib/apt/lists/* + +# Install Python dependencies +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application +COPY relay.py . +COPY run-relay.sh /usr/local/bin/ + +RUN chmod +x /usr/local/bin/run-relay.sh + +ENTRYPOINT ["/usr/local/bin/run-relay.sh"] diff --git a/hole-punch/images/python/v0.5/peer.py b/hole-punch/images/python/v0.5/peer.py new file mode 100644 index 00000000..90a8585b --- /dev/null +++ b/hole-punch/images/python/v0.5/peer.py @@ -0,0 +1,757 @@ +#!/usr/bin/env python3 +""" +py-libp2p hole-punch test peer implementation. +Handles both dialer and listener roles for DCUtR interop testing. +""" + +import logging +import os +import sys +import time + +import multiaddr +import redis +import trio + +from libp2p import new_host, create_yamux_muxer_option, create_mplex_muxer_option +from libp2p.crypto.ed25519 import create_new_key_pair +from libp2p.crypto.x25519 import create_new_key_pair as create_new_x25519_key_pair +from libp2p.peer.peerinfo import info_from_p2p_addr, PeerInfo +from libp2p.peer.id import ID +from libp2p.relay.circuit_v2.dcutr import DCUtRProtocol +from libp2p.relay.circuit_v2.protocol import CircuitV2Protocol, DEFAULT_RELAY_LIMITS +from libp2p.relay.circuit_v2.config import RelayConfig, RelayRole +import libp2p.relay.circuit_v2.transport as _circuit_transport_mod +from libp2p.relay.circuit_v2.transport import CircuitV2Transport +from libp2p.security.noise.transport import ( + PROTOCOL_ID as NOISE_PROTOCOL_ID, + Transport as NoiseTransport, +) +from libp2p.security.tls.transport import ( + PROTOCOL_ID as TLS_PROTOCOL_ID, + TLSTransport, +) +from libp2p.tools.async_service import background_trio_service +from libp2p.network.connection.raw_connection import RawConnection + +logger = logging.getLogger("hole-punch-peer") + +# --- Rust relay wire compatibility (length-delimited framing + rust-libp2p proto layout) --- +# When RELAY_ID is Rust we use varint(length) + message and Rust's HopMessage/Reservation/Status. + +# HopMessage type (rust-libp2p): RESERVE=0, CONNECT=1, STATUS=2 +# Status enum: OK=100 +RUST_STATUS_OK = 100 + + +def _varint_encode(n: int) -> bytes: + """Encode non-negative int as unsigned varint (same as quick-protobuf-codec).""" + if n < 0: + raise ValueError("varint must be non-negative") + buf = [] + while n > 0x7F: + buf.append((n & 0x7F) | 0x80) + n >>= 7 + buf.append(n & 0x7F) + return bytes(buf) + + +def _varint_decode(buf: bytes, start: int = 0) -> tuple[int, int]: + """Decode unsigned varint from buf starting at start. Returns (value, bytes_consumed).""" + n = 0 + shift = 0 + i = start + while i < len(buf): + b = buf[i] + n |= (b & 0x7F) << shift + i += 1 + if (b & 0x80) == 0: + return n, i - start + shift += 7 + if shift >= 64: + raise ValueError("varint too long") + raise ValueError("varint truncated") + + +def _encode_rust_reserve_request() -> bytes: + """Minimal RESERVE HopMessage (type=0 only).""" + # field 1, wire type 0 (varint), value 0 + return b"\x08\x00" + + +def _encode_rust_connect_request(peer_id_bytes: bytes) -> bytes: + """CONNECT HopMessage: type=1, peer=Peer{id=peer_id_bytes, addrs=[]}. + Rust proto: peer is message Peer { required bytes id=1; repeated bytes addrs=2; }.""" + # Peer message: field 1 (id): tag 0x0a, length, bytes + peer_inner = b"\x0a" + _varint_encode(len(peer_id_bytes)) + peer_id_bytes + # HopMessage: field 1 (type)=CONNECT, field 2 (peer)=Peer message + peer_outer_len = _varint_encode(len(peer_inner)) + return b"\x08\x01\x12" + peer_outer_len + peer_inner + + +def _parse_rust_hop_message(buf: bytes) -> dict: + """ + Parse Rust-shaped HopMessage from buf. Returns dict with: + type_pb (int), status (int | None), reservation (dict | None with expire, addrs). + """ + out = {"type_pb": None, "status": None, "reservation": None} + i = 0 + while i < len(buf): + if i >= len(buf): + break + tag = buf[i] + i += 1 + if i >= len(buf): + break + field_num = tag >> 3 + wire = tag & 7 + if wire == 0: # varint + n, consumed = _varint_decode(buf, i) + i += consumed + if field_num == 1: + out["type_pb"] = n + elif field_num == 5: + out["status"] = n + elif wire == 2: # length-delimited + length, consumed = _varint_decode(buf, i) + i += consumed + end = i + length + if end > len(buf): + break + chunk = buf[i:end] + i = end + if field_num == 3: # reservation message + out["reservation"] = _parse_rust_reservation(chunk) + # field 2 (peer) we ignore + return out + + +# Rust StopMessage: type (1), peer (2, Peer msg), limit (3), status (4). CONNECT=0, STATUS=1. +def _parse_rust_stop_message(buf: bytes) -> dict: + """Parse Rust StopMessage; return dict with type_pb, peer_id (bytes or None).""" + out = {"type_pb": None, "peer_id": None} + i = 0 + while i < len(buf): + if i >= len(buf): + break + tag = buf[i] + i += 1 + if i >= len(buf): + break + field_num = tag >> 3 + wire = tag & 7 + if wire == 0: # varint + n, consumed = _varint_decode(buf, i) + i += consumed + if field_num == 1: + out["type_pb"] = n + # field 4 status we don't need when parsing CONNECT + elif wire == 2: # length-delimited + length, consumed = _varint_decode(buf, i) + i += consumed + end = i + length + if end > len(buf): + break + if field_num == 2: # Peer message: field 1 = id (bytes) + chunk = buf[i:end] + pi = 0 + while pi < len(chunk): + if pi >= len(chunk): + break + pt = chunk[pi] + pi += 1 + if pi >= len(chunk): + break + pfn = pt >> 3 + pw = pt & 7 + if pw == 2 and pfn == 1: # id = field 1, length-delimited + plen, pc = _varint_decode(chunk, pi) + pi += pc + if pi + plen <= len(chunk): + out["peer_id"] = bytes(chunk[pi : pi + plen]) + break + elif pw == 0: + _, pc = _varint_decode(chunk, pi) + pi += pc + elif pw == 2: + plen, pc = _varint_decode(chunk, pi) + pi += pc + plen + break + i = end + return out + + +def _encode_rust_stop_status_ok() -> bytes: + """Rust StopMessage STATUS with status=100 (OK). type=1 (STATUS), status=100.""" + return b"\x08\x01\x20\x64" # field 1 (type)=1, field 4 (status)=100 + + +def _encode_rust_stop_status_error(status_code: int) -> bytes: + """Rust StopMessage STATUS with given status (e.g. 400 MALFORMED_MESSAGE).""" + # type=1 (STATUS), field 4 = varint status_code + return b"\x08\x01\x20" + _varint_encode(status_code) + + +def _parse_rust_reservation(buf: bytes) -> dict: + """Parse Rust Reservation: expire (1), addrs (2 repeated).""" + out = {"expire": None, "addrs": []} + i = 0 + while i < len(buf): + if i >= len(buf): + break + tag = buf[i] + i += 1 + if i >= len(buf): + break + field_num = tag >> 3 + wire = tag & 7 + if wire == 0: + n, consumed = _varint_decode(buf, i) + i += consumed + if field_num == 1: + out["expire"] = n + elif wire == 2: + length, consumed = _varint_decode(buf, i) + i += consumed + end = i + length + if end > len(buf): + break + if field_num == 2: + out["addrs"].append(buf[i:end]) + i = end + return out + + +async def _read_varint_prefixed_message(stream) -> bytes: + """Read unsigned_varint(length) then exactly length bytes from stream.""" + chunks = [] + total = 0 + while True: + chunk = await stream.read(1) + if not chunk: + raise ValueError("varint truncated") + chunks.append(chunk) + total += 1 + if total > 10: + raise ValueError("varint too long") + if (chunk[0] & 0x80) == 0: + break + buf = b"".join(chunks) + length, _ = _varint_decode(buf, 0) + payload = b"" + while len(payload) < length: + got = await stream.read(length - len(payload)) + if not got: + raise ValueError("message truncated") + payload += got + return payload + + +async def _write_varint_prefixed_message(stream, message: bytes) -> None: + """Write unsigned_varint(len(message)) then message to stream.""" + prefix = _varint_encode(len(message)) + await stream.write(prefix + message) + + +class HolePunchPeer: + def __init__(self): + # Required environment variables + self.is_dialer = os.getenv("IS_DIALER", "false").lower() == "true" + self.redis_addr = os.getenv("REDIS_ADDR") + self.test_key = os.getenv("TEST_KEY") + self.relay_id = os.getenv("RELAY_ID", "") + self.transport = os.getenv("TRANSPORT", "tcp") + self.secure_channel = os.getenv("SECURE_CHANNEL") # May be None for QUIC + self.muxer = os.getenv("MUXER") # May be None for QUIC + self.peer_ip = os.getenv("PEER_IP", "0.0.0.0") + self.router_ip = os.getenv("ROUTER_IP") + self.debug = os.getenv("DEBUG", "false").lower() == "true" + + # Validate required env vars + if not self.redis_addr: + raise ValueError("REDIS_ADDR environment variable is required") + if not self.test_key: + raise ValueError("TEST_KEY environment variable is required") + + # Parse Redis address + if ":" in self.redis_addr: + host, port = self.redis_addr.split(":") + self.redis_host, self.redis_port = host, int(port) + else: + self.redis_host = self.redis_addr + self.redis_port = 6379 + + self.redis_client = None + self.host = None + + def create_security_options(self): + """Create security transport options based on SECURE_CHANNEL.""" + key_pair = create_new_key_pair() + + # Standalone transports (QUIC) have built-in security + if self.transport == "quic-v1": + return {}, key_pair + + if self.secure_channel == "noise": + noise_key_pair = create_new_x25519_key_pair() + noise_transport = NoiseTransport( + libp2p_keypair=key_pair, + noise_privkey=noise_key_pair.private_key, + early_data=None, + ) + return {NOISE_PROTOCOL_ID: noise_transport}, key_pair + elif self.secure_channel == "tls": + tls_transport = TLSTransport( + libp2p_keypair=key_pair, + early_data=None, + muxers=None, + ) + return {TLS_PROTOCOL_ID: tls_transport}, key_pair + else: + raise ValueError(f"Unsupported secure channel: {self.secure_channel}") + + def create_muxer_options(self): + """Create muxer options based on MUXER.""" + if self.transport == "quic-v1": + return None # QUIC has built-in muxing + + if self.muxer == "yamux": + return create_yamux_muxer_option() + elif self.muxer == "mplex": + return create_mplex_muxer_option() + else: + raise ValueError(f"Unsupported muxer: {self.muxer}") + + def create_listen_address(self, port: int = 0): + """Create listen multiaddr based on transport.""" + if self.transport == "tcp": + return multiaddr.Multiaddr(f"/ip4/{self.peer_ip}/tcp/{port}") + elif self.transport == "quic-v1": + return multiaddr.Multiaddr(f"/ip4/{self.peer_ip}/udp/{port}/quic-v1") + elif self.transport == "ws": + return multiaddr.Multiaddr(f"/ip4/{self.peer_ip}/tcp/{port}/ws") + elif self.transport == "wss": + return multiaddr.Multiaddr(f"/ip4/{self.peer_ip}/tcp/{port}/wss") + else: + raise ValueError(f"Unsupported transport: {self.transport}") + + async def connect_redis(self): + """Connect to Redis with retry.""" + print(f"Connecting to Redis at {self.redis_host}:{self.redis_port}...", file=sys.stderr) + for attempt in range(10): + try: + self.redis_client = redis.Redis( + host=self.redis_host, + port=self.redis_port, + decode_responses=True + ) + self.redis_client.ping() + print(f"Connected to Redis on attempt {attempt + 1}", file=sys.stderr) + return + except Exception as e: + print(f"Redis connection attempt {attempt + 1} failed: {e}", file=sys.stderr) + if attempt < 9: + await trio.sleep(1) + raise RuntimeError("Failed to connect to Redis after 10 attempts") + + async def run_listener(self): + """Run as listener (IS_DIALER=false).""" + print("Starting as LISTENER...", file=sys.stderr) + print(f" TRANSPORT: {self.transport}", file=sys.stderr) + print(f" SECURE_CHANNEL: {self.secure_channel}", file=sys.stderr) + print(f" MUXER: {self.muxer}", file=sys.stderr) + print(f" PEER_IP: {self.peer_ip}", file=sys.stderr) + + sec_opt, key_pair = self.create_security_options() + muxer_opt = self.create_muxer_options() + listen_addr = self.create_listen_address(4001) + + self.host = new_host( + key_pair=key_pair, + sec_opt=sec_opt, + muxer_opt=muxer_opt, + enable_quic=(self.transport == "quic-v1"), + listen_addrs=[listen_addr], + ) + + # Configure relay client + relay_config = RelayConfig( + roles=RelayRole.STOP | RelayRole.CLIENT, + ) + relay_protocol = CircuitV2Protocol(self.host, DEFAULT_RELAY_LIMITS, allow_hop=False) + dcutr_protocol = DCUtRProtocol(self.host) + + async with self.host.run(listen_addrs=[listen_addr]): + async with background_trio_service(relay_protocol): + async with background_trio_service(dcutr_protocol): + await relay_protocol.event_started.wait() + await dcutr_protocol.event_started.wait() + + if self._is_rust_relay(): + self.host.set_stream_handler( + "/libp2p/circuit/relay/0.2.0/stop", + lambda s: self._handle_rust_stop_stream(s, relay_protocol), + ) + + # Initialize transport + transport = CircuitV2Transport(self.host, relay_protocol, relay_config) + + peer_id = str(self.host.get_id()) + print(f"Listener peer ID: {peer_id}", file=sys.stderr) + + # Wait for relay multiaddr + relay_key = f"{self.test_key}_relay_multiaddr" + relay_addr = None + print(f"Waiting for relay multiaddr at key: {relay_key}", file=sys.stderr) + for i in range(60): # 60 second timeout + relay_addr = self.redis_client.get(relay_key) + if relay_addr: + break + if i % 10 == 0: + print(f" Still waiting for relay... ({i}s)", file=sys.stderr) + await trio.sleep(1) + + if not relay_addr: + raise RuntimeError("Timeout waiting for relay multiaddr") + + print(f"Got relay multiaddr: {relay_addr}", file=sys.stderr) + + # Connect to relay + relay_maddr = multiaddr.Multiaddr(relay_addr) + relay_info = info_from_p2p_addr(relay_maddr) + await self.host.connect(relay_info) + print(f"Connected to relay: {relay_info.peer_id}", file=sys.stderr) + + # Add relay to peerstore for circuit reservation + self.host.get_peerstore().add_addrs( + relay_info.peer_id, relay_info.addrs, 3600 + ) + + # Make relay reservation BEFORE publishing peer_id. + # rust-libp2p relay requires an explicit RESERVE before it will + # forward CONNECT messages to this peer. py relay accepts it too. + print("Making relay reservation...", file=sys.stderr) + try: + hop_stream = await self.host.new_stream( + relay_info.peer_id, [_circuit_transport_mod.PROTOCOL_ID] + ) + if self._is_rust_relay(): + reserved = await self._make_reservation_rust_relay( + hop_stream, relay_info.peer_id + ) + else: + reserved = await transport._make_reservation( + hop_stream, relay_info.peer_id + ) + print( + f"Relay reservation: {'OK' if reserved else 'FAILED'}", + file=sys.stderr, + ) + try: + await hop_stream.close() + except Exception: + pass + except Exception as e: + print( + f"Warning: Could not make relay reservation: {e}", + file=sys.stderr, + ) + + # Publish our peer ID to Redis only AFTER reserving so the dialer + # can be sure the relay is ready to forward traffic to us. + redis_key = f"{self.test_key}_listener_peer_id" + self.redis_client.set(redis_key, peer_id) + print(f"Published peer ID to Redis key: {redis_key}", file=sys.stderr) + + # Wait for DCUtR to complete (dialer will initiate) + # The listener just needs to stay alive and respond + print("Listener ready, waiting for hole punch...", file=sys.stderr) + + # Run forever until container is shut down + await trio.sleep_forever() + + async def run_dialer(self): + """Run as dialer (IS_DIALER=true).""" + print("Starting as DIALER...", file=sys.stderr) + print(f" TRANSPORT: {self.transport}", file=sys.stderr) + print(f" SECURE_CHANNEL: {self.secure_channel}", file=sys.stderr) + print(f" MUXER: {self.muxer}", file=sys.stderr) + print(f" PEER_IP: {self.peer_ip}", file=sys.stderr) + + sec_opt, key_pair = self.create_security_options() + muxer_opt = self.create_muxer_options() + listen_addr = self.create_listen_address(4001) + + self.host = new_host( + key_pair=key_pair, + sec_opt=sec_opt, + muxer_opt=muxer_opt, + enable_quic=(self.transport == "quic-v1"), + listen_addrs=[listen_addr], + ) + + # Configure relay client. + # CLIENT enabled: the transport sends RESERVE then CONNECT on the same HOP stream. + # When RELAY_ID is rust, we use Rust protocol ID /libp2p/circuit/relay/2.0.0/hop + # (set in _use_rust_relay_protocol_id_if_needed) so py peers can talk to Rust relay. + relay_config = RelayConfig( + roles=RelayRole.STOP | RelayRole.CLIENT, + ) + relay_protocol = CircuitV2Protocol(self.host, DEFAULT_RELAY_LIMITS, allow_hop=False) + dcutr_protocol = DCUtRProtocol(self.host) + + async with self.host.run(listen_addrs=[listen_addr]): + async with background_trio_service(relay_protocol): + async with background_trio_service(dcutr_protocol): + await relay_protocol.event_started.wait() + await dcutr_protocol.event_started.wait() + + # Initialize transport + transport = CircuitV2Transport(self.host, relay_protocol, relay_config) + + peer_id = str(self.host.get_id()) + print(f"Dialer peer ID: {peer_id}", file=sys.stderr) + + # Wait for relay multiaddr + relay_key = f"{self.test_key}_relay_multiaddr" + relay_addr = None + print(f"Waiting for relay multiaddr at key: {relay_key}", file=sys.stderr) + for i in range(60): + relay_addr = self.redis_client.get(relay_key) + if relay_addr: + break + if i % 10 == 0: + print(f" Still waiting for relay... ({i}s)", file=sys.stderr) + await trio.sleep(1) + + if not relay_addr: + raise RuntimeError("Timeout waiting for relay multiaddr") + + print(f"Got relay multiaddr: {relay_addr}", file=sys.stderr) + + # Connect to relay + relay_maddr = multiaddr.Multiaddr(relay_addr) + relay_info = info_from_p2p_addr(relay_maddr) + await self.host.connect(relay_info) + print(f"Connected to relay: {relay_info.peer_id}", file=sys.stderr) + + # Wait for listener peer ID + listener_key = f"{self.test_key}_listener_peer_id" + listener_peer_id_str = None + print(f"Waiting for listener peer ID at key: {listener_key}", file=sys.stderr) + for i in range(60): + listener_peer_id_str = self.redis_client.get(listener_key) + if listener_peer_id_str: + break + if i % 10 == 0: + print(f" Still waiting for listener... ({i}s)", file=sys.stderr) + await trio.sleep(1) + + if not listener_peer_id_str: + raise RuntimeError("Timeout waiting for listener peer ID") + + print(f"Got listener peer ID: {listener_peer_id_str}", file=sys.stderr) + listener_peer_id = ID.from_base58(listener_peer_id_str) + + # Create circuit address to listener through relay + circuit_addr = multiaddr.Multiaddr( + f"{relay_addr}/p2p-circuit/p2p/{listener_peer_id_str}" + ) + print(f"Circuit address: {circuit_addr}", file=sys.stderr) + + # Start timing for DCUtR + handshake_start = time.time() + + # Connect through relay using the circuit transport + print("Connecting to listener through relay...", file=sys.stderr) + + if self._is_rust_relay(): + self.host.get_peerstore().add_addrs(listener_peer_id, [circuit_addr], 3600) + stream = await self._dial_rust_relay( + transport, relay_info, listener_peer_id + ) + if stream: + raw_conn = RawConnection(stream=stream, initiator=True) + await self.host.upgrade_outbound_connection( + raw_conn, listener_peer_id + ) + else: + await transport.dial(circuit_addr) + print("Connected to listener through relay", file=sys.stderr) + + # Initiate hole punch + print("Initiating hole punch...", file=sys.stderr) + result = await dcutr_protocol.initiate_hole_punch(listener_peer_id) + + handshake_end = time.time() + handshake_time_ms = (handshake_end - handshake_start) * 1000 + + # Verify direct connection + has_direct = await dcutr_protocol._have_direct_connection(listener_peer_id) + + print(f"Hole punch result: {result}", file=sys.stderr) + print(f"Has direct connection: {has_direct}", file=sys.stderr) + + if has_direct and result: + print(f"Hole punch SUCCESS! Direct connection established.", file=sys.stderr) + print(f"Handshake time: {handshake_time_ms:.2f}ms", file=sys.stderr) + + # Output results to stdout (YAML format) + print(f"handshakeTime: {handshake_time_ms:.2f}", file=sys.stdout) + + # Success - return cleanly to allow nursery to close + return + else: + print(f"Hole punch FAILED. result={result}, has_direct={has_direct}", file=sys.stderr) + raise SystemExit(1) + + def _use_rust_relay_protocol_id_if_needed(self): + """Use Rust relay HOP protocol ID when talking to a Rust relay. + rust-libp2p uses /libp2p/circuit/relay/0.2.0/hop (protocols/relay/src/protocol.rs). + STOP is handled by registering our own 0.2.0/stop handler in run_listener. + """ + if self.relay_id and self.relay_id.lower().startswith("rust"): + _circuit_transport_mod.PROTOCOL_ID = "/libp2p/circuit/relay/0.2.0/hop" + if self.debug: + print( + f"Using Rust relay protocol ID: {_circuit_transport_mod.PROTOCOL_ID}", + file=sys.stderr, + ) + + def _is_rust_relay(self) -> bool: + return bool(self.relay_id and self.relay_id.lower().startswith("rust")) + + async def _handle_rust_stop_stream(self, stream, relay_protocol) -> None: + """Handle incoming STOP stream from a Rust relay (length-delimited framing).""" + try: + payload = await _read_varint_prefixed_message(stream) + parsed = _parse_rust_stop_message(payload) + if parsed.get("type_pb") != 0: # CONNECT=0 + try: + await _write_varint_prefixed_message( + stream, _encode_rust_stop_status_error(400) + ) + except Exception: + pass + try: + await stream.close() + except Exception: + pass + return + peer_id_bytes = parsed.get("peer_id") + if not peer_id_bytes: + try: + await _write_varint_prefixed_message( + stream, _encode_rust_stop_status_error(400) + ) + except Exception: + pass + try: + await stream.close() + except Exception: + pass + return + await _write_varint_prefixed_message(stream, _encode_rust_stop_status_ok()) + peer_id = ID(peer_id_bytes) + remote_ma = f"/p2p/{peer_id.to_base58()}" + await relay_protocol.handle_incoming_connection(stream, remote_ma) + except Exception as e: + logger.debug("Rust STOP stream error: %s", e) + try: + await stream.close() + except Exception: + pass + + async def _make_reservation_rust_relay(self, hop_stream, relay_peer_id) -> bool: + """ + Make a reservation with a Rust relay using length-delimited framing and Rust proto layout. + Returns True if status is OK and reservation present. + """ + try: + msg = _encode_rust_reserve_request() + await _write_varint_prefixed_message(hop_stream, msg) + payload = await _read_varint_prefixed_message(hop_stream) + parsed = _parse_rust_hop_message(payload) + if parsed.get("status") != RUST_STATUS_OK: + return False + if parsed.get("reservation") is None: + return False + return True + except Exception as e: + logger.debug("Rust relay reservation failed: %s", e) + return False + + async def _dial_rust_relay( + self, + transport, + relay_info, + listener_peer_id, + ): + """ + Dial through a Rust relay: send CONNECT to relay targeting the listener + (who has already RESERVEd). Returns the open circuit stream. + """ + hop_stream = await self.host.new_stream( + relay_info.peer_id, [_circuit_transport_mod.PROTOCOL_ID] + ) + try: + connect_msg = _encode_rust_connect_request(listener_peer_id.to_bytes()) + await _write_varint_prefixed_message(hop_stream, connect_msg) + payload = await _read_varint_prefixed_message(hop_stream) + parsed = _parse_rust_hop_message(payload) + if parsed.get("status") != RUST_STATUS_OK: + await hop_stream.close() + raise RuntimeError( + f"Rust relay CONNECT failed: status={parsed.get('status')}" + ) + if hasattr(transport, "_build_connection_from_hop_stream"): + return await transport._build_connection_from_hop_stream( + hop_stream, relay_info.peer_id, listener_peer_id + ) + return hop_stream + except Exception: + try: + await hop_stream.close() + except Exception: + pass + raise + + async def run(self): + """Main entry point.""" + await self.connect_redis() + self._use_rust_relay_protocol_id_if_needed() + + if self.is_dialer: + await self.run_dialer() + else: + await self.run_listener() + + +def configure_logging(): + """Configure logging based on DEBUG env var.""" + debug = os.getenv("DEBUG", "false").lower() in ["true", "1", "yes", "debug"] + level = logging.DEBUG if debug else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s | %(name)s | %(levelname)s | %(message)s", + stream=sys.stderr, + ) + # Suppress overly verbose loggers + if not debug: + logging.getLogger("libp2p").setLevel(logging.WARNING) + + +async def main(): + configure_logging() + try: + peer = HolePunchPeer() + await peer.run() + except Exception as e: + print(f"Fatal error: {e}", file=sys.stderr) + import traceback + traceback.print_exc(file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + trio.run(main) diff --git a/hole-punch/images/python/v0.5/relay.py b/hole-punch/images/python/v0.5/relay.py new file mode 100644 index 00000000..46c60af7 --- /dev/null +++ b/hole-punch/images/python/v0.5/relay.py @@ -0,0 +1,687 @@ +#!/usr/bin/env python3 +""" +py-libp2p hole-punch test relay server implementation. + +Supports both Python (py-libp2p 2.0.0 wire format) and Rust (rust-libp2p +0.2.0 varint-length-delimited wire format) peers. +""" + +import logging +import os +import sys +import time + +import multiaddr +import redis +import trio + +from libp2p import new_host, create_yamux_muxer_option, create_mplex_muxer_option +from libp2p.crypto.ed25519 import create_new_key_pair +from libp2p.crypto.x25519 import create_new_key_pair as create_new_x25519_key_pair +from libp2p.peer.id import ID +from libp2p.relay.circuit_v2.protocol import ( + CircuitV2Protocol, + PROTOCOL_ID, + STOP_PROTOCOL_ID, +) +from libp2p.relay.circuit_v2.config import RelayConfig, RelayRole +from libp2p.relay.circuit_v2.resources import RelayLimits +from libp2p.relay.circuit_v2.transport import CircuitV2Transport +from libp2p.security.noise.transport import ( + PROTOCOL_ID as NOISE_PROTOCOL_ID, + Transport as NoiseTransport, +) +from libp2p.security.tls.transport import ( + PROTOCOL_ID as TLS_PROTOCOL_ID, + TLSTransport, +) +from libp2p.tools.async_service import background_trio_service + +logger = logging.getLogger("hole-punch-relay") + +# rust-libp2p uses these protocol IDs (version 0.2.0) +RUST_HOP_PROTOCOL_ID = "/libp2p/circuit/relay/0.2.0/hop" +RUST_STOP_PROTOCOL_ID = "/libp2p/circuit/relay/0.2.0/stop" + +RUST_STATUS_OK = 100 +RUST_STATUS_RESOURCE_LIMIT_EXCEEDED = 201 + + +# --- Varint + protobuf helpers (same wire format as quick_protobuf_codec) --- + +def _varint_encode(n: int) -> bytes: + buf = [] + while n > 0x7F: + buf.append((n & 0x7F) | 0x80) + n >>= 7 + buf.append(n & 0x7F) + return bytes(buf) + + +def _varint_decode(buf: bytes, start: int = 0) -> tuple[int, int]: + n = 0 + shift = 0 + i = start + while i < len(buf): + b = buf[i] + n |= (b & 0x7F) << shift + i += 1 + if (b & 0x80) == 0: + return n, i - start + shift += 7 + if shift >= 64: + raise ValueError("varint too long") + raise ValueError("varint truncated") + + +async def _read_varint_prefixed(stream) -> bytes: + chunks = [] + total = 0 + while True: + chunk = await stream.read(1) + if not chunk: + raise ValueError("varint truncated") + chunks.append(chunk) + total += 1 + if total > 10: + raise ValueError("varint too long") + if (chunk[0] & 0x80) == 0: + break + buf = b"".join(chunks) + length, _ = _varint_decode(buf, 0) + payload = b"" + while len(payload) < length: + got = await stream.read(length - len(payload)) + if not got: + raise ValueError("message truncated") + payload += got + return payload + + +async def _write_varint_prefixed(stream, message: bytes) -> None: + prefix = _varint_encode(len(message)) + await stream.write(prefix + message) + + +# --- Rust protobuf encoding/decoding for HopMessage/StopMessage --- +# Proto layout from circuit-relay-v2 message.proto: +# HopMessage: type(1), peer(2), reservation(3), limit(4), status(5) +# StopMessage: type(1), peer(2), limit(3), status(4) +# Peer: id(1), addrs(2) +# Reservation: expire(1), addrs(2), voucher(3) +# Limit: duration(1), data(2) + +def _parse_rust_hop_message(buf: bytes) -> dict: + out = {"type_pb": None, "status": None, "peer_id": None} + i = 0 + while i < len(buf): + tag = buf[i] + i += 1 + field_num = tag >> 3 + wire = tag & 7 + if wire == 0: + n, consumed = _varint_decode(buf, i) + i += consumed + if field_num == 1: + out["type_pb"] = n + elif field_num == 5: + out["status"] = n + elif wire == 2: + length, consumed = _varint_decode(buf, i) + i += consumed + end = i + length + if end > len(buf): + break + if field_num == 2: + out["peer_id"] = _extract_peer_id(buf[i:end]) + i = end + return out + + +def _extract_peer_id(peer_msg: bytes) -> bytes | None: + i = 0 + while i < len(peer_msg): + tag = peer_msg[i] + i += 1 + fn = tag >> 3 + w = tag & 7 + if w == 2 and fn == 1: + length, consumed = _varint_decode(peer_msg, i) + i += consumed + return bytes(peer_msg[i : i + length]) + elif w == 0: + _, consumed = _varint_decode(peer_msg, i) + i += consumed + elif w == 2: + length, consumed = _varint_decode(peer_msg, i) + i += consumed + length + return None + + +def _encode_hop_status_ok_with_reservation( + expire_unix: int, + relay_addrs: list[bytes], + duration: int, + data: int, +) -> bytes: + """Encode HopMessage { type=STATUS, status=OK, reservation={...}, limit={...} }.""" + parts = [] + # field 1: type = STATUS (2) + parts.append(b"\x08\x02") + # field 5: status = OK (100) + parts.append(b"\x28" + _varint_encode(RUST_STATUS_OK)) + # field 3: reservation message + res_parts = [] + # reservation.expire (field 1, varint) + res_parts.append(b"\x08" + _varint_encode(expire_unix)) + # reservation.addrs (field 2, repeated bytes) + for addr in relay_addrs: + res_parts.append(b"\x12" + _varint_encode(len(addr)) + addr) + res_inner = b"".join(res_parts) + parts.append(b"\x1a" + _varint_encode(len(res_inner)) + res_inner) + # field 4: limit message + lim_parts = [] + lim_parts.append(b"\x08" + _varint_encode(duration)) + lim_parts.append(b"\x10" + _varint_encode(data)) + lim_inner = b"".join(lim_parts) + parts.append(b"\x22" + _varint_encode(len(lim_inner)) + lim_inner) + return b"".join(parts) + + +def _encode_hop_status_error(status_code: int) -> bytes: + """HopMessage { type=STATUS, status= }.""" + return b"\x08\x02\x28" + _varint_encode(status_code) + + +def _encode_stop_connect(src_peer_id_bytes: bytes, duration: int, data: int) -> bytes: + """StopMessage { type=CONNECT, peer={id=src_peer_id}, limit={duration, data} }.""" + parts = [] + # field 1: type = CONNECT (0) + parts.append(b"\x08\x00") + # field 2: peer message + peer_inner = b"\x0a" + _varint_encode(len(src_peer_id_bytes)) + src_peer_id_bytes + parts.append(b"\x12" + _varint_encode(len(peer_inner)) + peer_inner) + # field 3: limit message + lim_parts = [] + lim_parts.append(b"\x08" + _varint_encode(duration)) + lim_parts.append(b"\x10" + _varint_encode(data)) + lim_inner = b"".join(lim_parts) + parts.append(b"\x1a" + _varint_encode(len(lim_inner)) + lim_inner) + return b"".join(parts) + + +def _parse_rust_stop_response(buf: bytes) -> dict: + """Parse StopMessage STATUS response: type(1), status(4).""" + out = {"type_pb": None, "status": None} + i = 0 + while i < len(buf): + if i >= len(buf): + break + tag = buf[i] + i += 1 + field_num = tag >> 3 + wire = tag & 7 + if wire == 0: + n, consumed = _varint_decode(buf, i) + i += consumed + if field_num == 1: + out["type_pb"] = n + elif field_num == 4: + out["status"] = n + elif wire == 2: + length, consumed = _varint_decode(buf, i) + i += consumed + i += length + return out + + +class HolePunchRelay: + def __init__(self): + self.redis_addr = os.getenv("REDIS_ADDR") + self.test_key = os.getenv("TEST_KEY") + self.transport = os.getenv("TRANSPORT", "tcp") + self.secure_channel = os.getenv("SECURE_CHANNEL") + self.muxer = os.getenv("MUXER") + self.relay_ip = os.getenv("RELAY_IP", "0.0.0.0") + self.debug = os.getenv("DEBUG", "false").lower() == "true" + + if not self.redis_addr: + raise ValueError("REDIS_ADDR environment variable is required") + if not self.test_key: + raise ValueError("TEST_KEY environment variable is required") + + if ":" in self.redis_addr: + host, port = self.redis_addr.split(":") + self.redis_host, self.redis_port = host, int(port) + else: + self.redis_host = self.redis_addr + self.redis_port = 6379 + + self.redis_client = None + self.host = None + # Peers that connected via 0.2.0/hop (Rust wire format) + self._rust_peers: set[str] = set() + self._relay_protocol = None + + def create_security_options(self): + key_pair = create_new_key_pair() + + if self.transport == "quic-v1": + return {}, key_pair + + if self.secure_channel == "noise": + noise_key_pair = create_new_x25519_key_pair() + noise_transport = NoiseTransport( + libp2p_keypair=key_pair, + noise_privkey=noise_key_pair.private_key, + early_data=None, + ) + return {NOISE_PROTOCOL_ID: noise_transport}, key_pair + elif self.secure_channel == "tls": + tls_transport = TLSTransport( + libp2p_keypair=key_pair, + early_data=None, + muxers=None, + ) + return {TLS_PROTOCOL_ID: tls_transport}, key_pair + else: + raise ValueError(f"Unsupported secure channel: {self.secure_channel}") + + def create_muxer_options(self): + if self.transport == "quic-v1": + return None + + if self.muxer == "yamux": + return create_yamux_muxer_option() + elif self.muxer == "mplex": + return create_mplex_muxer_option() + else: + raise ValueError(f"Unsupported muxer: {self.muxer}") + + def create_listen_address(self, port: int = 4001): + if self.transport == "tcp": + return multiaddr.Multiaddr(f"/ip4/{self.relay_ip}/tcp/{port}") + elif self.transport == "quic-v1": + return multiaddr.Multiaddr(f"/ip4/{self.relay_ip}/udp/{port}/quic-v1") + elif self.transport == "ws": + return multiaddr.Multiaddr(f"/ip4/{self.relay_ip}/tcp/{port}/ws") + elif self.transport == "wss": + return multiaddr.Multiaddr(f"/ip4/{self.relay_ip}/tcp/{port}/wss") + else: + raise ValueError(f"Unsupported transport: {self.transport}") + + async def connect_redis(self): + print(f"Connecting to Redis at {self.redis_host}:{self.redis_port}...", file=sys.stderr) + for attempt in range(10): + try: + self.redis_client = redis.Redis( + host=self.redis_host, + port=self.redis_port, + decode_responses=True + ) + self.redis_client.ping() + print(f"Connected to Redis on attempt {attempt + 1}", file=sys.stderr) + return + except Exception as e: + print(f"Redis connection attempt {attempt + 1} failed: {e}", file=sys.stderr) + if attempt < 9: + await trio.sleep(1) + raise RuntimeError("Failed to connect to Redis after 10 attempts") + + # ---- Rust HOP handler (varint-length-delimited, 0.2.0 proto layout) ---- + + async def _handle_rust_hop_stream(self, stream) -> None: + """Handle HOP streams from Rust peers (0.2.0/hop, varint framing).""" + remote_peer_id = stream.muxed_conn.peer_id + remote_id = str(remote_peer_id) + self._rust_peers.add(remote_id) + print(f"Rust HOP stream from {remote_id}", file=sys.stderr) + try: + payload = await _read_varint_prefixed(stream) + parsed = _parse_rust_hop_message(payload) + msg_type = parsed.get("type_pb") + + if msg_type == 0: # RESERVE + await self._handle_rust_reserve(stream, remote_peer_id) + elif msg_type == 1: # CONNECT + dst_peer_id_bytes = parsed.get("peer_id") + if not dst_peer_id_bytes: + await _write_varint_prefixed(stream, _encode_hop_status_error(400)) + return + dst_peer_id = ID(dst_peer_id_bytes) + await self._handle_rust_connect(stream, remote_peer_id, dst_peer_id) + else: + await _write_varint_prefixed(stream, _encode_hop_status_error(401)) + except Exception as e: + print(f"Error in Rust HOP handler from {remote_id}: {e}", file=sys.stderr) + try: + await _write_varint_prefixed(stream, _encode_hop_status_error(400)) + except Exception: + pass + + async def _handle_rust_reserve(self, stream, remote_peer_id) -> None: + """Process a Rust RESERVE request and send varint-delimited response.""" + rp = self._relay_protocol + peer_id = remote_peer_id + pid_str = str(peer_id) + + if rp.resource_manager.has_reservation(peer_id): + ttl = rp.resource_manager.refresh_reservation(peer_id) + elif rp.resource_manager.can_accept_reservation(peer_id): + ttl = rp.resource_manager.reserve(peer_id) + else: + print(f"Reservation limit exceeded for {pid_str}", file=sys.stderr) + await _write_varint_prefixed( + stream, _encode_hop_status_error(RUST_STATUS_RESOURCE_LIMIT_EXCEEDED) + ) + return + + expire_unix = int(time.time() + ttl) + relay_addrs = [] + try: + for addr in self.host.get_peerstore().addrs(self.host.get_id()): + relay_addrs.append( + multiaddr.Multiaddr( + str(addr) + "/p2p/" + str(self.host.get_id()) + ).to_bytes() + ) + except Exception: + pass + if not relay_addrs: + relay_addrs.append(self._listen_addr_with_p2p().to_bytes()) + + response = _encode_hop_status_ok_with_reservation( + expire_unix=expire_unix, + relay_addrs=relay_addrs, + duration=3600, + data=100 * 1024 * 1024, + ) + await _write_varint_prefixed(stream, response) + print(f"Reservation OK for Rust peer {pid_str} (ttl={ttl}s)", file=sys.stderr) + + async def _handle_rust_connect(self, hop_stream, src_peer_id, dst_peer_id) -> None: + """ + Handle a Rust CONNECT request: open STOP to destination, relay data. + The source (dialer) is Rust so HOP uses varint framing. + """ + dst_id_str = str(dst_peer_id) + is_rust_dst = dst_id_str in self._rust_peers + print( + f"Rust CONNECT: src={src_peer_id} -> dst={dst_peer_id} " + f"(dst_rust={is_rust_dst})", + file=sys.stderr, + ) + + try: + dst_stream = await self._open_stop_to_peer(dst_peer_id, src_peer_id, is_rust_dst) + except Exception as e: + print(f"Failed STOP to {dst_id_str}: {e}", file=sys.stderr) + await _write_varint_prefixed(hop_stream, _encode_hop_status_error(203)) + return + + # Send HOP STATUS OK to the Rust source (varint-delimited) + await _write_varint_prefixed(hop_stream, b"\x08\x02\x28\x64") + print(f"Relaying: {src_peer_id} <-> {dst_peer_id}", file=sys.stderr) + + async with trio.open_nursery() as nursery: + nursery.start_soon(self._relay_data, hop_stream, dst_stream) + nursery.start_soon(self._relay_data, dst_stream, hop_stream) + + # ---- Wrapped Python HOP CONNECT handler (destination may be Rust) ---- + + async def _handle_py_hop_stream(self, stream) -> None: + """ + Handle HOP from Python peers. Delegates to the default py-libp2p handler + for RESERVE but intercepts CONNECT to support Rust destinations. + """ + remote_peer_id = stream.muxed_conn.peer_id + from libp2p.relay.circuit_v2.protocol import HopMessage + try: + while True: + with trio.fail_after(60): + msg_bytes = await stream.read(1024) + if not msg_bytes: + return + + hop_msg = HopMessage() + hop_msg.ParseFromString(msg_bytes) + + if hop_msg.type == HopMessage.RESERVE: + await self._relay_protocol._handle_reserve(stream, hop_msg) + continue + + if hop_msg.type == HopMessage.CONNECT: + dst_peer_id = ID(hop_msg.peer) + dst_id_str = str(dst_peer_id) + + if dst_id_str in self._rust_peers: + await self._handle_py_connect_to_rust( + stream, remote_peer_id, dst_peer_id + ) + else: + await self._relay_protocol._handle_connect(stream, hop_msg) + return + + except trio.TooSlowError: + return + except Exception as e: + print( + f"Unexpected error handling hop stream from " + f"{remote_peer_id}: {e}", + file=sys.stderr, + ) + + async def _handle_py_connect_to_rust(self, hop_stream, src_peer_id, dst_peer_id) -> None: + """Python dialer CONNECT → Rust listener: open Rust STOP, relay.""" + print( + f"Py CONNECT to Rust dst: src={src_peer_id} -> dst={dst_peer_id}", + file=sys.stderr, + ) + try: + dst_stream = await self._open_stop_to_peer( + dst_peer_id, src_peer_id, is_rust_dst=True + ) + except Exception as e: + print(f"Failed STOP to Rust {dst_peer_id}: {e}", file=sys.stderr) + await self._send_py_hop_error(hop_stream, f"Failed: {e}") + return + + # Send py-format HOP STATUS OK to the Python source + await self._send_py_hop_ok(hop_stream) + print(f"Relaying: {src_peer_id} <-> {dst_peer_id}", file=sys.stderr) + + async with trio.open_nursery() as nursery: + nursery.start_soon(self._relay_data, hop_stream, dst_stream) + nursery.start_soon(self._relay_data, dst_stream, hop_stream) + + # ---- STOP stream opener (supports both Python and Rust destinations) ---- + + async def _open_stop_to_peer(self, dst_peer_id, src_peer_id, is_rust_dst: bool): + """Open a STOP stream to the destination and perform the CONNECT handshake.""" + if is_rust_dst: + return await self._open_rust_stop(dst_peer_id, src_peer_id) + else: + return await self._open_py_stop(dst_peer_id, src_peer_id) + + async def _open_rust_stop(self, dst_peer_id, src_peer_id): + """Open STOP to a Rust peer: 0.2.0/stop + varint framing.""" + dst_stream = await self.host.new_stream(dst_peer_id, [RUST_STOP_PROTOCOL_ID]) + stop_msg = _encode_stop_connect( + src_peer_id_bytes=src_peer_id.to_bytes(), + duration=3600, + data=100 * 1024 * 1024, + ) + await _write_varint_prefixed(dst_stream, stop_msg) + resp_payload = await _read_varint_prefixed(dst_stream) + parsed = _parse_rust_stop_response(resp_payload) + if parsed.get("status") != RUST_STATUS_OK: + await dst_stream.close() + raise ConnectionError( + f"Rust STOP rejected: status={parsed.get('status')}" + ) + return dst_stream + + async def _open_py_stop(self, dst_peer_id, src_peer_id): + """Open STOP to a Python peer: 2.0.0/stop + raw protobuf.""" + from libp2p.relay.circuit_v2.protocol import StopMessage + + dst_stream = await self.host.new_stream(dst_peer_id, [STOP_PROTOCOL_ID]) + stop_msg = StopMessage( + type=StopMessage.CONNECT, + peer=src_peer_id.to_bytes(), + ) + await dst_stream.write(stop_msg.SerializeToString()) + resp_bytes = await dst_stream.read(1024) + resp = StopMessage() + resp.ParseFromString(resp_bytes) + if resp.HasField("status"): + from libp2p.relay.circuit_v2.protocol import StatusCode + status_code = getattr(resp.status, "code", StatusCode.OK) + if status_code != StatusCode.OK: + await dst_stream.close() + raise ConnectionError(f"Py STOP rejected: {status_code}") + return dst_stream + + # ---- Helpers ---- + + async def _send_py_hop_ok(self, stream) -> None: + """Send a Python-format HOP STATUS OK response.""" + from libp2p.relay.circuit_v2.protocol import HopMessage, StatusCode + from libp2p.relay.circuit_v2.protocol import create_status + + status = create_status(code=StatusCode.OK, message="OK") + resp = HopMessage(type=HopMessage.STATUS, status=status) + await stream.write(resp.SerializeToString()) + + async def _send_py_hop_error(self, stream, message: str) -> None: + from libp2p.relay.circuit_v2.protocol import HopMessage, StatusCode + from libp2p.relay.circuit_v2.protocol import create_status + + status = create_status(code=StatusCode.CONNECTION_FAILED, message=message) + resp = HopMessage(type=HopMessage.STATUS, status=status) + try: + await stream.write(resp.SerializeToString()) + except Exception: + pass + + async def _relay_data(self, src, dst) -> None: + try: + while True: + data = await src.read(4096) + if not data: + break + await dst.write(data) + except Exception: + pass + + def _listen_addr_with_p2p(self): + return multiaddr.Multiaddr( + str(self.create_listen_address(4001)) + + "/p2p/" + + str(self.host.get_id()) + ) + + async def run(self): + print("Starting RELAY server...", file=sys.stderr) + print(f" TRANSPORT: {self.transport}", file=sys.stderr) + print(f" SECURE_CHANNEL: {self.secure_channel}", file=sys.stderr) + print(f" MUXER: {self.muxer}", file=sys.stderr) + print(f" RELAY_IP: {self.relay_ip}", file=sys.stderr) + + await self.connect_redis() + + sec_opt, key_pair = self.create_security_options() + muxer_opt = self.create_muxer_options() + listen_addr = self.create_listen_address(4001) + + self.host = new_host( + key_pair=key_pair, + sec_opt=sec_opt, + muxer_opt=muxer_opt, + enable_quic=(self.transport == "quic-v1"), + listen_addrs=[listen_addr], + ) + + limits = RelayLimits( + duration=3600, + data=100 * 1024 * 1024, + max_circuit_conns=10, + max_reservations=5, + ) + + relay_config = RelayConfig( + roles=RelayRole.HOP | RelayRole.STOP | RelayRole.CLIENT, + limits=limits, + ) + + self._relay_protocol = CircuitV2Protocol( + self.host, limits=limits, allow_hop=True + ) + + async with self.host.run(listen_addrs=[listen_addr]): + async with background_trio_service(self._relay_protocol): + await self._relay_protocol.event_started.wait() + + # Register handlers AFTER relay protocol startup so ours overwrite + # its defaults (otherwise CONNECT would use 2.0.0/stop for Rust dst). + self.host.set_stream_handler( + PROTOCOL_ID, self._handle_py_hop_stream + ) + self.host.set_stream_handler( + STOP_PROTOCOL_ID, + self._relay_protocol._handle_stop_stream, + ) + self.host.set_stream_handler( + RUST_HOP_PROTOCOL_ID, + self._handle_rust_hop_stream, + ) + self.host.set_stream_handler( + RUST_STOP_PROTOCOL_ID, + self._relay_protocol._handle_stop_stream, + ) + + CircuitV2Transport(self.host, self._relay_protocol, relay_config) + + peer_id = str(self.host.get_id()) + relay_multiaddr = f"{str(listen_addr)}/p2p/{peer_id}" + + print(f"Relay peer ID: {peer_id}", file=sys.stderr) + print(f"Relay multiaddr: {relay_multiaddr}", file=sys.stderr) + + redis_key = f"{self.test_key}_relay_multiaddr" + self.redis_client.set(redis_key, relay_multiaddr) + print( + f"Published relay multiaddr to Redis key: {redis_key}", + file=sys.stderr, + ) + + print("Relay server ready, accepting connections...", file=sys.stderr) + await trio.sleep_forever() + + +def configure_logging(): + debug = os.getenv("DEBUG", "false").lower() in ["true", "1", "yes", "debug"] + level = logging.DEBUG if debug else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s | %(name)s | %(levelname)s | %(message)s", + stream=sys.stderr, + ) + if not debug: + logging.getLogger("libp2p").setLevel(logging.WARNING) + + +async def main(): + configure_logging() + try: + relay = HolePunchRelay() + await relay.run() + except Exception as e: + print(f"Fatal error: {e}", file=sys.stderr) + import traceback + traceback.print_exc(file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + trio.run(main) diff --git a/hole-punch/images/python/v0.5/requirements.txt b/hole-punch/images/python/v0.5/requirements.txt new file mode 100644 index 00000000..71acbce8 --- /dev/null +++ b/hole-punch/images/python/v0.5/requirements.txt @@ -0,0 +1,14 @@ +# py-libp2p with DCUtR support +libp2p @ git+https://github.com/libp2p/py-libp2p.git@5723418a4a3dd8d8f7648e29f38a10ca650047a5 + +# Redis client for test coordination +redis>=4.0.0 + +# Async framework (py-libp2p uses trio) +trio>=0.26.0 + +# Multiaddr handling +multiaddr>=0.0.9 + +# HTTP requests (required by py-libp2p's auto-TLS/ACME) +requests>=2.25.0 diff --git a/hole-punch/images/python/v0.5/run-peer.sh b/hole-punch/images/python/v0.5/run-peer.sh new file mode 100644 index 00000000..c04fcbed --- /dev/null +++ b/hole-punch/images/python/v0.5/run-peer.sh @@ -0,0 +1,10 @@ +#!/usr/bin/env bash +set -e + +# Set route to WAN subnet via router (required for NAT traversal) +if [ -n "${WAN_SUBNET}" ] && [ -n "${WAN_ROUTER_IP}" ]; then + echo "Setting route to WAN subnet ${WAN_SUBNET} via ${WAN_ROUTER_IP}" >&2 + ip route add "${WAN_SUBNET}" via "${WAN_ROUTER_IP}" dev lan0 || true +fi + +exec python /app/peer.py "$@" diff --git a/hole-punch/images/python/v0.5/run-relay.sh b/hole-punch/images/python/v0.5/run-relay.sh new file mode 100644 index 00000000..5fd8588b --- /dev/null +++ b/hole-punch/images/python/v0.5/run-relay.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -e + +exec python /app/relay.py "$@" diff --git a/hole-punch/images/rust/v0.56/src/bin/peer.rs b/hole-punch/images/rust/v0.56/src/bin/peer.rs index b8b83d74..77a79501 100644 --- a/hole-punch/images/rust/v0.56/src/bin/peer.rs +++ b/hole-punch/images/rust/v0.56/src/bin/peer.rs @@ -204,7 +204,7 @@ async fn run_listener( } } - if sent_observed_addr && my_observed_addr.is_some() { + if my_observed_addr.is_some() { break my_observed_addr.unwrap() } }; @@ -221,11 +221,16 @@ async fn run_listener( swarm .listen_on(relayed_listener_addr.clone()) .expect("failed to listen on p2p circuit"); - - // Publish to Redis with TEST_KEY namespacing + let listener_peer_id_key = format!("{test_key}_listener_peer_id"); + + // Publish peer id so the dialer can discover us. With a Rust relay we get + // ReservationReqAccepted and could publish then; with a Python relay the + // wire format may not emit that event (§11 relay-reservation-wire-comparison.md), + // so we also publish after a short delay so py×rust tests pass with either relay. + tokio::time::sleep(Duration::from_secs(2)).await; let _: () = con - .set(&listener_peer_id_key, relayed_listener_addr.to_string()) + .set(&listener_peer_id_key, peer_id.to_string()) .await .expect(&format!( "Failed to publish peer id to Redis: (key: {listener_peer_id_key})" diff --git a/hole-punch/images/rust/v0.56/src/bin/relay.rs b/hole-punch/images/rust/v0.56/src/bin/relay.rs index 429c594e..f4caee82 100644 --- a/hole-punch/images/rust/v0.56/src/bin/relay.rs +++ b/hole-punch/images/rust/v0.56/src/bin/relay.rs @@ -118,6 +118,13 @@ async fn main() -> Result<()> { let full_multiaddr = format!("{address}/p2p/{peer_id}"); eprintln!("Listening on: {full_multiaddr}"); + // Register the listen address as an external address so the relay + // behaviour includes it in RESERVE OK responses. Without this, + // clients get NoAddressesInReservation because relay::Behaviour + // only includes confirmed external addresses. + swarm.add_external_address(address.clone()); + eprintln!("Registered external address: {address}"); + // Publish to Redis with TEST_KEY namespacing let relay_addr_key = format!("{test_key}_relay_multiaddr"); let _: () = con diff --git a/hole-punch/lib/generate-tests.sh b/hole-punch/lib/generate-tests.sh index 4df055c2..52538a1b 100755 --- a/hole-punch/lib/generate-tests.sh +++ b/hole-punch/lib/generate-tests.sh @@ -556,8 +556,8 @@ generate_tests_worker() { print_debug "selecting listener router: ${listener_router_id}" - # Iterate through all dialers - for dialer_id in "${all_image_ids[@]}"; do + # Iterate through this worker's chunk of dialers + for dialer_id in "${dialer_chunk[@]}"; do dialer_transports="${image_transports[$dialer_id]}" dialer_secure="${image_secure[$dialer_id]}" dialer_muxers="${image_muxers[$dialer_id]}" @@ -679,7 +679,7 @@ generate_tests_worker() { id: ${dialer_id} imageName: ${dialer_image_name} EOF - if [ -n "${dialer_commit}" ]; then + if [ -n "${dialer_commit}" ] && [ "${dialer_commit}" != "null" ]; then echo " snapshot: snapshots/${dialer_commit}.zip" >> "${worker_selected}" fi if [ "${dialer_legacy}" == "true" ]; then @@ -690,7 +690,7 @@ EOF id: ${listener_id} imageName: ${listener_image_name} EOF - if [ -n "${listener_commit}" ]; then + if [ -n "${listener_commit}" ] && [ "${listener_commit}" != "null" ]; then echo " snapshot: snapshots/${listener_commit}.zip" >> "${worker_selected}" fi if [ "${listener_legacy}" == "true" ]; then @@ -701,7 +701,7 @@ EOF id: ${relay_id} imageName: ${relay_image_name} EOF - if [ -n "${relay_commit}" ]; then + if [ -n "${relay_commit}" ] && [ "${relay_commit}" != "null" ]; then echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_selected}" fi cat >> "${worker_selected}" <> "${worker_selected}" fi cat >> "${worker_selected}" <> "${worker_selected}" fi @@ -736,7 +736,7 @@ EOF id: ${dialer_id} imageName: ${dialer_image_name} EOF - if [ -n "${dialer_commit}" ]; then + if [ -n "${dialer_commit}" ] && [ "${dialer_commit}" != "null" ]; then echo " snapshot: snapshots/${dialer_commit}.zip" >> "${worker_ignored}" fi if [ "${dialer_legacy}" == "true" ]; then @@ -747,7 +747,7 @@ EOF id: ${listener_id} imageName: ${listener_image_name} EOF - if [ -n "${listener_commit}" ]; then + if [ -n "${listener_commit}" ] && [ "${listener_commit}" != "null" ]; then echo " snapshot: snapshots/${listener_commit}.zip" >> "${worker_ignored}" fi if [ "${listener_legacy}" == "true" ]; then @@ -758,24 +758,24 @@ EOF id: ${relay_id} imageName: ${relay_image_name} EOF - if [ -n "${relay_commit}" ]; then - echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_selected}" + if [ -n "${relay_commit}" ] && [ "${relay_commit}" != "null" ]; then + echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_selected}" + if [ -n "${dialer_router_commit}" ] && [ "${dialer_router_commit}" != "null" ]; then + echo " snapshot: snapshots/${dialer_router_commit}.zip" >> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_selected}" + if [ -n "${listener_router_commit}" ] && [ "${listener_router_commit}" != "null" ]; then + echo " snapshot: snapshots/${listener_router_commit}.zip" >> "${worker_ignored}" fi fi @@ -861,7 +861,7 @@ EOF id: ${dialer_id} imageName: ${dialer_image_name} EOF - if [ -n "${dialer_commit}" ]; then + if [ -n "${dialer_commit}" ] && [ "${dialer_commit}" != "null" ]; then echo " snapshot: snapshots/${dialer_commit}.zip" >> "${worker_selected}" fi if [ "${dialer_legacy}" == "true" ]; then @@ -872,7 +872,7 @@ EOF id: ${listener_id} imageName: ${listener_image_name} EOF - if [ -n "${listener_commit}" ]; then + if [ -n "${listener_commit}" ] && [ "${listener_commit}" != "null" ]; then echo " snapshot: snapshots/${listener_commit}.zip" >> "${worker_selected}" fi if [ "${listener_legacy}" == "true" ]; then @@ -883,7 +883,7 @@ EOF id: ${relay_id} imageName: ${relay_image_name} EOF - if [ -n "${relay_commit}" ]; then + if [ -n "${relay_commit}" ] && [ "${relay_commit}" != "null" ]; then echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_selected}" fi cat >> "${worker_selected}" <> "${worker_selected}" fi cat >> "${worker_selected}" <> "${worker_selected}" fi else @@ -916,7 +916,7 @@ EOF id: ${dialer_id} imageName: ${dialer_image_name} EOF - if [ -n "${dialer_commit}" ]; then + if [ -n "${dialer_commit}" ] && [ "${dialer_commit}" != "null" ]; then echo " snapshot: snapshots/${dialer_commit}.zip" >> "${worker_ignored}" fi if [ "${dialer_legacy}" == "true" ]; then @@ -927,7 +927,7 @@ EOF id: ${listener_id} imageName: ${listener_image_name} EOF - if [ -n "${listener_commit}" ]; then + if [ -n "${listener_commit}" ] && [ "${listener_commit}" != "null" ]; then echo " snapshot: snapshots/${listener_commit}.zip" >> "${worker_ignored}" fi if [ "${listener_legacy}" == "true" ]; then @@ -938,24 +938,24 @@ EOF id: ${relay_id} imageName: ${relay_image_name} EOF - if [ -n "${relay_commit}" ]; then - echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_selected}" + if [ -n "${relay_commit}" ] && [ "${relay_commit}" != "null" ]; then + echo " snapshot: snapshots/${relay_commit}.zip" >> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_selected}" + if [ -n "${dialer_router_commit}" ] && [ "${dialer_router_commit}" != "null" ]; then + echo " snapshot: snapshots/${dialer_router_commit}.zip" >> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_selected}" + if [ -n "${listener_router_commit}" ] && [ "${listener_router_commit}" != "null" ]; then + echo " snapshot: snapshots/${listener_router_commit}.zip" >> "${worker_ignored}" fi fi done diff --git a/hole-punch/lib/run-single-test.sh b/hole-punch/lib/run-single-test.sh index d0157ed8..aff33f09 100755 --- a/hole-punch/lib/run-single-test.sh +++ b/hole-punch/lib/run-single-test.sh @@ -171,6 +171,7 @@ else - REDIS_ADDR=hole-punch-redis:6379 - TEST_KEY=${TEST_KEY} - TRANSPORT=${TRANSPORT_NAME} + - RELAY_ID=${RELAY_ID} - DIALER_IP=${DIALER_IP} - WAN_SUBNET=${WAN_SUBNET} - WAN_ROUTER_IP=${DIALER_ROUTER_LAN_IP} @@ -198,6 +199,7 @@ else - REDIS_ADDR=hole-punch-redis:6379 - TEST_KEY=${TEST_KEY} - TRANSPORT=${TRANSPORT_NAME} + - RELAY_ID=${RELAY_ID} - LISTENER_IP=${LISTENER_IP} - WAN_SUBNET=${WAN_SUBNET} - WAN_ROUTER_IP=${LISTENER_ROUTER_LAN_IP} From 9e22f1608a64ea05311748375f7c2b6251b62edd Mon Sep 17 00:00:00 2001 From: Sumanjeet Date: Wed, 17 Jun 2026 01:01:31 +0530 Subject: [PATCH 2/7] Update images.yaml --- hole-punch/images.yaml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/hole-punch/images.yaml b/hole-punch/images.yaml index 70259b3b..9027b6b5 100644 --- a/hole-punch/images.yaml +++ b/hole-punch/images.yaml @@ -33,10 +33,10 @@ relays: secureChannels: [noise, tls] muxers: [yamux, mplex] - - id: py-v0.5 + - id: py-v0.6 source: type: local - path: images/python/v0.5 + path: images/python/v0.6 dockerfile: Dockerfile.relay transports: [tcp, ws] secureChannels: [noise] @@ -53,10 +53,10 @@ implementations: secureChannels: [noise, tls] muxers: [yamux, mplex] - - id: py-v0.5 + - id: py-v0.6 source: type: local - path: images/python/v0.5 + path: images/python/v0.6 dockerfile: Dockerfile.peer transports: [tcp, ws] secureChannels: [noise] From fa1151aaf40a863396f32f7b47cd6e930759face Mon Sep 17 00:00:00 2001 From: sumanjeet0012 Date: Wed, 24 Jun 2026 00:00:34 +0530 Subject: [PATCH 3/7] updated py-libp2p reference commit --- hole-punch/README.md | 7 +- hole-punch/images.yaml | 4 +- .../python/{v0.5 => v0.6}/Dockerfile.peer | 0 .../python/{v0.5 => v0.6}/Dockerfile.relay | 0 .../images/python/{v0.5 => v0.6}/peer.py | 68 +++++++++++++++++-- .../images/python/{v0.5 => v0.6}/relay.py | 8 ++- .../python/{v0.5 => v0.6}/requirements.txt | 2 +- .../images/python/{v0.5 => v0.6}/run-peer.sh | 0 .../images/python/{v0.5 => v0.6}/run-relay.sh | 0 lib/check-dependencies.sh | 2 +- 10 files changed, 77 insertions(+), 14 deletions(-) rename hole-punch/images/python/{v0.5 => v0.6}/Dockerfile.peer (100%) rename hole-punch/images/python/{v0.5 => v0.6}/Dockerfile.relay (100%) rename hole-punch/images/python/{v0.5 => v0.6}/peer.py (91%) rename hole-punch/images/python/{v0.5 => v0.6}/relay.py (99%) rename hole-punch/images/python/{v0.5 => v0.6}/requirements.txt (72%) rename hole-punch/images/python/{v0.5 => v0.6}/run-peer.sh (100%) rename hole-punch/images/python/{v0.5 => v0.6}/run-relay.sh (100%) diff --git a/hole-punch/README.md b/hole-punch/README.md index ff463e09..6a0422fb 100644 --- a/hole-punch/README.md +++ b/hole-punch/README.md @@ -8,8 +8,9 @@ The hole-punch test suite verifies that libp2p implementations can establish dir **Current implementations**: - **rust-libp2p**: v0.56 +- **py-libp2p**: v0.6 - **Router**: Linux-based NAT router -- **Relay**: rust-libp2p v0.56 relay server +- **Relay**: rust-libp2p v0.56, py-libp2p v0.6 relay servers **Protocols tested**: - **Transports**: tcp, quic-v1, webrtc-direct, ws (WebSocket) @@ -175,7 +176,7 @@ Hole-punch tests support a powerful **two-stage filtering model** with both `--* ./run.sh --impl-select "rust-v0.56" # SELECT: Run multiple implementations (pipe-separated) -./run.sh --impl-select "rust-v0.56|go-v0.45" +./run.sh --impl-select "rust-v0.56|py-v0.6" # SELECT with alias: Run all rust versions ./run.sh --impl-select "~rust" @@ -771,7 +772,7 @@ docker ps | grep "hole-punch" # Check what will run ./run.sh --list-tests -# Run all tests (currently only rust-v0.56) +# Run all tests ./run.sh # Run specific transport only diff --git a/hole-punch/images.yaml b/hole-punch/images.yaml index 9027b6b5..3429822e 100644 --- a/hole-punch/images.yaml +++ b/hole-punch/images.yaml @@ -39,7 +39,7 @@ relays: path: images/python/v0.6 dockerfile: Dockerfile.relay transports: [tcp, ws] - secureChannels: [noise] + secureChannels: [noise, tls] muxers: [yamux] # Implementation definitions @@ -59,5 +59,5 @@ implementations: path: images/python/v0.6 dockerfile: Dockerfile.peer transports: [tcp, ws] - secureChannels: [noise] + secureChannels: [noise, tls] muxers: [yamux] diff --git a/hole-punch/images/python/v0.5/Dockerfile.peer b/hole-punch/images/python/v0.6/Dockerfile.peer similarity index 100% rename from hole-punch/images/python/v0.5/Dockerfile.peer rename to hole-punch/images/python/v0.6/Dockerfile.peer diff --git a/hole-punch/images/python/v0.5/Dockerfile.relay b/hole-punch/images/python/v0.6/Dockerfile.relay similarity index 100% rename from hole-punch/images/python/v0.5/Dockerfile.relay rename to hole-punch/images/python/v0.6/Dockerfile.relay diff --git a/hole-punch/images/python/v0.5/peer.py b/hole-punch/images/python/v0.6/peer.py similarity index 91% rename from hole-punch/images/python/v0.5/peer.py rename to hole-punch/images/python/v0.6/peer.py index 90a8585b..983acf9b 100644 --- a/hole-punch/images/python/v0.5/peer.py +++ b/hole-punch/images/python/v0.6/peer.py @@ -20,7 +20,7 @@ from libp2p.peer.id import ID from libp2p.relay.circuit_v2.dcutr import DCUtRProtocol from libp2p.relay.circuit_v2.protocol import CircuitV2Protocol, DEFAULT_RELAY_LIMITS -from libp2p.relay.circuit_v2.config import RelayConfig, RelayRole +from libp2p.relay.circuit_v2.config import RelayConfig, RelayRole, RelayLimits import libp2p.relay.circuit_v2.transport as _circuit_transport_mod from libp2p.relay.circuit_v2.transport import CircuitV2Transport from libp2p.security.noise.transport import ( @@ -31,7 +31,7 @@ PROTOCOL_ID as TLS_PROTOCOL_ID, TLSTransport, ) -from libp2p.tools.async_service import background_trio_service +from libp2p.tools.anyio_service import background_trio_service from libp2p.network.connection.raw_connection import RawConnection logger = logging.getLogger("hole-punch-peer") @@ -374,11 +374,19 @@ async def run_listener(self): listen_addrs=[listen_addr], ) + limits = RelayLimits( + duration=3600, + data=100 * 1024 * 1024, + max_circuit_conns=1000, + max_reservations=500, + ) + # Configure relay client relay_config = RelayConfig( roles=RelayRole.STOP | RelayRole.CLIENT, + limits=limits, ) - relay_protocol = CircuitV2Protocol(self.host, DEFAULT_RELAY_LIMITS, allow_hop=False) + relay_protocol = CircuitV2Protocol(self.host, limits, allow_hop=False) dcutr_protocol = DCUtRProtocol(self.host) async with self.host.run(listen_addrs=[listen_addr]): @@ -576,7 +584,24 @@ async def run_dialer(self): raw_conn, listener_peer_id ) else: - await transport.dial(circuit_addr) + for attempt in range(3): + try: + stream = await self._dial_python_relay( + relay_info, listener_peer_id + ) + if stream: + raw_conn = RawConnection(stream=stream, initiator=True) + await self.host.upgrade_outbound_connection( + raw_conn, listener_peer_id + ) + break + except Exception as e: + import traceback + print(f"Dial attempt {attempt + 1} failed: {e}", file=sys.stderr) + traceback.print_exc() + if attempt == 2: + raise + await trio.sleep(1.0) print("Connected to listener through relay", file=sys.stderr) # Initiate hole punch @@ -681,6 +706,41 @@ async def _make_reservation_rust_relay(self, hop_stream, relay_peer_id) -> bool: logger.debug("Rust relay reservation failed: %s", e) return False + async def _dial_python_relay(self, relay_info, listener_peer_id): + hop_stream = await self.host.new_stream( + relay_info.peer_id, [_circuit_transport_mod.PROTOCOL_ID] + ) + if not hop_stream: + raise ConnectionError("Failed to open stream to relay") + + try: + from libp2p.relay.circuit_v2.pb.circuit_pb2 import HopMessage + from libp2p.peer.peerstore import env_to_send_in_RPC + import trio + + envelope_bytes, _ = env_to_send_in_RPC(self.host) + connect_msg = HopMessage( + type=HopMessage.CONNECT, + peer=listener_peer_id.to_bytes(), + senderRecord=envelope_bytes, + ) + await hop_stream.write(connect_msg.SerializeToString()) + + with trio.fail_after(10): + resp_bytes = await hop_stream.read(1024) + resp = HopMessage() + resp.ParseFromString(resp_bytes) + status_code = getattr(resp.status, "code", 200) + if status_code not in (0, 200): + raise ConnectionError(f"Relay rejected CONNECT with status {status_code}") + return hop_stream + except Exception: + try: + await hop_stream.close() + except Exception: + pass + raise + async def _dial_rust_relay( self, transport, diff --git a/hole-punch/images/python/v0.5/relay.py b/hole-punch/images/python/v0.6/relay.py similarity index 99% rename from hole-punch/images/python/v0.5/relay.py rename to hole-punch/images/python/v0.6/relay.py index 46c60af7..2a7266a9 100644 --- a/hole-punch/images/python/v0.5/relay.py +++ b/hole-punch/images/python/v0.6/relay.py @@ -35,7 +35,7 @@ PROTOCOL_ID as TLS_PROTOCOL_ID, TLSTransport, ) -from libp2p.tools.async_service import background_trio_service +from libp2p.tools.anyio_service import background_trio_service logger = logging.getLogger("hole-punch-relay") @@ -605,8 +605,8 @@ async def run(self): limits = RelayLimits( duration=3600, data=100 * 1024 * 1024, - max_circuit_conns=10, - max_reservations=5, + max_circuit_conns=1000, + max_reservations=500, ) relay_config = RelayConfig( @@ -618,6 +618,8 @@ async def run(self): self.host, limits=limits, allow_hop=True ) + + async with self.host.run(listen_addrs=[listen_addr]): async with background_trio_service(self._relay_protocol): await self._relay_protocol.event_started.wait() diff --git a/hole-punch/images/python/v0.5/requirements.txt b/hole-punch/images/python/v0.6/requirements.txt similarity index 72% rename from hole-punch/images/python/v0.5/requirements.txt rename to hole-punch/images/python/v0.6/requirements.txt index 71acbce8..d5f607bf 100644 --- a/hole-punch/images/python/v0.5/requirements.txt +++ b/hole-punch/images/python/v0.6/requirements.txt @@ -1,5 +1,5 @@ # py-libp2p with DCUtR support -libp2p @ git+https://github.com/libp2p/py-libp2p.git@5723418a4a3dd8d8f7648e29f38a10ca650047a5 +git+https://github.com/sumanjeet0012/py-libp2p.git@64a9e5bb0d4c5262571570cadd3bd33b2f504655 # Redis client for test coordination redis>=4.0.0 diff --git a/hole-punch/images/python/v0.5/run-peer.sh b/hole-punch/images/python/v0.6/run-peer.sh similarity index 100% rename from hole-punch/images/python/v0.5/run-peer.sh rename to hole-punch/images/python/v0.6/run-peer.sh diff --git a/hole-punch/images/python/v0.5/run-relay.sh b/hole-punch/images/python/v0.6/run-relay.sh similarity index 100% rename from hole-punch/images/python/v0.5/run-relay.sh rename to hole-punch/images/python/v0.6/run-relay.sh diff --git a/lib/check-dependencies.sh b/lib/check-dependencies.sh index e89d93cf..9725793d 100755 --- a/lib/check-dependencies.sh +++ b/lib/check-dependencies.sh @@ -159,7 +159,7 @@ print_check() { # Versioned tools: "tool:min_version:version_type" declare -a VERSIONED_TOOLS=( - "bash:4.0:bash" + "bash:3.0:bash" "docker:20.10.0:standard" "yq:4.0.0:standard" "git:2.0.0:standard" From e0329f4298d4cfd01b361f410c6bcb57fe154a50 Mon Sep 17 00:00:00 2001 From: Sumanjeet Date: Wed, 24 Jun 2026 19:22:24 +0530 Subject: [PATCH 4/7] removed failing alias --- hole-punch/images.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hole-punch/images.yaml b/hole-punch/images.yaml index 3429822e..5e220b7a 100644 --- a/hole-punch/images.yaml +++ b/hole-punch/images.yaml @@ -5,7 +5,7 @@ # Usage: --test-select "~linux" or --test-ignore "!~linux" test-aliases: - alias: "failing" - value: "rust-v0.56 x rust-v0.56" + value: "" - alias: "python" value: "py-v0.6" From 6377cab56a03efcdf3588197472668bb2eea30a2 Mon Sep 17 00:00:00 2001 From: sumanjeet0012 Date: Wed, 24 Jun 2026 19:27:01 +0530 Subject: [PATCH 5/7] Fix CI test-aliases and add delay to fix rust relay race condition --- hole-punch/images/python/v0.6/peer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/hole-punch/images/python/v0.6/peer.py b/hole-punch/images/python/v0.6/peer.py index 983acf9b..5653ea43 100644 --- a/hole-punch/images/python/v0.6/peer.py +++ b/hole-punch/images/python/v0.6/peer.py @@ -677,6 +677,7 @@ async def _handle_rust_stop_stream(self, stream, relay_protocol) -> None: pass return await _write_varint_prefixed_message(stream, _encode_rust_stop_status_ok()) + import trio; await trio.sleep(0.5) peer_id = ID(peer_id_bytes) remote_ma = f"/p2p/{peer_id.to_base58()}" await relay_protocol.handle_incoming_connection(stream, remote_ma) From fbcc44c392692321eaf8c8a261642e7b902da5b2 Mon Sep 17 00:00:00 2001 From: sumanjeet0012 Date: Wed, 24 Jun 2026 19:32:43 +0530 Subject: [PATCH 6/7] Fix hole-punch-interop-pr.yml skipping all tests due to negated impls --- .github/workflows/hole-punch-interop-pr.yml | 24 ++++++--------------- 1 file changed, 6 insertions(+), 18 deletions(-) diff --git a/.github/workflows/hole-punch-interop-pr.yml b/.github/workflows/hole-punch-interop-pr.yml index ba046449..de2a93a1 100644 --- a/.github/workflows/hole-punch-interop-pr.yml +++ b/.github/workflows/hole-punch-interop-pr.yml @@ -128,27 +128,15 @@ jobs: # Automatic trigger - use change detection outputs echo "test-select=${{ steps.detect.outputs.changed-impls }}" >> $GITHUB_OUTPUT - # Negate each implementation ID in the pipe-separated list - # Always include ~failing to skip known-broken tests - CHANGED_IMPLS="${{ steps.detect.outputs.changed-impls }}" - if [ -n "$CHANGED_IMPLS" ]; then - NEGATED_IMPLS=$(echo "$CHANGED_IMPLS" | sed 's/|/|!/g;s/^/!/') - echo "test-ignore=~failing|$NEGATED_IMPLS" >> $GITHUB_OUTPUT - else - echo "test-ignore=~failing" >> $GITHUB_OUTPUT - fi + # Keep only the known failing filter. test-select already scopes + # to changed implementations, so adding negated impl patterns here + # would conflict and can yield zero selected tests. + echo "test-ignore=~failing" >> $GITHUB_OUTPUT echo "relay-select=${{ steps.detect.outputs.changed-relays }}" >> $GITHUB_OUTPUT - # Negate each relay ID in the pipe-separated list - # Always include ~failing to skip known-broken relays - CHANGED_RELAYS="${{ steps.detect.outputs.changed-relays }}" - if [ -n "$CHANGED_RELAYS" ]; then - NEGATED_RELAYS=$(echo "$CHANGED_RELAYS" | sed 's/|/|!/g;s/^/!/') - echo "relay-ignore=~failing|$NEGATED_RELAYS" >> $GITHUB_OUTPUT - else - echo "relay-ignore=~failing" >> $GITHUB_OUTPUT - fi + # Same for relay ignore + echo "relay-ignore=~failing" >> $GITHUB_OUTPUT echo "router-select=${{ steps.detect.outputs.changed-routers }}" >> $GITHUB_OUTPUT From ce086fb63826ce8e5c869706e3f3580dde4ed98f Mon Sep 17 00:00:00 2001 From: sumanjeet0012 Date: Thu, 25 Jun 2026 18:58:12 +0530 Subject: [PATCH 7/7] Revert changes to Rust peer.rs and relay.rs --- hole-punch/images/rust/v0.56/src/bin/peer.rs | 17 +---------------- hole-punch/images/rust/v0.56/src/bin/relay.rs | 7 ------- 2 files changed, 1 insertion(+), 23 deletions(-) diff --git a/hole-punch/images/rust/v0.56/src/bin/peer.rs b/hole-punch/images/rust/v0.56/src/bin/peer.rs index f7f0f02d..b9266851 100644 --- a/hole-punch/images/rust/v0.56/src/bin/peer.rs +++ b/hole-punch/images/rust/v0.56/src/bin/peer.rs @@ -203,7 +203,7 @@ async fn run_listener( } } - if my_observed_addr.is_some() { + if sent_observed_addr && my_observed_addr.is_some() { break my_observed_addr.unwrap() } }; @@ -221,21 +221,6 @@ async fn run_listener( .listen_on(relayed_listener_addr.clone()) .expect("failed to listen on p2p circuit"); - let listener_peer_id_key = format!("{test_key}_listener_peer_id"); - - // Publish peer id so the dialer can discover us. With a Rust relay we get - // ReservationReqAccepted and could publish then; with a Python relay the - // wire format may not emit that event (§11 relay-reservation-wire-comparison.md), - // so we also publish after a short delay so py×rust tests pass with either relay. - tokio::time::sleep(Duration::from_secs(2)).await; - let _: () = con - .set(&listener_peer_id_key, peer_id.to_string()) - .await - .expect(&format!( - "Failed to publish peer id to Redis: (key: {listener_peer_id_key})" - )); - eprintln!("Published peer id to Redis (key: {listener_peer_id_key})"); - let mut published_to_redis = false; let mut hole_punch_connection_id = None; diff --git a/hole-punch/images/rust/v0.56/src/bin/relay.rs b/hole-punch/images/rust/v0.56/src/bin/relay.rs index 81961018..41f5701f 100644 --- a/hole-punch/images/rust/v0.56/src/bin/relay.rs +++ b/hole-punch/images/rust/v0.56/src/bin/relay.rs @@ -120,13 +120,6 @@ async fn main() -> Result<()> { eprintln!("Confirmed external address: {address}"); eprintln!("Listening on: {full_multiaddr}"); - // Register the listen address as an external address so the relay - // behaviour includes it in RESERVE OK responses. Without this, - // clients get NoAddressesInReservation because relay::Behaviour - // only includes confirmed external addresses. - swarm.add_external_address(address.clone()); - eprintln!("Registered external address: {address}"); - // Publish to Redis with TEST_KEY namespacing let relay_addr_key = format!("{test_key}_relay_multiaddr"); let _: () = con