proofpoint-quarantine-portal/wsgi.py

"""Composition root and server entrypoint.

This is the ONLY module that constructs the store/queue/prefs and starts the worker
thread. `python wsgi.py` runs the app. Tests import `app.create_app` directly and never
touch this file, so importing the app never reads the real config or starts a thread.

Deployment invariant: run as a SINGLE process. The worker thread and the SQLite
connections live in-process; multiple processes would run competing queue workers and an
in-process config lock that doesn't coordinate file writes. Do not add web-server workers.
"""

from __future__ import annotations

import logging
import os
import sys
from logging.handlers import RotatingFileHandler

from app import create_app
from config_store import ConfigStore
from prefs import PrefStore
from worker import JobQueue

_PLACEHOLDER_SECRET = "change-me-to-a-random-string"


def _configure_logging(app_cfg) -> None:
    level = getattr(logging, str(app_cfg.get("log_level", "INFO")).upper(), logging.INFO)
    root = logging.getLogger()
    root.setLevel(level)
    fmt = logging.Formatter("%(asctime)s %(levelname)-7s %(name)s: %(message)s")
    sh = logging.StreamHandler()
    sh.setFormatter(fmt)
    root.addHandler(sh)
    # App log to a rotating file too, so admin errors are viewable in the panel.
    fh = RotatingFileHandler(
        app_cfg.get("app_log", "app.log"), maxBytes=5_000_000, backupCount=5, encoding="utf-8"
    )
    fh.setFormatter(fmt)
    root.addHandler(fh)


def build():
    store = ConfigStore()
    snap = store.snapshot()
    _configure_logging(snap.app)
    log = logging.getLogger("pps.wsgi")

    secret = snap.app.get("secret_key", "")
    if secret == _PLACEHOLDER_SECRET or len(secret) < 32:
        raise SystemExit(
            "app.secret_key is missing/weak. Generate one with:\n"
            "  python -c \"import secrets; print(secrets.token_urlsafe(48))\""
        )

    mode = snap.auth.get("mode", "static")
    if mode == "static":
        log.warning("=" * 72)
        log.warning("AUTH MODE = static — anyone with the shared password is an ADMIN.")
        log.warning('This is a DEVELOPMENT mode. Set [auth] mode = "oidc" for production.')
        log.warning("=" * 72)
        listen = snap.app.get("listen", "127.0.0.1")
        if listen not in ("127.0.0.1", "localhost", "::1") and not os.environ.get(
            "PPSQ_ALLOW_INSECURE_AUTH"
        ):
            raise SystemExit(
                f"Refusing to serve static auth on a non-loopback interface ({listen}). "
                'Use [auth] mode = "oidc", or set PPSQ_ALLOW_INSECURE_AUTH=1 for dev.'
            )

    db_path = snap.app.get("db_path", "jobs.db")
    queue = JobQueue(
        db_path,
        pps_provider=store.pps,
        cfg_provider=lambda: store.snapshot().quarantine,
        ops_log_path=snap.app.get("worker_log", "worker.log"),
    )
    prefs = PrefStore(db_path)
    app = create_app(store, queue, prefs)
    return app, queue


def main() -> int:
    from waitress import serve

    app, queue = build()
    queue.start()
    store = app.config["STORE"]
    snap = store.snapshot()
    host = snap.app.get("listen", "127.0.0.1")
    port = int(snap.app.get("port", 8080))
    logging.getLogger("pps.wsgi").info(
        "PPS Quarantine Manager listening on http://%s:%s", host, port
    )
    serve(app, host=host, port=port, threads=8)
    return 0


if __name__ == "__main__":
    sys.exit(main())