"""Router configuration with sensible defaults and environment variable overrides. Ported from net.i2p.router.RouterContext configuration handling. """ from __future__ import annotations import os from dataclasses import dataclass, field, fields @dataclass class RouterConfig: """I2P router configuration with sensible defaults.""" router_name: str = "i2p-python-router" listen_host: str = "0.0.0.0" listen_port: int = 9700 # Tunnel pool settings inbound_tunnel_count: int = 3 outbound_tunnel_count: int = 3 tunnel_length: int = 3 tunnel_lifetime_seconds: int = 600 # Bandwidth bandwidth_limit_kbps: int = 0 # 0 = unlimited # NetDB floodfill: bool = False reseed_urls: list[str] = field(default_factory=list) # Timeouts handshake_timeout_seconds: int = 30 idle_timeout_seconds: int = 300 # Health endpoint health_port: int = 9701 # Data directory data_dir: str = "~/.i2p-python" @classmethod def from_dict(cls, d: dict) -> "RouterConfig": """Create a RouterConfig from a dictionary, ignoring unknown keys. Parameters ---------- d: Dictionary of configuration values. Unknown keys are ignored. Returns ------- RouterConfig A new config instance with provided values overriding defaults. """ valid_keys = {f.name for f in fields(cls)} filtered = {k: v for k, v in d.items() if k in valid_keys} return cls(**filtered) @classmethod def with_env_overrides(cls, base: "RouterConfig | None" = None) -> "RouterConfig": """Apply environment variable overrides to a base config. Environment variable mapping: I2P_ROUTER_NAME -> router_name I2P_LISTEN_HOST -> listen_host I2P_LISTEN_PORT -> listen_port (int) I2P_INBOUND_TUNNEL_COUNT -> inbound_tunnel_count (int) I2P_OUTBOUND_TUNNEL_COUNT -> outbound_tunnel_count (int) I2P_TUNNEL_LENGTH -> tunnel_length (int) I2P_TUNNEL_LIFETIME -> tunnel_lifetime_seconds (int) I2P_BANDWIDTH_LIMIT_KBPS -> bandwidth_limit_kbps (int) I2P_FLOODFILL -> floodfill (bool: "true"/"false") I2P_HANDSHAKE_TIMEOUT -> handshake_timeout_seconds (int) I2P_IDLE_TIMEOUT -> idle_timeout_seconds (int) I2P_DATA_DIR -> data_dir Parameters ---------- base: Base config to apply overrides to. If None, uses defaults. Returns ------- RouterConfig Config with environment variable overrides applied. """ if base is None: base = cls() # String overrides _str_overrides = { "I2P_ROUTER_NAME": "router_name", "I2P_LISTEN_HOST": "listen_host", "I2P_DATA_DIR": "data_dir", } # Int overrides _int_overrides = { "I2P_LISTEN_PORT": "listen_port", "I2P_INBOUND_TUNNEL_COUNT": "inbound_tunnel_count", "I2P_OUTBOUND_TUNNEL_COUNT": "outbound_tunnel_count", "I2P_TUNNEL_LENGTH": "tunnel_length", "I2P_TUNNEL_LIFETIME": "tunnel_lifetime_seconds", "I2P_BANDWIDTH_LIMIT_KBPS": "bandwidth_limit_kbps", "I2P_HANDSHAKE_TIMEOUT": "handshake_timeout_seconds", "I2P_IDLE_TIMEOUT": "idle_timeout_seconds", "I2P_HEALTH_PORT": "health_port", } # Bool overrides _bool_overrides = { "I2P_FLOODFILL": "floodfill", } for env_key, attr in _str_overrides.items(): val = os.environ.get(env_key) if val is not None: object.__setattr__(base, attr, val) for env_key, attr in _int_overrides.items(): val = os.environ.get(env_key) if val is not None: object.__setattr__(base, attr, int(val)) for env_key, attr in _bool_overrides.items(): val = os.environ.get(env_key) if val is not None: object.__setattr__(base, attr, val.lower() == "true") return base def validate(self) -> None: """Validate configuration values. Raises ------ ValueError If any configuration value is invalid. """ if not (0 <= self.listen_port <= 65535): raise ValueError( f"listen_port must be 0-65535, got {self.listen_port}" ) if self.inbound_tunnel_count < 0: raise ValueError( f"inbound_tunnel_count must be non-negative, got {self.inbound_tunnel_count}" ) if self.outbound_tunnel_count < 0: raise ValueError( f"outbound_tunnel_count must be non-negative, got {self.outbound_tunnel_count}" ) if self.tunnel_length < 0: raise ValueError( f"tunnel_length must be non-negative, got {self.tunnel_length}" ) if self.tunnel_lifetime_seconds < 0: raise ValueError( f"tunnel_lifetime_seconds must be non-negative, got {self.tunnel_lifetime_seconds}" ) if self.bandwidth_limit_kbps < 0: raise ValueError( f"bandwidth_limit_kbps must be non-negative, got {self.bandwidth_limit_kbps}" ) if self.handshake_timeout_seconds < 0: raise ValueError( f"handshake_timeout_seconds must be non-negative, got {self.handshake_timeout_seconds}" ) if self.idle_timeout_seconds < 0: raise ValueError( f"idle_timeout_seconds must be non-negative, got {self.idle_timeout_seconds}" )