Skip to main content

Chapter 3: Security and Access Control

Security Concerns in RAG​

ConcernRiskMitigation
Unauthorized accessData leakAccess control
Prompt injectionManipulationInput validation
Data isolationCross-tenantNamespaces

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