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

"""Authentication for the Work Package Suite.

A self-contained username/password login. Passwords are stored only as bcrypt
hashes; a successful login 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.

Roles: 'admin' (may manage users) and 'user'.
"""
import os
import secrets
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional

import bcrypt
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 login lasts before the user must sign in again.
SESSION_HOURS = int(os.getenv("AUTH_SESSION_HOURS", "12"))

# Password policy (shared by the API and the CLI).
MIN_PASSWORD_LEN = int(os.getenv("AUTH_MIN_PASSWORD_LEN", "12"))
_COMMON_PASSWORDS = {
    "password", "password1", "password123", "passw0rd", "12345678", "123456789",
    "1234567890", "qwerty123", "letmein123", "changeme", "admin123", "welcome123",
    "iloveyou1", "abc12345", "qwertyuiop",
}


def password_problem(pw: str, username: str = "", email: str = "") -> Optional[str]:
    """Return a human-readable reason the password is unacceptable, or None if OK.
    Shared by the API endpoints and the CLI so the policy is enforced everywhere."""
    if len(pw) < MIN_PASSWORD_LEN:
        return f"Password must be at least {MIN_PASSWORD_LEN} characters."
    low = pw.lower()
    if username and low == username.strip().lower():
        return "Password must not be the same as the username."
    if email and low == email.strip().lower():
        return "Password must not be the same as the email."
    if low in _COMMON_PASSWORDS:
        return "That password is too common — choose something less guessable."
    return None

# Paths under /api that do NOT require a session (login itself, 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()


# ── password hashing ──────────────────────────────────────────────────────────
def hash_password(plain: str) -> str:
    # bcrypt operates on at most 72 bytes; longer inputs are truncated by the
    # algorithm. Encode explicitly so non-ASCII passwords hash consistently.
    return bcrypt.hashpw(plain.encode("utf-8")[:72], bcrypt.gensalt()).decode("ascii")


def verify_password(plain: str, hashed: str) -> bool:
    if not hashed:
        return False
    try:
        return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("ascii"))
    except (ValueError, TypeError):
        return False


# ── 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."""
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[JWT_ALG])
    except jwt.PyJWTError:
        return None


# ── 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 user.role != "admin":
        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()