🔒 AI 安全合规 / 数据隔离
🔐 数据隔离
防止用户 A 看到用户 B 的数据——多租户 AI 产品的基础安全要求。2023 年 ChatGPT 的缓存 bug 泄露对话就是数据隔离失败的典型案例。
数据隔离的三个维度
数据隔离维度
│
┌───────────┼───────────┐
│ │ │
租户隔离 会话隔离 时序隔离
(Tenant) (Session) (Temporal)
│ │ │
┌───────┤ ┌─────┤ ┌─────┤
│ │ │ │ │ │
存储 计算 对话 上下文 冷数据 热数据
隔离 隔离 历史 窗口 加密 缓存
租户隔离策略
| 策略 | 描述 | 隔离强度 | 成本 | 适用规模 |
|---|
| 共享数据库 + tenant_id | 所有租户共享表，用 tenant_id 过滤 | 🟡 中 | 低 | 小规模/内部工具 |
| Schema 隔离 | 同数据库，每个租户独立 schema | 🟢 较高 | 中 | 中等规模 |
| 数据库隔离 | 每个租户独立数据库实例 | 🔴 高 | 高 | 企业级/B2B |
| 物理隔离 | 不同服务器/区域 | 🔴 极高 | 极高 | 金融/医疗 |
方案1: 共享数据库 + tenant_id（最常见）
from sqlalchemy import Column, String, Integer, Text, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from contextlib import contextmanager
import uuid
Base = declarative_base()
class Conversation(Base):
"""对话记录表——包含 tenant_id"""
__tablename__ = 'conversations'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
tenant_id = Column(String(36), nullable=False, index=True) # 必须有索引！
user_id = Column(String(36), nullable=False, index=True)
title = Column(String(200))
created_at = Column(Integer, nullable=False)
class Message(Base):
"""消息表"""
__tablename__ = 'messages'
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
tenant_id = Column(String(36), nullable=False, index=True)
conversation_id = Column(String(36), nullable=False, index=True)
role = Column(String(20), nullable=False)
content = Column(Text, nullable=False)
created_at = Column(Integer, nullable=False)
# 关键：所有查询必须带 tenant_id 条件
class TenantAwareQuery:
"""租户感知查询——强制添加 tenant_id 过滤"""
def __init__(self, session_factory, tenant_id: str):
self.session_factory = session_factory
self.tenant_id = tenant_id
def get_conversations(self, user_id: str, limit: int = 20):
session = self.session_factory()
try:
return session.query(Conversation).filter(
Conversation.tenant_id == self.tenant_id, # 强制过滤
Conversation.user_id == user_id
).order_by(Conversation.created_at.desc()).limit(limit).all()
finally:
session.close()
def get_messages(self, conversation_id: str):
session = self.session_factory()
try:
# 双重检查：conversation_id + tenant_id
return session.query(Message).filter(
Message.tenant_id == self.tenant_id,
Message.conversation_id == conversation_id
).order_by(Message.created_at).all()
finally:
session.close()
# 使用
def get_tenant_query(tenant_id: str) -> TenantAwareQuery:
return TenantAwareQuery(SessionFactory, tenant_id)
# 中间件：从 JWT 提取 tenant_id
def auth_middleware(request):
token = request.headers.get('Authorization', '').replace('Bearer ', '')
payload = verify_jwt(token)
request.state.tenant_id = payload['tenant_id']
request.state.user_id = payload['user_id']
⚠️ 共享数据库最大的风险：忘记在查询中加 tenant_id 过滤。这是一个看似简单但极容易犯的错误。建议在 ORM 层强制注入 tenant_id，而不是依赖开发者每次都记得加。
方案2: Row-Level Security (PostgreSQL RLS)
-- PostgreSQL Row-Level Security：数据库层面的租户隔离
-- 即使应用层忘记过滤，数据库也会阻止跨租户访问
-- 1. 启用 RLS
ALTER TABLE conversations ENABLE ROW LEVEL SECURITY;
ALTER TABLE messages ENABLE ROW LEVEL SECURITY;
-- 2. 创建策略：只能看到自己租户的数据
CREATE POLICY tenant_isolation ON conversations
USING (tenant_id = current_setting('app.current_tenant')::text);
CREATE POLICY tenant_isolation ON messages
USING (tenant_id = current_setting('app.current_tenant')::text);
-- 3. 应用层设置当前租户
-- 在每个请求开始时：
-- SET app.current_tenant = 'tenant_123';
-- 之后所有查询自动过滤
-- Python 集成
"""
from sqlalchemy import text
@app.middleware("http")
async def set_tenant_context(request, call_next):
tenant_id = request.state.tenant_id
# 设置 PostgreSQL 会话变量
with engine.connect() as conn:
conn.execute(text(f"SET app.current_tenant = '{tenant_id}'"))
conn.commit()
response = await call_next(request)
return response
"""
对话历史加密
import os
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
import base64
class ConversationEncryptor:
"""对话历史加密器"""
def __init__(self, master_key: str = None):
"""使用主密钥派生加密密钥"""
if master_key is None:
master_key = os.environ.get('ENCRYPTION_MASTER_KEY')
if not master_key:
raise ValueError("必须提供加密主密钥")
# 使用 PBKDF2 派生密钥
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b'openclaw-conversation-salt', # 生产环境应该每租户不同
iterations=480000,
)
key = base64.urlsafe_b64encode(kdf.derive(master_key.encode()))
self.cipher = Fernet(key)
def encrypt(self, plaintext: str) -> str:
"""加密对话内容"""
return self.cipher.encrypt(plaintext.encode()).decode()
def decrypt(self, ciphertext: str) -> str:
"""解密对话内容"""
return self.cipher.decrypt(ciphertext.encode()).decode()
def encrypt_message(self, role: str, content: str) -> dict:
"""加密单条消息"""
return {
'role': role, # role 不加密（需要用于 API 调用）
'content_encrypted': self.encrypt(content),
'content_hash': hashlib.sha256(content.encode()).hexdigest()[:16]
}
# 使用
encryptor = ConversationEncryptor()
encrypted = encryptor.encrypt("用户输入的敏感内容")
decrypted = encryptor.decrypt(encrypted)
# 存储时加密
msg = encryptor.encrypt_message("user", "我的银行卡号是6222...")
# {'role': 'user', 'content_encrypted': 'gAAAA...', 'content_hash': 'a1b2c3d4e5f6g7h8'}
# 读取时解密
original = encryptor.decrypt(msg['content_encrypted'])
日志脱敏
import re
import json
import logging
from typing import Any
class SanitizingFormatter(logging.Formatter):
"""日志脱敏格式化器"""
PII_PATTERNS = {
'email': (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
'phone_cn': (r'1[3-9]\d{9}', '[PHONE]'),
'id_card': (r'\d{17}[\dXx]', '[ID_CARD]'),
'ip': (r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]'),
'api_key': (r'sk-[a-zA-Z0-9]{8,}[a-zA-Z0-9]*', 'sk-[REDACTED]'),
'credit_card': (r'\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}', '[CARD]'),
}
def format(self, record: logging.LogRecord) -> str:
# 先格式化
result = super().format(record)
# 再脱敏
for pii_type, (pattern, replacement) in self.PII_PATTERNS.items():
result = re.sub(pattern, replacement, result)
return result
class SanitizedLogger:
"""安全日志记录器"""
def __init__(self, name: str):
self.logger = logging.getLogger(name)
handler = logging.StreamHandler()
handler.setFormatter(SanitizingFormatter(
'%(asctime)s [%(levelname)s] %(message)s'
))
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def log_request(self, tenant_id: str, user_id: str,
model: str, token_count: int, **kwargs):
"""记录 API 请求日志（不含用户输入内容）"""
self.logger.info(json.dumps({
'event': 'llm_request',
'tenant_id': tenant_id[:8] + '***', # 部分脱敏
'user_id': user_id[:8] + '***',
'model': model,
'token_count': token_count,
# 注意：不记录用户输入和模型输出内容
}))
def log_error(self, error: str, tenant_id: str = None, **kwargs):
"""记录错误日志"""
safe_kwargs = {k: self._sanitize(v) for k, v in kwargs.items()}
self.logger.error(json.dumps({
'event': 'error',
'error': error,
'tenant_id': tenant_id[:8] + '***' if tenant_id else None,
**safe_kwargs
}))
def _sanitize(self, value: Any) -> Any:
"""通用脱敏"""
if isinstance(value, str):
for pattern, replacement in SanitizingFormatter.PII_PATTERNS.items():
value = re.sub(pattern[0], pattern[1], value) # pattern is (regex, replacement)
return value
return value
# 使用
logger = SanitizedLogger('ai_service')
logger.log_request(
tenant_id="tenant_abc123",
user_id="user_xyz789",
model="gpt-4o",
token_count=1500
)
# 输出: {"event":"llm_request","tenant_id":"tenant_a***","user_id":"user_xy***","model":"gpt-4o","token_count":1500}
向量数据库的租户隔离
💡 特别注意：向量数据库（Pinecone、Weaviate、Qdrant、Milvus）的隔离是 AI 产品独有的挑战。如果用户 A 的文档 Embedding 和用户 B 的混在一起，检索时可能返回错误租户的数据。
from dataclasses import dataclass
from typing import Optional
# 方案1: Metadata 过滤（Pinecone / Qdrant / Weaviate 都支持）
class TenantAwareVectorStore:
"""租户感知的向量存储"""
def __init__(self, client, index_name: str):
self.client = client
self.index = client.Index(index_name)
def upsert(self, tenant_id: str, doc_id: str,
embedding: list[float], metadata: dict = None):
"""插入向量——强制附加 tenant_id"""
vector = {
'id': f"{tenant_id}_{doc_id}", # 命名空间隔离
'values': embedding,
'metadata': {
**(metadata or {}),
'tenant_id': tenant_id # 强制设置 tenant_id
}
}
self.index.upsert(vectors=[vector])
def query(self, tenant_id: str, query_embedding: list[float],
top_k: int = 5, filter: dict = None):
"""查询向量——强制过滤 tenant_id"""
# 构建过滤条件
tenant_filter = {'tenant_id': {'$eq': tenant_id}}
if filter:
# 合并用户自定义过滤
tenant_filter = {'$and': [tenant_filter, filter]}
results = self.index.query(
vector=query_embedding,
top_k=top_k,
filter=tenant_filter, # 强制租户过滤！
include_metadata=True
)
return results
# Pinecone 具体实现
"""
import pinecone
pc = pinecone.Pinecone(api_key="your-key")
index = pc.Index("documents")
store = TenantAwareVectorStore(pc, "documents")
# 插入——自动附加 tenant_id
store.upsert(
tenant_id="tenant_abc",
doc_id="doc_123",
embedding=[0.1, 0.2, ...],
metadata={"source": "upload", "title": "合同文件"}
)
# 查询——自动过滤 tenant_id
results = store.query(
tenant_id="tenant_abc",
query_embedding=[0.15, 0.22, ...],
top_k=5
)
# 只返回 tenant_abc 的数据，不会泄露其他租户的
"""
# 方案2: 独立命名空间/Collection（更强的隔离）
class NamespacedVectorStore:
"""使用命名空间实现强隔离"""
def __init__(self, client):
self.client = client
def _get_namespace(self, tenant_id: str) -> str:
"""每个租户独立命名空间"""
return f"tenant_{tenant_id}"
def upsert(self, tenant_id: str, vectors: list[dict]):
"""在租户命名空间中插入"""
namespace = self._get_namespace(tenant_id)
index = self.client.Index("documents")
index.upsert(vectors=vectors, namespace=namespace)
def query(self, tenant_id: str, query_embedding: list[float], top_k: int = 5):
"""在租户命名空间中查询"""
namespace = self._get_namespace(tenant_id)
index = self.client.Index("documents")
return index.query(
vector=query_embedding,
top_k=top_k,
namespace=namespace # 只搜索该命名空间
)
# 方案3: 独立索引/Collection（最强隔离）
"""
# Qdrant: 每个 Collection 独立
from qdrant_client import QdrantClient
client = QdrantClient(url="http://localhost:6333")
def get_or_create_collection(tenant_id: str):
collection_name = f"docs_{tenant_id}"
try:
client.get_collection(collection_name)
except:
client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)
return collection_name
"""
| 向量库隔离方案 | 隔离强度 | 查询性能 | 成本 | 适用场景 |
|---|
| Metadata 过滤 | 🟡 中 | 略慢（需过滤） | 低 | 大多数 SaaS |
| 命名空间隔离 | 🟢 高 | 正常 | 中 | B2B 企业级 |
| 独立 Collection | 🔴 极高 | 正常 | 高 | 金融/医疗 |
数据隔离检查清单
## 数据隔离检查清单
### 存储层
[ ] 所有数据表包含 tenant_id 字段
[ ] tenant_id 字段有索引
[ ] 所有查询强制带 tenant_id 过滤（ORM 层或 RLS）
[ ] 考虑 PostgreSQL Row-Level Security
[ ] 对话历史加密存储
[ ] 加密密钥与租户绑定（而非全局共享）
[ ] 备份也按租户隔离
### 计算层
[ ] LLM 调用上下文不跨租户
[ ] Agent 工具调用范围限制在租户内
[ ] 缓存键包含 tenant_id（防止缓存混淆）
[ ] 临时文件/沙箱按租户隔离
### 日志层
[ ] 日志中不含用户原始输入/输出
[ ] PII 自动脱敏
[ ] tenant_id 部分脱敏
[ ] API Key 不出现在日志中
[ ] 日志访问有权限控制
### 向量数据库
[ ] 所有查询强制 metadata 过滤 tenant_id
[ ] 或使用命名空间/Collection 隔离
[ ] Embedding 元数据包含 tenant_id
[ ] 检索结果二次验证 tenant_id
[ ] 向量数据导出按租户隔离
### API 层
[ ] JWT/Token 包含 tenant_id
[ ] API 端点自动注入 tenant_id
[ ] 不允许客户端指定 tenant_id（从 Token 提取）
[ ] 批量操作限制在单个租户内
### 测试
[ ] 跨租户访问测试（A 不能访问 B 的数据）
[ ] IDOR 漏洞测试
[ ] 缓存混淆测试
[ ] 向量检索隔离测试
← 上一篇：输出过滤 | 下一篇：GDPR 合规 → | 返回首页