Módulo 4: Bedrock Guardrails And Defense In Depth

5. Manos a la obra: el scrubber de PII propio, determinista

Siguiente →

Descripción

La lección 4 dejó cuatro brechas concretas que Bedrock Guardrails, por diseño, no cubre. Esta lección construye la primera de las dos piezas que empiezan a cerrarlas: guardrails/pre_invoke_checks.py, un script de Python puro que detecta correos electrónicos y números de teléfono en el texto crudo de un manifiesto, antes de que extract-shipment-manifest-fields invoque a Bedrock. Corrió de verdad para escribir esta lección — cada bloque "Qué esperar" de aquí en adelante es salida literal, verificada con pytest sobre once casos fijos.

Conexión con el módulo

Esta es la primera mitad de la defensa en profundidad de este módulo. Complementa —no repite— el mecanismo 3 de la lección 2 (sensitive_information_policy_config, EMAIL/PHONE con ANONYMIZE): donde ese mecanismo es probabilístico y depende de contexto (Brecha 3 de la lección 4), este script es una expresión regular pura, con una garantía distinta y complementaria: el mismo texto de entrada produce, siempre, exactamente el mismo resultado, en cualquier máquina, sin depender de que Bedrock exista, responda, o esté correctamente configurado.

Analogía: el detector de humo de la cocina, no el sistema de videovigilancia del edificio

Un detector de humo doméstico no entiende de incendios complejos, no distingue un fuego eléctrico de uno de grasa, no tiene ninguna noción de contexto. Hace exactamente una cosa: si la concentración de humo cruza un umbral fijo, suena la alarma — siempre, de la misma forma, sin excepción, sin necesitar electricidad de respaldo, sin depender de una conexión a internet, sin que nadie tenga que configurarlo desde un panel remoto. El sistema de videovigilancia inteligente de un edificio entero (el equivalente al guardrail gestionado de Bedrock) es mucho más sofisticado —entiende patrones, distingue contextos, aprende de miles de casos—, pero también depende de electricidad, de una red, de que alguien lo haya configurado bien. pre_invoke_checks.py es el detector de humo de este módulo: simple, mecánico, siempre igual ante el mismo patrón — y precisamente por eso, nunca deja de correr, pase lo que pase con el sistema más sofisticado del edificio.

Paso 1 — El script completo

guardrails/pre_invoke_checks.py, en la raíz de andes-cargo-infra/:

#!/usr/bin/env python3
"""pre_invoke_checks.py -- a deterministic, local PII scrubber that runs BEFORE
extract-shipment-manifest-fields ever calls bedrock:InvokeModel.

This is defense in depth, not a replacement for Bedrock Guardrails' sensitive
information policy (Module 4, lesson 2/3): it is a second, independent check
that does not depend on Bedrock existing, being reachable, or being configured
correctly. See Module 4, lesson 4 for why the managed guardrail alone is not
enough.

Detects two PII patterns by regex over raw manifest text -- EMAIL and PHONE,
the same two entity types the managed guardrail's sensitive_information_policy_config
already covers with ANONYMIZE (Module 3/4, bedrock.tf). Finding the same class of
data twice, with two independently-written mechanisms, is the point: a bug in
one does not silently become the only line of defense.

Never uses random or datetime.now(). Given the same input text, this script
always produces the same output. Run the test suite with:
    pytest guardrails/test_pre_invoke_checks.py -v
"""

from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass, field

# Deliberately simple, readable patterns -- this is a pre-invoke SAFETY NET,
# not an attempt to exhaustively validate email/phone formats (that is not
# this script's job; RFC 5322-complete email matching is famously its own
# rabbit hole, and is not what a guardrail needs).
EMAIL_PATTERN = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")

# Matches phone numbers with at least 7 digits, allowing common separators
# (spaces, dashes, dots, parentheses) and an optional leading "+".
PHONE_PATTERN = re.compile(r"(?<!\d)(\+?\d[\d\-.\s()]{6,}\d)(?!\d)")

EMAIL_PLACEHOLDER = "[EMAIL_REDACTED]"
PHONE_PLACEHOLDER = "[PHONE_REDACTED]"


@dataclass(frozen=True)
class ScrubResult:
    original_text: str
    redacted_text: str
    emails_found: tuple[str, ...] = field(default_factory=tuple)
    phones_found: tuple[str, ...] = field(default_factory=tuple)

    @property
    def found_pii(self) -> bool:
        return bool(self.emails_found or self.phones_found)

    @property
    def entity_counts(self) -> dict[str, int]:
        return {"EMAIL": len(self.emails_found), "PHONE": len(self.phones_found)}


def scrub_pii(text: str) -> ScrubResult:
    """Find and redact EMAIL and PHONE patterns in text. Order matters: emails
    are redacted first, so a phone-like digit run inside a domain name (rare,
    but possible with numeric subdomains) is never double-matched."""
    emails_found = tuple(EMAIL_PATTERN.findall(text))
    redacted = EMAIL_PATTERN.sub(EMAIL_PLACEHOLDER, text)

    phones_found = tuple(m.strip() for m in PHONE_PATTERN.findall(redacted))
    redacted = PHONE_PATTERN.sub(PHONE_PLACEHOLDER, redacted)

    return ScrubResult(
        original_text=text,
        redacted_text=redacted,
        emails_found=emails_found,
        phones_found=phones_found,
    )


def format_report(result: ScrubResult) -> str:
    status = "PII FOUND" if result.found_pii else "CLEAN"
    lines = [
        f"pre_invoke_checks: {status}",
        f"  EMAIL matches: {len(result.emails_found)}",
        f"  PHONE matches: {len(result.phones_found)}",
    ]
    if result.found_pii:
        lines.append("  Redacted text sent onward:")
        lines.append(f"  {result.redacted_text!r}")
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Deterministic PII pre-check over raw manifest text, before bedrock:InvokeModel."
    )
    parser.add_argument("--text", required=True, help="raw manifest text to scan")
    args = parser.parse_args(argv)

    result = scrub_pii(args.text)
    print(format_report(result))
    # Exit code 0 always -- this check REDACTS and continues (defense in depth
    # for logging/downstream storage), it does not block the extraction attempt.
    # Module 4, lesson 4 explains why blocking here would duplicate, not add to,
    # what the managed guardrail's ANONYMIZE action already does at invoke time.
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Tres decisiones de diseño, cada una trazable a una brecha de la lección 4. Primero, scrub_pii nunca lanza una excepción por un texto sin PII — un manifiesto limpio (el caso más común, según la tasa de escalamiento del Módulo 1) pasa sin ninguna fricción. Segundo, el código de salida de main() es siempre 0 — este chequeo redacta y deja continuar, no bloquea, porque bloquear duplicaría exactamente lo que ANONYMIZE ya hace en el guardrail gestionado (comentario explícito en el código: "this check REDACTS and continues... blocking here would duplicate, not add to, what the managed guardrail's ANONYMIZE action already does"). Tercero, los correos se redactan antes que los teléfonos, en ese orden, para evitar que un patrón numérico dentro de un dominio se cuente dos veces.

Paso 2 — Corriendo el scrubber, de verdad

Contra el manifiesto de texto libre del Módulo 1, lección 3 (envío AC-4471), con un correo y un teléfono agregados a propósito:

python3 guardrails/pre_invoke_checks.py --text "Hi team, following up on shipment AC-4471. Contact me at ana.rojas@andescargo.com or call +51 987 654 321 if you need anything. AndesExpress handles pickup Thursday."

Qué esperar (literal — corrido de verdad, mismo entorno de esta guía):

pre_invoke_checks: PII FOUND
  EMAIL matches: 1
  PHONE matches: 1
  Redacted text sent onward:
  'Hi team, following up on shipment AC-4471. Contact me at [EMAIL_REDACTED] or call [PHONE_REDACTED] if you need anything. AndesExpress handles pickup Thursday.'

Fíjate en AC-4471 — la referencia del envío, dos letras y cuatro dígitos con un guion, sobrevive intacta en el texto redactado. PHONE_PATTERN exige una corrida de al menos ocho caracteres de dígitos y separadores, sin ninguna letra mezclada; AC-4471 tiene letras, así que nunca entra al patrón. Ahora, el mismo tipo de manifiesto que Andes Cargo procesa con más frecuencia —el formato clave=valor original, sin ningún dato personal incidental—:

python3 guardrails/pre_invoke_checks.py --text "shipmentId=4471
originCountry=Peru
destinationCountry=Chile
carrier=AndesExpress
weightKg=120"

Qué esperar (literal):

pre_invoke_checks: CLEAN
  EMAIL matches: 0
  PHONE matches: 0

Sin PII, sin ninguna acción de redacción — el caso que, según la tasa de escalamiento del Módulo 1, lección 3, cubre la enorme mayoría del tráfico real de Andes Cargo, incluso dentro de los manifiestos que sí llegan a extract-shipment-manifest-fields (un manifiesto en texto libre sin ningún dato de contacto incidental es perfectamente posible).

Paso 3 — El suite de pytest, once casos fijos

guardrails/test_pre_invoke_checks.py:

"""pytest suite for pre_invoke_checks.py -- fixed, deterministic manifest texts
only. No random, no datetime.now(): the same input text must always produce
the same PASS/FAIL result, on any machine. Run with:
    pytest guardrails/test_pre_invoke_checks.py -v
"""

import pytest

from pre_invoke_checks import format_report, main, scrub_pii

STRUCTURED_MANIFEST_NO_PII = """shipmentId=4471
originCountry=Peru
destinationCountry=Chile
carrier=AndesExpress
weightKg=120"""

FREE_TEXT_MANIFEST_WITH_EMAIL_AND_PHONE = (
    "Hi team, following up on shipment AC-4471. Contact me at "
    "ana.rojas@andescargo.com or call +51 987 654 321 if you need anything. "
    "AndesExpress handles pickup Thursday."
)

FREE_TEXT_MANIFEST_EMAIL_ONLY = (
    "Shipment reference AC-4472, 80kg of textile goods from Lima to Santiago. "
    "Reach the logistics desk at logistica@andescargo.com for questions."
)

FREE_TEXT_MANIFEST_PHONE_ONLY = (
    "Shipment AC-4473 ready for pickup. Call the warehouse at 011-4455-6677 "
    "to confirm the time window."
)

FREE_TEXT_MANIFEST_NO_PII = (
    "Shipment AC-4474: 45kg of electronics parts, Lima to Santiago, "
    "AndesExpress handling the pickup this Thursday."
)


def test_structured_manifest_has_no_pii():
    result = scrub_pii(STRUCTURED_MANIFEST_NO_PII)
    assert result.found_pii is False
    assert result.emails_found == ()
    assert result.phones_found == ()
    assert result.redacted_text == STRUCTURED_MANIFEST_NO_PII


def test_free_text_manifest_detects_email_and_phone():
    result = scrub_pii(FREE_TEXT_MANIFEST_WITH_EMAIL_AND_PHONE)
    assert result.found_pii is True
    assert result.emails_found == ("ana.rojas@andescargo.com",)
    assert len(result.phones_found) == 1
    assert "AC-4471" in result.redacted_text  # shipment reference preserved


def test_email_is_redacted_from_output_text():
    result = scrub_pii(FREE_TEXT_MANIFEST_EMAIL_ONLY)
    assert "logistica@andescargo.com" not in result.redacted_text
    assert "[EMAIL_REDACTED]" in result.redacted_text
    assert result.phones_found == ()


def test_phone_is_redacted_from_output_text():
    result = scrub_pii(FREE_TEXT_MANIFEST_PHONE_ONLY)
    assert "011-4455-6677" not in result.redacted_text
    assert "[PHONE_REDACTED]" in result.redacted_text
    assert result.emails_found == ()


def test_free_text_manifest_without_pii_is_clean():
    result = scrub_pii(FREE_TEXT_MANIFEST_NO_PII)
    assert result.found_pii is False
    assert result.redacted_text == FREE_TEXT_MANIFEST_NO_PII


def test_shipment_reference_is_never_mistaken_for_a_phone_number():
    """AC-4471 has a letter prefix -- the phone pattern requires an unbroken
    run of digits/separators, so a shipment reference like this must never
    trigger a false-positive PHONE match on its own."""
    result = scrub_pii("Shipment reference on our side is AC-4471.")
    assert result.phones_found == ()


def test_entity_counts_reflects_multiple_matches_of_the_same_type():
    text = "Primary contact: ana@andescargo.com. Backup contact: luis@andescargo.com."
    result = scrub_pii(text)
    assert result.entity_counts == {"EMAIL": 2, "PHONE": 0}


def test_report_labels_pii_found_case():
    result = scrub_pii(FREE_TEXT_MANIFEST_WITH_EMAIL_AND_PHONE)
    report = format_report(result)
    assert "PII FOUND" in report
    assert "EMAIL matches: 1" in report
    assert "PHONE matches: 1" in report


def test_report_labels_clean_case():
    result = scrub_pii(STRUCTURED_MANIFEST_NO_PII)
    report = format_report(result)
    assert "CLEAN" in report


def test_cli_end_to_end_with_pii(capsys):
    exit_code = main(["--text", FREE_TEXT_MANIFEST_WITH_EMAIL_AND_PHONE])
    captured = capsys.readouterr()
    assert exit_code == 0
    assert "PII FOUND" in captured.out


def test_cli_end_to_end_clean(capsys):
    exit_code = main(["--text", STRUCTURED_MANIFEST_NO_PII])
    captured = capsys.readouterr()
    assert exit_code == 0
    assert "CLEAN" in captured.out

Once casos: el manifiesto estructurado sin PII, el manifiesto de texto libre con ambos tipos de dato, cada tipo por separado, el caso limpio de texto libre, el caso borde de la referencia de envío que nunca debe confundirse con un teléfono, el conteo de múltiples coincidencias del mismo tipo, el formato del reporte en sus dos estados, y el flujo completo de la interfaz de línea de comandos en sus dos casos.

cd guardrails/
pytest test_pre_invoke_checks.py -v -p no:randomly

Qué esperar (literal — corrido de verdad):

============================= test session starts ==============================
platform darwin -- Python 3.14.0, pytest-9.1.1, pluggy-1.6.0
rootdir: guardrails/
collected 11 items

test_pre_invoke_checks.py::test_structured_manifest_has_no_pii PASSED    [  9%]
test_pre_invoke_checks.py::test_free_text_manifest_detects_email_and_phone PASSED [ 18%]
test_pre_invoke_checks.py::test_email_is_redacted_from_output_text PASSED [ 27%]
test_pre_invoke_checks.py::test_phone_is_redacted_from_output_text PASSED [ 36%]
test_pre_invoke_checks.py::test_free_text_manifest_without_pii_is_clean PASSED [ 45%]
test_pre_invoke_checks.py::test_shipment_reference_is_never_mistaken_for_a_phone_number PASSED [ 54%]
test_pre_invoke_checks.py::test_entity_counts_reflects_multiple_matches_of_the_same_type PASSED [ 63%]
test_pre_invoke_checks.py::test_report_labels_pii_found_case PASSED      [ 72%]
test_pre_invoke_checks.py::test_report_labels_clean_case PASSED          [ 81%]
test_pre_invoke_checks.py::test_cli_end_to_end_with_pii PASSED           [ 90%]
test_pre_invoke_checks.py::test_cli_end_to_end_clean PASSED              [100%]

============================== 11 passed in 0.01s ==============================

Once de once, en menos de un centésimo de segundo — sin red, sin disco más allá del propio código fuente, sin ninguna dependencia externa. La bandera -p no:randomly desactiva la aleatorización del orden de ejecución que el plugin pytest-randomly, instalado en este entorno, aplica por defecto — útil para verificar que ningún test depende del orden en que corre, pero innecesaria aquí para leer la salida en el mismo orden en que los casos aparecen en el archivo.

Errores comunes

Escribir un patrón de teléfono demasiado permisivo, que confunde un ID de envío con un número de teléfono (de un patrón regex mal acotado). Qué pasa: alguien, extendiendo este script, escribe un patrón de teléfono sin el (?<!\d)/(?!\d) (los llamados lookaround, que confirman que no hay otro dígito inmediatamente antes o después de la coincidencia) y termina marcando como "teléfono" cualquier corrida larga de dígitos, incluidos IDs internos con muchos dígitos. Cómo detectarlo: test_shipment_reference_is_never_mistaken_for_a_phone_number (Paso 3) empieza a fallar, o un ID de envío legítimo aparece redactado como [PHONE_REDACTED] en la salida. Cómo corregirlo: los lookaround de PHONE_PATTERN existen exactamente para este caso — confirman los límites exactos de la coincidencia, no solo su contenido. Cualquier patrón nuevo que agregues a este script debería, igual que este, tener al menos un caso de prueba dedicado a confirmar que NO coincide con algo que se le parece mucho, pero no lo es.

Asumir que main() debería devolver un código de salida distinto de cero cuando encuentra PII (de confundir "detectar" con "bloquear"). Qué pasa: alguien, integrando este script en un pipeline, espera que un exit code de 1 indique "hay PII, detén el proceso", y se sorprende cuando el script siempre devuelve 0. Cómo detectarlo: si tu lógica de integración verifica el código de salida de pre_invoke_checks.py para decidir si continuar o no. Cómo corregirlo: el comentario del propio código lo explica —este chequeo redacta y continúa, nunca bloquea, porque bloquear duplicaría el trabajo que ANONYMIZE ya hace en el guardrail gestionado (lección 3)—. El valor de este script no es decidir si la extracción continúa; es garantizar, de forma determinista e independiente del guardrail gestionado, que el texto que sale de aquí ya no lleva PII sin enmascarar, sin importar qué pase después con Bedrock.

Olvidar -p no:randomly y confundirse por qué el orden de la salida de pytest cambia entre corridas (de no revisar qué plugins están instalados). Qué pasa: alguien corre pytest -v sin la bandera, ve un orden distinto al de esta lección, y se pregunta si algo se rompió. Cómo detectarlo: el mismo comando, corrido dos veces seguidas sin -p no:randomly, muestra los mismos once PASSED, pero en un orden diferente cada vez. Cómo corregirlo: esto no es una falla de determinismo del código —cada test individual sigue produciendo, siempre, el mismo resultado (PASSED, con las mismas aserciones)—; es el plugin pytest-randomly reordenando la secuencia de ejecución a propósito, una práctica común para detectar dependencias ocultas entre tests. -p no:randomly desactiva solo el reordenamiento, no ninguna otra verificación.

Ejercicios

Ejercicio 1 — Corre pre_invoke_checks.py tú mismo con un manifiesto de texto libre propio, con un correo electrónico de un dominio que termine en más de tres letras (por ejemplo, .info o .technology). Antes de correrlo, predice si EMAIL_PATTERN lo detectaría, basándote en el patrón [A-Za-z]{2,} del final de la expresión regular.

Ver solución

Sí lo detectaría — {2,} significa "dos o más", sin límite superior, así que dominios de cualquier longitud (.com, .info, .technology, .io) coinciden igual. Es una decisión deliberada del patrón: muchos ejemplos de expresiones regulares para correos usan {2,4} o similar, asumiendo que los dominios son cortos — una asunción cada vez menos cierta con la proliferación de dominios de nivel superior largos. EMAIL_PATTERN, en este script, evita ese supuesto innecesario.

Ejercicio 2 — Explica por qué scrub_pii() redacta los correos ANTES que los teléfonos, no al revés ni simultáneamente. ¿Qué caso concreto, aunque poco común, podría salir mal si el orden se invirtiera?

Ver solución

Un dominio con un subdominio puramente numérico —poco común, pero válido, como contacto@192.168.mi-empresa.com— contiene una corrida larga de dígitos y puntos que, en teoría, podría coincidir con PHONE_PATTERN si ese patrón corriera primero, sobre el texto original completo, antes de que el correo hubiera sido identificado y removido de consideración. Al redactar los correos primero, ese fragmento completo desaparece del texto (reemplazado por [EMAIL_REDACTED]) antes de que PHONE_PATTERN tenga oportunidad de evaluarlo, eliminando la posibilidad de una doble coincidencia sobre el mismo fragmento de texto.

Ejercicio 3 — Predice qué pasaría si corrieras pytest test_pre_invoke_checks.py dos veces seguidas, sin cambiar ninguna línea de código, en dos máquinas distintas. Basándote en la ausencia de random y datetime.now() en pre_invoke_checks.py, ¿esperarías algún resultado distinto entre corridas o entre máquinas?

Ver solución

No, ningún resultado distinto en el contenido de cada test —los once casos deberían pasar exactamente igual, con las mismas aserciones exactas verificadas, en cualquier máquina, en cualquier momento, porque ningún cálculo de scrub_pii() depende de una fuente externa de aleatoriedad, del reloj del sistema, ni de ningún estado compartido entre corridas—. Lo único que podría variar, sin -p no:randomly, es el orden de ejecución (por el plugin pytest-randomly, ver Errores comunes de esta lección) — nunca el resultado PASS/FAIL de un test individual dado el mismo código fuente.

Resumen y siguiente paso

Esta lección construyó pre_invoke_checks.py, un scrubber de PII propio, determinista, que corre antes de cualquier llamada a Bedrock y nunca depende de que el guardrail gestionado exista o esté correctamente configurado. Corriste el script de verdad contra un manifiesto de texto libre con correo y teléfono (PII FOUND, ambos redactados) y contra el formato clave=valor original (CLEAN), y confirmaste, con once casos de pytest, que el comportamiento es correcto, incluido el caso borde de una referencia de envío que nunca debe confundirse con PII.

Antes de avanzar deberías poder: explicar por qué main() siempre devuelve código de salida 0; identificar, sin ayuda, por qué AC-4471 nunca se marca como teléfono; y correr tú mismo el suite completo con un manifiesto propio, añadido como un caso de prueba nuevo.

La lección 6 construye la segunda mitad de esta defensa: post_invoke_checks.py, el validador que confirma, después de la respuesta del modelo, que tiene exactamente la forma de ShipmentFields que la lección 4 demostró que Bedrock Guardrails nunca evalúa.

Recursos

  1. Python Docs — módulo re — referencia oficial de las expresiones regulares usadas en este script, incluidos los lookaround (?<!...)/(?!...).
  2. pytest — Anatomy of a test file — referencia de la estructura de test_pre_invoke_checks.py.
  3. pytest-randomly en PyPI — el plugin responsable del reordenamiento mencionado en Errores comunes de esta lección.
  4. Este módulo, lección 4 (04-why-a-managed-guardrail-is-not-enough-alone.md) — la Brecha 3, fuente directa de la decisión de construir un chequeo determinista independiente del mecanismo probabilístico de Bedrock Guardrails.
← Anterior4. Por qué un guardrail gestionado no basta soloSiguiente →6. Manos a la obra: el validador del esquema de salida