distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Router configuration with sensible defaults and environment variable overrides.
2
3Ported from net.i2p.router.RouterContext configuration handling.
4"""
5
6from __future__ import annotations
7
8import os
9from dataclasses import dataclass, field, fields
10
11
12@dataclass
13class RouterConfig:
14 """I2P router configuration with sensible defaults."""
15
16 router_name: str = "i2p-python-router"
17 listen_host: str = "0.0.0.0"
18 listen_port: int = 9700
19
20 # Tunnel pool settings
21 inbound_tunnel_count: int = 3
22 outbound_tunnel_count: int = 3
23 tunnel_length: int = 3
24 tunnel_lifetime_seconds: int = 600
25
26 # Bandwidth
27 bandwidth_limit_kbps: int = 0 # 0 = unlimited
28
29 # NetDB
30 floodfill: bool = False
31 reseed_urls: list[str] = field(default_factory=list)
32
33 # Timeouts
34 handshake_timeout_seconds: int = 30
35 idle_timeout_seconds: int = 300
36
37 # Health endpoint
38 health_port: int = 9701
39
40 # Data directory
41 data_dir: str = "~/.i2p-python"
42
43 @classmethod
44 def from_dict(cls, d: dict) -> "RouterConfig":
45 """Create a RouterConfig from a dictionary, ignoring unknown keys.
46
47 Parameters
48 ----------
49 d:
50 Dictionary of configuration values. Unknown keys are ignored.
51
52 Returns
53 -------
54 RouterConfig
55 A new config instance with provided values overriding defaults.
56 """
57 valid_keys = {f.name for f in fields(cls)}
58 filtered = {k: v for k, v in d.items() if k in valid_keys}
59 return cls(**filtered)
60
61 @classmethod
62 def with_env_overrides(cls, base: "RouterConfig | None" = None) -> "RouterConfig":
63 """Apply environment variable overrides to a base config.
64
65 Environment variable mapping:
66 I2P_ROUTER_NAME -> router_name
67 I2P_LISTEN_HOST -> listen_host
68 I2P_LISTEN_PORT -> listen_port (int)
69 I2P_INBOUND_TUNNEL_COUNT -> inbound_tunnel_count (int)
70 I2P_OUTBOUND_TUNNEL_COUNT -> outbound_tunnel_count (int)
71 I2P_TUNNEL_LENGTH -> tunnel_length (int)
72 I2P_TUNNEL_LIFETIME -> tunnel_lifetime_seconds (int)
73 I2P_BANDWIDTH_LIMIT_KBPS -> bandwidth_limit_kbps (int)
74 I2P_FLOODFILL -> floodfill (bool: "true"/"false")
75 I2P_HANDSHAKE_TIMEOUT -> handshake_timeout_seconds (int)
76 I2P_IDLE_TIMEOUT -> idle_timeout_seconds (int)
77 I2P_DATA_DIR -> data_dir
78
79 Parameters
80 ----------
81 base:
82 Base config to apply overrides to. If None, uses defaults.
83
84 Returns
85 -------
86 RouterConfig
87 Config with environment variable overrides applied.
88 """
89 if base is None:
90 base = cls()
91
92 # String overrides
93 _str_overrides = {
94 "I2P_ROUTER_NAME": "router_name",
95 "I2P_LISTEN_HOST": "listen_host",
96 "I2P_DATA_DIR": "data_dir",
97 }
98
99 # Int overrides
100 _int_overrides = {
101 "I2P_LISTEN_PORT": "listen_port",
102 "I2P_INBOUND_TUNNEL_COUNT": "inbound_tunnel_count",
103 "I2P_OUTBOUND_TUNNEL_COUNT": "outbound_tunnel_count",
104 "I2P_TUNNEL_LENGTH": "tunnel_length",
105 "I2P_TUNNEL_LIFETIME": "tunnel_lifetime_seconds",
106 "I2P_BANDWIDTH_LIMIT_KBPS": "bandwidth_limit_kbps",
107 "I2P_HANDSHAKE_TIMEOUT": "handshake_timeout_seconds",
108 "I2P_IDLE_TIMEOUT": "idle_timeout_seconds",
109 "I2P_HEALTH_PORT": "health_port",
110 }
111
112 # Bool overrides
113 _bool_overrides = {
114 "I2P_FLOODFILL": "floodfill",
115 }
116
117 for env_key, attr in _str_overrides.items():
118 val = os.environ.get(env_key)
119 if val is not None:
120 object.__setattr__(base, attr, val)
121
122 for env_key, attr in _int_overrides.items():
123 val = os.environ.get(env_key)
124 if val is not None:
125 object.__setattr__(base, attr, int(val))
126
127 for env_key, attr in _bool_overrides.items():
128 val = os.environ.get(env_key)
129 if val is not None:
130 object.__setattr__(base, attr, val.lower() == "true")
131
132 return base
133
134 def validate(self) -> None:
135 """Validate configuration values.
136
137 Raises
138 ------
139 ValueError
140 If any configuration value is invalid.
141 """
142 if not (0 <= self.listen_port <= 65535):
143 raise ValueError(
144 f"listen_port must be 0-65535, got {self.listen_port}"
145 )
146 if self.inbound_tunnel_count < 0:
147 raise ValueError(
148 f"inbound_tunnel_count must be non-negative, got {self.inbound_tunnel_count}"
149 )
150 if self.outbound_tunnel_count < 0:
151 raise ValueError(
152 f"outbound_tunnel_count must be non-negative, got {self.outbound_tunnel_count}"
153 )
154 if self.tunnel_length < 0:
155 raise ValueError(
156 f"tunnel_length must be non-negative, got {self.tunnel_length}"
157 )
158 if self.tunnel_lifetime_seconds < 0:
159 raise ValueError(
160 f"tunnel_lifetime_seconds must be non-negative, got {self.tunnel_lifetime_seconds}"
161 )
162 if self.bandwidth_limit_kbps < 0:
163 raise ValueError(
164 f"bandwidth_limit_kbps must be non-negative, got {self.bandwidth_limit_kbps}"
165 )
166 if self.handshake_timeout_seconds < 0:
167 raise ValueError(
168 f"handshake_timeout_seconds must be non-negative, got {self.handshake_timeout_seconds}"
169 )
170 if self.idle_timeout_seconds < 0:
171 raise ValueError(
172 f"idle_timeout_seconds must be non-negative, got {self.idle_timeout_seconds}"
173 )