#!/usr/bin/env python3 """Atomically replace only the SMTP section of a monitor configuration.""" from __future__ import annotations import argparse import json import os import shlex import stat import tempfile from pathlib import Path from typing import Any EXPECTED_KEYS = { "EMAIL_FROM_ADDRESS", "EMAIL_FROM_NAME", "SMTP_PASSWORD", "SMTP_PORT", "SMTP_RELAY", "SMTP_SSL", "SMTP_TLS", "SMTP_USERNAME", "SUPPORT_INBOX_ADDRESS", } def parse_values(path: Path) -> dict[str, str]: mode = stat.S_IMODE(path.stat().st_mode) if mode not in {0o400, 0o600}: raise ValueError("SMTP values file must have mode 0400 or 0600") values: dict[str, str] = {} with path.open(encoding="utf-8") as handle: for line_number, raw_line in enumerate(handle, start=1): line = raw_line.rstrip("\r\n") if not line or line.startswith("#"): continue if "=" not in line: raise ValueError(f"Malformed SMTP value on line {line_number}") key, raw_value = line.split("=", 1) if key in values: raise ValueError(f"Duplicate SMTP value: {key}") parsed = shlex.split(raw_value, comments=False, posix=True) if len(parsed) != 1 or not parsed[0]: raise ValueError(f"SMTP value must contain one non-empty token: {key}") values[key] = parsed[0] missing = sorted(EXPECTED_KEYS - values.keys()) extra = sorted(values.keys() - EXPECTED_KEYS) if missing or extra: details: list[str] = [] if missing: details.append("missing " + ", ".join(missing)) if extra: details.append("unexpected " + ", ".join(extra)) raise ValueError("Invalid SMTP values file: " + "; ".join(details)) return values def parse_bool(value: str, name: str) -> bool: normalized = value.lower() if normalized in {"true", "1"}: return True if normalized in {"false", "0"}: return False raise ValueError(f"{name} must be true, false, 1, or 0") def build_smtp(values: dict[str, str]) -> dict[str, Any]: try: port = int(values["SMTP_PORT"]) except ValueError as error: raise ValueError("SMTP_PORT must be an integer") from error if not 1 <= port <= 65535: raise ValueError("SMTP_PORT must be between 1 and 65535") tls = values["SMTP_TLS"].lower() if tls not in {"always", "never"}: raise ValueError("SMTP_TLS must be always or never for the external monitor") implicit_ssl = parse_bool(values["SMTP_SSL"], "SMTP_SSL") if implicit_ssl and tls != "never": raise ValueError("SMTP_TLS must be never when SMTP_SSL enables implicit TLS") return { "from_address": values["EMAIL_FROM_ADDRESS"], "from_name": values["EMAIL_FROM_NAME"], "implicit_ssl": implicit_ssl, "password": values["SMTP_PASSWORD"], "port": port, "recipient": values["SUPPORT_INBOX_ADDRESS"], "relay": values["SMTP_RELAY"], "starttls": tls == "always", "username": values["SMTP_USERNAME"], } def read_config(path: Path) -> dict[str, Any]: with path.open(encoding="utf-8") as handle: config = json.load(handle) if not isinstance(config, dict) or not isinstance(config.get("smtp"), dict): raise ValueError("Monitor configuration must contain an SMTP object") return config def write_atomic(path: Path, config: dict[str, Any]) -> None: descriptor, temporary_name = tempfile.mkstemp( dir=path.parent, prefix=f".{path.name}.smtp." ) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(config, handle, ensure_ascii=False, indent=2, sort_keys=True) handle.write("\n") temporary.chmod(0o600) temporary.replace(path) finally: temporary.unlink(missing_ok=True) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("config", type=Path) parser.add_argument("values", type=Path) args = parser.parse_args() try: config = read_config(args.config) config["smtp"] = build_smtp(parse_values(args.values)) write_atomic(args.config, config) except (OSError, ValueError, json.JSONDecodeError) as error: parser.error(str(error)) print("External monitor SMTP configuration updated without printing credentials.") return 0 if __name__ == "__main__": raise SystemExit(main())