La mayoría de los equipos descubren fugas de PII después de una queja de un cliente o una auditoría de cumplimiento. Para entonces, los datos ya han atravesado tu pipeline de ETL, aterrizado en logs de aplicación y sido indexados por tres herramientas de observabilidad diferentes.

Encontrarlo después del hecho es arqueología. Lo que realmente necesitas es un sistema que rastree PII en el momento en que entra en tu código, propague ese conocimiento a través de cada transformación y lo bloquee de salir por los canales equivocados. Este no es un problema resuelto, pero sí es uno soluble.

Qué significa realmente “rastrear PII”

El rastreo de PII a menudo se confunde con data masking o control de acceso. Esas son preocupaciones relacionadas, pero están aguas abajo del problema real. No puedes redactar, encriptar o restringir el acceso a datos que no puedes localizar.

Rastrear significa saber, en cada punto de tu pipeline, qué campos contienen datos sensibles. No adivinar basándote en nombres de columnas como email o ssn. Saber de verdad, porque los datos llevan una tag que sobrevive transformación, agregación y serialización.

Esto requiere tres cosas: un type system o una capa de metadatos que pueda etiquetar campos como sensibles, un mecanismo de propagación que lleve esas tags mientras los datos se mueven entre funciones y servicios, y enforcement points donde valides que los datos etiquetados solo se escriban en destinos aprobados.

La arquitectura: etiquetar, propagar, enforce

Las implementaciones más limpias que he visto usan un tipo de datos labeled a nivel de lenguaje. En Python, puedes envolver valores en una clase que lleve una tag de sensibilidad.

from dataclasses import dataclass
from enum import Enum, auto
from typing import Generic, TypeVar

T = TypeVar("T")

class Sensitivity(Enum):
    PLAIN = auto()
    PII = auto()
    PCI = auto()

@dataclass(frozen=True)
class Labeled(Generic[T]):
    value: T
    sensitivity: Sensitivity = Sensitivity.PLAIN

    def map(self, fn):
        return Labeled(fn(self.value), self.sensitivity)

Cuando un usuario registra una cuenta, tags las entradas raw en el boundary.

from labeled import Labeled, Sensitivity

def create_user(payload: dict) -> dict:
    email = Labeled(payload["email"], Sensitivity.PII)
    name = Labeled(payload["name"], Sensitivity.PII)
    age = Labeled(int(payload["age"]), Sensitivity.PLAIN)

    user = {
        "id": generate_uuid(),
        "email": email,
        "name": name,
        "age": age,
    }

    return user

La idea clave es que email y name ahora son auto-descriptivos. Cualquier función que reciba un valor Labeled puede inspeccionar su sensibilidad sin parsear el payload o adivinar a partir del nombre de la clave.

Propagando tags a través de transformaciones

Los datos labeled solo son útiles si la tag sobrevive a tu lógica de negocio. Cuando haces map, filter o aggregate, necesitas reglas para cómo se combina la sensibilidad.

El conjunto de reglas más simple es un join semilattice. Si fusionas dos valores, el resultado obtiene la tag más restrictiva.

class Sensitivity(Enum):
    PLAIN = 1
    PII = 2
    PCI = 3

    def join(self, other: "Sensitivity") -> "Sensitivity":
        return self if self.value >= other.value else other

Cuando serializas un registry para un servicio downstream, incluyes la tag en los metadatos. El consumer puede entonces decidir si escribir en una base de datos raw, un store de análitica masked, o un audit log.

def serialize(record: dict) -> dict:
    return {
        key: {
            "value": serialize_value(val.value),
            "sensitivity": val.sensitivity.name.lower(),
        }
        for key, val in record.items()
        if isinstance(val, Labeled)
    }

Esta es la parte que tropeza a la gente. No puedes simplemente etiquetar en la ingesta y esperar lo mejor. Las tags deben ser una preocupación de primera clase en tu serialization layer. Si tu formato interno de RPC descarta los metadatos, tu pipeline se queda ciega en el momento en que los datos cruzan un service boundary.

Enforcement: donde las tags realmente importan

Rastrear sin enforcement es teatro de observabilidad caro. Necesitas choke points donde los datos etiquetados sean inspeccionados antes de ser escritos en almacenamiento, enviados por una red o renderizados en una UI.

Un patrón común es un sink registry que mapea destinos a la sensibilidad máxima permitida.

SINK_POLICIES = {
    "raw_postgres": {Sensitivity.PLAIN, Sensitivity.PII, Sensitivity.PCI},
    "analytics_clickhouse": {Sensitivity.PLAIN},
    "application_logs": {Sensitivity.PLAIN},
    "third_party_api": set(),
}

def write(sink_name: str, record: dict):
    allowed = SINK_POLICIES[sink_name]
    for key, val in record.items():
        if isinstance(val, Labeled) and val.sensitivity not in allowed:
            raise ValueError(
                f"Field {key} with sensitivity {val.sensitivity.name} "
                f"is not allowed in sink {sink_name}"
            )
    # proceed with write

Si tu pipeline de análitica intenta ingerir un registry que contiene Sensitivity.PII en ClickHouse, la escritura falla en runtime con un error claro. Esto no es elegante, pero es explícito. Explícito es mejor que una violación del GDPR.

Específicamente para logging, puedes enforce esto a nivel del logger. El module logging de Python soporta filters que pueden inspeccionar log records antes de que se emitan.

import logging

class PiiFilter(logging.Filter):
    def filter(self, record):
        msg = record.getMessage()
        # In practice, use a more sophisticated check or structured logging
        if any(label in msg for label in ["email", "ssn", "phone"]):
            record.msg = "[REDACTED: PII detected]"
            record.args = ()
        return True

logger = logging.getLogger(__name__)
logger.addFilter(PiiFilter())

Esto es un instrumento tosco. Un enfoque mejor es el structured logging donde tu log payload es el mismo diccionario labeled, y el formatter redacta basándose en la tag de sensibilidad en lugar de regex.

Análisis estático como respaldo

El labeling en runtime es poderoso pero requiere disciplina. Alguien olvidará envolver un campo nuevo, o hará cast de un valor Labeled de vuelta a un primitivo para evitar un type error.

El análisis estático llena el vacío. Herramientas como Semgrep o linters personalizados pueden enforce que los strings raw de los bodies de peticiones HTTP nunca se pasen directamente a loggers o database sinks sin pasar por un constructor de labeling.

Una regla mínima de Semgrep podría verse así:

rules:
  - id: unlabeled-pii-log
    patterns:
      - pattern: logger.info($X)
      - metavariable-pattern:
          metavariable: $X
          pattern-either:
            - pattern: request.json[$FIELD]
            - pattern: request.form[$FIELD]
    message: "Logging raw request data without sensitivity labeling"
    languages: [python]
    severity: ERROR

Esto no detectará cada fuga, pero detectará las obvias, que es de donde vienen la mayoría de las fugas.

Los trade-offs de los que nadie quiere hablar

Este enfoque agrega fricción. Cada transformación de datos ahora involucra un campo de metadatos extra. Los payloads de serialización se hacen más grandes. Los code reviews se convierten en discusiones sobre si user_agent es PII. (Por cierto, usualmente lo es. La UE lo cree.)

El rendimiento es una preocupación real. Si estás procesando millones de events por segundo, asignar un wrapper Labeled para cada campo es una sobrecarga medible. En hot paths, puede que necesites hacer batch de chequeos de tags o mover el enforcement al edge.

También está la carga de mantenimiento. Las tags de sensibilidad son solo tan buenas como los humanos que las mantienen. Cuando cambian las regulaciones de privacidad, o cuando tu negocio se expande a una nueva jurisdicción, necesitas volver a etiquetar datos. Aquí no hay magia. Es trabajo.

Lo que no elegimos: escaneo de regex en reposo

Algunos equipos resuelven esto escaneando data stores después del hecho con regex, buscando patrones que parezcan emails o números de seguridad social. Esto es mejor que nada, pero es fundamentalmente reactivo.

Regex pierde datos ofuscados, identificadores hasheados y PII compuesto. También produce falsos positivos que erosionan la confianza en el sistema. Consideramos este enfoque desde el principio y lo rechazamos porque trata el síntoma (datos en el lugar equivocado) en lugar de la causa (datos saliendo del servicio sin etiquetar).

Empieza con los choke points

No necesitas etiquetar cada campo en cada servicio el primer día. Empieza con los boundaries. Tag los datos cuando entran en tu sistema desde usuarios, APIs de terceros y streams de events. Luego agrega enforcement en tus sinks de mayor riesgo: pipelines de análitica, infraestructura de logging e integraciones externas.

Una vez que esos están en su lugar, expande hacia adentro. El objetivo no es cobertura perfecta. El objetivo es hacer que las fugas de PII sean costosas de crear por accidente, y obvias de detectar en code review.

Si estás construyendo esto en Python, la clase Labeled de arriba es suficiente para empezar. En TypeScript, un branded type o un wrapper similar funciona de la misma manera. En Go, querrás un struct con un campo unexported para prevenir casting accidental.

Las herramientas son simples. La parte difícil es decidir que el PII no rastreado es un bug, y tratarlo como tal.