proofpoint-quarantine-portal/app.py

"""PPS Quarantine Manager — Flask app factory.

`create_app(store, queue, prefs)` wires the routes with no import-time side effects
(no config read, no worker thread, no network). The composition root is `wsgi.py`;
tests build an app directly with a temp config and a fake PPS client.

Slow actions (release/move/delete) are queued to the background JobQueue and return
immediately; the browser polls /api/jobs for progress.
"""

from __future__ import annotations

import email
import logging
import re
from email import policy

from flask import Flask, jsonify, render_template, request

import auth
from admin import init_admin
from pps_client import PPSError
from worker import VALID_ACTIONS, JobQueue

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


def create_app(store, queue: JobQueue, prefs) -> Flask:
    app = Flask(__name__)
    snap = store.snapshot()

    app.secret_key = snap.app["secret_key"]
    app.config.update(
        STORE=store,
        QUEUE=queue,
        PREFS=prefs,
        SESSION_COOKIE_HTTPONLY=True,
        # Lax (not Strict): the Okta callback is a cross-site top-level GET; Strict would
        # withhold the cookie and every login would fail with mismatching_state.
        SESSION_COOKIE_SAMESITE="Lax",
        SESSION_COOKIE_SECURE=bool(snap.app.get("cookie_secure", True)),
        SESSION_COOKIE_NAME="ppsq_session",
        MAX_CONTENT_LENGTH=4 * 1024 * 1024,
    )

    app.register_blueprint(auth.init_auth(app, store))
    app.register_blueprint(init_admin(store, queue))
    app.before_request(auth.check_csrf)

    @app.context_processor
    def _inject_user():
        return {"user": auth.current_user()}

    # ------------------------------------------------------------------ pages

    @app.route("/")
    @auth.login_required
    def index():
        return render_template("index.html")

    # ------------------------------------------------------------------ JSON API

    @app.route("/api/config")
    @auth.login_required
    def api_config():
        cfg = store.snapshot().quarantine
        user_email = auth.current_user()["email"]
        return jsonify(
            {
                "folders": list(cfg.get("folders", [])),
                "deleted_folder": cfg.get("deleted_folder"),
                "report_release": {
                    "steps": list(cfg.get("report_release", {}).get("steps", [])),
                    "move_target": cfg.get("report_release", {}).get("move_target"),
                },
                "prefs": prefs.effective(user_email, store.snapshot()),
                "csrf_token": auth.csrf_token(),
                "config_version": store.snapshot().version,
                "is_admin": auth.current_user().get("role") == "admin",
            }
        )

    @app.route("/api/messages")
    @auth.login_required
    def api_messages():
        """Return one page of a folder, newest-first.

        The client pages through the whole folder by passing `before` (the oldest date
        it has seen) as a cursor, since the PPS API caps each response at 1000 rows and
        offers no offset paging. `page.has_more`/`page.oldest` drive that loop.
        """
        cfg = store.snapshot().quarantine
        folder = request.args.get("folder") or cfg.get("default_folder", "Quarantine")
        page_size = _clamp_limit(request.args.get("limit"), cfg)
        before = request.args.get("before") or None
        try:
            records = store.pps().search(
                folder,
                cfg.get("list_query", "from=*"),
                limit=page_size,
                days_back=int(cfg.get("default_days_back", 7)),
                enddate=before,
            )
        except PPSError as exc:
            return _pps_error_response(exc, f"search folder={folder!r} before={before!r}")

        raw_count = len(records)
        hidden = queue.acted_localguids(folder)
        messages = [
            {
                "date": r.get("date"),
                "from": r.get("from"),
                "rcpts": r.get("rcpts", []),
                "subject": r.get("subject"),
                "guid": r.get("guid"),
                "localguid": r.get("localguid"),
                "size": r.get("size"),
            }
            for r in records
            if r.get("localguid") not in hidden
        ]
        # Cursor is the oldest date in the RAW batch (before the acted-on filter), so
        # paging never terminates early just because a page was mostly filtered out.
        dates = [r.get("date") for r in records if r.get("date")]
        oldest = min(dates) if dates else None
        return jsonify(
            {
                "folder": folder,
                "messages": messages,
                "page": {
                    "raw_count": raw_count,
                    "oldest": oldest,
                    "has_more": raw_count >= page_size,
                },
            }
        )

    @app.route("/api/message/<path:guid>")
    @auth.login_required
    def api_message(guid: str):
        try:
            raw = store.pps().get_raw(guid)
        except PPSError as exc:
            return _pps_error_response(exc, f"get_raw guid={guid!r}")
        return jsonify(_parse_message(raw))

    @app.route("/api/actions", methods=["POST"])
    @auth.login_required
    def api_actions():
        cfg = store.snapshot().quarantine
        folders = set(cfg.get("folders", []))
        body = request.get_json(silent=True) or {}
        action = body.get("action")
        folder = body.get("folder")
        messages = body.get("messages", [])
        targetfolder = body.get("targetfolder")

        if action not in VALID_ACTIONS:
            return jsonify({"error": f"unknown action: {action}"}), 400
        if not folder:
            return jsonify({"error": "folder is required"}), 400
        if not messages:
            return jsonify({"error": "no messages selected"}), 400
        if action == "move":
            if not targetfolder:
                return jsonify({"error": "move requires a targetfolder"}), 400
            # Validate against the folder list — otherwise any user could move mail to
            # an arbitrary folder name.
            if targetfolder not in folders:
                return jsonify({"error": f"unknown target folder: {targetfolder}"}), 400

        items = [
            {
                "localguid": m.get("localguid"),
                "guid": m.get("guid"),
                "subject": m.get("subject"),
                "sender": m.get("sender"),
                "recipient": m.get("recipient"),
            }
            for m in messages
            if m.get("localguid")
        ]
        if not items:
            return jsonify({"error": "no valid messages (missing localguid)"}), 400

        extra = {"targetfolder": targetfolder} if action == "move" else {}
        user = auth.current_user()["email"]
        job_id = queue.enqueue(action, folder, items, extra, user=user)
        return jsonify({"job_id": job_id, "queued": len(items)}), 202

    @app.route("/api/jobs")
    @auth.login_required
    def api_jobs():
        # Users see only their own jobs; admins see everything.
        user = auth.current_user()
        scope = None if user.get("role") == "admin" else user["email"]
        return jsonify(
            {"jobs": queue.recent_jobs(limit=25, user=scope),
             "config_version": store.snapshot().version}
        )

    @app.route("/api/prefs", methods=["GET", "PUT", "DELETE"])
    @auth.login_required
    def api_prefs():
        user_email = auth.current_user()["email"]
        snap = store.snapshot()
        if request.method == "GET":
            return jsonify(prefs.describe(user_email, snap))
        if request.method == "PUT":
            patch = request.get_json(silent=True) or {}
            try:
                effective = prefs.set_many(user_email, patch, snap)
            except ValueError as exc:
                return jsonify({"error": str(exc)}), 400
            return jsonify({"prefs": effective})
        # DELETE
        keys = (request.get_json(silent=True) or {}).get("keys")
        prefs.clear(user_email, keys)
        return jsonify({"prefs": prefs.effective(user_email, snap)})

    # ------------------------------------------------------------------ helpers

    def _pps_error_response(exc: PPSError, context: str):
        log.error("%s failed: %s", context, exc)
        return (
            jsonify(
                {"error": str(exc), "detail": exc.body, "status": exc.status, "reqid": exc.reqid}
            ),
            502,
        )

    return app


def _clamp_limit(raw: str | None, cfg) -> int:
    default = int(cfg.get("default_limit", 200))
    try:
        value = int(raw) if raw is not None else default
    except (TypeError, ValueError):
        value = default
    return max(1, min(value, 1000))


def _parse_message(raw: bytes) -> dict:
    """Parse raw RFC822 bytes into headers + text/html bodies + attachments."""
    msg = email.message_from_bytes(raw, policy=policy.default)
    headers = {
        "from": msg.get("From", ""),
        "to": msg.get("To", ""),
        "cc": msg.get("Cc", ""),
        "date": msg.get("Date", ""),
        "subject": msg.get("Subject", ""),
    }

    text_body = ""
    html_body = ""
    try:
        text_part = msg.get_body(preferencelist=("plain",))
        if text_part is not None:
            text_body = text_part.get_content()
    except Exception:  # noqa: BLE001 - malformed MIME shouldn't 500 the view
        text_body = ""
    try:
        html_part = msg.get_body(preferencelist=("html",))
        if html_part is not None:
            html_body = html_part.get_content()
    except Exception:  # noqa: BLE001
        html_body = ""

    if not text_body and not html_body:
        try:
            text_body = msg.get_content()
        except Exception:  # noqa: BLE001
            text_body = raw.decode("utf-8", errors="replace")

    return {
        "headers": headers,
        "text": text_body,
        "html": html_body,
        "attachments": _parse_attachments(msg),
        "raw_headers": _raw_headers(raw),
    }


def _parse_attachments(msg) -> list[dict]:
    """List attachment parts as {filename, content_type, size} (size in bytes)."""
    attachments = []
    try:
        for part in msg.iter_attachments():
            payload = part.get_payload(decode=True)
            attachments.append(
                {
                    "filename": part.get_filename() or "(unnamed)",
                    "content_type": part.get_content_type(),
                    "size": len(payload) if payload is not None else None,
                }
            )
    except Exception:  # noqa: BLE001 - malformed MIME shouldn't 500 the view
        pass
    return attachments


def _raw_headers(raw: bytes) -> str:
    """The literal RFC822 header block (everything before the first blank line)."""
    head = re.split(rb"\r?\n\r?\n", raw, maxsplit=1)[0]
    return head.decode("utf-8", errors="replace")