| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306 |
- """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")
|