distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for MessageWrapper, GarlicCloveBuilder, and GarlicMessageBuilder."""
2
3import os
4import time
5
6from i2p_crypto.session_key_manager import SessionKeyManager
7from i2p_router.message_wrapper import MessageWrapper
8from i2p_router.garlic_clove_builder import GarlicCloveBuilder
9from i2p_router.garlic_message_builder import GarlicMessageBuilder
10
11
12class TestMessageWrapper:
13 def test_wrap_produces_output(self):
14 key = os.urandom(32)
15 tag = os.urandom(32)
16 wrapped = MessageWrapper.wrap(b"hello", key, tag)
17 # Should start with the tag (32 bytes)
18 assert wrapped[:32] == tag
19 assert len(wrapped) > 32
20
21 def test_generate_session(self):
22 skm = SessionKeyManager()
23 key, tag = MessageWrapper.generate_session(skm)
24 assert len(key) == 32
25 assert len(tag) == 32
26 # Tag should be consumable
27 assert skm.consume_tag(tag) == key
28
29
30class TestGarlicCloveBuilder:
31 def test_build_data_clove(self):
32 dest_hash = os.urandom(32)
33 clove = GarlicCloveBuilder.build_data_clove(dest_hash, b"payload")
34 assert clove.delivery_type == 1 # DESTINATION
35 assert clove.dest_hash == dest_hash
36 assert len(clove.message_data) > 0
37
38 def test_build_ack_clove(self):
39 gw = os.urandom(32)
40 clove = GarlicCloveBuilder.build_ack_clove(
41 reply_token=12345, our_ib_gateway=gw, our_ib_tunnel_id=42
42 )
43 assert clove.delivery_type == 3 # TUNNEL
44 assert clove.router_hash == gw
45 assert clove.tunnel_id == 42
46
47 def test_build_leaseset_clove(self):
48 ls_bytes = os.urandom(200)
49 clove = GarlicCloveBuilder.build_leaseset_clove(ls_bytes)
50 assert clove.delivery_type == 0 # LOCAL
51 assert len(clove.message_data) > 0
52
53 def test_create_config_ordering(self):
54 dest_key = os.urandom(256)
55 data = GarlicCloveBuilder.build_data_clove(os.urandom(32), b"data")
56 ack = GarlicCloveBuilder.build_ack_clove(1, os.urandom(32), 1)
57 ls = GarlicCloveBuilder.build_leaseset_clove(b"ls")
58
59 config = GarlicCloveBuilder.create_garlic_config(
60 dest_key, data, ack_clove=ack, ls_clove=ls
61 )
62 # Order should be: ACK, LS, Data
63 assert config.clove_count() == 3
64 assert config.cloves[0] is ack
65 assert config.cloves[1] is ls
66 assert config.cloves[2] is data
67
68 def test_data_only_config(self):
69 dest_key = os.urandom(256)
70 data = GarlicCloveBuilder.build_data_clove(os.urandom(32), b"data")
71 config = GarlicCloveBuilder.create_garlic_config(dest_key, data)
72 assert config.clove_count() == 1
73
74
75class TestGarlicMessageBuilder:
76 def test_build_single_clove(self):
77 skm = SessionKeyManager()
78 dest_key = os.urandom(256)
79 dest_hash = dest_key[:32]
80 # Create a session so we have tags
81 skm.create_session(dest_hash)
82
83 from i2p_data.garlic_config import GarlicConfig, CloveConfig
84 config = GarlicConfig(
85 recipient_public_key=dest_key,
86 message_id=42,
87 expiration=int(time.time() * 1000) + 60_000,
88 )
89 config.add_clove(CloveConfig.for_local(
90 message_data=b"test", clove_id=1, expiration=int(time.time() * 1000) + 60_000
91 ))
92
93 encrypted, _ = GarlicMessageBuilder.build_message(config, skm)
94 assert len(encrypted) > 0
95
96 def test_build_pads_to_block_boundary(self):
97 data = b"x" * 17
98 padded = GarlicMessageBuilder._pad_to_block_boundary(data)
99 assert len(padded) % 16 == 0
100 assert len(padded) == 32
101
102 def test_build_clove_set_format(self):
103 from i2p_data.garlic_config import GarlicConfig, CloveConfig
104 config = GarlicConfig(
105 recipient_public_key=b"\x00" * 32,
106 message_id=1,
107 expiration=1000,
108 )
109 config.add_clove(CloveConfig.for_local(
110 message_data=b"hi", clove_id=1, expiration=1000
111 ))
112 raw = GarlicMessageBuilder._build_clove_set(config)
113 # First byte is clove count
114 assert raw[0] == 1