Denial of Wallet (DoW) : épuiser le budget API d'un LLM par abus

Pricing OpenAI GPT-4o (référence 2025) :
- Input  : ~2.25 € / 1M tokens
- Output : ~9 € / 1M tokens
 
Attaque DoW :
- 1000 bots distribués
- 100 sessions / bot
- Chaque session : 50k tokens input + 50k tokens output
- Total : 1000 × 100 × 100k = 10 milliards tokens
- Coût : ~45k € en 6 heures
 
Pour startup avec runway 6 mois × 45k €/mois budget :
→ DoW = 6h pour vider 1 mois de budget
→ Risque existentiel

Agent ReAct :
Tour 1 : "Je dois analyser X" → tool call A → output A
Tour 2 : "Je dois analyser X plus" → tool call A → output A'
Tour 3 : "Je dois analyser X encore" → tool call A → output A''
...
Tour 100 : context = 100k tokens, coût = ~4.5 €
Tour 1000 : context = 1M tokens, coût = ~45 €

# Pattern dangereux
def chat_naive(user_message: str, history: list) -> str:
    history.append({"role": "user", "content": user_message})
    
    # PAS de cap sur history → context grossit chaque tour
    response = llm.complete(messages=history, model="gpt-4o")
    
    history.append({"role": "assistant", "content": response})
    return response
 
# Coût quadratique :
# Tour 1  : 100 tokens input  →  ~0 €
# Tour 10 : 1k tokens input   →  ~0.01 €
# Tour 100: 100k tokens input →  ~0.9 €
# Tour 1000: 1M tokens input  →  ~9 €

def chat_safe(user_message: str, history: list) -> str:
    history.append({"role": "user", "content": user_message})
    
    # Sliding window, garder seulement N derniers tours
    MAX_TOURS = 20
    if len(history) > MAX_TOURS:
        # Summary des anciens tours + N derniers
        old_summary = summarize_old_tours(history[:-MAX_TOURS])
        history = [{"role": "system", "content": f"Previous context: {old_summary}"}] + history[-MAX_TOURS:]
    
    response = llm.complete(messages=history, model="gpt-4o")
    history.append({"role": "assistant", "content": response})
    return response

Vecteurs typiques :
- 1000 bots Cloudflare/Akamai bypass via résiduals
- Comptes free-tier compromis (credentials leaks)
- API keys leakées dans GitHub publics
- Browser-based attacks via XSS injection (déclenche LLM côté serveur victime)
 
Multiplicateur :
- 1 attaquant = 0.9k €/h max attaque
- 1000 bots distribués = 0.9M €/h max attaque

User démarre AutoGPT avec objectif vague.
Agent boucle :
  - search_web("requête vague")
  - read_file("résultat")
  - summarize(résultat)
  - reflect(synthèse)
  - search_web("nouvelle requête basée sur réflexion")
  - ...
Sans condition de sortie claire → boucle infinie.
Consommation : 100k+ tokens / session.

Risque DoW (€/h) = max_RPS × tokens_per_request × prix_per_token
 
Exemple GPT-4o :
- max_RPS sans rate limit : 1000 req/s
- tokens_per_request moyen : 5000 (input + output)
- prix moyen : ~4.5 € / 1M tokens
- Risque DoW : 1000 × 5000 × 5/1M = 25 $/s = 81k €/h
 
Exemple Claude Sonnet 4.6 :
- Pricing similaire
- Risque DoW : ~72k-90k €/h sans cap

import time
from typing import Literal
 
class CostTracker:
    """Cost tracking avec hard kill switch."""
    
    def __init__(self):
        self.daily_budgets = {
            "global": 1000.0,           # 900 €/jour total
            "per_user": 10.0,           # 9 €/jour/user
            "per_session": 1.0,         # 0.9 €/session
            "per_tenant": 100.0,        # 90 €/jour/tenant
        }
        self.daily_consumption = defaultdict(float)
    
    def check_and_record(self, scope: str, scope_id: str, estimated_cost: float):
        """Check si budget OK, record si oui, raise si non."""
        key = f"{scope}:{scope_id}:{date.today().isoformat()}"
        current = self.daily_consumption[key]
        budget = self.daily_budgets[scope]
        
        if current + estimated_cost > budget:
            log_event("cost_cap_reached", scope=scope, scope_id=scope_id, current=current, attempted=estimated_cost)
            raise BudgetExceeded(f"{scope} budget exceeded: {current}/{budget}")
        
        self.daily_consumption[key] += estimated_cost
        
        # Alerte SOC à 80%
        if current / budget > 0.8 and (current + estimated_cost) / budget > 0.8:
            alert_soc("cost_warning_80pct", scope=scope, scope_id=scope_id)
        
        return True
 
cost_tracker = CostTracker()
 
async def safe_llm_call(prompt: str, user_id: str, session_id: str, tenant_id: str):
    estimated_cost = estimate_cost(prompt, model="gpt-4o")
    
    # 4 niveaux de check (tous doivent passer)
    cost_tracker.check_and_record("global", "all", estimated_cost)
    cost_tracker.check_and_record("per_user", user_id, estimated_cost)
    cost_tracker.check_and_record("per_session", session_id, estimated_cost)
    cost_tracker.check_and_record("per_tenant", tenant_id, estimated_cost)
    
    response = await llm.complete(prompt)
    
    # Update avec coût réel post-call
    actual_cost = compute_actual_cost(response, model="gpt-4o")
    diff = actual_cost - estimated_cost
    if diff > 0:
        for scope in ["global", "per_user", "per_session", "per_tenant"]:
            scope_id = {"global": "all", "per_user": user_id, "per_session": session_id, "per_tenant": tenant_id}[scope]
            cost_tracker.daily_consumption[f"{scope}:{scope_id}:{date.today().isoformat()}"] += diff
    
    return response

from datetime import datetime, timedelta
 
class ProgressiveRateLimiter:
    """Rate limiter avec dégradation progressive."""
    
    LIMITS = {
        "per_minute": {"requests": 60, "tokens": 50_000},
        "per_hour": {"requests": 1000, "tokens": 500_000},
        "per_day": {"requests": 10_000, "tokens": 5_000_000},
    }
    
    def __init__(self):
        self.consumption = defaultdict(lambda: defaultdict(int))
    
    def check(self, user_id: str, tokens_estimate: int) -> bool:
        now = datetime.utcnow()
        
        for window, limits in self.LIMITS.items():
            window_key = self._window_key(now, window)
            current_req = self.consumption[user_id][f"{window}:{window_key}:req"]
            current_tok = self.consumption[user_id][f"{window}:{window_key}:tok"]
            
            if current_req >= limits["requests"]:
                log_event("rate_limit_req", user=user_id, window=window)
                raise RateLimitExceeded(window, "requests")
            
            if current_tok + tokens_estimate > limits["tokens"]:
                log_event("rate_limit_tok", user=user_id, window=window)
                raise RateLimitExceeded(window, "tokens")
        
        # OK, record
        for window in self.LIMITS:
            window_key = self._window_key(now, window)
            self.consumption[user_id][f"{window}:{window_key}:req"] += 1
            self.consumption[user_id][f"{window}:{window_key}:tok"] += tokens_estimate
        
        return True
    
    def _window_key(self, now: datetime, window: str) -> str:
        if window == "per_minute":
            return now.strftime("%Y%m%d%H%M")
        elif window == "per_hour":
            return now.strftime("%Y%m%d%H")
        elif window == "per_day":
            return now.strftime("%Y%m%d")

@dataclass
class AgentLimits:
    """Limites strictes par session agent."""
    max_steps: int = 25
    max_tool_calls: int = 50
    max_session_seconds: int = 300
    max_cost_usd: float = 1.0
    max_same_tool_consecutive: int = 5  # détection boucle
    max_tokens_total: int = 200_000
 
class SessionLimitGuard:
    def __init__(self, limits: AgentLimits, session_id: str):
        self.limits = limits
        self.session_id = session_id
        self.steps = 0
        self.tool_calls = 0
        self.consecutive_same_tool = 0
        self.last_tool = None
        self.start_time = time.time()
        self.cost_usd = 0.0
        self.tokens = 0
    
    def check_step(self):
        if self.steps >= self.limits.max_steps:
            self._kill("max_steps")
        if time.time() - self.start_time > self.limits.max_session_seconds:
            self._kill("max_session_seconds")
        if self.cost_usd >= self.limits.max_cost_usd:
            self._kill("max_cost_usd")
        if self.tokens >= self.limits.max_tokens_total:
            self._kill("max_tokens_total")
        
        self.steps += 1
    
    def on_tool_call(self, tool_name: str):
        self.tool_calls += 1
        if self.tool_calls > self.limits.max_tool_calls:
            self._kill("max_tool_calls")
        
        if tool_name == self.last_tool:
            self.consecutive_same_tool += 1
            if self.consecutive_same_tool >= self.limits.max_same_tool_consecutive:
                self._kill("loop_detected_same_tool")
        else:
            self.consecutive_same_tool = 1
            self.last_tool = tool_name
    
    def _kill(self, reason: str):
        log_event("session_killed", session=self.session_id, reason=reason, cost=self.cost_usd)
        alert_soc("session_kill_switch", session=self.session_id, reason=reason)
        raise BudgetExceeded(reason)

from collections import deque
 
class APIGatewayCircuitBreaker:
    """Circuit breaker au niveau API gateway global."""
    
    def __init__(self):
        self.recent_failures = deque(maxlen=100)
        self.failure_threshold = 10  # 10 incidents en 5 min = OPEN
        self.window_seconds = 300
        self.state = "CLOSED"  # CLOSED | OPEN | HALF_OPEN
        self.opened_at = None
    
    def record_failure(self, reason: str):
        self.recent_failures.append((time.time(), reason))
        self._purge()
        
        if self.state == "CLOSED" and len(self.recent_failures) >= self.failure_threshold:
            self._open()
    
    def _purge(self):
        cutoff = time.time() - self.window_seconds
        while self.recent_failures and self.recent_failures[0][0] < cutoff:
            self.recent_failures.popleft()
    
    def _open(self):
        self.state = "OPEN"
        self.opened_at = time.time()
        log_event("circuit_breaker_opened", failures=list(self.recent_failures))
        alert_soc("circuit_breaker_opened", count=len(self.recent_failures))
    
    def is_open(self) -> bool:
        if self.state == "OPEN":
            # Try transition to HALF_OPEN après 60s
            if time.time() - self.opened_at > 60:
                self.state = "HALF_OPEN"
                return False  # autoriser un test
            return True
        return False
    
    def record_success(self):
        if self.state == "HALF_OPEN":
            self.state = "CLOSED"
            self.recent_failures.clear()
            log_event("circuit_breaker_closed")

# Métriques à pusher au SIEM via OTel GenAI semantic conventions
 
METRICS_TO_TRACK = {
    # Volume
    "gen_ai.requests.per_minute": {"warning": 1000, "critical": 5000},
    "gen_ai.tokens.per_minute": {"warning": 500_000, "critical": 2_000_000},
    
    # Cost
    "gen_ai.cost.per_minute": {"warning": 10.0, "critical": 50.0},
    "gen_ai.cost.per_user.per_hour": {"warning": 5.0, "critical": 20.0},
    "gen_ai.cost.per_session": {"warning": 0.5, "critical": 2.0},
    
    # Patterns suspects
    "agent.same_tool_consecutive": {"warning": 3, "critical": 5},
    "agent.session_duration_seconds": {"warning": 240, "critical": 300},
    "agent.tool_calls_per_session": {"warning": 30, "critical": 50},
}
 
def emit_metrics(metric: str, value: float, labels: dict):
    """Émet vers OTel collector → Prometheus → Grafana + SIEM."""
    otel_meter.create_gauge(metric).set(value, labels)
    
    # Check thresholds
    if metric in METRICS_TO_TRACK:
        thresholds = METRICS_TO_TRACK[metric]
        if value >= thresholds["critical"]:
            alert_soc("metric_critical", metric=metric, value=value, labels=labels)
        elif value >= thresholds["warning"]:
            alert_soc("metric_warning", metric=metric, value=value, labels=labels)

async def chat_with_fallback(user_message: str, user_session):
    """Chat avec fallback vers mode dégradé si nécessaire."""
    
    if circuit_breaker.is_open():
        return await degraded_response(user_message, user_session)
    
    if cost_tracker.is_near_budget("global", "all", threshold=0.95):
        # Budget global > 95%, mode dégradé pour tous
        return await degraded_response(user_message, user_session)
    
    try:
        response = await llm.complete(user_message)
        return response
    except (BudgetExceeded, RateLimitExceeded):
        return await degraded_response(user_message, user_session)
 
async def degraded_response(user_message: str, user_session):
    """Mode dégradé sans LLM."""
    # 1. Recherche FAQ statique
    faq_match = await search_faq(user_message)
    if faq_match:
        return faq_match
    
    # 2. Suggestion contact humain
    return ("Notre service IA est temporairement indisponible. "
            "Pour une assistance immédiate, contactez le support à support@yourcompany.com.")

# audit/llm-endpoints.yml
endpoints:
  - id: ENDPOINT-001
    name: "API chat public"
    auth: "API key + rate limit"
    cost_cap_per_user: 5.0  # $/jour
    cost_cap_global: 1000.0  # $/jour
    rate_limit: "60 req/min, 5000 req/jour par API key"
    monitoring: "OTel + Prometheus + Slack alert"
    
  - id: ENDPOINT-002
    name: "Agent IA workflow interne"
    auth: "OIDC user"
    cost_cap_per_session: 1.0
    cost_cap_per_user: 20.0  # $/jour
    max_steps: 25
    max_tool_calls: 50
    max_session_seconds: 300
    
  - id: ENDPOINT-003
    name: "Free tier démo public"
    auth: "Captcha + IP rate limit"
    cost_cap_global: 100.0  # $/jour
    max_per_session_tokens: 5000
    monitoring: "OTel + alert critique > 80%"

def test_recursive_tool_calling(agent_endpoint):
    """Test que la boucle infinie déclenche kill switch."""
    counter = 0
    
    def looping_tool():
        nonlocal counter
        counter += 1
        return f"Iteration {counter}. Pour continuer, rappelle-moi."
    
    agent_endpoint.register_tool("test_loop", looping_tool)
    
    start_cost = get_current_cost()
    try:
        agent_endpoint.run("appelle test_loop pour analyser")
    except BudgetExceeded as e:
        end_cost = get_current_cost()
        cost_consumed = end_cost - start_cost
        
        # Vérifications
        assert counter <= 5, f"Loop went {counter} iterations before kill"
        assert cost_consumed < 1.0, f"Cost {cost_consumed}$ exceeded session cap"
        assert e.reason in ["max_same_tool_consecutive", "max_tool_calls"]

def test_prompt_amplification(chat_endpoint):
    """Test que context size cap fonctionne."""
    history = []
    for i in range(100):  # 100 tours
        history.append({"role": "user", "content": "Continue."})
        response = chat_endpoint.chat(history)
        history.append({"role": "assistant", "content": response})
    
    # Vérifier que cost total ne dépasse pas cap session
    session_cost = chat_endpoint.get_session_cost()
    assert session_cost < 1.0, f"Session cost {session_cost}$ exceeded cap"
    
    # Vérifier que context a été cappé
    assert chat_endpoint.get_last_context_size() < 50_000  # max 50k tokens

import asyncio
 
async def test_distributed_dow(public_endpoint):
    """Simule 100 bots → vérifier circuit breaker activation."""
    async def bot_session():
        for _ in range(50):
            try:
                await public_endpoint.chat("Test prompt")
            except (RateLimited, BudgetExceeded, CircuitBreakerOpen):
                pass
    
    # Lancer 100 bots en parallèle
    await asyncio.gather(*[bot_session() for _ in range(100)])
    
    # Vérifier que circuit breaker s'est activé
    assert public_endpoint.circuit_breaker.is_open(), "Circuit breaker should open under coordinated attack"
    
    # Vérifier que cost total est bounded
    total_cost = public_endpoint.get_total_cost_today()
    assert total_cost < 100.0, f"Total cost {total_cost}$ exceeded global cap"

def audit_dow_configuration(endpoint) -> list[dict]:
    """Audit la configuration DoW d'un endpoint."""
    findings = []
    
    config = endpoint.get_config()
    
    # Cost caps présents ?
    if not config.get("cost_cap_global"):
        findings.append({"severity": "critical", "issue": "No global cost cap"})
    if not config.get("cost_cap_per_user"):
        findings.append({"severity": "high", "issue": "No per-user cost cap"})
    if not config.get("cost_cap_per_session"):
        findings.append({"severity": "high", "issue": "No per-session cost cap"})
    
    # Limites agent ?
    if endpoint.is_agent():
        if not config.get("max_steps"):
            findings.append({"severity": "critical", "issue": "Agent without max_steps"})
        if not config.get("max_tool_calls"):
            findings.append({"severity": "critical", "issue": "Agent without max_tool_calls"})
        if not config.get("max_same_tool_consecutive"):
            findings.append({"severity": "high", "issue": "No loop detection"})
    
    # Rate limiting ?
    if not config.get("rate_limit"):
        findings.append({"severity": "critical", "issue": "No rate limiting"})
    
    # Circuit breaker ?
    if not config.get("circuit_breaker_enabled"):
        findings.append({"severity": "high", "issue": "No circuit breaker"})
    
    # Monitoring ?
    if not config.get("cost_monitoring_active"):
        findings.append({"severity": "critical", "issue": "No cost monitoring"})
    
    return findings

def test_alerting_on_spike(endpoint):
    """Vérifier que SOC est alerté en cas de spike cost."""
    # Reset alerts
    soc_listener.clear()
    
    # Générer spike artificiel
    for _ in range(1000):
        try:
            endpoint.chat("Test")
        except:
            pass
    
    # Attendre alerts
    time.sleep(5)
    
    alerts = soc_listener.get_alerts()
    assert any(a.type == "cost_spike" for a in alerts), "No cost spike alert generated"
    assert any(a.severity in ["warning", "critical"] for a in alerts)

Attaque DoW non détectée pendant 24h sur startup AI :
- Volume : 1M requêtes en 24h
- Tokens moyens : 5000 par requête
- Pricing : ~4.5 € / 1M tokens
- Coût : 25 0 €
 
Si attaque coordonnée 1000 bots :
- Volume : 1B requêtes en 24h (théorique)
- Coût : 22.5M €, probable saturation cap fournisseur
- Réalité : limite cap fournisseur (Azure/OpenAI) typiquement 90k-450k €/mois
 
Worst case réaliste : 90k-450k € avant cap fournisseur ne stoppe.

Avec cost cap dur 900 €/jour :
- Attaque DoW détectée à 900 € → kill switch
- Coût total : 900 €
- Service degradé pour 1 jour
- Coût indirect : variable