proofpoint-quarantine-portal/prefs.py

"""Per-user preferences.

Admin config sets the global default for each pref; a user may override it, and the
override persists across logins. Overrides live in the existing jobs.db (a separate
connection — WAL + busy_timeout are set by JobQueue), NOT in the shared config.toml.

Key-value schema so adding a new pref needs no migration. The effective value is the
admin default overridden by the user's stored value ONLY IF it still validates against
the live config — so if an admin deletes a folder a user had pinned, that user quietly
falls back to the default instead of hitting an unknown-folder error.
"""

from __future__ import annotations

import json
import logging
import sqlite3
import threading
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable, Mapping

log = logging.getLogger("pps.prefs")

_SORT_FIELDS = {"subject", "date", "from", "rcpt"}
_SORT_DIRS = {"asc", "desc"}


@dataclass(frozen=True)
class PrefSpec:
    default_path: tuple[str, ...]                     # where the admin default lives
    coerce: Callable[[Any], Any]
    validate: Callable[[Any, Mapping], bool]          # (value, live quarantine cfg) -> ok


def _in_folders(v, q) -> bool:
    return v in q.get("folders", [])


PREF_SPECS: dict[str, PrefSpec] = {
    "default_folder": PrefSpec(("default_folder",), str, _in_folders),
    "default_limit": PrefSpec(("default_limit",), int, lambda v, q: 1 <= v <= 1000),
    "sort_field": PrefSpec(("default_sort_field",), str, lambda v, q: v in _SORT_FIELDS),
    "sort_dir": PrefSpec(("default_sort_dir",), str, lambda v, q: v in _SORT_DIRS),
}


def _now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")


class PrefStore:
    def __init__(self, db_path: str):
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        with self._lock:
            self._conn.execute("PRAGMA busy_timeout=5000")
            self._conn.execute(
                """CREATE TABLE IF NOT EXISTS user_prefs (
                    email      TEXT NOT NULL,
                    key        TEXT NOT NULL,
                    value      TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (email, key)
                )"""
            )
            self._conn.commit()

    # ------------------------------------------------------------------ reads

    def raw(self, email: str) -> dict:
        """The user's stored overrides (decoded), unvalidated."""
        with self._lock:
            rows = self._conn.execute(
                "SELECT key, value FROM user_prefs WHERE email=?", (email,)
            ).fetchall()
        out = {}
        for r in rows:
            try:
                out[r["key"]] = json.loads(r["value"])
            except json.JSONDecodeError:
                continue
        return out

    def effective(self, email: str, snap) -> dict:
        """Admin defaults overridden by valid user values."""
        q = snap.quarantine
        overrides = self.raw(email)
        out = {}
        for key, spec in PREF_SPECS.items():
            default = _dig(q, spec.default_path)
            value = default
            if key in overrides:
                try:
                    cand = spec.coerce(overrides[key])
                    if spec.validate(cand, q):
                        value = cand
                except (TypeError, ValueError):
                    pass
            out[key] = value
        return out

    def describe(self, email: str, snap) -> dict:
        q = snap.quarantine
        defaults = {k: _dig(q, s.default_path) for k, s in PREF_SPECS.items()}
        overrides = self.raw(email)
        return {
            "prefs": self.effective(email, snap),
            "defaults": defaults,
            "overridden": sorted(k for k in overrides if k in PREF_SPECS),
        }

    # ------------------------------------------------------------------ writes

    def set_many(self, email: str, patch: dict, snap) -> dict:
        """Validate + persist a user's overrides. Returns the new effective prefs."""
        q = snap.quarantine
        to_write = []
        for key, value in patch.items():
            spec = PREF_SPECS.get(key)
            if spec is None:
                raise ValueError(f"unknown preference: {key}")
            try:
                coerced = spec.coerce(value)
            except (TypeError, ValueError):
                raise ValueError(f"invalid value for {key}: {value!r}")
            if not spec.validate(coerced, q):
                raise ValueError(f"invalid value for {key}: {value!r}")
            to_write.append((email, key, json.dumps(coerced), _now()))
        with self._lock:
            self._conn.executemany(
                "INSERT INTO user_prefs (email, key, value, updated_at) VALUES (?,?,?,?)"
                " ON CONFLICT(email, key) DO UPDATE SET value=excluded.value,"
                " updated_at=excluded.updated_at",
                to_write,
            )
            self._conn.commit()
        return self.effective(email, snap)

    def clear(self, email: str, keys: list[str] | None = None) -> None:
        with self._lock:
            if keys:
                self._conn.executemany(
                    "DELETE FROM user_prefs WHERE email=? AND key=?",
                    [(email, k) for k in keys],
                )
            else:
                self._conn.execute("DELETE FROM user_prefs WHERE email=?", (email,))
            self._conn.commit()


def _dig(m: Mapping, path: tuple[str, ...]):
    cur: Any = m
    for p in path:
        if not isinstance(cur, Mapping) or p not in cur:
            return None
        cur = cur[p]
    return cur