Coverage for src/local_deep_research/web/server_config.py: 100%
72 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +0000
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +0000
1"""Server configuration management for web app startup.
3This module handles server configuration that needs to be available before
4Flask app context is established. All settings are read from environment
5variables (LDR_* naming convention) with sensible defaults.
7During the deprecation period, legacy ``server_config.json`` files are read
8as a read-only fallback (env var > legacy file > default). No data is
9written back to the JSON file.
10"""
12import json
13import os
14from pathlib import Path
15from typing import Any, Dict, Optional
17from loguru import logger
19from ..config.paths import get_data_dir
20from ..settings.manager import get_typed_setting_value
22# Maps legacy JSON keys → (setting_key, env_var_name, is_security_critical)
23_LEGACY_KEY_MAP: Dict[str, tuple] = {
24 "host": ("web.host", "LDR_WEB_HOST", False),
25 "port": ("web.port", "LDR_WEB_PORT", False),
26 "debug": ("app.debug", "LDR_APP_DEBUG", False),
27 "use_https": ("web.use_https", "LDR_WEB_USE_HTTPS", False),
28 "allow_registrations": (
29 "app.allow_registrations",
30 "LDR_APP_ALLOW_REGISTRATIONS",
31 True,
32 ),
33 "rate_limit_default": (
34 "security.rate_limit_default",
35 "LDR_SECURITY_RATE_LIMIT_DEFAULT",
36 False,
37 ),
38 "rate_limit_login": (
39 "security.rate_limit_login",
40 "LDR_SECURITY_RATE_LIMIT_LOGIN",
41 False,
42 ),
43 "rate_limit_registration": (
44 "security.rate_limit_registration",
45 "LDR_SECURITY_RATE_LIMIT_REGISTRATION",
46 False,
47 ),
48 "rate_limit_settings": (
49 "security.rate_limit_settings",
50 "LDR_SECURITY_RATE_LIMIT_SETTINGS",
51 False,
52 ),
53 "rate_limit_upload_user": (
54 "security.rate_limit_upload_user",
55 "LDR_SECURITY_RATE_LIMIT_UPLOAD_USER",
56 False,
57 ),
58 "rate_limit_upload_ip": (
59 "security.rate_limit_upload_ip",
60 "LDR_SECURITY_RATE_LIMIT_UPLOAD_IP",
61 False,
62 ),
63}
66_DEFAULTS: Dict[str, Any] = {
67 "host": "0.0.0.0",
68 "port": 5000,
69 "debug": False,
70 "use_https": True,
71 "allow_registrations": True,
72 "rate_limit_default": "5000 per hour;50000 per day",
73 "rate_limit_login": "5 per 15 minutes",
74 "rate_limit_registration": "3 per hour",
75 "rate_limit_settings": "30 per minute",
76 "rate_limit_upload_user": "60 per minute;1000 per hour",
77 "rate_limit_upload_ip": "60 per minute;1000 per hour",
78}
81def get_server_config_path() -> Path:
82 """Return the path to the legacy server_config.json file."""
83 return Path(get_data_dir()) / "server_config.json"
86def has_legacy_customizations(config_path: Optional[Path] = None) -> bool:
87 """Return True if server_config.json exists with non-default values.
89 Parameters
90 ----------
91 config_path : Path, optional
92 Path to the legacy JSON file. Defaults to get_server_config_path().
93 """
94 path = config_path or get_server_config_path()
95 if not path.exists():
96 return False
97 try:
98 data = json.loads(path.read_text(encoding="utf-8-sig"))
99 except (json.JSONDecodeError, OSError, UnicodeDecodeError):
100 return False
101 if not isinstance(data, dict):
102 return False
103 for json_key in _LEGACY_KEY_MAP:
104 if json_key in data and data[json_key] != _DEFAULTS.get(json_key):
105 return True
106 unrecognized = set(data.keys()) - set(_LEGACY_KEY_MAP.keys())
107 return bool(unrecognized)
110def _load_legacy_config() -> Dict[str, Any]:
111 """Read legacy server_config.json as a read-only migration fallback.
113 Returns a dict of ``{json_key: value}`` for recognized keys found in the
114 file. Returns ``{}`` when the file does not exist or is malformed.
116 Logs deprecation warnings so users know to migrate to env vars.
117 """
118 config_path = get_server_config_path()
119 if not config_path.exists():
120 return {}
122 try:
123 data = json.loads(config_path.read_text(encoding="utf-8-sig"))
124 except (json.JSONDecodeError, OSError, UnicodeDecodeError):
125 logger.warning(
126 f"Could not read legacy server_config.json at {config_path}: "
127 f"Ignoring file."
128 )
129 return {}
131 if not isinstance(data, dict):
132 logger.warning(
133 f"Legacy server_config.json at {config_path} does not contain a "
134 f"JSON object. Ignoring file."
135 )
136 return {}
138 saved: Dict[str, Any] = {}
139 for json_key in _LEGACY_KEY_MAP:
140 if json_key in data:
141 saved[json_key] = data[json_key]
143 # Warn about unrecognized keys (likely typos)
144 unrecognized = set(data.keys()) - set(_LEGACY_KEY_MAP.keys())
145 if unrecognized:
146 logger.warning(
147 f"Legacy server_config.json contains unrecognized keys: "
148 f"{sorted(unrecognized)}. These will be ignored. "
149 f"Recognized keys: {sorted(_LEGACY_KEY_MAP.keys())}."
150 )
152 if saved:
153 logger.info(
154 f"server_config.json detected at {config_path}. "
155 f"Environment variables are the preferred configuration method."
156 )
158 return saved
161def load_server_config() -> Dict[str, Any]:
162 """Load server configuration from environment variables.
164 During the deprecation period, values from a legacy ``server_config.json``
165 are used as fallbacks when the corresponding env var is not set.
167 Priority: environment variable > legacy JSON file > built-in default.
169 Returns:
170 dict: Server configuration with keys: host, port, debug, use_https,
171 allow_registrations, rate_limit_default, rate_limit_login,
172 rate_limit_registration, rate_limit_settings
173 """
174 saved = _load_legacy_config()
176 config = {
177 "host": get_typed_setting_value(
178 "web.host", saved.get("host"), "text", default=_DEFAULTS["host"]
179 ),
180 "port": get_typed_setting_value(
181 "web.port", saved.get("port"), "number", default=_DEFAULTS["port"]
182 ),
183 "debug": get_typed_setting_value(
184 "app.debug",
185 saved.get("debug"),
186 "checkbox",
187 default=_DEFAULTS["debug"],
188 ),
189 "use_https": get_typed_setting_value(
190 "web.use_https",
191 saved.get("use_https"),
192 "checkbox",
193 default=_DEFAULTS["use_https"],
194 ),
195 "allow_registrations": get_typed_setting_value(
196 "app.allow_registrations",
197 saved.get("allow_registrations"),
198 "checkbox",
199 default=_DEFAULTS["allow_registrations"],
200 ),
201 "rate_limit_default": get_typed_setting_value(
202 "security.rate_limit_default",
203 saved.get("rate_limit_default"),
204 "text",
205 default=_DEFAULTS["rate_limit_default"],
206 ),
207 "rate_limit_login": get_typed_setting_value(
208 "security.rate_limit_login",
209 saved.get("rate_limit_login"),
210 "text",
211 default=_DEFAULTS["rate_limit_login"],
212 ),
213 "rate_limit_registration": get_typed_setting_value(
214 "security.rate_limit_registration",
215 saved.get("rate_limit_registration"),
216 "text",
217 default=_DEFAULTS["rate_limit_registration"],
218 ),
219 "rate_limit_settings": get_typed_setting_value(
220 "security.rate_limit_settings",
221 saved.get("rate_limit_settings"),
222 "text",
223 default=_DEFAULTS["rate_limit_settings"],
224 ),
225 "rate_limit_upload_user": get_typed_setting_value(
226 "security.rate_limit_upload_user",
227 saved.get("rate_limit_upload_user"),
228 "text",
229 default=_DEFAULTS["rate_limit_upload_user"],
230 ),
231 "rate_limit_upload_ip": get_typed_setting_value(
232 "security.rate_limit_upload_ip",
233 saved.get("rate_limit_upload_ip"),
234 "text",
235 default=_DEFAULTS["rate_limit_upload_ip"],
236 ),
237 }
239 # Log per-key messages for legacy values that differ from defaults
240 if saved:
241 for json_key in saved:
242 setting_key, env_var, is_critical = _LEGACY_KEY_MAP[json_key]
243 typed_value = config[json_key]
244 default_value = _DEFAULTS[json_key]
245 if typed_value == default_value:
246 continue # matches default — no noise
247 if is_critical:
248 logger.warning(
249 f"SECURITY: server_config.json sets '{json_key}' to a non-default value. "
250 f"Consider migrating to environment variable {env_var}."
251 )
252 else:
253 logger.info(
254 f"server_config.json sets '{json_key}' to a non-default value. "
255 f"Environment variable {env_var} is the preferred configuration method."
256 )
258 # Security: if allow_registrations is set to an unrecognized string value
259 # (e.g. "disabled", "nein", typos like "flase"), default to False
260 # (registrations disabled) rather than True. This "fail closed" approach
261 # prevents accidental open registration when an admin clearly intended to
262 # restrict it but used a non-standard boolean string.
263 _RECOGNIZED_BOOL_VALUES = {
264 "true",
265 "false",
266 "1",
267 "0",
268 "yes",
269 "no",
270 "on",
271 "off",
272 }
274 # Guard for env var path
275 raw_reg_env = os.getenv("LDR_APP_ALLOW_REGISTRATIONS")
276 if raw_reg_env is not None:
277 normalized = raw_reg_env.lower().strip()
278 if normalized not in _RECOGNIZED_BOOL_VALUES:
279 logger.warning(
280 f"LDR_APP_ALLOW_REGISTRATIONS='{raw_reg_env}' is not a "
281 f"recognized boolean value. Defaulting to FALSE "
282 f"(registrations disabled) for security. Use "
283 f"'true'/'false', '1'/'0', 'yes'/'no', or 'on'/'off'."
284 )
285 config["allow_registrations"] = False
287 # Guard for legacy JSON path: if legacy JSON had a string value for
288 # allow_registrations (e.g. "disabled") and no env var overrides it,
289 # parse_boolean would treat any non-empty string as True (HTML checkbox
290 # semantics). Fail closed to prevent accidental open registration.
291 elif (
292 "allow_registrations" in saved
293 and isinstance(saved["allow_registrations"], str)
294 and saved["allow_registrations"].lower().strip()
295 not in _RECOGNIZED_BOOL_VALUES
296 ):
297 logger.warning(
298 f"Legacy server_config.json has allow_registrations="
299 f"'{saved['allow_registrations']}' which is not a recognized "
300 f"boolean value. Defaulting to FALSE (registrations disabled) "
301 f"for security."
302 )
303 config["allow_registrations"] = False
305 return config