"""Tests for I2CP Tier 2: wire format fixes + new message types.""" import struct import os import pytest from i2p_client.i2cp_messages import ( SendMessageMessage, HostLookupMessage, HostReplyMessage, ReconfigureSessionMessage, SendMessageExpiresMessage, DateAndFlags, ) from i2p_client.session_config import WireSessionConfig def _make_dest_data() -> bytes: """Create valid-looking destination data (387 bytes with null cert).""" # 256 bytes pub key area + 128 bytes sig key area + null cert (type=0, len=0) return os.urandom(384) + b"\x00\x00\x00" # --- SendMessageMessage wire format fix --- class TestSendMessageMessageFixed: def test_roundtrip_with_nonce(self): dest = _make_dest_data() # typical destination size payload = b"hello" msg = SendMessageMessage(session_id=1, destination_data=dest, payload=payload, nonce=12345) wire = msg.payload_bytes() msg2 = SendMessageMessage._from_payload(wire) assert msg2.session_id == 1 assert msg2.destination_data == dest assert msg2.payload == payload assert msg2.nonce == 12345 def test_no_dest_len_prefix(self): """Wire format should NOT have a 2-byte dest_len prefix.""" dest = _make_dest_data() msg = SendMessageMessage(session_id=5, destination_data=dest, payload=b"x", nonce=0) wire = msg.payload_bytes() # First 2 bytes: session_id sid = struct.unpack("!H", wire[:2])[0] assert sid == 5 # Next 2 bytes should NOT be dest_len (390) — they should be part of dest # In old format, bytes 2-4 would be dest_len=390 (0x0186) # In new format, bytes 2+ are raw destination bytes assert wire[2:2+len(dest)] == dest def test_nonce_at_end(self): dest = _make_dest_data() msg = SendMessageMessage(session_id=1, destination_data=dest, payload=b"test", nonce=42) wire = msg.payload_bytes() # Last 4 bytes should be nonce nonce = struct.unpack("!I", wire[-4:])[0] assert nonce == 42 def test_default_nonce_zero(self): """Backward compat: nonce defaults to 0.""" msg = SendMessageMessage(session_id=1, destination_data=_make_dest_data(), payload=b"data") assert msg.nonce == 0 # --- SessionConfig --- class TestWireSessionConfig: def test_roundtrip(self): dest_data = _make_dest_data() opts = {"inbound.length": "3", "outbound.length": "3"} sc = WireSessionConfig(dest_data, options=opts) wire = sc.to_bytes() sc2 = WireSessionConfig.from_bytes(wire) assert sc2.destination_data == dest_data assert sc2.options == opts def test_date_field(self): dest_data = _make_dest_data() sc = WireSessionConfig(dest_data, date_ms=1700000000000) wire = sc.to_bytes() sc2 = WireSessionConfig.from_bytes(wire) assert sc2.date_ms == 1700000000000 def test_empty_options(self): dest_data = _make_dest_data() sc = WireSessionConfig(dest_data) wire = sc.to_bytes() sc2 = WireSessionConfig.from_bytes(wire) assert sc2.options == {} def test_signature_field(self): dest_data = _make_dest_data() sig = os.urandom(64) sc = WireSessionConfig(dest_data, signature=sig) wire = sc.to_bytes() sc2 = WireSessionConfig.from_bytes(wire) assert sc2.signature == sig # --- HostLookupMessage wire format fix --- class TestHostLookupMessageFixed: def test_roundtrip_by_name(self): msg = HostLookupMessage(session_id=1, request_id=100, hostname="example.i2p", timeout=15000) wire = msg.payload_bytes() msg2 = HostLookupMessage._from_payload(wire) assert msg2.session_id == 1 assert msg2.request_id == 100 assert msg2.hostname == "example.i2p" assert msg2.timeout == 15000 def test_roundtrip_by_hash(self): h = os.urandom(32) msg = HostLookupMessage(session_id=2, request_id=200, dest_hash=h, timeout=5000) wire = msg.payload_bytes() msg2 = HostLookupMessage._from_payload(wire) assert msg2.dest_hash == h assert msg2.timeout == 5000 def test_timeout_field_position(self): """Timeout should be at offset 6 (after session_id+request_id).""" msg = HostLookupMessage(session_id=0, request_id=0, dest_hash=os.urandom(32), timeout=9999) wire = msg.payload_bytes() timeout = struct.unpack("!I", wire[6:10])[0] assert timeout == 9999 def test_type_codes_match_java(self): """Java: 0=hash lookup, 1=host lookup.""" # Hash lookup h = os.urandom(32) msg = HostLookupMessage(session_id=0, request_id=0, dest_hash=h, timeout=10000) wire = msg.payload_bytes() lookup_type = wire[10] # after session_id(2)+req_id(4)+timeout(4) assert lookup_type == HostLookupMessage.LOOKUP_HASH # 0 # Host lookup msg2 = HostLookupMessage(session_id=0, request_id=0, hostname="test.i2p", timeout=10000) wire2 = msg2.payload_bytes() lookup_type2 = wire2[10] assert lookup_type2 == HostLookupMessage.LOOKUP_HOST # 1 # --- HostReplyMessage wire format fix --- class TestHostReplyMessageFixed: def test_no_dest_len_in_wire(self): """Success reply should NOT have 2-byte dest_len prefix.""" dest = _make_dest_data() msg = HostReplyMessage(session_id=1, request_id=100, result_code=0, destination_data=dest) wire = msg.payload_bytes() # After header (2+4+1=7 bytes), dest data starts directly assert wire[7:7+len(dest)] == dest def test_all_result_codes(self): for code in range(6): msg = HostReplyMessage(session_id=1, request_id=1, result_code=code) wire = msg.payload_bytes() msg2 = HostReplyMessage._from_payload(wire) assert msg2.result_code == code def test_roundtrip_success_with_dest(self): dest = _make_dest_data() msg = HostReplyMessage(session_id=5, request_id=42, result_code=0, destination_data=dest) wire = msg.payload_bytes() msg2 = HostReplyMessage._from_payload(wire) assert msg2.destination_data == dest assert msg2.result_code == 0 # --- ReconfigureSessionMessage --- class TestReconfigureSessionMessage: def test_roundtrip(self): sc = WireSessionConfig(_make_dest_data(), options={"foo": "bar"}) msg = ReconfigureSessionMessage(session_id=7, session_config=sc) wire = msg.payload_bytes() msg2 = ReconfigureSessionMessage._from_payload(wire) assert msg2.session_id == 7 assert msg2.session_config.options == {"foo": "bar"} def test_type_code(self): assert ReconfigureSessionMessage.TYPE == 2 # --- DateAndFlags --- class TestDateAndFlags: def test_roundtrip(self): df = DateAndFlags(date_ms=1700000000000, flags=0x0001) raw = df.to_bytes() assert len(raw) == 8 df2 = DateAndFlags.from_bytes(raw) assert df2.date_ms == 1700000000000 assert df2.flags == 0x0001 def test_zero_flags(self): df = DateAndFlags(date_ms=5000, flags=0) raw = df.to_bytes() df2 = DateAndFlags.from_bytes(raw) assert df2.date_ms == 5000 assert df2.flags == 0 # --- SendMessageExpiresMessage --- class TestSendMessageExpiresMessage: def test_roundtrip(self): dest = _make_dest_data() msg = SendMessageExpiresMessage( session_id=3, destination_data=dest, payload=b"data", nonce=99, expiration=DateAndFlags(date_ms=1700000000000, flags=0x0003), ) wire = msg.payload_bytes() msg2 = SendMessageExpiresMessage._from_payload(wire) assert msg2.session_id == 3 assert msg2.destination_data == dest assert msg2.payload == b"data" assert msg2.nonce == 99 assert msg2.expiration.date_ms == 1700000000000 assert msg2.expiration.flags == 0x0003 def test_type_code(self): assert SendMessageExpiresMessage.TYPE == 36 def test_expiration_at_end(self): dest = _make_dest_data() msg = SendMessageExpiresMessage( session_id=1, destination_data=dest, payload=b"x", nonce=0, expiration=DateAndFlags(date_ms=42, flags=0), ) wire = msg.payload_bytes() # Last 8 bytes are expiration exp_raw = wire[-8:] df = DateAndFlags.from_bytes(exp_raw) assert df.date_ms == 42