Chapter 3: Security and Access Control
Security Concerns in RAG
| Concern | Risk | Mitigation |
|---|---|---|
| Unauthorized access | Data leak | Access control |
| Prompt injection | Manipulation | Input validation |
| Data isolation | Cross-tenant | Namespaces |
Document-Level Permissions
def search_with_permissions(query, user):
"""Filter results based on user permissions"""
# Get user's allowed document IDs
allowed_docs = get_user_permissions(user.id)
# Search with filter
results = vectorstore.similarity_search(
query,
filter={"doc_id": {"$in": allowed_docs}},
k=10
)
return results
Multi-Tenant Architecture
Option 1: Namespace Isolation
# Each tenant gets a namespace
def get_tenant_retriever(tenant_id):
return vectorstore.as_retriever(
search_kwargs={
"namespace": f"tenant_{tenant_id}",
"k": 5
}
)
Option 2: Metadata Filtering
# Store tenant_id in metadata
def add_document(text, tenant_id):
vectorstore.add_texts(
[text],
metadatas=[{"tenant_id": tenant_id}]
)
def query_tenant(query, tenant_id):
return vectorstore.similarity_search(
query,
filter={"tenant_id": tenant_id}
)
Input Validation
def sanitize_query(query):
"""Prevent prompt injection"""
# Remove potential injection patterns
dangerous_patterns = [
"ignore previous instructions",
"disregard the above",
"system:"
]
for pattern in dangerous_patterns:
if pattern.lower() in query.lower():
raise ValueError("Invalid query")
return query
Audit Logging
def log_rag_access(user_id, query, docs_accessed, response):
"""Log all RAG interactions"""
log_entry = {
"timestamp": datetime.utcnow(),
"user_id": user_id,
"query": query,
"documents_accessed": [d.id for d in docs_accessed],
"response_length": len(response)
}
audit_logger.info(json.dumps(log_entry))
Security Checklist
- Implement authentication
- Enforce document permissions
- Isolate tenant data
- Validate/sanitize inputs
- Log all access
- Encrypt sensitive data
- Regular security audits