Skip to Content

SecurityService Sanitization Pipeline (PII Redact → Jailbreak Check)

Data Flow

SECURITY SERVICE SANITIZATION  —  Data Flow

   sanitize_input(text)
        │
        ▼  if not settings.security_enabled:  return passthrough
        │
        ▼  LAYER 1 — Presidio
            AnalyzerEngine().analyze(text, language="en")
            AnonymizerEngine().anonymize(text, results)  ──► safe_text
        │
        ▼  LAYER 2 — Lakera Guard (if LAKERA_GUARD_API_KEY set)
            POST https://api.lakera.ai/v1/prompt_injection
                  { "input": safe_text }
                  score < settings.security_jailbreak_threshold   (0.8)
        │
        ▼  LAYER 3 — local regex fallback (no Lakera key)
            re.search(JAILBREAK_REGEXES, safe_text)
        │
        ▼
   return {
       "safe_text":          "...",
       "is_injection":        bool,
       "redacted_entities":   [...],
   }

What? (Concept Overview)

SecurityService is the outbound safety gate that runs on every user message before it touches an LLM. It performs two composable checks: (1) PII redaction (Presidio-based with regex fallback, plus a custom UK_NINO PatternRecognizer); (2) jailbreak detection (Lakera Guard API first, then heuristic keyword blacklist). The service constructor takes no args — it eagerly reads from Settings so the safety posture is environment-driven.

Project Context

This file (app/services/security_service.py) is the only place that touches Presidio or Lakera. Every agent routes raw user input through security_service.sanitize_input(text) before any DB or prompt assembly; every LLM call is preceded by security_service.check_jailbreak(text) returning (is_safe, reason). Fail-open on missing Lakera API key (dev convenience); fail-closed on hard rule violations.

How? (Quick Reference Blocks)

3.1 Service Construction with Presidio + Custom Patterns

# app/services/security_service.py — SecurityService.__init__
from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern
from presidio_anonymizer import AnonymizerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
 
class SecurityService:
    def __init__(self):
        configuration = {
            "nlp_engine_name": "spacy",
            "models": [{"lang_code": "en", "model_name": "en_core_web_lg"}],
        }
        provider = NlpEngineProvider(nlp_configuration=configuration)
        nlp_engine = provider.create_engine()
 
        self.enabled = settings.security_enabled
        self.redact_pii = settings.pii_redaction_enabled
 
        if self.redact_pii:
            self.analyzer = AnalyzerEngine(
                nlp_engine=nlp_engine, supported_languages=["en"]
            )
            # Custom UK National Insurance regex: 2 letters + 6 digits + 1 [A-D]
            nino_regex = r"(?i)\b[A-Z]{2}\s?[0-9]{2}\s?[0-9]{2}\s?[0-9]{2}\s?[A-D]\b"
            nino_pattern = Pattern(name="uk_nino_pattern", regex=nino_regex, score=0.85)
            nino_recognizer = PatternRecognizer(
                supported_entity="UK_NINO", patterns=[nino_pattern]
            )
            self.analyzer.registry.add_recognizer(nino_recognizer)
            self.anonymizer = AnonymizerEngine()
        else:
            self.analyzer = None
            self.anonymizer = None
 
        # Auth setup
        self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
        self.secret_key = settings.secret_key
        self.algorithm = settings.jwt_algorithm
        self.expire_minutes = settings.access_token_expire_minutes
        self.lakera_guard_api_key = settings.lakera_guard_api_key
 
        # Regex PII fallback (when Presidio not available)
        self.pii_patterns = {
            "EMAIL":    r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            "PHONE_UK": r"(?:(?:\+44\s?|0)(?:7\d{3}|\d{4})\s?\d{6})",
            "CREDIT_CARD": r"\b(?:\d{4}[-\s]?){3}\d{4}\b",
        }
        self.injection_keywords = [
            "ignore previous instructions", "act as a pirate", "system prompt",
            "developer mode", "you are now", "launder money", "money laundering",
            "forge a check", "bypass 2fa", "base64", "encoded string",
            "system override", "disable_content_filter", "unrestrained ai",
            "forget you are", "simulated", "hypothetically",
        ]

3.2 Presidio-Driven Redaction

# app/services/security_service.py — _redact_with_presidio
def _redact_with_presidio(self, text: str) -> str:
    if not self.analyzer or not self.anonymizer:
        return text
    try:
        results = self.analyzer.analyze(
            text=text,
            entities=["PHONE_NUMBER", "EMAIL_ADDRESS", "CREDIT_CARD",
                      "IBAN_CODE", "US_BANK_NUMBER", "PERSON"],
            language="en",
        )
        from presidio_anonymizer.entities import OperatorConfig
 
        anonymized_result = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators={
                "DEFAULT": OperatorConfig("replace", {"new_value": "[CONFIDENTIAL_DATA]"}),
                "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL]"}),
                "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[PHONE]"}),
                "PERSON": OperatorConfig("replace", {"new_value": "[NAME]"}),
            },
        )
        return anonymized_result.text
    except Exception as e:
        logger.error(f"Presidio Redaction Error: {e}")
        return text    # fail-open: returning original text
 
def sanitize_input(self, text: str) -> str:
    if not self.enabled or not self.redact_pii:
        return text
    if self.analyzer:
        return self._redact_with_presidio(text)
    # Regex fallback when Presidio isn't loaded
    sanitized = text
    for label, pattern in self.pii_patterns.items():
        sanitized = re.sub(pattern, f"[{label}_REDACTED]", sanitized)
    return sanitized

3.3 Lakera Guard + Heuristic Jailbreak Check

# app/services/security_service.py — check_jailbreak
def _check_with_lakera(self, text: str) -> Tuple[bool, str]:
    if not self.lakera_guard_api_key:
        return True, ""   # dev: fail-open without key
    try:
        response = requests.post(
            "https://api.lakera.ai/v2/guard",
            headers={"Authorization": f"Bearer {self.lakera_guard_api_key}"},
            json={"messages": [{"role": "user", "content": text}]},
        )
        if response.status_code != 200:
            logger.warning(f"Lakera API Warning: {response.status_code} - {response.text}")
            return True, ""
        result = response.json()
        # 'flagged: True' means prompt-injection detected
        if result.get("flagged") is True:
            logger.info(f"Lakera Blocked Input: {result}")
            return False, "Lakera Guard: Prompt Injection Detected"
    except Exception as e:
        logger.error(f"Lakera Guard API Error: {e}")
        return True, ""
    return True, ""
 
def check_jailbreak(self, text: str) -> Tuple[bool, str]:
    if not self.enabled:
        return True, ""
 
    sanitized_text_for_check = self.sanitize_input(text)
 
    # 1. Lakera AI-based detection first
    is_safe, reason = self._check_with_lakera(sanitized_text_for_check)
    if not is_safe:
        return False, reason
 
    # 2. Heuristic keyword fallback
    text_lower = text.lower()    # check original text (jailbreaks hide in PII-redacted gaps)
    for keyword in self.injection_keywords:
        if keyword in text_lower:
            return False, f"Blocked keyword detected: '{keyword}'"
 
    if len(text) > 10000:
        return False, "Input exceeds safety length limits"
    return True, ""

Why? (Parameter Breakdown

Common Pitfalls

  1. Calling sanitize_input once before the LLM and forgetting to call it again in logs. The Postgres conversation log records the original text, leaking PII to the DB. Sanitise at the START of the pipeline AND replace at any logging site.
  2. Forgetting to add custom entity types when extending Presidio. New UK recognisers (UK_SORT_CODE, UK_POSTCODE) must be registered and listed in analyze(entities=[...]) to take effect.

Real-World Interview Prep

Q1: How do you handle the “PII desync” problem — redaction succeeds but logging happens BEFORE redaction?

A: Three fixes. (1) Sanitise at the input boundary — every request handler calls security.sanitize_input(message) first, then passes to all downstream code. (2) Sanitise before logging — log logger.info("...", extra={”message”: sanitized}) instead of raw text. (3) Audit trails — every call site that produces logs that might contain raw PII gets a lint check (grep -nE 'logger\.(info|error)\(\".*\"' app/). Production-readiness: log-scrub middleware that post-processes every log record.

Q2: How do you test the check_jailbreak function with known-bad + known-good inputs?

A: Three layers. (1) Unit: feed "ignore previous instructions" → assert (False, "Blocked keyword: '...'"). (2) Edge: feed "I really enjoy ignoring procrastination in my workflow" → assert keyword-substring match is fine or refined (e.g. requires word boundaries). (3) Lakera mocks via responses library: mock requests.post("https://api.lakera.ai/...") returning {”flagged”: True} → assert blocked. Without Lakera mock, integration tests require a real API key — expensive, skip with mocks.

Q3: Why is the keyword list hard-coded vs in Settings?

A: Two trade-offs. (a) Hard-coded: version-controlled, immutable, no config drift between environments. (b) Config-driven: SRE / compliance officer updates without PR; emergency jailbreak mitigation during incidents (e.g., new prompt-injection discovered in production). For the FCA posture, hard-coded makes sense because compliance reviews the list explicitly. For high-velocity product teams, config-driven is better. A middle-ground: hard-coded defaults in code, env-var override for incident response.

Top-to-Bottom Code Walkthrough (app/services/security_service.py)

SecurityService is the outbound safety gate. Every line of user text should pass through it before reaching the LLM, and every line of agent output before reaching the user or the database.

Imports

__init__

get_password_hash(plain) and verify_password(plain, hashed)

create_access_token(payload) -> str

decode_token(token) -> dict

sanitize_input(text) -> dict

This is the inbound pipeline. It runs in order, each step guarding against the next failure mode:

  1. Unicode bidi scrub: replaces LRE/RLE/PDF characters that could re-order display.
  2. Percent-decode to catch escape-encoded attacks.
  3. PII detection via Presidio: entities = self.analyzer.analyze(text=text, language="en"). Returns a list of detected entities (person, email, IBAN, etc.).
  4. Anonymisation: anonymised = self.anonymizer.anonymize(text=text, analyzer_results=entities). Returns text with &lt;PERSON>, &lt;EMAIL>, &lt;IBAN> placeholders.
  5. Jailbreak check (Lakera or fallback): see below.
  6. Returns {”safe_text”: anonymised.text, ”redacted_entities”: entities, ”is_injection”: bool}.

check_prompt_injection(text) -> tuple[bool, float]

Hybrid check:

  1. Fast regex pass: scan for "ignore previous", "ignore all", "DAN mode", "system prompt" — common jailbreak signatures.
  2. Optional Lakera call (if settings.lakera_guard_api_key): 
    async with httpx.AsyncClient() as client:
    r = await client.post(
        "https://api.lakera.ai/v1/prompt_injection",
        headers={"Authorization": f"Bearer {settings.lakera_guard_api_key}"},
        json={"input": text},
        timeout=2.0,
    )
    return r.json()["injection"] > 0.7, r.json()["score"]
  3. Fallback regex score if Lakera is unavailable — never block the request on the absence of a third-party API.

redact_pii(text) -> str

Convenience entry — runs only anonymize, skipping jailbreak detection. Used at logger-emit time, where PII must be stripped before persisting logs.

Common Pitfalls

Calling presidio at every log line is slow — its NLP models are heavy. Cache the AnalyzerEngine instance once at startup and reuse.

Lexicon-only jailbreak detection misses novel attacks. Always pair regex + the optional Lakera.

Logging the raw PII text before redact_pii — yes, the redactor is downstream of _prepare_log_message. The logger itself must NEVER log raw PII.

Real-World Interview Prep

Q1: What’s the trade-off between Presidio (local) and Lakera (cloud)?

A: Presidio is free and runs offline (data sovereignty is critical for FCA), but slower and easier to bypass. Lakera catches novel attacks in real-time but sends data to a third-party API. The hybrid is the right answer for regulated environments.

Q2: Why deprecated="auto" on the bcrypt context?

A: When you migrate to Argon2 (or any newer scheme), the same verify() call accepts both old (bcrypt) and new (Argon2) hashes. On the first verify() of an old hash against the new scheme, passlib transparently re-encodes and re-saves — zero-downtime migration.

Q3: How do you handle jailbreak detection when the LLM is the caller (not a human)?

A: Even LLM-generated prompts deserve jailbreak detection — a malicious tool result could contain a re-prompting attack. Apply check_prompt_injection to all text crossing into a prompt-template, regardless of source.

🛠️ Production Failure Playbook

🚨 Failure 1 — en_core_web_lg not loaded → OSError on first request

🚨 Failure 2 — Lakera quota exhausted → fallback to regex only

🚨 Failure 3 — PII not redacted before Lakera call (data sovereignty)

Advertisement