""" 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 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) 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) 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()] return cfg