""" Authentication Endpoints - ServiceManagerWeb Endpoints para autenticación y autorización """ from fastapi import APIRouter, HTTPException, status, Depends from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from sqlalchemy.orm import selectinload from pydantic import BaseModel, EmailStr from typing import Optional import structlog from app.core.database import get_db from app.core.security import security from app.core.config import get_settings from app.models.user import User from app.models.tenant import Tenant router = APIRouter() logger = structlog.get_logger(__name__) settings = get_settings() # OAuth2 scheme oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"/{settings.API_VERSION}/auth/login") # =================================== # PYDANTIC SCHEMAS # =================================== class LoginRequest(BaseModel): """Schema for login request.""" email: EmailStr password: str tenant_slug: str totp_code: Optional[str] = None class LoginResponse(BaseModel): """Schema for login response.""" access_token: str refresh_token: str token_type: str = "bearer" expires_in: int user: dict class RefreshTokenRequest(BaseModel): """Schema for refresh token request.""" refresh_token: str class TokenResponse(BaseModel): """Schema for token response.""" access_token: str token_type: str = "bearer" expires_in: int # =================================== # ENDPOINTS # =================================== @router.post("/login", response_model=LoginResponse) async def login( login_data: LoginRequest, db: AsyncSession = Depends(get_db) ): """ Authenticate user and return access/refresh tokens. Args: login_data: Login credentials db: Database session Returns: LoginResponse with tokens and user info Raises: HTTPException: If authentication fails """ logger.info( "Login attempt", email=login_data.email, tenant_slug=login_data.tenant_slug ) # 1. Buscar usuario en base de datos query = select(User).where(User.email == login_data.email) result = await db.execute(query) user = result.scalar_one_or_none() # 2. Verificar usuario y contraseña if not user or not security.verify_password(login_data.password, user.password_hash): logger.warning( "Login failed - invalid credentials", email=login_data.email ) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Credenciales inválidas" ) # 3. Verificar si está activo if not user.is_active: logger.warning( "Login failed - user inactive", email=login_data.email ) raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Usuario inactivo" ) # Create tokens token_data = { "sub": str(user.id), "email": user.email, "role": user.role.value if hasattr(user.role, "value") else user.role, "tenant_id": str(user.tenant_id) } access_token = security.create_access_token(token_data) refresh_token = security.create_refresh_token(token_data) logger.info( "Login successful", email=login_data.email, tenant_slug=login_data.tenant_slug, user_id=str(user.id) ) return LoginResponse( access_token=access_token, refresh_token=refresh_token, expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60, user={ "id": str(user.id), "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "role": user.role, "tenant_id": str(user.tenant_id), "is_active": user.is_active, "is_two_factor_enabled": user.totp_enabled or False, "created_at": user.created_at.isoformat() if user.created_at else None } ) @router.post("/refresh", response_model=TokenResponse) async def refresh_token( refresh_data: RefreshTokenRequest, db: AsyncSession = Depends(get_db) ): """ Refresh access token using refresh token. Args: refresh_data: Refresh token data db: Database session Returns: New access token Raises: HTTPException: If refresh token is invalid """ logger.info("Token refresh attempt") # Verify refresh token payload = security.verify_token(refresh_data.refresh_token) if not payload or payload.get("type") != "refresh": logger.warning("Token refresh failed - invalid token") raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token" ) # TODO: Check if refresh token exists in database and is not revoked # Create new access token token_data = { "sub": payload["sub"], "email": payload["email"], "role": payload["role"], "tenant_id": payload["tenant_id"] } access_token = security.create_access_token(token_data) logger.info("Token refresh successful", user_id=payload["sub"]) return TokenResponse( access_token=access_token, expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60 ) @router.post("/logout") async def logout( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db) ): """ Logout user and revoke refresh token. Args: token: Access token db: Database session Returns: Success message """ logger.info("Logout attempt") # Verify token payload = security.verify_token(token) if not payload: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token" ) # TODO: Revoke refresh token in database logger.info("Logout successful", user_id=payload["sub"]) return {"message": "Successfully logged out"} @router.get("/me") async def get_current_user( token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db) ): """ Get current user information. Args: token: Access token db: Database session Returns: Current user data Raises: HTTPException: If token is invalid """ # Verify token payload = security.verify_token(token) if not payload: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token" ) user_id = payload.get("sub") if not user_id: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload" ) # Fetch actual user from database query = select(User).where(User.id == user_id).options( selectinload(User.tenant) ) result = await db.execute(query) user = result.scalar_one_or_none() if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) if not user.is_active: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User account is disabled" ) return { "id": str(user.id), "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "role": user.role.value if hasattr(user.role, 'value') else user.role, "tenant_id": str(user.tenant_id), "tenant_name": user.tenant.name if user.tenant else None, "is_active": user.is_active, "is_two_factor_enabled": user.totp_secret is not None, "last_login": user.last_login.isoformat() if user.last_login else None, "created_at": user.created_at.isoformat() } # =================================== # DEPENDENCIES # =================================== # Dependencies are imported from app.api.deps to avoid duplication # Use get_current_user and get_current_active_superuser from deps.py