plantillas-proyectos/backend/core/security.py

"""
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

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)
) -> 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:
            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