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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
"""
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
|