proofpoint-quarantine-portal/config_store.py

"""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 `<name>_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",
)

# Report & Release step order before it flipped to move-first (`pipeline.ALLOWED_STEPS`).
# Configs written against it are migrated, not rejected — see `_migrate`.
_LEGACY_STEP_ORDER: tuple[str, ...] = ("release", "move", "delete")

# Steps the pipeline used to offer and no longer does; dropped from old configs on load.
_RETIRED_STEPS: frozenset[str] = frozenset({"delete"})

_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 `<name>_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, [], [])

            # Normalise the merged result (e.g. a step list still in the pre-flip
            # release-before-move order) before validating and writing it.
            _migrate(merged)
            _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 the
    move+release pipeline. Two shapes of retired config are rewritten rather than
    rejected: a `delete` step (Report & Release no longer deletes — Delete is its own
    action) and a step list in the pre-flip order (release before move).
    """
    import pipeline  # local import: pipeline imports nothing from us

    q = data.setdefault("quarantine", {})
    if "report_release" not in q:
        legacy = q.get("report_release_folder")
        q["report_release"] = {
            "steps": ["move", "release"] if legacy else ["release"],
            "move_target": legacy or "",
        }
    rr = q["report_release"]
    rr.setdefault("steps", ["move", "release"])
    steps = rr.get("steps")
    if isinstance(steps, (list, tuple)) and any(s in _RETIRED_STEPS for s in steps):
        steps = [s for s in steps if s not in _RETIRED_STEPS] or ["release"]
        log.warning("dropped retired Report & Release step(s) %s — Delete is a separate "
                    "action; steps are now %s", sorted(_RETIRED_STEPS), steps)
        rr["steps"] = steps
    if (isinstance(steps, (list, tuple))
            and _is_subsequence(steps, _LEGACY_STEP_ORDER)
            and not _is_subsequence(steps, pipeline.ALLOWED_STEPS)):
        rr["steps"] = sorted(steps, key=pipeline.ALLOWED_STEPS.index)
    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",
                )
    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)