Módulo 6: Data Privacy & PII Protection

6. Retention Policies y Encryption

Siguiente →

Descripción

Has construido un pipeline que detecta PII (cápsula 03), lo redacta (cápsula 04), y minimiza los datos enviados al LLM (cápsula 05). Pero hay una pregunta que aún no hemos abordado: ¿qué pasa con los datos después del procesamiento?

Los logs de tu sistema AI registran cada request. Los outputs del modelo se almacenan para auditoría. Los documentos de RAG persisten en tu vector store. Los mappings de redacción reversible contienen los datos originales. Las conversaciones de chat se guardan para contexto futuro. Cada uno de estos artefactos es un punto de almacenamiento de datos potencialmente sensibles — y cada uno necesita una política de retención y protección.

En esta cápsula construyes dos componentes: un Retention Scheduler que implementa políticas de eliminación automática, y las prácticas de encryption necesarias para proteger los datos mientras existen. Juntos, estos componentes cierran el ciclo de vida de los datos en tu sistema AI.

¿Por qué retention policies para AI?

En una aplicación web tradicional, los logs se rotan y los datos se archivan. En un sistema AI, la retención tiene consideraciones adicionales:

ai_retention_challenges = {
    "llm_logs": {
        "what": "Prompts completos + responses del LLM",
        "risk": "Contienen el contexto completo incluyendo cualquier PII que no se redactó",
        "retention_consideration": "¿Realmente necesitas guardar el prompt completo?",
        "recommended_retention": "30 días máximo, idealmente con PII redactado",
    },
    "chat_history": {
        "what": "Conversaciones multi-turno de usuarios",
        "risk": "Usuarios revelan información personal durante conversaciones",
        "retention_consideration": "¿Necesitas el historial completo o solo un resumen?",
        "recommended_retention": "90 días, luego resumir y eliminar raw",
    },
    "rag_documents": {
        "what": "Documentos indexados en vector store",
        "risk": "Documentos pueden contener PII de clientes, empleados, o socios",
        "retention_consideration": "¿Los documentos fuente aún son válidos?",
        "recommended_retention": "Según ciclo de vida del documento fuente",
    },
    "redaction_mappings": {
        "what": "Mappings placeholder → valor original de redacción reversible",
        "risk": "Contienen los datos originales sin redactar",
        "retention_consideration": "Solo deben existir durante el procesamiento",
        "recommended_retention": "Eliminación inmediata post-response",
    },
    "embeddings": {
        "what": "Vectores de embeddings de textos procesados",
        "risk": "Los embeddings pueden ser parcialmente reversibles",
        "retention_consideration": "¿Los embeddings corresponden a datos personales?",
        "recommended_retention": "Alineado con la retención del texto fuente",
    },
    "audit_logs": {
        "what": "Registros de detección y redacción de PII",
        "risk": "Pueden revelar qué datos se procesaron (metadata)",
        "retention_consideration": "Necesarios para compliance, pero minimizar metadata",
        "recommended_retention": "1-3 años según regulación aplicable",
    },
}

print("Desafíos de retención en sistemas AI:\n")
for artifact, info in ai_retention_challenges.items():
    print(f"  {artifact.upper()}")
    print(f"    Qué: {info['what']}")
    print(f"    Riesgo: {info['risk']}")
    print(f"    Retención: {info['recommended_retention']}")
    print()

Retention Scheduler

import time
import json
import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from typing import Optional, Callable
from enum import Enum


class RetentionAction(Enum):
    KEEP = "keep"
    ARCHIVE = "archive"
    DELETE = "delete"
    ANONYMIZE = "anonymize"


@dataclass
class RetentionPolicy:
    name: str
    data_type: str
    retention_days: int
    action_on_expiry: RetentionAction
    description: str


@dataclass
class DataRecord:
    record_id: str
    data_type: str
    created_at: datetime
    content: Optional[str] = None
    metadata: dict = field(default_factory=dict)
    
    @property
    def age_days(self) -> float:
        now = datetime.now(timezone.utc)
        return (now - self.created_at).total_seconds() / 86400


@dataclass
class RetentionResult:
    records_checked: int
    records_kept: int
    records_archived: int
    records_deleted: int
    records_anonymized: int
    details: list[dict] = field(default_factory=list)


class RetentionScheduler:
    """Implementa políticas de retención para datos de AI."""

    DEFAULT_POLICIES = [
        RetentionPolicy(
            name="llm_logs",
            data_type="llm_log",
            retention_days=30,
            action_on_expiry=RetentionAction.DELETE,
            description="Logs de prompts y responses del LLM",
        ),
        RetentionPolicy(
            name="chat_history",
            data_type="chat_message",
            retention_days=90,
            action_on_expiry=RetentionAction.ANONYMIZE,
            description="Historial de conversaciones de usuarios",
        ),
        RetentionPolicy(
            name="redaction_mappings",
            data_type="redaction_mapping",
            retention_days=0,
            action_on_expiry=RetentionAction.DELETE,
            description="Mappings de redacción reversible",
        ),
        RetentionPolicy(
            name="audit_logs",
            data_type="audit_log",
            retention_days=365,
            action_on_expiry=RetentionAction.ARCHIVE,
            description="Logs de auditoría de PII",
        ),
        RetentionPolicy(
            name="user_data_cache",
            data_type="user_cache",
            retention_days=7,
            action_on_expiry=RetentionAction.DELETE,
            description="Cache temporal de datos de usuario",
        ),
    ]

    def __init__(
        self,
        policies: Optional[list[RetentionPolicy]] = None,
        on_delete: Optional[Callable] = None,
        on_archive: Optional[Callable] = None,
        on_anonymize: Optional[Callable] = None,
    ):
        self.policies = {
            p.data_type: p for p in (policies or self.DEFAULT_POLICIES)
        }
        self.on_delete = on_delete or self._default_delete
        self.on_archive = on_archive or self._default_archive
        self.on_anonymize = on_anonymize or self._default_anonymize

    def evaluate(self, records: list[DataRecord]) -> RetentionResult:
        """Evalúa qué acción tomar para cada registro."""
        result = RetentionResult(records_checked=len(records), records_kept=0,
                                 records_archived=0, records_deleted=0,
                                 records_anonymized=0)

        for record in records:
            policy = self.policies.get(record.data_type)
            if not policy:
                result.records_kept += 1
                continue

            if record.age_days <= policy.retention_days:
                result.records_kept += 1
                result.details.append({
                    "record_id": record.record_id,
                    "action": "keep",
                    "age_days": round(record.age_days, 1),
                    "expires_in_days": round(
                        policy.retention_days - record.age_days, 1
                    ),
                })
            else:
                action = policy.action_on_expiry
                if action == RetentionAction.DELETE:
                    self.on_delete(record)
                    result.records_deleted += 1
                elif action == RetentionAction.ARCHIVE:
                    self.on_archive(record)
                    result.records_archived += 1
                elif action == RetentionAction.ANONYMIZE:
                    self.on_anonymize(record)
                    result.records_anonymized += 1

                result.details.append({
                    "record_id": record.record_id,
                    "action": action.value,
                    "age_days": round(record.age_days, 1),
                    "policy": policy.name,
                })

        return result

    def _default_delete(self, record: DataRecord):
        record.content = None
        record.metadata["deleted_at"] = datetime.now(timezone.utc).isoformat()

    def _default_archive(self, record: DataRecord):
        record.metadata["archived_at"] = datetime.now(timezone.utc).isoformat()
        record.metadata["archived"] = True

    def _default_anonymize(self, record: DataRecord):
        if record.content:
            record.content = hashlib.sha256(
                record.content.encode()
            ).hexdigest()[:16]
        record.metadata["anonymized_at"] = datetime.now(timezone.utc).isoformat()


# --- Demostración ---

scheduler = RetentionScheduler()

now = datetime.now(timezone.utc)
records = [
    DataRecord("log-1", "llm_log", now - timedelta(days=5), "Recent log"),
    DataRecord("log-2", "llm_log", now - timedelta(days=45), "Old log"),
    DataRecord("chat-1", "chat_message", now - timedelta(days=30), "Hi there"),
    DataRecord("chat-2", "chat_message", now - timedelta(days=120), "Old chat"),
    DataRecord("map-1", "redaction_mapping", now - timedelta(hours=2), "SSN:123"),
    DataRecord("audit-1", "audit_log", now - timedelta(days=200), "PII detected"),
    DataRecord("cache-1", "user_cache", now - timedelta(days=10), "user data"),
]

result = scheduler.evaluate(records)

print(f"Retention Evaluation:")
print(f"  Checked: {result.records_checked}")
print(f"  Kept: {result.records_kept}")
print(f"  Deleted: {result.records_deleted}")
print(f"  Archived: {result.records_archived}")
print(f"  Anonymized: {result.records_anonymized}")
print(f"\n  Details:")
for d in result.details:
    print(f"    {d['record_id']}: {d['action']} (age: {d['age_days']}d)")

# Output esperado:
# Retention Evaluation:
#   Checked: 7
#   Kept: 3
#   Deleted: 3
#   Archived: 0
#   Anonymized: 1

Log sanitization

Los logs son uno de los vectores de exposición más subestimados. Un log que registra el prompt completo contiene todo el PII que el usuario envió.

import re
import json
import logging
from typing import Optional


class SanitizedLogger:
    """Logger que sanitiza PII antes de escribir."""

    PII_PATTERNS = {
        "email": re.compile(
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
        ),
        "phone": re.compile(
            r"\b(?:\+\d{1,3}\s?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b"
        ),
        "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
        "credit_card": re.compile(r"\b(?:\d{4}[-\s]?){3}\d{4}\b"),
        "api_key": re.compile(r"\b(?:sk|pk|api)[_-][A-Za-z0-9]{20,}\b"),
    }

    def __init__(self, logger_name: str = "ai_system"):
        self.logger = logging.getLogger(logger_name)
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s [%(levelname)s] %(message)s"
                )
            )
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)

    def _sanitize(self, message: str) -> str:
        """Reemplaza PII en el mensaje de log."""
        sanitized = message
        for pii_type, pattern in self.PII_PATTERNS.items():
            sanitized = pattern.sub(f"[{pii_type.upper()}_REDACTED]", sanitized)
        return sanitized

    def info(self, message: str, **kwargs):
        self.logger.info(self._sanitize(message), **kwargs)

    def warning(self, message: str, **kwargs):
        self.logger.warning(self._sanitize(message), **kwargs)

    def error(self, message: str, **kwargs):
        self.logger.error(self._sanitize(message), **kwargs)

    def log_request(
        self,
        request_id: str,
        user_input: str,
        llm_output: str,
        pii_detected: int = 0,
    ):
        """Log de request con sanitización automática."""
        entry = {
            "request_id": request_id,
            "input_length": len(user_input),
            "output_length": len(llm_output),
            "pii_detected": pii_detected,
            "input_preview": self._sanitize(user_input[:100]),
        }
        self.logger.info(json.dumps(entry))


# --- Demostración ---

logger = SanitizedLogger("demo")

logger.info("User john@test.com called from 555-123-4567")
logger.info("Processing SSN 123-45-6789 for user")
logger.info("API key sk-abc123def456ghi789jkl012 detected in input")

logger.log_request(
    request_id="req-001",
    user_input="My email is maria@empresa.com and my SSN is 123-45-6789",
    llm_output="Your account status is active.",
    pii_detected=2,
)

# Output esperado:
# 2026-03-13 [INFO] User [EMAIL_REDACTED] called from [PHONE_REDACTED]
# 2026-03-13 [INFO] Processing SSN [SSN_REDACTED] for user
# 2026-03-13 [INFO] API key [API_KEY_REDACTED] detected in input
# 2026-03-13 [INFO] {"request_id": "req-001", "input_length": 55, ...}

Encryption at rest

Los datos almacenados (logs, base de datos, vector store) deben estar cifrados. Python proporciona herramientas para implementar encryption at rest.

import os
import json
import base64
from dataclasses import dataclass
from typing import Optional

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC


class DataEncryptor:
    """Cifra datos sensibles at rest."""

    def __init__(self, encryption_key: Optional[bytes] = None):
        if encryption_key:
            self.fernet = Fernet(encryption_key)
        else:
            self.key = Fernet.generate_key()
            self.fernet = Fernet(self.key)

    @classmethod
    def from_password(cls, password: str, salt: Optional[bytes] = None) -> "DataEncryptor":
        """Crea un encryptor derivando la key de un password."""
        salt = salt or os.urandom(16)
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        instance = cls(encryption_key=key)
        instance._salt = salt
        return instance

    def encrypt(self, data: str) -> str:
        """Cifra un string y retorna el ciphertext en base64."""
        return self.fernet.encrypt(data.encode()).decode()

    def decrypt(self, encrypted_data: str) -> str:
        """Descifra un ciphertext en base64."""
        return self.fernet.decrypt(encrypted_data.encode()).decode()

    def encrypt_dict(self, data: dict) -> str:
        """Cifra un diccionario completo."""
        json_str = json.dumps(data, ensure_ascii=False)
        return self.encrypt(json_str)

    def decrypt_dict(self, encrypted_data: str) -> dict:
        """Descifra un diccionario."""
        json_str = self.decrypt(encrypted_data)
        return json.loads(json_str)

    def encrypt_fields(
        self,
        data: dict,
        fields_to_encrypt: list[str],
    ) -> dict:
        """Cifra solo campos específicos de un dict."""
        result = data.copy()
        for field_name in fields_to_encrypt:
            if field_name in result and isinstance(result[field_name], str):
                result[field_name] = self.encrypt(result[field_name])
                result[f"_{field_name}_encrypted"] = True
        return result

    def decrypt_fields(
        self,
        data: dict,
        fields_to_decrypt: list[str],
    ) -> dict:
        """Descifra campos específicos de un dict."""
        result = data.copy()
        for field_name in fields_to_decrypt:
            if field_name in result and result.get(f"_{field_name}_encrypted"):
                result[field_name] = self.decrypt(result[field_name])
                del result[f"_{field_name}_encrypted"]
        return result


# --- Demostración ---

encryptor = DataEncryptor()

sensitive_data = {
    "user_id": "usr-123",
    "name": "María García",
    "email": "maria@empresa.com",
    "query": "¿Cuál es mi saldo?",
    "account_balance": "$15,234.50",
}

encrypted = encryptor.encrypt_fields(
    sensitive_data,
    fields_to_encrypt=["email", "account_balance"],
)

print("Original:")
for k, v in sensitive_data.items():
    print(f"  {k}: {v}")

print("\nEncrypted fields:")
for k, v in encrypted.items():
    display = v[:40] + "..." if isinstance(v, str) and len(v) > 40 else v
    print(f"  {k}: {display}")

decrypted = encryptor.decrypt_fields(
    encrypted,
    fields_to_decrypt=["email", "account_balance"],
)

print("\nDecrypted:")
for k, v in decrypted.items():
    print(f"  {k}: {v}")

print(f"\nMatch original: {decrypted['email'] == sensitive_data['email']}")

# Output esperado:
# Original:
#   email: maria@empresa.com
#   account_balance: $15,234.50
#
# Encrypted fields:
#   email: gAAAAABh... (encrypted)
#   _email_encrypted: True
#
# Decrypted:
#   email: maria@empresa.com
#   account_balance: $15,234.50
#
# Match original: True

Encryption in transit

Para datos en tránsito entre tu aplicación y APIs externas (OpenAI, vector stores), HTTPS/TLS es obligatorio. Verifica que todas las conexiones usen TLS.

import ssl
from urllib.parse import urlparse


def verify_tls_configuration(urls: list[str]) -> list[dict]:
    """Verifica que las URLs usen TLS."""
    results = []
    for url in urls:
        parsed = urlparse(url)
        result = {
            "url": url,
            "scheme": parsed.scheme,
            "uses_tls": parsed.scheme == "https",
            "hostname": parsed.hostname,
        }
        
        if not result["uses_tls"]:
            result["warning"] = (
                "INSECURE: Connection does not use TLS. "
                "Data in transit is not encrypted."
            )
            result["fix"] = f"Change to https://{parsed.hostname}{parsed.path}"
        else:
            result["status"] = "OK"
        
        results.append(result)
    return results


# --- Demostración ---

api_urls = [
    "https://api.openai.com/v1/chat/completions",
    "https://your-vector-store.com/api/search",
    "http://internal-service.local/process",
    "https://your-database.com:5432/query",
]

tls_results = verify_tls_configuration(api_urls)
for r in tls_results:
    status = "✅" if r["uses_tls"] else "❌"
    print(f"  {status} {r['url'][:50]}...")
    if not r["uses_tls"]:
        print(f"     ⚠ {r['warning']}")
        print(f"     Fix: {r['fix']}")

# Output esperado:
#   ✅ https://api.openai.com/v1/chat/completions...
#   ✅ https://your-vector-store.com/api/search...
#   ❌ http://internal-service.local/process...
#      ⚠ INSECURE: Connection does not use TLS...
#   ✅ https://your-database.com:5432/query...

Key management

Las claves de cifrado necesitan gestión adecuada. No las guardes en el código ni en variables de entorno en producción.

import os
import json
from datetime import datetime, timezone, timedelta


class KeyManager:
    """Gestión básica de claves de cifrado."""

    def __init__(self, keys_directory: str = ".keys"):
        self.keys_dir = keys_directory
        os.makedirs(keys_directory, exist_ok=True)

    def generate_key(self, key_name: str, ttl_days: int = 90) -> dict:
        """Genera una nueva clave con metadata."""
        key = Fernet.generate_key()
        metadata = {
            "key_name": key_name,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "expires_at": (
                datetime.now(timezone.utc) + timedelta(days=ttl_days)
            ).isoformat(),
            "ttl_days": ttl_days,
            "status": "active",
        }
        return {"key": key.decode(), "metadata": metadata}

    def rotate_key(
        self,
        old_key: str,
        new_key_name: str,
    ) -> dict:
        """Genera una nueva clave para rotación."""
        new_key_data = self.generate_key(new_key_name)
        return {
            "old_key_status": "deprecated",
            "new_key": new_key_data,
            "action": "Re-encrypt all data with new key",
        }

    def check_key_expiry(self, key_metadata: dict) -> dict:
        """Verifica si una clave está próxima a expirar."""
        expires = datetime.fromisoformat(key_metadata["expires_at"])
        now = datetime.now(timezone.utc)
        days_remaining = (expires - now).days

        return {
            "key_name": key_metadata["key_name"],
            "days_remaining": days_remaining,
            "expired": days_remaining <= 0,
            "needs_rotation": days_remaining <= 14,
            "status": (
                "EXPIRED" if days_remaining <= 0
                else "ROTATE_SOON" if days_remaining <= 14
                else "OK"
            ),
        }


# --- Demostración ---

km = KeyManager()

key_data = km.generate_key("pii_encryption", ttl_days=90)
print(f"Key generated: {key_data['metadata']['key_name']}")
print(f"  Expires: {key_data['metadata']['expires_at'][:10]}")

expiry = km.check_key_expiry(key_data['metadata'])
print(f"  Status: {expiry['status']}")
print(f"  Days remaining: {expiry['days_remaining']}")

Secure deletion

Eliminar datos no es simplemente del o os.remove(). En producción, necesitas asegurar que los datos no sean recuperables.

import os
import secrets
from typing import Optional


class SecureDeleter:
    """Eliminación segura de datos."""

    @staticmethod
    def secure_delete_string(sensitive_string: str) -> str:
        """
        Sobreescribe un string en memoria.
        Nota: Python no garantiza inmutabilidad de memoria,
        pero esta es la mejor práctica disponible.
        """
        length = len(sensitive_string)
        return secrets.token_hex(length // 2)

    @staticmethod
    def secure_delete_file(filepath: str, passes: int = 3):
        """
        Sobreescribe un archivo con datos aleatorios antes de eliminarlo.
        """
        if not os.path.exists(filepath):
            return {"status": "not_found", "path": filepath}

        file_size = os.path.getsize(filepath)

        for pass_num in range(passes):
            with open(filepath, "wb") as f:
                f.write(os.urandom(file_size))
                f.flush()
                os.fsync(f.fileno())

        os.remove(filepath)
        return {
            "status": "deleted",
            "path": filepath,
            "passes": passes,
            "size_overwritten": file_size,
        }

    @staticmethod
    def secure_delete_dict(data: dict) -> dict:
        """Sobreescribe valores de un diccionario."""
        for key in list(data.keys()):
            if isinstance(data[key], str):
                data[key] = secrets.token_hex(8)
            elif isinstance(data[key], (int, float)):
                data[key] = 0
            elif isinstance(data[key], dict):
                SecureDeleter.secure_delete_dict(data[key])
            elif isinstance(data[key], list):
                data[key] = []
        return data


# --- Demostración ---

deleter = SecureDeleter()

sensitive = {
    "ssn": "123-45-6789",
    "credit_card": "4111-1111-1111-1111",
    "email": "maria@empresa.com",
}

print("Before deletion:")
for k, v in sensitive.items():
    print(f"  {k}: {v}")

deleter.secure_delete_dict(sensitive)

print("\nAfter secure deletion:")
for k, v in sensitive.items():
    print(f"  {k}: {v}")

# Output esperado:
# Before deletion:
#   ssn: 123-45-6789
#   credit_card: 4111-1111-1111-1111
#   email: maria@empresa.com
#
# After secure deletion:
#   ssn: a1b2c3d4e5f6g7h8
#   credit_card: 9i0j1k2l3m4n5o6p
#   email: q7r8s9t0u1v2w3x4

Conexión con el proyecto

El Retention Scheduler y las prácticas de encryption se integran en el PII Protection Layer:

PII Protection Layer
├── PIIScanner (Cápsula 03)
├── PreLLMRedactor (Cápsula 04)
├── PostLLMRedactor (Cápsula 04)
├── DataMinimizer (Cápsula 05)
├── RetentionScheduler (ESTA CÁPSULA)  ← Retención
├── DataEncryptor (ESTA CÁPSULA)       ← Encryption
├── SanitizedLogger (ESTA CÁPSULA)     ← Log sanitization
└── Audit Logger

Troubleshooting

Problema 1: "La retention policy elimina datos que aún necesito para debugging"

Solución: Crea una política separada para datos de debugging con retención más larga (90 días), pero asegúrate de que los datos de debugging estén sanitizados (sin PII real). Usa logs con PII redactado para debugging.

Problema 2: "La encryption agrega latencia significativa"

Solución: Cifra solo los campos sensibles, no todo el registro. Usa encrypt_fields() en lugar de encrypt_dict(). La cifrado de un campo de texto típico toma <1ms con Fernet.

Problema 3: "Perdí la clave de cifrado y no puedo descifrar los datos"

Solución: Implementa key backup y escrow desde el inicio. Usa un servicio de key management (AWS KMS, GCP KMS, HashiCorp Vault del Módulo 5) en lugar de gestionar claves manualmente.

Problema 4: "No sé cuánto tiempo retener los datos según mi jurisdicción"

Solución: Como regla general: datos de transacciones 7 años (requisitos fiscales), logs de auditoría 1-3 años, datos de usuarios hasta que soliciten eliminación o terminen la relación. Consulta con tu equipo legal para requisitos específicos.

Ejercicios

Ejercicio 1: Retention scheduler con notificaciones

Extiende el RetentionScheduler para que envíe notificaciones cuando los datos estén próximos a expirar.

Ver solución 
class NotifyingRetentionScheduler(RetentionScheduler):
    def __init__(self, warning_days: int = 7, **kwargs):
        super().__init__(**kwargs)
        self.warning_days = warning_days
        self.notifications: list[dict] = []

    def evaluate_with_warnings(
        self, records: list[DataRecord],
    ) -> RetentionResult:
        result = self.evaluate(records)
        
        for record in records:
            policy = self.policies.get(record.data_type)
            if not policy:
                continue
            days_remaining = policy.retention_days - record.age_days
            if 0 < days_remaining <= self.warning_days:
                self.notifications.append({
                    "record_id": record.record_id,
                    "data_type": record.data_type,
                    "expires_in_days": round(days_remaining, 1),
                    "action_on_expiry": policy.action_on_expiry.value,
                })
        
        return result


scheduler = NotifyingRetentionScheduler(warning_days=7)
now = datetime.now(timezone.utc)
records = [
    DataRecord("log-1", "llm_log", now - timedelta(days=25)),
    DataRecord("log-2", "llm_log", now - timedelta(days=45)),
]
scheduler.evaluate_with_warnings(records)
print(f"Notifications: {scheduler.notifications}")

Ejercicio 2: Encrypted audit log

Crea un audit log que cifre las entradas antes de escribirlas.

Ver solución 
class EncryptedAuditLog:
    def __init__(self):
        self.encryptor = DataEncryptor()
        self.entries: list[str] = []

    def log(self, entry: dict):
        encrypted = self.encryptor.encrypt_dict(entry)
        self.entries.append(encrypted)

    def read(self, index: int) -> dict:
        return self.encryptor.decrypt_dict(self.entries[index])

    def read_all(self) -> list[dict]:
        return [self.encryptor.decrypt_dict(e) for e in self.entries]


audit = EncryptedAuditLog()
audit.log({"event": "pii_detected", "type": "email", "request_id": "req-1"})
audit.log({"event": "pii_redacted", "type": "ssn", "request_id": "req-2"})

print(f"Encrypted entries: {len(audit.entries)}")
print(f"Raw (encrypted): {audit.entries[0][:40]}...")
print(f"Decrypted: {audit.read(0)}")

Ejercicio 3: Log sanitizer con patrones custom

Extiende el SanitizedLogger para aceptar patrones de PII custom.

Ver solución 
class CustomSanitizedLogger(SanitizedLogger):
    def __init__(self, custom_patterns: Optional[dict] = None, **kwargs):
        super().__init__(**kwargs)
        if custom_patterns:
            for name, pattern_str in custom_patterns.items():
                self.PII_PATTERNS[name] = re.compile(pattern_str)

    def add_pattern(self, name: str, pattern: str):
        self.PII_PATTERNS[name] = re.compile(pattern)


logger = CustomSanitizedLogger(
    custom_patterns={
        "employee_id": r"\bEMP-\d{6}\b",
        "order_id": r"\bORD-\d{5,}\b",
    }
)

logger.info("Employee EMP-123456 processed order ORD-98765")
# Output: Employee [EMPLOYEE_ID_REDACTED] processed order [ORDER_ID_REDACTED]

Ejercicio 4: Data lifecycle tracker

Crea un tracker que registre el ciclo de vida completo de un dato.

Ver solución 
@dataclass
class DataLifecycleEvent:
    timestamp: str
    event: str
    details: str

class DataLifecycleTracker:
    def __init__(self):
        self.records: dict[str, list[DataLifecycleEvent]] = {}

    def track(self, record_id: str, event: str, details: str = ""):
        if record_id not in self.records:
            self.records[record_id] = []
        self.records[record_id].append(DataLifecycleEvent(
            timestamp=datetime.now(timezone.utc).isoformat(),
            event=event,
            details=details,
        ))

    def get_lifecycle(self, record_id: str) -> list[dict]:
        events = self.records.get(record_id, [])
        return [{"event": e.event, "details": e.details, "at": e.timestamp} for e in events]


tracker = DataLifecycleTracker()
tracker.track("data-001", "created", "User submitted form")
tracker.track("data-001", "pii_detected", "Email and phone found")
tracker.track("data-001", "redacted", "PII redacted pre-LLM")
tracker.track("data-001", "processed", "LLM generated response")
tracker.track("data-001", "scheduled_deletion", "Retention: 30 days")

for event in tracker.get_lifecycle("data-001"):
    print(f"  {event['event']}: {event['details']}")

Resumen

  • 🔑 Retention policies definen cuánto tiempo se almacenan datos de AI — los logs de LLM, chat histories, y redaction mappings necesitan políticas diferentes
  • 🔑 El RetentionScheduler evalúa registros contra políticas y ejecuta acciones automáticas: DELETE, ARCHIVE, o ANONYMIZE cuando expiran
  • 🔑 La sanitización de logs es crítica — un logger que registra prompts completos expone todo el PII del usuario. Usa SanitizedLogger con regex para redactar antes de escribir
  • 🔑 Encryption at rest con Fernet (symmetric) protege datos almacenados — cifra campos sensibles individualmente para minimizar latencia
  • 🔑 Encryption in transit requiere HTTPS/TLS para todas las conexiones a APIs externas — verifica que no haya endpoints HTTP sin cifrar
  • 🔑 Key management necesita rotación periódica (90 días), backup seguro, y idealmente un servicio de KMS (cubierto en Módulo 5)
  • 🔑 La eliminación segura requiere sobreescribir datos antes de eliminar — del y os.remove() no garantizan que los datos sean irrecuperables
  • 🔑 Los redaction mappings de redacción reversible son los datos más sensibles del pipeline — deben eliminarse inmediatamente después del procesamiento

Recursos adicionales

  1. GDPR Art. 5(1)(e) — Storage Limitation — Principio de limitación de almacenamiento en GDPR
  2. Python cryptography Library — Librería de cifrado usada en esta cápsula (Fernet)
  3. NIST Guidelines for Media Sanitization (SP 800-88) — Guía del NIST para eliminación segura de datos
  4. OWASP Logging Cheat Sheet — Mejores prácticas de logging seguro
  5. AWS KMS Documentation — Servicio de key management de AWS
  6. HashiCorp Vault — Encryption as a Service — Transit secrets engine de Vault para encryption
  7. PCI DSS — Key Management — Requisitos de gestión de claves para datos de tarjetas de crédito

Creado: Marzo 2026 Versión: 1.0

← Anterior5. Data MinimizationSiguiente →7. Compliance Basics: GDPR, CCPA y AI