feature/optimizacion-de-permisos

backend/core/celery_app.py

@@ -8,7 +8,7 @@ from core.config import settings
 
 logger = logging.getLogger(__name__)
 
-valkey_url = os.getenv("VALKEY_URL", "redis://valkey:6379/0")
+valkey_url = settings.VALKEY_URL
 print(f"DEBUG: Celery Broker URL: {valkey_url}")
 logger.info(
     "Initializing Celery app app_version=%s environment=%s broker=%s",

backend/core/config.py

@@ -24,13 +24,16 @@ class Settings(BaseSettings):
     CORE_DB_USER: str = "postgres"
     CORE_DB_PASSWORD: str = "postgres"
 
-
-
     # Security
     SECRET_KEY: str = "change-this-secret-key-in-production"
     ALGORITHM: str = "HS256"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
 
+    # Valkey / Redis
+    VALKEY_URL: str = "redis://valkey:6379/0"
+    PERMISSION_CACHE_ENABLED: bool = True
+    PERMISSION_CACHE_TTL_SECONDS: int = 300
+
     # Synchronization
     SYNC_SECRET_TOKEN: str = "change-this-sync-token-in-production"
     CENTRAL_SERVER_URL: str = "http://localhost:8000/api/v1/core/help-center/sync/"

backend/core/security.py

@@ -476,13 +476,59 @@ def resolve_effective_tenant_id_from_user(current_user: Dict[str, Any]) -> Optio
 
 
 def is_hub_admin(current_user: Dict[str, Any]) -> bool:
-    """True si el usuario tiene el rol hub_admin (super-admin del Hub con acceso global)."""
-    roles = current_user.get("roles")
-    if isinstance(roles, list) and "hub_admin" in roles:
-        return True
+    """True si el Hub atestigua que el usuario es hub_admin (super-admin global)."""
     return bool(current_user.get("is_hub_admin"))
 
 
+def _has_local_super_admin_role(
+    db: "Session", user_id: Optional[str], company_id: Optional[int]
+) -> bool:
+    """
+    True si el usuario tiene el rol local ``super_admin`` activo en la compañía.
+
+    Sustituye al antiguo bypass por rol ``admin`` del realm Keycloak para
+    autorización: la fuente de verdad es la BD local (``core.user_company_roles``
+    + ``core.company_roles``), no claims del JWT. La promoción automática de
+    admins de Keycloak a ``super_admin`` local sigue ocurriendo en el endpoint
+    ``/permissions/me`` (bootstrap), por lo que un admin del realm que entre
+    al sistema sigue obteniendo el bypass sin coordinación manual.
+    """
+    if not user_id or not company_id:
+        return False
+    try:
+        from api.v1.modules.core.permissions.models import (
+            CompanyRole,
+            UserCompanyRole,
+        )
+
+        return (
+            db.query(UserCompanyRole)
+            .join(CompanyRole, CompanyRole.id == UserCompanyRole.company_role_id)
+            .filter(
+                UserCompanyRole.user_id == user_id,
+                UserCompanyRole.company_id == company_id,
+                UserCompanyRole.is_active == True,
+                CompanyRole.code == "super_admin",
+                CompanyRole.is_active == True,
+            )
+            .first()
+            is not None
+        )
+    except Exception as exc:
+        # Un fallo de BD aquí no debe escalar a acceso silenciosamente:
+        # se loguea y se trata como "no es super_admin" (deniega bypass).
+        logger.warning(
+            "has_local_super_admin_role_failed",
+            extra={
+                "op": "has_local_super_admin_role",
+                "user_id": user_id,
+                "company_id": company_id,
+                "error": str(exc),
+            },
+        )
+        return False
+
+
 def resolve_tenant_id_required(
     current_user: Dict[str, Any],
     db: Optional["Session"] = None,
@@ -720,15 +766,20 @@ def validate_access_to_resource(
 
     tenant_id = resolve_effective_tenant_id_from_user(current_user)
 
-    # Admin global Keycloak / master, o hub_admin del Hub.
-    all_user_roles = collect_user_role_names(current_user)
-    is_keycloak_admin = "admin" in all_user_roles or is_hub_admin(current_user)
+    # Bypass de checks de permisos: hub_admin (atestado por el Hub en /auth/me)
+    # o rol local "super_admin" en la compañía (fuente de verdad: BD de a76).
+    # Se reemplazó el antiguo "admin" in realm_access.roles para que la
+    # autorización deje de depender de claims del JWT.
+    user_id = current_user.get("sub") or current_user.get("id")
+    is_global_admin = is_hub_admin(current_user) or _has_local_super_admin_role(
+        db, user_id, company_id
+    )
 
     # 🚪 EXCEPCIÓN ESPECIAL: Si es el endpoint /me, permitimos el paso para el Bootstrap
     # Detectamos si no se requieren permisos (típico de /me)
     is_me_endpoint = required_permissions is None
-    
-    if not is_keycloak_admin and not is_me_endpoint:
+
+    if not is_global_admin and not is_me_endpoint:
         if not validate_company_access(db, company_id, current_user):
             raise HTTPException(status_code=403, detail="Access denied to this company")
     
@@ -747,14 +798,15 @@ def validate_access_to_resource(
             raise HTTPException(status_code=500, detail="Error al resolver el tenant_id")
 
     # Si aún no hay tenant_id y no es admin, error 400
-    if not tenant_id and not is_keycloak_admin and not is_me_endpoint:
+    if not tenant_id and not is_global_admin and not is_me_endpoint:
         raise HTTPException(status_code=400, detail="Tenant ID not found in token")
 
     # Verificar permisos locales
     if required_permissions:
-        if is_keycloak_admin:
-            # hub_admin siempre debe tener tenant_id resuelto cuando se exigen permisos;
-            # retornar 1 silenciosamente sería acceso al tenant equivocado
+        if is_global_admin:
+            # hub_admin / super_admin local: siempre debe tener tenant_id resuelto
+            # cuando se exigen permisos; retornar 1 silenciosamente sería acceso
+            # al tenant equivocado.
             if tenant_id is None:
                 raise HTTPException(
                     status_code=400,
@@ -801,7 +853,7 @@ def validate_access_to_resource(
     # significaría acceso al tenant equivocado. Si llegamos aquí sin tenant_id
     # válido para un usuario no-admin, es un estado inconsistente que debe fallar.
     if not isinstance(tenant_id, int) or tenant_id <= 0:
-        if not is_keycloak_admin:
+        if not is_global_admin:
             raise HTTPException(
                 status_code=400,
                 detail="No se pudo determinar el tenant_id para la empresa especificada",