feat(pii): custom user-supplied regex patterns for redaction

Author: TheodoreSpeaks

apps/pii/server.py

@@ -10,7 +10,8 @@
 import time
 from typing import Any
 
-from fastapi import FastAPI
+import regex as regex_module
+from fastapi import FastAPI, HTTPException
 from presidio_analyzer import (
     AnalyzerEngine,
     BatchAnalyzerEngine,
@@ -220,9 +221,11 @@ def _analyze_one(
     entities: list[str] | None,
     score_threshold: float | None,
     return_decision_process: bool = False,
+    ad_hoc_recognizers: list[PatternRecognizer] | None = None,
 ):
     # Regex-only requests reuse a blank NlpArtifacts to skip the spaCy NLP pass;
-    # otherwise analyze() computes artifacts (runs spaCy) as usual.
+    # otherwise analyze() computes artifacts (runs spaCy) as usual. Custom-pattern
+    # recognizers are regex-based, so they run fine against the blank artifacts.
     nlp_artifacts = (
         _BLANK_ARTIFACTS.get(language) if _regex_only(entities, score_threshold) else None
     )
@@ -233,6 +236,7 @@ def _analyze_one(
         score_threshold=score_threshold,
         return_decision_process=return_decision_process,
         nlp_artifacts=nlp_artifacts,
+        ad_hoc_recognizers=ad_hoc_recognizers or None,
     )
 
 
@@ -241,6 +245,7 @@ def _analyze_many(
     language: str,
     entities: list[str] | None,
     score_threshold: float | None,
+    ad_hoc_recognizers: list[PatternRecognizer] | None = None,
 ):
     """Analyze many texts, skipping the spaCy pass for regex-only requests."""
     if _regex_only(entities, score_threshold):
@@ -252,6 +257,7 @@ def _analyze_many(
                 entities=entities,
                 score_threshold=score_threshold,
                 nlp_artifacts=blank,
+                ad_hoc_recognizers=ad_hoc_recognizers or None,
             )
             for text in texts
         ]
@@ -261,33 +267,97 @@ def _analyze_many(
             language=language,
             entities=entities or None,
             score_threshold=score_threshold,
+            ad_hoc_recognizers=ad_hoc_recognizers or None,
         )
     )
 
 
 app = FastAPI(title="Sim Presidio", docs_url=None, redoc_url=None)
 
+# Internal entity id assigned to the i-th user-supplied custom pattern. Never
+# surfaced: the anonymizer maps it back to the pattern's chosen `replacement`, and
+# callers relabel any leftover CUSTOM_<i> span to the pattern's display name.
+CUSTOM_ENTITY_PREFIX = "CUSTOM_"
+
+
+class CustomPattern(BaseModel):
+    """A user-supplied regex pattern. Matches are replaced with `replacement`."""
+
+    regex: str
+    replacement: str = ""
+    name: str = ""
+
+
+def custom_operators(patterns: list[CustomPattern] | None) -> dict[str, dict[str, Any]]:
+    """Raw replace-operator per custom pattern, keyed by its internal entity id."""
+    return {
+        f"{CUSTOM_ENTITY_PREFIX}{i}": {"type": "replace", "new_value": p.replacement}
+        for i, p in enumerate(patterns or [])
+    }
+
+
+def build_custom_recognizers(
+    patterns: list[CustomPattern] | None, language: str
+) -> tuple[list[PatternRecognizer], list[str]]:
+    """Ad-hoc PatternRecognizers + their entity ids for the given custom patterns.
+
+    Each regex is precompiled so a malformed pattern fails fast as a 400 rather
+    than surfacing later as an opaque analyze-time 500."""
+    recognizers: list[PatternRecognizer] = []
+    entity_ids: list[str] = []
+    for i, p in enumerate(patterns or []):
+        try:
+            regex_module.compile(p.regex)
+        except regex_module.error as exc:
+            raise HTTPException(
+                status_code=400, detail=f"Invalid custom pattern regex: {exc}"
+            ) from exc
+        entity = f"{CUSTOM_ENTITY_PREFIX}{i}"
+        recognizers.append(
+            PatternRecognizer(
+                supported_entity=entity,
+                patterns=[Pattern(name=p.name or entity, regex=p.regex, score=0.5)],
+                supported_language=language,
+            )
+        )
+        entity_ids.append(entity)
+    return recognizers, entity_ids
+
+
+def resolve_entities(
+    req_entities: list[str] | None, custom_entity_ids: list[str]
+) -> list[str] | None:
+    """Effective entity filter: the requested built-ins plus the custom entity ids.
+    `None` (detect all built-ins) is preserved only when neither is present, so a
+    custom-only request never widens into detecting every built-in entity."""
+    if req_entities is None and not custom_entity_ids:
+        return None
+    return list(req_entities or []) + custom_entity_ids
+
 
 class AnalyzeRequest(BaseModel):
     text: str
     language: str = "en"
     entities: list[str] | None = None
     score_threshold: float | None = None
     return_decision_process: bool = False
+    patterns: list[CustomPattern] | None = None
 
 
 class AnalyzeBatchRequest(BaseModel):
     texts: list[str]
     language: str = "en"
     entities: list[str] | None = None
     score_threshold: float | None = None
+    patterns: list[CustomPattern] | None = None
 
 
 class AnonymizeRequest(BaseModel):
     text: str
     analyzer_results: list[dict[str, Any]] = []
     anonymizers: dict[str, dict[str, Any]] | None = None
     operators: dict[str, dict[str, Any]] | None = None
+    patterns: list[CustomPattern] | None = None
 
 
 class AnonymizeBatchItem(BaseModel):
@@ -299,6 +369,7 @@ class AnonymizeBatchRequest(BaseModel):
     items: list[AnonymizeBatchItem] = []
     anonymizers: dict[str, dict[str, Any]] | None = None
     operators: dict[str, dict[str, Any]] | None = None
+    patterns: list[CustomPattern] | None = None
 
 
 class RedactRequest(BaseModel):
@@ -308,6 +379,7 @@ class RedactRequest(BaseModel):
     score_threshold: float | None = None
     anonymizers: dict[str, dict[str, Any]] | None = None
     operators: dict[str, dict[str, Any]] | None = None
+    patterns: list[CustomPattern] | None = None
 
 
 class RedactBatchRequest(BaseModel):
@@ -317,6 +389,7 @@ class RedactBatchRequest(BaseModel):
     score_threshold: float | None = None
     anonymizers: dict[str, dict[str, Any]] | None = None
     operators: dict[str, dict[str, Any]] | None = None
+    patterns: list[CustomPattern] | None = None
 
 
 def build_operators(
@@ -332,6 +405,17 @@ def build_operators(
     return operators
 
 
+def resolve_operators(
+    anonymizers: dict[str, dict[str, Any]] | None,
+    operators: dict[str, dict[str, Any]] | None,
+    patterns: list[CustomPattern] | None,
+) -> dict[str, OperatorConfig] | None:
+    """Merge the caller's operators with the per-custom-pattern replace operators."""
+    raw = dict(anonymizers or operators or {})
+    raw.update(custom_operators(patterns))
+    return build_operators(raw)
+
+
 def run_anonymize(
     text: str,
     raw_results: list[dict[str, Any]],
@@ -366,12 +450,15 @@ def supported_entities(language: str = "en") -> list[str]:
 @app.post("/analyze")
 def analyze(req: AnalyzeRequest) -> list[dict[str, Any]]:
     started = time.perf_counter()
+    recognizers, custom_ids = build_custom_recognizers(req.patterns, req.language)
+    entities = resolve_entities(req.entities, custom_ids)
     results = _analyze_one(
         req.text,
         req.language,
-        req.entities,
+        entities,
         req.score_threshold,
         req.return_decision_process,
+        recognizers,
     )
     logger.info(
         "analyze lang=%s chars=%d entities=%d duration_ms=%.1f",
@@ -387,14 +474,16 @@ def analyze(req: AnalyzeRequest) -> list[dict[str, Any]]:
 def analyze_batch(req: AnalyzeBatchRequest) -> list[list[dict[str, Any]]]:
     """Analyze many texts in one pass (spaCy nlp.pipe), returning one span list
     per input in request order — the batched counterpart to /analyze."""
-    results = _analyze_many(req.texts, req.language, req.entities, req.score_threshold)
+    recognizers, custom_ids = build_custom_recognizers(req.patterns, req.language)
+    entities = resolve_entities(req.entities, custom_ids)
+    results = _analyze_many(req.texts, req.language, entities, req.score_threshold, recognizers)
     return [[r.to_dict() for r in per_text] for per_text in results]
 
 
 @app.post("/anonymize")
 def anonymize(req: AnonymizeRequest) -> dict[str, Any]:
     started = time.perf_counter()
-    operators = build_operators(req.anonymizers or req.operators)
+    operators = resolve_operators(req.anonymizers, req.operators, req.patterns)
     result = run_anonymize(req.text, req.analyzer_results, operators)
     logger.info(
         "anonymize chars=%d spans=%d duration_ms=%.1f",
@@ -422,7 +511,7 @@ def anonymize_batch(req: AnonymizeBatchRequest) -> dict[str, list[str]]:
     """Mask many texts in one pass, returning masked text per item in request
     order — the batched counterpart to /anonymize. Anonymization is pure string
     work (no NLP), so callers should send only items with detected spans."""
-    operators = build_operators(req.anonymizers or req.operators)
+    operators = resolve_operators(req.anonymizers, req.operators, req.patterns)
     return {
         "texts": [
             run_anonymize(item.text, item.analyzer_results, operators).text
@@ -438,8 +527,12 @@ def redact(req: RedactRequest) -> dict[str, str]:
     with no detected PII passes through unchanged. The analyzer results feed the
     anonymizer directly (no dict round-trip)."""
     started = time.perf_counter()
-    operators = build_operators(req.anonymizers or req.operators)
-    results = _analyze_one(req.text, req.language, req.entities, req.score_threshold)
+    recognizers, custom_ids = build_custom_recognizers(req.patterns, req.language)
+    entities = resolve_entities(req.entities, custom_ids)
+    operators = resolve_operators(req.anonymizers, req.operators, req.patterns)
+    results = _analyze_one(
+        req.text, req.language, entities, req.score_threshold, ad_hoc_recognizers=recognizers
+    )
     text = (
         req.text
         if not results
@@ -466,8 +559,10 @@ def redact_batch(req: RedactBatchRequest) -> dict[str, list[str]]:
     the anonymizer directly (no dict round-trip), and anonymization runs only on
     texts that actually matched."""
     started = time.perf_counter()
-    operators = build_operators(req.anonymizers or req.operators)
-    analyzed = _analyze_many(req.texts, req.language, req.entities, req.score_threshold)
+    recognizers, custom_ids = build_custom_recognizers(req.patterns, req.language)
+    entities = resolve_entities(req.entities, custom_ids)
+    operators = resolve_operators(req.anonymizers, req.operators, req.patterns)
+    analyzed = _analyze_many(req.texts, req.language, entities, req.score_threshold, recognizers)
     masked: list[str] = []
     total_spans = 0
     for text, per_text in zip(req.texts, analyzed):
@@ -484,8 +579,8 @@ def redact_batch(req: RedactBatchRequest) -> dict[str, list[str]]:
         "redact_batch lang=%s texts=%d entities=%s nlp=%s spans=%d duration_ms=%.1f",
         req.language,
         len(req.texts),
-        len(req.entities) if req.entities else "all",
-        "skip" if _regex_only(req.entities, req.score_threshold) else "full",
+        len(entities) if entities else "all",
+        "skip" if _regex_only(entities, req.score_threshold) else "full",
         total_spans,
         (time.perf_counter() - started) * 1000,
     )

apps/sim/app/api/guardrails/mask-batch/route.ts

@@ -25,11 +25,11 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
   const parsed = await parseRequest(guardrailsMaskBatchContract, request, {})
   if (!parsed.success) return parsed.response
 
-  const { texts, entityTypes, language } = parsed.data.body
+  const { texts, entityTypes, language, customPatterns } = parsed.data.body
 
   try {
     const startedAt = performance.now()
-    const masked = await maskPIIBatch(texts, entityTypes, language)
+    const masked = await maskPIIBatch(texts, entityTypes, language, customPatterns)
     logger.info('Masked PII batch', {
       count: texts.length,
       durationMs: Math.round(performance.now() - startedAt),

apps/sim/app/api/guardrails/validate/route.ts

@@ -16,6 +16,7 @@ import {
 import { checkAndBillPayerOverageThreshold } from '@/lib/billing/threshold-billing'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import type { CustomPiiPattern } from '@/lib/guardrails/pii-entities'
 import { validateHallucination } from '@/lib/guardrails/validate_hallucination'
 import { validateJson } from '@/lib/guardrails/validate_json'
 import { validatePII } from '@/lib/guardrails/validate_pii'
@@ -63,6 +64,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       piiEntityTypes,
       piiMode,
       piiLanguage,
+      piiCustomPatterns,
     } = body
 
     if (!validationType) {
@@ -280,6 +282,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => {
       piiEntityTypes,
       piiMode,
       piiLanguage,
+      piiCustomPatterns,
       authHeaders,
       requestId
     )
@@ -392,6 +395,7 @@ async function executeValidation(
   piiEntityTypes: string[] | undefined,
   piiMode: string | undefined,
   piiLanguage: string | undefined,
+  piiCustomPatterns: CustomPiiPattern[] | undefined,
   authHeaders: { cookie?: string; authorization?: string; billingAttribution?: string } | undefined,
   requestId: string
 ): Promise<{
@@ -450,6 +454,7 @@ async function executeValidation(
       entityTypes: piiEntityTypes || [], // Empty array = detect all PII types
       mode: (piiMode as 'block' | 'mask') || 'block', // Default to block mode
       language: piiLanguage || 'en',
+      customPatterns: piiCustomPatterns,
       requestId,
     })
   }

apps/sim/blocks/blocks/guardrails.ts

@@ -214,6 +214,17 @@ Return ONLY the regex pattern - no explanations, no quotes, no forward slashes,
       },
       dependsOn: ['validationType'],
     },
+    {
+      id: 'piiCustomPatterns',
+      title: 'Custom Patterns',
+      type: 'table',
+      columns: ['Name', 'Pattern', 'Replacement'],
+      condition: {
+        field: 'validationType',
+        value: ['pii'],
+      },
+      dependsOn: ['validationType'],
+    },
   ],
   tools: {
     access: ['guardrails_validate'],
@@ -260,6 +271,10 @@ Return ONLY the regex pattern - no explanations, no quotes, no forward slashes,
       type: 'string',
       description: 'Language for PII detection (default: en)',
     },
+    piiCustomPatterns: {
+      type: 'json',
+      description: 'Custom regex patterns to detect and replace (name, pattern, replacement rows)',
+    },
   },
   outputs: {
     input: {