Implement multi-tenant permission management module
- Added Pydantic schemas for permission-related requests and responses in `schemas.py`. - Created seed data for various permissions and roles in `seed.py`. - Developed a `PermissionService` class in `service.py` to handle permission logic, including user role assignments and direct permission grants. - Implemented unit tests for permission functionalities in `test_permissions.py`, covering role assignments, permission checks, and expiration handling. - Seeded trailer types data in `trailer_types/seed.py`.
This commit is contained in:
324
backend/api/v1/modules/core/permissions/README.md
Normal file
324
backend/api/v1/modules/core/permissions/README.md
Normal file
@@ -0,0 +1,324 @@
|
||||
# Módulo de Permisos Multi-Tenant
|
||||
|
||||
Sistema completo de permisos granulares para aplicaciones multi-tenant con FastAPI y SQLAlchemy.
|
||||
|
||||
## 📁 Estructura del Módulo
|
||||
|
||||
```
|
||||
backend/api/v1/modules/core/permissions/
|
||||
├── __init__.py # Exports del módulo
|
||||
├── models.py # Modelos SQLAlchemy
|
||||
├── service.py # Lógica de negocio
|
||||
├── dependencies.py # Dependencias FastAPI
|
||||
├── schemas.py # Modelos Pydantic (request/response)
|
||||
└── routes.py # Endpoints de la API
|
||||
```
|
||||
|
||||
## 🎯 Componentes
|
||||
|
||||
### **models.py**
|
||||
|
||||
Define los modelos de base de datos:
|
||||
|
||||
- `Permission` - Permisos del sistema (ej: "invoice.view", "invoice.edit")
|
||||
- `ClientRole` - Roles personalizados por cliente
|
||||
- `RolePermission` - Relación roles-permisos
|
||||
- `UserClientRole` - Asignación usuario-rol-cliente
|
||||
- `UserClientPermission` - Permisos directos por usuario
|
||||
|
||||
### **service.py**
|
||||
|
||||
Contiene la clase `PermissionService` con métodos:
|
||||
|
||||
- `get_user_permissions()` - Obtiene todos los permisos de un usuario
|
||||
- `has_permission()` - Verifica un permiso específico
|
||||
- `has_all_permissions()` - Verifica múltiples permisos (AND)
|
||||
- `has_any_permission()` - Verifica múltiples permisos (OR)
|
||||
- `assign_role_to_user()` - Asigna roles a usuarios
|
||||
- `grant_direct_permission()` - Concede permisos directos
|
||||
|
||||
### **dependencies.py**
|
||||
|
||||
Dependencias para proteger rutas:
|
||||
|
||||
- `PermissionChecker` - Clase para verificar múltiples permisos
|
||||
- `RequirePermission` - Clase para verificar un solo permiso
|
||||
- `get_client_id()` - Extrae el ID del cliente del header
|
||||
- `get_permission_service()` - Proporciona instancia del servicio
|
||||
- `get_current_user_permissions()` - Devuelve permisos del usuario
|
||||
|
||||
### **schemas.py**
|
||||
|
||||
Modelos Pydantic para request/response:
|
||||
|
||||
- Responses: `PermissionResponse`, `ClientRoleResponse`, `UserPermissionsResponse`, etc.
|
||||
- Requests: `AssignRoleRequest`, `GrantPermissionRequest`, `CreateRoleRequest`, etc.
|
||||
|
||||
### **routes.py**
|
||||
|
||||
Endpoints de la API:
|
||||
|
||||
- `GET /permissions/me` - Permisos del usuario actual
|
||||
- `GET /permissions/available` - Lista todos los permisos
|
||||
- `GET /permissions/roles` - Lista roles del cliente
|
||||
- `POST /permissions/roles` - Crea un rol
|
||||
- `POST /permissions/assign-role` - Asigna rol a usuario
|
||||
- `POST /permissions/grant-permission` - Concede permiso directo
|
||||
- Ejemplos de rutas protegidas
|
||||
|
||||
## 🚀 Uso Rápido
|
||||
|
||||
### Importar el módulo
|
||||
|
||||
```python
|
||||
from api.v1.modules.core.permissions import (
|
||||
Permission,
|
||||
ClientRole,
|
||||
PermissionService,
|
||||
PermissionChecker,
|
||||
RequirePermission,
|
||||
router
|
||||
)
|
||||
```
|
||||
|
||||
### Registrar las rutas
|
||||
|
||||
```python
|
||||
# En backend/api/v1/router.py
|
||||
from api.v1.modules.core.permissions import router as permissions_router
|
||||
|
||||
api_router = APIRouter()
|
||||
api_router.include_router(permissions_router)
|
||||
```
|
||||
|
||||
### Proteger una ruta con permiso único
|
||||
|
||||
```python
|
||||
from fastapi import APIRouter, Depends
|
||||
from api.v1.modules.core.permissions import RequirePermission
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.get("/invoices")
|
||||
async def list_invoices(
|
||||
_: None = Depends(RequirePermission("invoice.view"))
|
||||
):
|
||||
return {"invoices": [...]}
|
||||
```
|
||||
|
||||
### Proteger con múltiples permisos
|
||||
|
||||
```python
|
||||
from api.v1.modules.core.permissions import PermissionChecker
|
||||
|
||||
@router.post("/invoices")
|
||||
async def create_invoice(
|
||||
_: None = Depends(PermissionChecker(
|
||||
["invoice.view", "invoice.create"],
|
||||
require_all=True # Requiere TODOS
|
||||
))
|
||||
):
|
||||
return {"created": True}
|
||||
```
|
||||
|
||||
### Usar permisos en la lógica
|
||||
|
||||
```python
|
||||
from api.v1.modules.core.permissions import get_current_user_permissions
|
||||
|
||||
@router.get("/dashboard")
|
||||
async def dashboard(
|
||||
permissions: set = Depends(get_current_user_permissions)
|
||||
):
|
||||
widgets = []
|
||||
|
||||
if "invoice.view" in permissions:
|
||||
widgets.append({"type": "invoices", "data": [...]})
|
||||
|
||||
return {"widgets": widgets}
|
||||
```
|
||||
|
||||
## 📊 Base de Datos
|
||||
|
||||
### Ejecutar migración
|
||||
|
||||
```bash
|
||||
cd backend
|
||||
alembic upgrade head
|
||||
```
|
||||
|
||||
Esto crea las tablas y permisos iniciales:
|
||||
|
||||
- **invoice.*** - view, create, edit, delete, approve
|
||||
- **user.*** - view, create, edit, delete
|
||||
- **report.*** - financial.view, admin.view, export
|
||||
- **roles.*** - view, create, edit, delete, assign
|
||||
- **permissions.*** - view, grant
|
||||
|
||||
## 🔐 Flujo de Autenticación
|
||||
|
||||
1. Usuario hace request con token JWT de Keycloak
|
||||
2. Header `X-Client-ID` indica el cliente/tenant
|
||||
3. Sistema extrae `user_id` del token
|
||||
4. Consulta permisos del usuario en ese cliente
|
||||
5. Valida si tiene el permiso requerido
|
||||
6. Devuelve 200 OK o 403 Forbidden
|
||||
|
||||
## 💡 Ejemplos Prácticos
|
||||
|
||||
### Crear un rol personalizado
|
||||
|
||||
```python
|
||||
from api.v1.modules.core.permissions import PermissionService
|
||||
from core.database import get_db
|
||||
|
||||
db = next(get_db())
|
||||
service = PermissionService(db)
|
||||
|
||||
# Crear rol
|
||||
role = ClientRole(
|
||||
client_id=1,
|
||||
name="Contador",
|
||||
code="accountant",
|
||||
description="Acceso a módulo contable"
|
||||
)
|
||||
db.add(role)
|
||||
db.commit()
|
||||
```
|
||||
|
||||
### Asignar permisos a un rol
|
||||
|
||||
```python
|
||||
from api.v1.modules.core.permissions.models import RolePermission
|
||||
|
||||
# Obtener permisos de facturación
|
||||
invoice_perms = db.query(Permission).filter(
|
||||
Permission.module == "invoice"
|
||||
).all()
|
||||
|
||||
# Asignar al rol
|
||||
for perm in invoice_perms:
|
||||
role_perm = RolePermission(
|
||||
client_role_id=role.id,
|
||||
permission_id=perm.id
|
||||
)
|
||||
db.add(role_perm)
|
||||
|
||||
db.commit()
|
||||
```
|
||||
|
||||
### Asignar rol a usuario
|
||||
|
||||
```python
|
||||
service.assign_role_to_user(
|
||||
user_id="user-uuid-from-keycloak",
|
||||
client_id=1,
|
||||
role_id=role.id,
|
||||
assigned_by="admin-uuid"
|
||||
)
|
||||
```
|
||||
|
||||
### Conceder permiso temporal
|
||||
|
||||
```python
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
service.grant_direct_permission(
|
||||
user_id="user-uuid",
|
||||
client_id=1,
|
||||
permission_code="invoice.delete",
|
||||
assigned_by="admin-uuid",
|
||||
expires_at=datetime.utcnow() + timedelta(days=7)
|
||||
)
|
||||
```
|
||||
|
||||
## ⚡ Optimización de Rendimiento
|
||||
|
||||
### 1. Caché con Redis
|
||||
|
||||
```python
|
||||
import redis
|
||||
from functools import lru_cache
|
||||
|
||||
redis_client = redis.Redis(host='localhost', port=6379)
|
||||
|
||||
def get_cached_permissions(user_id: str, client_id: int) -> set:
|
||||
cache_key = f"perms:{user_id}:{client_id}"
|
||||
|
||||
cached = redis_client.get(cache_key)
|
||||
if cached:
|
||||
return set(cached.decode().split(','))
|
||||
|
||||
# Consultar DB
|
||||
service = PermissionService(db)
|
||||
permissions = service.get_user_permissions(user_id, client_id)
|
||||
|
||||
# Cachear por 5 minutos
|
||||
redis_client.setex(cache_key, 300, ','.join(permissions))
|
||||
|
||||
return permissions
|
||||
```
|
||||
|
||||
### 2. Índices de Base de Datos
|
||||
|
||||
Ya están definidos en los modelos:
|
||||
|
||||
- Índices compuestos para consultas eficientes
|
||||
- Índices únicos para prevenir duplicados
|
||||
- Índices en foreign keys
|
||||
|
||||
### 3. Query Optimization
|
||||
|
||||
El servicio usa JOINs eficientes en lugar de N+1 queries.
|
||||
|
||||
## 🧪 Testing
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from api.v1.modules.core.permissions import PermissionService
|
||||
from api.v1.modules.core.permissions.models import Permission, ClientRole
|
||||
|
||||
def test_user_has_permission_from_role(db_session):
|
||||
# Setup
|
||||
perm = Permission(code="invoice.view", module="invoice", action="view")
|
||||
db_session.add(perm)
|
||||
|
||||
role = ClientRole(client_id=1, code="viewer", name="Viewer")
|
||||
db_session.add(role)
|
||||
db_session.commit()
|
||||
|
||||
# Test
|
||||
service = PermissionService(db_session)
|
||||
assert service.has_permission("user-123", 1, "invoice.view")
|
||||
```
|
||||
|
||||
## 📝 Notas Importantes
|
||||
|
||||
- **Client ID**: Por defecto se obtiene del header `X-Client-ID`, pero puede adaptarse a subdominios o JWT
|
||||
- **User ID**: Se extrae del campo `sub` del token JWT de Keycloak
|
||||
- **Permisos Directos**: Pueden revocar permisos heredados de roles (`is_granted=False`)
|
||||
- **Soft Delete**: Los roles y permisos se desactivan (`is_active=False`) en lugar de eliminarse
|
||||
|
||||
## 🔗 Integración con Keycloak
|
||||
|
||||
Los roles globales de Keycloak pueden coexistir con los roles locales:
|
||||
|
||||
```python
|
||||
@router.get("/protected")
|
||||
async def protected_route(
|
||||
current_user: dict = Depends(get_current_user),
|
||||
permissions: set = Depends(get_current_user_permissions)
|
||||
):
|
||||
# Verificar rol global de Keycloak
|
||||
keycloak_roles = current_user.get("realm_access", {}).get("roles", [])
|
||||
|
||||
if "super_admin" in keycloak_roles:
|
||||
# Super admin tiene acceso total
|
||||
return {"access": "granted", "level": "global"}
|
||||
|
||||
# Verificar permisos a nivel de cliente
|
||||
if "invoice.view" in permissions:
|
||||
return {"access": "granted", "level": "client"}
|
||||
|
||||
raise HTTPException(403, "No access")
|
||||
```
|
||||
38
backend/api/v1/modules/core/permissions/__init__.py
Normal file
38
backend/api/v1/modules/core/permissions/__init__.py
Normal file
@@ -0,0 +1,38 @@
|
||||
"""
|
||||
Módulo de permisos multi-tenant.
|
||||
Proporciona modelos, servicios y rutas para gestión de permisos granulares por companye.
|
||||
"""
|
||||
|
||||
from .models import (
|
||||
Permission,
|
||||
CompanyRole,
|
||||
RolePermission,
|
||||
UserCompanyRole,
|
||||
UserCompanyPermission,
|
||||
)
|
||||
from .service import PermissionService
|
||||
from .dependencies import (
|
||||
PermissionChecker,
|
||||
RequirePermission,
|
||||
get_permission_service,
|
||||
get_current_user_permissions,
|
||||
)
|
||||
from .routes import router
|
||||
|
||||
__all__ = [
|
||||
# Models
|
||||
"Permission",
|
||||
"CompanyRole",
|
||||
"RolePermission",
|
||||
"UserCompanyRole",
|
||||
"UserCompanyPermission",
|
||||
# Service
|
||||
"PermissionService",
|
||||
# Dependencies
|
||||
"PermissionChecker",
|
||||
"RequirePermission",
|
||||
"get_permission_service",
|
||||
"get_current_user_permissions",
|
||||
# Router
|
||||
"router",
|
||||
]
|
||||
231
backend/api/v1/modules/core/permissions/dependencies.py
Normal file
231
backend/api/v1/modules/core/permissions/dependencies.py
Normal file
@@ -0,0 +1,231 @@
|
||||
"""
|
||||
Dependencias de FastAPI para verificación de permisos multi-tenant.
|
||||
Proporciona decoradores y funciones para proteger rutas con permisos específicos.
|
||||
"""
|
||||
|
||||
from typing import List, Optional, Callable
|
||||
from fastapi import Depends, HTTPException, status, Header
|
||||
from sqlalchemy.orm import Session
|
||||
from functools import wraps
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user # Asumiendo que existe esta función
|
||||
from .service import PermissionService
|
||||
|
||||
|
||||
# Dependencia para obtener el ID del cliente del header o contexto
|
||||
async def get_client_id(
|
||||
x_client_id: Optional[str] = Header(None, alias="X-Client-ID")
|
||||
) -> int:
|
||||
"""
|
||||
Obtiene el ID del cliente desde el header de la petición.
|
||||
|
||||
En producción, esto podría obtenerse de:
|
||||
- Un header HTTP (X-Client-ID)
|
||||
- Un subdomain (cliente1.miapp.com)
|
||||
- El token JWT del usuario
|
||||
- La sesión del usuario
|
||||
"""
|
||||
if not x_client_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Client ID is required. Provide X-Client-ID header.",
|
||||
)
|
||||
|
||||
try:
|
||||
return int(x_client_id)
|
||||
except ValueError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid client ID format"
|
||||
)
|
||||
|
||||
|
||||
# Dependencia para obtener el servicio de permisos
|
||||
def get_permission_service(db: Session = Depends(get_core_db)) -> PermissionService:
|
||||
"""
|
||||
Crea una instancia del servicio de permisos con la sesión de base de datos.
|
||||
"""
|
||||
return PermissionService(db)
|
||||
|
||||
|
||||
# Clase para verificación de permisos (puede usarse como dependencia)
|
||||
class PermissionChecker:
|
||||
"""
|
||||
Verificador de permisos que puede usarse como dependencia de FastAPI.
|
||||
|
||||
Ejemplo de uso:
|
||||
@app.get("/invoices")
|
||||
async def list_invoices(
|
||||
_: None = Depends(PermissionChecker(["invoice.view"]))
|
||||
):
|
||||
return {"invoices": [...]}
|
||||
"""
|
||||
|
||||
def __init__(self, required_permissions: List[str], require_all: bool = True):
|
||||
"""
|
||||
Args:
|
||||
required_permissions: Lista de permisos requeridos
|
||||
require_all: Si True, requiere TODOS los permisos.
|
||||
Si False, requiere AL MENOS UNO.
|
||||
"""
|
||||
self.required_permissions = required_permissions
|
||||
self.require_all = require_all
|
||||
|
||||
async def __call__(
|
||||
self,
|
||||
current_user: dict = Depends(get_current_user),
|
||||
client_id: int = Depends(get_client_id),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
):
|
||||
"""
|
||||
Verifica que el usuario tenga los permisos requeridos.
|
||||
|
||||
Lanza HTTPException 403 si no tiene permisos.
|
||||
"""
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
if not user_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="User ID not found in token",
|
||||
)
|
||||
|
||||
# Verificar permisos
|
||||
if self.require_all:
|
||||
has_access = permission_service.has_all_permissions(
|
||||
user_id=user_id,
|
||||
client_id=client_id,
|
||||
permission_codes=self.required_permissions,
|
||||
)
|
||||
else:
|
||||
has_access = permission_service.has_any_permission(
|
||||
user_id=user_id,
|
||||
client_id=client_id,
|
||||
permission_codes=self.required_permissions,
|
||||
)
|
||||
|
||||
if not has_access:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Missing required permissions: {', '.join(self.required_permissions)}",
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# Función alternativa para verificar un solo permiso
|
||||
class RequirePermission:
|
||||
"""
|
||||
Verificador simple para un único permiso.
|
||||
|
||||
Ejemplo:
|
||||
@app.post("/invoices")
|
||||
async def create_invoice(
|
||||
_: None = Depends(RequirePermission("invoice.create"))
|
||||
):
|
||||
return {"created": True}
|
||||
"""
|
||||
|
||||
def __init__(self, permission_code: str):
|
||||
self.permission_code = permission_code
|
||||
|
||||
async def __call__(
|
||||
self,
|
||||
current_user: dict = Depends(get_current_user),
|
||||
client_id: int = Depends(get_client_id),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
):
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
if not user_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="User ID not found in token",
|
||||
)
|
||||
|
||||
has_permission = permission_service.has_permission(
|
||||
user_id=user_id, client_id=client_id, permission_code=self.permission_code
|
||||
)
|
||||
|
||||
if not has_permission:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Missing required permission: {self.permission_code}",
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# Decorador personalizado para aplicar a funciones (opcional)
|
||||
def require_permissions(*permissions: str, require_all: bool = True):
|
||||
"""
|
||||
Decorador para verificar permisos en funciones.
|
||||
Útil para lógica de negocio fuera de rutas FastAPI.
|
||||
|
||||
Ejemplo:
|
||||
@require_permissions("invoice.edit", "invoice.view")
|
||||
def update_invoice_logic(invoice_id: int, user_id: str, client_id: int, db: Session):
|
||||
# Lógica de actualización
|
||||
pass
|
||||
"""
|
||||
|
||||
def decorator(func: Callable):
|
||||
@wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
# Extraer user_id, client_id y db de los argumentos
|
||||
user_id = kwargs.get("user_id")
|
||||
client_id = kwargs.get("client_id")
|
||||
db = kwargs.get("db")
|
||||
|
||||
if not all([user_id, client_id, db]):
|
||||
raise ValueError(
|
||||
"Function must receive 'user_id', 'client_id', and 'db' as keyword arguments"
|
||||
)
|
||||
|
||||
# Verificar permisos
|
||||
permission_service = PermissionService(db)
|
||||
|
||||
if require_all:
|
||||
has_access = permission_service.has_all_permissions(
|
||||
user_id=user_id,
|
||||
client_id=client_id,
|
||||
permission_codes=list(permissions),
|
||||
)
|
||||
else:
|
||||
has_access = permission_service.has_any_permission(
|
||||
user_id=user_id,
|
||||
client_id=client_id,
|
||||
permission_codes=list(permissions),
|
||||
)
|
||||
|
||||
if not has_access:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Missing required permissions: {', '.join(permissions)}",
|
||||
)
|
||||
|
||||
return func(*args, **kwargs)
|
||||
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# Función helper para obtener permisos del usuario actual
|
||||
async def get_current_user_permissions(
|
||||
current_user: dict = Depends(get_current_user),
|
||||
client_id: int = Depends(get_client_id),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
) -> set:
|
||||
"""
|
||||
Devuelve todos los permisos del usuario actual en el cliente.
|
||||
Útil para endpoints que necesitan conocer los permisos disponibles.
|
||||
"""
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
if not user_id:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="User ID not found in token",
|
||||
)
|
||||
|
||||
return permission_service.get_user_permissions(user_id, client_id)
|
||||
153
backend/api/v1/modules/core/permissions/integration_example.py
Normal file
153
backend/api/v1/modules/core/permissions/integration_example.py
Normal file
@@ -0,0 +1,153 @@
|
||||
"""
|
||||
Ejemplo de integración del módulo de permisos en el router principal.
|
||||
Este archivo muestra cómo incluir las rutas de permisos en tu API.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from api.v1.modules.core.permissions import router as permissions_router
|
||||
|
||||
# Crear el router principal de la API v1
|
||||
api_v1_router = APIRouter(prefix="/api/v1")
|
||||
|
||||
# ============================================================================
|
||||
# REGISTRAR EL MÓDULO DE PERMISOS
|
||||
# ============================================================================
|
||||
|
||||
# Opción 1: Incluir directamente con el prefix ya definido en el módulo
|
||||
api_v1_router.include_router(permissions_router)
|
||||
# Esto creará rutas como: /api/v1/permissions/me, /api/v1/permissions/roles, etc.
|
||||
|
||||
|
||||
# Opción 2: Sobrescribir el prefix si lo necesitas
|
||||
# api_v1_router.include_router(
|
||||
# permissions_router,
|
||||
# prefix="/permisos", # Cambiar a español si prefieres
|
||||
# tags=["Permisos"]
|
||||
# )
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# IMPORTAR DEPENDENCIAS PARA USAR EN OTRAS RUTAS
|
||||
# ============================================================================
|
||||
|
||||
from api.v1.modules.core.permissions import (
|
||||
RequirePermission,
|
||||
PermissionChecker,
|
||||
get_current_user_permissions,
|
||||
PermissionService
|
||||
)
|
||||
|
||||
# ============================================================================
|
||||
# EJEMPLO: USAR PERMISOS EN OTROS MÓDULOS
|
||||
# ============================================================================
|
||||
|
||||
# Supongamos que tienes un módulo de facturas
|
||||
invoices_router = APIRouter(prefix="/invoices", tags=["Invoices"])
|
||||
|
||||
@invoices_router.get("/")
|
||||
async def list_invoices(
|
||||
_: None = Depends(RequirePermission("invoice.view"))
|
||||
):
|
||||
"""Lista todas las facturas (requiere permiso invoice.view)."""
|
||||
return {
|
||||
"invoices": [
|
||||
{"id": 1, "number": "INV-001", "amount": 1000},
|
||||
{"id": 2, "number": "INV-002", "amount": 2000}
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@invoices_router.post("/")
|
||||
async def create_invoice(
|
||||
_: None = Depends(PermissionChecker(["invoice.view", "invoice.create"], require_all=True))
|
||||
):
|
||||
"""Crea una factura (requiere invoice.view + invoice.create)."""
|
||||
return {"message": "Invoice created", "id": 3}
|
||||
|
||||
|
||||
@invoices_router.delete("/{invoice_id}")
|
||||
async def delete_invoice(
|
||||
invoice_id: int,
|
||||
_: None = Depends(RequirePermission("invoice.delete"))
|
||||
):
|
||||
"""Elimina una factura (requiere invoice.delete)."""
|
||||
return {"message": f"Invoice {invoice_id} deleted"}
|
||||
|
||||
|
||||
# Registrar el router de facturas
|
||||
api_v1_router.include_router(invoices_router)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# EJEMPLO: RUTA CON LÓGICA CONDICIONAL SEGÚN PERMISOS
|
||||
# ============================================================================
|
||||
|
||||
reports_router = APIRouter(prefix="/reports", tags=["Reports"])
|
||||
|
||||
@reports_router.get("/dashboard")
|
||||
async def get_reports_dashboard(
|
||||
user_permissions: set = Depends(get_current_user_permissions)
|
||||
):
|
||||
"""
|
||||
Dashboard de reportes que muestra diferentes secciones según los permisos.
|
||||
No requiere permisos específicos, pero el contenido varía.
|
||||
"""
|
||||
sections = []
|
||||
|
||||
if "report.financial.view" in user_permissions:
|
||||
sections.append({
|
||||
"type": "financial",
|
||||
"title": "Reportes Financieros",
|
||||
"reports": ["Balance General", "Estado de Resultados", "Flujo de Efectivo"]
|
||||
})
|
||||
|
||||
if "report.admin.view" in user_permissions:
|
||||
sections.append({
|
||||
"type": "administrative",
|
||||
"title": "Reportes Administrativos",
|
||||
"reports": ["Actividad de Usuarios", "Auditoría", "Logs del Sistema"]
|
||||
})
|
||||
|
||||
if "invoice.view" in user_permissions:
|
||||
sections.append({
|
||||
"type": "invoices",
|
||||
"title": "Reportes de Facturación",
|
||||
"reports": ["Facturas por Mes", "Facturas Pendientes", "Análisis de Ventas"]
|
||||
})
|
||||
|
||||
return {
|
||||
"user_permissions": list(user_permissions),
|
||||
"available_sections": sections
|
||||
}
|
||||
|
||||
|
||||
# Registrar el router de reportes
|
||||
api_v1_router.include_router(reports_router)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# EXPORTAR EL ROUTER PRINCIPAL
|
||||
# ============================================================================
|
||||
|
||||
__all__ = ["api_v1_router"]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# USO EN main.py
|
||||
# ============================================================================
|
||||
"""
|
||||
En tu backend/main.py, incluye el router así:
|
||||
|
||||
from fastapi import FastAPI
|
||||
from api.v1.integration_example import api_v1_router
|
||||
|
||||
app = FastAPI(title="Mi API Multi-Tenant")
|
||||
|
||||
# Registrar todas las rutas de API v1
|
||||
app.include_router(api_v1_router)
|
||||
|
||||
# Ahora tendrás:
|
||||
# - /api/v1/permissions/* (rutas de permisos)
|
||||
# - /api/v1/invoices/* (rutas de facturas protegidas)
|
||||
# - /api/v1/reports/* (rutas de reportes)
|
||||
"""
|
||||
247
backend/api/v1/modules/core/permissions/models.py
Normal file
247
backend/api/v1/modules/core/permissions/models.py
Normal file
@@ -0,0 +1,247 @@
|
||||
"""
|
||||
Modelos de permisos multi-tenant para el sistema.
|
||||
Este módulo define el sistema de permisos granular por compañia/tenant.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
from sqlalchemy import (
|
||||
String,
|
||||
Integer,
|
||||
ForeignKey,
|
||||
DateTime,
|
||||
Boolean,
|
||||
UniqueConstraint,
|
||||
Index,
|
||||
)
|
||||
from sqlalchemy.orm import relationship, Mapped, mapped_column
|
||||
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
|
||||
from core.database import Base
|
||||
|
||||
|
||||
# Modelo para permisos del sistema
|
||||
# Representa acciones específicas como "invoice.view", "invoice.edit", etc.
|
||||
class Permission(Base, TimestampMixin):
|
||||
__tablename__ = "permissions"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
code: Mapped[str] = mapped_column(
|
||||
String(100), unique=True, nullable=False, index=True
|
||||
)
|
||||
# Código único del permiso (ej: "invoice.view", "user.edit")
|
||||
|
||||
description: Mapped[Optional[str]] = mapped_column(String(255))
|
||||
# Descripción legible del permiso
|
||||
|
||||
module: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
|
||||
# Módulo al que pertenece (ej: "invoice", "user", "report")
|
||||
|
||||
action: Mapped[str] = mapped_column(String(50), nullable=False)
|
||||
# Acción específica (ej: "view", "edit", "delete", "create")
|
||||
|
||||
is_active: Mapped[bool] = mapped_column(
|
||||
Boolean, default=True, server_default="true", nullable=False
|
||||
)
|
||||
# Permite desactivar permisos sin eliminarlos
|
||||
|
||||
__table_args__ = {"schema": "core", "extend_existing": True}
|
||||
|
||||
# Relaciones
|
||||
role_permissions: Mapped[list["RolePermission"]] = relationship(
|
||||
"RolePermission", back_populates="permission", cascade="all, delete-orphan"
|
||||
)
|
||||
user_company_permissions: Mapped[list["UserCompanyPermission"]] = relationship(
|
||||
"UserCompanyPermission",
|
||||
back_populates="permission",
|
||||
cascade="all, delete-orphan",
|
||||
)
|
||||
|
||||
|
||||
# Modelo para roles personalizados por compañia/tenant
|
||||
# Cada compañia puede definir sus propios roles con nombres personalizados
|
||||
class CompanyRole(Base, TenantScopedMixin, TimestampMixin):
|
||||
__tablename__ = "company_roles"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
|
||||
name: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
# Nombre del rol (ej: "Administrador", "Contador", "Vendedor")
|
||||
|
||||
code: Mapped[str] = mapped_column(String(100), nullable=False)
|
||||
# Código único del rol dentro del compañia (ej: "admin", "accountant")
|
||||
|
||||
description: Mapped[Optional[str]] = mapped_column(String(255))
|
||||
# Descripción del rol
|
||||
|
||||
is_active: Mapped[bool] = mapped_column(
|
||||
Boolean, default=True, server_default="true", nullable=False
|
||||
)
|
||||
# Permite desactivar roles sin eliminarlos
|
||||
|
||||
# Restricción: el código del rol debe ser único por compañia
|
||||
__table_args__ = (
|
||||
UniqueConstraint(
|
||||
"company_id", "tenant_id", "code", name="uq_company_role_code"
|
||||
),
|
||||
Index(
|
||||
"ix_company_roles_company_id_is_active",
|
||||
"company_id",
|
||||
"tenant_id",
|
||||
"is_active",
|
||||
),
|
||||
{"schema": "core", "extend_existing": True},
|
||||
)
|
||||
|
||||
# Relaciones
|
||||
role_permissions: Mapped[list["RolePermission"]] = relationship(
|
||||
"RolePermission", back_populates="company_role", cascade="all, delete-orphan"
|
||||
)
|
||||
user_company_roles: Mapped[list["UserCompanyRole"]] = relationship(
|
||||
"UserCompanyRole",
|
||||
back_populates="company_role",
|
||||
cascade="all, delete-orphan",
|
||||
)
|
||||
|
||||
|
||||
# Tabla de relación entre roles de compañia y permisos
|
||||
# Define qué permisos tiene cada rol
|
||||
class RolePermission(Base, TenantScopedMixin, TimestampMixin):
|
||||
__tablename__ = "role_permissions"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
|
||||
company_role_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("core.company_roles.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
|
||||
permission_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("core.permissions.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
|
||||
# Restricción: un permiso no puede estar duplicado en el mismo rol
|
||||
__table_args__ = (
|
||||
UniqueConstraint("company_role_id", "permission_id", name="uq_role_permission"),
|
||||
Index("ix_role_permissions_composite", "company_role_id", "permission_id"),
|
||||
{"schema": "core", "extend_existing": True},
|
||||
)
|
||||
|
||||
# Relaciones
|
||||
company_role: Mapped["CompanyRole"] = relationship(
|
||||
"CompanyRole", back_populates="role_permissions"
|
||||
)
|
||||
permission: Mapped["Permission"] = relationship(
|
||||
"Permission", back_populates="role_permissions"
|
||||
)
|
||||
|
||||
|
||||
# Tabla de relación entre usuarios y roles de compañia
|
||||
# Define qué roles tiene cada usuario en cada compañia
|
||||
class UserCompanyRole(Base, TenantScopedMixin, TimestampMixin):
|
||||
__tablename__ = "user_company_roles"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
|
||||
user_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
|
||||
# ID del usuario (puede ser UUID de Keycloak u otro identificador)
|
||||
|
||||
company_role_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("core.company_roles.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
|
||||
is_active: Mapped[bool] = mapped_column(
|
||||
Boolean, default=True, server_default="true", nullable=False
|
||||
)
|
||||
# Permite desactivar asignaciones sin eliminarlas
|
||||
|
||||
assigned_by: Mapped[Optional[str]] = mapped_column(String(100))
|
||||
# ID del usuario que asignó este rol
|
||||
|
||||
# Restricción: un usuario no puede tener el mismo rol duplicado en un compañia
|
||||
__table_args__ = (
|
||||
UniqueConstraint(
|
||||
"user_id",
|
||||
"company_id",
|
||||
"tenant_id",
|
||||
"company_role_id",
|
||||
name="uq_user_company_role",
|
||||
),
|
||||
Index(
|
||||
"ix_user_company_roles_user_company",
|
||||
"user_id",
|
||||
"company_id",
|
||||
"tenant_id",
|
||||
"is_active",
|
||||
),
|
||||
{"schema": "core", "extend_existing": True},
|
||||
)
|
||||
|
||||
# Relaciones
|
||||
company_role: Mapped["CompanyRole"] = relationship(
|
||||
"CompanyRole", back_populates="user_company_roles"
|
||||
)
|
||||
|
||||
|
||||
# Tabla para permisos directos de usuario por compañia (opcional)
|
||||
# Permite asignar permisos específicos a un usuario sin necesidad de un rol
|
||||
# Útil para casos excepcionales o permisos temporales
|
||||
class UserCompanyPermission(Base, TenantScopedMixin, TimestampMixin):
|
||||
__tablename__ = "user_company_permissions"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
|
||||
user_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
|
||||
# ID del usuario
|
||||
|
||||
permission_id: Mapped[int] = mapped_column(
|
||||
Integer,
|
||||
ForeignKey("core.permissions.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
|
||||
is_granted: Mapped[bool] = mapped_column(
|
||||
Boolean, default=True, server_default="true", nullable=False
|
||||
)
|
||||
# True = permiso concedido, False = permiso revocado explícitamente
|
||||
# Permite revocar permisos que vienen de roles
|
||||
|
||||
is_active: Mapped[bool] = mapped_column(
|
||||
Boolean, default=True, server_default="true", nullable=False
|
||||
)
|
||||
|
||||
assigned_by: Mapped[Optional[str]] = mapped_column(String(100))
|
||||
expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime)
|
||||
# Permite permisos temporales con fecha de expiración
|
||||
|
||||
# Restricción: un usuario no puede tener el mismo permiso duplicado en un compañia
|
||||
__table_args__ = (
|
||||
UniqueConstraint(
|
||||
"user_id",
|
||||
"company_id",
|
||||
"tenant_id",
|
||||
"permission_id",
|
||||
name="uq_user_company_permission",
|
||||
),
|
||||
Index(
|
||||
"ix_user_company_permissions_composite",
|
||||
"user_id",
|
||||
"company_id",
|
||||
"tenant_id",
|
||||
"is_active",
|
||||
),
|
||||
{"schema": "core", "extend_existing": True},
|
||||
)
|
||||
|
||||
# Relaciones
|
||||
permission: Mapped["Permission"] = relationship(
|
||||
"Permission", back_populates="user_company_permissions"
|
||||
)
|
||||
445
backend/api/v1/modules/core/permissions/routes.py
Normal file
445
backend/api/v1/modules/core/permissions/routes.py
Normal file
@@ -0,0 +1,445 @@
|
||||
"""
|
||||
Rutas API para el sistema de permisos multi-tenant.
|
||||
Endpoints para gestión de permisos, roles y asignaciones.
|
||||
"""
|
||||
|
||||
from typing import List, Optional, Dict, Any
|
||||
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, validate_access_to_resource
|
||||
from .dependencies import (
|
||||
PermissionChecker,
|
||||
RequirePermission,
|
||||
get_permission_service,
|
||||
get_current_user_permissions,
|
||||
)
|
||||
from .service import PermissionService
|
||||
from .models import Permission, CompanyRole
|
||||
from .schemas import (
|
||||
PermissionResponse,
|
||||
CompanyRoleResponse,
|
||||
UserPermissionsResponse,
|
||||
AssignRoleRequest,
|
||||
GrantPermissionRequest,
|
||||
SuccessResponse,
|
||||
CreateRoleRequest,
|
||||
UpdateRoleRequest,
|
||||
AssignPermissionsToRoleRequest,
|
||||
PermissionListResponse,
|
||||
RoleListResponse,
|
||||
)
|
||||
|
||||
|
||||
router = APIRouter(prefix="/permissions", tags=["permissions"])
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# RUTAS DE CONSULTA DE PERMISOS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("/me", response_model=UserPermissionsResponse)
|
||||
async def get_my_permissions(
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
):
|
||||
"""
|
||||
Obtiene los permisos y roles del usuario actual en el company actual.
|
||||
No requiere permisos especiales ya que es información propia.
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
user_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
# Obtener permisos
|
||||
permissions = permission_service.get_user_permissions(user_id, company_id)
|
||||
|
||||
# Obtener roles
|
||||
roles = permission_service.get_user_roles(user_id, company_id)
|
||||
role_names = [role.name for role in roles]
|
||||
|
||||
return UserPermissionsResponse(
|
||||
user_id=user_id,
|
||||
company_id=company_id,
|
||||
permissions=list(permissions),
|
||||
roles=role_names,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/available", response_model=PermissionListResponse)
|
||||
async def list_available_permissions(
|
||||
db: Session = Depends(get_core_db),
|
||||
page: int = Query(1, ge=1, description="Número de página"),
|
||||
page_size: int = Query(100, ge=1, le=500, description="Tamaño de página"),
|
||||
module: Optional[str] = Query(None, description="Filtrar por módulo"),
|
||||
_: None = Depends(RequirePermission("permissions.view")),
|
||||
):
|
||||
"""
|
||||
Lista todos los permisos disponibles en el sistema.
|
||||
Requiere permiso: permissions.view
|
||||
"""
|
||||
query = db.query(Permission).filter(Permission.is_active == True)
|
||||
|
||||
if module:
|
||||
query = query.filter(Permission.module == module)
|
||||
|
||||
total = query.count()
|
||||
permissions = query.offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return PermissionListResponse(
|
||||
items=permissions, total=total, page=page, page_size=page_size
|
||||
)
|
||||
|
||||
|
||||
@router.get("/roles", response_model=RoleListResponse)
|
||||
async def list_company_roles(
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
page: int = Query(1, ge=1, description="Número de página"),
|
||||
page_size: int = Query(100, ge=1, le=500, description="Tamaño de página"),
|
||||
_: None = Depends(RequirePermission("roles.view")),
|
||||
):
|
||||
"""
|
||||
Lista todos los roles del companye actual.
|
||||
Requiere permiso: roles.view
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
query = db.query(CompanyRole).filter(
|
||||
CompanyRole.company_id == company_id, CompanyRole.is_active == True
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
roles = query.offset((page - 1) * page_size).limit(page_size).all()
|
||||
|
||||
return RoleListResponse(items=roles, total=total, page=page, page_size=page_size)
|
||||
|
||||
|
||||
@router.get("/users/{user_id}", response_model=UserPermissionsResponse)
|
||||
async def get_user_permissions(
|
||||
user_id: str,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
_: None = Depends(RequirePermission("user.view")),
|
||||
):
|
||||
"""
|
||||
Obtiene los permisos y roles de un usuario específico.
|
||||
Requiere permiso: user.view
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
permissions = permission_service.get_user_permissions(user_id, company_id)
|
||||
roles = permission_service.get_user_roles(user_id, company_id)
|
||||
role_names = [role.name for role in roles]
|
||||
|
||||
return UserPermissionsResponse(
|
||||
user_id=user_id,
|
||||
company_id=company_id,
|
||||
permissions=list(permissions),
|
||||
roles=role_names,
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# RUTAS DE GESTIÓN DE ROLES
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.post(
|
||||
"/roles", response_model=CompanyRoleResponse, status_code=status.HTTP_201_CREATED
|
||||
)
|
||||
async def create_role(
|
||||
request: CreateRoleRequest,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
_: None = Depends(RequirePermission("roles.create")),
|
||||
):
|
||||
"""
|
||||
Crea un nuevo rol personalizado para el companye.
|
||||
Requiere permiso: roles.create
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
# Verificar que el código no esté en uso
|
||||
existing = (
|
||||
db.query(CompanyRole)
|
||||
.filter(CompanyRole.company_id == company_id, CompanyRole.code == request.code)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"Role with code '{request.code}' already exists for this company",
|
||||
)
|
||||
|
||||
# Crear el rol
|
||||
role = CompanyRole(
|
||||
company_id=company_id,
|
||||
name=request.name,
|
||||
code=request.code,
|
||||
description=request.description,
|
||||
)
|
||||
|
||||
db.add(role)
|
||||
db.commit()
|
||||
db.refresh(role)
|
||||
|
||||
# Asignar permisos si se especificaron
|
||||
if request.permission_ids:
|
||||
from .models import RolePermission
|
||||
|
||||
for perm_id in request.permission_ids:
|
||||
role_perm = RolePermission(company_role_id=role.id, permission_id=perm_id)
|
||||
db.add(role_perm)
|
||||
db.commit()
|
||||
|
||||
return role
|
||||
|
||||
|
||||
@router.patch("/roles/{role_id}", response_model=CompanyRoleResponse)
|
||||
async def update_role(
|
||||
role_id: int,
|
||||
request: UpdateRoleRequest,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
_: None = Depends(RequirePermission("roles.edit")),
|
||||
):
|
||||
"""
|
||||
Actualiza un rol existente.
|
||||
Requiere permiso: roles.edit
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
role = (
|
||||
db.query(CompanyRole)
|
||||
.filter(CompanyRole.id == role_id, CompanyRole.company_id == company_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not role:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
|
||||
)
|
||||
|
||||
# Actualizar campos
|
||||
if request.name is not None:
|
||||
role.name = request.name
|
||||
if request.description is not None:
|
||||
role.description = request.description
|
||||
if request.is_active is not None:
|
||||
role.is_active = request.is_active
|
||||
|
||||
db.commit()
|
||||
db.refresh(role)
|
||||
|
||||
return role
|
||||
|
||||
|
||||
@router.delete("/roles/{role_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_role(
|
||||
role_id: int,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
_: None = Depends(RequirePermission("roles.delete")),
|
||||
):
|
||||
"""
|
||||
Elimina (desactiva) un rol.
|
||||
Requiere permiso: roles.delete
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
role = (
|
||||
db.query(CompanyRole)
|
||||
.filter(CompanyRole.id == role_id, CompanyRole.company_id == company_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not role:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND, detail="Role not found"
|
||||
)
|
||||
|
||||
role.is_active = False
|
||||
db.commit()
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# RUTAS DE ASIGNACIÓN DE ROLES Y PERMISOS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.post(
|
||||
"/assign-role", response_model=SuccessResponse, status_code=status.HTTP_201_CREATED
|
||||
)
|
||||
async def assign_role(
|
||||
request: AssignRoleRequest,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
_: None = Depends(RequirePermission("roles.assign")),
|
||||
):
|
||||
"""
|
||||
Asigna un rol a un usuario en el companye actual.
|
||||
Requiere permiso: roles.assign
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
assigner_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
try:
|
||||
user_role = permission_service.assign_role_to_user(
|
||||
user_id=request.user_id,
|
||||
company_id=company_id,
|
||||
role_id=request.role_id,
|
||||
assigned_by=assigner_id,
|
||||
)
|
||||
|
||||
return SuccessResponse(
|
||||
message=f"Role assigned to user {request.user_id}",
|
||||
data={"assignment_id": user_role.id},
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
||||
|
||||
|
||||
@router.post(
|
||||
"/grant-permission",
|
||||
response_model=SuccessResponse,
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
async def grant_permission(
|
||||
request: GrantPermissionRequest,
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
permission_service: PermissionService = Depends(get_permission_service),
|
||||
_: None = Depends(RequirePermission("permissions.grant")),
|
||||
):
|
||||
"""
|
||||
Concede un permiso directo a un usuario en el companye actual.
|
||||
Requiere permiso: permissions.grant
|
||||
"""
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
|
||||
assigner_id = current_user.get("sub") or current_user.get("id")
|
||||
|
||||
try:
|
||||
user_permission = permission_service.grant_direct_permission(
|
||||
user_id=request.user_id,
|
||||
company_id=company_id,
|
||||
permission_code=request.permission_code,
|
||||
assigned_by=assigner_id,
|
||||
expires_at=request.expires_at,
|
||||
)
|
||||
|
||||
return SuccessResponse(
|
||||
message=f"Permission {request.permission_code} granted to user {request.user_id}",
|
||||
data={
|
||||
"permission_id": user_permission.id,
|
||||
"expires_at": (
|
||||
user_permission.expires_at.isoformat()
|
||||
if user_permission.expires_at
|
||||
else None
|
||||
),
|
||||
},
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# EJEMPLOS DE RUTAS PROTEGIDAS CON PERMISOS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@router.get("/examples/invoices")
|
||||
async def example_view_invoices(_: None = Depends(RequirePermission("invoice.view"))):
|
||||
"""
|
||||
Endpoint de ejemplo que requiere el permiso invoice.view
|
||||
"""
|
||||
return {
|
||||
"message": "You have permission to view invoices",
|
||||
"invoices": [
|
||||
{"id": 1, "number": "INV-001", "amount": 1000},
|
||||
{"id": 2, "number": "INV-002", "amount": 2000},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@router.post("/examples/invoices")
|
||||
async def example_edit_invoice(
|
||||
_: None = Depends(
|
||||
PermissionChecker(["invoice.view", "invoice.edit"], require_all=True)
|
||||
)
|
||||
):
|
||||
"""
|
||||
Endpoint de ejemplo que requiere AMBOS permisos: invoice.view e invoice.edit
|
||||
"""
|
||||
return {
|
||||
"message": "Invoice updated successfully",
|
||||
"invoice": {"id": 1, "number": "INV-001", "amount": 1500},
|
||||
}
|
||||
|
||||
|
||||
@router.get("/examples/reports")
|
||||
async def example_view_reports(
|
||||
_: None = Depends(
|
||||
PermissionChecker(
|
||||
["report.financial.view", "report.admin.view"], require_all=False
|
||||
)
|
||||
)
|
||||
):
|
||||
"""
|
||||
Endpoint de ejemplo que requiere AL MENOS UNO de los permisos especificados.
|
||||
"""
|
||||
return {
|
||||
"message": "Financial reports",
|
||||
"reports": ["Monthly P&L", "Cash Flow", "Balance Sheet"],
|
||||
}
|
||||
|
||||
|
||||
@router.get("/examples/dashboard")
|
||||
async def example_dashboard(
|
||||
user_permissions: set = Depends(get_current_user_permissions),
|
||||
):
|
||||
"""
|
||||
Dashboard dinámico que muestra diferentes widgets según los permisos del usuario.
|
||||
"""
|
||||
widgets = []
|
||||
|
||||
if "invoice.view" in user_permissions:
|
||||
widgets.append(
|
||||
{
|
||||
"type": "invoices",
|
||||
"title": "Recent Invoices",
|
||||
"data": [{"id": 1, "number": "INV-001"}],
|
||||
}
|
||||
)
|
||||
|
||||
if "report.financial.view" in user_permissions:
|
||||
widgets.append(
|
||||
{
|
||||
"type": "financial",
|
||||
"title": "Financial Summary",
|
||||
"data": {"revenue": 50000, "expenses": 30000},
|
||||
}
|
||||
)
|
||||
|
||||
if "user.view" in user_permissions:
|
||||
widgets.append(
|
||||
{"type": "users", "title": "User Activity", "data": {"active_users": 42}}
|
||||
)
|
||||
|
||||
return {"permissions": list(user_permissions), "widgets": widgets}
|
||||
245
backend/api/v1/modules/core/permissions/schemas.py
Normal file
245
backend/api/v1/modules/core/permissions/schemas.py
Normal file
@@ -0,0 +1,245 @@
|
||||
"""
|
||||
Esquemas Pydantic para el módulo de permisos.
|
||||
Define los modelos de request/response para las APIs de permisos.
|
||||
"""
|
||||
|
||||
from typing import List, Optional
|
||||
from datetime import datetime
|
||||
from pydantic import BaseModel, Field, ConfigDict
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# SCHEMAS DE RESPONSE
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class PermissionResponse(BaseModel):
|
||||
"""Esquema de respuesta para un permiso individual."""
|
||||
|
||||
id: int
|
||||
code: str = Field(..., description="Código único del permiso (ej: 'invoice.view')")
|
||||
description: Optional[str] = Field(None, description="Descripción del permiso")
|
||||
module: str = Field(..., description="Módulo al que pertenece (ej: 'invoice')")
|
||||
action: str = Field(..., description="Acción específica (ej: 'view', 'edit')")
|
||||
is_active: bool = Field(..., description="Si el permiso está activo")
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
class CompanyRoleResponse(BaseModel):
|
||||
"""Esquema de respuesta para un rol de companye."""
|
||||
|
||||
id: int
|
||||
company_id: int = Field(..., description="ID del companye al que pertenece el rol")
|
||||
name: str = Field(..., description="Nombre del rol (ej: 'Administrador')")
|
||||
code: str = Field(..., description="Código del rol (ej: 'admin')")
|
||||
description: Optional[str] = Field(None, description="Descripción del rol")
|
||||
is_active: bool = Field(..., description="Si el rol está activo")
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
class CompanyRoleWithPermissionsResponse(CompanyRoleResponse):
|
||||
"""Esquema de respuesta para un rol con sus permisos incluidos."""
|
||||
|
||||
permissions: List[PermissionResponse] = Field(
|
||||
default_factory=list, description="Lista de permisos asignados a este rol"
|
||||
)
|
||||
|
||||
|
||||
class UserPermissionsResponse(BaseModel):
|
||||
"""Esquema de respuesta para los permisos de un usuario."""
|
||||
|
||||
user_id: str = Field(..., description="ID del usuario")
|
||||
company_id: int = Field(..., description="ID del companye")
|
||||
permissions: List[str] = Field(
|
||||
default_factory=list, description="Lista de códigos de permisos del usuario"
|
||||
)
|
||||
roles: List[str] = Field(
|
||||
default_factory=list, description="Lista de nombres de roles del usuario"
|
||||
)
|
||||
|
||||
|
||||
class UserCompanyRoleResponse(BaseModel):
|
||||
"""Esquema de respuesta para la asignación de rol a usuario."""
|
||||
|
||||
id: int
|
||||
user_id: str
|
||||
company_id: int
|
||||
company_role_id: int
|
||||
is_active: bool
|
||||
assigned_at: datetime
|
||||
assigned_by: Optional[str] = None
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
class UserCompanyPermissionResponse(BaseModel):
|
||||
"""Esquema de respuesta para un permiso directo de usuario."""
|
||||
|
||||
id: int
|
||||
user_id: str
|
||||
company_id: int
|
||||
permission_id: int
|
||||
permission_code: Optional[str] = None
|
||||
is_granted: bool = Field(
|
||||
..., description="True si está concedido, False si está revocado"
|
||||
)
|
||||
is_active: bool
|
||||
assigned_at: datetime
|
||||
assigned_by: Optional[str] = None
|
||||
expires_at: Optional[datetime] = Field(
|
||||
None, description="Fecha de expiración del permiso"
|
||||
)
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# SCHEMAS DE REQUEST
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class AssignRoleRequest(BaseModel):
|
||||
"""Esquema de request para asignar un rol a un usuario."""
|
||||
|
||||
user_id: str = Field(..., description="ID del usuario al que se asignará el rol")
|
||||
role_id: int = Field(..., description="ID del rol a asignar")
|
||||
|
||||
|
||||
class RemoveRoleRequest(BaseModel):
|
||||
"""Esquema de request para remover un rol de un usuario."""
|
||||
|
||||
user_id: str = Field(..., description="ID del usuario")
|
||||
role_id: int = Field(..., description="ID del rol a remover")
|
||||
|
||||
|
||||
class GrantPermissionRequest(BaseModel):
|
||||
"""Esquema de request para conceder un permiso directo a un usuario."""
|
||||
|
||||
user_id: str = Field(..., description="ID del usuario")
|
||||
permission_code: str = Field(
|
||||
..., description="Código del permiso a conceder (ej: 'invoice.delete')"
|
||||
)
|
||||
expires_at: Optional[datetime] = Field(
|
||||
None, description="Fecha de expiración del permiso (opcional)"
|
||||
)
|
||||
|
||||
|
||||
class RevokePermissionRequest(BaseModel):
|
||||
"""Esquema de request para revocar un permiso directo."""
|
||||
|
||||
user_id: str = Field(..., description="ID del usuario")
|
||||
permission_code: str = Field(..., description="Código del permiso a revocar")
|
||||
|
||||
|
||||
class CreateRoleRequest(BaseModel):
|
||||
"""Esquema de request para crear un rol personalizado."""
|
||||
|
||||
name: str = Field(..., min_length=1, max_length=100, description="Nombre del rol")
|
||||
code: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
max_length=100,
|
||||
description="Código único del rol (ej: 'custom_admin')",
|
||||
)
|
||||
description: Optional[str] = Field(
|
||||
None, max_length=255, description="Descripción del rol"
|
||||
)
|
||||
permission_ids: List[int] = Field(
|
||||
default_factory=list, description="IDs de permisos a asignar al rol"
|
||||
)
|
||||
|
||||
|
||||
class UpdateRoleRequest(BaseModel):
|
||||
"""Esquema de request para actualizar un rol existente."""
|
||||
|
||||
name: Optional[str] = Field(None, min_length=1, max_length=100)
|
||||
description: Optional[str] = Field(None, max_length=255)
|
||||
is_active: Optional[bool] = None
|
||||
|
||||
|
||||
class AssignPermissionsToRoleRequest(BaseModel):
|
||||
"""Esquema de request para asignar permisos a un rol."""
|
||||
|
||||
permission_ids: List[int] = Field(
|
||||
..., description="Lista de IDs de permisos a asignar al rol"
|
||||
)
|
||||
replace_existing: bool = Field(
|
||||
False,
|
||||
description="Si True, reemplaza los permisos existentes. Si False, los agrega.",
|
||||
)
|
||||
|
||||
|
||||
class CreatePermissionRequest(BaseModel):
|
||||
"""Esquema de request para crear un nuevo permiso (uso administrativo)."""
|
||||
|
||||
code: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
max_length=100,
|
||||
description="Código único del permiso (ej: 'custom_module.action')",
|
||||
)
|
||||
description: Optional[str] = Field(None, max_length=255)
|
||||
module: str = Field(
|
||||
..., min_length=1, max_length=50, description="Módulo del permiso"
|
||||
)
|
||||
action: str = Field(
|
||||
..., min_length=1, max_length=50, description="Acción del permiso"
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# SCHEMAS DE RESPUESTA GENÉRICOS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class SuccessResponse(BaseModel):
|
||||
"""Respuesta genérica de éxito."""
|
||||
|
||||
success: bool = True
|
||||
message: str = Field(..., description="Mensaje descriptivo de la operación")
|
||||
data: Optional[dict] = Field(None, description="Datos adicionales opcionales")
|
||||
|
||||
|
||||
class ErrorResponse(BaseModel):
|
||||
"""Respuesta genérica de error."""
|
||||
|
||||
success: bool = False
|
||||
detail: str = Field(..., description="Descripción del error")
|
||||
error_code: Optional[str] = Field(None, description="Código de error específico")
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# SCHEMAS DE PAGINACIÓN
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class PaginatedResponse(BaseModel):
|
||||
"""Esquema genérico para respuestas paginadas."""
|
||||
|
||||
items: List[dict] = Field(default_factory=list)
|
||||
total: int = Field(..., description="Total de items disponibles")
|
||||
page: int = Field(..., description="Página actual")
|
||||
page_size: int = Field(..., description="Tamaño de página")
|
||||
total_pages: int = Field(..., description="Total de páginas disponibles")
|
||||
|
||||
|
||||
class PermissionListResponse(BaseModel):
|
||||
"""Lista paginada de permisos."""
|
||||
|
||||
items: List[PermissionResponse]
|
||||
total: int
|
||||
page: int = 1
|
||||
page_size: int = 100
|
||||
|
||||
|
||||
class RoleListResponse(BaseModel):
|
||||
"""Lista paginada de roles."""
|
||||
|
||||
items: List[CompanyRoleResponse]
|
||||
total: int
|
||||
page: int = 1
|
||||
page_size: int = 100
|
||||
63
backend/api/v1/modules/core/permissions/seed.py
Normal file
63
backend/api/v1/modules/core/permissions/seed.py
Normal file
@@ -0,0 +1,63 @@
|
||||
seed_invoices = [
|
||||
("invoice.imp.tem.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("invoice.imp.tem.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("invoice.imp.tem.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.imp.tem.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.imp.tem.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
("invoice.imp.def.view", "Ver facturas definitivas de importación", "invoice", "view"),
|
||||
("invoice.imp.def.create", "Crear facturas definitivas de importación", "invoice", "create"),
|
||||
("invoice.imp.def.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.imp.def.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.imp.def.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
("invoice.imp.cm.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("invoice.imp.cm.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("invoice.imp.cm.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.imp.cm.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.imp.cm.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
("invoice.imp.cr.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("invoice.imp.cr.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("invoice.imp.cr.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.imp.cr.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.imp.cr.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
("invoice.exp.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("invoice.exp.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("invoice.exp.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.exp.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.exp.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
("invoice.exp.rep.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("invoice.exp.rep.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("invoice.exp.rep.edit", "Editar facturas", "invoice", "edit"),
|
||||
("invoice.exp.rep.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("invoice.exp.rep.process", "Procesar facturas", "invoice", "process"),
|
||||
|
||||
|
||||
]
|
||||
|
||||
seed_user = [
|
||||
("user.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("user.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("user.edit", "Editar facturas", "invoice", "edit"),
|
||||
("user.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("user.process", "Procesar facturas", "invoice", "process"),
|
||||
]
|
||||
|
||||
seed_report = [
|
||||
("report.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("report.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("report.edit", "Editar facturas", "invoice", "edit"),
|
||||
("report.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("report.process", "Procesar facturas", "invoice", "process"),
|
||||
]
|
||||
|
||||
seed_roles = [
|
||||
("roles.view", "Ver facturas temporales de importación", "invoice", "view"),
|
||||
("roles.create", "Crear facturas temporales de importación", "invoice", "create"),
|
||||
("roles.edit", "Editar facturas", "invoice", "edit"),
|
||||
("roles.delete", "Eliminar facturas", "invoice", "delete"),
|
||||
("roles.process", "Procesar facturas", "invoice", "process"),
|
||||
]
|
||||
298
backend/api/v1/modules/core/permissions/service.py
Normal file
298
backend/api/v1/modules/core/permissions/service.py
Normal file
@@ -0,0 +1,298 @@
|
||||
"""
|
||||
Servicio de gestión de permisos multi-tenant.
|
||||
Proporciona funciones para verificar y obtener permisos de usuarios por companye.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Set, Optional, List
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_, or_
|
||||
from .models import (
|
||||
Permission,
|
||||
CompanyRole,
|
||||
RolePermission,
|
||||
UserCompanyRole,
|
||||
UserCompanyPermission,
|
||||
)
|
||||
|
||||
|
||||
class PermissionService:
|
||||
"""
|
||||
Servicio para gestionar permisos de usuarios en contextos multi-tenant.
|
||||
Combina permisos de roles y permisos directos del usuario.
|
||||
"""
|
||||
|
||||
def __init__(self, db: Session):
|
||||
self.db = db
|
||||
|
||||
def get_user_permissions(
|
||||
self, user_id: str, company_id: int, use_cache: bool = False
|
||||
) -> Set[str]:
|
||||
"""
|
||||
Obtiene todos los permisos de un usuario para un companye específico.
|
||||
Combina permisos de roles asignados y permisos directos.
|
||||
|
||||
Args:
|
||||
user_id: ID del usuario
|
||||
company_id: ID del companye/tenant
|
||||
use_cache: Si True, intenta usar caché (útil en producción)
|
||||
|
||||
Returns:
|
||||
Set de códigos de permisos (ej: {"invoice.view", "invoice.edit"})
|
||||
"""
|
||||
# Obtener permisos de roles asignados al usuario
|
||||
role_permissions = self._get_permissions_from_roles(user_id, company_id)
|
||||
|
||||
# Obtener permisos directos del usuario
|
||||
direct_permissions = self._get_direct_permissions(user_id, company_id)
|
||||
|
||||
# Combinar ambos conjuntos
|
||||
# Los permisos directos revocados (is_granted=False) anulan los de roles
|
||||
all_permissions = role_permissions.copy()
|
||||
|
||||
for perm_code, is_granted in direct_permissions.items():
|
||||
if is_granted:
|
||||
all_permissions.add(perm_code)
|
||||
else:
|
||||
# Permiso explícitamente revocado
|
||||
all_permissions.discard(perm_code)
|
||||
|
||||
return all_permissions
|
||||
|
||||
def _get_permissions_from_roles(self, user_id: str, company_id: int) -> Set[str]:
|
||||
"""
|
||||
Obtiene permisos derivados de los roles del usuario en el companye.
|
||||
|
||||
Realiza un JOIN eficiente para obtener todos los permisos de los roles activos.
|
||||
"""
|
||||
query = (
|
||||
self.db.query(Permission.code)
|
||||
.join(RolePermission, RolePermission.permission_id == Permission.id)
|
||||
.join(CompanyRole, CompanyRole.id == RolePermission.company_role_id)
|
||||
.join(UserCompanyRole, UserCompanyRole.company_role_id == CompanyRole.id)
|
||||
.filter(
|
||||
and_(
|
||||
UserCompanyRole.user_id == user_id,
|
||||
UserCompanyRole.company_id == company_id,
|
||||
UserCompanyRole.is_active == True,
|
||||
CompanyRole.is_active == True,
|
||||
Permission.is_active == True,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
results = query.all()
|
||||
return {perm_code for (perm_code,) in results}
|
||||
|
||||
def _get_direct_permissions(self, user_id: str, company_id: int) -> dict:
|
||||
"""
|
||||
Obtiene permisos directos asignados al usuario.
|
||||
|
||||
Returns:
|
||||
Dict con código de permiso como key y is_granted como value
|
||||
{
|
||||
"invoice.delete": True, # Permiso concedido
|
||||
"user.delete": False # Permiso revocado explícitamente
|
||||
}
|
||||
"""
|
||||
now = datetime.utcnow()
|
||||
|
||||
query = (
|
||||
self.db.query(Permission.code, UserCompanyPermission.is_granted)
|
||||
.join(
|
||||
UserCompanyPermission,
|
||||
UserCompanyPermission.permission_id == Permission.id,
|
||||
)
|
||||
.filter(
|
||||
and_(
|
||||
UserCompanyPermission.user_id == user_id,
|
||||
UserCompanyPermission.company_id == company_id,
|
||||
UserCompanyPermission.is_active == True,
|
||||
Permission.is_active == True,
|
||||
or_(
|
||||
UserCompanyPermission.expires_at.is_(None),
|
||||
UserCompanyPermission.expires_at > now,
|
||||
),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
results = query.all()
|
||||
return {perm_code: is_granted for perm_code, is_granted in results}
|
||||
|
||||
def has_permission(
|
||||
self, user_id: str, company_id: int, permission_code: str
|
||||
) -> bool:
|
||||
"""
|
||||
Verifica si un usuario tiene un permiso específico en un companye.
|
||||
|
||||
Args:
|
||||
user_id: ID del usuario
|
||||
company_id: ID del companye/tenant
|
||||
permission_code: Código del permiso (ej: "invoice.edit")
|
||||
|
||||
Returns:
|
||||
True si el usuario tiene el permiso, False en caso contrario
|
||||
"""
|
||||
permissions = self.get_user_permissions(user_id, company_id)
|
||||
return permission_code in permissions
|
||||
|
||||
def has_any_permission(
|
||||
self, user_id: str, company_id: int, permission_codes: List[str]
|
||||
) -> bool:
|
||||
"""
|
||||
Verifica si el usuario tiene al menos uno de los permisos especificados.
|
||||
"""
|
||||
permissions = self.get_user_permissions(user_id, company_id)
|
||||
return any(perm in permissions for perm in permission_codes)
|
||||
|
||||
def has_all_permissions(
|
||||
self, user_id: str, company_id: int, permission_codes: List[str]
|
||||
) -> bool:
|
||||
"""
|
||||
Verifica si el usuario tiene todos los permisos especificados.
|
||||
"""
|
||||
permissions = self.get_user_permissions(user_id, company_id)
|
||||
return all(perm in permissions for perm in permission_codes)
|
||||
|
||||
def get_user_roles(self, user_id: str, company_id: int) -> List[CompanyRole]:
|
||||
"""
|
||||
Obtiene los roles activos de un usuario en un companye.
|
||||
"""
|
||||
query = (
|
||||
self.db.query(CompanyRole)
|
||||
.join(UserCompanyRole, UserCompanyRole.company_role_id == CompanyRole.id)
|
||||
.filter(
|
||||
and_(
|
||||
UserCompanyRole.user_id == user_id,
|
||||
UserCompanyRole.company_id == company_id,
|
||||
UserCompanyRole.is_active == True,
|
||||
CompanyRole.is_active == True,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
return query.all()
|
||||
|
||||
def assign_role_to_user(
|
||||
self,
|
||||
user_id: str,
|
||||
company_id: int,
|
||||
role_id: int,
|
||||
assigned_by: Optional[str] = None,
|
||||
) -> UserCompanyRole:
|
||||
"""
|
||||
Asigna un rol a un usuario en un companye específico.
|
||||
"""
|
||||
# Verificar que el rol pertenece al companye
|
||||
role = (
|
||||
self.db.query(CompanyRole)
|
||||
.filter(
|
||||
and_(
|
||||
CompanyRole.id == role_id,
|
||||
CompanyRole.company_id == company_id,
|
||||
CompanyRole.is_active == True,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not role:
|
||||
raise ValueError(f"Role {role_id} not found for company {company_id}")
|
||||
|
||||
# Verificar si ya existe la asignación
|
||||
existing = (
|
||||
self.db.query(UserCompanyRole)
|
||||
.filter(
|
||||
and_(
|
||||
UserCompanyRole.user_id == user_id,
|
||||
UserCompanyRole.company_id == company_id,
|
||||
UserCompanyRole.company_role_id == role_id,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing:
|
||||
if not existing.is_active:
|
||||
existing.is_active = True
|
||||
existing.assigned_at = datetime.utcnow()
|
||||
existing.assigned_by = assigned_by
|
||||
self.db.commit()
|
||||
return existing
|
||||
return existing
|
||||
|
||||
# Crear nueva asignación
|
||||
user_role = UserCompanyRole(
|
||||
user_id=user_id,
|
||||
company_id=company_id,
|
||||
company_role_id=role_id,
|
||||
assigned_by=assigned_by,
|
||||
)
|
||||
|
||||
self.db.add(user_role)
|
||||
self.db.commit()
|
||||
self.db.refresh(user_role)
|
||||
|
||||
return user_role
|
||||
|
||||
def grant_direct_permission(
|
||||
self,
|
||||
user_id: str,
|
||||
company_id: int,
|
||||
permission_code: str,
|
||||
assigned_by: Optional[str] = None,
|
||||
expires_at: Optional[datetime] = None,
|
||||
) -> UserCompanyPermission:
|
||||
"""
|
||||
Concede un permiso directo a un usuario en un companye.
|
||||
"""
|
||||
# Obtener el permiso por código
|
||||
permission = (
|
||||
self.db.query(Permission)
|
||||
.filter(
|
||||
and_(Permission.code == permission_code, Permission.is_active == True)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not permission:
|
||||
raise ValueError(f"Permission {permission_code} not found")
|
||||
|
||||
# Verificar si ya existe
|
||||
existing = (
|
||||
self.db.query(UserCompanyPermission)
|
||||
.filter(
|
||||
and_(
|
||||
UserCompanyPermission.user_id == user_id,
|
||||
UserCompanyPermission.company_id == company_id,
|
||||
UserCompanyPermission.permission_id == permission.id,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing:
|
||||
existing.is_granted = True
|
||||
existing.is_active = True
|
||||
existing.assigned_at = datetime.utcnow()
|
||||
existing.assigned_by = assigned_by
|
||||
existing.expires_at = expires_at
|
||||
self.db.commit()
|
||||
return existing
|
||||
|
||||
# Crear nuevo permiso directo
|
||||
user_permission = UserCompanyPermission(
|
||||
user_id=user_id,
|
||||
company_id=company_id,
|
||||
permission_id=permission.id,
|
||||
is_granted=True,
|
||||
assigned_by=assigned_by,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
self.db.add(user_permission)
|
||||
self.db.commit()
|
||||
self.db.refresh(user_permission)
|
||||
|
||||
return user_permission
|
||||
424
backend/api/v1/modules/core/permissions/test_permissions.py
Normal file
424
backend/api/v1/modules/core/permissions/test_permissions.py
Normal file
@@ -0,0 +1,424 @@
|
||||
"""
|
||||
Tests unitarios para el módulo de permisos multi-tenant.
|
||||
Ejecutar con: pytest backend/api/v1/modules/core/permissions/test_permissions.py -v
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker, Session
|
||||
from core.database import Base
|
||||
from api.v1.modules.core.permissions.models import (
|
||||
Permission,
|
||||
ClientRole,
|
||||
RolePermission,
|
||||
UserClientRole,
|
||||
UserClientPermission,
|
||||
)
|
||||
from api.v1.modules.core.permissions.service import PermissionService
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# FIXTURES
|
||||
# ============================================================================
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db_session():
|
||||
"""Crea una sesión de base de datos en memoria para testing."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(engine)
|
||||
Session = sessionmaker(bind=engine)
|
||||
session = Session()
|
||||
|
||||
yield session
|
||||
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_permissions(db_session: Session):
|
||||
"""Crea permisos de ejemplo."""
|
||||
permissions = [
|
||||
Permission(
|
||||
code="invoice.view",
|
||||
description="Ver facturas",
|
||||
module="invoice",
|
||||
action="view",
|
||||
),
|
||||
Permission(
|
||||
code="invoice.create",
|
||||
description="Crear facturas",
|
||||
module="invoice",
|
||||
action="create",
|
||||
),
|
||||
Permission(
|
||||
code="invoice.edit",
|
||||
description="Editar facturas",
|
||||
module="invoice",
|
||||
action="edit",
|
||||
),
|
||||
Permission(
|
||||
code="invoice.delete",
|
||||
description="Eliminar facturas",
|
||||
module="invoice",
|
||||
action="delete",
|
||||
),
|
||||
Permission(
|
||||
code="user.view", description="Ver usuarios", module="user", action="view"
|
||||
),
|
||||
]
|
||||
|
||||
for perm in permissions:
|
||||
db_session.add(perm)
|
||||
|
||||
db_session.commit()
|
||||
return permissions
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_role(db_session: Session, sample_permissions):
|
||||
"""Crea un rol de ejemplo con permisos."""
|
||||
role = ClientRole(
|
||||
client_id=1, name="Administrador", code="admin", description="Acceso completo"
|
||||
)
|
||||
db_session.add(role)
|
||||
db_session.commit()
|
||||
|
||||
# Asignar permisos al rol
|
||||
for perm in sample_permissions[:3]: # invoice.view, create, edit
|
||||
role_perm = RolePermission(client_role_id=role.id, permission_id=perm.id)
|
||||
db_session.add(role_perm)
|
||||
|
||||
db_session.commit()
|
||||
return role
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS DE PermissionService
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestPermissionService:
|
||||
"""Tests para el servicio de permisos."""
|
||||
|
||||
def test_get_permissions_from_role(self, db_session, sample_role):
|
||||
"""Test: obtener permisos desde roles asignados."""
|
||||
# Asignar rol a usuario
|
||||
user_role = UserClientRole(
|
||||
user_id="user-123", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
# Verificar permisos
|
||||
service = PermissionService(db_session)
|
||||
permissions = service.get_user_permissions("user-123", 1)
|
||||
|
||||
assert "invoice.view" in permissions
|
||||
assert "invoice.create" in permissions
|
||||
assert "invoice.edit" in permissions
|
||||
assert "invoice.delete" not in permissions # No está en el rol
|
||||
|
||||
def test_has_permission_single(self, db_session, sample_role):
|
||||
"""Test: verificar un permiso específico."""
|
||||
user_role = UserClientRole(
|
||||
user_id="user-123", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
|
||||
assert service.has_permission("user-123", 1, "invoice.view") is True
|
||||
assert service.has_permission("user-123", 1, "invoice.delete") is False
|
||||
|
||||
def test_has_all_permissions(self, db_session, sample_role):
|
||||
"""Test: verificar múltiples permisos (todos requeridos)."""
|
||||
user_role = UserClientRole(
|
||||
user_id="user-123", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Tiene todos
|
||||
assert (
|
||||
service.has_all_permissions(
|
||||
"user-123", 1, ["invoice.view", "invoice.create"]
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
# Falta uno
|
||||
assert (
|
||||
service.has_all_permissions(
|
||||
"user-123", 1, ["invoice.view", "invoice.delete"]
|
||||
)
|
||||
is False
|
||||
)
|
||||
|
||||
def test_has_any_permission(self, db_session, sample_role):
|
||||
"""Test: verificar múltiples permisos (al menos uno)."""
|
||||
user_role = UserClientRole(
|
||||
user_id="user-123", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Tiene al menos uno
|
||||
assert (
|
||||
service.has_any_permission(
|
||||
"user-123", 1, ["invoice.view", "invoice.delete"]
|
||||
)
|
||||
is True
|
||||
)
|
||||
|
||||
# No tiene ninguno
|
||||
assert (
|
||||
service.has_any_permission("user-123", 1, ["invoice.delete", "user.view"])
|
||||
is False
|
||||
)
|
||||
|
||||
def test_direct_permission_grant(self, db_session, sample_permissions):
|
||||
"""Test: conceder permiso directo a usuario."""
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Conceder permiso directo
|
||||
user_perm = service.grant_direct_permission(
|
||||
user_id="user-123", client_id=1, permission_code="invoice.delete"
|
||||
)
|
||||
|
||||
assert user_perm.is_granted is True
|
||||
assert user_perm.is_active is True
|
||||
|
||||
# Verificar que el usuario tiene el permiso
|
||||
permissions = service.get_user_permissions("user-123", 1)
|
||||
assert "invoice.delete" in permissions
|
||||
|
||||
def test_direct_permission_revoke(
|
||||
self, db_session, sample_role, sample_permissions
|
||||
):
|
||||
"""Test: revocar permiso que viene de un rol."""
|
||||
# Asignar rol con permisos
|
||||
user_role = UserClientRole(
|
||||
user_id="user-123", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Verificar que tiene el permiso del rol
|
||||
assert service.has_permission("user-123", 1, "invoice.view") is True
|
||||
|
||||
# Revocar explícitamente
|
||||
invoice_view = (
|
||||
db_session.query(Permission)
|
||||
.filter(Permission.code == "invoice.view")
|
||||
.first()
|
||||
)
|
||||
|
||||
user_perm = UserClientPermission(
|
||||
user_id="user-123",
|
||||
client_id=1,
|
||||
permission_id=invoice_view.id,
|
||||
is_granted=False, # Revocado
|
||||
)
|
||||
db_session.add(user_perm)
|
||||
db_session.commit()
|
||||
|
||||
# Ahora NO debe tener el permiso
|
||||
assert service.has_permission("user-123", 1, "invoice.view") is False
|
||||
|
||||
def test_permission_expiration(self, db_session, sample_permissions):
|
||||
"""Test: permisos con fecha de expiración."""
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Permiso expirado
|
||||
expired_perm = service.grant_direct_permission(
|
||||
user_id="user-123",
|
||||
client_id=1,
|
||||
permission_code="invoice.delete",
|
||||
expires_at=datetime.utcnow() - timedelta(days=1), # Ayer
|
||||
)
|
||||
|
||||
# No debe aparecer en los permisos
|
||||
permissions = service.get_user_permissions("user-123", 1)
|
||||
assert "invoice.delete" not in permissions
|
||||
|
||||
# Permiso futuro
|
||||
future_perm = service.grant_direct_permission(
|
||||
user_id="user-456",
|
||||
client_id=1,
|
||||
permission_code="invoice.delete",
|
||||
expires_at=datetime.utcnow() + timedelta(days=7), # En 7 días
|
||||
)
|
||||
|
||||
# Debe aparecer
|
||||
permissions = service.get_user_permissions("user-456", 1)
|
||||
assert "invoice.delete" in permissions
|
||||
|
||||
def test_assign_role_to_user(self, db_session, sample_role):
|
||||
"""Test: asignar un rol a un usuario."""
|
||||
service = PermissionService(db_session)
|
||||
|
||||
user_role = service.assign_role_to_user(
|
||||
user_id="user-789",
|
||||
client_id=1,
|
||||
role_id=sample_role.id,
|
||||
assigned_by="admin-001",
|
||||
)
|
||||
|
||||
assert user_role.user_id == "user-789"
|
||||
assert user_role.client_id == 1
|
||||
assert user_role.client_role_id == sample_role.id
|
||||
assert user_role.assigned_by == "admin-001"
|
||||
assert user_role.is_active is True
|
||||
|
||||
def test_assign_role_invalid_client(self, db_session, sample_role):
|
||||
"""Test: intentar asignar rol de otro cliente falla."""
|
||||
service = PermissionService(db_session)
|
||||
|
||||
with pytest.raises(ValueError, match="not found for client"):
|
||||
service.assign_role_to_user(
|
||||
user_id="user-789",
|
||||
client_id=999, # Cliente incorrecto
|
||||
role_id=sample_role.id,
|
||||
)
|
||||
|
||||
def test_multiple_roles_same_user(self, db_session, sample_permissions):
|
||||
"""Test: usuario con múltiples roles combina permisos."""
|
||||
# Crear dos roles diferentes
|
||||
role1 = ClientRole(client_id=1, code="viewer", name="Viewer")
|
||||
role2 = ClientRole(client_id=1, code="editor", name="Editor")
|
||||
db_session.add_all([role1, role2])
|
||||
db_session.commit()
|
||||
|
||||
# Role1: solo view
|
||||
rp1 = RolePermission(
|
||||
client_role_id=role1.id,
|
||||
permission_id=sample_permissions[0].id, # invoice.view
|
||||
)
|
||||
|
||||
# Role2: create y edit
|
||||
rp2 = RolePermission(
|
||||
client_role_id=role2.id,
|
||||
permission_id=sample_permissions[1].id, # invoice.create
|
||||
)
|
||||
rp3 = RolePermission(
|
||||
client_role_id=role2.id,
|
||||
permission_id=sample_permissions[2].id, # invoice.edit
|
||||
)
|
||||
db_session.add_all([rp1, rp2, rp3])
|
||||
db_session.commit()
|
||||
|
||||
# Asignar ambos roles al usuario
|
||||
ur1 = UserClientRole(user_id="user-multi", client_id=1, client_role_id=role1.id)
|
||||
ur2 = UserClientRole(user_id="user-multi", client_id=1, client_role_id=role2.id)
|
||||
db_session.add_all([ur1, ur2])
|
||||
db_session.commit()
|
||||
|
||||
# Verificar que tiene permisos de ambos roles
|
||||
service = PermissionService(db_session)
|
||||
permissions = service.get_user_permissions("user-multi", 1)
|
||||
|
||||
assert "invoice.view" in permissions
|
||||
assert "invoice.create" in permissions
|
||||
assert "invoice.edit" in permissions
|
||||
assert "invoice.delete" not in permissions
|
||||
|
||||
def test_inactive_role_no_permissions(self, db_session, sample_role):
|
||||
"""Test: roles inactivos no otorgan permisos."""
|
||||
# Asignar rol activo
|
||||
user_role = UserClientRole(
|
||||
user_id="user-inactive", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
|
||||
# Verificar que tiene permisos
|
||||
assert service.has_permission("user-inactive", 1, "invoice.view") is True
|
||||
|
||||
# Desactivar el rol
|
||||
sample_role.is_active = False
|
||||
db_session.commit()
|
||||
|
||||
# Ahora no debe tener permisos
|
||||
assert service.has_permission("user-inactive", 1, "invoice.view") is False
|
||||
|
||||
def test_get_user_roles(self, db_session, sample_role):
|
||||
"""Test: obtener roles de un usuario."""
|
||||
user_role = UserClientRole(
|
||||
user_id="user-roles", client_id=1, client_role_id=sample_role.id
|
||||
)
|
||||
db_session.add(user_role)
|
||||
db_session.commit()
|
||||
|
||||
service = PermissionService(db_session)
|
||||
roles = service.get_user_roles("user-roles", 1)
|
||||
|
||||
assert len(roles) == 1
|
||||
assert roles[0].code == "admin"
|
||||
assert roles[0].name == "Administrador"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# TESTS DE MODELOS
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestModels:
|
||||
"""Tests para los modelos de base de datos."""
|
||||
|
||||
def test_permission_creation(self, db_session):
|
||||
"""Test: crear un permiso."""
|
||||
perm = Permission(
|
||||
code="test.action",
|
||||
description="Permiso de prueba",
|
||||
module="test",
|
||||
action="action",
|
||||
)
|
||||
db_session.add(perm)
|
||||
db_session.commit()
|
||||
|
||||
assert perm.id is not None
|
||||
assert perm.is_active is True
|
||||
assert perm.created_at is not None
|
||||
|
||||
def test_client_role_unique_constraint(self, db_session):
|
||||
"""Test: código de rol debe ser único por cliente."""
|
||||
role1 = ClientRole(client_id=1, code="admin", name="Admin 1")
|
||||
db_session.add(role1)
|
||||
db_session.commit()
|
||||
|
||||
# Intentar crear otro con mismo código en mismo cliente
|
||||
role2 = ClientRole(client_id=1, code="admin", name="Admin 2")
|
||||
db_session.add(role2)
|
||||
|
||||
with pytest.raises(Exception): # Violación de constraint
|
||||
db_session.commit()
|
||||
|
||||
def test_role_permission_relationship(
|
||||
self, db_session, sample_role, sample_permissions
|
||||
):
|
||||
"""Test: relación entre roles y permisos."""
|
||||
# El rol ya tiene permisos del fixture
|
||||
assert len(sample_role.role_permissions) > 0
|
||||
|
||||
# Verificar que los permisos están relacionados
|
||||
permission_codes = [rp.permission.code for rp in sample_role.role_permissions]
|
||||
assert "invoice.view" in permission_codes
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# EJECUTAR TESTS
|
||||
# ============================================================================
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
@@ -0,0 +1,45 @@
|
||||
seed = [
|
||||
("20", "20 ft Sea container open top"),
|
||||
("2B", "20 ft Sea container - closed top"),
|
||||
("40", "40 ft Sea container - open top"),
|
||||
("4B", "40 ft Sea container - closed top"),
|
||||
("BI", "Beverage rack trailer"),
|
||||
("CB", "Gooseneck trailer"),
|
||||
("CH", "Chassis"),
|
||||
("CL", "Other lenght Sea container - closed top"),
|
||||
("CU", "Other length Sea container - open top"),
|
||||
("CZ", "Refrigerated Container"),
|
||||
("DD", "Double Drop Trailer"),
|
||||
("DT", "Drop Back Trailer"),
|
||||
("FR", "Flat rack trailer"),
|
||||
("FT", "Flatbed/platform trailer"),
|
||||
("HC", "Hopper Trailer (covered)"),
|
||||
("HE", "Horse trailer"),
|
||||
("HO", "Hopper Trailer (open)"),
|
||||
("HP", "Hopper Trailer (covered Pneumatic Discharged)"),
|
||||
("L1", "Tank trailer (liquids) not heated/not insulated"),
|
||||
("L2", "Tank trailer (liquids) heated/not insulated"),
|
||||
("L3", "Tank trailer (liquids) not heated/insulated"),
|
||||
("L4", "Tank trailer (liquids) heated/insulate"),
|
||||
("LP", "Logging, pipe or pole trailer"),
|
||||
("LT", "Livestock trailer"),
|
||||
("NC", "No Equipment"),
|
||||
("OE", "Other"),
|
||||
("RD", "Fixed Rack, Double Drop Trailer"),
|
||||
("RG", "Gondola closed"),
|
||||
("RO", "Gondola open"),
|
||||
("RS", "Fixed Rack, Single Drop Trailer"),
|
||||
("SD", "Single Drop Trailer"),
|
||||
("T1", "Tank trailer (gas) not heated/ not insulated"),
|
||||
("T2", "Tank trailer (gas) heated/not insulated"),
|
||||
("T3", "Tank trailer (gas) not heated/insulated"),
|
||||
("T4", "Tank trailer (gas) heated/insulate"),
|
||||
("T5", "Tank trailer (chemicals) not heated/not insulated"),
|
||||
("T6", "Tank trailer (chemicals) heated/not insulated"),
|
||||
("T7", "Tank trailer (chemicals) not heated/insulated"),
|
||||
("T8", "Tank trailer (chemicals) heated/insulate"),
|
||||
("TC", "Auto car/trailer"),
|
||||
("TK", "Tank trailer (food grade liquids)"),
|
||||
("TL", "Semi-truck trailer"),
|
||||
("TW", "Controlled temperature trailer"),
|
||||
]
|
||||
Reference in New Issue
Block a user