distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Enhanced diagnostic: connect to peers with detailed byte-level debugging."""
2import asyncio
3import base64
4import hashlib
5import logging
6import random
7import struct
8import sys
9import time
10
11sys.path.insert(0, "src")
12
13from i2p_data.router import RouterInfo
14from i2p_netdb.reseed import ReseedClient
15from i2p_router.identity import (
16 RouterKeyBundle,
17 create_full_router_identity,
18)
19from i2p_router.peer_connector import extract_ntcp2_address
20from i2p_transport.ntcp2_real_handshake import NTCP2RealHandshake
21from i2p_transport.ntcp2_blocks import BLOCK_TERMINATION, BLOCK_ROUTERINFO, decode_blocks
22
23logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(levelname)s %(name)s %(message)s")
24logger = logging.getLogger("diag2")
25
26
27async def raw_connect(host, port, our_static_key, our_ri_bytes, peer_static_pub, peer_ri_hash, peer_iv):
28 """Manual NTCP2 handshake with raw byte debugging at every step."""
29 reader, writer = await asyncio.open_connection(host, port)
30
31 try:
32 hs = NTCP2RealHandshake(
33 our_static=our_static_key,
34 peer_static_pub=peer_static_pub,
35 peer_ri_hash=peer_ri_hash,
36 peer_iv=peer_iv,
37 initiator=True,
38 )
39
40 # --- MSG1 ---
41 msg1 = hs.create_session_request(padding_len=0, router_info=our_ri_bytes)
42 logger.info("MSG1: %d bytes, msg3p2len=%d", len(msg1), hs._msg3p2len)
43 writer.write(msg1)
44 await writer.drain()
45
46 # --- MSG2 header (64 bytes) ---
47 msg2_header = await asyncio.wait_for(reader.readexactly(64), timeout=10)
48 logger.info("MSG2 header: %d bytes", len(msg2_header))
49 padlen2 = hs.process_session_created_header(msg2_header)
50 logger.info("MSG2 padlen2=%d", padlen2)
51
52 if padlen2 > 0:
53 msg2_padding = await reader.readexactly(padlen2)
54 else:
55 msg2_padding = b""
56 hs.process_session_created_padding(msg2_padding)
57 logger.info("MSG2 processed OK")
58
59 # --- MSG3 ---
60 msg3 = hs.create_session_confirmed(router_info=our_ri_bytes)
61 logger.info("MSG3: %d bytes (part1=48, part2=%d)", len(msg3), len(msg3) - 48)
62 logger.info("MSG3 expected part2=%d, actual part2=%d", hs._msg3p2len, len(msg3) - 48)
63
64 writer.write(msg3)
65 await writer.drain()
66 logger.info("MSG3 sent, draining...")
67
68 # --- POST-HANDSHAKE: try raw read ---
69 # Give peer a moment to process msg3
70 await asyncio.sleep(0.5)
71
72 # Try to read raw bytes to see what (if anything) comes back
73 try:
74 raw_data = await asyncio.wait_for(reader.read(4096), timeout=5)
75 if raw_data:
76 logger.info("RAW POST-HANDSHAKE: got %d bytes: %s", len(raw_data), raw_data[:64].hex())
77
78 # Try SipHash deobfuscation
79 keys = hs.split()
80 if len(raw_data) >= 2:
81 obf_len = raw_data[:2]
82 frame_len = keys.recv_siphash.deobfuscate_length(obf_len)
83 logger.info("SipHash deobfuscated length: %d", frame_len)
84 if frame_len <= len(raw_data) - 2:
85 encrypted = raw_data[2:2 + frame_len]
86 try:
87 plaintext = keys.recv_cipher.decrypt_with_ad(b"", encrypted)
88 blocks = decode_blocks(plaintext)
89 for block in blocks:
90 logger.info("Block type=%d len=%d", block.block_type, len(block.data))
91 if block.block_type == BLOCK_TERMINATION:
92 reason = block.data[-1] if block.data else -1
93 logger.info("TERMINATION reason=%d", reason)
94 except Exception as e:
95 logger.info("Decrypt failed: %s", e)
96 else:
97 logger.info("Need %d more bytes for frame", frame_len - (len(raw_data) - 2))
98 else:
99 logger.info("RAW POST-HANDSHAKE: 0 bytes (peer closed connection)")
100 except asyncio.TimeoutError:
101 logger.info("RAW POST-HANDSHAKE: timeout (peer sent nothing in 5s)")
102
103 writer.close()
104 await writer.wait_closed()
105
106 except Exception as e:
107 logger.error("Connection failed: %s", e, exc_info=True)
108 writer.close()
109
110
111async def main():
112 # Generate keys
113 bundle = RouterKeyBundle.generate()
114 identity, ri = create_full_router_identity(bundle, "0.0.0.0", 9000)
115 our_ri_bytes = ri.to_bytes()
116
117 logger.info("=== OUR ROUTERINFO ===")
118 logger.info("RI total bytes: %d", len(our_ri_bytes))
119 logger.info("Self-verify: %s", ri.verify())
120 logger.info("Options: %s", ri.options)
121
122 # Roundtrip check
123 ri2 = RouterInfo.from_bytes(our_ri_bytes)
124 signable1 = ri._signable_bytes()
125 signable2 = ri2._signable_bytes()
126 logger.info("Roundtrip signable match: %s", signable1 == signable2)
127 logger.info("Roundtrip verify: %s", ri2.verify())
128 logger.info("Roundtrip to_bytes match: %s", our_ri_bytes == ri2.to_bytes())
129
130 # Show the RI block as it would appear in msg3
131 from i2p_transport.ntcp2_blocks import router_info_block, options_block, padding_block, encode_blocks
132 blocks = [
133 router_info_block(our_ri_bytes),
134 options_block(),
135 padding_block(0),
136 ]
137 block_payload = encode_blocks(blocks)
138 logger.info("MSG3 block payload size: %d", len(block_payload))
139 logger.info("MSG3 expected p2 len (payload+16): %d", len(block_payload) + 16)
140
141 # Parse the block payload back to verify it's valid
142 parsed_blocks = decode_blocks(block_payload)
143 logger.info("Parsed %d blocks from msg3 payload", len(parsed_blocks))
144 for b in parsed_blocks:
145 logger.info(" Block type=%d datalen=%d", b.block_type, len(b.data))
146
147 # Verify the RI from the block matches
148 ri_block = parsed_blocks[0]
149 ri_flag = ri_block.data[0]
150 ri_data = ri_block.data[1:]
151 logger.info("RI block flag=%d, ri_data len=%d", ri_flag, len(ri_data))
152 ri3 = RouterInfo.from_bytes(ri_data)
153 logger.info("RI from msg3 block verify: %s", ri3.verify())
154 logger.info("RI from msg3 to_bytes match: %s", ri_data == ri3.to_bytes())
155
156 # Reseed
157 logger.info("=== RESEEDING ===")
158 client = ReseedClient(target_count=10, min_servers=1, timeout=15)
159 ri_list = await client.reseed()
160 logger.info("Got %d RIs from reseed", len(ri_list))
161
162 # Check what enc_types the peers use
163 for peer_bytes in ri_list[:5]:
164 try:
165 peer_ri = RouterInfo.from_bytes(peer_bytes)
166 cert = peer_ri.identity.certificate
167 from i2p_data.certificate import KeyCertificate
168 if isinstance(cert, KeyCertificate):
169 enc_type = cert.get_enc_type()
170 sig_type = cert.get_sig_type()
171 logger.info("Peer cert: enc_type=%s sig_type=%s version=%s",
172 enc_type, sig_type, peer_ri.options.get("router.version", "?"))
173 except Exception as e:
174 logger.debug("Skip peer cert check: %s", e)
175
176 # Try to connect to peers
177 connector_keypair = (bundle.ntcp2_private, bundle.ntcp2_public)
178
179 random.shuffle(ri_list)
180 attempts = 0
181 for peer_bytes in ri_list:
182 if attempts >= 5:
183 break
184 try:
185 peer_ri = RouterInfo.from_bytes(peer_bytes)
186 params = extract_ntcp2_address(peer_ri)
187 if params is None:
188 continue
189
190 host, port, peer_static_pub, peer_iv = params
191
192 # Fix 15-byte IVs by right-padding with 0x00
193 if len(peer_iv) < 16:
194 logger.info("Fixing short IV: %d bytes -> 16 bytes", len(peer_iv))
195 peer_iv = peer_iv + b"\x00" * (16 - len(peer_iv))
196 elif len(peer_iv) > 16:
197 logger.info("Truncating long IV: %d bytes -> 16 bytes", len(peer_iv))
198 peer_iv = peer_iv[:16]
199
200 peer_identity_bytes = peer_ri.identity.to_bytes()
201 peer_hash = hashlib.sha256(peer_identity_bytes).digest()
202
203 attempts += 1
204 logger.info("=== CONNECTING #%d TO %s:%d (hash=%s...) ===",
205 attempts, host, port, peer_hash[:4].hex())
206 logger.info("Peer version=%s caps=%s",
207 peer_ri.options.get("router.version", "?"),
208 peer_ri.options.get("caps", "?"))
209 logger.info("Peer static pub: %s", peer_static_pub.hex())
210 logger.info("Peer IV (%d bytes): %s", len(peer_iv), peer_iv.hex())
211 logger.info("Peer RI valid: %s", peer_ri.verify())
212
213 # Check peer's enc_type
214 from i2p_data.certificate import KeyCertificate
215 cert = peer_ri.identity.certificate
216 if isinstance(cert, KeyCertificate):
217 logger.info("Peer enc_type=%s sig_type=%s",
218 cert.get_enc_type(), cert.get_sig_type())
219
220 await raw_connect(
221 host=host,
222 port=port,
223 our_static_key=connector_keypair,
224 our_ri_bytes=our_ri_bytes,
225 peer_static_pub=peer_static_pub,
226 peer_ri_hash=peer_hash,
227 peer_iv=peer_iv,
228 )
229
230 except Exception as e:
231 logger.debug("Skip peer: %s", e)
232 continue
233
234 logger.info("=== DONE ===")
235
236
237if __name__ == "__main__":
238 asyncio.run(main())