"""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()