"""Configuration store: load, validate, atomically write, and hot-reload config.toml. Design notes (see AGENTS.md): * **tomlkit, not tomllib/tomli-w.** The admin panel writes this file back, and the file is heavily commented. tomlkit round-trips comments/ordering; a plain dict dump would destroy them on the first save. * **Immutable snapshots.** Config is never mutated in place. `apply()` builds a new frozen `Snapshot` and rebinds one attribute — atomic under the GIL, so readers need no lock. Views take one snapshot per request; the worker takes one per job. A job therefore runs under a single consistent config and an admin edit affects the *next* job, which removes every mid-job race without locking. * **Write-only secrets.** `redacted()` never emits a secret value; `apply()` treats a blank/absent secret as "leave unchanged". * **PPSQ_CONFIG** selects an alternate file. Tests rely on this to never touch the real config.toml. """ from __future__ import annotations import copy import logging import os import shutil import tempfile import threading from dataclasses import dataclass from pathlib import Path from types import MappingProxyType from typing import Any, Mapping from urllib.parse import urlparse import tomlkit from pps_client import PPSClient log = logging.getLogger("pps.config") # Dotted keys the admin panel may write. An explicit ALLOWLIST — never a denylist. ADMIN_EDITABLE: frozenset[str] = frozenset( { "pps.base_url", "pps.username", "pps.password", "pps.verify_tls", "pps.timeout", "pps.client_cert", "pps.client_key", "quarantine.default_folder", "quarantine.folders", "quarantine.deleted_folder", "quarantine.default_limit", "quarantine.list_query", "quarantine.default_days_back", "quarantine.chunk_size", "quarantine.default_sort_field", "quarantine.default_sort_dir", "quarantine.report_release.steps", "quarantine.report_release.move_target", "quarantine.report_release.step_delay_seconds", "auth.denied_message", "auth.users", "app.log_level", } ) # Written to disk but only picked up on restart; reported back so the UI can say so. RESTART_KEYS: frozenset[str] = frozenset( { "app.secret_key", "app.listen", "app.port", "app.db_path", "app.worker_log", "app.app_log", "app.cookie_secure", "auth.mode", } ) # Never leave the process. `redacted()` emits a `_set` boolean instead. SECRET_KEYS: frozenset[str] = frozenset( {"pps.password", "okta.client_secret", "auth.static.password", "app.secret_key"} ) # Connection identity — a change here means the PPSClient must be rebuilt. _PPS_FINGERPRINT = ( "base_url", "username", "password", "verify_tls", "timeout", "client_cert", "client_key", ) _SORT_FIELDS = frozenset({"subject", "date", "from", "rcpt"}) _SORT_DIRS = frozenset({"asc", "desc"}) _ROLES = frozenset({"admin", "user"}) class ConfigError(Exception): """Validation failure. Carries per-field messages for a 400 response.""" def __init__(self, errors: list[dict[str, str]]): self.errors = errors super().__init__("; ".join(f"{e['field']}: {e['message']}" for e in errors)) @dataclass(frozen=True) class Snapshot: version: int pps: Mapping[str, Any] quarantine: Mapping[str, Any] app: Mapping[str, Any] auth: Mapping[str, Any] okta: Mapping[str, Any] path: Path @dataclass(frozen=True) class ApplyResult: version: int applied: list[str] restart_required: list[str] def _freeze(d: Any) -> Any: """Deep-copy plain data and wrap mappings read-only, so a snapshot can't be mutated.""" if isinstance(d, Mapping): return MappingProxyType({k: _freeze(v) for k, v in d.items()}) if isinstance(d, (list, tuple)): return tuple(_freeze(v) for v in d) return d def _plain(doc: Any) -> Any: """tomlkit containers -> plain python (dict/list/scalars).""" if isinstance(doc, Mapping): return {k: _plain(v) for k, v in doc.items()} if isinstance(doc, (list, tuple)): return [_plain(v) for v in doc] return doc def _dig(d: Mapping, dotted: str, default=None): cur: Any = d for part in dotted.split("."): if not isinstance(cur, Mapping) or part not in cur: return default cur = cur[part] return cur def _flatten(d: Mapping, prefix: str = "") -> dict[str, Any]: """Flatten nested dicts to dotted keys. Lists are leaves (e.g. quarantine.folders).""" out: dict[str, Any] = {} for k, v in d.items(): key = f"{prefix}{k}" if isinstance(v, Mapping): out.update(_flatten(v, f"{key}.")) else: out[key] = v return out def resolve_config_path(explicit: str | Path | None = None) -> Path: """Explicit arg -> $PPSQ_CONFIG -> ./config.toml (next to this module).""" if explicit: return Path(explicit) env = os.environ.get("PPSQ_CONFIG") if env: return Path(env) return Path(__file__).with_name("config.toml") class ConfigStore: def __init__(self, path: str | Path | None = None): # Guard: a test that forgets PPSQ_CONFIG must never touch the real config.toml. if os.environ.get("PYTEST_CURRENT_TEST") and not (path or os.environ.get("PPSQ_CONFIG")): raise RuntimeError( "Refusing to open the default config.toml under pytest. " "Set PPSQ_CONFIG to a temp copy (see tests/conftest.py)." ) self.path = resolve_config_path(path) if not self.path.exists(): raise SystemExit( f"Missing {self.path}. Copy config.example.toml to config.toml and edit it." ) self._lock = threading.RLock() self._version = 0 doc = self._read_doc() data = _migrate(_plain(doc)) _validate(data) self._snapshot = self._build_snapshot(data) self._pps_fp: tuple | None = None self._pps: PPSClient | None = None self._rebuild_pps(data) # ------------------------------------------------------------------ reading def snapshot(self) -> Snapshot: """Current config. Lock-free: attribute reads are atomic under the GIL.""" return self._snapshot def pps(self) -> PPSClient: return self._pps # type: ignore[return-value] def redacted(self) -> dict: """Admin-panel payload. Secrets are replaced by a `_set` boolean.""" snap = self._snapshot data = { "pps": dict(_plain(snap.pps)), "quarantine": dict(_plain(snap.quarantine)), "app": dict(_plain(snap.app)), "auth": dict(_plain(snap.auth)), } for dotted in SECRET_KEYS: section, _, leaf = dotted.rpartition(".") parent = _dig(data, section) if section else data if isinstance(parent, dict) and leaf in parent: parent[f"{leaf}_set"] = bool(parent.pop(leaf)) # okta is server-only: expose nothing but whether it is configured. data["okta"] = {"configured": bool(_dig(snap.okta, "client_id"))} return data # ------------------------------------------------------------------ writing def apply(self, patch: dict, *, actor: str = "system") -> ApplyResult: """Validate + atomically write a nested patch, then swap in a new snapshot.""" with self._lock: flat = _flatten(patch) rejected = [k for k in flat if k not in ADMIN_EDITABLE] if rejected: raise ConfigError( [{"field": k, "message": "unknown or non-editable key"} for k in rejected] ) doc = self._read_doc() current = _migrate(_plain(doc)) merged = copy.deepcopy(current) applied: list[str] = [] for dotted, value in flat.items(): # Blank secret = leave unchanged (write-only fields). if dotted in SECRET_KEYS and (value is None or value == ""): continue if _dig(merged, dotted) == value: continue _set_dotted(merged, dotted, value) applied.append(dotted) if not applied: return ApplyResult(self._version, [], []) _validate(merged) for dotted in applied: _set_dotted_doc(doc, dotted, _dig(merged, dotted)) self._write_atomic(doc) self._rebuild_pps(merged) self._snapshot = self._build_snapshot(merged) log.info("config updated by %s: %s", actor, ", ".join(sorted(applied))) return ApplyResult( version=self._snapshot.version, applied=sorted(applied), restart_required=sorted(k for k in applied if k in RESTART_KEYS), ) def reload_from_disk(self) -> Snapshot: with self._lock: data = _migrate(_plain(self._read_doc())) _validate(data) self._rebuild_pps(data) self._snapshot = self._build_snapshot(data) log.info("config reloaded from %s (version %d)", self.path, self._snapshot.version) return self._snapshot # ------------------------------------------------------------------ internals def _read_doc(self): with self.path.open("r", encoding="utf-8") as fh: return tomlkit.parse(fh.read()) def _build_snapshot(self, data: dict) -> Snapshot: self._version += 1 return Snapshot( version=self._version, pps=_freeze(data.get("pps", {})), quarantine=_freeze(data.get("quarantine", {})), app=_freeze(data.get("app", {})), auth=_freeze(data.get("auth", {})), okta=_freeze(data.get("okta", {})), path=self.path, ) def _rebuild_pps(self, data: dict) -> None: """Rebuild the PPS client only when connection identity changed.""" p = data.get("pps", {}) fp = tuple(p.get(k) for k in _PPS_FINGERPRINT) if fp == self._pps_fp and self._pps is not None: return # Rebind the client before the snapshot; in-flight requests keep their own ref. self._pps = PPSClient( base_url=p["base_url"], username=p["username"], password=p["password"], verify_tls=p.get("verify_tls", False), timeout=int(p.get("timeout", 120)), client_cert=p.get("client_cert") or None, client_key=p.get("client_key") or None, ) self._pps_fp = fp def _write_atomic(self, doc) -> None: """Temp file in the same dir -> fsync -> backup -> atomic rename -> fsync dir.""" parent = self.path.parent fd, tmp = tempfile.mkstemp(dir=parent, prefix=".config.", suffix=".toml.tmp") try: os.fchmod(fd, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(tomlkit.dumps(doc)) fh.flush() os.fsync(fh.fileno()) if self.path.exists(): shutil.copy2(self.path, self.path.with_name(self.path.name + ".bak")) os.replace(tmp, self.path) # same filesystem by construction tmp = None dfd = os.open(parent, os.O_DIRECTORY) try: os.fsync(dfd) finally: os.close(dfd) finally: if tmp: Path(tmp).unlink(missing_ok=True) def _set_dotted(d: dict, dotted: str, value) -> None: parts = dotted.split(".") cur = d for part in parts[:-1]: cur = cur.setdefault(part, {}) cur[parts[-1]] = value def _set_dotted_doc(doc, dotted: str, value) -> None: """Set a key in a tomlkit document, creating intermediate tables as needed.""" parts = dotted.split(".") cur = doc for part in parts[:-1]: if part not in cur: cur[part] = tomlkit.table() cur = cur[part] cur[parts[-1]] = value def _migrate(data: dict) -> dict: """In-memory back-compat. Never rewrites the user's file on load. `quarantine.report_release_folder` (PoC) -> `[quarantine.report_release]` with steps=["release","move"], which reproduces the old hardcoded behaviour exactly. """ q = data.setdefault("quarantine", {}) if "report_release" not in q: legacy = q.get("report_release_folder") q["report_release"] = { "steps": ["release", "move"] if legacy else ["release"], "move_target": legacy or "", } rr = q["report_release"] rr.setdefault("steps", ["release", "move"]) rr.setdefault("move_target", q.get("report_release_folder", "") or "") rr.setdefault("step_delay_seconds", 60) q.setdefault("default_sort_field", "subject") q.setdefault("default_sort_dir", "asc") a = data.setdefault("auth", {}) # Fail closed: a config with no explicit mode defaults to oidc (which then requires # [okta] and fails loudly if missing), never to shared-password static admin. a.setdefault("mode", "oidc") a.setdefault("users", []) a.setdefault( "denied_message", "Your account is not authorised to use the PPS Quarantine Manager.", ) return data def _validate(data: dict) -> None: """Validate the merged config. Raises ConfigError with per-field messages.""" import pipeline # local import: pipeline imports nothing from us errors: list[dict[str, str]] = [] def bad(field: str, msg: str) -> None: errors.append({"field": field, "message": msg}) p = data.get("pps", {}) url = str(p.get("base_url", "")) parsed = urlparse(url) if parsed.scheme not in ("http", "https") or not parsed.netloc: bad("pps.base_url", "must be an http(s) URL with a host") elif parsed.scheme == "http" and not os.environ.get("PPSQ_ALLOW_INSECURE"): bad("pps.base_url", "must use https (set PPSQ_ALLOW_INSECURE=1 to override)") try: t = int(p.get("timeout", 120)) if not 5 <= t <= 600: bad("pps.timeout", "must be between 5 and 600 seconds") except (TypeError, ValueError): bad("pps.timeout", "must be an integer") for key in ("client_cert", "client_key"): val = p.get(key) if val and not Path(val).exists(): bad(f"pps.{key}", f"file not found: {val}") q = data.get("quarantine", {}) folders = q.get("folders", []) if not isinstance(folders, (list, tuple)) or not folders: bad("quarantine.folders", "at least one folder is required") folders = [] else: seen = set() for f in folders: if not isinstance(f, str) or not f.strip(): bad("quarantine.folders", "folder names must be non-empty strings") elif "," in f: # localguids are joined with "," in the POST payload (pps_client.act). bad("quarantine.folders", f"folder name may not contain a comma: {f!r}") elif f != f.strip(): bad("quarantine.folders", f"folder name has leading/trailing space: {f!r}") elif len(f) > 128: bad("quarantine.folders", f"folder name too long: {f[:20]!r}…") elif f in seen: bad("quarantine.folders", f"duplicate folder: {f!r}") seen.add(f) for field in ("default_folder", "deleted_folder"): val = q.get(field) if folders and val and val not in folders: bad(f"quarantine.{field}", f"{val!r} is not in the folder list") try: lim = int(q.get("default_limit", 200)) if not 1 <= lim <= 1000: bad("quarantine.default_limit", "must be between 1 and 1000") except (TypeError, ValueError): bad("quarantine.default_limit", "must be an integer") try: cs = int(q.get("chunk_size", 25)) if not 1 <= cs <= 500: bad("quarantine.chunk_size", "must be between 1 and 500") except (TypeError, ValueError): bad("quarantine.chunk_size", "must be an integer") try: db = int(q.get("default_days_back", 7)) if db < 1: bad("quarantine.default_days_back", "must be at least 1") except (TypeError, ValueError): bad("quarantine.default_days_back", "must be an integer") if not str(q.get("list_query", "")).strip(): bad("quarantine.list_query", "required (the PPS API rejects folder-only searches)") if q.get("default_sort_field") not in _SORT_FIELDS: bad("quarantine.default_sort_field", f"must be one of {sorted(_SORT_FIELDS)}") if q.get("default_sort_dir") not in _SORT_DIRS: bad("quarantine.default_sort_dir", f"must be one of {sorted(_SORT_DIRS)}") rr = q.get("report_release", {}) steps = rr.get("steps", []) if not isinstance(steps, (list, tuple)) or not steps: bad("quarantine.report_release.steps", "select at least one action") else: unknown = [s for s in steps if s not in pipeline.ALLOWED_STEPS] if unknown: bad("quarantine.report_release.steps", f"unknown step(s): {unknown}") elif len(set(steps)) != len(steps): bad("quarantine.report_release.steps", "duplicate steps") elif not _is_subsequence(steps, pipeline.ALLOWED_STEPS): bad( "quarantine.report_release.steps", f"must follow the fixed order {list(pipeline.ALLOWED_STEPS)}", ) if "move" in steps: target = rr.get("move_target") if not target: bad("quarantine.report_release.move_target", "required when 'move' is selected") elif folders and target not in folders: bad( "quarantine.report_release.move_target", f"{target!r} is not in the folder list", ) if "delete" in steps and not q.get("deleted_folder"): bad("quarantine.deleted_folder", "required when 'delete' is selected") try: d = int(rr.get("step_delay_seconds", 60)) if not 0 <= d <= 3600: bad("quarantine.report_release.step_delay_seconds", "must be between 0 and 3600") except (TypeError, ValueError): bad("quarantine.report_release.step_delay_seconds", "must be an integer") a = data.get("auth", {}) if a.get("mode") not in ("oidc", "static"): bad("auth.mode", "must be 'oidc' or 'static'") users = a.get("users", []) if not isinstance(users, (list, tuple)): bad("auth.users", "must be a list") else: seen_emails = set() for i, u in enumerate(users): if not isinstance(u, Mapping): bad(f"auth.users[{i}]", "must be a table with email and role") continue email = str(u.get("email", "")).strip() if "@" not in email or " " in email or len(email) < 3: bad(f"auth.users[{i}].email", f"invalid email: {email!r}") elif email.casefold() in seen_emails: bad(f"auth.users[{i}].email", f"duplicate user: {email!r}") seen_emails.add(email.casefold()) if u.get("role") not in _ROLES: bad(f"auth.users[{i}].role", f"must be one of {sorted(_ROLES)}") if errors: raise ConfigError(errors) def _is_subsequence(seq, universe) -> bool: it = iter(universe) return all(any(x == u for u in it) for x in seq)