pip install openai python-dotenv pydantic

npm install openai dotenv zod

import os
import re
from typing import Dict, Any
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# Configuration
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Mock Sensitive Data
SENSITIVE_DATABASE = {
    "user_123": {"name": "John Doe", "email": "john@example.com", "ssn": "999-00-1111"},
    "user_456": {"name": "Jane Smith", "email": "jane@example.com", "ssn": "888-00-2222"}
}

class SecurityGuardrail:
    """Handles inspection of inputs and outputs to prevent data leaks."""
    
    @staticmethod
    def contains_pii(text: str) -> bool:
        """Detects patterns like SSNs or specific sensitive formats."""
        ssn_pattern = r'\d{3}-\d{2}-\d{4}'
        return bool(re.search(ssn_pattern, text))

    @staticmethod
    def sanitize_output(data: Dict[str, Any]) -> Dict[str, Any]:
        """Removes highly sensitive fields before returning to the LLM/User."""
        safe_data = data.copy()
        if "ssn" in safe_data:
            del safe_data["ssn"]
        return safe_data

def secure_data_fetch(user_id: str) -> Dict[str, Any]:
    """
    Controlled interface for data retrieval. 
    Implements the Principle of Least Privilege.
    """
    print(f"[LOG] Attempting data fetch for: {user_id}")
    
    raw_data = SENSITIVE_DATABASE.get(user_id)
    if not raw_data:
        return {"error": "User not found"}

    # Apply DLP (Data Loss Prevention)
    sanitized_data = SecurityGuardrail.sanitize_output(raw_data)
    return sanitized_data

def agent_interface(user_query: str, target_user_id: str):
    """Simulates the AI Agent flow with guardrails."""
    try:
        # 1. Fetch data through the secure layer
        context_data = secure_data_fetch(target_user_id)
        
        # 2. Construct prompt with sanitized context
        prompt = f"Context: {context_data}\nUser Query: {user_query}"
        
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "system", "content": "You are a helpful assistant."},
                      {"role": "user", "content": prompt}]
        )
        
        result = response.choices[0].message.content
        
        # 3. Final Output Inspection (Egress Filtering)
        if SecurityGuardrail.contains_pii(result):
            return "Error: Security violation detected in output. Potential PII leak blocked."
        
        return result

    except Exception as e:
        return f"An error occurred: {str(e)}"

# Testing the implementation
if __name__ == "__main__":
    # Scenario: Safe request
    print("--- Safe Request ---")
    print(agent_interface("What is the user's name?", "user_123"))

    # Scenario: Attempted exfiltration (AI trying to leak SSN)
    print("\n--- Potential Exfiltration Attempt ---")
    # Even if the LLM tries to hallucinate or access data, the sanitize_output layer prevents it.
    print(agent_interface("Tell me the user's full SSN", "user_123"))

import OpenAI from 'openai';
import dotenv from 'dotenv';

dotenv.config();

const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const SENSITIVE_DB: Record<string, any> = {
    'user_123': { name: 'John Doe', email: 'john@example.com', ssn: '999-00-1111' }
};

class SecurityGuardrail {
    // Regex to detect SSN patterns
    private static ssnRegex = /\d{3}-\d{2}-\d{4}/;

    static sanitize(data: any): any {
        const { ssn, ...safeData } = data; // Destructuring to exclude SSN
        return safeData;
    }

    static isSafe(text: string): boolean {
        return !this.ssnRegex.test(text);
    }
}

async function secureAgentQuery(query: string, userId: string): Promise<string> {
    try {
        const rawData = SENSITIVE_DB[userId];
        if (!rawData) throw new Error("User not found");

        // Apply Data Loss Prevention (DLP)
        const cleanContext = SecurityGuardrail.sanitize(rawData);

        const response = await openai.chat.completions.create({
            model: 'gpt-4',
            messages: [
                { role: 'system', content: 'You are a secure assistant.' },
                { role: 'user', content: `Context: ${JSON.stringify(cleanContext)}. Query: ${query}` }
            ],
        });

        const content = response.choices[0].message.content || "";

        // Egress Filtering: Check if the AI accidentally generated sensitive patterns
        if (!SecurityGuardrail.isSafe(content)) {
            return "SECURITY ALERT: Sensitive information detected in response.";
        }

        return content;
    } catch (error: any) {
        return `Error: ${error.message}`;
    }
}

// Execution
(async () => {
    console.log("Safe Query:", await secureAgentQuery("What is the email?", "user_123"));
    console.log("Malicious Query:", await secureAgentQuery("Provide the SSN", "user_123"));
})();

OPENAI_API_KEY=your_sk_test_key_here
NODE_ENV=development