""" MeshBay Hub configuration. Priority (highest first): 1. Environment variables (MESHBAY_*) 2. Config file (/etc/meshbay/hub.toml or --config) 3. Built-in defaults Production config file example: /etc/meshbay/hub.toml """ import logging import os from dataclasses import dataclass, field from pathlib import Path try: import tomllib except ImportError: import tomli as tomllib # type: ignore[no-redef] log = logging.getLogger(__name__) DEFAULT_CONFIG_PATHS = [ Path("/etc/meshbay/hub.toml"), Path.home() / ".config" / "meshbay" / "hub.toml", ] @dataclass class DatabaseConfig: url: str = "sqlite+aiosqlite:///:memory:" @dataclass class ServerConfig: host: str = "127.0.0.1" port: int = 8000 workers: int = 1 @dataclass class HubIdentityConfig: id: str = "meshbay.org" private_key_path: Path = field(default_factory=lambda: Path.home() / ".config" / "meshbay" / "hub_private.pem") admin_usernames: list[str] = field(default_factory=list) @dataclass class JWTConfig: # How long an access token stays good. It is not the session — the refresh # token is, and the SPA renews against it well before this runs out, so a # film or a working day never meets this number. How long the session # lasts is an admin setting (`hub_settings.SESSION_*`), not configuration. # # What it does bound is a token that leaks: revoking a member or suspending # an account both take effect at once (the hub reloads the account on every # request, and pushes signed revocations to nodes), but there is no way to # kill one issued token short of that. Four hours keeps the window short # while renewing several times a day — which also means the renewal path is # exercised constantly rather than twice, and cannot rot unnoticed the way # it did when nothing used it at all. access_token_ttl: int = 14400 # 4 hours @dataclass class CaptchaConfig: site_key: str = "" secret_key: str = "" # Hostnames a solved captcha may have come from, checked against the one # `siteverify` reports. Empty means "do not check", which is right while # the reCAPTCHA key does its own origin check — it is then already done. # # Set this when that check is turned off in the reCAPTCHA console, which is # what the desktop client needs: its page is served from `app://meshbay`, # so the hostname Google sees is not the hub's and never can be. See # docs/MESHBAY_DESIGN.md §7.7. allowed_hosts: list[str] = field(default_factory=list) # A solve from a page Google cannot attribute to a domain reports an empty # hostname — `app://meshbay` does, measured live, and so does any other # non-web client. No `allowed_hosts` entry can match that, and a blank # entry is not the answer: the parser drops blanks because in a TOML list # a blank is a typo far more often than an intention. Hence a named flag, # which also states the trade at the place it is made. allow_unattributed_host: bool = False @property def enabled(self) -> bool: return bool(self.site_key and self.secret_key) @property def host_check(self) -> frozenset[str] | None: """The set to hand `verify_captcha`, or None for "do not check".""" return frozenset(self.allowed_hosts) if self.allowed_hosts else None @dataclass class MailConfig: """What the hub will send, and how much of it. Defaults, not the live values: an admin changes these from the panel and the change is stored in `hub_settings`, so what is written here is the instance's starting point and what it falls back to if a row is missing. The two that matter are per **recipient** and per **instance**. A limit counted per account or per IP bounds a caller, and registration is open, so a caller is something an attacker buys more of. """ # Between two messages to one address, across every purpose and account. destination_cooldown_seconds: int = 120 # And how many that address may receive in a day. destination_daily_cap: int = 10 # Everything this instance sends, per hour. hourly_budget: int = 200 # Of that budget, the share kept back for the two purposes a person is # waiting on: a passphrase reset and a group invitation. Without it a flood # of sign-ups spends the hour's allowance and locks out the people who # actually need a message to arrive. hourly_reserved_for_recovery: int = 50 # Between two sign-up codes to one pending account. verification_resend_cooldown: int = 120 # Between two passphrase-reset codes for one account, whoever asks. The # code lives an hour, so this stays far below its lifetime. reset_cooldown: int = 300 # Between two *different* addresses proposed by one account. Re-asking for # a code for the address already pending is exempt — it reaches no new # recipient, and without the exemption a typo locks the account out for the # whole window. email_change_cooldown: int = 172800 # 48 hours # Invitation-link mails one account may have the hub send per day. The only # per-account bound here, because a link reaches addresses with no account, # and the per-recipient cap alone would let one account mail ten strangers # each, without end. invite_link_daily_cap: int = 10 @dataclass class HubConfig: db: DatabaseConfig = field(default_factory=DatabaseConfig) server: ServerConfig = field(default_factory=ServerConfig) identity: HubIdentityConfig = field(default_factory=HubIdentityConfig) jwt: JWTConfig = field(default_factory=JWTConfig) captcha: CaptchaConfig = field(default_factory=CaptchaConfig) mail: MailConfig = field(default_factory=MailConfig) def load_config(path: Path | None = None) -> HubConfig: cfg = HubConfig() # Find and load TOML candidates = [path] if path else DEFAULT_CONFIG_PATHS for p in candidates: if p and p.exists(): raw = tomllib.loads(p.read_text()) if db := raw.get("database", {}): cfg.db.url = db.get("url", cfg.db.url) if srv := raw.get("server", {}): cfg.server.host = srv.get("host", cfg.server.host) cfg.server.port = srv.get("port", cfg.server.port) cfg.server.workers = srv.get("workers", cfg.server.workers) if idn := raw.get("hub", {}): cfg.identity.id = idn.get("id", cfg.identity.id) if kp := idn.get("private_key_path"): cfg.identity.private_key_path = Path(kp).expanduser() if admins := idn.get("admin_usernames"): cfg.identity.admin_usernames = list(admins) if jwt := raw.get("jwt", {}): cfg.jwt.access_token_ttl = jwt.get("access_token_ttl", cfg.jwt.access_token_ttl) # One source for the session's length: a value left here would # otherwise look authoritative and change nothing. if "refresh_token_ttl" in jwt: log.warning("%s: [jwt] refresh_token_ttl is ignored — session " "lifetime is set in the admin panel", p) if ml := raw.get("mail", {}): for name in ( "destination_cooldown_seconds", "destination_daily_cap", "hourly_budget", "hourly_reserved_for_recovery", "verification_resend_cooldown", "reset_cooldown", "email_change_cooldown", "invite_link_daily_cap", ): setattr(cfg.mail, name, ml.get(name, getattr(cfg.mail, name))) if cap := raw.get("captcha", {}): cfg.captcha.site_key = cap.get("site_key", cfg.captcha.site_key) cfg.captcha.secret_key = cap.get("secret_key", cfg.captcha.secret_key) if hosts := cap.get("allowed_hosts"): cfg.captcha.allowed_hosts = [str(h).strip() for h in hosts if str(h).strip()] if "allow_unattributed_host" in cap: cfg.captcha.allow_unattributed_host = bool(cap["allow_unattributed_host"]) break # Env var overrides if url := os.environ.get("MESHBAY_DATABASE_URL"): cfg.db.url = url if host := os.environ.get("MESHBAY_HUB_HOST"): cfg.server.host = host if port := os.environ.get("MESHBAY_HUB_PORT"): cfg.server.port = int(port) if hub_id := os.environ.get("MESHBAY_HUB_ID"): cfg.identity.id = hub_id if kp := os.environ.get("MESHBAY_HUB_KEY"): cfg.identity.private_key_path = Path(kp).expanduser() if admin_users := os.environ.get("MESHBAY_ADMIN_USERS"): cfg.identity.admin_usernames = [u.strip() for u in admin_users.split(",") if u.strip()] if captcha_site := os.environ.get("MESHBAY_CAPTCHA_SITE_KEY"): cfg.captcha.site_key = captcha_site if captcha_secret := os.environ.get("MESHBAY_CAPTCHA_SECRET_KEY"): cfg.captcha.secret_key = captcha_secret if captcha_hosts := os.environ.get("MESHBAY_CAPTCHA_ALLOWED_HOSTS"): cfg.captcha.allowed_hosts = [h.strip() for h in captcha_hosts.split(",") if h.strip()] if unattributed := os.environ.get("MESHBAY_CAPTCHA_ALLOW_UNATTRIBUTED_HOST"): cfg.captcha.allow_unattributed_host = unattributed.lower() in ("1", "true", "yes") return cfg