distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tunnel build pipeline wiring.
2
3Ported from:
4 net.i2p.router.tunnel.pool.BuildHandler (conceptual)
5
6Connects peer selection, hop configuration generation, build execution,
7crypto registration, and tunnel pool management into a single pipeline.
8"""
9
10from __future__ import annotations
11
12import os
13import time
14
15from i2p_data.tunnel import TunnelId, HopConfig
16from i2p_tunnel.build_executor import TunnelBuildExecutor, TunnelManager
17from i2p_tunnel.builder import TunnelEntry
18from i2p_tunnel.data_handler import TunnelCryptoRegistry
19from i2p_peer.profile import PeerSelector
20
21
22class TunnelBuildPipeline:
23 """Wire together tunnel building components into a cohesive pipeline.
24
25 Orchestrates the full tunnel build lifecycle:
26 1. Generate HopConfigs with random crypto keys for selected peers
27 2. Delegate to TunnelBuildExecutor to create build messages
28 3. Process replies — on acceptance, register keys and pool the tunnel
29 4. Report pool maintenance status
30 """
31
32 def __init__(
33 self,
34 build_executor: TunnelBuildExecutor,
35 tunnel_manager: TunnelManager,
36 crypto_registry: TunnelCryptoRegistry,
37 peer_selector: PeerSelector | None = None,
38 ) -> None:
39 self._executor = build_executor
40 self._manager = tunnel_manager
41 self._registry = crypto_registry
42 self._peer_selector = peer_selector
43
44 def build_tunnel(self, peer_profiles: list, is_inbound: bool = True) -> dict:
45 """Generate HopConfigs and build a tunnel request message.
46
47 Parameters
48 ----------
49 peer_profiles:
50 List of PeerProfile objects representing the peers to use
51 as hops, in order from gateway to endpoint.
52 is_inbound:
53 True for inbound tunnels, False for outbound.
54
55 Returns
56 -------
57 dict
58 ``{"hop_configs": [...], "build_msg": <TunnelBuildMessage or result>}``
59 """
60 hop_configs = []
61 public_keys = []
62
63 for i, profile in enumerate(peer_profiles):
64 # Generate random tunnel IDs and crypto keys for each hop.
65 receive_tid = TunnelId(int.from_bytes(os.urandom(4), "big") % (2**32 - 1) + 1)
66 send_tid = TunnelId(int.from_bytes(os.urandom(4), "big") % (2**32 - 1) + 1)
67
68 layer_key = os.urandom(32)
69 iv_key = os.urandom(32)
70 reply_key = os.urandom(32)
71 reply_iv = os.urandom(16)
72 receive_key = os.urandom(32)
73 send_key = os.urandom(32)
74
75 hop = HopConfig(
76 receive_tunnel_id=receive_tid,
77 send_tunnel_id=send_tid,
78 receive_key=receive_key,
79 send_key=send_key,
80 iv_key=iv_key,
81 reply_key=reply_key,
82 reply_iv=reply_iv,
83 layer_key=layer_key,
84 )
85 hop_configs.append(hop)
86
87 # In a full implementation, the public key would come from
88 # the peer's RouterInfo. For now, generate a placeholder
89 # 256-byte key so the executor has something to work with.
90 public_keys.append(os.urandom(256))
91
92 build_msg = self._executor.build_tunnel(hop_configs, public_keys, is_inbound)
93
94 return {"hop_configs": hop_configs, "build_msg": build_msg}
95
96 def process_reply(
97 self,
98 reply_result,
99 hop_configs: list,
100 tunnel_id: int,
101 is_inbound: bool,
102 expiration_ms: int,
103 ) -> TunnelEntry | None:
104 """Process a build reply and register the tunnel if accepted.
105
106 Parameters
107 ----------
108 reply_result:
109 A dict with an ``"accepted"`` boolean key indicating whether
110 all hops accepted the build request.
111 hop_configs:
112 The HopConfig list from build_tunnel.
113 tunnel_id:
114 The numeric tunnel identifier for this tunnel.
115 is_inbound:
116 True for inbound tunnels, False for outbound.
117 expiration_ms:
118 Absolute expiration time in milliseconds since epoch.
119
120 Returns
121 -------
122 TunnelEntry | None
123 A TunnelEntry if all hops accepted; None if any rejected.
124 """
125 if not reply_result.get("accepted", False):
126 return None
127
128 now_ms = int(time.time() * 1000)
129
130 entry = TunnelEntry(
131 tunnel_id=TunnelId(tunnel_id),
132 gateway=os.urandom(32),
133 length=len(hop_configs),
134 creation_time=now_ms,
135 expiration=expiration_ms,
136 )
137
138 # Register each hop's crypto keys in the registry.
139 num_hops = len(hop_configs)
140 for i, hop in enumerate(hop_configs):
141 is_endpoint = (i == num_hops - 1)
142 self._registry.register(
143 tunnel_id=int(hop.receive_tunnel_id),
144 layer_key=hop.layer_key,
145 iv_key=hop.iv_key,
146 is_endpoint=is_endpoint,
147 )
148
149 # Add the tunnel to the appropriate pool.
150 self._manager.add_tunnel(entry, is_inbound=is_inbound)
151
152 return entry
153
154 def maintain_pools(self) -> dict:
155 """Check tunnel pool status against targets.
156
157 Returns
158 -------
159 dict
160 ``{"needs_inbound": bool, "needs_outbound": bool,
161 "inbound_count": int, "outbound_count": int}``
162 """
163 return {
164 "needs_inbound": self._manager.needs_more_inbound(),
165 "needs_outbound": self._manager.needs_more_outbound(),
166 "inbound_count": self._manager.inbound_count(),
167 "outbound_count": self._manager.outbound_count(),
168 }