Chiffrer les prompts envoyés à OpenAI : analyse risque/conformité

# Classifier et router
def classify_data_for_llm(content: str, context: dict) -> dict:
    """Classifie la donnée pour décider quel LLM utiliser."""
    
    findings = {
        "rgpd_article_9": detect_health_political_religious(content),
        "secrets": detect_secrets(content),  # API keys, passwords, certs
        "code_proprietary": detect_proprietary_code(content),
        "pii_massive": count_pii(content) > 10,
        "ndatop": context.get("under_nda_strict", False),
    }
    
    if findings["secrets"]:
        return {"action": "REJECT", "reason": "secrets detected, never send to LLM"}
    
    if findings["rgpd_article_9"] or findings["code_proprietary"]:
        return {"action": "ROUTE_ONPREM", "reason": "high sensitivity"}
    
    if findings["pii_massive"]:
        return {"action": "PSEUDONYMIZE_AND_AZURE_EU", "reason": "minimization required"}
    
    # Default : tier OpenAI Enterprise OK
    return {"action": "OPENAI_ENTERPRISE", "reason": "standard"}

from openai import AsyncAzureOpenAI
 
client = AsyncAzureOpenAI(
    azure_endpoint="https://my-resource-france-central.openai.azure.com/",
    api_version="2024-08-01-preview",
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
)
 
response = await client.chat.completions.create(
    model="gpt-4o-deployment",  # nom du deployment, pas du model
    messages=[...],
    max_tokens=1500,
)

from mistralai import Mistral
 
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
 
response = await client.chat.complete_async(
    model="mistral-large-latest",
    messages=[...],
    max_tokens=1500,
)

import boto3
 
client = boto3.client(
    "bedrock-runtime",
    region_name="eu-central-1",  # Frankfurt
)
 
response = client.converse(
    modelId="anthropic.claude-sonnet-4-6-20250101-v2:0",
    messages=[{"role": "user", "content": [{"text": "..."}]}],
    inferenceConfig={"maxTokens": 1500},
)

from google.cloud import aiplatform
from vertexai.generative_models import GenerativeModel
 
aiplatform.init(project="my-project", location="europe-west4")  # Pays-Bas
 
model = GenerativeModel("gemini-2.0-pro-exp-2026")
response = model.generate_content(
    "...",
    generation_config={"max_output_tokens": 1500},
)

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
 
analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()
 
PII_ENTITIES = ["EMAIL_ADDRESS", "PHONE_NUMBER", "PERSON", "FR_NIR", "IBAN_CODE", "CREDIT_CARD"]
 
# Mapping local pour reverse-substitution
class PIIMapper:
    def __init__(self):
        self._map = {}  # token → real value
        self._counter = {}  # entity_type → counter
    
    def get_token(self, entity_type: str, real_value: str) -> str:
        if real_value in self._map.values():
            return [k for k, v in self._map.items() if v == real_value][0]
        
        self._counter[entity_type] = self._counter.get(entity_type, 0) + 1
        token = f"[{entity_type}_{self._counter[entity_type]:03d}]"
        self._map[token] = real_value
        return token
    
    def reverse(self, text: str) -> str:
        for token, real_value in self._map.items():
            text = text.replace(token, real_value)
        return text
 
 
def pseudonymize_for_llm(text: str, mapper: PIIMapper, language: str = "fr") -> str:
    results = analyzer.analyze(text=text, language=language, entities=PII_ENTITIES)
    
    # Replace from end to start to preserve indices
    text_pseudo = text
    for r in sorted(results, key=lambda x: x.start, reverse=True):
        original = text[r.start:r.end]
        token = mapper.get_token(r.entity_type, original)
        text_pseudo = text_pseudo[:r.start] + token + text_pseudo[r.end:]
    
    return text_pseudo
 
 
# Usage
mapper = PIIMapper()
 
prompt_user = "Bonjour, je suis Marie Dupont (marie.dupont@example.com), pour ma commande #1234..."
prompt_safe = pseudonymize_for_llm(prompt_user, mapper)
# → "Bonjour, je suis [PERSON_001] ([EMAIL_ADDRESS_001]), pour ma commande #1234..."
 
response = await openai_client.chat.completions.create(
    messages=[{"role": "user", "content": prompt_safe}],
    model="gpt-4o",
)
 
answer_safe = response.choices[0].message.content
# La réponse contient les tokens [PERSON_001], [EMAIL_ADDRESS_001]
 
# Re-substituer pour le user final
answer_real = mapper.reverse(answer_safe)
# OpenAI n'a jamais vu marie.dupont@example.com, l'utilisateur final reçoit la version normale

import hmac
import hashlib
import os
 
TOKEN_KEY = os.environ["TOKEN_HMAC_KEY"].encode()
 
def tokenize_id(identifier: str) -> str:
    h = hmac.new(TOKEN_KEY, identifier.encode(), hashlib.sha256).hexdigest()
    return f"id_{h[:12]}"
 
def detokenize_id(token: str, mapping: dict) -> str:
    return mapping.get(token, token)
 
 
# Usage : remplacer les IDs avant envoi LLM
order_id = "ORD-2026-001234"
order_token = tokenize_id(order_id)  # "id_a1b2c3d4e5f6"
 
# Mapper local
mapping[order_token] = order_id
 
prompt = f"Concernant la commande {order_token}..."
# → OpenAI voit le token, pas l'ID réel

def minimize_for_llm(full_context: dict, request_intent: str) -> dict:
    """N'inclure que les infos strictement nécessaires."""
    
    if request_intent == "summarize_email":
        return {"email_body": full_context["email_body"]}
        # Pas d'historique, pas de métadonnées user, pas d'IDs
    
    if request_intent == "classify_ticket":
        return {"ticket_subject": full_context["ticket_subject"], 
                "ticket_first_message": full_context["ticket_first_message"][:500]}
    
    # Refuse si intent inconnu
    raise ValueError(f"Unknown intent: {request_intent}")

def split_and_filter_relevant(document: str, query: str) -> str:
    """Split doc en chunks, n'envoyer que les chunks pertinents pour la query."""
    chunks = split_into_chunks(document, max_size=500)
    
    # Embedding similarity locale (ne va pas chez OpenAI)
    relevant = filter_by_local_embedding(chunks, query, top_k=3)
    
    return "\n".join(relevant)

# Anti-pattern : exposer noms individuels
prompt = "Qui parmi mes clients John Doe, Jane Smith, Marie Dupont a un crédit > 10k€ ?"
 
# Pattern : exposer info agrégée seulement
prompt = "Classer ces 3 montants en risque/non-risque selon le seuil 10k€ : 5000, 15000, 8000."
# Résoudre le mapping clients-montants en local

# DPIA, Usage OpenAI API pour [Application]
 
## 1. Description du traitement
 
- Application : ZerodaySupport ChatBot SAV v2.4
- Cas d'usage : Réponses automatiques niveau 1 SAV, escalation humaine si complexe
- Volume estimé : ~50 000 conversations/mois
- Données traitées :
  * PII clients (nom, email, téléphone)
  * Numéros commande
  * Historique conversation
- Catégories acteurs :
  * Utilisateurs : clients ZerodaySupport (particuliers EU)
  * Personnel : équipe SAV (visualisation)
  * Sous-traitants : OpenAI (API tier Enterprise)
 
## 2. Nécessité et proportionnalité
 
### Pourquoi ce traitement ?
Réduction temps réponse SAV (-60%), disponibilité 24/7, réduction coût support.
 
### Alternatives évaluées
- Mistral EU : qualité légèrement inférieure pour notre cas (test 2026-Q1, 8/10 vs 10/10)
- Azure OpenAI EU : équivalent qualité, choisie comme alternative en cours d'évaluation
- On-premise : non retenu (CapEx, expertise, volume insuffisant)
 
### Justification choix OpenAI standard (transitoire)
- Migration Azure OpenAI EU planifiée Q3 2026
- Tier API Enterprise + DPA + opt-out training
- Volume actuel ne justifie pas on-premise
 
## 3. Risques identifiés
 
### Confidentialité
- Transit UE→US (Schrems II) : risque MOYEN, atténué par DPA + SCC
- Logs OpenAI 30j : risque FAIBLE post-mitigation pseudonymisation
- Insider OpenAI : risque RÉSIDUEL après mitigations
- Subpoena US : risque RÉSIDUEL FAIBLE (données pseudonymisées)
 
### Intégrité
- Prompt injection : risque MOYEN, mitigation guardrails (classifier + filter)
- Hallucinations : risque MOYEN, disclaimer + escalation humaine
 
### Disponibilité
- Panne OpenAI : risque MOYEN, plan failover Azure OpenAI
 
### Conformité
- RGPD Art. 32 : OK avec mitigations
- RGPD Art. 6 : base légale = legitimate interest (intérêt légitime)
- EU AI Act : non haut risque (Annex III non applicable au SAV générique)
 
## 4. Mitigations implémentées
 
- [x] Pseudonymisation upstream (Presidio), emails, noms, téléphones, IBAN remplacés
- [x] Minimisation : context limité au strict nécessaire
- [x] DPA Enterprise OpenAI signé (réf : DPA-OAI-2025-0042)
- [x] Opt-out training par défaut
- [x] Tier API Enterprise (pas plain ChatGPT)
- [x] Audit logs structurés avec PII redaction
- [x] Disclaimer chatbot IA dans UI
- [x] Right to erasure implémenté (purge logs par user_id_hash)
- [x] Retention 1 an logs sécurité, 90j logs informationnels
- [x] Plan migration Azure OpenAI EU Q3 2026
 
## 5. Risque résiduel
 
Niveau : ACCEPTABLE TRANSITOIREMENT (jusqu'à migration Azure EU).
 
Justification : mitigations couvrent confidentialité courante, transit US 
limité par pseudonymisation, migration imminente vers EU.
 
## 6. Validation
 
- DPO : Marie Lambert, signé 2026-04-15
- RSSI : Jean Bernard, signé 2026-04-15
- AI Officer : Claire Roux, signée 2026-04-15
- Métier : Pierre Martin (Head of Customer Support), signé 2026-04-15
 
## 7. Revue
 
- Cadence : annuelle minimum
- Déclencheurs ad hoc : changement OpenAI ToS, jurisprudence Schrems III, 
  changement classification donnée, incident
- Prochaine review : 2027-04-15
- Migration Azure EU prévue : Q3 2026 → DPIA refresh post-migration

def should_use_openai_standard(data_classification: str, content: str) -> dict:
    """Aide-décision rapide."""
    
    # Bloqueurs absolus
    if data_classification == "rgpd_article_9":
        return {"decision": "NO", "alternative": "Azure OpenAI EU + DPIA, ou on-premise"}
    
    if contains_secrets(content):  # API keys, passwords
        return {"decision": "NO", "alternative": "Strip secrets first, never send"}
    
    if data_classification == "secret_defense":
        return {"decision": "NO", "alternative": "On-premise air-gap obligatoire"}
    
    if data_classification == "confidential" and not has_pseudonymization(content):
        return {"decision": "MAYBE", "condition": "Pseudonymize first, then OK with Enterprise tier"}
    
    # Acceptable
    if data_classification in {"public", "internal"}:
        return {"decision": "YES", "tier": "OpenAI Enterprise"}
    
    # Default : besoin DPIA
    return {"decision": "DPIA_REQUIRED", "next": "Run DPIA before decision"}