Servicio de anonimización para LLM (anonymize_for_ai)

Entidadactivecreado Thu Jun 11#security #core #monitoring #ia #multi-tenancy #data-isolation

Descripción

Módulo: core/workspace_api.py
Función: _anonymize_for_ai(data: dict) -> dict
Propósito: Preparar datos agregados de múltiples tenants para análisis de IA sin revelar identidad de clientes.

Qué hace

Anonimiza datos antes de enviar a un LLM externo (Google AI Studio, Claude, etc.). Preserva la distribución (números) que el LLM necesita para análisis, elimina identificadores que revelarían clientes.

Transformaciones

Campo	Antes	Después	Razón
`tenant_usage[].name`	`"ACME Corp"`	`"org-1"`	No revelar identidad de cliente
`tenant_usage[].id`	`1`	`1`	Preservado (referencia opaca)
`tenant_usage[].racks`	`5`	`5`	Preservado (distribución, no es secreto)
`slow_queries[].query`	`SELECT * FROM x WHERE password='hunter2'`	`SELECT * FROM x WHERE password='***'`	Redactar secretos
`db_stats.size_mb`	`100`	`100`	Preservado (no es sensible)

Algoritmo

def _anonymize_for_ai(data):
    """Quita identificadores de tenant y redacta secretos ANTES de enviar a un LLM externo.
    
    El análisis de rendimiento agrega datos de TODAS las organizaciones (nombres de
    cliente en ``tenant_usage``, texto de queries en ``slow_queries``). Mandar eso a
    Google AI Studio sería una fuga cross-tenant de toda la plataforma. El LLM solo
    necesita la *distribución* (conteos por org), no la identidad real: sustituimos el
    nombre por una etiqueta opaca ``org-<id>`` y redactamos posibles secretos del texto
    de las queries. Trabaja sobre una copia (no muta la respuesta devuelta al caller).
    """
    d = copy.deepcopy(data)
    
    # 1. Anonimizar names en tenant_usage
    for row in d.get("tenant_usage", []) or []:
        if isinstance(row, dict) and "name" in row:
            row["name"] = f"org-{row.get('id', '?')}"
    
    # 2. Redactar secretos en slow_queries
    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

Características de robustez:

copy.deepcopy() → no muta el diccionario original
Verificación de tipo: isinstance(row, dict) antes de acceder claves
Manejo de valores faltantes: .get(...) or []] (default a lista vacía)
Operación defensiva: no falla si faltan tenant_usage o slow_queries

Fuentes de datos

Input

tenant_usage — lista de dicts (modelo: Organization, agregado en memoria)
- Campos: id (int), name (str: nombre real, p.ej. “ACME Corp”), racks (int), devices (int)
- Origen: _get_tenant_usage() en core/workspace_api.py, que itera todas las organizaciones
slow_queries — lista de dicts (modelo: SystemLog, filtrado por duración)
- Campos: query (str: SQL crudo), calls (int), avg_duration_ms (float)
- Origen: _get_slow_queries() en core/workspace_api.py, que filtra SystemLog.query_text
db_stats — dict (estadísticas DB agregadas, no sensible)
- Campos: size_mb, indexes_count, table_count
- Origen: Connection pool stats, no necesita anonimización

Output

Diccionario transformado, listo para pasar a AIOperations:

# Llamada (perf-review)
return AIOperations.detect_anomalies(_anonymize_for_ai(data), tone=tone)

# Llamada (finops)
return AIOperations.capacity_recommendation(_anonymize_for_ai(data), tone=tone)

El LLM recibe datos sin identificadores, pero con la misma forma y distribución.

Dependencias

copy (stdlib)
monitoring.services.ai_providers.sanitize.redact_secrets() — función de redacción genérica

from monitoring.services.ai_providers.sanitize import redact_secrets

Invocaciones

_ai_analyze_perf(data, tone="technical") (core/workspace_api.py)
- Llamada: /api/workspace/perf-review?ai=true
- Paso: AIOperations.detect_anomalies(_anonymize_for_ai(data), tone=tone)
_ai_analyze_finops(data, tone="technical") (core/workspace_api.py)
- Llamada: /api/workspace/finops?ai=true
- Paso: AIOperations.capacity_recommendation(_anonymize_for_ai(data), tone=tone)

Ambas envueltas en try/except que logea errores redactados:

except Exception as e:
    logger.warning("workspace perf-review AI analysis failed: %s", redact_secrets(str(e)))
    return "AI analysis temporarily unavailable"

Tests

tests/api/test_config_ia.py:

test_anonymize_for_ai_strips_org_names_and_redacts_queries()
- Verifica: tenant_usage[0].name == "org-1" (redactado)
- Verifica: tenant_usage[0].racks == 5 (preservado)
- Verifica: "hunter2" no está en slow_queries[0].query (secreto redactado)
- Verifica: db_stats.size_mb == 100 (no sensible, preservado)
test_anonymize_for_ai_does_not_mutate_original()
- Input: {"tenant_usage": [{"id": 1, "name": "ACME Corp"}]}
- Llamada a _anonymize_for_ai(data)
- Verifica: El dict original mantiene "ACME Corp" (deep copy, no mutación)
test_anonymize_for_ai_tolerates_missing_keys()
- Entrada incompleta: {"costs": {}} (no tenant_usage/slow_queries)
- Verifica: No crashea, devuelve igual

Seguridad

Threat Model

Atacante: Google, proveedor de LLM, o intermediario (ISP, proxy)
Objetivo: Identificar qué empresas usan CreaRack y sus patrones
Mitigación: Sustituir nombres por etiquetas opacas; si alguien ve el tráfico, ve “org-1, org-2, org-3…” sin saber quiénes son

No es una panacea

El ID numérico (org-<id>) sigue siendo un identificador (aunque opaco)
Si el atacante tiene acceso a ambos lados (logs de Google + BD de CreaRack), puede correlacionar
Defensa complementaria: TLS/cifrado en tránsito, logs con acceso restringido

Alternativas consideradas

Agregar por rango de ID (p.ej. org-group-1): Peor para el LLM (pierde precisión)
Ofuscación del ID (hash, XOR): Más complejo, no aporta más seguridad si el hash es predecible
No enviar a LLM externo (on-premise): Out of scope (decisión de arquitectura, T2)

Véase también

[[incident—20260611—fuga-cross-tenant-llm]]
[[feature—config-ia—auditoria-suprema-etapa-3]]
[[concept—security—data-isolation]]
[[entity—monitoring—service—redact-secrets]]
[[entity—core—endpoint—workspace-perf-review]]