Módulo 6: Data Privacy & PII Protection

8. Proyecto: PII Protection Layer

Descripción del proyecto

Este proyecto cierra el Módulo 6 con la implementación de un PII Protection Layer completo — una capa de protección de datos personales que se integra con tu Sanitization Pipeline del Módulo 4. No es un ejercicio parcial — es el artefacto production-ready que demuestra que tu sistema AI no solo resiste ataques (M3), mantiene la higiene de datos (M4), y asegura credenciales (M5), sino que protege la información personal de tus usuarios.

En las cápsulas anteriores construiste las piezas individuales: PIIScanner (03), PreLLMRedactor y PostLLMRedactor (04), DataMinimizer (05), RetentionScheduler y DataEncryptor (06), y ConsentManager (07). Ahora todo se consolida en un solo sistema que:

Escanea PII en inputs y outputs con Presidio + reconocedores custom
Redacta PII antes de enviar al LLM (pre-LLM) y después de recibir la respuesta (post-LLM)
Minimiza datos enviados al LLM según políticas por endpoint
Programa retención con eliminación automática de datos expirados
Registra auditoría de cada detección, redacción y acción de retención

Al terminar vas a tener un directorio con código Python ejecutable y un servidor FastAPI que demuestra el PII Protection Layer en acción. Este es el sexto artefacto de la guía y se integra con el Sanitization Pipeline (M4) y el Injection Defense Pipeline (M3) en el Módulo 8 (Secured AI System).

Objetivo del proyecto

Construir un middleware FastAPI de protección de PII que procese cada request a través de 5 capas (PII scanner → pre-LLM redactor → data minimizer → post-LLM redactor → audit logger), con configuración centralizada, retención automática, y métricas de performance — todo en un paquete reutilizable.

Especificaciones técnicas

Stack

Python >= 3.10
pydantic >= 2.0
fastapi >= 0.100
uvicorn >= 0.20
presidio-analyzer >= 2.2
presidio-anonymizer >= 2.2
spacy >= 3.5
cryptography >= 41.0

Estructura del entregable

pii-protection-layer/
├── pii_layer/
│   ├── __init__.py
│   ├── scanner.py            # PIIScanner
│   ├── redactor.py           # PreLLMRedactor + PostLLMRedactor
│   ├── minimizer.py          # DataMinimizer
│   ├── retention.py          # RetentionScheduler
│   ├── audit.py              # PIIAuditLogger
│   ├── pipeline.py           # PIIProtectionPipeline (integración)
│   └── config.py             # Configuración centralizada
├── app.py                    # FastAPI application
├── tests/
│   ├── test_scanner.py
│   ├── test_redactor.py
│   ├── test_minimizer.py
│   ├── test_retention.py
│   └── test_pipeline.py
├── requirements.txt
└── README.md

Funcionalidades obligatorias

1. PIIScanner

✅ Presidio AnalyzerEngine con modelo spaCy
✅ Reconocedores custom (mínimo 2: EMPLOYEE_ID y un tipo de tu dominio)
✅ Configurable: score_threshold, entities to scan, language
✅ Resultado con entities, severities, scan_time_ms

2. PreLLMRedactor

✅ Redacción con Presidio Anonymizer
✅ Operators configurables por tipo de entidad
✅ Modo reversible con mappings
✅ Resultado con redacted_text, redaction_count, mappings

3. PostLLMRedactor

✅ Scan del output del LLM para PII
✅ Block automático si se detecta PII critical (SSN, credit card)
✅ Redacción de PII no-critical en output
✅ Resultado con action (pass/redacted/blocked)

4. DataMinimizer

✅ Clasificación de campos (required/useful/unnecessary/forbidden)
✅ Políticas por endpoint/tarea
✅ Generalización de datos sensibles
✅ Resultado con minimized_data, reduction_percent

5. RetentionScheduler

✅ Políticas por tipo de dato (llm_log, chat, redaction_mapping)
✅ Evaluación de registros contra políticas
✅ Acciones: delete, archive, anonymize
✅ Resultado con records_kept/deleted/archived/anonymized

6. PIIProtectionPipeline

✅ Integración de los 5 componentes en flujo secuencial
✅ Audit logging de cada etapa
✅ Error handling por etapa
✅ Métricas de timing por componente

Código de implementación

Paso 1: Setup del proyecto

mkdir pii-protection-layer && cd pii-protection-layer
mkdir pii_layer tests

Crea requirements.txt:

pydantic>=2.0
fastapi>=0.100
uvicorn>=0.20
presidio-analyzer>=2.2
presidio-anonymizer>=2.2
spacy>=3.5
cryptography>=41.0
pytest>=7.0
pytest-asyncio>=0.21

pip install -r requirements.txt
python -m spacy download en_core_web_lg

Paso 2: Configuración centralizada

Crea pii_layer/config.py:

from pydantic import BaseModel, Field
from typing import Optional


class ScannerConfig(BaseModel):
    score_threshold: float = 0.5
    languages: list[str] = Field(default_factory=lambda: ["en"])
    entities: Optional[list[str]] = None
    custom_recognizers_enabled: bool = True


class RedactorConfig(BaseModel):
    strategy: str = "mask"
    reversible: bool = False
    block_on_critical: bool = True
    critical_entities: list[str] = Field(
        default_factory=lambda: ["US_SSN", "CREDIT_CARD", "IBAN_CODE"]
    )
    fallback_message: str = (
        "Lo siento, no puedo proporcionar esa información "
        "por razones de privacidad."
    )


class MinimizerConfig(BaseModel):
    forbidden_fields: list[str] = Field(
        default_factory=lambda: [
            "ssn", "credit_card", "password", "api_key",
            "social_security", "medical_record",
        ]
    )
    sensitive_fields: list[str] = Field(
        default_factory=lambda: [
            "email", "phone", "address", "date_of_birth",
            "full_name", "last_name",
        ]
    )
    include_sensitive: bool = False


class RetentionConfig(BaseModel):
    llm_log_days: int = 30
    chat_history_days: int = 90
    redaction_mapping_days: int = 0
    audit_log_days: int = 365
    user_cache_days: int = 7


class PIILayerConfig(BaseModel):
    scanner: ScannerConfig = Field(default_factory=ScannerConfig)
    redactor: RedactorConfig = Field(default_factory=RedactorConfig)
    minimizer: MinimizerConfig = Field(default_factory=MinimizerConfig)
    retention: RetentionConfig = Field(default_factory=RetentionConfig)
    enable_audit_log: bool = True
    enable_pre_llm_redaction: bool = True
    enable_post_llm_redaction: bool = True
    enable_minimization: bool = True

Paso 3: PIIScanner

Crea pii_layer/scanner.py:

import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

from presidio_analyzer import (
    AnalyzerEngine,
    PatternRecognizer,
    Pattern,
    RecognizerRegistry,
)

from .config import ScannerConfig


class PIISeverity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


ENTITY_SEVERITY = {
    "PERSON": PIISeverity.MEDIUM,
    "EMAIL_ADDRESS": PIISeverity.MEDIUM,
    "PHONE_NUMBER": PIISeverity.MEDIUM,
    "US_SSN": PIISeverity.CRITICAL,
    "CREDIT_CARD": PIISeverity.CRITICAL,
    "IBAN_CODE": PIISeverity.HIGH,
    "IP_ADDRESS": PIISeverity.LOW,
    "LOCATION": PIISeverity.LOW,
    "DATE_TIME": PIISeverity.LOW,
    "NRP": PIISeverity.MEDIUM,
    "URL": PIISeverity.LOW,
    "EMPLOYEE_ID": PIISeverity.MEDIUM,
    "SUPPORT_TICKET": PIISeverity.LOW,
}

SEVERITY_ORDER = {
    PIISeverity.LOW: 0, PIISeverity.MEDIUM: 1,
    PIISeverity.HIGH: 2, PIISeverity.CRITICAL: 3,
}


@dataclass
class PIIEntity:
    entity_type: str
    text: str
    start: int
    end: int
    score: float
    severity: PIISeverity


@dataclass
class ScanResult:
    text: str
    entities: list[PIIEntity] = field(default_factory=list)
    entity_count: int = 0
    max_severity: PIISeverity = PIISeverity.LOW
    scan_time_ms: float = 0.0

    @property
    def has_pii(self) -> bool:
        return self.entity_count > 0

    @property
    def has_critical(self) -> bool:
        return self.max_severity == PIISeverity.CRITICAL


def _create_custom_recognizers() -> list[PatternRecognizer]:
    return [
        PatternRecognizer(
            supported_entity="EMPLOYEE_ID",
            name="Employee ID",
            patterns=[Pattern("emp_id", r"\bEMP-\d{6}\b", 0.95)],
        ),
        PatternRecognizer(
            supported_entity="SUPPORT_TICKET",
            name="Support Ticket",
            patterns=[Pattern("ticket", r"\bTKT-\d{8}\b", 0.90)],
        ),
    ]


class PIIScanner:
    def __init__(self, config: Optional[ScannerConfig] = None):
        self.config = config or ScannerConfig()

        registry = RecognizerRegistry()
        registry.load_predefined_recognizers()

        if self.config.custom_recognizers_enabled:
            for rec in _create_custom_recognizers():
                registry.add_recognizer(rec)

        self.analyzer = AnalyzerEngine(registry=registry)

    def scan(
        self,
        text: str,
        language: Optional[str] = None,
        entities: Optional[list[str]] = None,
    ) -> ScanResult:
        start = time.perf_counter()

        lang = language or self.config.languages[0]
        ents = entities or self.config.entities

        results = self.analyzer.analyze(
            text=text,
            language=lang,
            entities=ents,
            score_threshold=self.config.score_threshold,
        )

        pii_entities = []
        max_sev = PIISeverity.LOW

        for r in results:
            sev = ENTITY_SEVERITY.get(r.entity_type, PIISeverity.MEDIUM)
            pii_entities.append(PIIEntity(
                entity_type=r.entity_type,
                text=text[r.start:r.end],
                start=r.start, end=r.end,
                score=r.score, severity=sev,
            ))
            if SEVERITY_ORDER[sev] > SEVERITY_ORDER[max_sev]:
                max_sev = sev

        elapsed = (time.perf_counter() - start) * 1000

        return ScanResult(
            text=text,
            entities=sorted(pii_entities, key=lambda e: e.start),
            entity_count=len(pii_entities),
            max_severity=max_sev,
            scan_time_ms=round(elapsed, 2),
        )

Paso 4: Redactors

Crea pii_layer/redactor.py:

from dataclasses import dataclass, field
from typing import Optional

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

from .config import RedactorConfig


@dataclass
class RedactionMapping:
    placeholder: str
    original: str
    entity_type: str


@dataclass
class PreLLMResult:
    original_text: str
    redacted_text: str
    redaction_count: int
    mappings: list[RedactionMapping] = field(default_factory=list)
    reversible: bool = False


@dataclass
class PostLLMResult:
    original_output: str
    redacted_output: str
    pii_found: int
    entities_redacted: list[dict] = field(default_factory=list)
    action: str = "pass"


class PreLLMRedactor:
    def __init__(self, config: Optional[RedactorConfig] = None):
        self.config = config or RedactorConfig()
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, text: str, language: str = "en") -> PreLLMResult:
        results = self.analyzer.analyze(
            text=text, language=language,
            score_threshold=0.5,
        )

        if not results:
            return PreLLMResult(
                original_text=text, redacted_text=text, redaction_count=0,
            )

        mappings = []

        if self.config.reversible:
            counter = {}
            redacted = text
            for r in sorted(results, key=lambda x: x.end, reverse=True):
                entity_text = text[r.start:r.end]
                count = counter.get(r.entity_type, 0) + 1
                counter[r.entity_type] = count
                placeholder = f"<{r.entity_type}_{count}>"
                mappings.append(RedactionMapping(
                    placeholder=placeholder,
                    original=entity_text,
                    entity_type=r.entity_type,
                ))
                redacted = redacted[:r.start] + placeholder + redacted[r.end:]

            return PreLLMResult(
                original_text=text, redacted_text=redacted,
                redaction_count=len(results),
                mappings=mappings, reversible=True,
            )

        anonymized = self.anonymizer.anonymize(
            text=text, analyzer_results=results,
        )

        return PreLLMResult(
            original_text=text,
            redacted_text=anonymized.text,
            redaction_count=len(results),
        )


class PostLLMRedactor:
    def __init__(self, config: Optional[RedactorConfig] = None):
        self.config = config or RedactorConfig()
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def redact(self, llm_output: str, language: str = "en") -> PostLLMResult:
        results = self.analyzer.analyze(
            text=llm_output, language=language,
            score_threshold=0.5,
        )

        if not results:
            return PostLLMResult(
                original_output=llm_output,
                redacted_output=llm_output,
                pii_found=0, action="pass",
            )

        has_critical = any(
            r.entity_type in self.config.critical_entities
            for r in results
        )

        entities_info = [
            {
                "type": r.entity_type,
                "text": llm_output[r.start:r.end],
                "score": r.score,
            }
            for r in results
        ]

        if has_critical and self.config.block_on_critical:
            return PostLLMResult(
                original_output=llm_output,
                redacted_output=self.config.fallback_message,
                pii_found=len(results),
                entities_redacted=entities_info,
                action="blocked",
            )

        anonymized = self.anonymizer.anonymize(
            text=llm_output, analyzer_results=results,
        )

        return PostLLMResult(
            original_output=llm_output,
            redacted_output=anonymized.text,
            pii_found=len(results),
            entities_redacted=entities_info,
            action="redacted",
        )

Paso 5: DataMinimizer

Crea pii_layer/minimizer.py:

from dataclasses import dataclass, field
from typing import Optional, Any
from enum import Enum
import re

from .config import MinimizerConfig


class MinAction(Enum):
    INCLUDED = "included"
    EXCLUDED = "excluded"
    BLOCKED = "blocked"
    GENERALIZED = "generalized"


@dataclass
class MinResult:
    original_field_count: int
    minimized_data: dict
    excluded_count: int = 0
    blocked_count: int = 0
    data_reduction_percent: float = 0.0


class DataMinimizer:
    def __init__(self, config: Optional[MinimizerConfig] = None):
        self.config = config or MinimizerConfig()

    def minimize(
        self,
        data: dict,
        allowed_fields: Optional[list[str]] = None,
    ) -> MinResult:
        minimized = {}
        excluded = 0
        blocked = 0

        original_size = sum(len(str(v)) for v in data.values() if v)

        for key, value in data.items():
            norm_key = key.lower().replace("-", "_").replace(" ", "_")

            if norm_key in self.config.forbidden_fields:
                blocked += 1
                continue

            if allowed_fields and key not in allowed_fields:
                excluded += 1
                continue

            if norm_key in self.config.sensitive_fields:
                if not self.config.include_sensitive:
                    excluded += 1
                    continue
                minimized[key] = self._generalize(key, value)
                continue

            minimized[key] = value

        min_size = sum(len(str(v)) for v in minimized.values() if v)
        reduction = (1 - min_size / original_size) * 100 if original_size > 0 else 0

        return MinResult(
            original_field_count=len(data),
            minimized_data=minimized,
            excluded_count=excluded,
            blocked_count=blocked,
            data_reduction_percent=round(reduction, 1),
        )

    def _generalize(self, key: str, value: Any) -> Any:
        if value is None:
            return None
        str_val = str(value)
        norm = key.lower()

        if norm == "email" and "@" in str_val:
            return f"***@{str_val.split('@')[1]}"
        if norm in ("phone", "telephone") and len(str_val) >= 4:
            return "***" + str_val[-4:]
        if norm in ("full_name",):
            parts = str_val.split()
            return parts[0] if parts else "[Name]"
        if norm in ("date_of_birth", "dob"):
            match = re.search(r"(19|20)\d{2}", str_val)
            if match:
                return f"{(int(match.group()) // 10) * 10}s"

        return value

Paso 6: RetentionScheduler

Crea pii_layer/retention.py:

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

from .config import RetentionConfig


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


@dataclass
class RetentionPolicy:
    data_type: str
    retention_days: int
    action: RetAction


@dataclass
class DataRecord:
    record_id: str
    data_type: str
    created_at: datetime
    content: Optional[str] = None

    @property
    def age_days(self) -> float:
        return (datetime.now(timezone.utc) - self.created_at).total_seconds() / 86400


@dataclass
class RetentionResult:
    records_checked: int
    kept: int = 0
    deleted: int = 0
    archived: int = 0
    anonymized: int = 0
    details: list[dict] = field(default_factory=list)


class RetentionScheduler:
    def __init__(self, config: Optional[RetentionConfig] = None):
        cfg = config or RetentionConfig()
        self.policies = {
            "llm_log": RetentionPolicy("llm_log", cfg.llm_log_days, RetAction.DELETE),
            "chat_message": RetentionPolicy("chat_message", cfg.chat_history_days, RetAction.ANONYMIZE),
            "redaction_mapping": RetentionPolicy("redaction_mapping", cfg.redaction_mapping_days, RetAction.DELETE),
            "audit_log": RetentionPolicy("audit_log", cfg.audit_log_days, RetAction.ARCHIVE),
            "user_cache": RetentionPolicy("user_cache", cfg.user_cache_days, RetAction.DELETE),
        }

    def evaluate(self, records: list[DataRecord]) -> RetentionResult:
        result = RetentionResult(records_checked=len(records))

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

            if record.age_days <= policy.retention_days:
                result.kept += 1
                result.details.append({
                    "record_id": record.record_id,
                    "action": "keep",
                    "age_days": round(record.age_days, 1),
                })
            else:
                if policy.action == RetAction.DELETE:
                    record.content = None
                    result.deleted += 1
                elif policy.action == RetAction.ARCHIVE:
                    result.archived += 1
                elif policy.action == RetAction.ANONYMIZE:
                    if record.content:
                        import hashlib
                        record.content = hashlib.sha256(
                            record.content.encode()
                        ).hexdigest()[:16]
                    result.anonymized += 1

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

        return result

Paso 7: PIIAuditLogger

Crea pii_layer/audit.py:

import json
import logging
import re
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class PIIAuditEntry:
    timestamp: str
    request_id: str
    stage: str
    action: str
    pii_count: int = 0
    details: dict = field(default_factory=dict)
    timing_ms: float = 0.0


PII_LOG_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"),
}


class PIIAuditLogger:
    def __init__(self, logger_name: str = "pii_audit"):
        self.logger = logging.getLogger(logger_name)
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter(
                "%(asctime)s [PII_AUDIT] %(message)s"
            ))
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)
        self.entries: list[PIIAuditEntry] = []

    def _sanitize_log(self, message: str) -> str:
        sanitized = message
        for pii_type, pattern in PII_LOG_PATTERNS.items():
            sanitized = pattern.sub(f"[{pii_type.upper()}_REDACTED]", sanitized)
        return sanitized

    def log(
        self,
        request_id: str,
        stage: str,
        action: str,
        pii_count: int = 0,
        details: Optional[dict] = None,
        timing_ms: float = 0.0,
    ):
        entry = PIIAuditEntry(
            timestamp=datetime.now(timezone.utc).isoformat(),
            request_id=request_id,
            stage=stage,
            action=action,
            pii_count=pii_count,
            details=details or {},
            timing_ms=round(timing_ms, 2),
        )
        self.entries.append(entry)

        log_data = {
            "request_id": request_id,
            "stage": stage,
            "action": action,
            "pii_count": pii_count,
            "timing_ms": round(timing_ms, 2),
        }
        self.logger.info(self._sanitize_log(json.dumps(log_data)))

    def get_stats(self) -> dict:
        if not self.entries:
            return {"total_entries": 0}

        pii_detected = sum(e.pii_count for e in self.entries if e.stage == "scan")
        blocked = sum(1 for e in self.entries if e.action == "blocked")

        return {
            "total_entries": len(self.entries),
            "total_pii_detected": pii_detected,
            "total_blocked": blocked,
            "avg_timing_ms": round(
                sum(e.timing_ms for e in self.entries) / len(self.entries), 2
            ),
        }

Paso 8: Pipeline integrado

Crea pii_layer/pipeline.py:

import time
import uuid
from dataclasses import dataclass, field
from typing import Optional, Callable

from .scanner import PIIScanner
from .redactor import PreLLMRedactor, PostLLMRedactor
from .minimizer import DataMinimizer
from .audit import PIIAuditLogger
from .config import PIILayerConfig


@dataclass
class PIIContext:
    request_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    user_id: str = "anonymous"
    endpoint: str = ""
    stages_passed: list[str] = field(default_factory=list)
    stages_failed: list[str] = field(default_factory=list)
    total_pii_detected: int = 0
    total_pii_redacted: int = 0
    timings: dict[str, float] = field(default_factory=dict)
    total_time_ms: float = 0.0


class PIIProtectionPipeline:
    def __init__(
        self,
        config: Optional[PIILayerConfig] = None,
        llm_caller: Optional[Callable] = None,
    ):
        self.config = config or PIILayerConfig()
        self.scanner = PIIScanner(self.config.scanner)
        self.pre_redactor = PreLLMRedactor(self.config.redactor)
        self.post_redactor = PostLLMRedactor(self.config.redactor)
        self.minimizer = DataMinimizer(self.config.minimizer)
        self.audit = PIIAuditLogger()
        self.llm_caller = llm_caller

    async def process(
        self,
        user_input: str,
        user_data: Optional[dict] = None,
        context: Optional[PIIContext] = None,
    ) -> dict:
        ctx = context or PIIContext()
        start = time.perf_counter()

        # Stage 1: Scan input for PII
        t = time.perf_counter()
        scan_result = self.scanner.scan(user_input)
        ctx.timings["scan"] = (time.perf_counter() - t) * 1000
        ctx.total_pii_detected = scan_result.entity_count
        ctx.stages_passed.append("scan")

        self.audit.log(
            ctx.request_id, "scan", "completed",
            pii_count=scan_result.entity_count,
            details={"max_severity": scan_result.max_severity.value},
            timing_ms=ctx.timings["scan"],
        )

        # Stage 2: Pre-LLM Redaction
        processed_input = user_input
        if self.config.enable_pre_llm_redaction and scan_result.has_pii:
            t = time.perf_counter()
            redact_result = self.pre_redactor.redact(user_input)
            ctx.timings["pre_redact"] = (time.perf_counter() - t) * 1000
            processed_input = redact_result.redacted_text
            ctx.total_pii_redacted += redact_result.redaction_count
            ctx.stages_passed.append("pre_redact")

            self.audit.log(
                ctx.request_id, "pre_redact", "redacted",
                pii_count=redact_result.redaction_count,
                timing_ms=ctx.timings["pre_redact"],
            )

        # Stage 3: Data Minimization
        if self.config.enable_minimization and user_data:
            t = time.perf_counter()
            min_result = self.minimizer.minimize(user_data)
            ctx.timings["minimize"] = (time.perf_counter() - t) * 1000
            ctx.stages_passed.append("minimize")

            self.audit.log(
                ctx.request_id, "minimize", "minimized",
                details={
                    "reduction": f"{min_result.data_reduction_percent}%",
                    "blocked": min_result.blocked_count,
                },
                timing_ms=ctx.timings["minimize"],
            )

        # Stage 4: LLM Call
        t = time.perf_counter()
        try:
            raw_output = await self._call_llm(processed_input)
            ctx.stages_passed.append("llm_call")
        except Exception as e:
            ctx.stages_failed.append("llm_call")
            ctx.total_time_ms = (time.perf_counter() - start) * 1000
            return {
                "answer": "Error procesando tu solicitud. Intenta de nuevo.",
                "pii_protected": True,
            }
        finally:
            ctx.timings["llm_call"] = (time.perf_counter() - t) * 1000

        # Stage 5: Post-LLM Redaction
        final_output = raw_output
        if self.config.enable_post_llm_redaction:
            t = time.perf_counter()
            post_result = self.post_redactor.redact(raw_output)
            ctx.timings["post_redact"] = (time.perf_counter() - t) * 1000
            final_output = post_result.redacted_output
            ctx.stages_passed.append("post_redact")

            if post_result.pii_found > 0:
                ctx.total_pii_redacted += post_result.pii_found
                self.audit.log(
                    ctx.request_id, "post_redact", post_result.action,
                    pii_count=post_result.pii_found,
                    timing_ms=ctx.timings["post_redact"],
                )

        ctx.total_time_ms = (time.perf_counter() - start) * 1000

        if self.config.enable_audit_log:
            self.audit.log(
                ctx.request_id, "pipeline", "completed",
                details={
                    "stages_passed": ctx.stages_passed,
                    "total_pii": ctx.total_pii_detected,
                    "total_redacted": ctx.total_pii_redacted,
                },
                timing_ms=ctx.total_time_ms,
            )

        return {
            "answer": final_output,
            "pii_protected": True,
            "pii_stats": {
                "detected": ctx.total_pii_detected,
                "redacted": ctx.total_pii_redacted,
            },
        }

    async def _call_llm(self, user_input: str) -> str:
        if self.llm_caller:
            return await self.llm_caller(user_input)

        from openai import AsyncOpenAI
        client = AsyncOpenAI()
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "Eres un asistente útil. Responde de forma concisa."},
                {"role": "user", "content": user_input},
            ],
            temperature=0.3,
        )
        return response.choices[0].message.content

Paso 9: `pii_layer/init.py`

from .scanner import PIIScanner, ScanResult, PIIEntity, PIISeverity
from .redactor import PreLLMRedactor, PostLLMRedactor, PreLLMResult, PostLLMResult
from .minimizer import DataMinimizer, MinResult
from .retention import RetentionScheduler, RetentionResult, DataRecord
from .audit import PIIAuditLogger
from .pipeline import PIIProtectionPipeline, PIIContext
from .config import PIILayerConfig

Paso 10: FastAPI Application

Crea app.py:

import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

from pii_layer import (
    PIIProtectionPipeline, PIIContext, PIILayerConfig,
)

logging.basicConfig(level=logging.INFO)

app = FastAPI(title="PII Protected AI API", version="1.0")


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=10000)
    user_id: str = "anonymous"
    user_data: Optional[dict] = None


class ChatResponse(BaseModel):
    answer: str
    pii_protected: bool = True
    pii_stats: Optional[dict] = None


config = PIILayerConfig()

pipeline = PIIProtectionPipeline(config=config)


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    context = PIIContext(user_id=request.user_id, endpoint="/chat")
    try:
        result = await pipeline.process(
            user_input=request.message,
            user_data=request.user_data,
            context=context,
        )
        return ChatResponse(**result)
    except Exception:
        raise HTTPException(status_code=500, detail="Error interno.")


@app.get("/health")
async def health():
    return {"status": "ok", "pii_layer_version": "1.0"}


@app.get("/audit/stats")
async def audit_stats():
    return pipeline.audit.get_stats()

Paso 11: Tests

Crea tests/test_scanner.py:

from pii_layer.scanner import PIIScanner, PIISeverity
from pii_layer.config import ScannerConfig


def test_clean_text():
    scanner = PIIScanner()
    result = scanner.scan("The weather is nice today.")
    assert not result.has_pii


def test_email_detection():
    scanner = PIIScanner()
    result = scanner.scan("Contact me at test@example.com")
    assert result.has_pii
    assert any(e.entity_type == "EMAIL_ADDRESS" for e in result.entities)


def test_ssn_critical():
    scanner = PIIScanner()
    result = scanner.scan("SSN: 123-45-6789")
    assert result.has_critical
    assert result.max_severity == PIISeverity.CRITICAL


def test_custom_recognizer():
    scanner = PIIScanner()
    result = scanner.scan("Employee EMP-123456 reported.")
    assert any(e.entity_type == "EMPLOYEE_ID" for e in result.entities)


def test_score_threshold():
    strict = PIIScanner(ScannerConfig(score_threshold=0.9))
    lenient = PIIScanner(ScannerConfig(score_threshold=0.3))
    text = "Call Robert at 555-0100"
    strict_result = strict.scan(text)
    lenient_result = lenient.scan(text)
    assert lenient_result.entity_count >= strict_result.entity_count

Crea tests/test_redactor.py:

from pii_layer.redactor import PreLLMRedactor, PostLLMRedactor
from pii_layer.config import RedactorConfig


def test_pre_llm_redaction():
    redactor = PreLLMRedactor()
    result = redactor.redact("My email is john@test.com")
    assert result.redaction_count > 0
    assert "john@test.com" not in result.redacted_text


def test_pre_llm_no_pii():
    redactor = PreLLMRedactor()
    result = redactor.redact("The sky is blue")
    assert result.redaction_count == 0
    assert result.redacted_text == "The sky is blue"


def test_reversible_redaction():
    redactor = PreLLMRedactor(RedactorConfig(reversible=True))
    result = redactor.redact("Contact John at john@test.com")
    assert result.reversible
    assert len(result.mappings) > 0


def test_post_llm_pass():
    redactor = PostLLMRedactor()
    result = redactor.redact("The product costs $49.99.")
    assert result.action == "pass"
    assert result.pii_found == 0


def test_post_llm_block_critical():
    redactor = PostLLMRedactor(RedactorConfig(block_on_critical=True))
    result = redactor.redact("Your SSN is 123-45-6789")
    assert result.action == "blocked"


def test_post_llm_redact_non_critical():
    redactor = PostLLMRedactor()
    result = redactor.redact("Contact support@acme.com for help")
    assert result.action in ("redacted", "pass")

Crea tests/test_minimizer.py:

from pii_layer.minimizer import DataMinimizer
from pii_layer.config import MinimizerConfig


def test_blocks_forbidden_fields():
    minimizer = DataMinimizer()
    data = {"query": "hello", "ssn": "123-45-6789", "password": "secret"}
    result = minimizer.minimize(data)
    assert "ssn" not in result.minimized_data
    assert "password" not in result.minimized_data
    assert result.blocked_count == 2


def test_allowed_fields_filter():
    minimizer = DataMinimizer()
    data = {"query": "hello", "name": "John", "extra": "data"}
    result = minimizer.minimize(data, allowed_fields=["query"])
    assert "extra" not in result.minimized_data
    assert "query" in result.minimized_data


def test_reduction_percent():
    minimizer = DataMinimizer()
    data = {
        "query": "hi", "ssn": "123-45-6789",
        "credit_card": "4111111111111111",
    }
    result = minimizer.minimize(data)
    assert result.data_reduction_percent > 0


def test_excludes_sensitive_by_default():
    minimizer = DataMinimizer()
    data = {"query": "hello", "email": "test@test.com"}
    result = minimizer.minimize(data)
    assert "email" not in result.minimized_data

Crea tests/test_retention.py:

from datetime import datetime, timezone, timedelta
from pii_layer.retention import RetentionScheduler, DataRecord


def test_keep_recent_records():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "llm_log", now - timedelta(days=5), "content")]
    result = scheduler.evaluate(records)
    assert result.kept == 1
    assert result.deleted == 0


def test_delete_expired_records():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "llm_log", now - timedelta(days=45), "content")]
    result = scheduler.evaluate(records)
    assert result.deleted == 1


def test_anonymize_old_chat():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    record = DataRecord("r1", "chat_message", now - timedelta(days=120), "Hello world")
    result = scheduler.evaluate([record])
    assert result.anonymized == 1
    assert record.content != "Hello world"


def test_immediate_delete_redaction_mapping():
    scheduler = RetentionScheduler()
    now = datetime.now(timezone.utc)
    records = [DataRecord("r1", "redaction_mapping", now - timedelta(hours=1), "data")]
    result = scheduler.evaluate(records)
    assert result.deleted == 1

Crea tests/test_pipeline.py:

import pytest
from pii_layer import PIIProtectionPipeline, PIIContext, PIILayerConfig


@pytest.fixture
def pipeline():
    config = PIILayerConfig()

    async def mock_llm(user_input: str) -> str:
        return "Your order is being processed and will arrive tomorrow."

    return PIIProtectionPipeline(config=config, llm_caller=mock_llm)


@pytest.mark.asyncio
async def test_pipeline_clean_input(pipeline):
    context = PIIContext()
    result = await pipeline.process("What is the weather today?", context=context)
    assert result["pii_protected"]
    assert "scan" in context.stages_passed


@pytest.mark.asyncio
async def test_pipeline_with_pii(pipeline):
    context = PIIContext()
    result = await pipeline.process(
        "My name is John Smith and my email is john@test.com",
        context=context,
    )
    assert result["pii_protected"]
    assert context.total_pii_detected > 0


@pytest.mark.asyncio
async def test_pipeline_llm_failure():
    async def failing_llm(user_input):
        raise Exception("API down")

    pipeline = PIIProtectionPipeline(llm_caller=failing_llm)
    context = PIIContext()
    result = await pipeline.process("Hello", context=context)
    assert "Error" in result["answer"] or "error" in result["answer"].lower()
    assert "llm_call" in context.stages_failed


@pytest.mark.asyncio
async def test_pipeline_with_user_data(pipeline):
    context = PIIContext()
    user_data = {
        "query": "order status",
        "order_id": "ORD-123",
        "ssn": "123-45-6789",
    }
    result = await pipeline.process(
        "What is my order status?",
        user_data=user_data,
        context=context,
    )
    assert result["pii_protected"]
    assert "minimize" in context.stages_passed

Ejecución

Correr los tests

cd pii-protection-layer
pytest tests/ -v

# Output esperado:
# tests/test_scanner.py::test_clean_text PASSED
# tests/test_scanner.py::test_email_detection PASSED
# tests/test_scanner.py::test_ssn_critical PASSED
# tests/test_scanner.py::test_custom_recognizer PASSED
# tests/test_scanner.py::test_score_threshold PASSED
# tests/test_redactor.py::test_pre_llm_redaction PASSED
# tests/test_redactor.py::test_pre_llm_no_pii PASSED
# tests/test_redactor.py::test_reversible_redaction PASSED
# tests/test_redactor.py::test_post_llm_pass PASSED
# tests/test_redactor.py::test_post_llm_block_critical PASSED
# tests/test_redactor.py::test_post_llm_redact_non_critical PASSED
# tests/test_minimizer.py::test_blocks_forbidden_fields PASSED
# tests/test_minimizer.py::test_allowed_fields_filter PASSED
# tests/test_minimizer.py::test_reduction_percent PASSED
# tests/test_minimizer.py::test_excludes_sensitive_by_default PASSED
# tests/test_retention.py::test_keep_recent_records PASSED
# tests/test_retention.py::test_delete_expired_records PASSED
# tests/test_retention.py::test_anonymize_old_chat PASSED
# tests/test_retention.py::test_immediate_delete_redaction_mapping PASSED
# tests/test_pipeline.py::test_pipeline_clean_input PASSED
# tests/test_pipeline.py::test_pipeline_with_pii PASSED
# tests/test_pipeline.py::test_pipeline_llm_failure PASSED
# tests/test_pipeline.py::test_pipeline_with_user_data PASSED
# All tests passed!

Correr el servidor

uvicorn app:app --reload --port 8001

Probar con curl

# Clean input (sin PII)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What is the weather today?", "user_id": "test-user"}'

# Input con PII (se redacta antes del LLM)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "My name is John Smith and my SSN is 123-45-6789. What is my order status?", "user_id": "test-user"}'

# Input con user_data (se minimiza)
curl -X POST http://localhost:8001/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Order status?", "user_id": "test-user", "user_data": {"query": "order status", "order_id": "ORD-123", "ssn": "123-45-6789", "email": "john@test.com"}}'

# Audit stats
curl http://localhost:8001/audit/stats

# Health check
curl http://localhost:8001/health

Criterios de éxito

Tu proyecto está completo cuando puedas verificar estos puntos:

Rúbrica de evaluación

Total: 100 puntos

Categoría	Puntos	Criterios clave
PIIScanner	15	Presidio integration (4), custom recognizers (3), severities (3), configurable (3), scan_time_ms (2)
PreLLMRedactor	15	Presidio Anonymizer (4), configurable operators (3), reversible mode (4), mappings (4)
PostLLMRedactor	10	Output scan (3), block on critical (3), redact non-critical (2), fallback message (2)
DataMinimizer	10	Forbidden fields (3), sensitive exclusion (2), generalización (3), reduction metrics (2)
RetentionScheduler	10	Policies per type (3), delete/archive/anonymize (3), evaluation (2), immediate delete for mappings (2)
PIIAuditLogger	10	Log sanitization (3), audit entries (3), stats (2), PII pattern redaction (2)
Pipeline Integration	15	5 stages integrated (5), error handling (3), timing metrics (3), context tracking (4)
Tests	10	Scanner tests (2), redactor tests (2), minimizer tests (2), retention tests (2), pipeline tests (2)
Code Quality	5	Clean structure (2), typing (1), config centralized (1), no hardcoded values (1)

Distribución de notas

Rango	Calificación
90-100	Excelente — PII Protection Layer production-ready
80-89	Muy bien — Protección sólida con mejoras menores
70-79	Bien — Cubre lo básico pero necesita más robustez
60-69	Aceptable — Faltan componentes o profundidad
< 60	Necesita revisión — Brechas en la protección

Errores comunes

1. No cachear el AnalyzerEngine de Presidio

❌ Crear un nuevo AnalyzerEngine en cada scan
✅ Crear uno al inicializar y reutilizarlo

Crear el engine incluye cargar el modelo de spaCy (~200MB). Hazlo una vez en __init__ y reutiliza.

2. Loggear PII en los logs de auditoría

❌ logger.info(f"PII found: {entity.text}")
✅ logger.info(f"PII found: [{entity.entity_type}]")

El audit log registra qué tipo de PII se encontró, no qué valor tenía. El valor es exactamente lo que proteges.

3. No manejar fallos del LLM con fallback

❌ raise Exception("LLM call failed")
✅ return {"answer": "Error procesando.", "pii_protected": True}

Si la API del LLM falla, retorna un fallback seguro, no un error 500.

4. Guardar redaction mappings en logs persistentes

Los mappings de redacción reversible contienen los datos originales. Si se persisten en logs, la redacción fue inútil.

5. No validar que el PostLLMRedactor realmente se ejecuta

En producción, un bug que deshabilita el PostLLMRedactor expone PII en outputs sin que nadie lo note. Los tests deben verificar que PII en outputs se detecta y se actúa.

6. Score threshold demasiado alto

Un threshold de 0.9 pierde muchas detecciones legítimas. Empieza con 0.5 en producción y ajusta según métricas de falsos positivos/negativos.

7. No integrar el DataMinimizer con el pipeline

El minimizer es inútil si los datos completos del usuario se envían al LLM por otra ruta (por ejemplo, en el system prompt o como contexto RAG sin filtrar).

8. Ignorar PII en mensajes de error

Los mensajes de error que incluyen el input del usuario ("Error processing: My SSN is 123-45-6789") filtran PII a los logs y potencialmente al usuario.

Conexión con los módulos siguientes

Tu PII Protection Layer es el sexto artefacto. Cuando avances:

Módulo	Cómo se conecta
Módulo 7: Security Testing	Testeas tu PII layer con adversarial inputs, prompts que intentan extraer PII, y verificas que la redacción funciona
Módulo 8: Integración	Tu PII layer se integra con Injection Defense (M3) + Sanitization (M4) + Secrets (M5) en el Secured AI System

La integración en el Módulo 8 combina todos los pipelines:

Request → Sanitizer (M4) → Injection Detector (M3) → PII Scanner (M6) →
Pre-LLM Redactor (M6) → Data Minimizer (M6) → LLM →
Output Validator (M4) → Post-LLM Redactor (M6) → Content Filter (M4) →
Audit Logger (M4+M6) → Response

Resumen

El PII Protection Layer es el artefacto central del Módulo 6 — integra PII scanning, pre/post-LLM redaction, data minimization, retention, y audit logging en un middleware FastAPI reutilizable
6 componentes trabajan en secuencia: PIIScanner → PreLLMRedactor → DataMinimizer → LLM → PostLLMRedactor → PIIAuditLogger
Configuración centralizada con Pydantic permite ajustar el pipeline sin cambiar código — thresholds, entities, retention periods, strategies
Error handling distingue entre stages que pueden fallar silenciosamente (scan, minimize) y stages que deben fallar con fallback (LLM call, post-redaction)
Tests unitarios verifican cada componente de forma aislada y el pipeline integrado
El pipeline se integra con el Sanitization Pipeline del Módulo 4 y se consolida en el Módulo 8 como parte del Secured AI System
Presidio es el motor de detección y redacción — soporta 30+ tipos de PII, reconocedores custom, y operators configurables

Recursos para el proyecto

Microsoft Presidio Documentation — Documentación completa de Presidio
Presidio Anonymizer Operators — Referencia de operators para redacción
FastAPI Documentation — Framework web para el servidor del pipeline
Pytest Documentation — Framework de testing
OWASP LLM02: Sensitive Information Disclosure — La vulnerabilidad que el pipeline mitiga
Python cryptography Library — Librería de cifrado para encryption at rest

Creado: Marzo 2026 Versión: 1.0

← Anterior7. Compliance Basics: GDPR, CCPA y AI Siguiente →1. Introducción: Security Testing & Auditing