RAG Security Engineering: Preventing Data Leakage and Prompt Injection in Production RAG Systems

# Secure API Key Rotation Script for HolySheep AI Migration
import requests
import json
from datetime import datetime
from cryptography.fernet import Fernet

class HolySheepKeyRotation:
    """
    Secure API key rotation for HolySheep AI endpoints.
    Implements key versioning and automatic rollback on failure.
    """
    
    def __init__(self, base_url: str, admin_key: str):
        self.base_url = base_url
        self.admin_key = admin_key
        self.key_version = 1
        self.encryption_key = Fernet.generate_key()
        self.fernet = Fernet(self.encryption_key)
        
    def create_new_key(self, scopes: list = None) -> dict:
        """
        Generate a new API key with specified scopes.
        Scopes follow principle of least privilege.
        """
        if scopes is None:
            scopes = ["chat:write", "embeddings:read", "files:upload"]
        
        headers = {
            "Authorization": f"Bearer {self.admin_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "name": f"production-key-v{self.key_version}",
            "scopes": scopes,
            "rate_limit": {
                "requests_per_minute": 1000,
                "tokens_per_minute": 150000
            },
            "allowed_ips": [  # IP whitelisting for additional security
                "203.0.113.0/24",  # Production subnet
                "198.51.100.0/24"  # DR subnet
            ],
            "expires_at": datetime.now().timestamp() + (90 * 24 * 60 * 60)  # 90 days
        }
        
        response = requests.post(
            f"{self.base_url}/api-keys",
            headers=headers,
            json=payload
        )
        
        if response.status_code == 201:
            result = response.json()
            # Encrypt key at rest before storing
            result["encrypted_key"] = self.fernet.encrypt(
                result["secret"].encode()
            ).decode()
            return result
        
        raise Exception(f"Key creation failed: {response.text}")
    
    def revoke_old_key(self, key_id: str) -> bool:
        """
        Revoke a previous API key after successful migration.
        Immediate revocation ensures no downtime windows.
        """
        headers = {
            "Authorization": f"Bearer {self.admin_key}"
        }
        
        response = requests.delete(
            f"{self.base_url}/api-keys/{key_id}",
            headers=headers
        )
        
        return response.status_code == 204
    
    def verify_key_permissions(self, new_key: str) -> dict:
        """
        Verify new key has correct permissions before full migration.
        Tests all required scopes in a sandbox environment.
        """
        test_headers = {
            "Authorization": f"Bearer {new_key}"
        }
        
        results = {
            "chat:write": False,
            "embeddings:read": False,
            "files:upload": False
        }
        
        # Test chat completion
        try:
            chat_response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=test_headers,
                json={
                    "model": "deepseek-v3",
                    "messages": [{"role": "user", "content": "test"}],
                    "max_tokens": 10
                },
                timeout=10
            )
            results["chat:write"] = chat_response.status_code == 200
        except:
            pass
        
        # Test embeddings
        try:
            embed_response = requests.post(
                f"{self.base_url}/embeddings",
                headers=test_headers,
                json={
                    "model": "text-embedding-3-small",
                    "input": "test"
                },
                timeout=10
            )
            results["embeddings:read"] = embed_response.status_code == 200
        except:
            pass
        
        return results


Execution example
def execute_migration():
    """
    Complete migration workflow with verification checkpoints.
    """
    holy_sheep = HolySheepKeyRotation(
        base_url="https://api.holysheep.ai/v1",
        admin_key=os.environ.get("HOLYSHEEP_ADMIN_KEY")
    )
    
    # Step 1: Create new key with production scopes
    print("Creating new production API key...")
    new_key_data = holy_sheep.create_new_key(scopes=[
        "chat:write",
        "embeddings:read",
        "files:upload"
    ])
    
    # Step 2: Verify permissions before use
    print("Verifying key permissions...")
    permissions = holy_sheep.verify_key_permissions(new_key_data["secret"])
    
    if not all(permissions.values()):
        print(f"Permission check failed: {permissions}")
        raise Exception("Key verification failed - aborting migration")
    
    # Step 3: Store encrypted key securely
    print("Storing encrypted key...")
    store_secure_key(
        key_id=new_key_data["id"],
        encrypted_key=new_key_data["encrypted_key"]
    )
    
    # Step 4: Update application configuration
    print("Updating application configuration...")
    update_app_config(new_key_data["secret"])
    
    # Step 5: Gradual traffic migration via feature flags
    print("Starting canary traffic migration...")
    increment_canary_percentage(from_percent=5, to_percent=100, step=5)
    
    print("Migration complete!")
    return new_key_data

30-Day Post-Launch Metrics

Understanding RAG Security Threats

Prompt Injection: The Invisible Attacker

Data Leakage Vectors in RAG Systems

1. Unsanitized Retrieval: Vector search returns sensitive documents without proper access control checks
2. Excessive Context Inclusion: Including too much retrieved context increases exposure surface
3. Training Data Contamination: Model inadvertently memorizes and reveals sensitive training data
4. Log Leakage: API responses, including retrieved context, are logged without sanitization
5. Incomplete Output Filtering: Generated responses contain verbatim excerpts from restricted documents

HolySheep AI Security Architecture

1. Semantic Injection Detection

• Pattern-based injection detection with 99.2% precision
• Semantic anomaly scoring for novel attack vectors
• Automatic redaction of detected injection attempts
• Real-time alerting to security operations teams

2. Isolated Context Processing

3. PII Detection and Filtering

• Email addresses, phone numbers, and national IDs
• Payment card numbers (with automatic masking)
• API keys and authentication tokens
• Medical record numbers and insurance IDs

Implementing Defense in Depth: A Production Framework

Layer 1: Input Validation and Sanitization

# Comprehensive RAG Security Framework
Defense in Depth: Input → Retrieval → Generation → Output

import re
import hashlib
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum

class ThreatLevel(Enum):
    SAFE = "safe"
    SUSPICIOUS = "suspicious"
    DANGEROUS = "dangerous"
    BLOCKED = "blocked"

@dataclass
class SecurityResult:
    threat_level: ThreatLevel
    sanitized_content: str
    detected_patterns: List[str]
    confidence_score: float

class RAGInputValidator:
    """
    Multi-layer input validation for RAG systems.
    Combines pattern matching, semantic analysis, and behavioral detection.
    """
    
    # Injection patterns with severity weighting
    INJECTION_PATTERNS = {
        # Critical severity (immediate block)
        "critical": [
            (r"ignore\s+(all\s+)?(previous|prior|above)\s+instructions?", 0.99),
            (r"(forget|disregard)\s+everything", 0.98),
            (r"you\s+are\s+now\s+", 0.97),
            (r"new\s+system\s+instruction", 0.96),
            (r"<\s*script", 0.99),
            (r" SecurityResult:
        """
        Comprehensive validation of user input.
        Returns sanitized content and threat assessment.
        """
        threat_score = 0.0
        detected_patterns = []
        sanitized = user_input
        
        # Rate limiting check
        if not self._check_rate_limit(user_id):
            return SecurityResult(
                threat_level=ThreatLevel.BLOCKED,
                sanitized_content="",
                detected_patterns=["RATE_LIMIT_EXCEEDED"],
                confidence_score=1.0
            )
        
        # Pattern-based detection
        for severity, patterns in self.INJECTION_PATTERNS.items():
            for pattern, weight in patterns:
                matches = re.findall(pattern, sanitized, re.IGNORECASE)
                if matches:
                    detected_patterns.extend(matches)
                    threat_score = max(threat_score, weight)
                    sanitized = re.sub(pattern, "[RESTRICTED_CONTENT]", 
                                      sanitized, flags=re.IGNORECASE)
        
        # Behavioral anomaly detection
        behavioral_score = self._analyze_behavior(user_id, sanitized)
        threat_score = max(threat_score, behavioral_score)
        
        # Semantic analysis via HolySheep
        if self.enable_semantic_analysis and threat_score < 0.5:
            semantic_result = self._semantic_check(sanitized)
            if semantic_result:
                threat_score = max(threat_score, semantic_result["score"])
                detected_patterns.extend(semantic_result
Related Resources
📚 AI API Tutorials
💰 View Pricing
📖 Developer Docs
🚀 Sign Up Free
Related Articles
Meta-Prompting: Let AI Optimize Its Own Prompts (Complete Be
AI Agent Exception Recovery: Smart Retry Logic and Human-in-
Agent Evaluation Framework: Building Automated Testing and Q