Project-SDE-WP-Suite/server/auth.py

"""Authentication for the Work Package Suite.

Identity is confirmed by Okta (OIDC authorization-code flow, see server/okta_auth.py
and the routes in server/app.py); there is no local password anywhere in this app
(D15, D16 — T10.4 removed the last of it). A successful sign-in issues a signed JWT
that rides in an HttpOnly cookie (`wp_session`). Because the token is signed and
self-validating, there is no server-side session store — every request is checked by
verifying the cookie's signature and expiry (see `auth_gate` and `get_current_user`).

Security model:
  • The real boundary is `auth_gate` (middleware in app.py): every /api/ data
    route is refused with 401 unless a valid session cookie is present.
  • The cookie is HttpOnly (JS can't read it → XSS can't steal the session),
    SameSite=Lax (blunts CSRF), and Secure whenever the request arrives over
    HTTPS (detected via X-Forwarded-Proto behind NGINX).
  • The signing secret comes from AUTH_SECRET_KEY. In production this MUST be
    set; if it is missing we fall back to a random per-process key (which logs a
    warning and invalidates every session on restart) so dev still works.

Permissions roles (`User.role`) — distinct from a person's job function on the
project, which lives in `User.project_role` and grants nothing. Decided entirely
locally: Okta gates WHO can authenticate at all, this app decides what an
authenticated account may do — see D15/D16.
  • admin          application administrator: user administration, app settings,
                   and implicit access to every project.
  • project_super_user
                   everything a project_admin may do, plus USER ADMINISTRATION
                   scoped to the projects they hold the role on: they create and
                   manage the accounts on their own jobs without an app admin
                   having to do it for them. They cannot reach app settings, and
                   they cannot create or alter an admin / super-user account.
  • project_admin  within their assigned projects: may delete work packages,
                   modify a SOP after it has been completed, and delete projects.
  • project_user   normal member: creates and edits work packages, authors a SOP
                   up to completion. May NOT delete WPs or change a completed SOP.
                   Also where Okta JIT provisioning (T10.3) lands a brand-new
                   account — the lowest-privilege role, promoted locally from
                   there by an admin (see manage_users.py for the bootstrap case).

The user-administration SCOPE of a super user is worked out in server/app.py
(`managed_project_ids`, `manage_user_problem`), because it depends on project
membership rows — this module only decides which roles carry the power at all.
"""
import os
import secrets
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional

import jwt
from fastapi import Depends, HTTPException, Request, Response, status
from sqlalchemy import select, func
from sqlalchemy.orm import Session

from .db import get_db, DATABASE_URL
from . import models

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

COOKIE_NAME = "wp_session"
JWT_ALG = "HS256"
# How long a session lasts before the person must sign in again.
SESSION_HOURS = int(os.getenv("AUTH_SESSION_HOURS", "12"))

# ── permissions roles ─────────────────────────────────────────────────────────
ROLE_ADMIN = "admin"
ROLE_PROJECT_SUPER = "project_super_user"
ROLE_PROJECT_ADMIN = "project_admin"
ROLE_PROJECT_USER = "project_user"
# Ordered most- to least-privileged; the console renders dropdowns in this order.
ROLES = (ROLE_ADMIN, ROLE_PROJECT_SUPER, ROLE_PROJECT_ADMIN, ROLE_PROJECT_USER)
ROLE_LABELS = {
    ROLE_ADMIN: "Administrator",
    ROLE_PROJECT_SUPER: "Project Super User",
    ROLE_PROJECT_ADMIN: "Project Admin",
    ROLE_PROJECT_USER: "Project User",
}
# Roles that may be held ON A SINGLE PROJECT via ProjectMember.role, so someone can
# run the users on one job and be an ordinary member of the next. '' means "inherit
# the account's own role" and is always allowed alongside these.
PROJECT_SCOPED_ROLES = (ROLE_PROJECT_SUPER, ROLE_PROJECT_ADMIN, ROLE_PROJECT_USER)
# Job functions offered in the admin console. Free text underneath, so a project
# can use a title that isn't on this list.
PROJECT_ROLES = (
    "Project Manager", "Assistant Project Manager", "Construction Manager",
    "Quality Manager", "Superintendent", "General Foreman", "Foreman",
    "Planner / Scheduler", "BIM / VDC Coordinator", "Engineer",
    "Safety (HSE)", "Warehouse / Materials", "Commissioning", "Field Technician",
)


def normalize_role(role: Optional[str]) -> str:
    """Map a stored/incoming role onto the current vocabulary.

    Accounts created before permissions roles existed carry the legacy value
    'user', which means exactly what 'project_user' means now."""
    r = (role or "").strip()
    if r == "user":
        return ROLE_PROJECT_USER
    return r if r in ROLES else ROLE_PROJECT_USER


def is_admin(user: "models.User") -> bool:
    return normalize_role(user.role) == ROLE_ADMIN


def is_project_admin(user: "models.User") -> bool:
    """True for the roles allowed to delete work packages and change a completed
    SOP. A super user is a project admin with user administration on top, so it is
    included here — never enumerate the two roles by hand."""
    return normalize_role(user.role) in (ROLE_ADMIN, ROLE_PROJECT_SUPER, ROLE_PROJECT_ADMIN)



# NOTE: "may this account administer users?" is deliberately NOT answered here. The
# super-user role can be held per project (ProjectMember.role), so the question needs
# membership rows to answer and lives in app.py — `is_user_manager` /
# `require_user_manager` / `managed_project_ids`. An account-role-only version of the
# same question used to exist here and silently disagreed with the scoped one, which
# locked per-project super users out of the routes they were entitled to.

# Paths under /api that do NOT require a session (the Okta routes themselves, health, docs).
_EXEMPT_PREFIXES = ("/api/auth/",)
_EXEMPT_EXACT = {
    "/api/health",
    "/api/docs",
    "/api/openapi.json",
    "/api/docs/oauth2-redirect",
    "/api/redoc",
}


def _load_secret() -> str:
    s = os.getenv("AUTH_SECRET_KEY")
    if s:
        return s
    # No key configured. In production (a real database is configured via
    # POSTGRES_* / DATABASE_URL) this is FATAL — refuse to start rather than sign
    # sessions with a throwaway key that silently rotates on every restart. In
    # local dev (SQLite, no DB env) fall back to an ephemeral key so the app still
    # runs zero-config.
    # "Prod" = a real (non-SQLite) database is in use — matches exactly the
    # condition db.py uses to pick Postgres, so we don't wrongly block a
    # zero-config SQLite dev run just because a stray POSTGRES_USER is exported.
    is_prod = not str(DATABASE_URL).startswith("sqlite")
    if is_prod:
        raise RuntimeError(
            "AUTH_SECRET_KEY is not set. Refusing to start in production with an "
            "ephemeral signing key — set a strong fixed AUTH_SECRET_KEY "
            "(see server/.env.example / DEPLOYMENT.md)."
        )
    log.warning(
        "AUTH_SECRET_KEY is not set — using a random ephemeral key for local dev. "
        "Logins reset on restart. Set AUTH_SECRET_KEY for anything non-dev."
    )
    return secrets.token_urlsafe(48)


SECRET_KEY = _load_secret()


# ── tokens ──────────────────────────────────────────────────────────────────
def create_token(user: "models.User") -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user.id,
        "username": user.username,
        "role": user.role,
        "ver": user.token_version or 0,
        "iat": now,
        "exp": now + timedelta(hours=SESSION_HOURS),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=JWT_ALG)


def decode_token(token: str) -> Optional[dict]:
    """Return the token claims if the signature and expiry are valid, else None.
    Session cookies only — a token of any other type is rejected."""
    try:
        claims = jwt.decode(token, SECRET_KEY, algorithms=[JWT_ALG])
    except jwt.PyJWTError:
        return None
    # No route issues a `typ`-carrying token anymore (that was the password-reset
    # token, removed in T10.4), but a session token still must never validate as
    # one — kept as a defensive check, cheap insurance against a future token type
    # riding the same cookie.
    if claims.get("typ"):
        return None
    return claims


# ── cookie helpers ────────────────────────────────────────────────────────────
def _is_https(request: Request) -> bool:
    # Behind NGINX, TLS is terminated at the proxy and forwarded as plain HTTP,
    # so trust X-Forwarded-Proto (set in nginx-wp-suite.conf) when present.
    xfp = request.headers.get("x-forwarded-proto", "")
    if xfp:
        return xfp.split(",")[0].strip().lower() == "https"
    return request.url.scheme == "https"


def set_session_cookie(response: Response, request: Request, token: str) -> None:
    response.set_cookie(
        key=COOKIE_NAME,
        value=token,
        max_age=SESSION_HOURS * 3600,
        httponly=True,
        secure=_is_https(request),
        samesite="lax",
        path="/",
    )


def clear_session_cookie(response: Response) -> None:
    response.delete_cookie(COOKIE_NAME, path="/")


# ── request gate (used as middleware in app.py) ─────────────────────────────────
def _needs_auth(path: str) -> bool:
    if not path.startswith("/api/"):
        return False  # static assets are served by NGINX, not this app
    if path in _EXEMPT_EXACT:
        return False
    return not any(path.startswith(p) for p in _EXEMPT_PREFIXES)


def is_request_authenticated(request: Request) -> Optional[dict]:
    """Validate the session cookie on a raw request. Returns claims or None.
    Used by the middleware gate, which has no dependency-injection context."""
    token = request.cookies.get(COOKIE_NAME)
    if not token:
        return None
    return decode_token(token)


# ── dependencies (used inside route handlers) ───────────────────────────────────
def get_current_user(request: Request, db: Session = Depends(get_db)) -> "models.User":
    """Resolve the logged-in user from the session cookie, or raise 401.

    Unlike the middleware gate (which only checks the token signature), this also
    confirms the account still exists and is active — so disabling a user takes
    effect on their next request."""
    claims = is_request_authenticated(request)
    if not claims:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
    user = db.get(models.User, claims.get("sub"))
    if not user or not user.is_active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Account is inactive")
    # Session revocation: a mismatch means the token was invalidated (e.g. the
    # password was changed after this token was issued).
    if (claims.get("ver", 0) or 0) != (user.token_version or 0):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Session expired")
    return user


def require_admin(user: "models.User" = Depends(get_current_user)) -> "models.User":
    if not is_admin(user):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
    return user




# ── account helpers (shared by routes and the CLI) ──────────────────────────────
def find_user(db: Session, username: str) -> Optional["models.User"]:
    """Look up by username, case-insensitively (also matches on email)."""
    uname = (username or "").strip().lower()
    if not uname:
        return None
    return db.scalars(
        select(models.User).where(
            (func.lower(models.User.username) == uname)
            | (func.lower(models.User.email) == uname)
        )
    ).first()