refactor: update role handling in security and permissions modules, and change hub-net to non-external in docker-compose

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

@@ -1,6 +1,6 @@
 from typing import Dict, Any, Optional
 from core.database import get_core_db
-from core.security import get_current_user, validate_access_to_resource
+from core.security import collect_user_role_names, get_current_user, validate_access_to_resource
 from fastapi import APIRouter, Depends, HTTPException, Query, Path
 from sqlalchemy import func, or_, and_
 from sqlalchemy.orm import Session
@@ -201,8 +201,9 @@ def list_invoices(
     List invoices with optional filters and granular permission enforcement.
     """
     tenant_id = validate_access_to_resource(db, company_id, current_user)
-    user_roles = current_user.get("realm_access", {}).get("roles", [])
-    allowed_filters = [] 
+    # 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')
@@ -246,8 +247,8 @@ def list_invoices(
 
         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:
+            # 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:

backend/api/v1/modules/core/permissions/routes.py

@@ -8,7 +8,7 @@ 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 core.security import collect_user_role_names, get_current_user, validate_access_to_resource
 from .dependencies import (
     PermissionChecker,
     RequirePermission,
@@ -64,11 +64,8 @@ async def get_my_permissions(
     user_id = current_user.get("sub") or current_user.get("id")
 
     # 2. Determinar si es un admin de Keycloak para forzar bootstrap si es necesario
-    realm_roles = current_user.get("realm_access", {}).get("roles", [])
-    client_roles = []
-    for client in current_user.get("resource_access", {}).values():
-        client_roles.extend(client.get("roles", []))
-    is_keycloak_admin = "admin" in realm_roles or "admin" in client_roles
+    # Roles vienen del Hub (/auth/me vía verify_token); ver collect_user_role_names.
+    is_keycloak_admin = "admin" in collect_user_role_names(current_user)
 
     # 3. Bootstrap: si la compañía no tiene roles, o si el usuario es admin, o si estamos en desarrollo y el usuario no tiene roles
     from .models import CompanyRole, UserCompanyRole

backend/core/middleware.py

@@ -144,6 +144,16 @@ class LicenseValidationMiddleware(BaseHTTPMiddleware):
             # Fallback para flujos SSO cuando el override no viaja en header.
             tenant_override = request.cookies.get("sso_tenant_id") or request.cookies.get("sso_tenant_pub")
 
+        # TenantMiddleware (corre antes) ya resolvió el token y dejó tenant en user_info.
+        # Sin esto, Swagger/curl sin cookies SSO llaman verify-license sin contexto y el Hub
+        # puede devolver 401 aunque /auth/me con el mismo Bearer responda 200.
+        if not tenant_override:
+            user_info = getattr(request.state, "user_info", None)
+            if isinstance(user_info, dict):
+                tid = user_info.get("tenant_id")
+                if tid is not None and str(tid).strip() != "":
+                    tenant_override = str(tid)
+
         hub_headers = {"Authorization": f"Bearer {token}"}
         if tenant_override:
             hub_headers["X-Tenant-Override"] = str(tenant_override)

backend/core/security.py

@@ -9,7 +9,7 @@ from fastapi import Depends, HTTPException, Request, Security
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from jose import JWTError, jwt
 import httpx
-from cachetools import TTLCache
+from cachetools import TTLCache 
 from sqlalchemy.orm import Session
 from sqlalchemy.exc import IntegrityError
 
@@ -229,13 +229,14 @@ def has_role(required_role: str):
     async def role_checker(
         current_user: Dict[str, Any] = Depends(get_current_user),
     ) -> Dict[str, Any]:
-        user_roles = current_user.get("realm_access", {}).get("roles", [])
+        user_roles = collect_user_role_names(current_user)
 
         if required_role not in user_roles:
-            logger.warning(f"Role denied. Required: {required_role}. User actually has: {user_roles}")
-            # Also check client roles as a debug fallback
-            client_roles = current_user.get("resource_access", {})
-            logger.warning(f"User client roles: {client_roles}")
+            logger.warning(
+                "Role denied. Required: %s. User has: %s",
+                required_role,
+                sorted(user_roles),
+            )
             raise HTTPException(
                 status_code=403,
                 detail=f"User does not have required role: {required_role}",
@@ -267,6 +268,29 @@ def get_tenant_from_token(user_info: Dict[str, Any]) -> Optional[int]:
     return None
 
 
+def collect_user_role_names(current_user: Dict[str, Any]) -> Set[str]:
+    """
+    Roles del usuario: primero la lista ``roles`` del Hub (GET /api/v1/auth/me
+    vía verify_token). Si no hay lista no vacía, se unen realm_access y
+    resource_access del JWT Keycloak clásico.
+    """
+    names: Set[str] = set()
+    hub_roles = current_user.get("roles")
+    if isinstance(hub_roles, list):
+        names.update(str(r) for r in hub_roles if r is not None)
+
+    if names:
+        return names
+
+    realm = current_user.get("realm_access")
+    if isinstance(realm, dict):
+        names.update(str(r) for r in (realm.get("roles") or []) if r is not None)
+    for client in (current_user.get("resource_access") or {}).values():
+        if isinstance(client, dict):
+            names.update(str(r) for r in (client.get("roles") or []) if r is not None)
+    return names
+
+
 def validate_company_access(
     db: Session, company_id: int, current_user: Dict[str, Any]
 ) -> bool:
@@ -334,23 +358,9 @@ def validate_access_to_resource(
     tenant_id = get_tenant_from_token(current_user)
     if not tenant_id:
         tenant_id = current_user.get("tenant_id")
-    
-    # 🕵️ DEBUG ULTRA-DETALLADO (Ver en consola del backend)
-    print("--- TOKEN DEBUG START ---")
-    print(f"Usuario: {current_user.get('preferred_username')}")
-    print(f"Sub: {current_user.get('sub')}")
-    print(f"Realm Roles: {current_user.get('realm_access', {}).get('roles', [])}")
-    print(f"Resource Access: {current_user.get('resource_access', {})}")
-    print("--- TOKEN DEBUG END ---")
 
-    # 🛡️ DETERMINAR SI ES ADMIN DE KEYCLOAK
-    realm_roles = current_user.get("realm_access", {}).get("roles", [])
-    # Buscamos en todos los clientes posibles por si acaso
-    all_client_roles = []
-    for client in current_user.get("resource_access", {}).values():
-        all_client_roles.extend(client.get("roles", []))
-    
-    all_user_roles = set(realm_roles + all_client_roles)
+    # Admin global Keycloak / master: lista ``roles`` del Hub (/auth/me), con fallback JWT.
+    all_user_roles = collect_user_role_names(current_user)
     is_keycloak_admin = "admin" in all_user_roles
 
     # 🚪 EXCEPCIÓN ESPECIAL: Si es el endpoint /me, permitimos el paso para el Bootstrap

docker-compose.yml

@@ -310,5 +310,5 @@ networks:
     driver: bridge
 
   hub-net:
-    external: true
+    external: false
     name: aduanasoft-hub_default