Sécurité IA pour architectes : SI résistant aux usages non maîtrisés

# Avec LiteLLM : un seul SDK pour tous fournisseurs
from litellm import completion
 
# Le gateway route selon configuration
response = completion(
    model="gpt-4o",  # ou "claude-sonnet-4-6", "gemini-2-pro", "mistral-large", "llama-3-70b" (self-hosted)
    messages=[{"role": "user", "content": "..."}],
    fallbacks=["claude-sonnet-4-6", "gemini-2-pro"],  # fallback automatique
)

┌──────────────────────────────────────────────────────────────────┐
│                  Utilisateurs / Applications                       │
└──────────────────────────────────┬───────────────────────────────┘
                                   │
┌──────────────────────────────────▼───────────────────────────────┐
│  AI Gateway (LiteLLM / Portkey / Cloudflare AI Gateway)          │
│  - Auth (OIDC/JWT)                                                │
│  - Rate limiting + cost tracking                                  │
│  - Routage multi-modèles                                          │
│  - Fallback + circuit breaker                                     │
│  - Input filter (LLM Guard, Lakera)                              │
└──────────────────┬───────────────────────────────────────────────┘
                   │
       ┌───────────┴──────────────┬─────────────────┐
       │                          │                 │
┌──────▼────────┐       ┌─────────▼────┐    ┌──────▼─────────┐
│ LLM API       │       │ LLM API       │    │ Self-hosted    │
│ externes      │       │ régions UE    │    │ (vLLM/TGI/     │
│ (OpenAI, etc.)│       │ (Azure OpenAI)│    │  Ollama)       │
└───────────────┘       └───────────────┘    └────────────────┘
 
┌──────────────────────────────────────────────────────────────────┐
│  Vector DB partagée (Pinecone / Weaviate / Qdrant / pgvector)    │
│  - Multi-tenancy native                                           │
│  - ACL propagées (ABAC commun)                                    │
│  - Audit logs                                                     │
└──────────────────────────────────────────────────────────────────┘
 
┌──────────────────────────────────────────────────────────────────┐
│  IAM / ABAC service (OpenFGA / Cedar / Permit.io)                 │
│  - Politiques fine-grained                                        │
│  - Tokens éphémères                                               │
│  - Source-of-truth pour permissions                               │
└──────────────────────────────────────────────────────────────────┘
 
┌──────────────────────────────────────────────────────────────────┐
│  Observabilité centralisée                                        │
│  Langfuse (LLM-spécifique) + OTel collector + SIEM                │
│  - Logs OpenTelemetry GenAI semantic conventions                  │
│  - Métriques cost/latency/quality                                 │
│  - Alertes SOC                                                    │
└──────────────────────────────────────────────────────────────────┘
 
┌──────────────────────────────────────────────────────────────────┐
│  GitOps (ArgoCD / Flux)                                           │
│  - System prompts versionnés                                      │
│  - Configurations agents                                          │
│  - ABAC policies                                                  │
│  - AI BOM                                                         │
└──────────────────────────────────────────────────────────────────┘

# litellm-config.yaml
model_list:
  - model_name: "smart-fallback"
    litellm_params:
      model: gpt-4o
      api_key: os.environ/OPENAI_API_KEY
    fallbacks: ["claude-sonnet-4-6", "gemini-2-pro"]
  
  - model_name: "claude-sonnet-4-6"
    litellm_params:
      model: anthropic/claude-sonnet-4-6
      api_key: os.environ/ANTHROPIC_API_KEY
  
  - model_name: "gemini-2-pro"
    litellm_params:
      model: vertex_ai/gemini-2-pro
      vertex_project: yourcompany-project
      vertex_location: europe-west1  # région UE explicite
 
router_settings:
  routing_strategy: "least-busy"
  fallbacks: 
    - model: "smart-fallback"
      max_attempts: 3
 
guardrails:
  - guardrail_name: "lakera_pre_call"
    litellm_params:
      guardrail: lakera_v2
      mode: pre_call
      api_key: os.environ/LAKERA_API_KEY
  - guardrail_name: "presidio_pii_post"
    litellm_params:
      guardrail: presidio
      mode: post_call
      output_parse_pii: true
 
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL
  cache: true
  cache_params:
    type: redis
    host: redis.internal

# Architecture retrieval avec ABAC commun
def secure_retrieve(query: str, user_context: dict) -> list[Chunk]:
    # 1. Filter au niveau VDB (filtre dur)
    candidates = vector_db.query(
        namespace=f"tenant_{user_context['tenant_id']}",
        vector=embed(query),
        filter={
            "tenant_id": user_context['tenant_id'],
            "sensitivity_score": {"$lte": user_context['clearance_score']},
            "$or": [
                {"acl_public": True},
                {"acl_users": {"$in": [user_context['user_id']]}},
                {"acl_groups": {"$in": user_context['groups']}},
            ],
        },
        top_k=20,
    )
    
    # 2. Post-filter via ABAC service (source-of-truth)
    verified = []
    for c in candidates:
        decision = abac_service.check(
            principal=user_context['user_id'],
            action="read",
            resource=c.metadata["doc_id"],
            context={
                "tenant": user_context['tenant_id'],
                "time": datetime.utcnow(),
                "ip": user_context['ip'],
            }
        )
        if decision.allowed:
            verified.append(c)
        else:
            log_event("acl_drift_detected", doc_id=c.metadata["doc_id"], user=user_context['user_id'])
    
    return verified[:5]

# Pour agents qui exécutent du code généré : sandbox kernel-level
apiVersion: node.k8s.io/v1
kind: RuntimeClass
metadata:
  name: kata-fc
handler: kata-fc  # Firecracker via Kata
---
apiVersion: v1
kind: Pod
metadata:
  name: code-executor
  namespace: ai-platform
spec:
  runtimeClassName: kata-fc
  securityContext:
    runAsNonRoot: true
    readOnlyRootFilesystem: true
  containers:
    - name: executor
      image: registry.internal/code-executor:audited
      resources:
        limits:
          cpu: "1"
          memory: "512Mi"
      env:
        # Pas de secrets ici, ephemeral via mediator
        - name: REQUEST_ID
          valueFrom:
            fieldRef:
              fieldPath: metadata.annotations['request-id']

Application LLM (LangChain, LlamaIndex, custom)
        │
        ▼ OTel SDK avec GenAI semantic conventions
┌──────────────────────────┐
│ Langfuse (LLM-spécifique)│ ─── Visualisation LLM-aware
└─────────────┬────────────┘
              │ OTel export
              ▼
┌──────────────────────────┐
│ OTel Collector central   │
└─────────────┬────────────┘
              │
   ┌──────────┴──────────┐
   │                     │
   ▼                     ▼
┌──────────┐       ┌──────────┐
│ SIEM     │       │ APM      │
│ Splunk/  │       │ Datadog/ │
│ Sentinel │       │ Grafana  │
└──────────┘       └──────────┘

                  ┌──────────────────────────┐
                  │  Plateforme IA interne    │
                  │  - UX comparable          │
                  │  - Performance comparable │
                  │  - Conformité gérée       │
                  │  - Logging transparent    │
                  │  - Cost tracking          │
                  │  - Multi-modèles          │
                  └──────────┬───────────────┘
                             │
                ┌────────────┼────────────┐
                │            │            │
        Employés ─────► Plateforme  ◄──── DPO/RSSI
        utilisateurs    (gouvernance     (visibilité)
                         intégrée)
                             │
                             │ logs
                             ▼
                  ┌──────────────────────────┐
                  │ Observabilité + SIEM      │
                  │ + Alertes shadow AI       │
                  │ détecté hors plateforme   │
                  └──────────────────────────┘

# Pattern : dégradation gracieuse
class ResilientLLMService:
    def __init__(self):
        self.primary = OpenAIClient()
        self.fallback = AnthropicClient()
        self.cache = RedisCache(ttl=3600)
        self.circuit_breaker = CircuitBreaker(failure_threshold=5)
    
    async def query(self, prompt: str) -> str:
        # 1. Cache hit ?
        cached = await self.cache.get(prompt)
        if cached:
            return cached
        
        # 2. Circuit breaker ?
        if self.circuit_breaker.is_open():
            return self.degraded_response(prompt)  # mode FAQ statique, etc.
        
        # 3. Primary
        try:
            response = await self.primary.complete(prompt, timeout=15)
            await self.cache.set(prompt, response)
            return response
        except (TimeoutError, ServiceUnavailable):
            self.circuit_breaker.record_failure()
        
        # 4. Fallback
        try:
            return await self.fallback.complete(prompt, timeout=15)
        except Exception:
            return self.degraded_response(prompt)
    
    def degraded_response(self, prompt: str) -> str:
        # Mode dégradé : recherche FAQ statique, suggestion contact humain
        return search_faq(prompt) or "Service temporairement indisponible. Contactez le support."

# Pipeline ETL classique étendu IA
@dag(schedule_interval="@daily")
def rag_corpus_pipeline():
    # ETL classique
    raw_documents = extract_from_warehouse(...)
    cleaned = transform_documents(raw_documents)
    
    # Extension IA : sanitization + embedding
    sanitized = sanitize_for_llm(cleaned)  # strip control chars, instruction markers
    chunks = chunk_documents(sanitized)
    embeddings = generate_embeddings(chunks, model="text-embedding-3-large")
    
    # Indexation vector DB avec metadata complète
    upsert_to_vector_db(
        chunks_with_embeddings=embeddings,
        metadata={
            "source_doc_id": ...,
            "tenant_id": ...,
            "sensitivity_score": ...,
            "acl_users": ...,
            "acl_groups": ...,
            "ingested_at": datetime.utcnow(),
            "lineage_id": ...,  # lien vers data lineage
        },
    )
    
    # Logs vers même OTel collector
    record_otel_event("rag_pipeline_completed", chunks=len(embeddings))