app/auth.py

"""Authentication: password hashing, server-side sessions, and the FastAPI
dependencies that enforce role boundaries on every protected route.

Sessions, not JWT (per user decision): a session is an opaque random token
(`secrets.token_urlsafe`) stored in the `sessions` table and in an httpOnly
cookie. Every authenticated request does one DB lookup (token -> user, with
an `expires_at` check) -- logout, or any future "disable this account"
action, is an instant, unconditional revocation (delete the row), which a
JWT can't give you without an extra denylist table.

Password hashing uses the `bcrypt` library directly (not passlib -- passlib
is unmaintained and breaks against modern bcrypt releases; bcrypt is what
passlib wraps internally anyway).
"""

from __future__ import annotations

import os
import secrets
from datetime import datetime, timedelta, timezone

import bcrypt
from fastapi import Cookie, Depends, HTTPException, Response, status
from sqlalchemy import select
from sqlalchemy.orm import Session as DBSession

from app.config import settings
from app.db.models import Client, Session as SessionModel, User, UserRole
from app.db.session import get_session

SESSION_COOKIE_NAME = "session_token"

# Secure-by-default: the cookie is only ever sent back by a browser over
# HTTPS, which is correct once this app sits behind the reverse proxy that
# terminates TLS externally. Set COOKIE_SECURE=false only for local dev
# when hitting the app directly over http:// (e.g. http://127.0.0.1:8811),
# bypassing the proxy -- never in a deployed environment.
COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "true").strip().lower() != "false"
BCRYPT_MAX_PASSWORD_BYTES = 72  # bcrypt's own hard limit


def hash_password(password: str) -> str:
    if len(password.encode("utf-8")) > BCRYPT_MAX_PASSWORD_BYTES:
        raise ValueError(f"Password must be at most {BCRYPT_MAX_PASSWORD_BYTES} bytes.")
    hashed = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
    return hashed.decode("utf-8")


def verify_password(password: str, password_hash: str) -> bool:
    try:
        return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
    except ValueError:
        # Malformed/foreign hash -- never let this raise into a 500 that
        # might hint at *why* verification failed.
        return False


def create_session(db: DBSession, user: User) -> tuple[str, datetime]:
    """Create a session row for `user`, returning (token, expires_at).

    Also opportunistically clears this user's already-expired sessions --
    a lightweight tidy-up rather than a background job, adequate at this
    scale.
    """
    now = datetime.now(timezone.utc)
    db.query(SessionModel).filter(
        SessionModel.user_id == user.id, SessionModel.expires_at < now
    ).delete()

    token = secrets.token_urlsafe(32)
    expires_at = now + timedelta(minutes=settings.SESSION_EXPIRE_MINUTES)
    db.add(SessionModel(token=token, user_id=user.id, expires_at=expires_at))
    db.flush()
    return token, expires_at


def revoke_session(db: DBSession, token: str) -> None:
    db.query(SessionModel).filter(SessionModel.token == token).delete()


def set_session_cookie(response: Response, token: str, expires_at: datetime) -> None:
    max_age = max(0, int((expires_at - datetime.now(timezone.utc)).total_seconds()))
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=token,
        httponly=True,
        secure=COOKIE_SECURE,
        samesite="lax",
        max_age=max_age,
        path="/",
    )


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


def get_current_user(
    session_token: str | None = Cookie(default=None, alias=SESSION_COOKIE_NAME),
    db: DBSession = Depends(get_session),
) -> User:
    """FastAPI dependency: resolve the logged-in user from the session
    cookie. Raises 401 if there's no valid, unexpired session -- the same
    error regardless of *why* (missing cookie, unknown token, expired
    token) so nothing about the failure reason leaks.
    """
    unauthorized = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Not authenticated.",
        headers={"WWW-Authenticate": "Cookie"},
    )
    if not session_token:
        raise unauthorized

    now = datetime.now(timezone.utc)
    stmt = select(SessionModel).where(
        SessionModel.token == session_token, SessionModel.expires_at > now
    )
    db_session = db.execute(stmt).scalar_one_or_none()
    if db_session is None:
        raise unauthorized

    user = db.get(User, db_session.user_id)
    if user is None:
        raise unauthorized
    return user


def require_officer(user: User = Depends(get_current_user)) -> User:
    if user.role != UserRole.OFFICER:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="This action requires a banking officer account.",
        )
    return user


def require_client(user: User = Depends(get_current_user)) -> User:
    if user.role != UserRole.CLIENT:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="This action requires a client account.",
        )
    return user


def get_current_client_profile(
    user: User = Depends(require_client), db: DBSession = Depends(get_session)
) -> Client:
    """Resolves the logged-in client's own `clients` row. A client user
    without a linked profile is a data-integrity bug, not a client mistake
    -- 500s loudly rather than pretending they're unauthenticated.
    """
    stmt = select(Client).where(Client.user_id == user.id)
    client = db.execute(stmt).scalar_one_or_none()
    if client is None:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Client account has no linked profile.",
        )
    return client


def seed_officer_if_missing(db: DBSession) -> None:
    """Idempotent startup step (EXPL-04-style: runs once, at lifespan
    startup): ensures `SEED_OFFICER_EMAIL` exists as an officer account, so
    there's always a way in on first launch.
    """
    stmt = select(User).where(User.email == settings.SEED_OFFICER_EMAIL)
    existing = db.execute(stmt).scalar_one_or_none()
    if existing is not None:
        return

    officer = User(
        name="Seed Officer",
        email=settings.SEED_OFFICER_EMAIL,
        password_hash=hash_password(settings.SEED_OFFICER_PASSWORD),
        role=UserRole.OFFICER,
    )
    db.add(officer)
    db.commit()