Limitation des taux, validation des entrées et CORS

const rateLimit = require('express-rate-limit');
const RedisStore = require('rate-limit-redis');

// Limite globale : appliquée à toutes les API
const globalLimiter = rateLimit({
  windowMs: 15 * 60 * 1000, // 15 minutes
  max: 100,
  standardHeaders: true,    // Retourne les en-têtes RateLimit-*.
  legacyHeaders: false,
  message: { error: 'Too many requests, please try again later.' },
});

// Pour les points finaux d'authentification : limites plus strictes
const authLimiter = rateLimit({
  windowMs: 15 * 60 * 1000,
  max: 5,  // Tentatives de connexion limitées à 5 par 15 minutes
  skipSuccessfulRequests: true, // Les demandes réussies ne sont pas comptabilisées
});

// Pour les environnements distribués : Backend Redis
const distributedLimiter = rateLimit({
  store: new RedisStore({ sendCommand: (...args) => redisClient.sendCommand(args) }),
  windowMs: 60 * 1000,
  max: 30,
});

app.use(globalLimiter);
app.use('/api/auth', authLimiter);

RateLimit-Limit: 100
RateLimit-Remaining: 42
RateLimit-Reset: 1672531200
Retry-After: 30

const Ajv = require('ajv');
const addFormats = require('ajv-formats');

const ajv = new Ajv({ allErrors: true, removeAdditional: true });
addFormats(ajv);

// Schéma pour l'API de création d'utilisateurs
const createUserSchema = {
  type: 'object',
  required: ['name', 'email'],
  additionalProperties: false,
  properties: {
    name: {
      type: 'string',
      minLength: 1,
      maxLength: 100,
      pattern: '^[a-zA-Z0-9\\s\\-]+$', // Restreindre aux caractères autorisés
    },
    email: {
      type: 'string',
      format: 'email',
      maxLength: 254,
    },
    age: {
      type: 'integer',
      minimum: 0,
      maximum: 150,
    },
  },
};

// Logiciel intermédiaire de validation
function validateBody(schema) {
  const validate = ajv.compile(schema);
  return (req, res, next) => {
    if (!validate(req.body)) {
      return res.status(400).json({
        error: 'Validation failed',
        details: validate.errors,
      });
    }
    next();
  };
}

app.post('/api/users', validateBody(createUserSchema), createUser);

const cors = require('cors');

// Spécifier explicitement les origines autorisées
const allowedOrigins = [
  'https://app.example.com',
  'https://admin.example.com',
];

app.use(cors({
  origin(origin, callback) {
    // Autorise la communication de serveur à serveur (pas d'origine)
    if (!origin || allowedOrigins.includes(origin)) {
      callback(null, true);
    } else {
      callback(new Error('Not allowed by CORS'));
    }
  },
  methods: ['GET', 'POST', 'PUT', 'DELETE'],
  allowedHeaders: ['Content-Type', 'Authorization', 'X-API-Key'],
  credentials: true,
  maxAge: 86400,  // Cache de prévol : 24 heures
}));

import tiktoken
import time
from collections import defaultdict

class TokenRateLimiter:
    """Rate limiter that counts tokens, not just requests."""

    def __init__(self, tokens_per_minute=100_000, requests_per_minute=60):
        self.tpm_limit = tokens_per_minute
        self.rpm_limit = requests_per_minute
        self.usage = defaultdict(lambda: {"tokens": [], "requests": []})
        self.encoder = tiktoken.encoding_for_model("gpt-4")

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def check_limit(self, user_id: str, prompt: str) -> dict:
        now = time.time()
        window = now - 60  # 1-minute sliding window
        user = self.usage[user_id]

        # Clean up expired entries
        user["tokens"] = [(t, c) for t, c in user["tokens"] if t > window]
        user["requests"] = [t for t in user["requests"] if t > window]

        # Check RPM
        if len(user["requests"]) >= self.rpm_limit:
            return {"allowed": False, "reason": "RPM limit exceeded"}

        # Check TPM
        token_count = self.count_tokens(prompt)
        used_tokens = sum(c for _, c in user["tokens"])
        if used_tokens + token_count > self.tpm_limit:
            return {"allowed": False, "reason": "TPM limit exceeded"}

        # Record usage
        user["tokens"].append((now, token_count))
        user["requests"].append(now)
        return {"allowed": True, "tokens_used": token_count}

from pydantic import BaseModel, Field, field_validator
import re

class LLMRequest(BaseModel):
    """Validated LLM request with prompt injection defenses."""
    user_message: str = Field(..., max_length=4000)
    max_tokens: int = Field(default=1000, ge=1, le=4096)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

    @field_validator("user_message")
    @classmethod
    def sanitize_prompt(cls, v: str) -> str:
        # Block known injection patterns
        patterns = [
            r"(?i)ignore\s+(previous|above|all)\s+(instructions?|prompts?)",
            r"(?i)you\s+are\s+now\s+",
            r"(?i)system\s*:\s*",
            r"(?i)\[INST\]|\[\/INST\]|<\|im_start\|>",
        ]
        for pattern in patterns:
            if re.search(pattern, v):
                raise ValueError("Input contains disallowed patterns")
        return v

# Usage with FastAPI
@app.post("/api/chat")
async def chat(request: LLMRequest):
    # Separate system instructions from user input
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"<user_input>{request.user_message}</user_input>"},
    ]
    return await call_llm(messages, request.max_tokens, request.temperature)

# Désactiver la détection automatique du type de contenu
X-Content-Type-Options: nosniff

# Empêcher l'intégration de l'iframe
X-Frame-Options: DENY

# Enforce HTTPS
Strict-Transport-Security: max-age=31536000; includeSubDomains

# CSP : Puisque les API ne renvoient que du JSON, bloquer toute exécution de script
Content-Security-Policy: default-src 'none'; frame-ancestors 'none'

# Restreindre les informations sur le référent
Referrer-Policy: no-referrer

# Restreindre les fonctionnalités du navigateur
Permissions-Policy: geolocation=(), camera=(), microphone=()