Module 4: Bedrock Guardrails And Defense In Depth

5. Hands-on: the custom, deterministic PII scrubber

Next →

Description

Lesson 4 left four concrete gaps Bedrock Guardrails, by design, doesn't cover. This lesson builds the first of the two pieces starting to close them: guardrails/pre_invoke_checks.py, a pure Python script detecting email addresses and phone numbers in a manifest's raw text, before extract-shipment-manifest-fields invokes Bedrock. It ran for real to write this lesson — every "What to expect" block from here on is literal output, verified with pytest against eleven fixed cases.

Connection to the module

This is the first half of this module's defense in depth. It complements — doesn't repeat — lesson 2's mechanism 3 (sensitive_information_policy_config, EMAIL/PHONE with ANONYMIZE): where that mechanism is probabilistic and context-dependent (lesson 4's Gap 3), this script is a pure regular expression, with a different, complementary guarantee: the same input text always produces exactly the same result, on any machine, with no dependence on Bedrock existing, responding, or being correctly configured.

Analogy: the kitchen's smoke detector, not the building's video surveillance system

A household smoke detector doesn't understand complex fires, doesn't distinguish an electrical fire from a grease fire, has no notion of context at all. It does exactly one thing: if smoke concentration crosses a fixed threshold, the alarm sounds — always, the same way, without exception, needing no backup power, no internet connection, no one having to configure it from a remote panel. An entire building's smart video surveillance system (the equivalent of Bedrock's managed guardrail) is much more sophisticated — it understands patterns, distinguishes contexts, learns from thousands of cases —, but it also depends on electricity, a network, and someone having configured it well. pre_invoke_checks.py is this module's smoke detector: simple, mechanical, always the same in the face of the same pattern — and precisely because of that, it never stops running, no matter what happens to the building's more sophisticated system.

Step 1 — The complete script

guardrails/pre_invoke_checks.py, at the root of 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())

Three design decisions, each traceable to a gap from lesson 4. First, scrub_pii never raises an exception for a text with no PII — a clean manifest (the most common case, per Module 1's escalation rate) passes with no friction. Second, main()'s exit code is always 0 — this check redacts and lets things continue, it doesn't block, because blocking would duplicate exactly what ANONYMIZE already does in the managed guardrail (explicit comment in the code: "this check REDACTS and continues... blocking here would duplicate, not add to, what the managed guardrail's ANONYMIZE action already does"). Third, emails get redacted before phone numbers, in that order, to prevent a numeric pattern inside a domain from being counted twice.

Step 2 — Running the scrubber, for real

Against Module 1, lesson 3's free-text manifest (shipment AC-4471), with an email and a phone number added on purpose:

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."

What to expect (literal — run for real, same environment as this guide):

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.'

Notice AC-4471 — the shipment reference, two letters and four digits with a dash, survives intact in the redacted text. PHONE_PATTERN requires a run of at least eight digit-and-separator characters, with no letters mixed in; AC-4471 has letters, so it never matches the pattern. Now, the same kind of manifest Andes Cargo processes most frequently — the original key=value format, with no incidental personal data:

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

What to expect (literal):

pre_invoke_checks: CLEAN
  EMAIL matches: 0
  PHONE matches: 0

No PII, no redaction action — the case that, per Module 1, lesson 3's escalation rate, covers the vast majority of Andes Cargo's real traffic, even within the manifests that do reach extract-shipment-manifest-fields (a free-text manifest with no incidental contact data is perfectly possible).

Step 3 — The pytest suite, eleven fixed cases

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

Eleven cases: the structured manifest with no PII, the free-text manifest with both data types, each type separately, the clean free-text case, the edge case of a shipment reference that must never be mistaken for a phone number, the count of multiple matches of the same type, the report's format in both states, and the command-line interface's complete flow in both cases.

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

What to expect (literal — run for real):

============================= 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 ==============================

Eleven out of eleven, in under a hundredth of a second — no network, no disk beyond the source code itself, no external dependency. The -p no:randomly flag disables the execution-order randomization the pytest-randomly plugin, installed in this environment, applies by default — useful for verifying no test depends on the order it runs in, but unnecessary here to read the output in the same order the cases appear in the file.

Common mistakes

Writing a phone pattern that's too permissive, confusing a shipment ID with a phone number (poorly-bounded regex pattern mistake). What happens: someone, extending this script, writes a phone pattern with no (?<!\d)/(?!\d) (the so-called lookarounds, confirming there's no other digit immediately before or after the match) and ends up marking any long run of digits as a "phone number," including internal IDs with many digits. How to spot it: test_shipment_reference_is_never_mistaken_for_a_phone_number (Step 3) starts failing, or a legitimate shipment ID shows up redacted as [PHONE_REDACTED] in the output. How to fix it: PHONE_PATTERN's lookarounds exist exactly for this case — they confirm the match's exact boundaries, not just its content. Any new pattern you add to this script should, like this one, have at least one dedicated test case confirming it does NOT match something that closely resembles it, but isn't.

Assuming main() should return a nonzero exit code when it finds PII (confusing "detect" with "block" mistake). What happens: someone, integrating this script into a pipeline, expects an exit code of 1 to mean "there's PII, stop the process," and is surprised when the script always returns 0. How to spot it: if your integration logic checks pre_invoke_checks.py's exit code to decide whether to continue. How to fix it: the code's own comment explains it — this check redacts and continues, it never blocks, because blocking would duplicate the work ANONYMIZE already does in the managed guardrail (lesson 3). This script's value isn't deciding whether the extraction continues; it's guaranteeing, deterministically and independently of the managed guardrail, that the text leaving here no longer carries unmasked PII, regardless of what happens next with Bedrock.

Forgetting -p no:randomly and getting confused about why pytest's output order changes between runs (not checking which plugins are installed mistake). What happens: someone runs pytest -v without the flag, sees a different order than this lesson's, and wonders if something broke. How to spot it: the same command, run twice in a row without -p no:randomly, shows the same eleven PASSED, but in a different order each time. How to fix it: this isn't a determinism failure in the code — every individual test still always produces the same result (PASSED, with the same assertions) — it's the pytest-randomly plugin deliberately reordering the execution sequence, a common practice for detecting hidden dependencies between tests. -p no:randomly disables only the reordering, not any other check.

Exercises

Exercise 1 — Run pre_invoke_checks.py yourself with your own free-text manifest, with an email from a domain ending in more than three letters (for example, .info or .technology). Before running it, predict whether EMAIL_PATTERN would detect it, based on the [A-Za-z]{2,} pattern at the end of the regular expression.

See solution

Yes, it would detect it — {2,} means "two or more," with no upper limit, so domains of any length (.com, .info, .technology, .io) match equally. It's a deliberate pattern decision: many example regular expressions for emails use {2,4} or similar, assuming domains are short — an assumption that's less and less true with the proliferation of long top-level domains. EMAIL_PATTERN, in this script, avoids that unnecessary assumption.

Exercise 2 — Explain why scrub_pii() redacts emails BEFORE phone numbers, not the other way around or simultaneously. What specific, though uncommon, case could go wrong if the order were reversed?

See solution

A domain with a purely numeric subdomain — uncommon, but valid, like contact@192.168.my-company.com — contains a long run of digits and dots that, in theory, could match PHONE_PATTERN if that pattern ran first, on the complete original text, before the email had been identified and removed from consideration. By redacting emails first, that entire fragment disappears from the text (replaced by [EMAIL_REDACTED]) before PHONE_PATTERN gets a chance to evaluate it, eliminating the possibility of a double match on the same piece of text.

Exercise 3 — Predict what would happen if you ran pytest test_pre_invoke_checks.py twice in a row, with no code changes, on two different machines. Based on the absence of random and datetime.now() in pre_invoke_checks.py, would you expect any different result between runs or between machines?

See solution

No, no different result in the content of any test — the eleven cases should pass exactly the same way, with the same exact assertions verified, on any machine, at any time, because no scrub_pii() calculation depends on an external source of randomness, the system clock, or any state shared between runs. The only thing that could vary, without -p no:randomly, is the execution order (because of the pytest-randomly plugin, see this lesson's Common mistakes) — never a given individual test's PASS/FAIL result given the same source code.

Summary and next step

This lesson built pre_invoke_checks.py, a custom, deterministic PII scrubber that runs before any call to Bedrock and never depends on the managed guardrail existing or being correctly configured. You ran the script for real against a free-text manifest with an email and a phone number (PII FOUND, both redacted) and against the original key=value format (CLEAN), and confirmed, with eleven pytest cases, that the behavior is correct, including the edge case of a shipment reference that must never be mistaken for PII.

Before moving on you should be able to: explain why main() always returns exit code 0; identify, unaided, why AC-4471 is never marked as a phone number; and run the complete suite yourself with your own manifest, added as a new test case.

Lesson 6 builds this defense's second half: post_invoke_checks.py, the validator confirming, after the model's response, that it has exactly the ShipmentFields shape lesson 4 demonstrated Bedrock Guardrails never evaluates.

Resources

  1. Python Docs — re module — official reference for the regular expressions used in this script, including the (?<!...)/(?!...) lookarounds.
  2. pytest — Anatomy of a test file — reference for test_pre_invoke_checks.py's structure.
  3. pytest-randomly on PyPI — the plugin responsible for the reordering mentioned in this lesson's Common mistakes.
  4. This module, lesson 4 (04-why-a-managed-guardrail-is-not-enough-alone.md) — Gap 3, the direct source for the decision to build a deterministic check independent of Bedrock Guardrails' probabilistic mechanism.
← Previous4. Why a managed guardrail isn't enough aloneNext →6. Hands-on: the output schema validator