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 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 4fd494c6..5e220b7a 100644 --- a/hole-punch/images.yaml +++ b/hole-punch/images.yaml @@ -6,6 +6,8 @@ test-aliases: - alias: "failing" value: "" + - alias: "python" + value: "py-v0.6" # Router configurations - Multiple router types for testing different NAT implementations routers: @@ -31,6 +33,15 @@ relays: secureChannels: [noise, tls] muxers: [yamux, mplex] + - id: py-v0.6 + source: + type: local + path: images/python/v0.6 + dockerfile: Dockerfile.relay + transports: [tcp, ws] + secureChannels: [noise, tls] + muxers: [yamux] + # Implementation definitions implementations: - id: rust-v0.56 @@ -41,3 +52,12 @@ implementations: transports: [quic-v1, tcp] secureChannels: [noise, tls] muxers: [yamux, mplex] + + - id: py-v0.6 + source: + type: local + path: images/python/v0.6 + dockerfile: Dockerfile.peer + transports: [tcp, ws] + secureChannels: [noise, tls] + muxers: [yamux] diff --git a/hole-punch/images/python/v0.6/Dockerfile.peer b/hole-punch/images/python/v0.6/Dockerfile.peer new file mode 100644 index 00000000..68d2ae84 --- /dev/null +++ b/hole-punch/images/python/v0.6/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.6/Dockerfile.relay b/hole-punch/images/python/v0.6/Dockerfile.relay new file mode 100644 index 00000000..db750cb4 --- /dev/null +++ b/hole-punch/images/python/v0.6/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.6/peer.py b/hole-punch/images/python/v0.6/peer.py new file mode 100644 index 00000000..5653ea43 --- /dev/null +++ b/hole-punch/images/python/v0.6/peer.py @@ -0,0 +1,818 @@ +#!/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, 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 ( + 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.anyio_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], + ) + + 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, 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: + 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 + 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()) + 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) + 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_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, + 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.6/relay.py b/hole-punch/images/python/v0.6/relay.py new file mode 100644 index 00000000..2a7266a9 --- /dev/null +++ b/hole-punch/images/python/v0.6/relay.py @@ -0,0 +1,689 @@ +#!/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.anyio_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=1000, + max_reservations=500, + ) + + 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.6/requirements.txt b/hole-punch/images/python/v0.6/requirements.txt new file mode 100644 index 00000000..d5f607bf --- /dev/null +++ b/hole-punch/images/python/v0.6/requirements.txt @@ -0,0 +1,14 @@ +# py-libp2p with DCUtR support +git+https://github.com/sumanjeet0012/py-libp2p.git@64a9e5bb0d4c5262571570cadd3bd33b2f504655 + +# 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.6/run-peer.sh b/hole-punch/images/python/v0.6/run-peer.sh new file mode 100644 index 00000000..c04fcbed --- /dev/null +++ b/hole-punch/images/python/v0.6/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.6/run-relay.sh b/hole-punch/images/python/v0.6/run-relay.sh new file mode 100644 index 00000000..5fd8588b --- /dev/null +++ b/hole-punch/images/python/v0.6/run-relay.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -e + +exec python /app/relay.py "$@" diff --git a/hole-punch/lib/generate-tests.sh b/hole-punch/lib/generate-tests.sh index 6ca4ed15..0b444c24 100755 --- a/hole-punch/lib/generate-tests.sh +++ b/hole-punch/lib/generate-tests.sh @@ -559,8 +559,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]}" @@ -682,7 +682,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 @@ -693,7 +693,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 @@ -704,7 +704,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 @@ -739,7 +739,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 @@ -750,7 +750,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 @@ -761,7 +761,8 @@ 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_ignored}" fi cat >> "${worker_ignored}" <> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_ignored}" fi fi @@ -864,7 +867,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 @@ -875,7 +878,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 @@ -886,7 +889,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 @@ -919,7 +922,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 @@ -930,7 +933,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 @@ -941,7 +944,8 @@ 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_ignored}" fi cat >> "${worker_ignored}" <> "${worker_ignored}" fi cat >> "${worker_ignored}" <> "${worker_ignored}" fi fi diff --git a/hole-punch/lib/run-single-test.sh b/hole-punch/lib/run-single-test.sh index cc81b8e2..ff875529 100755 --- a/hole-punch/lib/run-single-test.sh +++ b/hole-punch/lib/run-single-test.sh @@ -225,6 +225,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} @@ -252,6 +253,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} 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"