""" Servicio de autenticación con Keycloak """ from keycloak import KeycloakOpenID, KeycloakAdmin from keycloak.exceptions import KeycloakError from fastapi import HTTPException from sqlalchemy.orm import Session import logging from core.config import settings from .dto import ( LoginRequestDTO, TokenResponseDTO, RefreshTokenRequestDTO, UserInfoResponseDTO, LogoutRequestDTO, RegisterRequestDTO, RegisterResponseDTO ) logger = logging.getLogger(__name__) class AuthService: """Servicio de autenticación""" def __init__(self, db: Session): self.db = db self.keycloak_openid = KeycloakOpenID( server_url=settings.KEYCLOAK_SERVER_URL, client_id=settings.KEYCLOAK_CLIENT_ID, realm_name=settings.KEYCLOAK_REALM, client_secret_key=settings.KEYCLOAK_CLIENT_SECRET ) def login(self, login_data: LoginRequestDTO) -> TokenResponseDTO: """ Autentica usuario y obtiene tokens Args: login_data: Credenciales de login Returns: TokenResponseDTO con access_token y refresh_token Raises: HTTPException: Si las credenciales son inválidas """ try: # Verificar que el tenant existe from api.v1.modules.a76.tenants.service import TenantService tenant_service = TenantService(self.db) tenant = tenant_service.get_tenant_by_slug(login_data.tenant_slug) if not tenant: raise HTTPException(status_code=404, detail="Tenant not found") if not tenant.is_active: raise HTTPException(status_code=403, detail="Tenant is not active") # Crear nueva instancia de KeycloakOpenID con el realm del tenant keycloak_client = KeycloakOpenID( server_url=settings.KEYCLOAK_SERVER_URL, client_id=settings.KEYCLOAK_CLIENT_ID, realm_name=tenant.keycloak_realm, client_secret_key=settings.KEYCLOAK_CLIENT_SECRET ) # Obtener token de Keycloak token_response = keycloak_client.token( username=login_data.username, password=login_data.password, grant_type=["password"] ) logger.info(f"User logged in: {login_data.username} (tenant: {tenant.slug})") return TokenResponseDTO( access_token=token_response["access_token"], refresh_token=token_response["refresh_token"], token_type="bearer", expires_in=token_response["expires_in"] ) except KeycloakError as e: logger.warning(f"Keycloak authentication failed: {str(e)}") raise HTTPException(status_code=401, detail="Invalid credentials") except HTTPException: raise except Exception as e: logger.error(f"Login error: {str(e)}") raise HTTPException(status_code=500, detail="Authentication error") def refresh_token(self, refresh_data: RefreshTokenRequestDTO) -> TokenResponseDTO: """ Refresca el access token usando refresh token Args: refresh_data: Refresh token Returns: TokenResponseDTO con nuevos tokens """ try: token_response = self.keycloak_openid.refresh_token( refresh_data.refresh_token ) return TokenResponseDTO( access_token=token_response["access_token"], refresh_token=token_response["refresh_token"], token_type="bearer", expires_in=token_response["expires_in"] ) except KeycloakError as e: logger.warning(f"Token refresh failed: {str(e)}") raise HTTPException(status_code=401, detail="Invalid or expired refresh token") except Exception as e: logger.error(f"Token refresh error: {str(e)}") raise HTTPException(status_code=500, detail="Token refresh error") def get_user_info(self, access_token: str) -> UserInfoResponseDTO: """ Obtiene información del usuario desde el token Args: access_token: Access token JWT Returns: UserInfoResponseDTO con información del usuario """ try: user_info = self.keycloak_openid.userinfo(access_token) # Extraer roles roles = [] if "realm_access" in user_info: roles = user_info["realm_access"].get("roles", []) # Extraer tenant_id si está presente tenant_id = user_info.get("tenant_id") if not tenant_id and "attributes" in user_info: tenant_id = user_info["attributes"].get("tenant_id") return UserInfoResponseDTO( sub=user_info.get("sub"), email=user_info.get("email"), name=user_info.get("name"), preferred_username=user_info.get("preferred_username"), tenant_id=int(tenant_id) if tenant_id else None, roles=roles ) except KeycloakError as e: logger.warning(f"Get user info failed: {str(e)}") raise HTTPException(status_code=401, detail="Invalid token") except Exception as e: logger.error(f"Get user info error: {str(e)}") raise HTTPException(status_code=500, detail="Error retrieving user info") def logout(self, logout_data: LogoutRequestDTO) -> dict: """ Cierra sesión invalidando el refresh token Args: logout_data: Refresh token a invalidar Returns: Dict con mensaje de éxito """ try: self.keycloak_openid.logout(logout_data.refresh_token) logger.info("User logged out successfully") return {"message": "Logged out successfully"} except KeycloakError as e: logger.warning(f"Logout failed: {str(e)}") # No lanzamos error aquí, el logout puede fallar si el token ya expiró return {"message": "Logged out"} except Exception as e: logger.error(f"Logout error: {str(e)}") raise HTTPException(status_code=500, detail="Logout error") def register(self, register_data: RegisterRequestDTO) -> RegisterResponseDTO: """ Registra un nuevo usuario en Keycloak Args: register_data: Datos del usuario a registrar Returns: RegisterResponseDTO con información del usuario creado Raises: HTTPException: Si el registro falla """ try: # Verificar que el tenant existe from api.v1.modules.a76.tenants.service import TenantService tenant_service = TenantService(self.db) tenant = tenant_service.get_tenant_by_slug(register_data.tenant_slug) if not tenant: raise HTTPException(status_code=404, detail="Tenant not found") if not tenant.is_active: raise HTTPException(status_code=403, detail="Tenant is not active") # Crear instancia de KeycloakAdmin para gestión de usuarios keycloak_admin = KeycloakAdmin( server_url=settings.KEYCLOAK_SERVER_URL, username=settings.KEYCLOAK_ADMIN_USERNAME, password=settings.KEYCLOAK_ADMIN_PASSWORD, realm_name=tenant.keycloak_realm, user_realm_name="master", # El admin suele estar en master realm verify=True ) # Preparar datos del usuario para Keycloak user_data = { "username": register_data.username, "email": register_data.email, "firstName": register_data.first_name, "lastName": register_data.last_name, "enabled": True, "emailVerified": False, "credentials": [{ "type": "password", "value": register_data.password, "temporary": False }], "attributes": { "tenant_id": str(tenant.id), "tenant_slug": tenant.slug } } # Crear usuario en Keycloak user_id = keycloak_admin.create_user(user_data) # Asignar rol por defecto (user) - opcional, solo si existe try: user_role = keycloak_admin.get_realm_role("user") if user_role: keycloak_admin.assign_realm_roles(user_id, [user_role]) logger.info(f"Assigned 'user' role to {register_data.username}") except KeycloakError as e: # El rol 'user' no existe, no es un error crítico logger.warning(f"Could not assign 'user' role: {str(e)}") logger.info(f"User registered: {register_data.username} (tenant: {tenant.slug}, user_id: {user_id})") return RegisterResponseDTO( user_id=user_id, username=register_data.username, email=register_data.email, message="User registered successfully" ) except KeycloakError as e: error_message = str(e) logger.warning(f"Keycloak registration failed: {error_message}") # Mensajes de error más específicos if "User exists" in error_message or "409" in error_message: raise HTTPException(status_code=409, detail="Username or email already exists") elif "Invalid" in error_message: raise HTTPException(status_code=400, detail="Invalid user data") else: raise HTTPException(status_code=500, detail="Registration error") except HTTPException: raise except Exception as e: logger.error(f"Registration error: {str(e)}") raise HTTPException(status_code=500, detail="Registration error") def exchange_code(self, exchange_data) -> TokenResponseDTO: """ Intercambia un authorization code por tokens Este método se usa cuando el frontend recibe un código de autorización después de un login con proveedor externo (Microsoft, Google, etc.) a través de Keycloak. Args: exchange_data: Datos del código y redirect_uri Returns: TokenResponseDTO con access_token y refresh_token Raises: HTTPException: Si el código es inválido o expiró """ try: # Importar el DTO aquí para evitar referencias circulares from .dto import ExchangeCodeRequestDTO # Intercambiar código por tokens usando Keycloak token_response = self.keycloak_openid.token( grant_type='authorization_code', code=exchange_data.code, redirect_uri=exchange_data.redirect_uri ) logger.info(f"Code exchanged successfully") # Si se proporciona tenant_slug, podríamos validar que el usuario pertenece a ese tenant # Por ahora simplemente retornamos los tokens if exchange_data.tenant_slug: # Decodificar token para obtener tenant_id del usuario user_info = self.keycloak_openid.introspect(token_response['access_token']) user_tenant_id = user_info.get('tenant_id') # Validar que el tenant existe y está activo from api.v1.modules.a76.tenants.service import TenantService tenant_service = TenantService(self.db) tenant = tenant_service.get_tenant_by_slug(exchange_data.tenant_slug) if not tenant: raise HTTPException(status_code=404, detail="Tenant not found") if not tenant.is_active: raise HTTPException(status_code=403, detail="Tenant is not active") # Opcional: Verificar que el usuario pertenece al tenant # Esto depende de cómo manejes los tenants en tu aplicación return TokenResponseDTO( access_token=token_response['access_token'], refresh_token=token_response['refresh_token'], token_type=token_response.get('token_type', 'bearer'), expires_in=token_response.get('expires_in', 3600) ) except KeycloakError as e: error_message = str(e) logger.warning(f"Code exchange failed: {error_message}") if "invalid_grant" in error_message.lower(): raise HTTPException(status_code=400, detail="Invalid or expired authorization code") elif "invalid_client" in error_message.lower(): raise HTTPException(status_code=401, detail="Invalid client credentials") else: raise HTTPException(status_code=500, detail="Token exchange error") except HTTPException: raise except Exception as e: logger.error(f"Code exchange error: {str(e)}") raise HTTPException(status_code=500, detail="Code exchange error")