feat(security): implement permission checks in tenant CRUD routes and enhance API error handling

This commit is contained in:
2026-01-14 17:34:02 -06:00
parent 4f38328031
commit 86e1f7e8f6
13 changed files with 346 additions and 202 deletions

View File

@@ -176,29 +176,64 @@ def validate_company_access(
def validate_access_to_resource(
db: Session, company_id: int, current_user: Dict[str, Any]
db: Session,
company_id: int,
current_user: Dict[str, Any],
required_permissions: Optional[list[str]] = None,
require_all: bool = False,
) -> int:
"""
Valida que el usuario tenga acceso a un recurso específico basado en company_id
y regresa el tenant_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 o no tiene acceso
HTTPException: Si no hay tenant_id, no tiene acceso o no tiene los permisos requeridos
"""
tenant_id = get_tenant_from_token(current_user)
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")
# Validar que el tenant_id del usuario coincida con el del recurso
# 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