from datetime import datetime from fastapi import APIRouter, HTTPException, Depends, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from jose import JWTError, jwt from passlib.context import CryptContext from app.models.auth_models import ( LoginRequest, RegisterRequest, EmailRequest, RefreshTokenRequest, LoginResponse, MessageResponse, ErrorResponse ) from app.services.auth import AuthService from app.services.token import TokenService from app.services.mail import MailService from app.utils.jwt_utils import create_access_token, create_refresh_token, decode_token from app.config.settings import Config router = APIRouter(prefix="/auth", tags=["Authentication"]) security = HTTPBearer() config = Config() async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)): """Dependency to get current user from JWT token""" try: token = credentials.credentials payload = decode_token(token) email: str = payload.get("sub") if email is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) user = AuthService.get_user_by_email(email) if user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found", headers={"WWW-Authenticate": "Bearer"}, ) return user except JWTError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) @router.post("/login", response_model=LoginResponse) async def login(login_data: LoginRequest): """ User Login Endpoint: Authenticate and obtain access & refresh tokens. This endpoint allows users to log in by providing their email and password. Upon successful authentication, it returns access and refresh tokens that can be used for authorized access to protected resources. """ try: # Check if the provided email is valid user = AuthService.get_user_by_email(login_data.email) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Access denied: User with this email does not exist." ) # Authenticate user authenticated_user = AuthService.authenticate(login_data.email, login_data.password) if not authenticated_user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Access denied: Incorrect password." ) # Check if user is active if not authenticated_user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied: Account disabled, contact admin." ) # Check if user is blocked if authenticated_user.is_blocked: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Access denied: Account Forbidden." ) # Update user login timestamps authenticated_user.update_last_login() authenticated_user.update_last_seen() authenticated_user.save() # Create additional claims additional_claims = { "is_superuser": authenticated_user.is_superuser, "is_active": authenticated_user.is_active, "is_staff": authenticated_user.is_staff, "is_admin": authenticated_user.is_admin, } # Generate tokens access_token = create_access_token( data={"sub": authenticated_user.email}, additional_claims=additional_claims ) refresh_token = create_refresh_token(data={"sub": authenticated_user.email}) return LoginResponse( msg="Login successful", user_id=authenticated_user.public_id, token={ "access": access_token, "refresh": refresh_token } ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.post("/register", response_model=MessageResponse) async def register(register_data: RegisterRequest): """ User Registration Endpoint: Create a new user account. This endpoint allows users to create a new account by providing their registration details, including username, company info, email, and password. Upon successful registration, a user account is created, and they can use it to log in. """ try: # Check if user already exists existing_user = AuthService.get_user_by_email(register_data.email) if existing_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="User with this email already exists" ) # Create new user user = AuthService.create_user( email=register_data.email, password=register_data.password, username=register_data.username, company_name=register_data.company_name, company_type=register_data.company_type ) # Send verification email try: MailService.send_verification_email(user) except Exception as e: # Log the error but don't fail registration print(f"Failed to send verification email: {str(e)}") return MessageResponse( msg="Registration successful. Please check your email for verification instructions." ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.post("/logout", response_model=MessageResponse) async def logout(refresh_data: RefreshTokenRequest, current_user=Depends(get_current_user)): """ User Logout Endpoint: Revoke access and refresh tokens. This endpoint allows users to log out, revoking their access and refresh tokens. """ try: # Decode and validate refresh token payload = decode_token(refresh_data.refresh) refresh_exp = payload.get('exp') if refresh_exp is not None and datetime.utcnow().timestamp() >= refresh_exp: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Refresh token has expired" ) # Add tokens to blacklist jti = payload.get("jti") if jti: TokenService.blacklist_token(current_user, jti) return MessageResponse(msg="Logout successful") except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.get("/token/verify", response_model=MessageResponse) async def verify_token(current_user=Depends(get_current_user)): """ Token Verification Endpoint: Verify the validity of an access token. This endpoint allows users to verify the validity of their access token. """ return MessageResponse(msg="Token is valid") @router.post("/token/refresh", response_model=dict) async def refresh_token(refresh_data: RefreshTokenRequest): """ Token Refresh Endpoint: Refresh an access token using a valid refresh token. This endpoint allows users to refresh their access token using a valid refresh token. """ try: # Decode refresh token payload = decode_token(refresh_data.refresh) email = payload.get("sub") if not email: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token" ) # Get user user = AuthService.get_user_by_email(email) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found" ) # Create new access token additional_claims = { "is_superuser": user.is_superuser, "is_active": user.is_active, "is_staff": user.is_staff, "is_admin": user.is_admin, } new_access_token = create_access_token( data={"sub": user.email}, additional_claims=additional_claims ) return { "msg": "Token refreshed successfully", "token": { "access": new_access_token } } except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.get("/activate/{activation_token}", response_model=MessageResponse) async def verify_email(activation_token: str): """ Activate user account: verify email """ try: # Verify activation token and activate user result = AuthService.verify_email_token(activation_token) if result: return MessageResponse(msg="Email verified successfully") else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid or expired activation token" ) except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.post("/resend-email-verify", response_model=MessageResponse) async def resend_verification_email(email_data: EmailRequest): """ Resend Verification Email Endpoint: Resend a verification email to the user. """ try: user = AuthService.get_user_by_email(email_data.email) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) if user.is_verified: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email is already verified" ) # Send verification email MailService.send_verification_email(user) return MessageResponse(msg="Verification email sent successfully") except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" ) @router.post("/forgot-password", response_model=MessageResponse) async def forgot_password(email_data: EmailRequest): """ Forgot Password Endpoint: Initiate the password reset process """ try: user = AuthService.get_user_by_email(email_data.email) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Send password reset email MailService.send_password_reset_email(user) return MessageResponse(msg="Password reset instructions have been sent to your email") except HTTPException: raise except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"An error occurred while processing your request: {str(e)}" )