from __future__ import annotations import time import uuid from fastapi import APIRouter, Depends, HTTPException, Request, Response, status from sqlalchemy import select from sqlalchemy.orm import Session from app.api.deps import get_current_user, get_db from app.core.config import get_settings from app.core.security import ( SessionPayload, create_session_token, hash_password, read_session_token, verify_password, ) from app.models.user import User from app.schemas import CurrentUser, PasswordLoginRequest from app.services.audit import record_audit_event router = APIRouter(prefix="/api/v1/auth", tags=["auth"]) settings = get_settings() def _current_user_out(user: User) -> CurrentUser: return CurrentUser( public_ref=user.public_ref, display_name=user.display_name, role=user.role, # type: ignore[arg-type] ) def _set_session(response: Response, user: User) -> None: token = create_session_token( SessionPayload( user_id=str(user.id), public_ref=user.public_ref, role=user.role, display_name=user.display_name, issued_at=int(time.time()), ) ) response.set_cookie( settings.session_cookie_name, token, httponly=True, samesite="lax", secure=settings.session_cookie_secure, max_age=settings.session_ttl_seconds, ) def bootstrap_initial_admin(db: Session) -> None: """Create or rotate the explicitly configured first manager in operational mode.""" if ( settings.mobilityops_demo_mode or not settings.initial_admin_email or not settings.initial_admin_password ): return email = settings.initial_admin_email.strip().lower() user = db.scalar(select(User).where(User.email == email)) if user is None: user = User( public_ref="USR-ADMIN", email=email, password_hash=hash_password(settings.initial_admin_password), display_name=settings.initial_admin_display_name, role="operations_manager", active=True, ) db.add(user) record_audit_event( db, actor_type="system", actor_label="bootstrap", action="operational_admin_created", entity_type="user", entity_id=user.id, ) db.commit() @router.post("/login", response_model=CurrentUser) def password_login( body: PasswordLoginRequest, response: Response, db: Session = Depends(get_db), ) -> CurrentUser: if settings.mobilityops_demo_mode: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Password login is unavailable in demo mode", ) user = db.scalar(select(User).where(User.email == body.email.strip().lower())) if user is None or not user.active or not verify_password(body.password, user.password_hash): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") _set_session(response, user) record_audit_event( db, actor_type="user", actor_id=user.id, actor_label=user.display_name, action="password_login", entity_type="user", entity_id=user.id, ) db.commit() return _current_user_out(user) @router.get("/session", response_model=CurrentUser) def get_session(response: Response, user: CurrentUser = Depends(get_current_user)) -> CurrentUser: response.headers["Cache-Control"] = "no-store" return user @router.post("/logout") def logout(request: Request, response: Response, db: Session = Depends(get_db)) -> dict: token = request.cookies.get(settings.session_cookie_name) payload = read_session_token(token) if token else None if payload is not None: record_audit_event( db, actor_type="user", actor_id=uuid.UUID(payload.user_id), actor_label=payload.display_name, action="logout", entity_type="user", ) db.commit() response.delete_cookie(settings.session_cookie_name) return {"status": "logged_out"}