🛡️Güvenlik ve Guardrails

Komtaş CoPilot'ta güvenlik, çok katmanlı bir yaklaşımla uygulanmaktadır. Bu bölüm, prompt injection savunmasından OWASP LLM Top 10'a, halüsinasyon kontrolünden PII korumasına kadar tüm güvenlik boyutlarını kapsamaktadır.

5.1 Prompt Injection Savunması

Direkt Prompt Injection Tipleri

Dolaylı (RAG) Prompt Injection

2025 araştırmalarına göre, sadece 5 zararlı belge %90 oranında AI yanıtlarını manipüle edebilmektedir. RAG sistemleri bu açıdan kritik risk taşır.

# RAG güvenlik katmanı — doküman tarama
from lakera_guard import LakeraGuard

guard = LakeraGuard(api_key=os.getenv("LAKERA_API_KEY"))

async def safe_rag_retrieval(query: str, chunks: list[str]) -> list[str]:
    """RAG dokümanlarını injection için tara"""
    safe_chunks = []
    
    for chunk in chunks:
        # Her chunk'ı injection için kontrol et
        result = await guard.check_input(
            user_input=chunk,
            external_inputs=[],
            check_types=["prompt_injection", "jailbreak"]
        )
        
        if not result.flagged:
            safe_chunks.append(chunk)
        else:
            # Şüpheli chunk'ı logla ama kullanma
            await security_log.warn(
                f"Şüpheli RAG chunk tespit edildi: {result.categories}",
                chunk_preview=chunk[:100]
            )
    
    return safe_chunks

# Canary token ile sistem prompt sızıntısı tespiti
CANARY_TOKEN = "KOMTAS-CANARY-XK7P2"

SYSTEM_PROMPT = f"""Sen Komtaş asistanısın...
{CANARY_TOKEN}
[Bu token sistem dışına çıkmamalı]"""

def detect_prompt_leak(response: str) -> bool:
    """Sistem prompt sızıntısını tespit et"""
    return CANARY_TOKEN in response

5.2 Güvenlik Araçları

import httpx
import os

LAKERA_API_KEY = os.getenv("LAKERA_GUARD_API_KEY")
LAKERA_BASE_URL = "https://api.lakera.ai/v1"

async def lakera_check(user_input: str, context: list[str] = None) -> dict:
    """Lakera Guard API çağrısı"""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{LAKERA_BASE_URL}/guard",
            headers={"Authorization": f"Bearer {LAKERA_API_KEY}"},
            json={
                "input": user_input,
                "context": context or [],
                "policies": ["prompt_injection", "jailbreak", "pii", "moderation"]
            },
            timeout=2.0  # 2 saniye timeout — latency kritik
        )
    
    result = response.json()
    
    if result["flagged"]:
        raise SecurityException(
            f"Güvenlik ihlali tespit edildi: {result['categories']}",
            categories=result["categories"],
            confidence=result["confidence"]
        )
    
    return result

# FastAPI middleware
@app.middleware("http")
async def security_middleware(request: Request, call_next):
    if request.method == "POST" and "/chat" in request.url.path:
        body = await request.body()
        data = json.loads(body)
        user_message = data.get("message", "")
        
        try:
            await lakera_check(user_message)
        except SecurityException as e:
            return JSONResponse(
                status_code=400,
                content={"error": "Güvenlik politikası ihlali", "detail": str(e)}
            )
    
    return await call_next(request)

from llm_guard import scan_prompt, scan_output
from llm_guard.input_scanners import (
    Anonymize, BanTopics, Code, PromptInjection, Secrets, Toxicity
)
from llm_guard.output_scanners import (
    Deanonymize, NoRefusal, FactualConsistency, Relevance, Sensitive
)
from llm_guard.vault import Vault

# Vault: PII'yı güvenli sakla ve geri yükle
vault = Vault()

# Input scanner konfigürasyonu
input_scanners = [
    Anonymize(
        vault=vault,
        entity_types=["PERSON", "EMAIL", "PHONE", "TC_KIMLIK_NO", "CREDIT_CARD"],
        preamble="Türkçe içerik:"
    ),
    BanTopics(
        topics=["siyasi görüş", "rakip firmalar"],
        threshold=0.75
    ),
    PromptInjection(
        threshold=0.7,
        model="laiyer-ai/prompt-injection-v1"
    ),
    Secrets(redact_mode="partial"),
    Toxicity(threshold=0.7)
]

# Output scanner konfigürasyonu
output_scanners = [
    Deanonymize(vault=vault),  # PII geri yükle (yetkili kullanıcılar için)
    FactualConsistency(
        model="vectara/hallucination_evaluation_model",
        minimum_score=0.6
    ),
    NoRefusal(threshold=0.5),
    Relevant(threshold=0.5)
]

def secure_llm_call(user_message: str, system_prompt: str) -> str:
    # Input tarama
    sanitized_prompt, is_valid, risk_score = scan_prompt(
        scanners=input_scanners,
        prompt=user_message,
        fail_fast=True
    )
    
    if not is_valid:
        return "Güvenlik politikası: Bu istek işlenemiyor."
    
    # LLM çağrısı
    raw_output = call_llm(sanitized_prompt, system_prompt)
    
    # Output tarama
    sanitized_output, is_valid, risk_score = scan_output(
        scanners=output_scanners,
        prompt=sanitized_prompt,
        output=raw_output,
        fail_fast=False
    )
    
    return sanitized_output

# komtas_rails.co — Colang konfigürasyonu
define user ask about competitors
  "Rakip firma hakkında bilgi ver"
  "X şirketi ile karşılaştır"

define bot decline competitor questions
  "Üzgünüm, rakip firmalar hakkında bilgi veremiyorum. 
   Komtaş ürünleri hakkında yardımcı olmaktan memnuniyet duyarım."

define flow competitor question protection
  user ask about competitors
  bot decline competitor questions

define user ask for harmful content
  "nasıl hack yapılır"
  "zararlı kod yaz"

define bot refuse harmful request
  "Bu tür içerik üretemem. Güvenli ve etik kullanım politikamıza aykırıdır."

define flow safety guardrail
  user ask for harmful content
  bot refuse harmful request

# Python entegrasyonu
from nemoguardrails import RailsConfig, LLMRails

config = RailsConfig.from_path("./komtas_rails/")
rails = LLMRails(config)

async def guarded_chat(user_message: str) -> str:
    """NeMo Guardrails ile korumalı chat"""
    response = await rails.generate_async(
        messages=[{"role": "user", "content": user_message}]
    )
    return response["content"]

5.3 OWASP LLM Top 10 (2025)

5.4 Halüsinasyon Kontrolü

Halüsinasyon, LLM'lerin olgusal olmayan veya bağlamdan sapan içerik üretmesidir. İki ana kategorisi vardır:

• İçsel Halüsinasyon: Modelin kendi eğitim verisiyle çelişen çıktı üretmesi
• Dışsal Halüsinasyon: Sağlanan bağlamla çelişen çıktı (RAG'da kritik)
• Olgusal Halüsinasyon: Gerçek dünyadaki olgularla çelişen ifadeler
• Sadakat Halüsinasyonu: Kaynakta olmayan bilgileri kaynak gibi sunma

from sentence_transformers import CrossEncoder
import anthropic

# NLI tabanlı halüsinasyon tespiti
nli_model = CrossEncoder("cross-encoder/nli-deberta-v3-base")

def detect_hallucination_nli(claim: str, context_chunks: list[str]) -> float:
    """
    NLI modeli ile claim'in context tarafından desteklenip desteklenmediğini ölç.
    Returns: 0 (tamamen çelişki) - 1 (tam destek)
    """
    scores = []
    for chunk in context_chunks:
        # NLI: entailment, neutral, contradiction
        nli_scores = nli_model.predict([(chunk, claim)])
        entailment_score = nli_scores[0][2]  # entailment sınıfı
        scores.append(entailment_score)
    
    return max(scores) if scores else 0.0

# LLM-as-Judge ile halüsinasyon tespiti
def llm_judge_faithfulness(answer: str, contexts: list[str]) -> dict:
    """Claude ile faithfulness kontrolü"""
    client = anthropic.Anthropic()
    
    context_text = "\n\n".join([f"[KAYNAK {i+1}]\n{c}" for i, c in enumerate(contexts)])
    
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="""Sen bir kalite kontrol ajansın. Verilen cevabın, 
        kaynak metinlerde desteklenip desteklenmediğini değerlendir.
        Her ifade için "destekleniyor" veya "desteklenmiyor" belirt.
        JSON formatında yanıt ver: {"score": 0-1, "unsupported_claims": [...]}""",
        messages=[{
            "role": "user",
            "content": f"KAYNAKLAR:\n{context_text}\n\nCEVAP:\n{answer}"
        }]
    )
    
    return json.loads(response.content[0].text)

# Çoklu model konsensüsü (Ensemble Verification)
async def ensemble_verification(claim: str, contexts: list[str]) -> float:
    """
    Birden fazla yöntemle halüsinasyon tespiti yaparak ortalama al
    """
    scores = await asyncio.gather(
        asyncio.to_thread(detect_hallucination_nli, claim, contexts),
        asyncio.to_thread(llm_judge_faithfulness_score, claim, contexts),
        asyncio.to_thread(cross_reference_check, claim, contexts)
    )
    
    # Ağırlıklı ortalama: NLI 0.3, LLM-Judge 0.5, Cross-ref 0.2
    weights = [0.3, 0.5, 0.2]
    return sum(s * w for s, w in zip(scores, weights))

5.5 PII Koruması ve Veri Gizliliği

KVKK'ya Özgü PII Tipleri

from presidio_analyzer import AnalyzerEngine, RecognizerRegistry, PatternRecognizer
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
import re

# Türkçe NLP engine
nlp_config = {"nlp_engine_name": "spacy", "models": [{"lang_code": "tr", "model_name": "tr_core_news_lg"}]}
provider = NlpEngineProvider(nlp_configuration=nlp_config)

# TC Kimlik No özel tanıyıcı
class TcKimlikNoRecognizer(PatternRecognizer):
    def __init__(self):
        patterns = [
            Pattern("TC_KIMLIK_HIGH", r"\b[1-9][0-9]{10}\b", 0.85),
        ]
        super().__init__(supported_entity="TR_TC_KIMLIK_NO", patterns=patterns)
    
    def validate_result(self, pattern_text: str) -> bool:
        """TC Kimlik No algoritma doğrulaması"""
        digits = [int(d) for d in pattern_text]
        if len(digits) != 11 or digits[0] == 0:
            return False
        
        # Standart TC algoritma kontrolü
        d10 = (7 * sum(digits[:9:2]) - sum(digits[1:8:2])) % 10
        d11 = sum(digits[:10]) % 10
        return digits[9] == d10 and digits[10] == d11

# Analyzer ve anonymizer kur
registry = RecognizerRegistry()
registry.add_recognizer(TcKimlikNoRecognizer())
# Vergi Kimlik No
registry.add_recognizer(PatternRecognizer(
    supported_entity="TR_VERGI_KIMLIK_NO",
    patterns=[Pattern("HIGH", r"\b[0-9]{10}\b", 0.7)],
    context=["vergi", "vkn", "vergi kimlik"]
))

analyzer = AnalyzerEngine(
    nlp_engine=provider.create_engine(),
    registry=registry
)
anonymizer = AnonymizerEngine()

def anonymize_for_llm(text: str) -> tuple[str, dict]:
    """LLM'e göndermeden önce PII'yı maskele"""
    results = analyzer.analyze(
        text=text,
        language="tr",
        entities=["TR_TC_KIMLIK_NO", "TR_VERGI_KIMLIK_NO", "PERSON", 
                  "EMAIL_ADDRESS", "PHONE_NUMBER", "IBAN_CODE"]
    )
    
    if not results:
        return text, {}
    
    operators = {
        "PERSON": OperatorConfig("replace", {"new_value": "[KIŞI]"}),
        "EMAIL_ADDRESS": OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 5, "from_end": False}),
        "TR_TC_KIMLIK_NO": OperatorConfig("replace", {"new_value": "[TC-KIMLIK]"}),
        "PHONE_NUMBER": OperatorConfig("mask", {"masking_char": "*", "chars_to_mask": 6, "from_end": True}),
    }
    
    anonymized = anonymizer.anonymize(
        text=text,
        analyzer_results=results,
        operators=operators
    )
    
    return anonymized.text, {"pii_detected": len(results), "entities": [r.entity_type for r in results]}

5.6 5 Katmanlı Güvenlik Mimarisi