Building a Regex Library for Email and SSN Detection

When a Data Subject Request (DSR) lands, the first thing an engineer has to prove is that every structurally unambiguous identifier for that subject was found before any probabilistic model spent a cycle. Email addresses and US Social Security Numbers are the two most common of these: both are fixed grammars you either match or leak. This page is the concrete implementation the Regex Pattern Libraries for PII cluster wraps in a validated schema, and it sits inside the broader PII Extraction & Redaction Pipelines architecture as the deterministic first pass. Anyone building the extraction layer of a DSR pipeline encounters the same three traps: patterns that backtrack catastrophically on hostile input, boundary assertions that let a nine-digit invoice number masquerade as an SSN, and matches emitted without the stable offset and confidence that downstream redaction and audit need. The goal here is a small, versioned matcher that eliminates catastrophic backtracking, enforces regulatory range exclusions, and hands every match to a routing layer with an auditable confidence score before anything is redacted.

Deterministic email and SSN detection flow from raw payload to audit log A raw DSR payload is split into overlapping chunks so identifiers straddling a chunk edge are not lost. Each chunk is scanned in parallel by a precompiled email regex and a precompiled SSN regex whose negative lookaheads exclude the SSA 000, 666 and 900-999 ranges and zeroed group or serial components. Matches are emitted as MatchResult records keyed by absolute start and end offset with a confidence score. A confidence router compares each score to a threshold: matches at or above it flow to deterministic auto-redaction, and matches below it flow to a role-gated manual-review queue. Every decision is written to an append-only audit log carrying the pattern-library version hash, offsets, and a salted hash of the value only. Raw DSR payload multi-GB, untrusted text Overlapping chunk stream overlap = _MAX_MATCH_LEN memory-bounded, edge-safe Precompiled email regex (?!.*\.\.) (?!.*@example\.com) re.ASCII · Punycode-normalised Precompiled SSN regex (?!(000|666|9\d{2})-) & more SSA area / group / serial excl. MatchResult (start_idx, end_idx) deduped by absolute offset confidence ≥ threshold? at / above Auto-redact deterministic mask below Manual review role-gated queue Append-only audit log pattern_library_version offsets + salted hash only

Prerequisites

  • Python 3.11+ — for zoneinfo (audit timestamps), pattern-matching, and the improved re error messages.
  • pydantic>=2.6 — pattern records are modelled as frozen Pydantic v2 models so a malformed expression fails at definition time rather than silently matching nothing.
  • The third-party regex module (optional) — only if you need atomic groups (?>...) or possessive quantifiers to guarantee linear-time execution on adversarial input; the standard-library re is sufficient for the patterns below.
  • Infra: a dead-letter queue (Kafka topic or SQS) for documents that trip the per-document timeout, and an append-only audit sink. Identifiers must never be written to operational logs in plaintext, per the data-minimization duty of GDPR Art. 5(1)©.

No NLP dependencies belong in this layer. Contextually ambiguous matches are handed off to the NLP-Based Entity Recognition stage; this module stays deterministic.

Step-by-step implementation

Step 1 — Engineer the pattern matrix

Production-grade detection balances syntactic compliance with pragmatic precision. The canonical RFC 5322 specification for email is computationally prohibitive and produces unacceptable false-positive rates against unstructured telemetry. A tiered approach isolates high-confidence syntactic matches and defers ambiguous cases to review.

For email, the baseline pattern prioritises alphanumeric local parts, explicit subaddressing (+), and standard domain structures:

\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b

Edge-case hardening requires explicit exclusion of common false positives — internal routing tokens, base64 fragments, and log identifiers that mimic email syntax. A negative lookahead strips matches containing consecutive dots or known test environments (@example.com). Internationalised domain names (IDNs) must be normalised to Punycode before regex evaluation to prevent Unicode homograph spoofing.

SSN detection demands stricter structural validation and regulatory exclusion. The AAA-GG-SSSS format is universally recognised, but compliance requires filtering invalid ranges per SSA issuance and randomization rules:

\b(?!(000|666|9\d{2})-)(?!\d{3}-00-)(?!\d{3}-\d{2}-0000)\d{3}-\d{2}-\d{4}\b

This excludes 000 area codes, 666 (historically reserved), 900–999 (never issued), and zeroed group or serial components. In DSR corpora, SSNs also appear masked (***-**-1234) or concatenated without delimiters in legacy exports; those are handled by a secondary fallback pattern with length constraints rather than by loosening the primary boundary assertions.

Step 2 — Model each pattern as a validated, versioned record

Compile-time validation is what turns a folder of ad-hoc re.compile() calls into a governed library. Modelling each pattern as a frozen Pydantic v2 record means a malformed expression raises a ValidationError at import, and the whole set can be hashed into a version string that every later match cites for GDPR Art. 30 accountability.

import re
import hashlib
from typing import Pattern
from pydantic import BaseModel, ConfigDict, field_validator, computed_field


class PatternSpec(BaseModel):
    """A single, immutable, validated PII pattern."""

    model_config = ConfigDict(frozen=True)

    pattern_type: str
    source: str
    base_confidence: float

    @field_validator("source")
    @classmethod
    def _must_compile(cls, v: str) -> str:
        """Reject an expression that will not compile at definition time."""
        re.compile(v, re.ASCII | re.VERBOSE)
        return v

    @computed_field  # type: ignore[prop-decorator]
    @property
    def compiled(self) -> Pattern[str]:
        return re.compile(self.source, re.ASCII | re.VERBOSE)

Step 3 — Compile once and scan a memory-bounded stream

Deterministic execution requires precompilation, thread-safe invocation, and memory-bounded streaming — never re.finditer against a raw multi-gigabyte payload held whole in memory. See the Python re module documentation for flag semantics. Chunking introduces one hazard: an identifier split across a chunk edge. The matcher below carries an overlap window equal to the longest possible match so boundary-straddling identifiers are neither missed nor double-counted.

from dataclasses import dataclass
from typing import Iterator


@dataclass(frozen=True)
class MatchResult:
    pattern_type: str
    match_value: str
    confidence: float
    start_idx: int
    end_idx: int


class DeterministicPIIMatcher:
    """Precompiled, ReDoS-safe matcher for email and SSN identifiers."""

    # Longest realistic match; used as the inter-chunk overlap window.
    _MAX_MATCH_LEN = 320

    def __init__(self) -> None:
        self.email_re = re.compile(
            r"""
            \b
            (?!.*\.\.)               # block consecutive dots
            (?!.*@example\.com)      # block reserved test domain
            [A-Za-z0-9._%+-]+
            @
            [A-Za-z0-9.-]+
            \.[A-Za-z]{2,}
            \b
            """,
            re.ASCII | re.VERBOSE,
        )
        self.ssn_re = re.compile(
            r"""
            \b
            (?!(000|666|9\d{2})-)    # SSA area-code exclusions
            (?!\d{3}-00-)            # group-code exclusion
            (?!\d{3}-\d{2}-0000)     # serial-code exclusion
            \d{3}-\d{2}-\d{4}
            \b
            """,
            re.ASCII | re.VERBOSE,
        )

    def _chunks(self, text: str, size: int = 8192) -> Iterator[tuple[int, str]]:
        """Yield (absolute_offset, window) with overlap to catch edge matches."""
        step = size - self._MAX_MATCH_LEN
        start = 0
        while start < len(text):
            yield start, text[start : start + size]
            start += step

    def scan_stream(self, payload: str) -> list[MatchResult]:
        results: dict[tuple[int, int], MatchResult] = {}
        for offset, window in self._chunks(payload):
            for m in self.email_re.finditer(window):
                key = (offset + m.start(), offset + m.end())
                results[key] = MatchResult(
                    "EMAIL", m.group(0), 0.95, *key
                )
            for m in self.ssn_re.finditer(window):
                key = (offset + m.start(), offset + m.end())
                results[key] = MatchResult(
                    "SSN", m.group(0), 1.0, *key
                )
        return [results[k] for k in sorted(results)]

Keying results by absolute (start, end) offset makes the overlap idempotent: an identifier seen in two adjacent windows collapses to a single MatchResult.

Step 4 — Route matches by confidence

Not every syntactic match warrants automatic redaction. A routing layer evaluates confidence against a configurable threshold; matches below it go to a role-gated review queue rather than triggering irreversible masking. Ambiguous tokens such as user_12345@internal.corp are exactly what belong in the Confidence Scoring Thresholds workflow instead of being auto-redacted.

def route_matches(
    matches: list[MatchResult], threshold: float = 0.90
) -> tuple[list[MatchResult], list[MatchResult]]:
    """Split matches into auto-redact and manual-review buckets."""
    auto_redact: list[MatchResult] = []
    manual_review: list[MatchResult] = []
    for match in matches:
        adjusted = match.confidence
        # Numeric-heavy local parts often indicate system IDs, not people.
        if match.pattern_type == "EMAIL":
            local = match.match_value.split("@", 1)[0]
            if any(c.isdigit() for c in local):
                adjusted -= 0.05
        (auto_redact if adjusted >= threshold else manual_review).append(match)
    return auto_redact, manual_review

The routing logic must be version-controlled alongside the pattern matrix so an audit can reproduce exactly why a given token was auto-redacted or held.

Configuration reference

Parameter Type Default Compliance note
chunk_size int 8192 Bounds peak memory on multi-GB payloads; too small inflates overlap re-scans.
_MAX_MATCH_LEN int 320 Inter-chunk overlap; must exceed the longest possible identifier or edge matches are lost.
route.threshold float 0.90 Matches below this route to review, not redaction — preserves data utility under GDPR Art. 5(1)© minimization.
email base_confidence float 0.95 Syntactic-only; contextual confirmation deferred to NLP.
ssn base_confidence float 1.0 Structural + SSA range exclusion make a valid match near-certain.
per_document_timeout float (s) 2.0 Hostile input past this is quarantined to the DLQ, protecting the DSR statutory deadline.
pattern_library_version str (hash) computed Logged with every match for GDPR Art. 30 reproducibility.

Hash the compiled library at startup to produce pattern_library_version:

def library_version(specs: list[PatternSpec]) -> str:
    """Stable hash proving which pattern set produced a match."""
    joined = "|".join(sorted(f"{s.pattern_type}:{s.source}" for s in specs))
    return hashlib.sha256(joined.encode("utf-8")).hexdigest()[:16]

Verification

Confirm correctness with a positive/negative matrix that pins down the boundary and range exclusions — the two places these patterns most often regress.

import pytest

matcher = DeterministicPIIMatcher()


@pytest.mark.parametrize(
    "text, expect_type",
    [
        ("contact jane.doe+dsr@corp.io today", "EMAIL"),
        ("file under 123-45-6789 please", "SSN"),
        ("ping ops@example.com now", None),       # reserved test domain
        ("invoice 000-12-3456 paid", None),       # invalid SSA area code
        ("ref 900-11-2222 archived", None),       # never-issued range
        ("case 123-00-4567 closed", None),        # zeroed group code
    ],
)
def test_boundaries(text: str, expect_type: str | None) -> None:
    found = matcher.scan_stream(text)
    if expect_type is None:
        assert found == []
    else:
        assert found and found[0].pattern_type == expect_type


def test_edge_straddling_match_not_lost() -> None:
    """An SSN split across a chunk boundary is found exactly once."""
    payload = ("x" * 8190) + "123-45-6789" + ("y" * 100)
    results = [r for r in matcher.scan_stream(payload) if r.pattern_type == "SSN"]
    assert len(results) == 1

A correct run emits an audit record per match carrying pattern_library_version, the (start_idx, end_idx) offsets, and a salted hash of the value — never the value itself. The compliance assertion is simple: given the same input and the same library version, the offsets and routing decision must be byte-for-byte reproducible.

Troubleshooting

Catastrophic backtracking stalls a worker : Root cause: nested or ambiguous quantifiers such as (.*)* or (a+)+ on adversarial input. Fix: keep quantifiers flat, wrap each scan in per_document_timeout, and quarantine offenders to the DLQ. Where lookbehind is unavoidable, switch to the regex module’s atomic groups (?>...) for guaranteed linear time.

SSN false positives from invoice or phone numbers : Root cause: the fallback unhyphenated pattern matching any nine-digit run. Fix: apply the SSA range exclusions to the fallback too and require a modulo/length constraint before emitting a match; downgrade fallback matches below route.threshold so they land in review.

Identifier missed at a chunk boundary : Root cause: _MAX_MATCH_LEN smaller than the longest real identifier, so the overlap window doesn’t cover it. Fix: raise _MAX_MATCH_LEN above your longest expected match and re-run test_edge_straddling_match_not_lost.

Unicode homograph email bypasses the pattern : Root cause: an IDN evaluated as raw Unicode. Fix: normalise the domain to Punycode before scanning and keep the matcher on re.ASCII so mixed-script spoofs cannot slip a lookalike character past the domain class.

Audit trail itself becomes a PII store : Root cause: logging match_value in plaintext. Fix: store only derived values — the library-version hash, source-record reference, offsets, and a salted hash of the matched string — keeping identifiable data out of operational logs per GDPR Art. 5(1)©.