""" Utilidades de seguridad y autenticación con Keycloak """ import logging from typing import Any, Dict, Optional from fastapi import Depends, HTTPException, Security from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from jose import JWTError, jwt from keycloak import KeycloakOpenID from sqlalchemy.orm import Session from .config import settings logger = logging.getLogger(__name__) # Configuración de Keycloak keycloak_openid = KeycloakOpenID( server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth", 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), ) -> Dict[str, Any]: """ Dependency para obtener el usuario actual desde el token JWT Uso en FastAPI: current_user: dict = Depends(get_current_user) """ token = credentials.credentials user_info = verify_token(token) 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: logger.warning(f"Role denied. Required: {required_role}. User actually has: {user_roles}") # Also check client roles as a debug fallback client_roles = current_user.get("resource_access", {}) logger.warning(f"User client roles: {client_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( db: Session, 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: Verifica que la compañía pertenezca al tenant del usuario consultando la BD. """ tenant_id = get_tenant_from_token(current_user) # Si no hay tenant_id en el token, denegar acceso if not tenant_id: return False # Consultar si la compañía pertenece al tenant try: from api.v1.modules.a76.general_catalogs.company.models import Company company = ( db.query(Company) .filter(Company.id == company_id, Company.tenant_id == tenant_id) .first() ) return company is not None except Exception as e: logger.error(f"Error validating company access: {str(e)}") return False def validate_access_to_resource( db: Session, company_id: int, current_user: Dict[str, Any], required_permissions: Optional[list[str]] = None, require_all: bool = True, ) -> int: """ Valida que el usuario tenga acceso a un recurso específico basado en company_id y regresa el tenant_id. Opcionalmente verifica permisos. Args: db: Sesión de base de datos company_id: company_id asociado al recurso current_user: Información del usuario actual desde el token required_permissions: Lista opcional de permisos requeridos. Si es None, no verifica permisos. require_all: Si True, requiere TODOS los permisos. Si False, requiere AL MENOS UNO. Returns: tenant_id si el usuario tiene acceso Raises: HTTPException: Si no hay tenant_id, no tiene acceso o no tiene los permisos requeridos """ tenant_id = get_tenant_from_token(current_user) if not tenant_id: # Fallback para desarrollo o tokens mal formados que sí tienen el atributo pero en otro lado # Esto evita el 400 si get_tenant_from_token falla pero el usuario es válido tenant_id = current_user.get("tenant_id") if not tenant_id: raise HTTPException(status_code=400, detail="Tenant ID not found in token") if not validate_company_access(db, company_id, current_user): raise HTTPException(status_code=403, detail="Access denied to this company") # Verificar permisos si se proporcionaron if required_permissions: from api.v1.modules.core.permissions.service import PermissionService user_id = current_user.get("sub") or current_user.get("id") if not user_id: raise HTTPException(status_code=401, detail="User ID not found in token") permission_service = PermissionService(db) if require_all: has_access = permission_service.has_all_permissions( user_id=user_id, company_id=company_id, permission_codes=required_permissions, ) else: has_access = permission_service.has_any_permission( user_id=user_id, company_id=company_id, permission_codes=required_permissions, ) if not has_access: raise HTTPException( status_code=403, detail=f"Missing required permissions: {', '.join(required_permissions)}", ) return tenant_id