Skip to content

Commit 5d9cca2

Browse files
lla-danella-dane
committed
updated as per #664 
1 parent 59a7ee0 commit 5d9cca2

File tree

2 files changed

+240
-76
lines changed

2 files changed

+240
-76
lines changed

‎interop/arch.py‎

Lines changed: 61 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
11
from dataclasses import (
22
dataclass,
33
)
4+
import logging
45

6+
from cryptography.hazmat.primitives.asymmetric import (
7+
x25519,
8+
)
59
import multiaddr
610
import redis
711
import trio
@@ -15,7 +19,6 @@
1519
from libp2p.crypto.rsa import (
1620
create_new_key_pair,
1721
)
18-
from libp2p.crypto.x25519 import create_new_key_pair as create_new_x25519_key_pair
1922
from libp2p.custom_types import (
2023
TProtocol,
2124
)
@@ -24,7 +27,6 @@
2427
InsecureTransport,
2528
)
2629
from libp2p.security.noise.transport import (
27-
PROTOCOL_ID as NOISE_PROTOCOL_ID,
2830
Transport as NoiseTransport,
2931
)
3032
import libp2p.security.secio.transport as secio
@@ -37,11 +39,52 @@
3739
Yamux,
3840
)
3941

42+
# Configure detailed logging
43+
logging.basicConfig(
44+
level=logging.DEBUG,
45+
format="%(asctime)s - %(levelname)s - %(message)s",
46+
handlers=[
47+
logging.StreamHandler(),
48+
logging.FileHandler("ping_debug.log", mode="w", encoding="utf-8"),
49+
],
50+
)
51+
4052

4153
def generate_new_rsa_identity() -> KeyPair:
4254
return create_new_key_pair()
4355

4456

57+
def create_noise_keypair():
58+
"""Create a Noise protocol keypair for secure communication"""
59+
try:
60+
x25519_private_key = x25519.X25519PrivateKey.generate()
61+
62+
class NoisePrivateKey:
63+
def __init__(self, key):
64+
self._key = key
65+
66+
def to_bytes(self):
67+
return self._key.private_bytes_raw()
68+
69+
def public_key(self):
70+
return NoisePublicKey(self._key.public_key())
71+
72+
def get_public_key(self):
73+
return NoisePublicKey(self._key.public_key())
74+
75+
class NoisePublicKey:
76+
def __init__(self, key):
77+
self._key = key
78+
79+
def to_bytes(self):
80+
return self._key.public_bytes_raw()
81+
82+
return NoisePrivateKey(x25519_private_key)
83+
except Exception as e:
84+
logging.error(f"Failed to create Noise keypair: {e}")
85+
return None
86+
87+
4588
async def build_host(transport: str, ip: str, port: str, sec_protocol: str, muxer: str):
4689
match (sec_protocol, muxer):
4790
case ("insecure", "mplex"):
@@ -69,18 +112,23 @@ async def build_host(transport: str, ip: str, port: str, sec_protocol: str, muxe
69112
muladdr = multiaddr.Multiaddr(f"/ip4/{ip}/tcp/{port}")
70113
return (host, muladdr)
71114
case ("noise", "yamux"):
72-
key_pair = create_new_key_pair()
73-
noise_key_pair = create_new_x25519_key_pair()
115+
key_pair = generate_new_rsa_identity()
116+
logging.debug("Generated RSA keypair")
74117

75-
host = new_host(
76-
key_pair,
77-
{TProtocol(YAMUX_PROTOCOL_ID): Yamux},
78-
{
79-
NOISE_PROTOCOL_ID: NoiseTransport(
80-
key_pair, noise_privkey=noise_key_pair.private_key
81-
)
82-
},
83-
)
118+
noise_privkey = create_noise_keypair()
119+
if not noise_privkey:
120+
print("[ERROR] Failed to create Noise keypair")
121+
return 1
122+
logging.debug("Generated Noise keypair")
123+
124+
noise_transport = NoiseTransport(key_pair, noise_privkey)
125+
logging.debug(f"Noise transport initialized: {noise_transport}")
126+
sec_opt = {TProtocol("/noise"): noise_transport}
127+
muxer_opt = {TProtocol(YAMUX_PROTOCOL_ID): Yamux}
128+
129+
logging.info(f"Using muxer: {muxer_opt}")
130+
131+
host = new_host(key_pair=key_pair, sec_opt=sec_opt, muxer_opt=muxer_opt)
84132
muladdr = multiaddr.Multiaddr(f"/ip4/{ip}/tcp/{port}")
85133
return (host, muladdr)
86134
case _:

‎interop/lib.py‎

Lines changed: 179 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,3 @@
1-
from dataclasses import (
2-
dataclass,
3-
)
4-
import json
51
import logging
62

73
import multiaddr
@@ -38,92 +34,212 @@
3834

3935

4036
async def handle_ping(stream: INetStream) -> None:
41-
while True:
42-
try:
43-
payload = await stream.read(PING_LENGTH)
44-
peer_id = stream.muxed_conn.peer_id
45-
if payload is not None:
46-
print(f"received ping from {peer_id}")
37+
"""Handle incoming ping requests from rust-libp2p clients"""
38+
peer_id = stream.muxed_conn.peer_id
39+
print(f"[INFO] New ping stream opened by {peer_id}")
40+
logging.info(f"Ping handler called for peer {peer_id}")
4741

48-
await stream.write(payload)
49-
print(f"responded with pong to {peer_id}")
42+
ping_count = 0
5043

51-
except Exception:
52-
await stream.reset()
53-
break
44+
try:
45+
while True:
46+
try:
47+
print(f"[INFO] Wailting for ping data from {peer_id}...")
48+
logging.debug(f"Stream state: {stream}")
49+
data = await stream.read(PING_LENGTH)
50+
51+
if not data:
52+
print(
53+
f"[INFO] No data received,conneciton likely closed by {peer_id}"
54+
)
55+
logging.debug("No data received, stream closed")
56+
break
57+
58+
if len(data) == 0:
59+
print(f"[INFO] Empty data received, connection closed by {peer_id}")
60+
logging.debug("Empty data received")
61+
break
62+
63+
ping_count += 1
64+
print(
65+
f"[PING {ping_count}] Received ping from {peer_id}:"
66+
f" {len(data)} bytes"
67+
)
68+
logging.debug(f"Ping data: {data.hex()}")
69+
70+
# Echo the data back (this is what ping protocol does)
71+
await stream.write(data)
72+
print(f"[PING {ping_count}] Echoed ping back to {peer_id}")
73+
74+
except Exception as e:
75+
print(f"[ERROR] Error in ping loop with {peer_id}: {e}")
76+
logging.exception("Ping loop error")
77+
break
5478

79+
except Exception as e:
80+
print(f"[ERROR] Error handling ping from {peer_id}: {e}")
81+
logging.exception("Ping handler error")
82+
finally:
83+
try:
84+
print(f"[INFO] Closing ping stream with {peer_id}")
85+
await stream.close()
86+
except Exception as e:
87+
logging.debug(f"Error closing stream: {e}")
5588

56-
async def send_ping(stream: INetStream) -> None:
57-
try:
58-
payload = b"\x01" * PING_LENGTH
59-
print(f"sending ping to {stream.muxed_conn.peer_id}")
89+
print(f"[INFO] Ping session completed with {peer_id} ({ping_count} pings)")
6090

61-
await stream.write(payload)
6291

63-
with trio.fail_after(RESP_TIMEOUT):
64-
response = await stream.read(PING_LENGTH)
92+
async def send_ping(stream: INetStream, count: int = 1) -> None:
93+
"""Send a sequence of pings compatible with rust-libp2p."""
94+
peer_id = stream.muxed_conn.peer_id
95+
print(f"[INFO] Starting ping sequence to {peer_id} ({count} pings)")
6596

66-
if response == payload:
67-
print(f"received pong from {stream.muxed_conn.peer_id}")
97+
import os
98+
import time
6899

69-
except Exception as e:
70-
print(f"error occurred: {e}")
100+
rtts = []
101+
102+
for i in range(1, count + 1):
103+
try:
104+
# Generate random 32-byte payload as per ping protcol spec
105+
payload = os.urandom(PING_LENGTH)
106+
print(f"[PING {i}/{count}] Sending ping to {peer_id}")
107+
logging.debug(f"Sending payload: {payload.hex()}")
108+
start_time = time.time()
109+
110+
await stream.write(payload)
111+
112+
with trio.fail_after(RESP_TIMEOUT):
113+
response = await stream.read(PING_LENGTH)
114+
115+
end_time = time.time()
116+
rtt = (end_time - start_time) * 1000
117+
118+
if (
119+
response
120+
and len(response) >= PING_LENGTH
121+
and response[:PING_LENGTH] == payload
122+
):
123+
rtts.append(rtt)
124+
print(f"[PING {i}] Successful RTT: {rtt:.2f}ms")
125+
else:
126+
print(f"[ERROR] Ping {i} failed: response mismatch or incomplete")
127+
if response:
128+
logging.debug(f"Expecte: {payload.hex()}")
129+
logging.debug(f"Received: {response.hex()}")
130+
131+
if i < count:
132+
await trio.sleep(1)
133+
134+
except trio.TooSlowError:
135+
print(f"[ERROR] Ping {i} timed out after {RESP_TIMEOUT}s")
136+
except Exception as e:
137+
print(f"[ERROR] Ping {i} failed: {e}")
138+
logging.exception(f"Ping {i} error")
139+
140+
# Print statistics
141+
if rtts:
142+
avg_rtt = sum(rtts) / len(rtts)
143+
min_rtt = min(rtts)
144+
max_rtt = max(rtts) # Fixed typo: was max_rtts
145+
success_count = len(rtts)
146+
loss_rate = ((count - success_count) / count) * 100
147+
148+
print("\n[STATS] Ping Statistics:")
149+
print(
150+
f" Packets: Sent={count}, Received={success_count},"
151+
f" Lost={count - success_count}"
152+
)
153+
print(f" Loss rate: {loss_rate:.1f}%")
154+
print(f" RTT: min={min_rtt:.2f}ms, avg={avg_rtt:.2f}ms, max={max_rtt:.2f}ms")
155+
else:
156+
print(f"\n[STATS] All pings failed ({count} attempts)")
71157

72158

73159
async def run_test(
74160
transport, ip, port, is_dialer, test_timeout, redis_addr, sec_protocol, muxer
75161
):
76-
import time
77-
78-
logging.info("Starting run_test")
79-
80162
redis_client = RedisClient(
81163
redis.Redis(host="localhost", port=int(redis_addr), db=0)
82164
)
83165
(host, listen_addr) = await build_host(transport, ip, port, sec_protocol, muxer)
84-
logging.info(f"Running ping test local_peer={host.get_id()}")
85-
86-
async with host.run(listen_addrs=[listen_addr]), trio.open_nursery() as nursery:
166+
async with host.run(listen_addrs=[listen_addr]):
87167
if not is_dialer:
168+
print("[INFO] Starting py-libp2p ping server...")
169+
170+
print(f"[INFO] Registering ping handler for protocol: {PING_PROTOCOL_ID}")
88171
host.set_stream_handler(PING_PROTOCOL_ID, handle_ping)
172+
173+
# Also register alternative protocol IDs for better compatibilty
174+
alt_protcols = [
175+
TProtocol("/ping/1.0.0"),
176+
TProtocol("/libp2p/ping/1.0.0"),
177+
]
178+
179+
for alt_proto in alt_protcols:
180+
print(f"[INFO] Also registering handler for: {alt_proto}")
181+
host.set_stream_handler(alt_proto, handle_ping)
182+
183+
print("[INFO] Server started successfully!")
184+
print(f"[INFO] Peer ID: {host.get_id()}")
185+
print(f"[INFO] Listening: /ip4/{ip}/tcp/{port}")
186+
print(f"[INFO] Primary Protocol: {PING_PROTOCOL_ID}")
187+
89188
ma = f"{listen_addr}/p2p/{host.get_id().pretty()}"
90189
redis_client.rpush("listenerAddr", ma)
91190

92-
logging.info(f"Test instance, listening: {ma}")
191+
print("[INFO] Pushed address to Redis database")
192+
await trio.sleep_forever()
93193
else:
194+
print("[INFO] Starting py-libp2p ping client...")
195+
196+
print("[INFO] Fetching remote address from Redis database...")
94197
redis_addr = redis_client.brpop("listenerAddr", timeout=5)
95198
destination = redis_addr[0].decode()
96199
maddr = multiaddr.Multiaddr(destination)
97200
info = info_from_p2p_addr(maddr)
201+
target_peer_id = info.peer_id
98202

99-
handshake_start = time.perf_counter()
203+
print(f"[INFO] Our Peer ID: {host.get_id()}")
204+
print(f"[INFO] Target: {destination}")
205+
print(f"[INFO] Target Peer ID: {target_peer_id}")
206+
print("[INFO] Connecting to peer...")
100207

101-
logging.info("GETTING READY FOR CONNECTION")
102208
await host.connect(info)
103-
logging.info("HOST CONNECTED")
104-
105-
# TILL HERE EVERYTHING IS FINE
106-
107-
stream = await host.new_stream(info.peer_id, [PING_PROTOCOL_ID])
108-
logging.info("CREATED NEW STREAM")
109-
110-
# DOES NOT MORE FORWARD FROM THIS
111-
logging.info("Remote conection established")
112-
113-
nursery.start_soon(send_ping, stream)
114-
115-
handshake_plus_ping = (time.perf_counter() - handshake_start) * 1000.0
116-
117-
logging.info(f"handshake time: {handshake_plus_ping:.2f}ms")
118-
return
119-
120-
await trio.sleep_forever()
121-
122-
123-
@dataclass
124-
class Report:
125-
handshake_plus_one_rtt_millis: float
126-
ping_rtt_millis: float
127-
128-
def gen_report(self):
129-
return json.dumps(self.__dict__)
209+
print("[INFO] Connection established!")
210+
211+
# Try protocols in order of preference
212+
# Start with the standard libp2p ping protocol
213+
protocols_to_try = [
214+
PING_PROTOCOL_ID, # /ipfs/ping/1.0.0 - standard protocol
215+
TProtocol("/ping/1.0.0"), # Alternative
216+
TProtocol("/libp2p/ping/1.0.0"), # Another alternative
217+
]
218+
219+
stream = None
220+
221+
for proto in protocols_to_try:
222+
try:
223+
print(f"[INFO] Trying to open stream with protocol: {proto}")
224+
stream = await host.new_stream(target_peer_id, [proto])
225+
print(f"[INFO] Stream opened with protocol: {proto}")
226+
break
227+
except Exception as e:
228+
print(f"[ERROR] Failed to open stream with {proto}: {e}")
229+
logging.debug(f"Protocol {proto} failed: {e}")
230+
continue
231+
232+
if not stream:
233+
print("[ERROR] Failed to open stream with any ping protocol")
234+
print("[ERROR] Ensure the target peer supports one of these protocols")
235+
for proto in protocols_to_try:
236+
print(f"[ERROR] - {proto}")
237+
return 1
238+
239+
await send_ping(stream)
240+
241+
await stream.close()
242+
print("[INFO] Stream closed successfully")
243+
244+
print("\n[INFO] Client stopped")
245+
return 0

0 commit comments

Comments
 (0)