Coverage for nexusLIMS/cli/config.py: 100%

151 statements  

« prev     ^ index     » next       coverage.py v7.11.3, created at 2026-03-24 05:23 +0000

1""" 

2CLI commands for dumping and loading NexusLIMS configuration. 

3 

4Provides ``nexuslims config dump`` and ``nexuslims config load`` for exporting 

5the current effective configuration to JSON and importing a previously 

6dumped configuration back into a ``.env`` file. The JSON format uses a nested 

7structure for NEMO harvesters and email config so that the file is both 

8human-readable and unambiguous. 

9 

10Usage 

11----- 

12 

13.. code-block:: bash 

14 

15 # Dump current config to stdout (default) 

16 nexuslims config dump 

17 

18 # Dump current config to a file 

19 nexuslims config dump --output /path/to/nexuslims_config.json 

20 

21 # Load a previously dumped config into .env 

22 nexuslims config load nexuslims_config.json 

23 nexuslims config load nexuslims_config.json --env-path /path/to/.env 

24 nexuslims config load nexuslims_config.json --force 

25 

26Security 

27-------- 

28The ``dump`` output is **not** sanitised — it contains live API tokens and 

29passwords exactly as they appear in the running configuration. A warning is 

30printed to stderr every time ``dump`` is invoked. The ``load`` command will 

31back up any pre-existing ``.env`` file before overwriting. 

32""" 

33 

34import json 

35import logging 

36from copy import deepcopy 

37from datetime import datetime 

38from pathlib import Path 

39 

40import click 

41 

42from nexusLIMS.cli import _format_version 

43 

44# Heavy NexusLIMS imports are lazy-loaded inside functions to keep 

45# --help / --version fast (same pattern as process_records.py). 

46 

47logger = logging.getLogger(__name__) 

48 

49 

50# --------------------------------------------------------------------------- 

51# Secret-field definitions 

52# --------------------------------------------------------------------------- 

53# Top-level keys in the Settings model_dump() that must be redacted for log 

54# display. The nested paths for NEMO tokens and email password are handled 

55# explicitly in _sanitize_config(). 

56_SECRET_TOP_LEVEL_KEYS = {"NX_CDCS_TOKEN", "NX_CERT_BUNDLE", "NX_ELABFTW_API_KEY"} 

57 

58# Substrings that indicate a key contains sensitive data 

59_SECRET_SUBSTRINGS = {"TOKEN", "PASSWORD"} 

60 

61_REDACTED = "***" 

62 

63# --------------------------------------------------------------------------- 

64# Core helpers (imported by process_records for the verbose log dump) 

65# --------------------------------------------------------------------------- 

66 

67 

68def _build_config_dict(settings) -> dict: 

69 """ 

70 Assemble the full nested configuration dictionary from *settings*. 

71 

72 The returned dict is the canonical representation used by both ``dump`` 

73 (written to disk unsanitised) and the verbose log dump in 

74 ``process_records`` (sanitised before logging). 

75 

76 Parameters 

77 ---------- 

78 settings : Settings 

79 The NexusLIMS settings instance (or proxy). 

80 

81 Returns 

82 ------- 

83 dict 

84 Full configuration. Top-level keys match the ``Settings`` field names. 

85 ``"nemo_harvesters"`` is a dict keyed by harvester number (as strings). 

86 ``"email_config"`` is the ``EmailConfig`` dict. Either nested key is 

87 omitted when the corresponding config is empty / ``None``. 

88 """ 

89 config = settings.model_dump() 

90 

91 # Merge dynamically-assembled NEMO harvesters 

92 harvesters = settings.nemo_harvesters() 

93 if harvesters: 

94 config["nemo_harvesters"] = { 

95 str(num): hvst.model_dump() for num, hvst in harvesters.items() 

96 } 

97 

98 # Merge email config 

99 email = settings.email_config() 

100 if email is not None: 

101 config["email_config"] = email.model_dump() 

102 

103 return config 

104 

105 

106def _sanitize_config(config_dict: dict) -> dict: 

107 """ 

108 Return a deep copy of *config_dict* with all secret values replaced. 

109 

110 Secrets that are redacted: 

111 

112 * Top-level: ``NX_CDCS_TOKEN``, ``NX_CERT_BUNDLE``, ``NX_ELABFTW_API_KEY`` 

113 * Any top-level key containing ``TOKEN`` or ``PASSWORD`` (case-insensitive) 

114 * ``nemo_harvesters.<N>.token`` for every harvester 

115 * ``email_config.smtp_password`` 

116 

117 Parameters 

118 ---------- 

119 config_dict : dict 

120 The full config dict as returned by :func:`_build_config_dict`. 

121 

122 Returns 

123 ------- 

124 dict 

125 A new dict with secrets replaced by ``"***"``. 

126 """ 

127 sanitized = deepcopy(config_dict) 

128 

129 # Redact explicitly listed secret keys 

130 for key in _SECRET_TOP_LEVEL_KEYS: 

131 if key in sanitized: 

132 sanitized[key] = _REDACTED 

133 

134 # Redact any key containing sensitive substrings 

135 for key in list(sanitized.keys()): 

136 if any(substring in key.upper() for substring in _SECRET_SUBSTRINGS): 

137 sanitized[key] = _REDACTED 

138 

139 for hvst in sanitized.get("nemo_harvesters", {}).values(): 

140 if "token" in hvst: 

141 hvst["token"] = _REDACTED 

142 

143 if ( 

144 "email_config" in sanitized 

145 and sanitized["email_config"] is not None 

146 and "smtp_password" in sanitized["email_config"] 

147 ): 

148 sanitized["email_config"]["smtp_password"] = _REDACTED 

149 

150 return sanitized 

151 

152 

153# --------------------------------------------------------------------------- 

154# Flattening (nested JSON -> env-var key/value pairs) 

155# --------------------------------------------------------------------------- 

156 

157# Mapping from email_config dict keys to the env var names that 

158# settings.email_config() reads. 

159_EMAIL_KEY_TO_ENV = { 

160 "smtp_host": "NX_EMAIL_SMTP_HOST", 

161 "smtp_port": "NX_EMAIL_SMTP_PORT", 

162 "smtp_username": "NX_EMAIL_SMTP_USERNAME", 

163 "smtp_password": "NX_EMAIL_SMTP_PASSWORD", 

164 "use_tls": "NX_EMAIL_USE_TLS", 

165 "sender": "NX_EMAIL_SENDER", 

166 "recipients": "NX_EMAIL_RECIPIENTS", 

167} 

168 

169# Mapping from NemoHarvesterConfig dict keys to the env-var suffix 

170# (the harvester number is appended after the underscore). 

171_NEMO_KEY_TO_SUFFIX = { 

172 "address": "NX_NEMO_ADDRESS_", 

173 "token": "NX_NEMO_TOKEN_", 

174 "strftime_fmt": "NX_NEMO_STRFTIME_FMT_", 

175 "strptime_fmt": "NX_NEMO_STRPTIME_FMT_", 

176 "tz": "NX_NEMO_TZ_", 

177} 

178 

179 

180def _flatten_to_env(config_dict: dict) -> dict[str, str]: 

181 """ 

182 Convert a nested config dict back into flat ``{ENV_VAR: value}`` pairs. 

183 

184 This is the inverse of the nesting performed by :func:`_build_config_dict`. 

185 ``None`` values are omitted (they would just be commented-out lines in a 

186 ``.env`` file). List values (``NX_IGNORE_PATTERNS``) are JSON-encoded. 

187 Boolean values are lowercased. ``recipients`` is joined with commas. 

188 

189 Parameters 

190 ---------- 

191 config_dict : dict 

192 The nested config dict (as dumped to JSON). 

193 

194 Returns 

195 ------- 

196 dict[str, str] 

197 Flat mapping of environment variable names to string values. 

198 """ 

199 env_vars: dict[str, str] = {} 

200 

201 # --- top-level scalars / lists ---------------------------------------- 

202 nested_keys = {"nemo_harvesters", "email_config"} 

203 for key, value in config_dict.items(): 

204 if key in nested_keys or value is None: 

205 continue 

206 if isinstance(value, list): 

207 env_vars[key] = json.dumps(value) 

208 elif isinstance(value, bool): 

209 env_vars[key] = str(value).lower() 

210 else: 

211 env_vars[key] = str(value) 

212 

213 # --- NEMO harvesters -------------------------------------------------- 

214 env_vars.update(_flatten_nemo_harvesters(config_dict.get("nemo_harvesters", {}))) 

215 

216 # --- email config ----------------------------------------------------- 

217 email = config_dict.get("email_config") 

218 if email is not None: 

219 env_vars.update(_flatten_email_config(email)) 

220 

221 return env_vars 

222 

223 

224def _flatten_nemo_harvesters(harvesters: dict) -> dict[str, str]: 

225 """Expand the nested ``nemo_harvesters`` dict into ``NX_NEMO_*_N`` pairs.""" 

226 env_vars: dict[str, str] = {} 

227 for num_str, hvst in harvesters.items(): 

228 for hvst_key, env_prefix in _NEMO_KEY_TO_SUFFIX.items(): 

229 value = hvst.get(hvst_key) 

230 if value is not None: 

231 env_vars[f"{env_prefix}{num_str}"] = str(value) 

232 return env_vars 

233 

234 

235def _flatten_email_config(email: dict) -> dict[str, str]: 

236 """Expand the nested ``email_config`` dict into ``NX_EMAIL_*`` pairs.""" 

237 env_vars: dict[str, str] = {} 

238 for email_key, env_name in _EMAIL_KEY_TO_ENV.items(): 

239 value = email.get(email_key) 

240 if value is None: 

241 continue 

242 if email_key == "recipients": 

243 env_vars[env_name] = ",".join(str(v) for v in value) 

244 elif isinstance(value, bool): 

245 env_vars[env_name] = str(value).lower() 

246 else: 

247 env_vars[env_name] = str(value) 

248 return env_vars 

249 

250 

251# --------------------------------------------------------------------------- 

252# .env file writer 

253# --------------------------------------------------------------------------- 

254 

255 

256def _write_env_file(env_vars: dict[str, str], path: Path) -> None: 

257 """ 

258 Write *env_vars* to *path* as a ``.env`` file. 

259 

260 Each value is single-quoted so that spaces, commas and other shell 

261 metacharacters are preserved verbatim. 

262 

263 Parameters 

264 ---------- 

265 env_vars : dict[str, str] 

266 Flat environment variable mapping. 

267 path : Path 

268 Destination file path. 

269 """ 

270 lines = [f"{key}='{value}'" for key, value in sorted(env_vars.items())] 

271 path.write_text("\n".join(lines) + "\n") 

272 

273 

274# --------------------------------------------------------------------------- 

275# Click CLI 

276# --------------------------------------------------------------------------- 

277 

278 

279@click.group() 

280@click.version_option(version=None, message=_format_version("nexuslims config")) 

281def main() -> None: 

282 """Manage NexusLIMS configuration files.""" 

283 

284 

285@main.command() 

286@click.option( 

287 "--output", 

288 "-o", 

289 type=click.Path(dir_okay=False, writable=True), 

290 default=None, 

291 help="Path to write the JSON config file. If omitted, prints to stdout.", 

292) 

293def dump(output: str | None) -> None: 

294 """Dump the current effective configuration to JSON. 

295 

296 By default, prints the configuration to stdout. Use --output to write to a file. 

297 

298 The output contains live API tokens and passwords — handle it like a 

299 secret. Use ``load`` to import a previously dumped file back into a 

300 ``.env``. 

301 """ 

302 from nexusLIMS.cli import handle_config_error # noqa: PLC0415 

303 

304 with handle_config_error(): 

305 from nexusLIMS.config import settings # noqa: PLC0415 

306 

307 # Accessing settings attributes triggers validation; the context 

308 # manager turns any ValidationError into a friendly message. 

309 config_dict = _build_config_dict(settings) 

310 

311 click.echo( 

312 "WARNING: The output will contain live credentials " 

313 "(API tokens, passwords, certificates). " 

314 "Handle it with the same care as your .env file.", 

315 err=True, 

316 ) 

317 json_output = json.dumps(config_dict, indent=2, default=str) 

318 

319 if output is None: 

320 # Print to stdout 

321 click.echo(json_output) 

322 else: 

323 # Write to file 

324 output_path = Path(output) 

325 output_path.write_text(json_output + "\n") 

326 click.echo(f"Configuration dumped to {output_path}", err=True) 

327 

328 

329@main.command() 

330@click.option( 

331 "--env-path", 

332 type=click.Path(dir_okay=False, writable=True), 

333 default=".env", 

334 show_default=True, 

335 help="Path to the .env file to edit (created if absent).", 

336) 

337def edit(env_path: str) -> None: 

338 """Interactively edit the NexusLIMS configuration in a terminal UI. 

339 

340 Opens a tabbed form pre-populated from the existing .env file (if any). 

341 Press Ctrl+S to save and write the .env file, or Escape to cancel without 

342 saving. 

343 """ 

344 from nexusLIMS.tui.apps.config.app import ConfiguratorApp # noqa: PLC0415 

345 

346 ConfiguratorApp(env_path=Path(env_path)).run() 

347 

348 

349@main.command("labarchives-get-uid") 

350@click.option( 

351 "--email", 

352 required=True, 

353 help="Your LabArchives login email address.", 

354) 

355@click.option( 

356 "--la-password", 

357 required=True, 

358 help=( 

359 "Your LabArchives account password or app token (visible in LabArchives " 

360 "under Settings > API). This is NOT the NX_LABARCHIVES_ACCESS_PASSWORD " 

361 "(HMAC signing secret) — it is your personal LabArchives login credential." 

362 ), 

363) 

364@click.option( 

365 "--env-path", 

366 type=click.Path(dir_okay=False), 

367 default=".env", 

368 show_default=True, 

369 help="Path to the .env file to read LabArchives credentials from.", 

370) 

371@click.option( 

372 "--verbose", 

373 "-v", 

374 is_flag=True, 

375 default=False, 

376 help="Enable debug logging to see full HTTP request/response details.", 

377) 

378def labarchives_get_uid( 

379 email: str, la_password: str, env_path: str, *, verbose: bool 

380) -> None: 

381 """Look up your LabArchives user ID (UID) using your login credentials. 

382 

383 Reads NX_LABARCHIVES_ACCESS_KEY_ID, NX_LABARCHIVES_ACCESS_PASSWORD, and 

384 NX_LABARCHIVES_URL from your .env file (or environment), then calls the 

385 LabArchives API to retrieve the UID for the provided email address. 

386 

387 This command does NOT require a fully-configured NexusLIMS environment — 

388 only the three NX_LABARCHIVES_* fields listed above need to be set. 

389 

390 The UID is printed along with a ready-to-paste .env line: 

391 

392 \b 

393 NX_LABARCHIVES_USER_ID=<uid> 

394 

395 Note: --la-password is your personal LabArchives account password or 

396 app token, NOT the NX_LABARCHIVES_ACCESS_PASSWORD signing secret. 

397 """ # noqa: D301 

398 import logging # noqa: PLC0415 

399 

400 from nexusLIMS.config import read_labarchives_env # noqa: PLC0415 

401 from nexusLIMS.utils.labarchives import ( # noqa: PLC0415 

402 LabArchivesClient, 

403 LabArchivesError, 

404 ) 

405 

406 if verbose: 

407 logging.basicConfig(level=logging.DEBUG) 

408 logging.getLogger("nexusLIMS.utils.labarchives").setLevel(logging.DEBUG) 

409 

410 la = read_labarchives_env(env_path) 

411 

412 if not la["NX_LABARCHIVES_ACCESS_KEY_ID"]: 

413 msg = ( 

414 "NX_LABARCHIVES_ACCESS_KEY_ID is not configured. " 

415 "Run 'nexuslims config edit' to set it up." 

416 ) 

417 raise click.ClickException(msg) 

418 if not la["NX_LABARCHIVES_ACCESS_PASSWORD"]: 

419 msg = ( 

420 "NX_LABARCHIVES_ACCESS_PASSWORD is not configured. " 

421 "Run 'nexuslims config edit' to set it up." 

422 ) 

423 raise click.ClickException(msg) 

424 if not la["NX_LABARCHIVES_URL"]: 

425 msg = ( 

426 "NX_LABARCHIVES_URL is not configured. " 

427 "Run 'nexuslims config edit' to set it up." 

428 ) 

429 raise click.ClickException(msg) 

430 

431 click.echo(f"Connecting to {la['NX_LABARCHIVES_URL']} ...", err=True) 

432 

433 # Use a placeholder uid since get_user_info does not require one 

434 client = LabArchivesClient( 

435 base_url=la["NX_LABARCHIVES_URL"], 

436 akid=la["NX_LABARCHIVES_ACCESS_KEY_ID"], 

437 password=la["NX_LABARCHIVES_ACCESS_PASSWORD"], 

438 uid="", 

439 ) 

440 

441 try: 

442 info = client.get_user_info(email, la_password) 

443 except LabArchivesError as exc: 

444 raise click.ClickException(str(exc)) from exc 

445 

446 uid = info.get("uid") 

447 if not uid: 

448 msg = ( 

449 "LabArchives returned no UID for the provided credentials. " 

450 "Check your email and password." 

451 ) 

452 raise click.ClickException(msg) 

453 

454 click.echo(f"UID: {uid}") 

455 click.echo() 

456 click.echo("Add this line to your .env file:") 

457 click.echo(f" NX_LABARCHIVES_USER_ID={uid}") 

458 

459 

460@main.command() 

461@click.argument("input", type=click.Path(exists=True, dir_okay=False, readable=True)) 

462@click.option( 

463 "--env-path", 

464 type=click.Path(dir_okay=False, writable=True), 

465 default=".env", 

466 show_default=True, 

467 help="Path to write the .env file.", 

468) 

469@click.option( 

470 "--force", 

471 is_flag=True, 

472 default=False, 

473 help="Skip confirmation prompt when overwriting an existing .env file.", 

474) 

475def load(input: str, env_path: str, *, force: bool) -> None: # noqa: A002 

476 """Load a previously dumped JSON config into a .env file. 

477 

478 INPUT is the path to the JSON file produced by ``dump``. 

479 

480 If a .env file already exists at ENV_PATH a timestamped backup is created 

481 before it is overwritten. 

482 """ 

483 input_path = Path(input) 

484 env_file = Path(env_path) 

485 

486 config_dict = json.loads(input_path.read_text()) 

487 

488 # --- back up existing .env if present ---------------------------------- 

489 if env_file.exists(): 

490 click.echo( 

491 f"WARNING: A .env file already exists at {env_file}. " 

492 "It will be overwritten.", 

493 err=True, 

494 ) 

495 if not force: 

496 click.confirm("Create a backup and proceed?", abort=True) 

497 

498 timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") 

499 backup_path = env_file.with_name(f"{env_file.name}.bak.{timestamp}") 

500 env_file.rename(backup_path) 

501 click.echo(f"Existing .env backed up to {backup_path}") 

502 

503 # --- flatten and write ------------------------------------------------- 

504 env_vars = _flatten_to_env(config_dict) 

505 _write_env_file(env_vars, env_file) 

506 click.echo(f"Configuration loaded into {env_file}")