Real-Time PII Prevencija: AI Agenti od Curenja Podataka

Real-Time PII Prevencija: AI Agenti od Curenja Podataka

Problem Curenja Podataka AI

AI agenti mogu slučajno curiti PII:

Korisnik: "Koji je imejl John Doe-a?"
AI Odgovor: "John Doe-ov imejl je john.doe@company.com" ← CURENJE!

Rešenje: Real-Time Detekcija

from presidio_analyzer import AnalyzerEngine

def safe_ai_response(user_query: str, ai_response: str) -> str:
    """
    Analizi AI odgovor pre nego što ga prikaži korisniku
    """
    analyzer = AnalyzerEngine()
    
    # Detektuj PII u AI odgovoru
    pii_results = analyzer.analyze(
        text=ai_response,
        language="sr",
        entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER"]
    )
    
    if pii_results:
        # Blokiraj odgovor ako sadrži PII
        return "❌ Odgovor sadrži osjetljive podatke. Pokušaj sa specifičnijom pretragom."
    
    # Siguran odgovor
    return ai_response

AI Agent Pristup

class PIIAwareAIAgent:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.memory = {}  # Token mapiranje
    
    def generate_safe_response(self, query: str) -> str:
        # 1. Anonimizuj upit pre nego što ga pošalješ AI-u
        anon_query = self.anonymize_query(query)
        
        # 2. AI generiše odgovor sa token-ima
        ai_response = call_ai_api(anon_query)
        
        # 3. Deanonimizuj odgovor
        safe_response = self.deanonymize_response(ai_response)
        
        # 4. Validiraj da nema novog curenja
        if self.has_pii(safe_response):
            return "❌ Bezbednosni problem."
        
        return safe_response
    
    def anonymize_query(self, text: str) -> str:
        analyzer_results = self.analyzer.analyze(text, language="sr")
        
        anon_text = text
        for result in sorted(analyzer_results, key=lambda x: x.start, reverse=True):
            token = f"[{result.entity_type}_{len(self.memory)}]"
            original = text[result.start:result.end]
            self.memory[token] = original
            
            anon_text = anon_text[:result.start] + token + anon_text[result.end:]
        
        return anon_text
    
    def deanonymize_response(self, text: str) -> str:
        for token, original in self.memory.items():
            text = text.replace(token, original)
        return text
    
    def has_pii(self, text: str) -> bool:
        results = self.analyzer.analyze(text, language="sr")
        return len(results) > 0

Proactive Curenja

def detect_pii_injection_attempts(user_query: str) -> bool:
    """
    Detektuj pokušaje da se AI prosledi PII kroz prompt injection
    """
    suspicious_patterns = [
        r"ignore previous instructions",
        r"show me the raw data",
        r"what is .*email|phone|ssn",
    ]
    
    for pattern in suspicious_patterns:
        if re.search(pattern, user_query, re.IGNORECASE):
            return True
    
    return False

Rezultati

Zaključak

Real-time PII prevencija sa AI agentima sprečava curenja podataka kroz anonimizaciju/deanonimizaciju + post-response validaciju.