""" Servicio de autenticación con Keycloak """ import logging from api.v1.modules.a76.tenants.service import TenantService from api.v1.modules.a76.user_tenant.service import UserTenantService from core.config import settings from fastapi import HTTPException from keycloak import KeycloakAdmin, KeycloakOpenID from keycloak.exceptions import KeycloakError from sqlalchemy.orm import Session from .dto import ( LoginRequestDTO, LogoutRequestDTO, RefreshTokenRequestDTO, RegisterRequestDTO, RegisterResponseDTO, TokenResponseDTO, UserInfoResponseDTO, ) 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 tenant_service = TenantService(self.db) user_tenant_service = UserTenantService(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, ) # PASO 1: Primero actualizamos los atributos del usuario ANTES de autenticar # Esto es necesario para que los Protocol Mappers incluyan los valores correctos # en el token que se generará a continuación # Para obtener el user_id, necesitamos hacer una autenticación temporal # o buscar el usuario por username try: 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", verify=True, ) # Buscar usuario por username users = keycloak_admin.get_users({"username": login_data.username}) if users and len(users) > 0: user_id = users[0]["id"] # Verificar si el usuario tiene acceso a este tenant has_access = user_tenant_service.user_has_access_to_tenant( user_id, tenant.id ) if not has_access: logger.warning( f"User {user_id} tried to access tenant {tenant.id} without permission" ) raise HTTPException( status_code=403, detail="You don't have access to this tenant", ) # Obtener los datos actuales del usuario current_user = keycloak_admin.get_user(user_id) current_attributes = current_user.get("attributes", {}) # Actualizar los atributos de tenant current_attributes["tenant_id"] = [str(tenant.id)] current_attributes["tenant_slug"] = [tenant.slug] # Actualizar el usuario con los nuevos atributos update_payload = { "email": current_user.get("email"), "firstName": current_user.get("firstName"), "lastName": current_user.get("lastName"), "enabled": current_user.get("enabled", True), "emailVerified": current_user.get("emailVerified", False), "attributes": current_attributes, } keycloak_admin.update_user(user_id=user_id, payload=update_payload) except KeycloakError as e: logger.warning(f"Could not pre-update user attributes: {str(e)}") # Continuamos con el login aunque falle la actualización except HTTPException: raise # Re-lanzamos las excepciones HTTP (como acceso denegado) except Exception as e: logger.warning(f"Error pre-updating user attributes: {str(e)}") # PASO 2: Ahora autenticamos al usuario # Si los Protocol Mappers están configurados, el token incluirá # automáticamente los atributos tenant_id y tenant_slug actualizados token_response = keycloak_client.token( username=login_data.username, password=login_data.password, grant_type=["password"], ) 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) 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]) 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)}") # Agregar el usuario al tenant en la base de datos try: from api.v1.modules.a76.user_tenant.service import UserTenantService user_tenant_service = UserTenantService(self.db) user_tenant_service.add_user_to_tenant( keycloak_user_id=user_id, tenant_id=tenant.id, role="user", # Rol por defecto ) except Exception as e: # Si falla, hacer rollback del usuario en Keycloak logger.error(f"Failed to add user to tenant in database: {str(e)}") try: keycloak_admin.delete_user(user_id) except Exception as e: pass raise HTTPException( status_code=500, detail="Failed to register user in database" ) 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 # 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, ) # 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: # Validar que el tenant existe y está activo 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")