"""Tests for MessageWrapper, GarlicCloveBuilder, and GarlicMessageBuilder.""" import os import time from i2p_crypto.session_key_manager import SessionKeyManager from i2p_router.message_wrapper import MessageWrapper from i2p_router.garlic_clove_builder import GarlicCloveBuilder from i2p_router.garlic_message_builder import GarlicMessageBuilder class TestMessageWrapper: def test_wrap_produces_output(self): key = os.urandom(32) tag = os.urandom(32) wrapped = MessageWrapper.wrap(b"hello", key, tag) # Should start with the tag (32 bytes) assert wrapped[:32] == tag assert len(wrapped) > 32 def test_generate_session(self): skm = SessionKeyManager() key, tag = MessageWrapper.generate_session(skm) assert len(key) == 32 assert len(tag) == 32 # Tag should be consumable assert skm.consume_tag(tag) == key class TestGarlicCloveBuilder: def test_build_data_clove(self): dest_hash = os.urandom(32) clove = GarlicCloveBuilder.build_data_clove(dest_hash, b"payload") assert clove.delivery_type == 1 # DESTINATION assert clove.dest_hash == dest_hash assert len(clove.message_data) > 0 def test_build_ack_clove(self): gw = os.urandom(32) clove = GarlicCloveBuilder.build_ack_clove( reply_token=12345, our_ib_gateway=gw, our_ib_tunnel_id=42 ) assert clove.delivery_type == 3 # TUNNEL assert clove.router_hash == gw assert clove.tunnel_id == 42 def test_build_leaseset_clove(self): ls_bytes = os.urandom(200) clove = GarlicCloveBuilder.build_leaseset_clove(ls_bytes) assert clove.delivery_type == 0 # LOCAL assert len(clove.message_data) > 0 def test_create_config_ordering(self): dest_key = os.urandom(256) data = GarlicCloveBuilder.build_data_clove(os.urandom(32), b"data") ack = GarlicCloveBuilder.build_ack_clove(1, os.urandom(32), 1) ls = GarlicCloveBuilder.build_leaseset_clove(b"ls") config = GarlicCloveBuilder.create_garlic_config( dest_key, data, ack_clove=ack, ls_clove=ls ) # Order should be: ACK, LS, Data assert config.clove_count() == 3 assert config.cloves[0] is ack assert config.cloves[1] is ls assert config.cloves[2] is data def test_data_only_config(self): dest_key = os.urandom(256) data = GarlicCloveBuilder.build_data_clove(os.urandom(32), b"data") config = GarlicCloveBuilder.create_garlic_config(dest_key, data) assert config.clove_count() == 1 class TestGarlicMessageBuilder: def test_build_single_clove(self): skm = SessionKeyManager() dest_key = os.urandom(256) dest_hash = dest_key[:32] # Create a session so we have tags skm.create_session(dest_hash) from i2p_data.garlic_config import GarlicConfig, CloveConfig config = GarlicConfig( recipient_public_key=dest_key, message_id=42, expiration=int(time.time() * 1000) + 60_000, ) config.add_clove(CloveConfig.for_local( message_data=b"test", clove_id=1, expiration=int(time.time() * 1000) + 60_000 )) encrypted, _ = GarlicMessageBuilder.build_message(config, skm) assert len(encrypted) > 0 def test_build_pads_to_block_boundary(self): data = b"x" * 17 padded = GarlicMessageBuilder._pad_to_block_boundary(data) assert len(padded) % 16 == 0 assert len(padded) == 32 def test_build_clove_set_format(self): from i2p_data.garlic_config import GarlicConfig, CloveConfig config = GarlicConfig( recipient_public_key=b"\x00" * 32, message_id=1, expiration=1000, ) config.add_clove(CloveConfig.for_local( message_data=b"hi", clove_id=1, expiration=1000 )) raw = GarlicMessageBuilder._build_clove_set(config) # First byte is clove count assert raw[0] == 1