proofpoint-quarantine-portal/admin.py

"""Admin panel — detached page at /admin plus /api/admin/* endpoints.

All routes require the admin role. Settings are read/written through the ConfigStore
(atomic write, comment-preserving, secrets never returned). Users and the denied message
are ordinary config keys, so they go through the same single PUT /api/admin/config write
path — one atomic write, no interleaving between "save users" and "save folders".
"""

from __future__ import annotations

import logging

from flask import Blueprint, jsonify, render_template, request

import logs
import pipeline
from auth import admin_required, current_user
from config_store import ADMIN_EDITABLE, RESTART_KEYS, ConfigError
from pps_client import PPSClient, PPSError

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


def init_admin(store, queue) -> Blueprint:
    bp = Blueprint("admin", __name__)

    @bp.route("/admin")
    @admin_required
    def admin_page():
        return render_template("admin.html")

    @bp.route("/api/admin/config", methods=["GET", "PUT"])
    @admin_required
    def admin_config():
        if request.method == "GET":
            snap = store.snapshot()
            return jsonify(
                {
                    "config": store.redacted(),
                    "schema": _schema(snap),
                    "version": snap.version,
                }
            )
        # PUT
        patch = request.get_json(silent=True) or {}
        try:
            result = store.apply(patch, actor=current_user()["email"])
        except ConfigError as exc:
            return jsonify({"errors": exc.errors}), 400
        return jsonify(
            {
                "version": result.version,
                "applied": result.applied,
                "restart_required": result.restart_required,
                "config": store.redacted(),
            }
        )

    @bp.route("/api/admin/pps-test", methods=["POST"])
    @admin_required
    def admin_pps_test():
        """Build a throwaway client from the PENDING form values and probe PPS.

        Never persists, never touches the live client. Blank password falls back to the
        stored one so the admin can test other fields without re-typing the secret.
        """
        body = request.get_json(silent=True) or {}
        snap = store.snapshot()
        p = snap.pps
        password = body.get("password") or p.get("password")
        try:
            client = PPSClient(
                base_url=body.get("base_url") or p["base_url"],
                username=body.get("username") or p["username"],
                password=password,
                verify_tls=body.get("verify_tls", p.get("verify_tls", False)),
                timeout=int(body.get("timeout") or p.get("timeout", 120)),
                client_cert=(body.get("client_cert") or p.get("client_cert")) or None,
                client_key=(body.get("client_key") or p.get("client_key")) or None,
            )
            q = snap.quarantine
            records = client.search(
                q.get("default_folder", "Quarantine"),
                q.get("list_query", "from=*"),
                limit=1,
                days_back=int(q.get("default_days_back", 7)),
            )
            return jsonify({"ok": True, "detail": f"Connected. {len(records)} record(s) sampled."})
        except PPSError as exc:
            return jsonify({"ok": False, "detail": str(exc)})
        except Exception as exc:  # noqa: BLE001 - surface any client/build error to the admin
            return jsonify({"ok": False, "detail": f"{type(exc).__name__}: {exc}"})

    @bp.route("/api/admin/jobs")
    @admin_required
    def admin_jobs():
        limit = _int(request.args.get("limit"), 50, 1, 500)
        return jsonify({"jobs": queue.recent_jobs(limit=limit), "active": queue.has_active_jobs()})

    @bp.route("/api/admin/jobs/<int:job_id>/retry", methods=["POST"])
    @admin_required
    def admin_retry(job_id: int):
        ok = queue.retry(job_id)
        if not ok:
            return jsonify({"error": "job not found or not failed"}), 404
        return jsonify({"job_id": job_id, "status": "pending"})

    @bp.route("/api/admin/logs")
    @admin_required
    def admin_logs():
        source = request.args.get("source", "ops")
        if source not in logs.LOG_SOURCES:
            return jsonify({"error": f"unknown log source: {source}"}), 400
        lines = _int(request.args.get("lines"), 200, 1, logs.MAX_LINES)
        query = request.args.get("q") or None
        app_cfg = store.snapshot().app
        return jsonify(logs.read_log(source, app_cfg, lines=lines, query=query))

    return bp


def _schema(snap) -> dict:
    """Static choices the admin UI needs to render generically."""
    return {
        "folders": list(snap.quarantine.get("folders", [])),
        "steps": list(pipeline.ALLOWED_STEPS),
        "roles": ["admin", "user"],
        "sort_fields": ["subject", "date", "from", "rcpt"],
        "sort_dirs": ["asc", "desc"],
        "auth_mode": snap.auth.get("mode", "static"),
        "editable_keys": sorted(ADMIN_EDITABLE),
        "restart_keys": sorted(RESTART_KEYS),
        "pipeline_sentence": pipeline.describe_pipeline(snap.quarantine),
    }


def _int(raw, default: int, lo: int, hi: int) -> int:
    try:
        return max(lo, min(int(raw), hi))
    except (TypeError, ValueError):
        return default