338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""
|
|
Utilidades de seguridad y autenticación con Keycloak
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any, Dict, Optional, Set
|
|
|
|
from fastapi import Depends, HTTPException, Security
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from jose import JWTError, jwt
|
|
import httpx
|
|
from cachetools import TTLCache
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from .config import settings
|
|
from .database import get_core_db
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Cache para tokens verificados (1 minuto de TTL, máximo 1000 tokens)
|
|
token_cache = TTLCache(maxsize=1000, ttl=60)
|
|
|
|
# IDs de tenants ya sincronizados en este proceso (evita consultas repetidas)
|
|
_synced_tenant_ids: Set[int] = set()
|
|
|
|
# Security scheme
|
|
security = HTTPBearer()
|
|
|
|
|
|
async def verify_token(token: str) -> Dict[str, Any]:
|
|
"""
|
|
Verifica un token JWT llamando al Hub central.
|
|
"""
|
|
# Check cache first
|
|
if token in token_cache:
|
|
return token_cache[token]
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
response = await client.get(
|
|
f"{settings.HUB_URL}api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
user_info = response.json()
|
|
token_cache[token] = user_info
|
|
return user_info
|
|
|
|
logger.error(f"Hub token verification failed with status {response.status_code}")
|
|
raise HTTPException(status_code=401, detail="Could not validate credentials")
|
|
|
|
except httpx.HTTPError as e:
|
|
logger.error(f"Hub unreachable or error during token verification: {str(e)}")
|
|
raise HTTPException(status_code=503, detail="Authentication service unavailable")
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error during token verification: {str(e)}")
|
|
raise HTTPException(status_code=401, detail="Authentication error")
|
|
|
|
|
|
def _ensure_company_exists(db: Session, tenant_id: int, tenant_name: str) -> None:
|
|
"""
|
|
Garantiza que exista al menos una empresa en a76.company para el tenant.
|
|
El tenant IS la empresa — se crea automáticamente al primer login.
|
|
"""
|
|
try:
|
|
from api.v1.modules.a76.general_catalogs.company.models import Company
|
|
exists = db.query(Company).filter(Company.tenant_id == tenant_id).first()
|
|
if not exists:
|
|
company = Company(tenant_id=tenant_id, name=tenant_name)
|
|
db.add(company)
|
|
db.commit()
|
|
logger.info(f"Empresa creada automáticamente para tenant id={tenant_id}: '{tenant_name}'")
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.warning(f"No se pudo crear empresa automática para tenant {tenant_id}: {e}")
|
|
|
|
|
|
def _ensure_tenant_synced(db: Session, tenant_id: int, tenant_slug: str) -> None:
|
|
"""
|
|
Garantiza que el tenant del Hub exista en core.tenants local.
|
|
Se ejecuta una sola vez por tenant_id por ciclo de vida del proceso.
|
|
El Hub es la fuente de verdad — este método solo sincroniza en una dirección.
|
|
"""
|
|
if tenant_id in _synced_tenant_ids:
|
|
return
|
|
|
|
try:
|
|
# Importación local para evitar imports circulares
|
|
from api.v1.modules.core.tenants.models import Tenant, TenantType
|
|
|
|
name = " ".join(word.capitalize() for word in tenant_slug.replace("-", " ").split())
|
|
|
|
existing = db.query(Tenant).filter(Tenant.id == tenant_id).first()
|
|
if existing:
|
|
# Update name/slug if they differ (Hub is source of truth)
|
|
if existing.slug != tenant_slug or existing.name != name:
|
|
existing.slug = tenant_slug
|
|
existing.name = name
|
|
db.commit()
|
|
logger.info(f"Tenant id={tenant_id} actualizado: slug='{tenant_slug}'")
|
|
_synced_tenant_ids.add(tenant_id)
|
|
# Garantizar empresa aunque el tenant ya existiera
|
|
_ensure_company_exists(db, tenant_id, name)
|
|
return
|
|
|
|
# Crear el tenant local con los datos disponibles del token.
|
|
# El Hub siempre crea el realm de Keycloak con el mismo nombre que el slug.
|
|
tenant = Tenant(
|
|
id=tenant_id,
|
|
name=name,
|
|
slug=tenant_slug,
|
|
type=TenantType.SHARED,
|
|
keycloak_realm=tenant_slug,
|
|
is_active=True,
|
|
)
|
|
db.add(tenant)
|
|
db.commit()
|
|
_synced_tenant_ids.add(tenant_id)
|
|
logger.info(f"Tenant '{tenant_slug}' (id={tenant_id}) sincronizado desde Hub a core.tenants")
|
|
# Crear la empresa correspondiente al tenant recién sincronizado
|
|
_ensure_company_exists(db, tenant_id, name)
|
|
|
|
except IntegrityError:
|
|
# Puede ser concurrencia o colisión de slug (id diferente, mismo slug)
|
|
db.rollback()
|
|
from api.v1.modules.core.tenants.models import Tenant
|
|
# Si el slug ya existe con diferente id, el tenant real del Hub no está registrado aún.
|
|
# Logueamos el conflicto para depuración; el sistema continuará con tenant_id vacío.
|
|
stale = db.query(Tenant).filter(Tenant.slug == tenant_slug).first()
|
|
if stale and stale.id != tenant_id:
|
|
logger.error(
|
|
f"Conflicto de tenant: JWT dice id={tenant_id} slug='{tenant_slug}', "
|
|
f"pero core.tenants tiene id={stale.id} mismo slug. "
|
|
f"Elimine el registro obsoleto con: "
|
|
f"DELETE FROM core.tenants WHERE id={stale.id};"
|
|
)
|
|
else:
|
|
_synced_tenant_ids.add(tenant_id)
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.warning(f"No se pudo sincronizar tenant {tenant_id} ({tenant_slug}): {e}")
|
|
|
|
|
|
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.
|
|
Auto-sincroniza el tenant en core.tenants si fue creado en el Hub
|
|
pero aún no existe en la BD local.
|
|
|
|
Uso en FastAPI:
|
|
current_user: dict = Depends(get_current_user)
|
|
"""
|
|
token = credentials.credentials
|
|
user_info = await verify_token(token)
|
|
|
|
# Sincronizar tenant desde Hub a BD local (solo la primera vez por tenant)
|
|
tenant_id = user_info.get("tenant_id")
|
|
tenant_slug = user_info.get("tenant_slug")
|
|
if tenant_id and tenant_slug:
|
|
_ensure_tenant_synced(db, int(tenant_id), str(tenant_slug))
|
|
|
|
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
|