Sécuriser les API LLM : rate limiting, quotas, anti-abuse

import time
import redis.asyncio as redis
 
r = redis.Redis(host="localhost", port=6379)
 
async def check_rate_limit(
    key: str,
    max_count: int,
    window_seconds: int,
) -> tuple[bool, int]:
    """
    Sliding window rate limit avec Redis sorted set.
    Retourne (allowed, count_in_window).
    """
    now = time.time()
    window_start = now - window_seconds
    
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, window_start)
    pipe.zcard(key)
    pipe.zadd(key, {f"{now}:{id(now)}": now})
    pipe.expire(key, window_seconds + 1)
    _, count, _, _ = await pipe.execute()
    
    if count >= max_count:
        return False, count
    return True, count + 1
 
 
# Application
async def check_user_limits(user_id: str) -> dict:
    checks = {
        "req_per_second": await check_rate_limit(f"rl:s:{user_id}", 5, 1),
        "req_per_minute": await check_rate_limit(f"rl:m:{user_id}", 30, 60),
        "req_per_day": await check_rate_limit(f"rl:d:{user_id}", 1000, 86400),
    }
    
    blocked = [k for k, (allowed, _) in checks.items() if not allowed]
    return {"allowed": len(blocked) == 0, "blocked_dimensions": blocked}

async def check_token_budget(
    user_id: str,
    estimated_input_tokens: int,
    max_output_tokens: int,
) -> bool:
    """Vérifie tokens input/output budget avant appel LLM."""
    # Limite par minute
    INPUT_TOKENS_PER_MIN = 50000
    OUTPUT_TOKENS_PER_MIN = 30000
    
    in_key = f"tokens:in:{user_id}:{int(time.time()) // 60}"
    out_key = f"tokens:out:{user_id}:{int(time.time()) // 60}"
    
    current_in = int(await r.get(in_key) or 0)
    current_out = int(await r.get(out_key) or 0)
    
    if current_in + estimated_input_tokens > INPUT_TOKENS_PER_MIN:
        return False
    if current_out + max_output_tokens > OUTPUT_TOKENS_PER_MIN:
        return False
    
    # Réserver
    pipe = r.pipeline()
    pipe.incrby(in_key, estimated_input_tokens)
    pipe.incrby(out_key, max_output_tokens)
    pipe.expire(in_key, 70)
    pipe.expire(out_key, 70)
    await pipe.execute()
    
    return True
 
 
async def update_actual_token_usage(
    user_id: str,
    actual_input_tokens: int,
    actual_output_tokens: int,
    estimated_max_output_tokens: int,
):
    """Réajuste après le vrai usage (refund la différence sur output max)."""
    refund = estimated_max_output_tokens - actual_output_tokens
    if refund > 0:
        out_key = f"tokens:out:{user_id}:{int(time.time()) // 60}"
        await r.decrby(out_key, refund)

PRICING = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "claude-sonnet-4-6": {"input": 3.00 / 1_000_000, "output": 15.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}
 
async def check_cost_budget(
    user_id: str,
    org_id: str,
    model: str,
    input_tokens: int,
    estimated_output_tokens: int,
) -> dict:
    cost = (
        input_tokens * PRICING[model]["input"]
        + estimated_output_tokens * PRICING[model]["output"]
    )
    
    user_key = f"cost:user:{user_id}:{datetime.utcnow().date()}"
    org_key = f"cost:org:{org_id}:{datetime.utcnow().date()}"
    
    user_spent = float(await r.get(user_key) or 0)
    org_spent = float(await r.get(org_key) or 0)
    
    USER_DAILY_BUDGET = 10.0
    ORG_DAILY_BUDGET = 1000.0
    
    if user_spent + cost > USER_DAILY_BUDGET:
        return {"allowed": False, "reason": "user daily budget exceeded"}
    if org_spent + cost > ORG_DAILY_BUDGET:
        return {"allowed": False, "reason": "org daily budget exceeded"}
    
    # Soft warn à 80%
    if user_spent + cost > USER_DAILY_BUDGET * 0.8:
        await alert_soc(f"User {user_id} at 80% daily budget")
    
    return {"allowed": True, "estimated_cost": cost}

class RequestBudget:
    def __init__(
        self,
        max_tool_calls: int = 10,
        max_total_tokens: int = 50000,
        max_duration_s: int = 60,
    ):
        self.max_tool_calls = max_tool_calls
        self.max_total_tokens = max_total_tokens
        self.max_duration_s = max_duration_s
        
        self.tool_calls = 0
        self.total_tokens = 0
        self.start = time.time()
    
    def check(self):
        if self.tool_calls >= self.max_tool_calls:
            raise BudgetExceeded(f"max {self.max_tool_calls} tool calls")
        if self.total_tokens >= self.max_total_tokens:
            raise BudgetExceeded(f"max {self.max_total_tokens} tokens")
        if time.time() - self.start >= self.max_duration_s:
            raise BudgetExceeded(f"max {self.max_duration_s}s duration")
    
    def add_tool_call(self):
        self.tool_calls += 1
        self.check()
    
    def add_tokens(self, n: int):
        self.total_tokens += n
        self.check()
 
 
# Usage dans agent loop
budget = RequestBudget(max_tool_calls=10)
 
while not done:
    budget.check()
    response = await llm_call(...)
    budget.add_tokens(response.usage.total_tokens)
    
    if response.tool_calls:
        for tool_call in response.tool_calls:
            budget.add_tool_call()
            result = await execute_tool(tool_call)
            messages.append(result)

async def chat(req: ChatReq):
    # JAMAIS faire confiance à req.max_tokens
    SERVER_MAX_TOKENS = 1500
    
    response = await openai_client.chat.completions.create(
        model=req.model,
        messages=req.messages,
        max_tokens=SERVER_MAX_TOKENS,  # forçage côté serveur
    )

def estimate_input_tokens(messages: list[dict]) -> int:
    text = " ".join(m["content"] for m in messages if isinstance(m["content"], str))
    return len(text) // 4  # approximation rapide
 
@app.post("/chat")
async def chat(req: ChatReq):
    estimated = estimate_input_tokens(req.messages)
    
    if estimated > 50000:
        # Cas exceptionnel, demander review
        await log_anomaly("large_input", user_id, estimated)
        if estimated > 100000:
            raise HTTPException(413, "Input too large")

# AWS Budgets via CDK
aws budgets create-budget \
    --account-id $ACCOUNT_ID \
    --budget '{
        "BudgetName": "OpenAI-Monthly",
        "BudgetLimit": {"Amount": "5000", "Unit": "EUR"},
        "TimeUnit": "MONTHLY",
        "BudgetType": "COST"
    }' \
    --notifications-with-subscribers '[
        {
            "Notification": {
                "NotificationType": "ACTUAL",
                "ComparisonOperator": "GREATER_THAN",
                "Threshold": 80
            },
            "Subscribers": [{"SubscriptionType": "EMAIL", "Address": "soc@example.com"}]
        },
        {
            "Notification": {
                "NotificationType": "FORECASTED",
                "ComparisonOperator": "GREATER_THAN",
                "Threshold": 100
            },
            "Subscribers": [{"SubscriptionType": "SNS", "Address": "arn:aws:sns:...:incident-response"}]
        }
    ]'

async def detect_extraction_attempt(user_id: str, query: str):
    # 1. Diversity / entropy de queries
    user_queries_24h = await get_user_queries(user_id, hours=24)
    if len(set(user_queries_24h)) / len(user_queries_24h) > 0.95 and len(user_queries_24h) > 100:
        await flag_user(user_id, "high_query_diversity")
    
    # 2. Patterns systématiques (variations paramétriques)
    if detect_parametric_pattern(user_queries_24h):
        await flag_user(user_id, "parametric_scraping")
    
    # 3. Volume disproportionné
    user_profile = await get_user_profile(user_id)
    if len(user_queries_24h) > user_profile.baseline_volume * 10:
        await flag_user(user_id, "volume_anomaly")
    
    # 4. User agent / IP suspects
    if user_profile.ip in CLOUD_PROVIDER_IPS:
        await flag_user(user_id, "cloud_ip")

# Pseudo-code (implémentations open-source disponibles)
from watermark import WatermarkProcessor
 
processor = WatermarkProcessor(
    vocab=tokenizer.vocab,
    gamma=0.25,  # fraction green list
    delta=2.0,   # logit boost
)
 
logits_processor = LogitsProcessorList([processor])
output = model.generate(input_ids, logits_processor=logits_processor)
 
# Détection a posteriori sur output suspect
detector = WatermarkDetector(vocab, gamma=0.25, delta=2.0)
score = detector.detect(suspicious_output)
if score.z_score > 4:
    print("Watermark detected with high confidence")

async def maybe_challenge(user_id: str, request):
    flags = await get_user_flags(user_id)
    if "high_query_diversity" in flags or "parametric_scraping" in flags:
        if not request.cookies.get("captcha_solved_recently"):
            return require_captcha()
    return None

import httpx
import hashlib
 
async def check_password_pwned(password: str) -> bool:
    sha1 = hashlib.sha1(password.encode()).hexdigest().upper()
    prefix, suffix = sha1[:5], sha1[5:]
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.pwnedpasswords.com/range/{prefix}")
    return suffix in r.text
 
 
@app.post("/auth/login")
async def login(req: LoginReq):
    # 1. Rate limit login
    rl = await check_rate_limit(f"login:{req.email}", 10, 3600)
    if not rl[0]:
        raise HTTPException(429, "Too many attempts")
    
    # 2. Check pwned
    if await check_password_pwned(req.password):
        await alert_user_password_compromised(req.email)
        raise HTTPException(401, "Password found in breach. Reset required.")
    
    # 3. Verify password
    user = await verify_credentials(req.email, req.password)
    if not user:
        await record_failed_attempt(req.email, request.client.host)
        raise HTTPException(401)
    
    # 4. Anomaly detection
    if await is_new_device_or_location(user, request):
        await send_mfa_challenge(user)
        return {"mfa_required": True}
    
    # 5. MFA
    if user.mfa_enabled:
        await send_mfa_challenge(user)
        return {"mfa_required": True}
    
    return await issue_token(user)

services:
  - name: llm-api
    url: http://app-internal:8000
    routes:
      - name: chat
        paths: ["/chat"]
    plugins:
      - name: rate-limiting
        config:
          minute: 30
          hour: 200
          day: 1000
          policy: redis
          redis_host: redis.internal
      
      - name: jwt
        config:
          secret_is_base64: false
          claims_to_verify: ["exp"]
      
      - name: prometheus
        config:
          per_consumer: true
      
      - name: request-size-limiting
        config:
          allowed_payload_size: 256  # KB max

http {
    limit_req_zone $http_x_user_id zone=user_per_min:10m rate=30r/m;
    limit_req_zone $binary_remote_addr zone=ip_per_sec:10m rate=10r/s;
    
    server {
        listen 443 ssl;
        server_name api.example.com;
        
        location /chat {
            limit_req zone=user_per_min burst=10 nodelay;
            limit_req zone=ip_per_sec burst=5;
            
            client_max_body_size 256K;
            client_body_timeout 30s;
            
            proxy_pass http://app-internal:8000;
            proxy_set_header X-Request-Id $request_id;
            proxy_read_timeout 60s;
        }
    }
}

import structlog
 
logger = structlog.get_logger()
 
@app.post("/chat")
async def chat(req: ChatReq, request: Request):
    request_id = str(uuid.uuid4())
    
    # ... process ...
    
    logger.info(
        "llm_request_completed",
        request_id=request_id,
        user_id=user_id,
        org_id=org_id,
        model=req.model,
        tokens_in=actual_tokens_in,
        tokens_out=actual_tokens_out,
        cost_usd=actual_cost,
        duration_ms=duration_ms,
        guardrail_blocks=guardrail_blocks,
        tool_calls=tool_calls,
    )

from prometheus_client import Counter, Histogram
 
llm_requests = Counter("llm_requests_total", "LLM requests", ["model", "status"])
llm_tokens = Counter("llm_tokens_total", "Tokens consumed", ["model", "direction"])
llm_cost = Counter("llm_cost_usd_total", "EUR cost", ["model", "user", "org"])
llm_duration = Histogram("llm_duration_seconds", "Request duration", ["model"])

# alertmanager rules
groups:
  - name: llm_cost
    rules:
      - alert: LLMOrgCostHigh
        expr: increase(llm_cost_usd_total[1h]) > 100
        for: 5m
        annotations:
          summary: "Org {{ $labels.org }} spent > 90 € in 1h"
      
      - alert: LLMUserCostHigh
        expr: increase(llm_cost_usd_total{user!=""}[1h]) > 20
        for: 5m
        annotations:
          summary: "User {{ $labels.user }} spent > 18 € in 1h"
      
      - alert: LLMRecursiveLoop
        expr: rate(llm_tool_calls_total[1m]) > 100
        annotations:
          summary: "Possible recursive tool calling on {{ $labels.user }}"