from fastapi import APIRouter, Depends, HTTPException, status, Request, Response from sqlalchemy.orm import Session from datetime import datetime, timedelta, timezone from typing import Optional from slowapi import Limiter from slowapi.util import get_remote_address from ..database import get_db from ..models.user import User from ..models.session import Session as SessionModel from ..models.permission import Role from ..schemas.user import UserCreate, UserLogin, UserResponse, Token, TokenRefresh from ..auth import ( verify_password, get_password_hash, create_access_token, create_refresh_token, get_current_active_user, get_current_user_optional ) from ..config import settings router = APIRouter(prefix="/api/auth", tags=["auth"]) limiter = Limiter(key_func=get_remote_address) @router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED) @limiter.limit("5/minute") async def register(request: Request, user_data: UserCreate, db: Session = Depends(get_db)): # Check if user exists existing_user = db.query(User).filter( (User.username == user_data.username) | (User.email == user_data.email) ).first() if existing_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Registration failed. Please check your input." ) # Check if this is the first user - if so, make them admin user_count = db.query(User).count() if user_count == 0: # First user gets admin role admin_role = db.query(Role).filter(Role.name == "admin").first() role_id = admin_role.id if admin_role else None else: # Default to viewer role for new users viewer_role = db.query(Role).filter(Role.name == "viewer").first() role_id = viewer_role.id if viewer_role else None # Create new user new_user = User( username=user_data.username, email=user_data.email, password_hash=get_password_hash(user_data.password), role_id=role_id ) db.add(new_user) db.commit() db.refresh(new_user) return new_user @router.post("/login") @limiter.limit("5/minute") async def login( request: Request, response: Response, user_data: UserLogin, db: Session = Depends(get_db) ): # Find user user = db.query(User).filter(User.username == user_data.username).first() if not user or not verify_password(user_data.password, user.password_hash): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication failed" ) if not user.is_active: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Authentication failed" ) # Create tokens access_token = create_access_token(data={"sub": str(user.id)}) refresh_token = create_refresh_token(data={"sub": str(user.id)}) # Save refresh token to database expires_at = datetime.now(timezone.utc) + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS) session = SessionModel( user_id=user.id, refresh_token=refresh_token, expires_at=expires_at, ip_address=request.client.host if request.client else None, user_agent=request.headers.get("user-agent") ) db.add(session) db.commit() # Set HttpOnly cookies response.set_cookie( key="access_token", value=access_token, httponly=True, secure=True, samesite="lax", max_age=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60, path="/" ) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=True, samesite="lax", max_age=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60, path="/" ) return {"user": UserResponse.from_orm_with_role(user), "access_token": access_token} @router.post("/refresh") async def refresh( request: Request, response: Response, db: Session = Depends(get_db) ): # Get refresh token from cookie refresh_token = request.cookies.get("refresh_token") if not refresh_token: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication failed" ) # Find session session = db.query(SessionModel).filter( SessionModel.refresh_token == refresh_token ).first() if not session or session.expires_at < datetime.now(timezone.utc): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication failed" ) user = db.query(User).filter(User.id == session.user_id).first() if not user or not user.is_active: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication failed" ) # Create new tokens access_token = create_access_token(data={"sub": str(user.id)}) new_refresh_token = create_refresh_token(data={"sub": str(user.id)}) # Update session session.refresh_token = new_refresh_token session.expires_at = datetime.now(timezone.utc) + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS) db.commit() # Set HttpOnly cookies response.set_cookie( key="access_token", value=access_token, httponly=True, secure=True, samesite="lax", max_age=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES * 60, path="/" ) response.set_cookie( key="refresh_token", value=new_refresh_token, httponly=True, secure=True, samesite="lax", max_age=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60, path="/" ) return {"user": UserResponse.from_orm_with_role(user), "access_token": access_token} @router.post("/logout") async def logout( response: Response, request: Request, db: Session = Depends(get_db) ): # Try to get current user from token (optional) try: token = get_token_from_cookie(request) if token: try: payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) user_id = payload.get("sub") if user_id: # Remove all sessions for user db.query(SessionModel).filter( SessionModel.user_id == user_id ).delete() db.commit() except JWTError: pass # Token invalid, just clear cookies except: pass # Any error, just clear cookies # Clear cookies regardless of auth status response.delete_cookie(key="access_token", path="/") response.delete_cookie(key="refresh_token", path="/") return {"message": "Successfully logged out"} @router.get("/me", response_model=UserResponse | None) async def get_me(current_user: Optional[User] = Depends(get_current_user_optional)): if not current_user: return None return UserResponse.from_orm_with_role(current_user)