"""Tests for I2CP Tier 3: lease set message types.""" import struct import os from i2p_client.i2cp_messages import ( I2CPMessage, RequestLeaseSetMessage, RequestVariableLeaseSetMessage, CreateLeaseSetMessage, CreateLeaseSet2Message, ) # --- RequestLeaseSetMessage (Type 21) --- class TestRequestLeaseSetMessage: def test_roundtrip_single_tunnel(self): router_hash = os.urandom(32) msg = RequestLeaseSetMessage( session_id=1, tunnels=[(router_hash, 9999)], end_date=1700000000000) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert isinstance(msg2, RequestLeaseSetMessage) assert msg2.session_id == 1 assert len(msg2.tunnels) == 1 assert msg2.tunnels[0][0] == router_hash assert msg2.tunnels[0][1] == 9999 assert msg2.end_date == 1700000000000 def test_roundtrip_multiple_tunnels(self): tunnels = [(os.urandom(32), i * 100) for i in range(5)] msg = RequestLeaseSetMessage( session_id=42, tunnels=tunnels, end_date=9999999999999) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert msg2.session_id == 42 assert len(msg2.tunnels) == 5 for orig, parsed in zip(tunnels, msg2.tunnels): assert parsed[0] == orig[0] assert parsed[1] == orig[1] def test_end_date_encoding(self): """end_date is an 8-byte big-endian value at the end of the payload.""" msg = RequestLeaseSetMessage( session_id=0, tunnels=[(os.urandom(32), 1)], end_date=0xDEADBEEFCAFE) payload = msg.payload_bytes() end_date = struct.unpack("!Q", payload[-8:])[0] assert end_date == 0xDEADBEEFCAFE def test_type_code(self): assert RequestLeaseSetMessage.TYPE == 21 def test_zero_tunnels(self): msg = RequestLeaseSetMessage( session_id=7, tunnels=[], end_date=100) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert msg2.tunnels == [] assert msg2.end_date == 100 # --- RequestVariableLeaseSetMessage (Type 37) --- class TestRequestVariableLeaseSetMessage: def test_roundtrip_single_lease(self): gw = os.urandom(32) msg = RequestVariableLeaseSetMessage( session_id=3, leases=[(gw, 555, 1700000000000)]) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert isinstance(msg2, RequestVariableLeaseSetMessage) assert msg2.session_id == 3 assert len(msg2.leases) == 1 assert msg2.leases[0] == (gw, 555, 1700000000000) def test_roundtrip_multiple_leases(self): leases = [ (os.urandom(32), i * 10, 1700000000000 + i * 60000) for i in range(4) ] msg = RequestVariableLeaseSetMessage(session_id=10, leases=leases) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert msg2.session_id == 10 assert len(msg2.leases) == 4 for orig, parsed in zip(leases, msg2.leases): assert parsed == orig def test_type_code(self): assert RequestVariableLeaseSetMessage.TYPE == 37 # --- CreateLeaseSetMessage (Type 4) --- class TestCreateLeaseSetMessage: def test_roundtrip(self): spk = os.urandom(64) pk = os.urandom(32) ls_data = os.urandom(200) msg = CreateLeaseSetMessage( session_id=5, signing_private_key=spk, private_key=pk, lease_set_data=ls_data) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert isinstance(msg2, CreateLeaseSetMessage) assert msg2.session_id == 5 assert msg2.signing_private_key == spk assert msg2.private_key == pk assert msg2.lease_set_data == ls_data def test_key_length_fields(self): """Key lengths are encoded as 2-byte big-endian prefixes.""" spk = os.urandom(48) pk = os.urandom(128) msg = CreateLeaseSetMessage( session_id=0, signing_private_key=spk, private_key=pk, lease_set_data=b"") payload = msg.payload_bytes() # session_id(2) then spk_len(2) spk_len = struct.unpack("!H", payload[2:4])[0] assert spk_len == 48 # After spk, pk_len(2) pk_offset = 4 + 48 pk_len = struct.unpack("!H", payload[pk_offset:pk_offset + 2])[0] assert pk_len == 128 def test_type_code(self): assert CreateLeaseSetMessage.TYPE == 4 def test_empty_lease_set_data(self): msg = CreateLeaseSetMessage( session_id=1, signing_private_key=b"\x01" * 20, private_key=b"\x02" * 10, lease_set_data=b"") wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert msg2.lease_set_data == b"" assert msg2.signing_private_key == b"\x01" * 20 assert msg2.private_key == b"\x02" * 10 # --- CreateLeaseSet2Message (Type 41) --- class TestCreateLeaseSet2Message: def test_roundtrip(self): ls_data = os.urandom(300) keys = [(0, os.urandom(32)), (4, os.urandom(64))] msg = CreateLeaseSet2Message( session_id=8, ls_type=CreateLeaseSet2Message.LS_TYPE_LS2, lease_set_data=ls_data, private_keys=keys) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert isinstance(msg2, CreateLeaseSet2Message) assert msg2.session_id == 8 assert msg2.ls_type == 3 assert msg2.lease_set_data == ls_data assert len(msg2.private_keys) == 2 assert msg2.private_keys[0] == keys[0] assert msg2.private_keys[1] == keys[1] def test_type_code(self): assert CreateLeaseSet2Message.TYPE == 41 def test_multiple_private_keys(self): keys = [(enc_type, os.urandom(16 + enc_type)) for enc_type in (0, 4, 6)] msg = CreateLeaseSet2Message( session_id=2, ls_type=1, lease_set_data=b"test", private_keys=keys) wire = msg.to_wire() msg2 = I2CPMessage.from_wire(wire) assert len(msg2.private_keys) == 3 for orig, parsed in zip(keys, msg2.private_keys): assert parsed[0] == orig[0] assert parsed[1] == orig[1] def test_ls_type_constants(self): assert CreateLeaseSet2Message.LS_TYPE_LEASESET == 1 assert CreateLeaseSet2Message.LS_TYPE_LS2 == 3 assert CreateLeaseSet2Message.LS_TYPE_ENCRYPTED_LS2 == 5 assert CreateLeaseSet2Message.LS_TYPE_META_LS2 == 7 def test_registry_lookup(self): """Type 41 should be in the I2CP registry.""" assert I2CPMessage._registry[41] is CreateLeaseSet2Message