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
« 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.
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.
10Usage
11-----
13.. code-block:: bash
15 # Dump current config to stdout (default)
16 nexuslims config dump
18 # Dump current config to a file
19 nexuslims config dump --output /path/to/nexuslims_config.json
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
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"""
34import json
35import logging
36from copy import deepcopy
37from datetime import datetime
38from pathlib import Path
40import click
42from nexusLIMS.cli import _format_version
44# Heavy NexusLIMS imports are lazy-loaded inside functions to keep
45# --help / --version fast (same pattern as process_records.py).
47logger = logging.getLogger(__name__)
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"}
58# Substrings that indicate a key contains sensitive data
59_SECRET_SUBSTRINGS = {"TOKEN", "PASSWORD"}
61_REDACTED = "***"
63# ---------------------------------------------------------------------------
64# Core helpers (imported by process_records for the verbose log dump)
65# ---------------------------------------------------------------------------
68def _build_config_dict(settings) -> dict:
69 """
70 Assemble the full nested configuration dictionary from *settings*.
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).
76 Parameters
77 ----------
78 settings : Settings
79 The NexusLIMS settings instance (or proxy).
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()
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 }
98 # Merge email config
99 email = settings.email_config()
100 if email is not None:
101 config["email_config"] = email.model_dump()
103 return config
106def _sanitize_config(config_dict: dict) -> dict:
107 """
108 Return a deep copy of *config_dict* with all secret values replaced.
110 Secrets that are redacted:
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``
117 Parameters
118 ----------
119 config_dict : dict
120 The full config dict as returned by :func:`_build_config_dict`.
122 Returns
123 -------
124 dict
125 A new dict with secrets replaced by ``"***"``.
126 """
127 sanitized = deepcopy(config_dict)
129 # Redact explicitly listed secret keys
130 for key in _SECRET_TOP_LEVEL_KEYS:
131 if key in sanitized:
132 sanitized[key] = _REDACTED
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
139 for hvst in sanitized.get("nemo_harvesters", {}).values():
140 if "token" in hvst:
141 hvst["token"] = _REDACTED
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
150 return sanitized
153# ---------------------------------------------------------------------------
154# Flattening (nested JSON -> env-var key/value pairs)
155# ---------------------------------------------------------------------------
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}
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}
180def _flatten_to_env(config_dict: dict) -> dict[str, str]:
181 """
182 Convert a nested config dict back into flat ``{ENV_VAR: value}`` pairs.
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.
189 Parameters
190 ----------
191 config_dict : dict
192 The nested config dict (as dumped to JSON).
194 Returns
195 -------
196 dict[str, str]
197 Flat mapping of environment variable names to string values.
198 """
199 env_vars: dict[str, str] = {}
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)
213 # --- NEMO harvesters --------------------------------------------------
214 env_vars.update(_flatten_nemo_harvesters(config_dict.get("nemo_harvesters", {})))
216 # --- email config -----------------------------------------------------
217 email = config_dict.get("email_config")
218 if email is not None:
219 env_vars.update(_flatten_email_config(email))
221 return env_vars
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
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
251# ---------------------------------------------------------------------------
252# .env file writer
253# ---------------------------------------------------------------------------
256def _write_env_file(env_vars: dict[str, str], path: Path) -> None:
257 """
258 Write *env_vars* to *path* as a ``.env`` file.
260 Each value is single-quoted so that spaces, commas and other shell
261 metacharacters are preserved verbatim.
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")
274# ---------------------------------------------------------------------------
275# Click CLI
276# ---------------------------------------------------------------------------
279@click.group()
280@click.version_option(version=None, message=_format_version("nexuslims config"))
281def main() -> None:
282 """Manage NexusLIMS configuration files."""
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.
296 By default, prints the configuration to stdout. Use --output to write to a file.
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
304 with handle_config_error():
305 from nexusLIMS.config import settings # noqa: PLC0415
307 # Accessing settings attributes triggers validation; the context
308 # manager turns any ValidationError into a friendly message.
309 config_dict = _build_config_dict(settings)
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)
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)
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.
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
346 ConfiguratorApp(env_path=Path(env_path)).run()
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.
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.
387 This command does NOT require a fully-configured NexusLIMS environment —
388 only the three NX_LABARCHIVES_* fields listed above need to be set.
390 The UID is printed along with a ready-to-paste .env line:
392 \b
393 NX_LABARCHIVES_USER_ID=<uid>
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
400 from nexusLIMS.config import read_labarchives_env # noqa: PLC0415
401 from nexusLIMS.utils.labarchives import ( # noqa: PLC0415
402 LabArchivesClient,
403 LabArchivesError,
404 )
406 if verbose:
407 logging.basicConfig(level=logging.DEBUG)
408 logging.getLogger("nexusLIMS.utils.labarchives").setLevel(logging.DEBUG)
410 la = read_labarchives_env(env_path)
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)
431 click.echo(f"Connecting to {la['NX_LABARCHIVES_URL']} ...", err=True)
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 )
441 try:
442 info = client.get_user_info(email, la_password)
443 except LabArchivesError as exc:
444 raise click.ClickException(str(exc)) from exc
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)
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}")
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.
478 INPUT is the path to the JSON file produced by ``dump``.
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)
486 config_dict = json.loads(input_path.read_text())
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)
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}")
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}")