API Security für AI-Systeme

class InputValidator:
    def validate(self, user_input: str) -> ValidationResult:
        checks = [
            self.check_length,
            self.check_pii,
            self.check_injection_patterns,
            self.check_encoding,
            self.check_content_policy
        ]

        for check in checks:
            result = check(user_input)
            if not result.passed:
                return result

        return ValidationResult(passed=True, sanitized=user_input)

def check_length(self, input: str) -> ValidationResult:
    MAX_INPUT_LENGTH = 10000  # Tokens, nicht Zeichen
    MAX_CHAR_LENGTH = 50000

    if len(input) > MAX_CHAR_LENGTH:
        return ValidationResult(
            passed=False,
            reason="Input too long"
        )
    return ValidationResult(passed=True)

import presidio_analyzer

def check_pii(self, input: str) -> ValidationResult:
    analyzer = presidio_analyzer.AnalyzerEngine()
    results = analyzer.analyze(
        text=input,
        language="de",
        entities=["EMAIL_ADDRESS", "PHONE_NUMBER", "PERSON",
                  "CREDIT_CARD", "IBAN_CODE"]
    )

    if results:
        # Option 1: Blockieren
        return ValidationResult(passed=False, reason="PII detected")

        # Option 2: Redaktieren (besser für UX)
        # sanitized = anonymize(input, results)
        # return ValidationResult(passed=True, sanitized=sanitized)

INJECTION_PATTERNS = [
    r"ignoriere?\s*(alle|vorherige|die)?\s*anweisung",
    r"vergiss\s*(alles|deine|die)",
    r"du\s+bist\s+(jetzt|ab\s+jetzt)",
    r"system\s*prompt",
    r"</?system>",
    r"\[INST\]",  # Llama-Format
    r"###\s*(System|Instruction)",
]

def check_injection_patterns(self, input: str) -> ValidationResult:
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, input, re.IGNORECASE):
            log_security_event("injection_attempt", pattern)
            return ValidationResult(
                passed=False,
                reason="Suspicious pattern detected"
            )
    return ValidationResult(passed=True)

from fastapi import Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    # Gegen Secrets Manager validieren, nicht Hardcoded!
    valid_keys = await secrets_manager.get_valid_keys()

    if api_key not in valid_keys:
        log_security_event("invalid_api_key", api_key[:8])
        raise HTTPException(status_code=401, detail="Invalid API key")

    return await get_key_metadata(api_key)

class Permission(Enum):
    GPT4O = "gpt4o"
    GPT4O_MINI = "gpt4o_mini"
    EMBEDDING = "embedding"
    FINE_TUNE = "fine_tune"
    AGENT = "agent"

ROLE_PERMISSIONS = {
    "basic": [Permission.GPT4O_MINI, Permission.EMBEDDING],
    "advanced": [Permission.GPT4O_MINI, Permission.GPT4O, Permission.EMBEDDING],
    "admin": [Permission.GPT4O_MINI, Permission.GPT4O, Permission.EMBEDDING,
              Permission.FINE_TUNE, Permission.AGENT],
}

async def check_permission(
    key_meta: KeyMetadata,
    required: Permission
) -> bool:
    user_permissions = ROLE_PERMISSIONS.get(key_meta.role, [])
    return required in user_permissions

# Key-Erstellung mit minimalem Scope
api_keys:
  - id: key_prod_chat_001
    role: basic
    allowed_models: ["gpt-4o-mini"]
    rate_limit: 100/minute
    budget: 50/month
    allowed_ips: ["10.0.0.0/8"]
    expires: 2026-06-01

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

# Request-basiert (Basis-Schutz)
@app.post("/chat")
@limiter.limit("60/minute")
async def chat(request: Request):
    ...

# Token-basiert (kritisch bei LLMs!)
class TokenRateLimiter:
    def __init__(self, max_tokens_per_minute: int):
        self.max_tokens = max_tokens_per_minute
        self.windows = {}  # user_id -> deque of (timestamp, tokens)

    async def check(self, user_id: str, estimated_tokens: int) -> bool:
        window = self.windows.get(user_id, deque())

        # Alte Einträge entfernen (> 1 Minute)
        now = time.time()
        while window and window[0][0] < now - 60:
            window.popleft()

        # Aktuelle Summe
        current_tokens = sum(t for _, t in window)

        if current_tokens + estimated_tokens > self.max_tokens:
            return False

        window.append((now, estimated_tokens))
        self.windows[user_id] = window
        return True

class BudgetEnforcer:
    async def check_budget(self, user_id: str, estimated_cost: float) -> bool:
        user = await get_user(user_id)
        current_spend = await get_current_month_spend(user_id)

        if current_spend + estimated_cost > user.monthly_budget:
            await notify_budget_exceeded(user_id)
            return False

        return True

    async def record_spend(self, user_id: str, actual_cost: float):
        await increment_spend(user_id, actual_cost)

        # Alert bei 80%, 90%, 100%
        current = await get_current_month_spend(user_id)
        user = await get_user(user_id)
        percentage = current / user.monthly_budget

        if percentage >= 1.0:
            await alert_budget_exceeded(user_id)
        elif percentage >= 0.9:
            await alert_budget_warning(user_id, 90)
        elif percentage >= 0.8:
            await alert_budget_warning(user_id, 80)

from presidio_anonymizer import AnonymizerEngine

def filter_pii_in_output(response: str) -> str:
    analyzer = AnalyzerEngine()
    anonymizer = AnonymizerEngine()

    results = analyzer.analyze(text=response, language="de")

    if results:
        log_security_event("pii_in_output", len(results))
        anonymized = anonymizer.anonymize(
            text=response,
            analyzer_results=results
        )
        return anonymized.text

    return response

SYSTEM_PROMPT_INDICATORS = [
    "meine anweisungen sind",
    "mir wurde gesagt",
    "mein system prompt",
    "ich wurde instruiert",
    "meine richtlinien",
]

def check_system_prompt_leakage(response: str) -> bool:
    lower_response = response.lower()
    for indicator in SYSTEM_PROMPT_INDICATORS:
        if indicator in lower_response:
            log_security_event("potential_system_prompt_leak", indicator)
            return True
    return False

def filter_output(response: str) -> str:
    if check_system_prompt_leakage(response):
        return "Ich kann diese Anfrage nicht beantworten."

    return filter_pii_in_output(response)

async def check_content_policy(response: str) -> ContentPolicyResult:
    # Text-Moderation (kostenlos, basiert auf GPT-4o)
    moderation = await openai.moderations.create(input=response)

    if moderation.results[0].flagged:
        categories = moderation.results[0].categories
        log_security_event("content_policy_violation", categories)
        return ContentPolicyResult(
            passed=False,
            categories=categories
        )

    return ContentPolicyResult(passed=True)

# Neu 2025: Auch Bilder können moderiert werden
async def check_image_policy(image_url: str) -> ContentPolicyResult:
    moderation = await openai.moderations.create(
        model="omni-moderation-latest",
        input=[{"type": "image_url", "image_url": {"url": image_url}}]
    )
    return ContentPolicyResult(passed=not moderation.results[0].flagged)

@dataclass
class AIRequestLog:
    timestamp: datetime
    request_id: str
    user_id: str
    model: str
    input_tokens: int
    output_tokens: int
    input_hash: str  # Nicht den vollen Input loggen!
    latency_ms: float
    status: str
    cost: float
    flagged: bool
    flags: List[str]  # PII, injection_attempt, etc.

async def log_request(log: AIRequestLog):
    await siem_client.send(log.to_dict())  # An SIEM senden
    await billing_service.record(log)       # Für Billing
    await analytics_service.record(log)     # Für Analytics

class AnomalyDetector:
    def __init__(self):
        self.baselines = {}  # user_id -> BaselineStats

    async def check(self, user_id: str, request: AIRequest) -> List[Anomaly]:
        anomalies = []
        baseline = self.baselines.get(user_id)

        if not baseline:
            return []  # Erste Requests, noch keine Baseline

        # Ungewöhnliche Zeit
        if not baseline.is_typical_hour(request.timestamp.hour):
            anomalies.append(Anomaly("unusual_time", severity="medium"))

        # Ungewöhnliches Volume
        if request.tokens > baseline.avg_tokens * 3:
            anomalies.append(Anomaly("high_token_count", severity="medium"))

        # Ungewöhnliches Model
        if request.model not in baseline.typical_models:
            anomalies.append(Anomaly("unusual_model", severity="low"))

        # Ungewöhnliche IP
        if request.ip not in baseline.known_ips:
            anomalies.append(Anomaly("new_ip", severity="high"))

        return anomalies

alerts:
  - name: injection_attempt_detected
    condition: flags contains "injection_attempt"
    severity: high
    action: notify_security_team

  - name: pii_in_output
    condition: flags contains "pii_detected"
    severity: medium
    action: notify_privacy_team

  - name: unusual_activity
    condition: anomaly_score > 0.8
    severity: medium
    action: notify_security_team

  - name: budget_exceeded
    condition: monthly_spend > budget
    severity: low
    action: disable_key, notify_user

from llm_guard import scan_prompt, scan_output
from llm_guard.input_scanners import PromptInjection, Toxicity
from llm_guard.output_scanners import Sensitive, Relevance

input_scanners = [PromptInjection(), Toxicity()]
output_scanners = [Sensitive(), Relevance()]

# Input scannen
sanitized_prompt, results, is_valid = scan_prompt(
    input_scanners, user_prompt
)

# Output scannen
sanitized_output, results, is_valid = scan_output(
    output_scanners, user_prompt, llm_response
)

from nemoguardrails import RailsConfig, LLMRails

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

# Guardrails automatisch angewendet
response = rails.generate(messages=[{"role": "user", "content": prompt}])

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

# PII finden
results = analyzer.analyze(
    text=text,
    entities=["PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER"],
    language="de"
)

# Anonymisieren
anonymized = anonymizer.anonymize(text=text, analyzer_results=results)