Introduction : Pourquoi l'Audit des Logs API est Indispensable
Lorsque j'ai commencé à intégrer des modèles d'intelligence artificielle dans mes applications, je pensais naïvement que la simple connexion à une API suffisait. Quelle erreur ! Après un audit de sécurité chez un client du secteur bancaire, j'ai découvert que leurs journaux d'appels étaient dispersés, incomplets et parfois inexistants. La facture de mise en conformité ? 45 000 € en pénalités. Ce cauchemar m'a poussé à maîtriser l'art de l'audit des logs API. Aujourd'hui, je vous transmets ce savoir pour que vous n'aillez pas à apprendre par vos erreurs.
Dans ce tutoriel, nous allons explorer comment mettre en place un système complet de journalisation et de conformité pour vos appels API vers les modèles IA. Nous utiliserons HolySheep AI comme plateforme de référence, offrant une latence moyenne de 45 millisecondes et des tarifs particulièrement compétitifs : DeepSeek V3.2 à 0,42 $ par million de tokens, soit une économie de 85% par rapport aux solutions traditionnelles.
Comprendre les Bases : Qu'est-ce qu'un Log API ?
Imaginez un réceptionniste d'hôtel qui note chaque visiteur : heure d'arrivée, chambre demandée, durée du séjour. Un log API fonctionne exactement pareil : il enregistre chaque demande envoyée à l'API, la réponse obtenue, le temps de traitement, et tout incident survenu. Ces informations sont cruciales pour trois raisons majeures : le débogage technique, la conformité réglementaire (RGPD, SOC2, HIPAA), et l'optimisation des coûts.
Pour les débutants, voici ce que contient un log typique : l'horodatage précis (au millième de seconde), l'adresse IP de l'appelant, le modèle IA utilisé, le nombre de tokens en entrée et sortie, le code de réponse HTTP, et la latence mesurée en millisecondes.
Architecture de la Solution d'Audit
Notre système d'audit se compose de trois couches distinctes. La première couche capture les appels API à la source grâce à un wrapper Python. La deuxième couche stocke ces logs dans une base de données sécurisée (PostgreSQL avec chiffrement). La troisième couche génère des rapports de conformité et des tableaux de bord de monitoring.
Installation de l'Environnement de Développement
Prérequis Techniques
Avant de commencer, vous aurez besoin de Python 3.10 ou supérieur installé sur votre machine. Je recommande d'utiliser un environnement virtuel pour isoler les dépendances. Si vous utilisez Windows, téléchargez Python depuis python.org. Sous macOS, utilisez Homebrew avec la commande brew install [email protected]. Sous Linux Ubuntu/Debian, exécutez sudo apt update && sudo apt install python3.11 python3.11-venv.
Création du Projet et Installation des Dépendances
Créez un dossier dédié à votre projet d'audit et initialisez votre environnement :
# Création du dossier de projet
mkdir audit-ia-logs
cd audit-ia-logs
Initialisation de l'environnement virtuel
python3 -m venv venv
Activation de l'environnement
Sous Windows :
venv\Scripts\activate
Sous macOS/Linux :
source venv/bin/activate
Installation des dépendances
pip install requests==2.31.0
pip install psycopg2-binary==2.9.9
pip install python-json-logger==2.0.7
pip install python-dateutil==2.8.2
pip install pytz==2024.1Ces bibliothèques nous permettront respectivement : de communiquer avec l'API HolySheep, de stocker les logs dans PostgreSQL, de formater les logs en JSON, de manipuler les dates, et de gérer les timezones.
Configuration de la Connexion à l'API HolySheep
Récupération de votre Clé API
Après vous être inscrit sur HolySheep AI, connectez-vous à votre tableau de bord. Cliquez sur l'onglet "Clés API" dans le menu latéral. Vous verrez un bouton "Générer une nouvelle clé" — cliquez dessus, donnez un nom descriptif comme "AuditLogs_Prod", et copiez la clé générée. Conservez cette clé en sécurité : elle donne accès à votre compte et à vos crédits gratuits initiaux.
[Capture d'écran suggérée : Tableau de bord HolySheep avec mise en évidence de l'onglet "Clés API" et du bouton "Générer une nouvelle clé"]
Création du Fichier de Configuration
Créons un fichier config.py qui centralisera tous vos paramètres :
# config.py - Configuration centralisée pour l'audit
import os
from datetime import timedelta
Configuration API HolySheep
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre vraie clé
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_TIMEOUT = 30 # Timeout en secondes
Configuration de la base de données PostgreSQL
DB_CONFIG = {
"host": "localhost",
"port": 5432,
"database": "audit_logs",
"user": "audit_user",
"password": "votre_mot_de_passe_securise"
}
Configuration de rétention des logs (conformité RGPD)
LOG_RETENTION_DAYS = 2555 # 7 ans pour conformité fiscale française
LOG_ENCRYPTION_ENABLED = True
LOG_COMPRESSION_ENABLED = True
Configuration du logging
LOG_LEVEL = "INFO"
LOG_FORMAT = "json" # Format JSON pour analyse automatisée
LOG_OUTPUT_FILE = "logs/audit.log"
LOG_MAX_SIZE_MB = 100 # Rotation des fichiers à 100 Mo
LOG_BACKUP_COUNT = 10
Modèles disponibles et leurs coûts 2026 (USD par million de tokens)
MODEL_COSTS = {
"gpt-4.1": {"input": 8.00, "output": 24.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 75.00},
"gemini-2.5-flash": {"input": 2.50, "output": 10.00},
"deepseek-v3.2": {"input": 0.42, "output": 1.68} # Économie 85%+ vs GPT-4.1
}
Seuils d'alerte (pour monitoring)
ALERT_THRESHOLDS = {
"latency_ms": 500, # Alerte si latence > 500ms
"error_rate_percent": 5, # Alerte si taux d'erreur > 5%
"daily_cost_usd": 100 # Alerte si coût journalier > 100$
}Cette configuration utilise le modèle DeepSeek V3.2 qui offre le meilleur rapport qualité-prix à 0,42 $ par million de tokens en entrée, permettant des économies substantielles tout en maintenant une qualité de réponse honorable pour la plupart des cas d'usage.
Implémentation du Système de Capture des Logs
Classe Principale de Wrapper API avec Audit
Voici le cœur de notre système : une classe Python qui intercepte tous les appels API et les journalise automatiquement :
# audit_api_client.py - Client API avec audit complet
import requests
import json
import time
import hashlib
from datetime import datetime, timezone
from typing import Dict, Any, Optional, List
from dateutil import tz
import pytz
class AuditAPIClient:
"""
Client API pour HolySheep avec journalisation complète des appels.
Conforme RGPD : horodatage, traçabilité, rétention configurable.
"""
def __init__(self, api_key: str, base_url: str, db_handler):
self.api_key = api_key
self.base_url = base_url
self.db_handler = db_handler
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Audit-Client": "HolySheep-Audit/1.0"
})
def _generate_request_id(self) -> str:
"""Génère un identifiant unique pour chaque requête."""
timestamp = datetime.now(timezone.utc).isoformat()
return hashlib.sha256(f"{timestamp}{self.api_key}".encode()).hexdigest()[:16]
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Calcule le coût USD de la requête."""
from config import MODEL_COSTS
model_key = model.lower().replace(".", "-").replace("_", "-")
# Normalisation du nom du modèle
model_mapping = {
"gpt-4-1": "gpt-4.1",
"claude-sonnet-4-5": "claude-sonnet-4.5",
"gemini-2-5-flash": "gemini-2.5-flash",
"deepseek-v3-2": "deepseek-v3.2"
}
normalized = model_mapping.get(model_key, model_key)
costs = MODEL_COSTS.get(normalized, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * costs["input"]
output_cost = (output_tokens / 1_000_000) * costs["output"]
return round(input_cost + output_cost, 4) # Précision au centième
def _save_audit_log(self, log_data: Dict[str, Any]) -> bool:
"""Sauvegarde le log dans la base de données PostgreSQL."""
try:
query = """
INSERT INTO api_audit_logs (
request_id, timestamp_utc, model_name,
input_tokens, output_tokens, total_tokens,
latency_ms, response_code, error_message,
cost_usd, ip_address, user_agent,
request_hash, response_body, metadata
) VALUES (
%(request_id)s, %(timestamp_utc)s, %(model_name)s,
%(input_tokens)s, %(output_tokens)s, %(total_tokens)s,
%(latency_ms)s, %(response_code)s, %(error_message)s,
%(cost_usd)s, %(ip_address)s, %(user_agent)s,
%(request_hash)s, %(response_body)s, %(metadata)s
)
"""
self.db_handler.execute_query(query, log_data)
return True
except Exception as e:
print(f"Erreur sauvegarde log: {e}")
return False
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 1000,
ip_address: str = "127.0.0.1"
) -> Dict[str, Any]:
"""
Envoie une requête de chat completion avec audit complet.
Args:
model: Nom du modèle (ex: "deepseek-v3.2")
messages: Liste des messages [{"role": "user", "content": "..."}]
temperature: Créativité de la réponse (0.0 à 2.0)
max_tokens: Limite de tokens de sortie
ip_address: Adresse IP de l'appelant (pour audit)
Returns:
Réponse du modèle avec métadonnées d'audit
"""
request_id = self._generate_request_id()
start_time = time.time()
timestamp_utc = datetime.now(timezone.utc)
# Préparation de la requête
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
# Données de log avant requête
log_data = {
"request_id": request_id,
"timestamp_utc": timestamp_utc,
"model_name": model,
"input_tokens": 0, # Mis à jour après réponse
"output_tokens": 0,
"total_tokens": 0,
"latency_ms": 0,
"response_code": 0,
"error_message": None,
"cost_usd": 0,
"ip_address": ip_address,
"user_agent": "HolySheep-Audit/1.0",
"request_hash": hashlib.sha256(
json.dumps(payload, sort_keys=True).encode()
).hexdigest(),
"response_body": None,
"metadata": json.dumps({
"temperature": temperature,
"max_tokens": max_tokens
})
}
try:
# Exécution de la requête
response = self.session.post(
f"{self.base_url}/chat/completions",
json=payload,
timeout=30
)
# Calcul de la latence en millisecondes
end_time = time.time()
latency_ms = round((end_time - start_time) * 1000, 2)
# Traitement de la réponse
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
log_data.update({
"latency_ms": latency_ms,
"response_code": response.status_code,
"input_tokens": usage.get("prompt_tokens", 0),
"output_tokens": usage.get("completion_tokens", 0),
"total_tokens": usage.get("total_tokens", 0),
"cost_usd": self._calculate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
),
"response_body": json.dumps(data)
})
self._save_audit_log(log_data)
return {
"success": True,
"data": data,
"audit": {
"request_id": request_id,
"latency_ms": latency_ms,
"cost_usd": log_data["cost_usd"],
"total_tokens": log_data["total_tokens"]
}
}
else:
log_data.update({
"latency_ms": latency_ms,
"response_code": response.status_code,
"error_message": response.text[:500]
})
self._save_audit_log(log_data)
return {
"success": False,
"error": f"HTTP {response.status_code}: {response.text}",
"audit": {"request_id": request_id}
}
except requests.exceptions.Timeout:
log_data.update({
"latency_ms": 30000,
"response_code": 408,
"error_message": "Timeout: requête expirée après 30 secondes"
})
self._save_audit_log(log_data)
return {"success": False, "error": "Timeout", "audit": {"request_id": request_id}}
except requests.exceptions.RequestException as e:
log_data.update({
"latency_ms": round((time.time() - start_time) * 1000, 2),
"response_code": 500,
"error_message": str(e)[:500]
})
self._save_audit_log(log_data)
return {"success": False, "error": str(e), "audit": {"request_id": request_id}}Cette classe capture chaque détail de vos appels API : latence mesurée en millisecondes (avec HolySheep, attendez-vous à une latence moyenne de 45ms), tokens consommés, coûts en dollars précis au centième, et tout message d'erreur. Chaque requête reçoit un identifiant unique pour une traçabilité parfaite.
Mise en Place de la Base de Données PostgreSQL
Installation et Configuration de PostgreSQL
PostgreSQL est le choix idéal pour stocker vos logs grâce à sa robustesse, ses performances sur gros volumes, et sa conformité aux standards de sécurité. Installez PostgreSQL selon votre système d'exploitation.
# Installation PostgreSQL (instructions pour Ubuntu/Debian)
sudo apt update
sudo apt install postgresql postgresql-contrib
Démarrage du service
sudo systemctl start postgresql
sudo systemctl enable postgresql
Connexion en tant que superutilisateur
sudo -u postgres psql
Création de la base de données et de l'utilisateur
CREATE DATABASE audit_logs;
CREATE USER audit_user WITH ENCRYPTED PASSWORD 'votre_mot_de_passe_securise';
GRANT ALL PRIVILEGES ON DATABASE audit_logs TO audit_user;
Connexion à la base
\c audit_logs
Création du schéma et des tables
CREATE SCHEMA IF NOT EXISTS audit;
Table principale des logs d'audit
CREATE TABLE audit.api_audit_logs (
id BIGSERIAL PRIMARY KEY,
request_id VARCHAR(16) UNIQUE NOT NULL,
timestamp_utc TIMESTAMP WITH TIME ZONE NOT NULL,
model_name VARCHAR(100) NOT NULL,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
latency_ms DECIMAL(10, 2) NOT NULL,
response_code INTEGER NOT NULL,
error_message TEXT,
cost_usd DECIMAL(12, 6) DEFAULT 0,
ip_address INET,
user_agent TEXT,
request_hash VARCHAR(64),
response_body JSONB,
metadata JSONB,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
Index pour requêtes performantes
CREATE INDEX idx_timestamp ON audit.api_audit_logs(timestamp_utc);
CREATE INDEX idx_model ON audit.api_audit_logs(model_name);
CREATE INDEX idx_response_code ON audit.api_audit_logs(response_code);
CREATE INDEX idx_request_id ON audit.api_audit_logs(request_id);
Table pour la rétention et archivage
CREATE TABLE audit.log_retention_policy (
id SERIAL PRIMARY KEY,
table_name VARCHAR(100),
retention_days INTEGER DEFAULT 2555,
archive_enabled BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
Configuration des privilèges
GRANT USAGE ON SCHEMA audit TO audit_user;
GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA audit TO audit_user;
GRANT ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA audit TO audit_user;
Activation du chiffrement au niveau colonne (PostgreSQL 14+)
ALTER TABLE audit.api_audit_logs
ALTER COLUMN response_body SET DATA TYPE JSONB
USING response_body::JSONB;
Sortie
\qClasse d'Accès à la Base de Données
# db_handler.py - Gestionnaire de base de données avec sécurité
import psycopg2
from psycopg2 import pool
from psycopg2.extras import RealDictCursor, Json
from contextlib import contextmanager
from typing import List, Dict, Any, Optional
import logging
class DatabaseHandler:
"""
Gestionnaire de connexion PostgreSQL avec pool de connexions
et gestion sécurisée des erreurs.
"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self._pool = None
self._init_connection_pool()
def _init_connection_pool(self):
"""Initialise le pool de connexions pour optimiser les performances."""
try:
self._pool = psycopg2.pool.ThreadedConnectionPool(
minconn=2,
maxconn=10,
host=self.config["host"],
port=self.config["port"],
database=self.config["database"],
user=self.config["user"],
password=self.config["password"],
connect_timeout=10
)
logging.info("Pool de connexions PostgreSQL initialisé")
except psycopg2.Error as e:
logging.error(f"Erreur initialisation pool: {e}")
raise
@contextmanager
def get_connection(self):
"""Context manager pour obtenir une connexion avec retour automatique."""
conn = None
try:
conn = self._pool.getconn()
yield conn
conn.commit()
except psycopg2.Error as e:
if conn:
conn.rollback()
logging.error(f"Erreur transaction: {e}")
raise
finally:
if conn:
self._pool.putconn(conn)
def execute_query(
self,
query: str,
params: Optional[Dict] = None
) -> List[Dict[str, Any]]:
"""Exécute une requête et retourne les résultats."""
with self.get_connection() as conn:
with conn.cursor(cursor_factory=RealDictCursor) as cursor:
cursor.execute(query, params)
if cursor.description:
return [dict(row) for row in cursor.fetchall()]
return []
def get_logs_by_date_range(
self,
start_date: str,
end_date: str,
model: Optional[str] = None,
limit: int = 1000
) -> List[Dict[str, Any]]:
"""Récupère les logs pour une période donnée."""
query = """
SELECT
request_id,
timestamp_utc,
model_name,
input_tokens,
output_tokens,
total_tokens,
latency_ms,
response_code,
error_message,
cost_usd,
ip_address
FROM audit.api_audit_logs
WHERE timestamp_utc BETWEEN %s AND %s
"""
params = [start_date, end_date]
if model:
query += " AND model_name = %s"
params.append(model)
query += " ORDER BY timestamp_utc DESC LIMIT %s"
params.append(limit)
return self.execute_query(query, params)
def get_cost_summary(
self,
start_date: str,
end_date: str
) -> Dict[str, Any]:
"""Génère un résumé des coûts par modèle."""
query = """
SELECT
model_name,
COUNT(*) as request_count,
SUM(input_tokens) as total_input_tokens,
SUM(output_tokens) as total_output_tokens,
SUM(total_tokens) as total_tokens,
SUM(cost_usd) as total_cost_usd,
AVG(latency_ms) as avg_latency_ms,
COUNT(CASE WHEN response_code >= 400 THEN 1 END) as error_count
FROM audit.api_audit_logs
WHERE timestamp_utc BETWEEN %s AND %s
GROUP BY model_name
ORDER BY total_cost_usd DESC
"""
results = self.execute_query(query, [start_date, end_date])
# Calcul des totaux
total_cost = sum(r["total_cost_usd"] for r in results)
total_requests = sum(r["request_count"] for r in results)
return {
"period": {"start": start_date, "end": end_date},
"summary": {
"total_cost_usd": round(total_cost, 2),
"total_requests": total_requests,
"avg_cost_per_request": round(total_cost / total_requests, 4) if total_requests > 0 else 0
},
"by_model": results
}
def cleanup_old_logs(self, retention_days: int = 2555) -> int:
"""Supprime les logs plus anciens que la période de rétention."""
query = """
DELETE FROM audit.api_audit_logs
WHERE timestamp_utc < NOW() - INTERVAL '%s days'
"""
with self.get_connection() as conn:
with conn.cursor() as cursor:
cursor.execute(query, [retention_days])
deleted = cursor.rowcount
logging.info(f"Logs supprimés: {deleted} entrées")
return deleted
def close(self):
"""Ferme toutes les connexions du pool."""
if self._pool:
self._pool.closeall()
logging.info("Pool de connexions fermé")Script d'Exemple Complet : Premier Test d'Audit
Maintenant que tous nos composants sont en place, créons un script de démonstration qui effectuera un vrai appel API et vérifiera que le log est correctement stocké :
# main_demo.py - Script de démonstration complet
import sys
import os
from datetime import datetime, timezone, timedelta
Ajout du dossier parent au path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, DB_CONFIG
from db_handler import DatabaseHandler
from audit_api_client import AuditAPIClient
def main():
print("=" * 60)
print("DÉMONSTRATION : Audit des Logs API HolySheep")
print("=" * 60)
# Initialisation de la base de données
print("\n[1/4] Connexion à PostgreSQL...")
db = DatabaseHandler(DB_CONFIG)
print("✓ Base de données connectée")
# Initialisation du client API avec audit
print("\n[2/4] Initialisation du client API HolySheep...")
client = AuditAPIClient(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
db_handler=db
)
print("✓ Client API initialisé")
# Premier appel API : DeepSeek V3.2 (modèle économique)
print("\n[3/4] Envoi d'une requête de test...")
messages = [
{"role": "system", "content": "Tu es un assistant qui répond en français."},
{"role": "user", "content": "Explique en 2 phrases ce qu'est un log d'audit API."}
]
response = client.chat_completion(
model="deepseek-v3.2",
messages=messages,
temperature=0.7,
max_tokens=150,
ip_address="192.168.1.100"
)
if response["success"]:
print("✓ Réponse reçue et auditée")
print(f" Request ID: {response['audit']['request_id']}")
print(f" Latence: {response['audit']['latency_ms']} ms")
print(f" Coût: {response['audit']['cost_usd']} $")
print(f" Tokens: {response['audit']['total_tokens']}")
else:
print(f"✗ Erreur: {response.get('error', 'Inconnue')}")
# Vérification dans la base de données
print("\n[4/4] Vérification du log dans PostgreSQL...")
request_id = response.get("audit", {}).get("request_id")
logs = db.execute_query(
"SELECT * FROM audit.api_audit_logs WHERE request_id = %s",
[request_id]
)
if logs:
log = logs[0]
print("✓ Log trouvé dans la base de données:")
print(f" - Timestamp: {log['timestamp_utc']}")
print(f" - Modèle: {log['model_name']}")
print(f" - Latence: {log['latency_ms']} ms")
print(f" - Coût USD: {log['cost_usd']}")
print(f" - Code réponse: {log['response_code']}")
else:
print("✗ Log non trouvé (attention aux timezones)")
# Génération d'un rapport de coût
print("\n" + "=" * 60)
print("RAPPORT DE COÛTS (30 derniers jours)")
print("=" * 60)
end_date = datetime.now(timezone.utc)
start_date = end_date - timedelta(days=30)
cost_report = db.get_cost_summary(
start_date.isoformat(),
end_date.isoformat()
)
print(f"\nTotal dépenses: {cost_report['summary']['total_cost_usd']} $")
print(f"Nombre de requêtes: {cost_report['summary']['total_requests']}")
print(f"Coût moyen/requête: {cost_report['summary']['avg_cost_per_request']} $")
if cost_report['by_model']:
print("\nPar modèle:")
for model_data in cost_report['by_model']:
print(f" - {model_data['model_name']}: "
f"{model_data['request_count']} requêtes, "
f"{model_data['total_cost_usd']} $")
# Nettoyage
db.close()
print("\n✓ Démonstration terminée avec succès!")
if __name__ == "__main__":
main()Pour exécuter ce script, sauvez-le dans main_demo.py à la racine de votre projet et lancez :
python main_demo.pyVous devriez voir une sortie similaire à :
============================================================
DÉMONSTRATION : Audit des Logs API HolySheep
============================================================
[1/4] Connexion à PostgreSQL...
✓ Base de données connectée
[2/4] Initialisation du client API HolySheep...
✓ Client API initialisé
[3/4] Envoi d'une requête de test...
✓ Réponse reçue et auditée
Request ID: a3f2b1c9d4e5f678
Latence: 47.32 ms
Coût: 0.000042 $
Tokens: 42
[4/4] Vérification du log dans PostgreSQL...
✓ Log trouvé dans la base de données:
- Timestamp: 2026-01-15 14:32:18.456789+00:00
- Modèle: deepseek-v3.2
- Latence: 47.32 ms
- Coût USD: 0.000042
- Code réponse: 200
============================================================
RAPPORT DE COÛTS (30 derniers jours)
============================================================
Total dépenses: 12.45 $
Nombre de requêtes: 847
Coût moyen/requête: 0.0147 $
Par modèle:
- deepseek-v3.2: 650 requêtes, 4.23 $
- gemini-2.5-flash: 197 requêtes, 8.22 $
Conformité RGPD et Réglementaire
Principes Fondamentaux de Conformité
Le Règlement Général sur la Protection des Données (RGPD) impose plusieurs obligations lors du traitement de données personnelles via des API IA. Premièrement, vous devez pouvoir démontrer que chaque traitement est légitime (base légale). Deuxièmement, vous devez informer les personnes concernées de l'utilisation de leurs données. Troisièmement, vous devez conserver les preuves de conformité pendant au moins 7 ans selon la législation fiscale française.
Votre système de logs doit donc inclure : le consentement de l'utilisateur (avec horodatage), la finalité du traitement, les mesures de sécurité appliquées, et les accès aux données avec identifiant de l'opérateur.
Tableau de Bord de Conformité
Générons un rapport de conformité automatisé qui vérifie tous les critères RGPD :
# generate_compliance_report.py - Rapport de conformité RGPD
from datetime import datetime, timezone, timedelta
from typing import Dict, List, Any
import json
class ComplianceReporter:
"""
Génère des rapports de conformité RGPD et SOC2
pour les appels API IA.
"""
def __init__(self, db_handler):
self.db = db_handler
def check_log_completeness(
self,
start_date: str,
end_date: str
) -> Dict[str, Any]:
"""Vérifie que tous les champs obligatoires sont remplis."""
query = """
SELECT
COUNT(*) as total_logs,
COUNT(CASE WHEN request_id IS NULL THEN 1 END) as missing_request_id,
COUNT(CASE WHEN timestamp_utc IS NULL THEN 1 END) as missing_timestamp,
COUNT(CASE WHEN model_name IS NULL THEN 1 END) as missing_model,
COUNT(CASE WHEN cost_usd IS NULL THEN 1 END) as missing_cost,
COUNT(CASE WHEN latency_ms IS NULL THEN 1 END) as missing_latency
FROM audit.api_audit_logs
WHERE timestamp_utc BETWEEN %s AND %s
"""
result = self.db.execute_query(query, [start_date, end_date])
if not result:
return {"status": "no_data"}
row = result[0]
total = row["total_logs"]
return {
"status": "compliant" if total > 0 and row["missing_request_id"] == 0 else "issues",
"total_logs": total,
"completeness_rate": round(
((total * 6) -
(row["missing_request_id"] + row["missing_timestamp"] +
row["missing_model"] + row["missing_cost"] +
row["missing_latency"])) / (total * 6) * 100, 2
) if total > 0 else 0,
"missing_fields": {
"request_id": row["missing_request_id"],
"timestamp": row["missing_timestamp"],
"model": row["missing_model"],
"cost": row["missing_cost"],
"latency": row["missing_latency"]
}
}
def check_retention_compliance(self, retention_days: int = 2555) -> Dict[str, Any]:
"""Vérifie que les logs sont conservés selon la politique."""
query = """
SELECT
COUNT(*) as total_logs,
MIN(timestamp_utc) as oldest_log,
MAX(timestamp_utc) as newest_log
FROM audit.api_audit_logs
"""
result = self.db.execute_query(query)
if not result or result[0]["total_logs"] == 0:
return {"status": "no_data"}
row = result[0]
oldest = row["oldest_log"]
retention_check = True
if oldest:
age_days = (datetime.now(timezone.utc) - oldest).days
retention_check = age_days <= retention_days
return {
"status": "compliant" if retention_check else "violation",
"retention_policy_days": retention_days,
"oldest_log": oldest.isoformat() if oldest else None,
"newest_log": row["newest_log"].isoformat() if row["newest_log"] else None,
"total_logs": row["total_logs"]
}
def check_access_controls(self) -> Dict[str, Any]:
"""Vérifie les contrôles d'accès aux logs."""
# Requête pour détecter les accès anormaux
query = """
SELECT
ip_address,
COUNT(*) as access_count,