Module 3.7: Compulsory Security — Moderation & Defensive Schemas

Curriculum Alignment: docs/plan/03_phase3_modular_chaining_lcel.md
Topic Scope: Content Moderation Middleware, Schema Enforcement for Security, PII Filtering, Preventing Unauthorized Tool Calls
Level: Advanced AI Engineering / Architecture

---

1. Security in Composed Pipelines: Beyond Syntax

In multi-stage LCEL pipelines, security cannot be treated as an afterthought. Unsanitized user inputs can trigger toxicity, jailbreaks, or prompt injection at Step 1, which then propagates through parallel branches and compromises downstream databases or tools.

Phase 3 mandates two core defensive primitives:

1. Content Moderation Middleware: An automated gate intercepting both incoming inputs and outgoing completions to block toxic, hateful, violent, or self-harm content before it enters the chain or reaches end users.
2. Schema Enforcement for Security: Using Pydantic validators not merely for structure, but as cryptographic and regex firewalls that reject PII (Personally Identifiable Information), sensitive keys, or unauthorized system actions.

---

2. Content Moderation Middleware in LCEL

Using OpenAI's free Moderation API or dedicated classification models, we insert a moderation check as a first-class RunnableLambda at the boundary of our LCEL chain:

---

3. Schema Enforcement: Pydantic as a Security Firewall

In production, models can hallucinate dangerous outputs: emitting API keys found in retrieved context, outputting real user emails/SSNs, or generating destructive tool actions (e.g. DROP TABLE).

By embedding Pydantic field validators and root validators, we enforce strict defensive constraints directly on model outputs:

"""
Defensive Security Middleware & Schema Enforcement
Demonstrates automated content moderation and PII blocking in LCEL.
"""
import re
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
from langchain_core.exceptions import OutputParserException
from langchain_core.runnables import RunnableLambda
from openai import OpenAI

class SecurityViolationException(Exception):
    """Raised when moderation middleware flags content."""
    pass

def check_content_moderation(text: str) -> str:
    """Evaluates text against the OpenAI Moderation endpoint."""
    client = OpenAI()
    response = client.moderations.create(input=text)
    result = response.results[0]
    
    if result.flagged:
        flagged_categories = [
            cat for cat, flagged in result.categories.model_dump().items() if flagged
        ]
        raise SecurityViolationException(
            f"Security Violation: Content blocked due to policy violations: {flagged_categories}"
        )
    return text

# LCEL Inbound Moderation Middleware
inbound_moderation = RunnableLambda(
    lambda x: {**x, "user_input": check_content_moderation(x["user_input"])}
)

# -------------------------------------------------------------
# Defensive Schema: Sanitizes PII and Blocks Destructive Actions
# -------------------------------------------------------------
class SafeExtractedProfile(BaseModel):
    """Extracts customer profile while strictly forbidding PII leakage."""
    model_config = ConfigDict(frozen=True, extra="forbid")
    
    username: str = Field(min_length=3, max_length=50)
    organization_role: str = Field(description="Role within enterprise.")
    technical_skills: list[str] = Field(description="List of verified skills.")
    public_summary: str = Field(description="Clean, non-sensitive summary.")

    @field_validator("public_summary")
    @classmethod
    def validate_no_pii(cls, value: str) -> str:
        """Enforces zero PII leakage (Emails, SSN, Credit Cards, API Keys)."""
        # Email pattern
        if re.search(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+", value):
            raise ValueError("Security Policy Violation: Output contains unredacted email address.")
        
        # SSN pattern
        if re.search(r"\b\d{3}-\d{2}-\d{4}\b", value):
            raise ValueError("Security Policy Violation: Output contains Social Security Number.")
        
        # Secret/Key pattern (e.g. sk-..., bearer tokens)
        if re.search(r"(sk-[a-zA-Z0-9]{20,}|bearer\s+[a-zA-Z0-9_\-\.]+)", value, re.IGNORECASE):
            raise ValueError("Security Policy Violation: Output contains raw credential token.")
            
        return value

Full Composed Secure Pipeline:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
structured_model = model.with_structured_output(SafeExtractedProfile)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract customer profile from raw text. Never output personal email addresses or credentials."),
    ("human", "Raw Text:\n<raw_data>\n{user_input}\n</raw_data>")
])

# Secure Chain: Input Moderation -> Prompt -> Structured Extraction with PII Defense
secure_extraction_pipeline = (
    inbound_moderation 
    | prompt 
    | structured_model
)

Conceptual Mindmap: Compulsory Security in LCEL

---

4. Curated Reading & Canonical References

Resource	Canonical Reference & Link	Specific Focus Areas
Primary Curriculum Book	AI Agents and Applications (Google Drive)	Part 4, Chapter 14, Section 14.2 (pp. 364–373): Implementing guardrails, filtering unauthorized queries, and defensive agent design.
OpenAI Security API	OpenAI Moderation Guide	Classifying input/output text against harmful content policies.
OWASP GenAI Standard	OWASP Top 10 for LLMs — LLM06: Sensitive Information Disclosure	Defenses against training data leakage and accidental PII emission.

---

5. Active Recall (Module 3.7 Flashcards)

Content ModerationClick or press Space to flip ↺

Why should Content Moderation Middleware be placed at the input boundary of an LCEL chain?

Question / Scenario

Content Moderation • AnswerClick to flip back ↻

Running moderation early prevents adversarial or toxic prompts from consuming downstream LLM tokens, executing expensive parallel branches, or compromising internal tools. Failing early protects both system compute budgets and backend integrity.

💡 Architect Takeaway: Always evaluate safety at the entry boundary before executing model inference.

Technical Explanation

Defensive SchemasClick or press Space to flip ↺

How can Pydantic validators protect against Sensitive Information Disclosure (LLM06)?

Question / Scenario

Defensive Schemas • AnswerClick to flip back ↻

By implementing @field_validator rules that inspect string outputs with regex patterns for emails, SSNs, credit cards, or API keys, raising a validation error and halting delivery before sensitive data reaches external users or databases.

💡 Architect Takeaway: Pydantic models serve as deterministic security firewalls against model leakage.

Technical Explanation

---

6. Hands-on Engineering Exercises

Exercise 3.7: The Secure Schema Validator Drill

• Goal: Build the foundation for your Secure Schema Validator deliverable.
• Task: Create an input test case containing a simulated enterprise leak: "User Bob can be reached at bob.private@corp.com with secret sk-live998811223344".
• Requirements:
  • Run the input through the secure_extraction_pipeline.
  • Prove that the Pydantic validator intercepts the email/key pattern, raises a ValueError, and prevents the leaked data from being emitted.