distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for RouterContext — the central coordinator."""
2
3import os
4
5import pytest
6
7from i2p_router.core import RouterContext
8from i2p_crypto.session_key_manager import SessionKeyManager
9from i2p_crypto.garlic_crypto import GarlicEncryptor, GarlicDecryptor
10from i2p_data.message_router import (
11 InboundMessageHandler,
12 OutboundMessageRouter,
13 MessageDispatcher,
14)
15from i2p_data.garlic_handler import GarlicMessageHandler
16from i2p_netdb.datastore import DataStore
17from i2p_netdb.netdb_handler import NetDBHandler
18from i2p_tunnel.data_handler import TunnelCryptoRegistry, TunnelDataHandler
19from i2p_tunnel.build_executor import TunnelManager
20
21
22class TestRouterContextCreation:
23 """RouterContext creates all subsystems on init."""
24
25 def test_creates_with_random_hash(self):
26 ctx = RouterContext()
27 assert isinstance(ctx.router_hash, bytes)
28 assert len(ctx.router_hash) == 32
29
30 def test_creates_with_given_hash(self):
31 h = os.urandom(32)
32 ctx = RouterContext(router_hash=h)
33 assert ctx.router_hash == h
34
35 def test_owns_session_key_mgr(self):
36 ctx = RouterContext()
37 assert isinstance(ctx.session_key_mgr, SessionKeyManager)
38
39 def test_owns_garlic_encryptor(self):
40 ctx = RouterContext()
41 assert isinstance(ctx.garlic_encryptor, GarlicEncryptor)
42
43 def test_owns_garlic_decryptor(self):
44 ctx = RouterContext()
45 assert isinstance(ctx.garlic_decryptor, GarlicDecryptor)
46
47 def test_owns_inbound_handler(self):
48 ctx = RouterContext()
49 assert isinstance(ctx.inbound_handler, InboundMessageHandler)
50
51 def test_owns_outbound_router(self):
52 ctx = RouterContext()
53 assert isinstance(ctx.outbound_router, OutboundMessageRouter)
54
55 def test_owns_message_dispatcher(self):
56 ctx = RouterContext()
57 assert isinstance(ctx.message_dispatcher, MessageDispatcher)
58
59 def test_owns_datastore(self):
60 ctx = RouterContext()
61 assert isinstance(ctx.datastore, DataStore)
62
63 def test_owns_netdb_handler(self):
64 ctx = RouterContext()
65 assert isinstance(ctx.netdb_handler, NetDBHandler)
66
67 def test_owns_crypto_registry(self):
68 ctx = RouterContext()
69 assert isinstance(ctx.crypto_registry, TunnelCryptoRegistry)
70
71 def test_owns_tunnel_data_handler(self):
72 ctx = RouterContext()
73 assert isinstance(ctx.tunnel_data_handler, TunnelDataHandler)
74
75 def test_owns_tunnel_manager(self):
76 ctx = RouterContext()
77 assert isinstance(ctx.tunnel_manager, TunnelManager)
78
79 def test_owns_garlic_handler(self):
80 ctx = RouterContext()
81 assert isinstance(ctx.garlic_handler, GarlicMessageHandler)
82
83
84class TestGetStatus:
85 """get_status returns a valid dict with expected keys."""
86
87 def test_status_keys(self):
88 ctx = RouterContext()
89 status = ctx.get_status()
90 assert "router_hash" in status
91 assert "tunnels_registered" in status
92 assert "netdb_entries" in status
93 assert "inbound_count" in status
94 assert "outbound_count" in status
95
96 def test_status_initial_values(self):
97 ctx = RouterContext()
98 status = ctx.get_status()
99 assert status["tunnels_registered"] == 0
100 assert status["netdb_entries"] == 0
101 assert status["inbound_count"] == 0
102 assert status["outbound_count"] == 0
103
104 def test_status_router_hash_is_hex(self):
105 ctx = RouterContext()
106 status = ctx.get_status()
107 # Should be a valid hex string of the 32-byte hash
108 assert len(status["router_hash"]) == 64
109 bytes.fromhex(status["router_hash"]) # should not raise
110
111
112class TestNetDBRoundtrip:
113 """store_netdb_entry and lookup_netdb_entry roundtrip."""
114
115 def test_store_returns_true(self):
116 ctx = RouterContext()
117 key = os.urandom(32)
118 data = b"router-info-payload"
119 assert ctx.store_netdb_entry(key, data) is True
120
121 def test_lookup_after_store(self):
122 ctx = RouterContext()
123 key = os.urandom(32)
124 data = b"router-info-payload"
125 ctx.store_netdb_entry(key, data)
126 assert ctx.lookup_netdb_entry(key) == data
127
128 def test_lookup_missing_returns_none(self):
129 ctx = RouterContext()
130 key = os.urandom(32)
131 assert ctx.lookup_netdb_entry(key) is None
132
133 def test_status_reflects_netdb_count(self):
134 ctx = RouterContext()
135 ctx.store_netdb_entry(os.urandom(32), b"a")
136 ctx.store_netdb_entry(os.urandom(32), b"b")
137 assert ctx.get_status()["netdb_entries"] == 2
138
139
140class TestTunnelRegistration:
141 """register_tunnel and process_tunnel_data work together."""
142
143 def test_register_and_process(self):
144 ctx = RouterContext()
145 tunnel_id = 42
146 layer_key = os.urandom(32)
147 iv_key = os.urandom(32)
148 # Register as endpoint so we get a "deliver" action
149 ctx.register_tunnel(tunnel_id, layer_key, iv_key, is_endpoint=True)
150
151 # Encrypted data must be multiple of 16 bytes
152 encrypted_data = os.urandom(1024)
153 result = ctx.process_tunnel_data(tunnel_id, encrypted_data)
154 assert result["action"] == "deliver"
155 assert "data" in result
156
157 def test_process_unknown_tunnel(self):
158 ctx = RouterContext()
159 result = ctx.process_tunnel_data(999, os.urandom(1024))
160 assert result["action"] == "unknown"
161 assert result["tunnel_id"] == 999
162
163 def test_status_reflects_registered_tunnels(self):
164 ctx = RouterContext()
165 ctx.register_tunnel(1, os.urandom(32), os.urandom(32))
166 ctx.register_tunnel(2, os.urandom(32), os.urandom(32))
167 assert ctx.get_status()["tunnels_registered"] == 2
168
169 def test_intermediate_hop_forwards(self):
170 ctx = RouterContext()
171 tunnel_id = 7
172 ctx.register_tunnel(tunnel_id, os.urandom(32), os.urandom(32), is_endpoint=False)
173 result = ctx.process_tunnel_data(tunnel_id, os.urandom(1024))
174 assert result["action"] == "forward"
175
176
177class TestInboundDispatch:
178 """process_inbound dispatches to registered handlers."""
179
180 def test_dispatch_delivery_status(self):
181 ctx = RouterContext()
182 # Delivery status: 4-byte message ID
183 msg_id = (12345).to_bytes(4, "big")
184 payload = msg_id + b"\x00" * 8
185 # Type 10 = DELIVERY_STATUS — uses default handler
186 result = ctx.process_inbound(10, payload)
187 assert result == 12345
188
189 def test_dispatch_unknown_type_returns_none(self):
190 ctx = RouterContext()
191 result = ctx.process_inbound(255, b"unknown")
192 assert result is None
193
194 def test_garlic_handler_wired(self):
195 """GARLIC type (11) should be wired to garlic_handler.handle."""
196 ctx = RouterContext()
197 # Sending garbage to garlic handler should return empty list
198 # (tag won't be found in session key manager)
199 result = ctx.process_inbound(11, os.urandom(128))
200 assert result == []
201
202 def test_database_store_wired(self):
203 """DATABASE_STORE type (1) should be wired to netdb_handler.handle_store."""
204 ctx = RouterContext()
205 key = os.urandom(32)
206 data = b"stored-via-inbound"
207 # Type 1 payload: key(32) || data
208 result = ctx.process_inbound(1, key + data)
209 assert result is True
210 # Verify it's in the datastore
211 assert ctx.lookup_netdb_entry(key) == data
212
213 def test_database_lookup_wired(self):
214 """DATABASE_LOOKUP type (2) should be wired to netdb_handler.handle_lookup."""
215 ctx = RouterContext()
216 key = os.urandom(32)
217 data = b"lookup-target"
218 ctx.store_netdb_entry(key, data)
219 # Type 2 payload: key(32)
220 result = ctx.process_inbound(2, key)
221 assert result == data
222
223
224class TestOutboundRouting:
225 """route_outbound returns correct routing dict."""
226
227 def test_local_delivery(self):
228 ctx = RouterContext()
229 result = ctx.route_outbound(0, b"hello")
230 assert result["type"] == "local"
231 assert result["payload"] == b"hello"
232
233 def test_router_delivery(self):
234 ctx = RouterContext()
235 rh = os.urandom(32)
236 result = ctx.route_outbound(1, b"data", router_hash=rh)
237 assert result["type"] == "router"
238 assert result["router_hash"] == rh
239
240 def test_tunnel_delivery(self):
241 ctx = RouterContext()
242 result = ctx.route_outbound(
243 2, b"tdata", tunnel_id=42, gateway=os.urandom(32)
244 )
245 assert result["type"] == "tunnel"
246 assert result["tunnel_id"] == 42
247
248 def test_destination_delivery(self):
249 ctx = RouterContext()
250 dest = os.urandom(32)
251 result = ctx.route_outbound(3, b"dest", destination=dest)
252 assert result["type"] == "destination"
253 assert result["destination"] == dest
254
255
256class TestFullWiring:
257 """Full wiring: store in NetDB then lookup via inbound handler."""
258
259 def test_store_then_lookup_via_inbound(self):
260 ctx = RouterContext()
261 key = os.urandom(32)
262 data = b"full-wiring-test"
263
264 # Store via direct API
265 ctx.store_netdb_entry(key, data)
266
267 # Lookup via inbound dispatch (type 2 = DATABASE_LOOKUP)
268 result = ctx.process_inbound(2, key)
269 assert result == data
270
271 def test_store_via_inbound_then_lookup_direct(self):
272 ctx = RouterContext()
273 key = os.urandom(32)
274 data = b"reverse-wiring-test"
275
276 # Store via inbound dispatch (type 1 = DATABASE_STORE)
277 ctx.process_inbound(1, key + data)
278
279 # Lookup via direct API
280 assert ctx.lookup_netdb_entry(key) == data
281
282 def test_tunnel_registration_reflected_in_status(self):
283 ctx = RouterContext()
284 ctx.register_tunnel(100, os.urandom(32), os.urandom(32))
285 ctx.store_netdb_entry(os.urandom(32), b"x")
286 status = ctx.get_status()
287 assert status["tunnels_registered"] == 1
288 assert status["netdb_entries"] == 1