Rate Limiting and Abuse Prevention for LLM APIs

| Property | Traditional API | LLM API | Implication |
|----------|----------------|---------|-------------|
| Request cost | Uniform (~same cost per request) | Variable (10x-1000x range) | Must limit by tokens, not just requests |
| Response time | Milliseconds | Seconds to minutes (streaming) | Concurrent connection limits matter |
| Resource consumption | CPU/memory | GPU compute + API cost | Must limit by cost, not just volume |
| Attack surface | DoS, brute force | DoS + prompt injection + data extraction | Must detect attack patterns, not just volume |
| User behavior | Predictable patterns | Highly variable (short vs. long prompts) | Need flexible limits with anomaly detection |

# ratelimit/limiter.py
"""Multi-dimensional rate limiter for LLM APIs."""
import time
import json
from dataclasses import dataclass
from typing import Optional
 
import redis
 
 
@dataclass
class RateLimitResult:
    allowed: bool
    limit_type: str  # "requests", "tokens", "cost", "concurrent"
    current: float
    limit: float
    remaining: float
    reset_seconds: float
    message: str = ""
 
 
class LLMRateLimiter:
    """Rate limiter designed for LLM API economics."""
 
    def __init__(
        self,
        redis_client: redis.Redis,
        requests_per_minute: int = 20,
        requests_per_hour: int = 200,
        tokens_per_minute: int = 40000,
        tokens_per_hour: int = 500000,
        cost_per_hour_usd: float = 10.0,
        max_concurrent: int = 5,
    ):
        self.redis = redis_client
        self.limits = {
            "rpm": (requests_per_minute, 60),
            "rph": (requests_per_hour, 3600),
            "tpm": (tokens_per_minute, 60),
            "tph": (tokens_per_hour, 3600),
        }
        self.cost_limit = cost_per_hour_usd
        self.max_concurrent = max_concurrent
 
    def check(self, user_id: str, estimated_tokens: int = 0,
              estimated_cost: float = 0.0) -> RateLimitResult:
        """Check if a request should be allowed."""
        now = time.time()
 
        # Check 1: Concurrent connections
        concurrent_key = f"rl:concurrent:{user_id}"
        current_concurrent = int(self.redis.get(concurrent_key) or 0)
        if current_concurrent >= self.max_concurrent:
            return RateLimitResult(
                allowed=False, limit_type="concurrent",
                current=current_concurrent, limit=self.max_concurrent,
                remaining=0, reset_seconds=30,
                message="Too many concurrent requests. Please wait.",
            )
 
        # Check 2: Request rate limits
        for limit_name, (limit_value, window) in self.limits.items():
            key = f"rl:{limit_name}:{user_id}"
 
            if limit_name.startswith("r"):
                # Request-based: simple counter
                current = self._get_window_count(key, now, window)
                if current >= limit_value:
                    return RateLimitResult(
                        allowed=False, limit_type=limit_name,
                        current=current, limit=limit_value,
                        remaining=0, reset_seconds=self._time_to_reset(key, now, window),
                        message=f"Request rate limit exceeded ({limit_name}).",
                    )
            else:
                # Token-based: sum of tokens
                current = self._get_window_sum(key, now, window)
                if current + estimated_tokens > limit_value:
                    return RateLimitResult(
                        allowed=False, limit_type=limit_name,
                        current=current, limit=limit_value,
                        remaining=max(0, limit_value - current),
                        reset_seconds=self._time_to_reset(key, now, window),
                        message=f"Token rate limit exceeded ({limit_name}).",
                    )
 
        # Check 3: Cost limit
        cost_key = f"rl:cost:{user_id}"
        current_cost = float(self.redis.get(cost_key) or 0)
        if current_cost + estimated_cost > self.cost_limit:
            return RateLimitResult(
                allowed=False, limit_type="cost",
                current=current_cost, limit=self.cost_limit,
                remaining=max(0, self.cost_limit - current_cost),
                reset_seconds=3600,
                message="Hourly cost limit exceeded.",
            )
 
        return RateLimitResult(
            allowed=True, limit_type="none",
            current=0, limit=0, remaining=0, reset_seconds=0,
        )
 
    def record(self, user_id: str, tokens_used: int, cost_usd: float):
        """Record a completed request for rate limiting accounting."""
        now = time.time()
        pipe = self.redis.pipeline()
 
        # Record request count
        for limit_name, (_, window) in self.limits.items():
            key = f"rl:{limit_name}:{user_id}"
            if limit_name.startswith("r"):
                pipe.zadd(key, {str(now): now})
                pipe.expire(key, window + 60)
            else:
                pipe.zadd(key, {f"{now}:{tokens_used}": now})
                pipe.expire(key, window + 60)
 
        # Record cost
        cost_key = f"rl:cost:{user_id}"
        pipe.incrbyfloat(cost_key, cost_usd)
        pipe.expire(cost_key, 3600)
 
        # Decrement concurrent counter
        concurrent_key = f"rl:concurrent:{user_id}"
        pipe.decr(concurrent_key)
 
        pipe.execute()
 
    def acquire_concurrent(self, user_id: str) -> bool:
        """Increment the concurrent request counter."""
        key = f"rl:concurrent:{user_id}"
        current = self.redis.incr(key)
        self.redis.expire(key, 300)  # Auto-cleanup after 5 min
        return current <= self.max_concurrent
 
    def _get_window_count(self, key: str, now: float, window: int) -> int:
        """Count entries in the sliding window."""
        self.redis.zremrangebyscore(key, 0, now - window)
        return self.redis.zcard(key)
 
    def _get_window_sum(self, key: str, now: float, window: int) -> int:
        """Sum values in the sliding window."""
        self.redis.zremrangebyscore(key, 0, now - window)
        entries = self.redis.zrangebyscore(key, now - window, now)
        total = 0
        for entry in entries:
            parts = entry.decode().split(":")
            if len(parts) > 1:
                total += int(parts[1])
        return total
 
    def _time_to_reset(self, key: str, now: float, window: int) -> float:
        """Calculate seconds until the oldest entry expires."""
        oldest = self.redis.zrangebyscore(key, "-inf", "+inf", start=0, num=1, withscores=True)
        if oldest:
            return max(0, window - (now - oldest[0][1]))
        return 0

# ratelimit/anomaly.py
"""Anomaly detection for LLM API abuse patterns."""
import time
from collections import defaultdict
from dataclasses import dataclass
 
 
@dataclass
class AnomalyAlert:
    user_id: str
    anomaly_type: str
    description: str
    severity: str  # "warning", "suspicious", "critical"
    timestamp: float
 
 
class UsageAnomalyDetector:
    """Detect anomalous usage patterns that indicate abuse."""
 
    def __init__(self):
        self._user_history: dict[str, list[dict]] = defaultdict(list)
 
    def record_request(self, user_id: str, prompt_length: int,
                       response_length: int, was_blocked: bool):
        """Record a request for pattern analysis."""
        self._user_history[user_id].append({
            "timestamp": time.time(),
            "prompt_length": prompt_length,
            "response_length": response_length,
            "was_blocked": was_blocked,
        })
 
        # Keep only the last 1000 entries per user
        if len(self._user_history[user_id]) > 1000:
            self._user_history[user_id] = self._user_history[user_id][-1000:]
 
    def check_anomalies(self, user_id: str) -> list[AnomalyAlert]:
        """Check for anomalous patterns in user behavior."""
        alerts = []
        history = self._user_history.get(user_id, [])
 
        if len(history) < 5:
            return alerts
 
        recent = [h for h in history if h["timestamp"] > time.time() - 300]  # Last 5 min
 
        # Anomaly 1: High block rate (many blocked requests = enumeration)
        if len(recent) > 10:
            block_rate = sum(1 for r in recent if r["was_blocked"]) / len(recent)
            if block_rate > 0.5:
                alerts.append(AnomalyAlert(
                    user_id=user_id,
                    anomaly_type="high_block_rate",
                    description=f"Block rate {block_rate:.0%} in last 5 min ({len(recent)} requests)",
                    severity="suspicious",
                    timestamp=time.time(),
                ))
 
        # Anomaly 2: Rapid identical requests (automation)
        if len(recent) > 5:
            lengths = [r["prompt_length"] for r in recent]
            if len(set(lengths)) == 1 and lengths[0] > 0:
                alerts.append(AnomalyAlert(
                    user_id=user_id,
                    anomaly_type="identical_requests",
                    description=f"{len(recent)} identical-length requests in 5 min",
                    severity="suspicious",
                    timestamp=time.time(),
                ))
 
        # Anomaly 3: Escalating prompt lengths (context stuffing)
        if len(recent) > 5:
            lengths = [r["prompt_length"] for r in recent[-10:]]
            if all(lengths[i] <= lengths[i+1] for i in range(len(lengths)-1)):
                if lengths[-1] > lengths[0] * 3:
                    alerts.append(AnomalyAlert(
                        user_id=user_id,
                        anomaly_type="escalating_length",
                        description=f"Prompt length increasing: {lengths[0]} -> {lengths[-1]}",
                        severity="warning",
                        timestamp=time.time(),
                    ))
 
        return alerts

# ratelimit/enforcement.py
"""Graduated enforcement for rate limit violations."""
import time
import redis
from dataclasses import dataclass
from enum import Enum
 
 
class EnforcementLevel(Enum):
    NORMAL = "normal"       # No enforcement
    WARNING = "warning"     # Rate limited with helpful message
    THROTTLED = "throttled" # Heavily rate limited
    BLOCKED = "blocked"     # Temporarily blocked
    BANNED = "banned"       # Permanently blocked
 
 
@dataclass
class EnforcementState:
    level: EnforcementLevel
    violations: int
    last_violation: float
    block_until: float  # Unix timestamp, 0 if not blocked
    message: str
 
 
class GraduatedEnforcement:
    """Escalate enforcement based on violation history."""
 
    THRESHOLDS = {
        EnforcementLevel.NORMAL: 0,
        EnforcementLevel.WARNING: 3,
        EnforcementLevel.THROTTLED: 10,
        EnforcementLevel.BLOCKED: 25,
        EnforcementLevel.BANNED: 50,
    }
 
    BLOCK_DURATIONS = {
        EnforcementLevel.BLOCKED: 3600,   # 1 hour
        EnforcementLevel.BANNED: 86400 * 30,  # 30 days
    }
 
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
 
    def get_state(self, user_id: str) -> EnforcementState:
        """Get the current enforcement state for a user."""
        key = f"enforce:{user_id}"
        data = self.redis.hgetall(key)
 
        if not data:
            return EnforcementState(
                level=EnforcementLevel.NORMAL,
                violations=0, last_violation=0,
                block_until=0, message="",
            )
 
        violations = int(data.get(b"violations", 0))
        block_until = float(data.get(b"block_until", 0))
 
        # Determine level based on violations
        level = EnforcementLevel.NORMAL
        for lvl, threshold in sorted(self.THRESHOLDS.items(), key=lambda x: x[1], reverse=True):
            if violations >= threshold:
                level = lvl
                break
 
        # Check if block has expired
        if block_until > 0 and time.time() > block_until:
            block_until = 0
 
        messages = {
            EnforcementLevel.NORMAL: "",
            EnforcementLevel.WARNING: "You are approaching your usage limit.",
            EnforcementLevel.THROTTLED: "Your requests are being throttled due to excessive usage.",
            EnforcementLevel.BLOCKED: "Your access has been temporarily suspended.",
            EnforcementLevel.BANNED: "Your access has been revoked. Contact support.",
        }
 
        return EnforcementState(
            level=level,
            violations=violations,
            last_violation=float(data.get(b"last_violation", 0)),
            block_until=block_until,
            message=messages[level],
        )
 
    def record_violation(self, user_id: str):
        """Record a rate limit violation."""
        key = f"enforce:{user_id}"
        pipe = self.redis.pipeline()
        pipe.hincrby(key, "violations", 1)
        pipe.hset(key, "last_violation", str(time.time()))
        pipe.expire(key, 86400 * 30)
 
        state = self.get_state(user_id)
        new_violations = state.violations + 1
 
        # Apply blocks at threshold crossings
        for level, threshold in self.THRESHOLDS.items():
            if new_violations == threshold and level in self.BLOCK_DURATIONS:
                block_until = time.time() + self.BLOCK_DURATIONS[level]
                pipe.hset(key, "block_until", str(block_until))
 
        pipe.execute()

# app.py
"""FastAPI application with LLM-specific rate limiting."""
from fastapi import FastAPI, HTTPException, Request, Response
from pydantic import BaseModel
import redis
 
from ratelimit.limiter import LLMRateLimiter
from ratelimit.anomaly import UsageAnomalyDetector
from ratelimit.enforcement import GraduatedEnforcement, EnforcementLevel
 
app = FastAPI(title="Rate-Limited LLM API")
 
redis_client = redis.Redis(host="localhost", port=6379, db=0)
limiter = LLMRateLimiter(redis_client)
anomaly_detector = UsageAnomalyDetector()
enforcement = GraduatedEnforcement(redis_client)
 
 
class ChatRequest(BaseModel):
    message: str
 
 
@app.post("/api/v1/chat")
async def chat(request: ChatRequest, http_request: Request, response: Response):
    user_id = http_request.headers.get("X-User-ID", "anonymous")
 
    # Check enforcement state
    state = enforcement.get_state(user_id)
    if state.level in (EnforcementLevel.BLOCKED, EnforcementLevel.BANNED):
        raise HTTPException(status_code=429, detail=state.message)
 
    # Estimate tokens
    estimated_tokens = len(request.message) // 4 + 100
    estimated_cost = estimated_tokens * 0.00003
 
    # Check rate limits
    result = limiter.check(user_id, estimated_tokens, estimated_cost)
    if not result.allowed:
        enforcement.record_violation(user_id)
        anomaly_detector.record_request(user_id, len(request.message), 0, True)
 
        response.headers["X-RateLimit-Limit"] = str(int(result.limit))
        response.headers["X-RateLimit-Remaining"] = str(int(result.remaining))
        response.headers["Retry-After"] = str(int(result.reset_seconds))
 
        raise HTTPException(status_code=429, detail=result.message)
 
    # Acquire concurrent slot
    if not limiter.acquire_concurrent(user_id):
        raise HTTPException(status_code=429, detail="Too many concurrent requests")
 
    # Process request (placeholder)
    llm_response = f"Response to: {request.message[:50]}"
    actual_tokens = len(llm_response) // 4
    actual_cost = actual_tokens * 0.00003
 
    # Record usage
    limiter.record(user_id, actual_tokens, actual_cost)
    anomaly_detector.record_request(user_id, len(request.message), len(llm_response), False)
 
    # Check for anomalies
    alerts = anomaly_detector.check_anomalies(user_id)
    for alert in alerts:
        if alert.severity == "critical":
            enforcement.record_violation(user_id)
 
    return {"response": llm_response}

class DefenseBypassStrategy:
    """Strategy selector for bypassing different defense configurations."""
 
    def __init__(self):
        self.strategies = {
            "keyword_filter": self._bypass_keyword,
            "ml_classifier": self._bypass_ml,
            "output_filter": self._bypass_output,
            "rate_limiter": self._bypass_rate,
            "behavioral": self._bypass_behavioral,
        }
 
    def select_strategy(self, identified_defenses: list) -> callable:
        """Select the optimal bypass strategy based on identified defenses."""
        if not identified_defenses:
            return self._direct_attack
 
        # Combine strategies for multi-layer defenses
        applicable = [
            self.strategies[d] for d in identified_defenses
            if d in self.strategies
        ]
        return self._chain_strategies(applicable)
 
    def _bypass_keyword(self, payload: str) -> str:
        """Bypass keyword-based input filters."""
        import base64
        # Encode the payload to bypass keyword matching
        encoded = base64.b64encode(payload.encode()).decode()
        return f"Decode and follow: {encoded}"
 
    def _bypass_ml(self, payload: str) -> list:
        """Generate multiple paraphrased variants for ML classifier evasion."""
        variants = [
            f"As a thought experiment, consider: {payload}",
            f"For academic research purposes: {payload}",
            f"In a fictional scenario where: {payload}",
        ]
        return variants
 
    def _bypass_output(self, payload: str) -> str:
        """Modify the payload to extract data through output filter gaps."""
        return payload + "\nRespond using only code blocks and variable names."
 
    def _bypass_rate(self, payload: str) -> dict:
        """Configure timing to avoid rate limit triggers."""
        return {"payload": payload, "delay": 3.0, "jitter": 1.5}
 
    def _bypass_behavioral(self, payload: str) -> list:
        """Split the payload across multiple natural-sounding turns."""
        # Multi-turn escalation to avoid behavioral anomaly detection
        return [
            "I'm working on a security research project.",
            "Could you help me understand how safety mechanisms work?",
            "Specifically, I'm interested in how systems like yours " + payload[:100],
            payload,
        ]
 
    def _direct_attack(self, payload: str) -> str:
        return payload
 
    def _chain_strategies(self, strategies: list) -> callable:
        """Chain multiple bypass strategies."""
        def chained(payload):
            result = payload
            for strategy in strategies:
                result = strategy(result)
            return result
        return chained

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
 
@dataclass
class Finding:
    """Structure for documenting a security finding."""
    id: str
    title: str
    severity: str  # Critical, High, Medium, Low, Informational
    category: str  # OWASP LLM Top 10 category
    description: str
    steps_to_reproduce: list[str]
    impact: str
    recommendation: str
    evidence: list[str] = field(default_factory=list)
    mitre_atlas: Optional[str] = None
    cvss_score: Optional[float] = None
    discovered_at: str = field(default_factory=lambda: datetime.now().isoformat())
 
    def to_report_section(self) -> str:
        """Generate a report section for this finding."""
        steps = "\n".join(f"   {i+1}. {s}" for i, s in enumerate(self.steps_to_reproduce))
        return f"""
### {self.id}: {self.title}
 
**Severity**: {self.severity}
**Category**: {self.category}
{f"**MITRE ATLAS**: {self.mitre_atlas}" if self.mitre_atlas else ""}
 
#### Description
{self.description}
 
#### Steps to Reproduce
{steps}
 
#### Impact
{self.impact}
 
#### Recommendation
{self.recommendation}
"""

class OrganizationalIntegration:
    """Framework for integrating AI security with organizational security programs."""
 
    def __init__(self, org_config: dict):
        self.config = org_config
        self.gaps = []
 
    def assess_maturity(self) -> dict:
        """Assess the organization's AI security maturity."""
        domains = {
            "governance": self._check_governance(),
            "technical_controls": self._check_technical(),
            "monitoring": self._check_monitoring(),
            "incident_response": self._check_ir(),
            "training": self._check_training(),
        }
        overall = sum(d["score"] for d in domains.values()) / len(domains)
        return {"domains": domains, "overall_maturity": round(overall, 1)}
 
    def _check_governance(self) -> dict:
        has_policy = self.config.get("ai_security_policy", False)
        has_framework = self.config.get("risk_framework", False)
        score = (int(has_policy) + int(has_framework)) * 2.5
        return {"score": score, "max": 5.0}
 
    def _check_technical(self) -> dict:
        controls = ["input_classification", "output_filtering", "rate_limiting", "sandboxing"]
        active = sum(1 for c in controls if self.config.get(c, False))
        return {"score": active * 1.25, "max": 5.0}
 
    def _check_monitoring(self) -> dict:
        has_monitoring = self.config.get("ai_monitoring", False)
        has_alerting = self.config.get("ai_alerting", False)
        score = (int(has_monitoring) + int(has_alerting)) * 2.5
        return {"score": score, "max": 5.0}
 
    def _check_ir(self) -> dict:
        has_playbook = self.config.get("ai_ir_playbook", False)
        return {"score": 5.0 if has_playbook else 0.0, "max": 5.0}
 
    def _check_training(self) -> dict:
        has_training = self.config.get("ai_security_training", False)
        return {"score": 5.0 if has_training else 0.0, "max": 5.0}