feature/permisos-invoices

This commit is contained in:
2026-05-04 14:06:44 -06:00
parent 97bb5c23b6
commit aac5607e05
10 changed files with 569 additions and 238 deletions

View File

@@ -1,144 +1,125 @@
from typing import List, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, 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 api.v1.modules.a76.invoice_settings import services
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, InvoiceSettingsResponse, OperationType
from api.v1.modules.core.permissions.service import PermissionService
router = APIRouter(
prefix="/a76/invoice-settings",
tags=["a76/invoice-settings"]
)
def _invoice_perm_base_for_settings(operation_type: OperationType, invoice_type: str) -> str:
"""Alineado con get_invoice_permission_base en invoices/routes (defaults por tipo)."""
op = (
operation_type.value
if hasattr(operation_type, "value")
else str(operation_type).lower().split(".")[-1]
)
inv = (invoice_type or "").upper()
if op == "imp":
if inv == "TEM":
return "invoice.imp.tem"
if inv == "DEF":
return "invoice.imp.def"
if inv == "MEX":
return "invoice.imp.cm"
if inv == "CR":
return "invoice.imp.cr"
return "invoice.imp.tem"
if op == "exp":
if inv == "REPAR":
return "invoice.exp.rep"
return "invoice.exp"
return "invoice.imp.tem"
def _can_read_invoice_settings_row(
db: Session,
company_id: int,
current_user: Dict[str, Any],
invoice_type: str,
operation_type: OperationType,
) -> bool:
"""
Ver configuración por tipo/op: settings_general.view O ver facturas de ese mismo contexto
(para cargar defaults en alta/edición sin abrir la pantalla de parámetros).
"""
user_roles = current_user.get("realm_access", {}).get("roles", [])
if "admin" in user_roles:
return True
user_id = current_user.get("sub") or current_user.get("id")
if not user_id:
return False
ps = PermissionService(db)
if ps.has_permission(str(user_id), company_id, "settings_general.view"):
return True
base = _invoice_perm_base_for_settings(operation_type, invoice_type)
return ps.has_permission(str(user_id), company_id, f"{base}.view")
@router.get("/{invoice_type}", response_model=InvoiceSettingsResponse)
def get_invoice_settings(
invoice_type: str,
operation_type: OperationType = Query(...),
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get settings for a specific invoice type and operation"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not _can_read_invoice_settings_row(
db, company_id, current_user, invoice_type, operation_type
):
raise HTTPException(
status_code=403,
detail="Missing required permissions: settings_general.view "
f"(o permiso de vista del tipo de factura solicitado, p. ej. {_invoice_perm_base_for_settings(operation_type, invoice_type)}.view)",
)
settings = services.get_settings(
db,
tenant_id,
company_id,
invoice_type,
operation_type
)
if not settings:
# Return empty default if not found, to simplify frontend logic
return InvoiceSettingsResponse(
invoice_type=invoice_type,
operation_type=operation_type,
settings={},
id=0,
tenant_id=tenant_id,
company_id=company_id
)
return InvoiceSettingsResponse.model_validate(settings)
@router.get("/", response_model=List[InvoiceSettingsResponse])
def list_invoice_settings(
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""List all configured settings for validation or overview"""
tenant_id = validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=["settings_general.view"],
)
return services.list_settings(
db,
tenant_id,
company_id
)
@router.put("/", response_model=InvoiceSettingsResponse)
def save_invoice_settings(
settings_data: InvoiceSettingsRequest,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Create or update invoice settings"""
tenant_id = validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=["settings_general.edit"],
)
return services.upsert_settings(
db,
tenant_id,
company_id,
settings_data
)
from typing import List, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, 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 api.v1.modules.a76.invoice_settings import services
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, InvoiceSettingsResponse, OperationType
from api.v1.modules.core.permissions.service import PermissionService
from api.v1.modules.a76.invoices.permission_map import get_invoice_permission_base
router = APIRouter(
prefix="/a76/invoice-settings",
tags=["a76/invoice-settings"]
)
def _invoice_perm_base_for_settings(operation_type: OperationType, invoice_type: str) -> str:
"""Alineado con `permission_map.get_invoice_permission_base`."""
return get_invoice_permission_base(operation_type, invoice_type)
def _can_read_invoice_settings_row(
db: Session,
company_id: int,
current_user: Dict[str, Any],
invoice_type: str,
operation_type: OperationType,
) -> bool:
"""
Ver configuración por tipo/op: settings_general.view O ver facturas de ese mismo contexto
(para cargar defaults en alta/edición sin abrir la pantalla de parámetros).
"""
user_roles = current_user.get("realm_access", {}).get("roles", [])
if "admin" in user_roles:
return True
user_id = current_user.get("sub") or current_user.get("id")
if not user_id:
return False
ps = PermissionService(db)
if ps.has_permission(str(user_id), company_id, "settings_general.view"):
return True
base = _invoice_perm_base_for_settings(operation_type, invoice_type)
return ps.has_permission(str(user_id), company_id, f"{base}.view")
@router.get("/{invoice_type}", response_model=InvoiceSettingsResponse)
def get_invoice_settings(
invoice_type: str,
operation_type: OperationType = Query(...),
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get settings for a specific invoice type and operation"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
if not _can_read_invoice_settings_row(
db, company_id, current_user, invoice_type, operation_type
):
raise HTTPException(
status_code=403,
detail="Missing required permissions: settings_general.view "
f"(o permiso de vista del tipo de factura solicitado, p. ej. {_invoice_perm_base_for_settings(operation_type, invoice_type)}.view)",
)
settings = services.get_settings(
db,
tenant_id,
company_id,
invoice_type,
operation_type
)
if not settings:
# Return empty default if not found, to simplify frontend logic
return InvoiceSettingsResponse(
invoice_type=invoice_type,
operation_type=operation_type,
settings={},
id=0,
tenant_id=tenant_id,
company_id=company_id
)
return InvoiceSettingsResponse.model_validate(settings)
@router.get("/", response_model=List[InvoiceSettingsResponse])
def list_invoice_settings(
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""List all configured settings for validation or overview"""
tenant_id = validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=["settings_general.view"],
)
return services.list_settings(
db,
tenant_id,
company_id
)
@router.put("/", response_model=InvoiceSettingsResponse)
def save_invoice_settings(
settings_data: InvoiceSettingsRequest,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Create or update invoice settings"""
tenant_id = validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=["settings_general.edit"],
)
return services.upsert_settings(
db,
tenant_id,
company_id,
settings_data
)

View File

@@ -9,7 +9,7 @@ from core.security import get_current_user, validate_access_to_resource
from api.v1.modules.core.tasks_tracking import track_and_dispatch
from api.v1.modules.a76.invoices.models import InvoiceHeader, OperationType
from api.v1.modules.a76.invoices.routes import get_invoice_permission_base
from api.v1.modules.a76.invoices.permission_map import get_invoice_permission_base
from .task import process_invoice_task
from ...exports.process.task import process_export_invoice_task

View File

@@ -9,7 +9,7 @@ from core.security import get_current_user, validate_access_to_resource
from api.v1.modules.core.tasks_tracking import track_and_dispatch
from api.v1.modules.a76.invoices.models import InvoiceHeader, OperationType
from api.v1.modules.a76.invoices.routes import get_invoice_permission_base
from api.v1.modules.a76.invoices.permission_map import get_invoice_permission_base
from .task import revert_invoice_task as revert_import_invoice_task
from ...exports.revert.task import revert_invoice_task as revert_export_invoice_task

View File

@@ -0,0 +1,91 @@
"""
Única fuente de verdad para permisos de facturas por (operation_type, invoice_type).
Debe mantenerse alineado con el seed en api/v1/modules/core/permissions/seed.py
y con frontend/src/lib/permissions/invoice-permissions.ts.
"""
from __future__ import annotations
from typing import Any, Iterable, Optional
# Tipos de exportación que cubre únicamente invoice.exp.view (histórico list_invoices).
EXP_TYPES_FROM_GENERAL_EXPORT_VIEW: tuple[str, ...] = (
"EXDEF",
"MATDE",
"NODES",
"PTERM",
"SCRAP",
"VEMEX",
"VIRTU",
"AFIJO",
"REEXP",
)
def get_invoice_permission_base(operation_type: Any, invoice_type: Any) -> str:
"""Devuelve la clave base del permiso (sin .view/.create/...) según tipo de factura."""
op = str(operation_type).lower().split(".")[-1] if operation_type else ""
inv = str(invoice_type).upper() if invoice_type else ""
if op == "imp":
if inv == "TEM":
return "invoice.imp.tem"
if inv == "DEF":
return "invoice.imp.def"
if inv == "MEX":
return "invoice.imp.cm"
if inv == "CR":
return "invoice.imp.cr"
if inv == "REP":
return "invoice.imp.rep"
return "invoice.imp.tem"
if op == "exp":
if inv == "REPAR":
return "invoice.exp.rep"
if inv == "DONAC":
return "invoice.exp.donac"
return "invoice.exp"
return "invoice.imp.tem"
def build_allowed_types_from_view_permissions(
perm_codes: Iterable[str],
) -> list[tuple[str, str]]:
"""
Construye la lista de tuplas (operation_type en minúsculas, invoice_type en MAYÚSCULAS)
permitidas por los permisos *.view del usuario. Misma lógica que el antiguo list_invoices
(sin permiso global invoice.view_all — eliminado de producto).
"""
codes = set(perm_codes)
allowed: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set()
def add_pair(op: str, inv: str) -> None:
pair = (op.lower(), inv.upper())
if pair not in seen:
seen.add(pair)
allowed.append(pair)
if "invoice.imp.tem.view" in codes:
add_pair("imp", "TEM")
if "invoice.imp.def.view" in codes:
add_pair("imp", "DEF")
if "invoice.imp.cm.view" in codes:
add_pair("imp", "MEX")
if "invoice.imp.cr.view" in codes:
add_pair("imp", "CR")
if "invoice.imp.rep.view" in codes:
add_pair("imp", "REP")
if "invoice.exp.rep.view" in codes:
add_pair("exp", "REPAR")
if "invoice.exp.donac.view" in codes:
add_pair("exp", "DONAC")
if "invoice.exp.view" in codes:
for t in EXP_TYPES_FROM_GENERAL_EXPORT_VIEW:
add_pair("exp", t)
return allowed

View File

@@ -1,4 +1,7 @@
import logging
from typing import Dict, Any, Optional
from core.config import settings
from core.database import get_core_db
from core.security import collect_user_role_names, get_current_user, validate_access_to_resource
from fastapi import APIRouter, Depends, HTTPException, Query, Path
@@ -7,28 +10,14 @@ from sqlalchemy.orm import Session
from . import schemas, services, models
from .catalog_service import InvoiceCatalogService
from .permission_map import (
build_allowed_types_from_view_permissions,
get_invoice_permission_base,
)
# Create main router
router = APIRouter()
# --- 🛡️ FUNCIÓN EVALUADORA DE PERMISOS DINÁMICOS ---
def get_invoice_permission_base(operation_type: Any, invoice_type: Any) -> str:
"""Devuelve la clave base del permiso dependiendo del tipo de factura"""
# Limpiamos los valores por si vienen como Enums
op = str(operation_type).lower().split('.')[-1] if operation_type else ''
inv = str(invoice_type).upper() if invoice_type else ''
if op == 'imp':
if inv == 'TEM': return 'invoice.imp.tem'
if inv == 'DEF': return 'invoice.imp.def'
if inv == 'MEX': return 'invoice.imp.cm'
if inv == 'CR': return 'invoice.imp.cr'
return 'invoice.imp.tem' # Fallback
elif op == 'exp':
if inv == 'REPAR': return 'invoice.exp.rep'
return 'invoice.exp'
return 'invoice.imp.tem' # Fallback general
logger = logging.getLogger(__name__)
# --- RUTAS DE UTILIDAD ---
@@ -36,17 +25,50 @@ def get_invoice_permission_base(operation_type: Any, invoice_type: Any) -> str:
@router.get("/invoices/creation-data", response_model=schemas.InvoiceCreationResponse)
def get_creation_data(
company_id: int = Query(..., description="Company ID"),
operation_type: schemas.OperationType = Query(..., description="Operation type for the new invoice"),
invoice_type: Optional[str] = Query(
None,
description="Invoice type key (required for import; optional for export — uses invoice.exp.create)",
),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get consolidated data for creating a new invoice"""
try:
tenant_id = validate_access_to_resource(db, company_id, current_user)
ot = operation_type.value if hasattr(operation_type, "value") else operation_type
if ot == "imp":
if not invoice_type:
raise HTTPException(
status_code=422,
detail="invoice_type is required for import invoices",
)
perm_base = get_invoice_permission_base(operation_type, invoice_type)
validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=[f"{perm_base}.create"],
)
else:
if invoice_type:
perm_base = get_invoice_permission_base(operation_type, invoice_type)
validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=[f"{perm_base}.create"],
)
else:
validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=["invoice.exp.create"],
)
return InvoiceCatalogService.get_creation_data(db, tenant_id, company_id)
except Exception as e:
import traceback
print(f"[ERROR] get_creation_data failed: {str(e)}")
traceback.print_exc()
logger.exception("get_creation_data failed: %s", e)
raise HTTPException(status_code=500, detail=f"Error al cargar datos de creación: {str(e)}")
@router.get("/invoices/{invoice_id}/edition-data", response_model=schemas.InvoiceEditionResponse)
@@ -74,19 +96,68 @@ def get_edition_data(
except HTTPException:
raise
except Exception as e:
import traceback
print(f"[ERROR] get_edition_data failed: {str(e)}")
traceback.print_exc()
logger.exception("get_edition_data failed: %s", e)
raise HTTPException(status_code=500, detail=f"Error al cargar datos de edición: {str(e)}")
@router.get("/invoices/remesa-suggestion", response_model=Dict[str, int])
def get_remesa_suggestion(
pedimento_id: int = Query(..., description="Pedimento ID"),
company_id: int = Query(..., description="Company ID"),
operation_type: Optional[schemas.OperationType] = Query(
None,
description="Required when no invoice exists yet for this pedimento (context for permission)",
),
invoice_type: Optional[str] = Query(
None,
description="Required when no invoice exists yet for this pedimento (context for permission)",
),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""
Sugiere siguiente remesa. Si ya hay facturas ligadas al pedimento, el permiso se infiere de ellas.
Si no hay facturas en compliance para ese pedimento, pasar operation_type e invoice_type del alta en curso.
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
invoice_for_perm = (
db.query(models.InvoiceHeader)
.join(
models.InvoiceComplianceMx,
models.InvoiceComplianceMx.invoice_id == models.InvoiceHeader.id,
)
.filter(
models.InvoiceComplianceMx.pedimento_id == pedimento_id,
models.InvoiceHeader.tenant_id == tenant_id,
models.InvoiceHeader.company_id == company_id,
)
.order_by(models.InvoiceHeader.id.desc())
.first()
)
if invoice_for_perm:
perm_base = get_invoice_permission_base(
invoice_for_perm.operation_type, invoice_for_perm.invoice_type
)
validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=[f"{perm_base}.edit"],
)
elif operation_type is not None and invoice_type:
perm_base = get_invoice_permission_base(operation_type, invoice_type)
validate_access_to_resource(
db,
company_id,
current_user,
required_permissions=[f"{perm_base}.edit"],
)
else:
raise HTTPException(
status_code=422,
detail="operation_type and invoice_type are required when no invoice is linked to this pedimento yet",
)
max_rem = (
db.query(func.max(models.InvoiceComplianceMx.remesa))
.filter(
@@ -117,9 +188,7 @@ def create_invoice(
return services.InvoiceService.create(db, data, tenant_id, company_id)
except Exception as e:
import traceback
print(f"[ERROR] create_invoice failed: {str(e)}")
traceback.print_exc()
logger.exception("create_invoice failed: %s", e)
raise HTTPException(status_code=500, detail=f"Error al guardar factura: {str(e)}")
@router.get("/invoices/{invoice_id}", response_model=schemas.InvoiceHeaderResponse)
@@ -203,56 +272,38 @@ def list_invoices(
tenant_id = validate_access_to_resource(db, company_id, current_user)
# Roles vienen del Hub (/auth/me), no de realm_access del JWT crudo.
is_hub_admin = "admin" in collect_user_role_names(current_user)
allowed_filters = []
# Información del usuario para debugging (se ve en los logs del servidor)
user_name = current_user.get('preferred_username') or current_user.get('email', 'Desconocido')
from api.v1.modules.core.permissions.service import PermissionService
user_id = current_user.get("sub") or current_user.get("id")
perm_service = PermissionService(db)
perm_codes = perm_service.get_user_permissions(user_id, company_id)
# 🕵️ DEBUG LOGS - Cruciales para diagnosticar filtrado que no funciona
print(f"[AUTH] User: {user_name} (ID: {user_id})")
print(f"[AUTH] App Permissions (C{company_id}): {perm_codes}")
# Definimos si debe saltar el filtrado granular (SOLO con permiso explícito)
has_global_view = "invoice.view_all" in perm_codes
# El rol de admin de Keycloak ya NO otorga bypass automático si hay permisos granulares
if has_global_view:
print(f"[AUTH] GLOBAL ACCESS for {user_name}")
allowed_filters = None
else:
# Aplicamos filtros basados en permisos específicos
# Importaciones
if "invoice.imp.tem.view" in perm_codes: allowed_filters.append(("imp", "TEM"))
if "invoice.imp.def.view" in perm_codes: allowed_filters.append(("imp", "DEF"))
if "invoice.imp.cm.view" in perm_codes: allowed_filters.append(("imp", "MEX"))
if "invoice.imp.cr.view" in perm_codes: allowed_filters.append(("imp", "CR"))
if "invoice.imp.rep.view" in perm_codes: allowed_filters.append(("imp", "REP"))
# Exportaciones
if "invoice.exp.rep.view" in perm_codes: allowed_filters.append(("exp", "REPAR"))
if "invoice.exp.donac.view" in perm_codes: allowed_filters.append(("exp", "DONAC"))
# Permiso general de exportación
if "invoice.exp.view" in perm_codes:
for t in ["EXDEF", "MATDE", "NODES", "PTERM", "SCRAP", "VEMEX", "VIRTU", "AFIJO", "REEXP"]:
if ("exp", t) not in allowed_filters:
allowed_filters.append(("exp", t))
if settings.ENVIRONMENT == "development":
user_name = current_user.get("preferred_username") or current_user.get(
"email", "Desconocido"
)
logger.debug(
"[invoices.list] user=%s id=%s company=%s perms=%s",
user_name,
user_id,
company_id,
perm_codes,
)
print(f"[AUTH] Filtered access for {user_name}. Allowed types count: {len(allowed_filters)}")
allowed_filters = build_allowed_types_from_view_permissions(perm_codes)
if not allowed_filters:
# Si no tiene ningún permiso de factura, bloqueamos
# Excepto si es admin (Hub roles / Keycloak), fallback a ver todo.
if is_hub_admin:
print(f"[AUTH] Keycloak Admin {user_name} has no app permissions. Granting view_all as fallback.")
allowed_filters = None
else:
raise HTTPException(status_code=403, detail="No tienes permisos para ver facturas en esta empresa")
if not allowed_filters:
if is_hub_admin:
logger.debug(
"[invoices.list] hub admin without invoice view perms — unfiltered types fallback"
)
allowed_filters = None
else:
raise HTTPException(
status_code=403,
detail="No tienes permisos para ver facturas en esta empresa",
)
try:
skip = (page - 1) * page_size
@@ -283,9 +334,7 @@ def list_invoices(
"page_size": page_size
}
except Exception as e:
import traceback
print(f"[ERROR] list_invoices failed: {str(e)}")
traceback.print_exc()
logger.exception("list_invoices failed: %s", e)
raise HTTPException(status_code=500, detail=f"Internal server error in invoices list: {str(e)}")