distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Full NTCP2 connection test: handshake + data exchange."""
2import asyncio
3import base64
4import hashlib
5import logging
6import random
7import sys
8import time
9
10sys.path.insert(0, "src")
11
12from i2p_data.router import RouterInfo
13from i2p_netdb.reseed import ReseedClient
14from i2p_router.identity import (
15 RouterKeyBundle,
16 create_full_router_identity,
17)
18from i2p_router.peer_connector import extract_ntcp2_address
19from i2p_transport.ntcp2_real_server import NTCP2RealConnector
20from i2p_transport.ntcp2_blocks import (
21 BLOCK_DATETIME, BLOCK_I2NP, BLOCK_ROUTERINFO, BLOCK_TERMINATION,
22 BLOCK_PADDING, BLOCK_OPTIONS,
23)
24
25logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
26logger = logging.getLogger("full")
27
28BLOCK_NAMES = {
29 BLOCK_DATETIME: "DateTime",
30 BLOCK_I2NP: "I2NP",
31 BLOCK_ROUTERINFO: "RouterInfo",
32 BLOCK_TERMINATION: "Termination",
33 BLOCK_PADDING: "Padding",
34 BLOCK_OPTIONS: "Options",
35}
36
37
38async def main():
39 bundle = RouterKeyBundle.generate()
40 identity, ri = create_full_router_identity(bundle, "0.0.0.0", 9000)
41 our_ri_bytes = ri.to_bytes()
42
43 logger.info("Our RI: %d bytes, verify=%s", len(our_ri_bytes), ri.verify())
44
45 # Reseed
46 client = ReseedClient(target_count=20, min_servers=1, timeout=15)
47 ri_list = await client.reseed()
48 logger.info("Got %d peer RIs from reseed", len(ri_list))
49
50 connector_keypair = (bundle.ntcp2_private, bundle.ntcp2_public)
51 random.shuffle(ri_list)
52
53 connected = 0
54 attempts = 0
55 for peer_bytes in ri_list:
56 if connected >= 3 or attempts >= 10:
57 break
58 try:
59 peer_ri = RouterInfo.from_bytes(peer_bytes)
60 params = extract_ntcp2_address(peer_ri)
61 if params is None:
62 continue
63
64 host, port, peer_static_pub, peer_iv = params
65 if len(peer_static_pub) != 32 or len(peer_iv) != 16:
66 continue
67
68 peer_identity_bytes = peer_ri.identity.to_bytes()
69 peer_hash = hashlib.sha256(peer_identity_bytes).digest()
70
71 attempts += 1
72 logger.info("Connecting to %s:%d (v%s, caps=%s)...",
73 host, port,
74 peer_ri.options.get("router.version", "?"),
75 peer_ri.options.get("caps", "?"))
76
77 connector = NTCP2RealConnector()
78 try:
79 conn = await asyncio.wait_for(
80 connector.connect(
81 host=host,
82 port=port,
83 our_static_key=connector_keypair,
84 our_ri_bytes=our_ri_bytes,
85 peer_static_pub=peer_static_pub,
86 peer_ri_hash=peer_hash,
87 peer_iv=peer_iv,
88 ),
89 timeout=10,
90 )
91 connected += 1
92 logger.info("CONNECTED to %s:%d!", host, port)
93
94 # Read frames
95 frames_read = 0
96 try:
97 while frames_read < 5:
98 blocks = await asyncio.wait_for(conn.recv_frame(), timeout=10)
99 frames_read += 1
100 for block in blocks:
101 name = BLOCK_NAMES.get(block.block_type, f"Unknown({block.block_type})")
102 logger.info(" Frame %d: %s (%d bytes)",
103 frames_read, name, len(block.data))
104 if block.block_type == BLOCK_TERMINATION:
105 reason = block.data[-1] if block.data else -1
106 logger.info(" -> Termination reason=%d", reason)
107 elif block.block_type == BLOCK_I2NP and len(block.data) >= 11:
108 msg_type = block.data[0]
109 logger.info(" -> I2NP msg_type=%d", msg_type)
110 elif block.block_type == BLOCK_ROUTERINFO and len(block.data) > 1:
111 flag = block.data[0]
112 ri_data = block.data[1:]
113 try:
114 peer_ri2 = RouterInfo.from_bytes(ri_data)
115 logger.info(" -> RI: %d addrs, v%s, verify=%s",
116 len(peer_ri2.addresses),
117 peer_ri2.options.get("router.version", "?"),
118 peer_ri2.verify())
119 except Exception as e:
120 logger.info(" -> RI parse error: %s", e)
121 elif block.block_type == BLOCK_DATETIME:
122 import struct
123 ts = struct.unpack("!I", block.data[:4])[0]
124 logger.info(" -> DateTime: %d", ts)
125 except asyncio.TimeoutError:
126 logger.info(" No more frames within timeout (read %d frames)", frames_read)
127 except Exception as e:
128 logger.info(" Error reading frame: %s", e)
129
130 try:
131 await conn.close()
132 except Exception:
133 pass
134
135 except asyncio.TimeoutError:
136 logger.info("Connection timed out")
137 except Exception as e:
138 logger.info("Connection failed: %s", e)
139
140 except Exception as e:
141 continue
142
143 logger.info("=== DONE: %d/%d connections successful ===", connected, attempts)
144
145
146if __name__ == "__main__":
147 asyncio.run(main())