""" Utilidades de seguridad y autenticación con Keycloak """ from fastapi import HTTPException, Security, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from keycloak import KeycloakOpenID from jose import jwt, JWTError from typing import Optional, Dict, Any from .config import settings import logging from sqlalchemy.orm import Session from core.database import get_core_db from api.v1.modules.a76.user_tenant.models import UserTenant logger = logging.getLogger(__name__) # Configuración de Keycloak 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 ) # Security scheme security = HTTPBearer() def verify_token(token: str) -> Dict[str, Any]: """ Verifica y decodifica un token JWT de Keycloak Args: token: Token JWT Returns: Payload del token decodificado Raises: HTTPException: Si el token es inválido """ try: # Obtener clave pública de Keycloak KEYCLOAK_PUBLIC_KEY = ( "-----BEGIN PUBLIC KEY-----\n" + keycloak_openid.public_key() + "\n-----END PUBLIC KEY-----" ) # Decodificar y verificar token options = { "verify_signature": True, "verify_aud": False, "verify_exp": True } decoded_token = jwt.decode( token, KEYCLOAK_PUBLIC_KEY, algorithms=["RS256"], options=options ) return decoded_token except JWTError as e: logger.error(f"Token verification failed: {str(e)}") raise HTTPException( status_code=401, detail="Could not validate credentials" ) except Exception as e: logger.error(f"Unexpected error during token verification: {str(e)}") raise HTTPException( status_code=401, detail="Authentication error" ) async def get_current_user( credentials: HTTPAuthorizationCredentials = Security(security), db: Session = Depends(get_core_db) ) -> Dict[str, Any]: """ Dependency para obtener el usuario actual desde el token JWT Enriquecido con tenant_id y company_id desde la tabla user_tenant Uso en FastAPI: current_user: dict = Depends(get_current_user) """ token = credentials.credentials user_info = verify_token(token) # Obtener user_id desde el token user_id = user_info.get("sub") if not user_id: raise HTTPException(status_code=401, detail="User ID not found in token") # Consultar la tabla user_tenant para obtener tenant_id y company_id user_tenant = db.query(UserTenant).filter(UserTenant.keycloak_user_id == user_id).first() if not user_tenant: raise HTTPException(status_code=403, detail="User does not have access to any tenant or company") # Enriquecer user_info con tenant_id y company_id user_info["tenant_id"] = user_tenant.tenant_id user_info["company_id"] = user_tenant.company_id return user_info async def get_current_active_user( current_user: Dict[str, Any] = Depends(get_current_user) ) -> Dict[str, Any]: """ Dependency para obtener usuario activo (puede incluir validaciones adicionales) """ # Aquí se pueden agregar validaciones adicionales # Por ejemplo, verificar si el usuario está activo en la BD return current_user def has_role(required_role: str): """ Decorator/Dependency para verificar roles de usuario Uso: @router.get("/admin") async def admin_endpoint(user = Depends(has_role("admin"))): ... """ async def role_checker( current_user: Dict[str, Any] = Depends(get_current_user) ) -> Dict[str, Any]: user_roles = current_user.get("realm_access", {}).get("roles", []) if required_role not in user_roles: raise HTTPException( status_code=403, detail=f"User does not have required role: {required_role}" ) return current_user return role_checker def get_tenant_from_token(user_info: Dict[str, Any]) -> Optional[int]: """ Extrae el tenant_id del token JWT El tenant_id puede estar en diferentes lugares según configuración de Keycloak: - En claims personalizados - En el realm - En atributos del usuario """ # Intentar obtener de claims personalizados tenant_id = user_info.get("tenant_id") if not tenant_id: # Intentar obtener de atributos tenant_id = user_info.get("attributes", {}).get("tenant_id") if tenant_id: return int(tenant_id) return None def validate_company_access( company_id: int, current_user: Dict[str, Any] ) -> bool: """ Valida que el usuario tenga acceso a la compañía solicitada Args: company_id: ID de la compañía a la que se quiere acceder current_user: Información del usuario actual desde el token Returns: True si el usuario tiene acceso, False en caso contrario Nota: Por ahora solo verifica que el tenant_id del usuario coincida con el company_id. Se puede extender para validar permisos específicos por compañía. """ tenant_id = get_tenant_from_token(current_user) # Si no hay tenant_id en el token, denegar acceso if not tenant_id: return False # Validar que el company_id pertenezca al tenant del usuario # Por ahora asumimos que company_id == tenant_id # Esto se puede modificar si hay una tabla de relación tenant-company return tenant_id == company_id def validate_access_to_resource( company_id: int, current_user: dict = Depends(get_current_user) ) -> bool: """ Valida que el usuario tenga acceso a un recurso específico basado en company_id y regresa el tenant_id Args: company_id: company_id asociado al recurso current_user: Información del usuario actual desde el token Returns: True si el usuario tiene acceso, False en caso contrario """ tenant_id = get_tenant_from_token(current_user) if not tenant_id: raise HTTPException(status_code=400, detail="Tenant ID not found in token") if not validate_company_access(company_id, current_user): raise HTTPException(status_code=403, detail="Access denied to this company") # Validar que el tenant_id del usuario coincida con el del recurso return tenant_id class KeycloakClient: """Cliente para interactuar con Keycloak Admin API""" def __init__(self): self.openid = keycloak_openid def create_user(self, email: str, password: str, tenant_id: int, **kwargs): """Crea un usuario en Keycloak""" # Implementar lógica para crear usuario usando keycloak admin pass def assign_role(self, user_id: str, role: str): """Asigna un rol a un usuario""" pass def create_tenant_realm(self, tenant_name: str): """Crea un realm para un nuevo tenant""" pass