distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Configuration objects for garlic message construction.
2
3Ported from net.i2p.router.crypto.GarlicConfig and
4PayloadGarlicConfig. These are builder-side data structures
5used before encryption — not wire format.
6"""
7
8from __future__ import annotations
9
10from dataclasses import dataclass, field
11
12NULL_CERT = b"\x00\x00\x00"
13
14
15@dataclass
16class CloveConfig:
17 """Configuration for a single garlic clove.
18
19 delivery_type values: 0=LOCAL, 1=DESTINATION, 2=ROUTER, 3=TUNNEL
20 """
21
22 delivery_type: int
23 message_data: bytes
24 clove_id: int
25 expiration: int
26 dest_hash: bytes | None = None
27 router_hash: bytes | None = None
28 tunnel_id: int | None = None
29 certificate: bytes = field(default_factory=lambda: NULL_CERT)
30 request_ack: bool = False
31
32 @classmethod
33 def for_local(
34 cls,
35 message_data: bytes,
36 clove_id: int,
37 expiration: int,
38 **kwargs,
39 ) -> CloveConfig:
40 """Create a LOCAL delivery clove."""
41 return cls(
42 delivery_type=0,
43 message_data=message_data,
44 clove_id=clove_id,
45 expiration=expiration,
46 **kwargs,
47 )
48
49 @classmethod
50 def for_destination(
51 cls,
52 dest_hash: bytes,
53 message_data: bytes,
54 clove_id: int,
55 expiration: int,
56 **kwargs,
57 ) -> CloveConfig:
58 """Create a DESTINATION delivery clove."""
59 return cls(
60 delivery_type=1,
61 dest_hash=dest_hash,
62 message_data=message_data,
63 clove_id=clove_id,
64 expiration=expiration,
65 **kwargs,
66 )
67
68 @classmethod
69 def for_tunnel(
70 cls,
71 router_hash: bytes,
72 tunnel_id: int,
73 message_data: bytes,
74 clove_id: int,
75 expiration: int,
76 **kwargs,
77 ) -> CloveConfig:
78 """Create a TUNNEL delivery clove."""
79 return cls(
80 delivery_type=3,
81 router_hash=router_hash,
82 tunnel_id=tunnel_id,
83 message_data=message_data,
84 clove_id=clove_id,
85 expiration=expiration,
86 **kwargs,
87 )
88
89
90@dataclass
91class GarlicConfig:
92 """Top-level garlic message configuration.
93
94 Holds the recipient key and a list of cloves to encrypt together.
95 """
96
97 recipient_public_key: bytes
98 message_id: int
99 expiration: int
100 cloves: list[CloveConfig] = field(default_factory=list)
101
102 def add_clove(self, clove: CloveConfig) -> None:
103 self.cloves.append(clove)
104
105 def clove_count(self) -> int:
106 return len(self.cloves)