proofpoint-quarantine-portal/auth.py

"""Authentication and authorisation.

Two modes, selected by `[auth] mode` in config (server-file only, never admin-editable):

* **oidc** — Okta via Authlib. `/login` redirects to Okta; `/authorize` validates the
  token, reads the email, and checks it against the admin-managed allowed-users list.
* **static** — a single shared credential for local development, so the app runs and is
  testable without an Okta tenant. Loud startup warning; refuses non-loopback binds.

Roles come from the app's own users list (`[[auth.users]]`), re-resolved on every
request against the live config snapshot, so removing/demoting a user takes effect
immediately without them re-logging-in.

`is_allowed` and `safe_next` are pure and Flask-free — the security-critical logic is
unit-testable without a browser or an Okta tenant.
"""

from __future__ import annotations

import hmac
import logging
import secrets
from functools import wraps
from urllib.parse import urlparse

from authlib.integrations.flask_client import OAuth
from flask import (
    Blueprint,
    abort,
    current_app,
    redirect,
    render_template,
    request,
    session,
    url_for,
)

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

_oauth = OAuth()


# ------------------------------------------------------------------ pure helpers

def is_allowed(email: str, users) -> tuple[bool, str | None]:
    """Return (allowed, role). Case-insensitive email match against the users list."""
    key = (email or "").strip().casefold()
    if not key:
        return False, None
    for u in users:
        if str(u.get("email", "")).strip().casefold() == key:
            return True, u.get("role", "user")
    return False, None


def safe_next(target: str | None) -> str:
    """Sanitise a post-login redirect target to a local path (blocks open redirects)."""
    if not target:
        return "/"
    # Reject anything that could send the browser off-site: absolute URLs, protocol-
    # relative (//host), and backslash tricks that some browsers treat as a slash.
    if not target.startswith("/") or target.startswith("//") or target.startswith("/\\"):
        return "/"
    parsed = urlparse(target)
    if parsed.scheme or parsed.netloc:
        return "/"
    return target


# ------------------------------------------------------------------ session/user

def current_user() -> dict | None:
    return session.get("user")


def _resolve_role(store) -> str | None:
    """The logged-in user's current role.

    In OIDC mode this is re-resolved against the LIVE users list, so a removed/demoted
    user loses access immediately. In static (dev) mode the shared account isn't in the
    users list, so the role fixed at login (from [auth.static]) is trusted.
    """
    user = current_user()
    if not user:
        return None
    if store.snapshot().auth.get("mode") == "static":
        return user.get("role")
    allowed, role = is_allowed(user.get("email", ""), store.snapshot().auth.get("users", []))
    return role if allowed else None


def login_required(view):
    @wraps(view)
    def wrapped(*args, **kwargs):
        if not current_user():
            if request.path.startswith("/api/"):
                abort(401)
            return redirect(url_for("auth.login", next=request.path))
        return view(*args, **kwargs)

    return wrapped


def admin_required(view):
    @wraps(view)
    def wrapped(*args, **kwargs):
        if not current_user():
            if request.path.startswith("/api/"):
                abort(401)
            return redirect(url_for("auth.login", next=request.path))
        store = current_app.config["STORE"]
        if _resolve_role(store) != "admin":
            abort(403)
        return view(*args, **kwargs)

    return wrapped


# ------------------------------------------------------------------ CSRF

def csrf_token() -> str:
    tok = session.get("csrf")
    if not tok:
        tok = secrets.token_urlsafe(32)
        session["csrf"] = tok
    return tok


def check_csrf() -> None:
    """before_request hook: require a matching X-CSRF-Token on state-changing /api calls."""
    if request.method in ("GET", "HEAD", "OPTIONS"):
        return
    if not request.path.startswith("/api/"):
        return  # static-mode /login form post is exempt (has no session yet)
    tok = session.get("csrf")
    if not tok or not hmac.compare_digest(tok, request.headers.get("X-CSRF-Token", "")):
        abort(400, "CSRF token missing or invalid")


# ------------------------------------------------------------------ blueprint

def init_auth(app, store) -> Blueprint:
    cfg = store.snapshot()
    mode = cfg.auth.get("mode", "static")
    bp = Blueprint("auth", __name__)

    if mode == "oidc":
        okta = cfg.okta
        _oauth.init_app(app)
        _oauth.register(
            name="okta",
            server_metadata_url=str(okta["issuer"]).rstrip("/")
            + "/.well-known/openid-configuration",
            client_id=okta["client_id"],
            client_secret=okta["client_secret"],
            client_kwargs={"scope": "openid email profile"},
        )

    @bp.route("/login", methods=["GET", "POST"])
    def login():
        store = current_app.config["STORE"]
        snap = store.snapshot()
        target = safe_next(request.args.get("next"))
        if snap.auth.get("mode") == "static":
            return _static_login(snap, target)
        # OIDC: stash next in the session and bounce to Okta.
        session["next"] = target
        return _oauth.okta.authorize_redirect(redirect_uri=snap.okta["redirect_uri"])

    @bp.route("/authorize")
    def authorize():
        store = current_app.config["STORE"]
        snap = store.snapshot()
        token = _oauth.okta.authorize_access_token()  # validates state/nonce/id_token
        info = token.get("userinfo") or _oauth.okta.userinfo(token=token)
        email = (info or {}).get("email", "")
        name = (info or {}).get("name", email)
        return _finish_login(snap, email, name, safe_next(session.pop("next", None)))

    @bp.route("/logout")
    def logout():
        session.clear()
        return redirect(url_for("auth.login"))

    return bp


def _static_login(snap, target: str):
    static = snap.auth.get("static", {})
    error = None
    if request.method == "POST":
        u = request.form.get("username", "")
        p = request.form.get("password", "")
        ok_u = hmac.compare_digest(u, str(static.get("username", "")))
        ok_p = hmac.compare_digest(p, str(static.get("password", "")))
        if ok_u and ok_p:
            return _finish_login(
                snap,
                static.get("email", "admin@example.invalid"),
                static.get("username", "admin"),
                target,
                forced_role=static.get("role", "admin"),
            )
        error = "Invalid credentials"
    return render_template("login.html", error=error, next_url=target)


def _finish_login(snap, email: str, name: str, target: str, forced_role: str | None = None):
    """Common tail: check access, set the session, redirect (or render denied)."""
    if forced_role is not None:
        allowed, role = True, forced_role
    else:
        allowed, role = is_allowed(email, snap.auth.get("users", []))
    if not allowed:
        log.warning("access denied for %r", email)
        return render_template(
            "denied.html",
            message=snap.auth.get("denied_message", "Your account is not authorised."),
            email=email,
        ), 403
    session.clear()  # drop any OIDC state; prevent session fixation
    session["user"] = {"email": email, "role": role, "name": name}
    session["csrf"] = secrets.token_urlsafe(32)
    session.permanent = False  # browser-session only (decision)
    log.info("login: %s (%s)", email, role)
    return redirect(target)