fix: solución de bloqueos y estandarización de permisos

backend/api/v1/modules/a76/invoices/imports/process/routes.py

@@ -9,6 +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 .task import process_invoice_task
 from ...exports.process.task import process_export_invoice_task
 
@@ -34,6 +35,10 @@ def trigger_invoice_process(
     if invoice is None:
         raise HTTPException(status_code=404, detail=f"Factura {invoice_id} no encontrada.")
 
+    # 🛡️ Validación de permisos granulares
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.process"])
+
     if invoice.operation_type == OperationType.EXP:
         task = track_and_dispatch(
             db=db,

backend/api/v1/modules/a76/invoices/imports/revert/routes.py

@@ -9,6 +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 .task import revert_invoice_task as revert_import_invoice_task
 from ...exports.revert.task import revert_invoice_task as revert_export_invoice_task
 
@@ -40,6 +41,10 @@ def trigger_invoice_revert(
     if invoice is None:
         raise HTTPException(status_code=404, detail=f"Factura {invoice_id} no encontrada.")
 
+    # 🛡️ Validación de permisos granulares: des-actualizar requiere el mismo permiso que procesar
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.process"])
+
     if invoice.operation_type == OperationType.EXP:
         task = track_and_dispatch(
             db=db,

backend/api/v1/modules/a76/invoices/models.py

@@ -773,4 +773,3 @@ class InvoiceCollections(Base, TenantScopedMixin, TimestampMixin):
 
     # Relationship
     header: Mapped["InvoiceHeader"] = relationship(back_populates="collections")
-    concept: Mapped[Optional[str]] = mapped_column(String(100))  # CONCEPTO / Conce

backend/api/v1/modules/a76/invoices/routes.py

@@ -1,9 +1,8 @@
 from typing import Dict, Any, Optional
-from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
 from core.database import get_core_db
 from core.security import get_current_user, validate_access_to_resource
 from fastapi import APIRouter, Depends, HTTPException, Query, Path
-from sqlalchemy import func
+from sqlalchemy import func, or_, and_
 from sqlalchemy.orm import Session
 
 from . import schemas, services, models
@@ -12,6 +11,28 @@ from .catalog_service import InvoiceCatalogService
 # 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
+
+
+# --- RUTAS DE UTILIDAD ---
+
 @router.get("/invoices/creation-data", response_model=schemas.InvoiceCreationResponse)
 def get_creation_data(
     company_id: int = Query(..., description="Company ID"),
@@ -19,8 +40,14 @@ def get_creation_data(
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     """Get consolidated data for creating a new invoice"""
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
-    return InvoiceCatalogService.get_creation_data(db, tenant_id, company_id)
+    try:
+        tenant_id = validate_access_to_resource(db, company_id, current_user)
+        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()
+        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)
 def get_edition_data(
@@ -30,12 +57,27 @@ def get_edition_data(
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     """Get consolidated data for editing an existing invoice"""
-    tenant_id = validate_access_to_resource(db, company_id, current_user)
-    data = InvoiceCatalogService.get_edition_data(db, invoice_id, tenant_id, company_id)
-    if not data:
-        raise HTTPException(status_code=404, detail="Invoice not found")
-    return data
+    try:
+        tenant_id = validate_access_to_resource(db, company_id, current_user)
+        
+        # 1. Traer la factura para saber su tipo
+        invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+        if not invoice:
+            raise HTTPException(status_code=404, detail="Invoice not found")
+            
+        # 2. Validar permiso dinámico de Edición
+        perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+        validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
+        data = InvoiceCatalogService.get_edition_data(db, invoice_id, tenant_id, company_id)
+        return data
+    except HTTPException:
+        raise
+    except Exception as e:
+        import traceback
+        print(f"[ERROR] get_edition_data failed: {str(e)}")
+        traceback.print_exc()
+        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(
@@ -44,7 +86,6 @@ def get_remesa_suggestion(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Suggest next remesa for selected pedimento (max+1)."""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
     max_rem = (
         db.query(func.max(models.InvoiceComplianceMx.remesa))
@@ -58,31 +99,84 @@ def get_remesa_suggestion(
     return {"next_remesa": int((max_rem or 0) + 1)}
 
 
-# Create CRUD routes for Invoice Header using TenantCRUDRoutes
-invoice_crud = TenantCRUDRoutes(
-    service=services.InvoiceService,
-    create_schema=schemas.InvoiceHeaderCreate,
-    update_schema=schemas.InvoiceHeaderUpdate,
-    response_schema=schemas.InvoiceHeaderResponse,
-    prefix="/invoices",
-    tags=[],
-    resource_name="Invoice",
-    id_name="invoice_id",
-    id_type=int,    
-    enable_list=False,  # Disable auto-list to override with custom filter
-    enable_filters=True,  # Enable filters for status, operation_type, etc.
-    list_permissions=[],
-    get_permissions=[],
-    create_permissions=[],
-    update_permissions=[],
-    delete_permissions=[],
-    default_page_size=50,
-    max_page_size=200,
-)
+# --- RUTAS CRUD MANUALES (Sustituyen al TenantCRUDRoutes por seguridad) ---
 
-# Include the main CRUD routes
-router.include_router(invoice_crud.router)
+@router.post("/invoices/", response_model=schemas.InvoiceHeaderResponse)
+def create_invoice(
+    data: schemas.InvoiceHeaderCreate,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    try:
+        tenant_id = validate_access_to_resource(db, company_id, current_user)
+        
+        # Validamos usando los datos que vienen en el body (payload)
+        perm_base = get_invoice_permission_base(data.operation_type, data.invoice_type)
+        validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.create"])
+        
+        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()
+        raise HTTPException(status_code=500, detail=f"Error al guardar factura: {str(e)}")
 
+@router.get("/invoices/{invoice_id}", response_model=schemas.InvoiceHeaderResponse)
+def get_invoice(
+    invoice_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice:
+        raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
+    
+    return invoice
+
+@router.put("/invoices/{invoice_id}", response_model=schemas.InvoiceHeaderResponse)
+def update_invoice(
+    invoice_id: int,
+    data: schemas.InvoiceHeaderUpdate,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice:
+        raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
+    
+    return services.InvoiceService.update(db, invoice_id, tenant_id, data, company_id)
+
+@router.delete("/invoices/{invoice_id}")
+def delete_invoice(
+    invoice_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice:
+        raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.delete"])
+    
+    success = services.InvoiceService.delete(db, invoice_id, tenant_id, company_id)
+    return {"success": success}
+
+
+# --- RUTA DE LISTADO (FILTROS) ---
 
 @router.get("/invoices/", response_model=schemas.InvoiceHeaderListResponse)
 def list_invoices(
@@ -104,101 +198,133 @@ def list_invoices(
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     """
-    List invoices with optional filters, including manifest_number.
+    List invoices with optional filters and granular permission enforcement.
     """
-    print(f"DEBUG: list_invoices called with manifest_number={manifest_number}")
     tenant_id = validate_access_to_resource(db, company_id, current_user)
-    print(f"DEBUG: tenant_id={tenant_id}, company_id={company_id}")
+    user_roles = current_user.get("realm_access", {}).get("roles", [])
+    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')
     
-    skip = (page - 1) * page_size
-    filters = {
-        "invoice_number": invoice_number or search,
-        "status": status,
-        "operation_type": operation_type,
-        "invoice_type": invoice_type,
-        "manifest_number": manifest_number,
-        "pedimento": pedimento,
-        "project_number": project_number,
-        "year": year,
-    }
+    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)
     
-    # Remove None values
-    filters = {k: v for k, v in filters.items() if v is not None}
+    # 🕵️ 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}")
 
-    items, total = services.InvoiceService.get_all(
-        db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters, sort_by=sort_by, sort_order=sort_order
-    )
-    print(f"DEBUG: InvoiceService returned {len(items)} items, total={total}")
+    # Definimos si debe saltar el filtrado granular (SOLO con permiso explícito)
+    has_global_view = "invoice.view_all" in perm_codes
     
-    return {
-        "items": items,
-        "total": total,
-        "page": page,
-        "page_size": page_size
-    }
+    # 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))
+
+        print(f"[AUTH] Filtered access for {user_name}. Allowed types count: {len(allowed_filters)}")
+
+        if not allowed_filters:
+            # Si no tiene ningún permiso de factura, bloqueamos
+            # Excepto si es un admin de Keycloak, le damos el beneficio de la duda pero logeamos
+            if "admin" in user_roles:
+                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")
+
+    try:
+        skip = (page - 1) * page_size
+        
+        # Combinar filtros de búsqueda con filtros granulares de permisos
+        filters = {
+            "invoice_number": invoice_number,
+            "status": status,
+            "operation_type": operation_type.value if operation_type else None,
+            "invoice_type": invoice_type,
+            "manifest_number": manifest_number,
+            "pedimento": pedimento,
+            "project_number": project_number,
+            "year": year,
+            "allowed_types": allowed_filters
+        }
+        
+        filters = {k: v for k, v in filters.items() if v is not None}
+
+        items, total = services.InvoiceService.get_all(
+            db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters, sort_by=sort_by, sort_order=sort_order
+        )
+        
+        return {
+            "items": items,
+            "total": total,
+            "page": page,
+            "page_size": page_size
+        }
+    except Exception as e:
+        import traceback
+        print(f"[ERROR] list_invoices failed: {str(e)}")
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"Internal server error in invoices list: {str(e)}")
 
 
-# Additional nested routes for child resources
+# --- RUTAS DE LOGÍSTICA ---
 
-# --- Logistics Routes ---
-
-@router.get(
-    "/invoices/{invoice_id}/logistics",
-    response_model=list[schemas.InvoiceLogisticsResponse],
-    summary="Get all logistics for an invoice",
-)
+@router.get("/invoices/{invoice_id}/logistics", response_model=list[schemas.InvoiceLogisticsResponse])
 def get_invoice_logistics(
     invoice_id: int = Path(..., description="Invoice ID"),
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Get all logistics entries for a specific invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
+    return services.InvoiceLogisticsService.get_all_by_invoice(db, invoice_id)
 
-    logistics = services.InvoiceLogisticsService.get_all_by_invoice(
-        db, invoice_id)
-    return logistics
-
-
-@router.post(
-    "/invoices/{invoice_id}/logistics",
-    response_model=schemas.InvoiceLogisticsResponse,
-    status_code=201,
-    summary="Add logistics to an invoice",
-)
+@router.post("/invoices/{invoice_id}/logistics", response_model=schemas.InvoiceLogisticsResponse, status_code=201)
 def create_invoice_logistics(
+    logistics_data: schemas.InvoiceLogisticsCreate,
     invoice_id: int = Path(..., description="Invoice ID"),
-    logistics_data: schemas.InvoiceLogisticsCreate = ...,
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Add a new logistics entry to an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    # Editar los hijos cuenta como editar la factura padre
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
+    return services.InvoiceLogisticsService.create(db, logistics_data, invoice_id, tenant_id, company_id)
 
-    logistics = services.InvoiceLogisticsService.create(
-        db, logistics_data, invoice_id, tenant_id, company_id)
-    return logistics
-
-
-@router.delete(
-    "/invoices/{invoice_id}/logistics/{logistics_id}",
-    status_code=204,
-    summary="Delete logistics from an invoice",
-)
+@router.delete("/invoices/{invoice_id}/logistics/{logistics_id}", status_code=204)
 def delete_invoice_logistics(
     invoice_id: int = Path(..., description="Invoice ID"),
     logistics_id: int = Path(..., description="Logistics ID"),
@@ -206,84 +332,54 @@ def delete_invoice_logistics(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Delete a logistics entry from an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
-
-    success = services.InvoiceLogisticsService.delete(
-        db, logistics_id, invoice_id)
-    if not success:
-        raise HTTPException(
-            status_code=404, detail="Logistics entry not found")
-
+    if not services.InvoiceLogisticsService.delete(db, logistics_id, invoice_id):
+        raise HTTPException(status_code=404, detail="Logistics entry not found")
     return None
 
 
-# --- Sales Details Routes ---
+# --- RUTAS DE DETALLES DE VENTA (PARTIDAS) ---
 
-@router.get(
-    "/invoices/{invoice_id}/details",
-    response_model=list[schemas.InvoiceSalesDetailsResponse],
-    summary="Get all sales details for an invoice",
-)
+@router.get("/invoices/{invoice_id}/details", response_model=list[schemas.InvoiceSalesDetailsResponse])
 def get_invoice_details(
     invoice_id: int = Path(..., description="Invoice ID"),
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Get all sales details for a specific invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
-
-    details = services.InvoiceSalesDetailsService.get_all_by_invoice(
-        db, invoice_id)
-    return details
-
-
-@router.post(
-    "/invoices/{invoice_id}/details",
-    response_model=schemas.InvoiceSalesDetailsResponse,
-    status_code=201,
-    summary="Add sales detail to an invoice",
-)
+    return services.InvoiceSalesDetailsService.get_all_by_invoice(db, invoice_id)
 
+@router.post("/invoices/{invoice_id}/details", response_model=schemas.InvoiceSalesDetailsResponse, status_code=201)
 def create_invoice_detail(
+    detail_data: schemas.InvoiceSalesDetailsCreate,
     invoice_id: int = Path(..., description="Invoice ID"),
-    detail_data: schemas.InvoiceSalesDetailsCreate = ...,
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Add a new sales detail to an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
+    return services.InvoiceSalesDetailsService.create(db, detail_data, invoice_id, tenant_id, company_id)
 
-    detail = services.InvoiceSalesDetailsService.create(
-        db, detail_data, invoice_id, tenant_id, company_id)
-    return detail
-
-
-@router.delete(
-    "/invoices/{invoice_id}/details/{detail_id}",
-    status_code=204,
-    summary="Delete sales detail from an invoice",
-)
+@router.delete("/invoices/{invoice_id}/details/{detail_id}", status_code=204)
 def delete_invoice_detail(
     invoice_id: int = Path(..., description="Invoice ID"),
     detail_id: int = Path(..., description="Detail ID"),
@@ -291,82 +387,54 @@ def delete_invoice_detail(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Delete a sales detail from an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
-
-    success = services.InvoiceSalesDetailsService.delete(
-        db, detail_id, invoice_id)
-    if not success:
+    if not services.InvoiceSalesDetailsService.delete(db, detail_id, invoice_id):
         raise HTTPException(status_code=404, detail="Sales detail not found")
-
     return None
 
 
-# --- Collections Routes ---
+# --- RUTAS DE COBRANZA (COLLECTIONS) ---
 
-@router.get(
-    "/invoices/{invoice_id}/collections",
-    response_model=list[schemas.InvoiceCollectionsResponse],
-    summary="Get all collections for an invoice",
-)
+@router.get("/invoices/{invoice_id}/collections", response_model=list[schemas.InvoiceCollectionsResponse])
 def get_invoice_collections(
     invoice_id: int = Path(..., description="Invoice ID"),
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Get all collections for a specific invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.view"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
+    return services.InvoiceCollectionsService.get_all_by_invoice(db, invoice_id)
 
-    collections = services.InvoiceCollectionsService.get_all_by_invoice(
-        db, invoice_id)
-    return collections
-
-
-@router.post(
-    "/invoices/{invoice_id}/collections",
-    response_model=schemas.InvoiceCollectionsResponse,
-    status_code=201,
-    summary="Add collection to an invoice",
-)
+@router.post("/invoices/{invoice_id}/collections", response_model=schemas.InvoiceCollectionsResponse, status_code=201)
 def create_invoice_collection(
+    collection_data: schemas.InvoiceCollectionsCreate,
     invoice_id: int = Path(..., description="Invoice ID"),
-    collection_data: schemas.InvoiceCollectionsCreate = ...,
     company_id: int = Query(..., description="Company ID"),
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Add a new collection to an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
+    return services.InvoiceCollectionsService.create(db, collection_data, invoice_id, tenant_id, company_id)
 
-    collection = services.InvoiceCollectionsService.create(
-        db, collection_data, invoice_id, tenant_id, company_id)
-    return collection
-
-
-@router.delete(
-    "/invoices/{invoice_id}/collections/{collection_id}",
-    status_code=204,
-    summary="Delete collection from an invoice",
-)
+@router.delete("/invoices/{invoice_id}/collections/{collection_id}", status_code=204)
 def delete_invoice_collection(
     invoice_id: int = Path(..., description="Invoice ID"),
     collection_id: int = Path(..., description="Collection ID"),
@@ -374,18 +442,13 @@ def delete_invoice_collection(
     db: Session = Depends(get_core_db),
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
-    """Delete a collection from an invoice"""
     tenant_id = validate_access_to_resource(db, company_id, current_user)
+    invoice = services.InvoiceService.get_by_id(db, invoice_id, tenant_id, company_id)
+    if not invoice: raise HTTPException(status_code=404, detail="Invoice not found")
+        
+    perm_base = get_invoice_permission_base(invoice.operation_type, invoice.invoice_type)
+    validate_access_to_resource(db, company_id, current_user, required_permissions=[f"{perm_base}.edit"])
 
-    # Verify the invoice exists and belongs to the tenant/company
-    invoice = services.InvoiceService.get_by_id(
-        db, invoice_id, tenant_id, company_id)
-    if not invoice:
-        raise HTTPException(status_code=404, detail="Invoice not found")
-
-    success = services.InvoiceCollectionsService.delete(
-        db, collection_id, invoice_id)
-    if not success:
+    if not services.InvoiceCollectionsService.delete(db, collection_id, invoice_id):
         raise HTTPException(status_code=404, detail="Collection not found")
-
-    return None
+    return None

backend/api/v1/modules/a76/invoices/services.py

@@ -326,6 +326,35 @@ class InvoiceService:
             if not filters.get("invoice_type") and ot_exp_val == "exp":
                 query = query.filter(models.InvoiceHeader.invoice_type != "REPAR")
 
+            # Filtro por permisos granulares (allowed_types)
+            if "allowed_types" in filters:
+                from sqlalchemy import or_, and_
+                allowed = filters["allowed_types"]
+                if allowed is None:
+                    # Acceso global (admin o view_all) - no filtramos por tipos
+                    pass
+                elif not allowed:
+                    # Seguridad: Si el usuario NO tiene permisos para ningún tipo específico
+                    query = query.filter(models.InvoiceHeader.id == -1)
+                else:
+                    from sqlalchemy import func
+                    conditions = []
+                    for op, inv in allowed:
+                        # Aseguramos comparación insensible a mayúsculas para mayor robustez con la DB
+                        op_str = str(op).lower()
+                        inv_str = str(inv).lower()
+                        conditions.append(
+                            and_(
+                                func.lower(models.InvoiceHeader.operation_type) == op_str,
+                                func.lower(models.InvoiceHeader.invoice_type) == inv_str
+                            )
+                        )
+                    if conditions:
+                        query = query.filter(or_(*conditions))
+                    else:
+                        # Seguridad: Si tiene allowed_types pero no generamos condiciones, no debe ver nada
+                        query = query.filter(models.InvoiceHeader.id == -1)
+
         # Apply sorting
         if sort_by:
             # Simple column mapping