Data Leakage Prevention

# Using Microsoft Presidio for PII Detection
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def sanitize_input(text: str) -> str:
    """Detect and redact PII before sending to LLM"""
    
    # Analyze for PII entities
    results = analyzer.analyze(
        text=text,
        entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER", 
                  "CREDIT_CARD", "US_SSN", "IP_ADDRESS"],
        language="en"
    )
    
    # Anonymize detected entities
    anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
    
    return anonymized.text

# Example usage
user_input = "My SSN is 123-45-6789 and email is john@example.com"
safe_input = sanitize_input(user_input)
# Output: "My SSN is <US_SSN> and email is <EMAIL_ADDRESS>"

# Now safe to send to LLM
response = llm.generate(safe_input)

# Document-level access control in RAG
from typing import List, Dict

def retrieve_with_permissions(
    query: str, 
    user_id: str, 
    user_roles: List[str]
) -> List[Dict]:
    """Retrieve documents respecting user permissions"""
    
    # Get user's accessible document IDs
    allowed_doc_ids = get_user_document_access(user_id, user_roles)
    
    # Search with metadata filter
    results = vector_store.similarity_search(
        query=query,
        k=5,
        filter={
            "document_id": {"$in": allowed_doc_ids},
            "classification": {"$lte": get_user_clearance(user_id)}
        }
    )
    
    # Log access for audit trail
    log_document_access(user_id, [r.metadata["document_id"] for r in results])
    
    return results