CreaRack-SL

Auditoría Suprema Etapa 3: Correcciones de seguridad en config/IA

Funcionalidadactivecreado Thu Jun 11#security#audit#monitoring#ia#config#multi-tenancy

Resumen

Sesión 127 (2026-06-11): Primera tanda de arreglos tras la Auditoría Suprema, enfocada en el dominio configuración e inteligencia artificial (30 hallazgos totales).

Resultados:

  • ✅ 1 ALTA cerrada (fuga cross-tenant a Google AI)
  • ✅ 8 MEDIA/BAJA cerradas (timing attacks, clave JWT dedicada, redacción de logs, robustez)
  • ✅ 8 tests nuevos en tests/api/test_config_ia.py
  • ⏳ 5 ítems diferidos con motivo documentado (carve-out, T2, T4, orden middleware)

Lo que cambia

1. Anonimización de datos antes de LLM (ALTA)

Endpoints /api/workspace/perf-review?ai=true y /api/workspace/finops?ai=true:

Antes: Mandaban a Google AI Studio todos los datos agregados de todas las organizaciones (nombres reales + SQL crudo).

Ahora: Función _anonymize_for_ai() en core/workspace_api.py:

  • Sustituye nombres de org: "ACME Corp" → "org-1"
  • Redacta secretos en queries: password='hunter2' → password='***'
  • Preserva conteos (lo que el LLM necesita para analizar)
  • Trabaja sobre copia (response al caller mantiene nombres reales)
def _anonymize_for_ai(data):
    """Anonimiza tenant_usage + redacta secrets en slow_queries ANTES de LLM."""
    d = copy.deepcopy(data)
    for row in d.get("tenant_usage", []) or []:
        if isinstance(row, dict) and "name" in row:
            row["name"] = f"org-{row.get('id', '?')}"
    for q in d.get("slow_queries", []) or []:
        if isinstance(q, dict) and q.get("query"):
            q["query"] = redact_secrets(str(q["query"]))
    return d

Invocación:

def _ai_analyze_perf(data, tone="technical"):
    return AIOperations.detect_anomalies(_anonymize_for_ai(data), tone=tone)

2. Clave JWT dedicada (AGENT_JWT_SECRET)

terminal/api/auth.py + config/settings/base.py:

  • Nueva env var AGENT_JWT_SECRET (vacía por defecto = fallback inerte a SECRET_KEY)
  • Helper _agent_jwt_secret() usado en emisión + verificación REST + WebSocket
  • Fallback permite activación sin breaking change: declarar en Dokploy fuerza re-auth de flota

Beneficio: Rotación de claves sin cascada; si SECRET_KEY se filtra, tokens del Agent no se ven comprometidos.

3. Comparación de claves en tiempo constante

core/workspace_api.py:

  • _check_key() (validación de WORKSPACE_API_KEY) → hmac.compare_digest() (anti-timing)
  • config/urls.py (_InternalToolsAuth.authenticate() para INTERNAL_TOOLS_TOKEN) → hmac.compare_digest() (anti-timing)

Antes: if token == expected vulnerable a timing attacks (un atacante podría adivinar byte a byte).

Ahora: Comparación en tiempo constante.

4. Redacción de secretos en logs de providers IA

monitoring/services/ai_providers/:

  • google_genai.py, claude.py, ollama.py → errores del SDK pasan por redact_secrets() antes de loguear
  • Riesgo: El error original podría contener API keys o datos del tenant
except Exception as e:
    logger.error("Claude API error: %s", redact_secrets(str(e)))
    raise

5. Saneo del context en ai_fallback

core/services/ai_providers/router.py:

  • El dict de contexto (telemetría, nombres de device) se redacta antes de incrustarse en prompt
from monitoring.services.ai_providers.sanitize import redact_secrets
context_json = redact_secrets(json.dumps(context, indent=2, default=str))
prompt = f"Context:\n{context_json}\n\n{prompt}"

6. Mensajes de error genéricos al cliente

core/workspace_api.py + core/services/ai_operations.py:

  • Endpoints devuelven "AI analysis temporarily unavailable" en lugar de str(exc)
  • El detalle redactado va solo a logs
def _ai_analyze_perf(data, tone="technical"):
    try:
        return AIOperations.detect_anomalies(_anonymize_for_ai(data), tone=tone)
    except Exception as e:
        logger.warning("workspace perf-review AI analysis failed: %s", redact_secrets(str(e)))
        return "AI analysis temporarily unavailable"

7. Robustez de drivers IA

monitoring/services/ai_providers/google_genai.py:

  • Guard response.text or "" (evita AttributeError si respuesta está vacía/bloqueada)
  • Guard de claves antes de Diagnosis(**...)/Recommendation(**...) (KeyError)

monitoring/services/ai_providers/static_rules.py:

  • Instanciación de learned-patterns en try/except → un patrón malformado en Valkey no tumba fallback estático
try:
    result = AIInsightResult(
        diagnosis=Diagnosis(**pattern["diagnosis"]),
        recommendation=Recommendation(**pattern["recommendation"]),
    )
except Exception as e:
    logger.warning("Skipping malformed learned pattern: %s", e)
    continue
return result

8. datetime aware

config/settings/base.py (JsonFormatter):

  • datetime.utcnow() → datetime.now(UTC) (deprecado → modern)

terminal/api/auth.py (JWT iat/exp):

  • datetime.utcnow() → datetime.now(UTC)

Por qué importa

  1. ALTA: Evita fuga de identidad de clientes a servicios externos
  2. MEDIA/BAJA: Cierra vectores de timing attacks, reduce SPOF de claves, hardena logs
  3. Primeros pasos: Esto es lo que hay que hacer antes de abrir el producto a clientes reales
  4. Hoy seguro, mañana certificable: Sin riesgo actual (solo uso interno), pero necesario para compliance

Cambios en archivos

ArchivoCambiosLOC
core/workspace_api.py_anonymize_for_ai(), _check_key(), _ai_analyze_perf(), _ai_analyze_finops()+40, redacción
config/settings/base.pyAGENT_JWT_SECRET env var+9
terminal/api/auth.py_agent_jwt_secret(), generate_*_token() → datetime.now(UTC), verify_agent_token()+7, refactor
terminal/consumers.pyImporta _agent_jwt_secret() para verificación WebSocket+2
config/urls.py_InternalToolsAuth.authenticate() → hmac.compare_digest()+3
monitoring/services/ai_providers/google_genai.pyGuards + redacción en logs+8
monitoring/services/ai_providers/claude.pyRedacción en logs+2
monitoring/services/ai_providers/ollama.pyRedacción en logs+2
monitoring/services/ai_providers/router.pySaneo del context+7
monitoring/services/ai_providers/static_rules.pyTry/except en learned-patterns+7
tests/api/test_config_ia.pyNuevos tests (anonymize, check_key, agent_jwt_secret)+92

Total: ~180 LOC, 8 tests.

Tests

tests/api/test_config_ia.py:

  • test_anonymize_for_ai_strips_org_names_and_redacts_queries() — redacción correcta
  • test_anonymize_for_ai_does_not_mutate_original() — no muta response al caller
  • test_anonymize_for_ai_tolerates_missing_keys() — robusto
  • test_check_key_constant_time() — hmac.compare_digest
  • test_check_key_denies_when_unconfigured()
  • test_agent_jwt_secret_fallback() — fallback inerte a SECRET_KEY
  • test_agent_jwt_roundtrip_with_dedicated_key() — token round-trip
  • (Implicit en otros tests de terminal/consumer WebSocket JWT)

Diferidos (T2, T4, carve-out)

Abiertos en Auditoría pero NO cerrados en Etapa 3 (motivo documentado):

  1. /media/ sin auth (config/urls.py) — carve-out: acceso anónimo para render frontend de imágenes. Tocar gate rompería UX completa.

  2. global_exception_handler filtra str(exc) (config/urls.py, ~509 endpoints) — transversal T4 (decisión Edu). Requiere barrido de cómo frontend consume cuerpo de error.

  3. CSRF deshabilitado en API Ninja — auditar que ningún GET mute estado. (Defensa SameSite=Strict ya presente.)

  4. Llamada IA síncrona en request (workspace_api.py, perf-review/finops) — transversal T2 (síncrono→Huey), toca contrato de la API.

  5. Orden de RateLimitMiddleware (capa per-tenant = código muerto) — mover el middleware conocido riesgo de orden, requiere validación.

Relación con Auditoría Suprema

Véase [[entity—core—doc—auditoria-suprema]] (30 hallazgos):

  • Dominio: config/IA (1 ALTA, 8 MEDIA, 21 BAJA)
  • Etapa 3 tanda: ALTA + MEDIA/BAJA de seguridad/robustez
  • Próximas tandas: T2 (IA síncrona→Huey), T4 (global_exception_handler), carve-outs

Véase también

  • [[incident—20260611—fuga-cross-tenant-llm]]
  • [[decision—20260611—agent-jwt-secret-dedicada]]
  • [[entity—core—service—anonymize-for-ai]]
  • [[entity—terminal—function—agent-jwt-secret]]
  • [[concept—security—timing-attack-prevention]]