distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for I2CP Tier 2: wire format fixes + new message types."""
2
3import struct
4import os
5import pytest
6
7from i2p_client.i2cp_messages import (
8 SendMessageMessage,
9 HostLookupMessage,
10 HostReplyMessage,
11 ReconfigureSessionMessage,
12 SendMessageExpiresMessage,
13 DateAndFlags,
14)
15from i2p_client.session_config import WireSessionConfig
16
17
18def _make_dest_data() -> bytes:
19 """Create valid-looking destination data (387 bytes with null cert)."""
20 # 256 bytes pub key area + 128 bytes sig key area + null cert (type=0, len=0)
21 return os.urandom(384) + b"\x00\x00\x00"
22
23
24# --- SendMessageMessage wire format fix ---
25
26class TestSendMessageMessageFixed:
27 def test_roundtrip_with_nonce(self):
28 dest = _make_dest_data() # typical destination size
29 payload = b"hello"
30 msg = SendMessageMessage(session_id=1, destination_data=dest,
31 payload=payload, nonce=12345)
32 wire = msg.payload_bytes()
33 msg2 = SendMessageMessage._from_payload(wire)
34 assert msg2.session_id == 1
35 assert msg2.destination_data == dest
36 assert msg2.payload == payload
37 assert msg2.nonce == 12345
38
39 def test_no_dest_len_prefix(self):
40 """Wire format should NOT have a 2-byte dest_len prefix."""
41 dest = _make_dest_data()
42 msg = SendMessageMessage(session_id=5, destination_data=dest,
43 payload=b"x", nonce=0)
44 wire = msg.payload_bytes()
45 # First 2 bytes: session_id
46 sid = struct.unpack("!H", wire[:2])[0]
47 assert sid == 5
48 # Next 2 bytes should NOT be dest_len (390) — they should be part of dest
49 # In old format, bytes 2-4 would be dest_len=390 (0x0186)
50 # In new format, bytes 2+ are raw destination bytes
51 assert wire[2:2+len(dest)] == dest
52
53 def test_nonce_at_end(self):
54 dest = _make_dest_data()
55 msg = SendMessageMessage(session_id=1, destination_data=dest,
56 payload=b"test", nonce=42)
57 wire = msg.payload_bytes()
58 # Last 4 bytes should be nonce
59 nonce = struct.unpack("!I", wire[-4:])[0]
60 assert nonce == 42
61
62 def test_default_nonce_zero(self):
63 """Backward compat: nonce defaults to 0."""
64 msg = SendMessageMessage(session_id=1, destination_data=_make_dest_data(),
65 payload=b"data")
66 assert msg.nonce == 0
67
68
69# --- SessionConfig ---
70
71class TestWireSessionConfig:
72 def test_roundtrip(self):
73 dest_data = _make_dest_data()
74 opts = {"inbound.length": "3", "outbound.length": "3"}
75 sc = WireSessionConfig(dest_data, options=opts)
76 wire = sc.to_bytes()
77 sc2 = WireSessionConfig.from_bytes(wire)
78 assert sc2.destination_data == dest_data
79 assert sc2.options == opts
80
81 def test_date_field(self):
82 dest_data = _make_dest_data()
83 sc = WireSessionConfig(dest_data, date_ms=1700000000000)
84 wire = sc.to_bytes()
85 sc2 = WireSessionConfig.from_bytes(wire)
86 assert sc2.date_ms == 1700000000000
87
88 def test_empty_options(self):
89 dest_data = _make_dest_data()
90 sc = WireSessionConfig(dest_data)
91 wire = sc.to_bytes()
92 sc2 = WireSessionConfig.from_bytes(wire)
93 assert sc2.options == {}
94
95 def test_signature_field(self):
96 dest_data = _make_dest_data()
97 sig = os.urandom(64)
98 sc = WireSessionConfig(dest_data, signature=sig)
99 wire = sc.to_bytes()
100 sc2 = WireSessionConfig.from_bytes(wire)
101 assert sc2.signature == sig
102
103
104# --- HostLookupMessage wire format fix ---
105
106class TestHostLookupMessageFixed:
107 def test_roundtrip_by_name(self):
108 msg = HostLookupMessage(session_id=1, request_id=100,
109 hostname="example.i2p", timeout=15000)
110 wire = msg.payload_bytes()
111 msg2 = HostLookupMessage._from_payload(wire)
112 assert msg2.session_id == 1
113 assert msg2.request_id == 100
114 assert msg2.hostname == "example.i2p"
115 assert msg2.timeout == 15000
116
117 def test_roundtrip_by_hash(self):
118 h = os.urandom(32)
119 msg = HostLookupMessage(session_id=2, request_id=200,
120 dest_hash=h, timeout=5000)
121 wire = msg.payload_bytes()
122 msg2 = HostLookupMessage._from_payload(wire)
123 assert msg2.dest_hash == h
124 assert msg2.timeout == 5000
125
126 def test_timeout_field_position(self):
127 """Timeout should be at offset 6 (after session_id+request_id)."""
128 msg = HostLookupMessage(session_id=0, request_id=0,
129 dest_hash=os.urandom(32), timeout=9999)
130 wire = msg.payload_bytes()
131 timeout = struct.unpack("!I", wire[6:10])[0]
132 assert timeout == 9999
133
134 def test_type_codes_match_java(self):
135 """Java: 0=hash lookup, 1=host lookup."""
136 # Hash lookup
137 h = os.urandom(32)
138 msg = HostLookupMessage(session_id=0, request_id=0,
139 dest_hash=h, timeout=10000)
140 wire = msg.payload_bytes()
141 lookup_type = wire[10] # after session_id(2)+req_id(4)+timeout(4)
142 assert lookup_type == HostLookupMessage.LOOKUP_HASH # 0
143
144 # Host lookup
145 msg2 = HostLookupMessage(session_id=0, request_id=0,
146 hostname="test.i2p", timeout=10000)
147 wire2 = msg2.payload_bytes()
148 lookup_type2 = wire2[10]
149 assert lookup_type2 == HostLookupMessage.LOOKUP_HOST # 1
150
151
152# --- HostReplyMessage wire format fix ---
153
154class TestHostReplyMessageFixed:
155 def test_no_dest_len_in_wire(self):
156 """Success reply should NOT have 2-byte dest_len prefix."""
157 dest = _make_dest_data()
158 msg = HostReplyMessage(session_id=1, request_id=100,
159 result_code=0, destination_data=dest)
160 wire = msg.payload_bytes()
161 # After header (2+4+1=7 bytes), dest data starts directly
162 assert wire[7:7+len(dest)] == dest
163
164 def test_all_result_codes(self):
165 for code in range(6):
166 msg = HostReplyMessage(session_id=1, request_id=1, result_code=code)
167 wire = msg.payload_bytes()
168 msg2 = HostReplyMessage._from_payload(wire)
169 assert msg2.result_code == code
170
171 def test_roundtrip_success_with_dest(self):
172 dest = _make_dest_data()
173 msg = HostReplyMessage(session_id=5, request_id=42,
174 result_code=0, destination_data=dest)
175 wire = msg.payload_bytes()
176 msg2 = HostReplyMessage._from_payload(wire)
177 assert msg2.destination_data == dest
178 assert msg2.result_code == 0
179
180
181# --- ReconfigureSessionMessage ---
182
183class TestReconfigureSessionMessage:
184 def test_roundtrip(self):
185 sc = WireSessionConfig(_make_dest_data(), options={"foo": "bar"})
186 msg = ReconfigureSessionMessage(session_id=7, session_config=sc)
187 wire = msg.payload_bytes()
188 msg2 = ReconfigureSessionMessage._from_payload(wire)
189 assert msg2.session_id == 7
190 assert msg2.session_config.options == {"foo": "bar"}
191
192 def test_type_code(self):
193 assert ReconfigureSessionMessage.TYPE == 2
194
195
196# --- DateAndFlags ---
197
198class TestDateAndFlags:
199 def test_roundtrip(self):
200 df = DateAndFlags(date_ms=1700000000000, flags=0x0001)
201 raw = df.to_bytes()
202 assert len(raw) == 8
203 df2 = DateAndFlags.from_bytes(raw)
204 assert df2.date_ms == 1700000000000
205 assert df2.flags == 0x0001
206
207 def test_zero_flags(self):
208 df = DateAndFlags(date_ms=5000, flags=0)
209 raw = df.to_bytes()
210 df2 = DateAndFlags.from_bytes(raw)
211 assert df2.date_ms == 5000
212 assert df2.flags == 0
213
214
215# --- SendMessageExpiresMessage ---
216
217class TestSendMessageExpiresMessage:
218 def test_roundtrip(self):
219 dest = _make_dest_data()
220 msg = SendMessageExpiresMessage(
221 session_id=3, destination_data=dest,
222 payload=b"data", nonce=99,
223 expiration=DateAndFlags(date_ms=1700000000000, flags=0x0003),
224 )
225 wire = msg.payload_bytes()
226 msg2 = SendMessageExpiresMessage._from_payload(wire)
227 assert msg2.session_id == 3
228 assert msg2.destination_data == dest
229 assert msg2.payload == b"data"
230 assert msg2.nonce == 99
231 assert msg2.expiration.date_ms == 1700000000000
232 assert msg2.expiration.flags == 0x0003
233
234 def test_type_code(self):
235 assert SendMessageExpiresMessage.TYPE == 36
236
237 def test_expiration_at_end(self):
238 dest = _make_dest_data()
239 msg = SendMessageExpiresMessage(
240 session_id=1, destination_data=dest,
241 payload=b"x", nonce=0,
242 expiration=DateAndFlags(date_ms=42, flags=0),
243 )
244 wire = msg.payload_bytes()
245 # Last 8 bytes are expiration
246 exp_raw = wire[-8:]
247 df = DateAndFlags.from_bytes(exp_raw)
248 assert df.date_ms == 42