distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Peer connection management — address extraction, connection pooling, and NTCP2 connector.
2
3Provides utilities for connecting to I2P peers via the real NTCP2 transport:
4- extract_ntcp2_address(): parse NTCP2 params from a RouterInfo
5- ConnectionPool: track active connections by peer hash
6- PeerConnector: initiate NTCP2 handshakes to remote peers
7"""
8
9from __future__ import annotations
10
11import asyncio
12import hashlib
13import logging
14from dataclasses import dataclass
15
16from i2p_data.data_helper import from_base64
17
18from i2p_transport.ntcp2_real_server import NTCP2RealConnector
19
20logger = logging.getLogger(__name__)
21
22
23def extract_ntcp2_address(router_info) -> tuple[str, int, bytes, bytes] | None:
24 """Extract NTCP2 connection params from a RouterInfo.
25
26 Returns (host, port, static_key_32bytes, iv_16bytes) or None.
27 Looks for address with transport="NTCP2" and options containing "s" and "i".
28 """
29 for addr in router_info.addresses:
30 if addr.transport != "NTCP2":
31 continue
32 opts = addr.options
33 s_b64 = opts.get("s")
34 i_b64 = opts.get("i")
35 if s_b64 is None or i_b64 is None:
36 continue
37 host = opts.get("host")
38 port_str = opts.get("port")
39 if host is None or port_str is None:
40 continue
41 try:
42 static_key = from_base64(s_b64)
43 iv = from_base64(i_b64)
44 port = int(port_str)
45 except (ValueError, Exception):
46 continue
47 return (host, port, static_key, iv)
48 return None
49
50
51class ConnectionPool:
52 """Tracks active NTCP2 connections by peer hash."""
53
54 def __init__(self, max_connections: int = 50):
55 self._connections: dict[bytes, object] = {}
56 self._max = max_connections
57
58 def add(self, peer_hash: bytes, connection) -> bool:
59 """Add connection. Returns False if at capacity."""
60 if len(self._connections) >= self._max:
61 return False
62 self._connections[peer_hash] = connection
63 return True
64
65 def get(self, peer_hash: bytes):
66 """Get connection by peer hash, or None."""
67 return self._connections.get(peer_hash)
68
69 def remove(self, peer_hash: bytes) -> None:
70 """Remove a connection by peer hash. No-op if not present."""
71 self._connections.pop(peer_hash, None)
72
73 def is_connected(self, peer_hash: bytes) -> bool:
74 """Return True if peer_hash has an active connection."""
75 return peer_hash in self._connections
76
77 def get_all_peer_hashes(self) -> list[bytes]:
78 """Return all connected peer hashes."""
79 return list(self._connections.keys())
80
81 @property
82 def active_count(self) -> int:
83 """Number of active connections."""
84 return len(self._connections)
85
86
87class PeerConnector:
88 """Connects to I2P peers via NTCP2.
89
90 Extracts NTCP2 address from RouterInfo, initiates handshake,
91 and returns established connection.
92 """
93
94 def __init__(
95 self,
96 our_static_keypair: tuple[bytes, bytes],
97 our_iv: bytes,
98 our_ri_bytes: bytes | None = None,
99 ):
100 self._our_static = our_static_keypair
101 self._our_iv = our_iv
102 self._our_ri_bytes = our_ri_bytes or b""
103
104 def set_our_router_info(self, ri_bytes: bytes) -> None:
105 """Set our own serialized RouterInfo to send in msg3."""
106 self._our_ri_bytes = ri_bytes
107
108 async def connect(self, router_info) -> object | None:
109 """Connect to a peer using NTCP2.
110
111 1. Extract NTCP2 address from RouterInfo
112 2. Create NTCP2RealConnector
113 3. Perform handshake (sends OUR RouterInfo in msg3)
114 4. Return connection (or None on failure)
115 """
116 params = extract_ntcp2_address(router_info)
117 if params is None:
118 logger.debug("No NTCP2 address found in RouterInfo")
119 return None
120
121 host, port, peer_static_pub, peer_iv = params
122
123 # Compute the SHA-256 hash of the peer's RouterIdentity for handshake obfuscation.
124 # Java: _bobHash = new SessionKey(peer.calculateHash().getData())
125 # calculateHash() = SHA-256 of the RouterIdentity bytes (NOT the full RouterInfo)
126 identity_bytes = router_info.identity.to_bytes()
127 peer_ri_hash = hashlib.sha256(identity_bytes).digest()
128
129 try:
130 connector = NTCP2RealConnector()
131 conn = await connector.connect(
132 host=host,
133 port=port,
134 our_static_key=self._our_static,
135 our_ri_bytes=self._our_ri_bytes,
136 peer_static_pub=peer_static_pub,
137 peer_ri_hash=peer_ri_hash,
138 peer_iv=peer_iv,
139 )
140 logger.info("Connected to peer at %s:%d", host, port)
141 return conn
142 except Exception:
143 logger.exception("Failed to connect to peer at %s:%d", host, port)
144 return None