Add channel_keys dict to NetworkConfig for storing per-channel keys. Keys are used in KICK rejoin, passed via AUTOJOIN +#channel key syntax, supported in ADDNETWORK channel_keys= parameter, and propagated through REHASH. Extract rehash() as reusable async function for SIGHUP reuse. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
190 lines
6.4 KiB
Python
190 lines
6.4 KiB
Python
"""Configuration loader and validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
if sys.version_info >= (3, 11):
|
|
import tomllib
|
|
else:
|
|
try:
|
|
import tomllib
|
|
except ModuleNotFoundError:
|
|
import tomli as tomllib # type: ignore[no-redef]
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class BacklogConfig:
|
|
"""Backlog storage settings."""
|
|
|
|
max_messages: int = 10000
|
|
replay_on_connect: bool = True
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class ProxyConfig:
|
|
"""SOCKS5 proxy settings."""
|
|
|
|
host: str = "127.0.0.1"
|
|
port: int = 1080
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class NetworkConfig:
|
|
"""IRC network connection settings."""
|
|
|
|
name: str
|
|
host: str
|
|
port: int = 6667
|
|
tls: bool = False
|
|
nick: str = ""
|
|
user: str = ""
|
|
realname: str = ""
|
|
channels: list[str] = field(default_factory=list)
|
|
channel_keys: dict[str, str] = field(default_factory=dict)
|
|
autojoin: bool = False
|
|
password: str | None = None
|
|
proxy_host: str | None = None
|
|
proxy_port: int | None = None
|
|
auth_service: str = "nickserv" # "nickserv", "qbot", or "none"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class BouncerConfig:
|
|
"""Main bouncer settings."""
|
|
|
|
bind: str = "127.0.0.1"
|
|
port: int = 6667
|
|
password: str = "changeme"
|
|
backlog: BacklogConfig = field(default_factory=BacklogConfig)
|
|
|
|
# Captcha solving (NoCaptchaAI)
|
|
captcha_api_key: str = ""
|
|
captcha_poll_interval: int = 3
|
|
captcha_poll_timeout: int = 120
|
|
|
|
# Connection tuning
|
|
probation_seconds: int = 45
|
|
backoff_steps: list[int] = field(default_factory=lambda: [5, 10, 30, 60, 120, 300])
|
|
nick_timeout: int = 10
|
|
rejoin_delay: int = 3
|
|
http_timeout: int = 15
|
|
|
|
# Email verification
|
|
email_poll_interval: int = 15
|
|
email_max_polls: int = 30
|
|
email_request_timeout: int = 20
|
|
|
|
# Certificate generation
|
|
cert_validity_days: int = 3650
|
|
|
|
# PING watchdog
|
|
ping_interval: int = 120 # seconds of silence before sending PING
|
|
ping_timeout: int = 30 # seconds to wait for PONG after PING
|
|
|
|
# Push notifications
|
|
notify_url: str = "" # ntfy/webhook URL (empty = disabled)
|
|
notify_on_highlight: bool = True
|
|
notify_on_privmsg: bool = True
|
|
notify_cooldown: int = 60 # min seconds between notifications
|
|
notify_proxy: bool = False # route notifications through SOCKS5
|
|
|
|
# Client TLS
|
|
client_tls: bool = False # enable TLS for client listener
|
|
client_tls_cert: str = "" # path to PEM cert (auto-generated if empty)
|
|
client_tls_key: str = "" # path to PEM key (or same file as cert)
|
|
|
|
# Background account farming
|
|
farm_enabled: bool = False
|
|
farm_interval: int = 3600 # seconds between attempts per network
|
|
farm_max_accounts: int = 10 # max verified accounts per network
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Config:
|
|
"""Top-level configuration."""
|
|
|
|
bouncer: BouncerConfig
|
|
proxy: ProxyConfig
|
|
networks: dict[str, NetworkConfig]
|
|
|
|
|
|
def load(path: Path) -> Config:
|
|
"""Load and validate configuration from a TOML file."""
|
|
with open(path, "rb") as f:
|
|
raw = tomllib.load(f)
|
|
|
|
bouncer_raw = raw.get("bouncer", {})
|
|
backlog_raw = bouncer_raw.pop("backlog", {})
|
|
|
|
bouncer = BouncerConfig(
|
|
bind=bouncer_raw.get("bind", "127.0.0.1"),
|
|
port=bouncer_raw.get("port", 6667),
|
|
password=bouncer_raw.get("password", "changeme"),
|
|
backlog=BacklogConfig(**backlog_raw),
|
|
captcha_api_key=bouncer_raw.get("captcha_api_key", ""),
|
|
captcha_poll_interval=bouncer_raw.get("captcha_poll_interval", 3),
|
|
captcha_poll_timeout=bouncer_raw.get("captcha_poll_timeout", 120),
|
|
probation_seconds=bouncer_raw.get("probation_seconds", 45),
|
|
backoff_steps=bouncer_raw.get("backoff_steps", [5, 10, 30, 60, 120, 300]),
|
|
nick_timeout=bouncer_raw.get("nick_timeout", 10),
|
|
rejoin_delay=bouncer_raw.get("rejoin_delay", 3),
|
|
http_timeout=bouncer_raw.get("http_timeout", 15),
|
|
email_poll_interval=bouncer_raw.get("email_poll_interval", 15),
|
|
email_max_polls=bouncer_raw.get("email_max_polls", 30),
|
|
email_request_timeout=bouncer_raw.get("email_request_timeout", 20),
|
|
cert_validity_days=bouncer_raw.get("cert_validity_days", 3650),
|
|
ping_interval=bouncer_raw.get("ping_interval", 120),
|
|
ping_timeout=bouncer_raw.get("ping_timeout", 30),
|
|
notify_url=bouncer_raw.get("notify_url", ""),
|
|
notify_on_highlight=bouncer_raw.get("notify_on_highlight", True),
|
|
notify_on_privmsg=bouncer_raw.get("notify_on_privmsg", True),
|
|
notify_cooldown=bouncer_raw.get("notify_cooldown", 60),
|
|
notify_proxy=bouncer_raw.get("notify_proxy", False),
|
|
client_tls=bouncer_raw.get("client_tls", False),
|
|
client_tls_cert=bouncer_raw.get("client_tls_cert", ""),
|
|
client_tls_key=bouncer_raw.get("client_tls_key", ""),
|
|
farm_enabled=bouncer_raw.get("farm_enabled", False),
|
|
farm_interval=bouncer_raw.get("farm_interval", 3600),
|
|
farm_max_accounts=bouncer_raw.get("farm_max_accounts", 10),
|
|
)
|
|
|
|
proxy_raw = raw.get("proxy", {})
|
|
proxy = ProxyConfig(
|
|
host=proxy_raw.get("host", "127.0.0.1"),
|
|
port=proxy_raw.get("port", 1080),
|
|
)
|
|
|
|
networks: dict[str, NetworkConfig] = {}
|
|
for name, net_raw in raw.get("networks", {}).items():
|
|
networks[name] = NetworkConfig(
|
|
name=name,
|
|
host=net_raw["host"],
|
|
port=net_raw.get("port", 6697 if net_raw.get("tls", False) else 6667),
|
|
tls=net_raw.get("tls", False),
|
|
nick=net_raw.get("nick", ""),
|
|
user=net_raw.get("user", ""),
|
|
realname=net_raw.get("realname", ""),
|
|
channels=net_raw.get("channels", []),
|
|
channel_keys=dict(net_raw.get("channel_keys", {})),
|
|
autojoin=net_raw.get("autojoin", True),
|
|
password=net_raw.get("password"),
|
|
proxy_host=net_raw.get("proxy_host"),
|
|
proxy_port=net_raw.get("proxy_port"),
|
|
auth_service=net_raw.get("auth_service", "nickserv"),
|
|
)
|
|
|
|
if not networks:
|
|
raise ValueError("at least one network must be configured")
|
|
|
|
for name in networks:
|
|
if "/" in name:
|
|
raise ValueError(
|
|
f"network name {name!r} must not contain '/' "
|
|
"(reserved for namespace separator)"
|
|
)
|
|
|
|
return Config(bouncer=bouncer, proxy=proxy, networks=networks)
|