""" 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 os from dataclasses import dataclass, field from pathlib import Path try: import tomllib except ImportError: import tomli as tomllib # type: ignore[no-redef] 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 below is, and the SPA renews against it well before this runs out, # so a film or a working day never meets this number. # # 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 refresh_token_ttl: int = 86400 * 30 # 30 days @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/captcha.md §6. 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 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) 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) cfg.jwt.refresh_token_ttl = jwt.get("refresh_token_ttl", cfg.jwt.refresh_token_ttl) 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