1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
"""
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 = ""
@property
def enabled(self) -> bool:
return bool(self.site_key and self.secret_key)
@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)
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
return cfg
|