Architecture RAG sécurisée : segmentation, accès, traçabilité

# Pattern d'isolation tenant strict (Pinecone-like)
class TenantScopedRAG:
    def __init__(self, vector_db, tenant_id: str):
        self.tenant_id = tenant_id
        # Connexion préfixée — impossible d'oublier le tenant
        self.namespace = f"tenant_{tenant_id}"
    
    def query(self, query_text: str, user_context: dict, top_k: int = 5):
        # Le namespace est injecté côté API. Pas de override possible côté client.
        return self.vector_db.query(
            namespace=self.namespace,
            vector=embed(query_text),
            filter=self._build_user_filter(user_context),
            top_k=top_k,
        )
    
    def _build_user_filter(self, ctx: dict) -> dict:
        # ACL filter additionnel au-dessus de la segmentation tenant
        return {
            "$or": [
                {"acl_public": True},
                {"acl_users": {"$in": [ctx["user_id"]]}},
                {"acl_groups": {"$in": ctx["groups"]}},
            ]
        }

# Politique de sensibilité (exemple)
sensitivity_levels:
  - name: public
    score: 0
    description: "Doc accessible à tous"
  - name: internal
    score: 10
    description: "Doc interne entreprise"
  - name: confidential
    score: 50
    description: "Doc confidentiel — RH, Finance"
  - name: restricted
    score: 90
    description: "Doc restreint — direction, audit"
 
user_clearance:
  default: 10  # internal
  hr_admin: 50
  cfo: 90
  external: 0

def get_user_clearance(user) -> int:
    return max([CLEARANCE_BY_ROLE.get(r, 0) for r in user.roles] + [0])
 
results = vdb.query(
    namespace=tenant_namespace,
    vector=embedding,
    filter={
        "sensitivity_score": {"$lte": get_user_clearance(current_user)},
        # ... autres ACLs
    },
    top_k=10,
)

# Exemple : domaine encodé en metadata
chunk_metadata = {
    "tenant_id": "company_a",
    "domain": "hr",         # finance, tech, legal, sales, etc.
    "sensitivity": "internal",
    "owner_user_id": "u_42",
    "owner_groups": ["hr_team"],
    # ...
}
 
# Filter retrieval domaine + sensibilité + ACL
filter_query = {
    "domain": {"$in": current_user.allowed_domains},
    "sensitivity_score": {"$lte": current_user.clearance},
    # ... acl
}

def build_retrieval_filter(user, query_context):
    """Construit un filter retrieval combinant tous les axes."""
    return {
        "tenant_id": user.tenant_id,                                  # axe 1
        "sensitivity_score": {"$lte": user.clearance_score},          # axe 2
        "domain": {"$in": user.allowed_domains},                      # axe 3
        "region": {"$in": user.allowed_regions},                      # axe 4 si applicable
        "$or": [                                                       # ACL fine
            {"acl_public": True},
            {"acl_users": {"$in": [user.id]}},
            {"acl_groups": {"$in": user.groups}},
        ],
    }

# Exemple Cedar-like policy
permit (
    principal in Group::"hr_team",
    action == Action::"read_document",
    resource
)
when {
    resource.tenant_id == principal.tenant_id &&
    resource.domain == "hr" &&
    resource.sensitivity_score <= principal.clearance_score &&
    (resource.acl_public ||
     principal in resource.acl_groups ||
     principal == resource.owner)
};

def index_document_with_acl(doc, vector_db):
    """Indexe un document en propageant ses ACL source au niveau chunk."""
    source_acl = get_native_acl(doc.source_id, doc.source_type)
    
    chunks = chunk_document(doc.content)
    
    for chunk in chunks:
        embedding = embed(chunk.text)
        vector_db.upsert(
            id=f"{doc.id}_chunk_{chunk.idx}",
            vector=embedding,
            namespace=f"tenant_{doc.tenant_id}",
            metadata={
                "doc_id": doc.id,
                "tenant_id": doc.tenant_id,
                "domain": doc.domain,
                "sensitivity": doc.sensitivity,
                "sensitivity_score": SENSITIVITY_SCORES[doc.sensitivity],
                "acl_public": source_acl.is_public,
                "acl_users": source_acl.users,
                "acl_groups": source_acl.groups,
                "owner": source_acl.owner,
                "indexed_at": datetime.utcnow().isoformat(),
                "acl_version": source_acl.version,  # pour invalidation
            },
        )

def secure_retrieve(query: str, user) -> list[Chunk]:
    # 1. Pre-filter au niveau VDB
    candidates = vector_db.query(
        namespace=f"tenant_{user.tenant_id}",
        vector=embed(query),
        filter=build_retrieval_filter(user),
        top_k=20,  # un peu plus large pour absorber le post-filter
    )
    
    # 2. Post-filter contre IAM source-of-truth
    verified = []
    for c in candidates:
        if iam_service.can_access(
            user=user,
            resource_id=c.metadata["doc_id"],
            action="read",
        ):
            verified.append(c)
        else:
            log_event("acl_drift_detected", doc_id=c.metadata["doc_id"], user=user.id)
    
    return verified[:5]  # top-5 après post-filter

def fetch_chunk_content(chunk_id: str, user) -> str:
    # Émettre un token éphémère scopé à ce chunk + cet utilisateur
    eph_token = iam_service.issue_ephemeral_token(
        subject=user.id,
        resource=chunk_id,
        action="read",
        ttl_seconds=30,
    )
    return content_store.fetch(chunk_id, auth_token=eph_token)

{
  "timestamp": "2026-04-30T11:23:45.123Z",
  "request_id": "req_8a92c4d1",
  "session_id": "sess_91fe...",
  "user": {
    "id": "u_42",
    "tenant_id": "company_a",
    "ip": "203.0.113.10",
    "auth_method": "oidc"
  },
  "rag": {
    "query_text_hash": "sha256:abc...",
    "embedding_model": "text-embedding-3-large",
    "retrieved_chunks": [
      {"id": "doc1_chunk_3", "score": 0.89, "sensitivity": "internal"},
      {"id": "doc2_chunk_1", "score": 0.85, "sensitivity": "internal"}
    ],
    "filter_applied": {
      "tenant_id": "company_a",
      "sensitivity_max": 10,
      "domains": ["hr"]
    }
  },
  "llm": {
    "model": "claude-sonnet-4-6",
    "input_tokens": 1234,
    "output_tokens": 287,
    "response_hash": "sha256:def..."
  },
  "security": {
    "input_score": 0.05,
    "output_score": 0.0,
    "canary_leak": false,
    "action_taken": "allowed"
  },
  "duration_ms": 412
}

chunk_audit_trail = {
    "chunk_id": "doc1_chunk_3",
    "events": [
        {"type": "indexed", "doc_id": "doc1", "by": "ingestion_pipeline", "at": "2026-04-15T..."},
        {"type": "acl_updated", "by": "iam_sync", "at": "2026-04-20T..."},
        {"type": "retrieved", "session_id": "sess_91fe", "user": "u_42", "at": "2026-04-30T..."},
        {"type": "flagged", "reason": "outlier_query_match", "at": "2026-04-29T..."},
    ]
}

┌─────────────────────────────────────────────────────────────┐
│                    Authentification (OIDC / SAML)            │
│                    + Service IAM (OpenFGA / Cedar)           │
└─────────────────────────────────┬───────────────────────────┘
                                  │ user context
┌─────────────────────────────────▼───────────────────────────┐
│ API Gateway                                                  │
│  - Rate limiting                                             │
│  - Detection extraction (volume, diversité)                  │
│  - Logging structuré OTel                                    │
└─────────────────────────────────┬───────────────────────────┘
                                  │
┌─────────────────────────────────▼───────────────────────────┐
│ RAG Orchestrator                                             │
│  - Build filter (tenant + sensibilité + domaine + ACL)       │
│  - Pre-filter VDB                                            │
│  - Similarity search                                         │
│  - Post-filter via IAM (re-vérification)                     │
│  - Sanitization chunks                                       │
└──────────┬──────────────────────────────────┬──────────────┘
           │                                  │
┌──────────▼─────────┐              ┌─────────▼─────────────┐
│ Vector DB          │              │ IAM service           │
│ - Index par tenant │              │ - ABAC policies       │
│ - Encrypt at-rest  │              │ - Ephemeral tokens    │
│ - Audit logs natif │              │ - Audit logs          │
└────────────────────┘              └───────────────────────┘
           │                                  │
┌──────────▼──────────────────────────────────▼─────────────┐
│ LLM avec system prompt durci                              │
│ + Output filter (DLP, canary, allowlist domaines)         │
└──────────┬────────────────────────────────────────────────┘
           │
┌──────────▼────────────────────────────────────────────────┐
│ UI sanitization (DOMPurify, CSP, no markdown image)        │
└──────────┬────────────────────────────────────────────────┘
           │
┌──────────▼────────────────────────────────────────────────┐
│ Telemetry → OTel collector → SIEM (Splunk/Sentinel/Elastic)│
└────────────────────────────────────────────────────────────┘

def audit_segmentation(rag_app):
    issues = []
    
    # Test cross-tenant
    canary = "TENANT_A_CANARY_3f4d92"
    rag_app.index_for_tenant("tenant_a", document=canary)
    for resp in rag_app.query_as_tenant("tenant_b", queries=SAMPLE_QUERIES):
        if canary in resp:
            issues.append({"type": "cross_tenant_leak"})
    
    # Test sensibilité
    classified_doc = "RESTRICTED-INFO-ONLY-LEVEL-90"
    rag_app.index_doc(classified_doc, sensitivity_score=90)
    for resp in rag_app.query_as_user(user_clearance=10, queries=...):
        if classified_doc in resp:
            issues.append({"type": "sensitivity_leak"})
    
    return issues

def audit_access_control(rag_app, iam):
    # Test ACL propagation : créer doc avec ACL stricte, requérir depuis user sans droit
    doc_id = rag_app.index_doc(content="...", acl_users=["u_42"])
    resp = rag_app.query_as_user("u_other", "give me content of " + doc_id)
    assert doc_id not in resp
    
    # Test ACL drift : doc avec ACL périmée
    iam.revoke(user="u_42", doc=doc_id)
    resp = rag_app.query_as_user("u_42", ...)
    assert "content of doc" not in resp  # post-filter doit attraper

def audit_traceability(rag_app, siem):
    # Tester complétude des logs
    rag_app.query_as_user("u_42", "test query")
    logs = siem.fetch_recent(user="u_42", limit=1)
    assert logs[0].user.id == "u_42"
    assert logs[0].rag.retrieved_chunks  # liste présente
    assert logs[0].llm.model            # info LLM présente
    assert logs[0].security.action_taken
    
    # Tester reproductibilité d'un incident
    incident_id = "inc_test_001"
    rag_app.simulate_incident(incident_id)
    reconstructed = siem.reconstruct_session(incident_id)
    assert reconstructed.is_complete