distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Tests for Garlic encryption types."""
2
3import os
4import struct
5
6import pytest
7
8
9class TestDeliveryInstructions:
10 def test_enum_values(self):
11 from i2p_data.garlic import DeliveryType
12 assert DeliveryType.LOCAL == 0
13 assert DeliveryType.DESTINATION == 1
14 assert DeliveryType.ROUTER == 2
15 assert DeliveryType.TUNNEL == 3
16
17 def test_local_roundtrip(self):
18 from i2p_data.garlic import DeliveryInstructions, DeliveryType
19 di = DeliveryInstructions(DeliveryType.LOCAL)
20 data = di.to_bytes()
21 di2, consumed = DeliveryInstructions.from_bytes(data)
22 assert di2.delivery_type == DeliveryType.LOCAL
23 assert di2.destination is None
24 assert di2.router is None
25 assert di2.tunnel_id is None
26
27 def test_destination_roundtrip(self):
28 from i2p_data.garlic import DeliveryInstructions, DeliveryType
29 dest = os.urandom(32)
30 di = DeliveryInstructions(DeliveryType.DESTINATION, destination=dest)
31 data = di.to_bytes()
32 di2, consumed = DeliveryInstructions.from_bytes(data)
33 assert di2.delivery_type == DeliveryType.DESTINATION
34 assert di2.destination == dest
35
36 def test_router_roundtrip(self):
37 from i2p_data.garlic import DeliveryInstructions, DeliveryType
38 router = os.urandom(32)
39 di = DeliveryInstructions(DeliveryType.ROUTER, router=router)
40 data = di.to_bytes()
41 di2, consumed = DeliveryInstructions.from_bytes(data)
42 assert di2.delivery_type == DeliveryType.ROUTER
43 assert di2.router == router
44
45 def test_tunnel_roundtrip(self):
46 from i2p_data.garlic import DeliveryInstructions, DeliveryType
47 router = os.urandom(32)
48 di = DeliveryInstructions(DeliveryType.TUNNEL, router=router, tunnel_id=42)
49 data = di.to_bytes()
50 di2, consumed = DeliveryInstructions.from_bytes(data)
51 assert di2.delivery_type == DeliveryType.TUNNEL
52 assert di2.router == router
53 assert di2.tunnel_id == 42
54
55
56class TestGarlicClove:
57 def test_construct(self):
58 from i2p_data.garlic import GarlicClove, DeliveryInstructions, DeliveryType
59 di = DeliveryInstructions(DeliveryType.LOCAL)
60 msg_data = os.urandom(64)
61 clove = GarlicClove(di, msg_data, clove_id=123, expiration=1710000000000)
62 assert clove.delivery_instructions.delivery_type == DeliveryType.LOCAL
63 assert clove.message_data == msg_data
64 assert clove.clove_id == 123
65 assert clove.expiration == 1710000000000
66
67 def test_roundtrip_local(self):
68 from i2p_data.garlic import GarlicClove, DeliveryInstructions, DeliveryType
69 di = DeliveryInstructions(DeliveryType.LOCAL)
70 msg = os.urandom(32)
71 clove = GarlicClove(di, msg, clove_id=999, expiration=5000)
72 data = clove.to_bytes()
73 clove2 = GarlicClove.from_bytes(data)
74 assert clove2.clove_id == 999
75 assert clove2.expiration == 5000
76 assert clove2.message_data == msg
77 assert clove2.delivery_instructions.delivery_type == DeliveryType.LOCAL
78
79 def test_roundtrip_tunnel(self):
80 from i2p_data.garlic import GarlicClove, DeliveryInstructions, DeliveryType
81 router = os.urandom(32)
82 di = DeliveryInstructions(DeliveryType.TUNNEL, router=router, tunnel_id=777)
83 msg = os.urandom(100)
84 clove = GarlicClove(di, msg, clove_id=42, expiration=9999)
85 data = clove.to_bytes()
86 clove2 = GarlicClove.from_bytes(data)
87 assert clove2.delivery_instructions.delivery_type == DeliveryType.TUNNEL
88 assert clove2.delivery_instructions.router == router
89 assert clove2.delivery_instructions.tunnel_id == 777
90 assert clove2.message_data == msg
91
92
93class TestGarlicMessage:
94 def test_construct(self):
95 from i2p_data.garlic import GarlicMessage, GarlicClove, DeliveryInstructions, DeliveryType
96 di = DeliveryInstructions(DeliveryType.LOCAL)
97 clove = GarlicClove(di, b"msg1", clove_id=1, expiration=1000)
98 gm = GarlicMessage([clove])
99 assert gm.clove_count() == 1
100
101 def test_multiple_cloves_roundtrip(self):
102 from i2p_data.garlic import GarlicMessage, GarlicClove, DeliveryInstructions, DeliveryType
103 cloves = []
104 for i in range(3):
105 di = DeliveryInstructions(DeliveryType.LOCAL)
106 cloves.append(GarlicClove(di, os.urandom(16), clove_id=i, expiration=1000 + i))
107 gm = GarlicMessage(cloves)
108 data = gm.to_bytes()
109 gm2 = GarlicMessage.from_bytes(data)
110 assert gm2.clove_count() == 3
111 for i in range(3):
112 assert gm2.cloves[i].clove_id == i
113 assert gm2.cloves[i].expiration == 1000 + i
114
115 def test_empty_cloves(self):
116 from i2p_data.garlic import GarlicMessage
117 gm = GarlicMessage([])
118 assert gm.clove_count() == 0
119 data = gm.to_bytes()
120 gm2 = GarlicMessage.from_bytes(data)
121 assert gm2.clove_count() == 0
122
123 def test_mixed_delivery_types(self):
124 from i2p_data.garlic import GarlicMessage, GarlicClove, DeliveryInstructions, DeliveryType
125 cloves = [
126 GarlicClove(DeliveryInstructions(DeliveryType.LOCAL),
127 b"local", clove_id=1, expiration=100),
128 GarlicClove(DeliveryInstructions(DeliveryType.ROUTER, router=os.urandom(32)),
129 b"router", clove_id=2, expiration=200),
130 GarlicClove(DeliveryInstructions(DeliveryType.TUNNEL, router=os.urandom(32), tunnel_id=55),
131 b"tunnel", clove_id=3, expiration=300),
132 ]
133 gm = GarlicMessage(cloves)
134 data = gm.to_bytes()
135 gm2 = GarlicMessage.from_bytes(data)
136 assert gm2.cloves[0].delivery_instructions.delivery_type == DeliveryType.LOCAL
137 assert gm2.cloves[1].delivery_instructions.delivery_type == DeliveryType.ROUTER
138 assert gm2.cloves[2].delivery_instructions.delivery_type == DeliveryType.TUNNEL
139 assert gm2.cloves[2].delivery_instructions.tunnel_id == 55