第 8 课：多租户 + RBAC / ACL —— 企业级 RAG 的"围墙"

本节目标：理解认证/RBAC/ACL 三层安全模型、掌握文档级权限控制、两阶段 ACL 过滤设计、以及权限粒度与成本的权衡。

前 7 课假设"所有用户看到所有文档"。企业不行：HR 政策只给 HR 看，财务数据只给财务看，A 公司知识库不能让 B 公司搜到。

1. 三层安全模型

Code

认证（Authentication）    "你是谁？"  → JWT → User 对象
        ↓
授权（RBAC）              "你能做什么？" → Role → Permission
        ↓
数据隔离（ACL）           "你能看哪些文档？" → 租户 + 可见性 + 白名单

类比：认证 = 进大楼刷门禁、RBAC = 楼层权限、ACL = 文件柜锁。

2. 四层数据模型：Tenant → User → Role → Permission

python

class User(BaseModel):
    id: str
    tenant_id: str           # 属于哪个租户
    roles: list[str]         # ["analyst", "viewer"]
    is_active: bool

    def permissions(self) -> set[str]:
        return permissions_of(self.roles)

    def has(self, permission: str) -> bool:
        perms = self.permissions()
        return "*" in perms or permission in perms

Code

Tenant("acme-corp")
  ├── alice, roles=["kb_admin"]  → kb:read, kb:write, kb:delete, agent:run
  ├── bob,   roles=["viewer"]    → kb:read, agent:run
  └── admin, roles=["admin"]     → * (通配符，全权限)

Tenant("beta-inc")
  └── dave, roles=["analyst"]    → kb:read, agent:run, tool:use:medium

权限命名规则：{domain}:{action}:{scope}

权限	含义
`kb:read`	读知识库
`kb:write`	写入文档
`kb:delete`	删除文档
`tool:use:low`	使用低风险工具
`tool:use:high`	使用高风险工具
`approval:approve`	审批通过
`conv:read:self`	只看自己的对话
`*`	通配符，超管全权限

3. 认证层：JWT → User

python

async def current_user(token: str = Depends(oauth2_scheme)) -> User:
    if not settings.enable_auth:
        return _DEV_USER   # 开发模式：跳过认证，返回全权限 dev 用户

    payload = decode_token(token)            # JWT 解码
    user = users_db.get_by_id(payload["sub"])
    if not user or not user.is_active:
        raise HTTPException(401)
    return user

enable_auth=False → 向后兼容，老接口不需要改动。安全功能默认关闭、按需开启。

4. RBAC 层：require_permission

python

def require_permission(permission: str):
    def _dep(user: User = Depends(current_user)) -> User:
        if not access_control_service.can(user, permission):
            raise HTTPException(403, detail=f"permission denied: {permission}")
        return user
    return _dep

# 用法
@router.post("/documents")
async def upload_doc(
    ...,
    user: User = Depends(require_permission("kb:write")),
): ...

一行 Depends 完成 RBAC。can() 逻辑：没用户 → False、有 * 超管 → True、权限集合里有 → True。

还有 require_any_permission("a", "b") 支持 OR 逻辑——有 a 或有 b 就行。

5. 多租户隔离：tenant_id 穿透所有表

users、conversations、agent_runs、approvals、document_acl — 每张表都有 tenant_id 列 + 索引。查询时 WHERE tenant_id = ? 做行级过滤。

python

def must_match_tenant(self, user, tenant_id):
    if self.is_super_admin(user):
        return              # 超管可跨租户
    if user.tenant_id != tenant_id:
        raise PermissionError("tenant mismatch")

6. ACL 层：文档级权限

三档可见性

python

class Visibility(str, Enum):
    PUBLIC = "public"         # 同租户登录用户
    INTERNAL = "internal"     # 同租户登录用户（默认）
    RESTRICTED = "restricted" # 白名单（allowed_users 或 allowed_roles）

can_access_doc 判断链

python

def can_access_doc(self, user, acl):
    if user is None or not user.is_active:  return False
    if "*" in user.permissions():           return True    # 超管
    if acl.tenant_id != user.tenant_id:     return False   # 核心隔离！
    if acl.visibility == RESTRICTED:
        return user.id in acl.allowed_users or \
               bool(set(user.roles) & set(acl.allowed_roles))
    return self.can(user, "kb:read")  # PUBLIC/INTERNAL

判断顺序：每步都可能直接拒绝或放行。

7. 检索时的两阶段 ACL 过滤（核心！）

Code

第 1 阶段（向量库层）：where tenant_id = user.tenant_id
  → 粗过滤，排除其他租户，性能好
        ↓ 候选 chunks（同租户）
第 2 阶段（应用层）：逐条 can_access_doc(user, acl)
  → 精细过滤，排除 RESTRICTED 文档中用户不在白名单的
        ↓
  最终 top-K

第 1 阶段：

python

def build_retrieval_filter(self, user):
    if user is None or self.is_super_admin(user):
        return {}                            # 超管不做租户过滤
    return {"tenant_id": user.tenant_id}     # Chroma where 参数

第 2 阶段：

python

def filter_hits_by_acl(self, user, hits):
    if user is None or self.is_super_admin(user):
        return hits                          # 向后兼容

    kept = []
    for h in hits:
        acl = acl_db.get(h["metadata"]["doc_id"])
        if acl is None:                      # 老文档兜底
            acl = DocumentACL(               # 从 metadata 临时构造
                tenant_id=h["metadata"].get("tenant_id"),
                visibility=Visibility.INTERNAL,  # 保守策略
            )
        if self.can_access_doc(user, acl):
            kept.append(h)
    return kept

从 metadata 兜底构造 ACL 是过渡方案——生产应跑 migration 给所有老文档补 ACL 记录。

为什么冗余取 3x

假设 top_k=5，ACL 过滤可能丢掉一些。只取 5 条 → 过滤后可能 2 条 → 不够用。多取 3 倍（15 条）→ 过滤后 8 条 → 截断到 5 → 保证结果充足。

8. 完整场景：bob 搜"年假政策"

Code

认证: bob(viewer, acme-corp) → require_permission("kb:read") 第1阶段: tenant_id=acme-corp → Chroma 排除 beta-inc 全部文档 → 15 条候选

第2阶段:
  chunk "hr-handbook":  INTERNAL  → bob有kb:read → chunk "salary-data":  RESTRICTED, allowed_roles=["hr_admin"]
                        → bob是viewer不在白名单 → 丢弃
  chunk "annual-leave": PUBLIC → → 过滤后 8 条 → 截断 5 条 → 送入生成

bob 永远看不到 salary-data，即使它和"年假"高度相关。安全优先于召回率。

9. 工程教训

安全默认关闭→ enable_auth=False / user=None 跳过所有 ACL，向后兼容老接口。安全功能必须能关，否则一开全崩
粗过滤只做"绝对确定"的排除→ 租户不同肯定不能看。visibility 判断放第 2 阶段（白名单逻辑粗过滤做不了）
metadata 兜底是过渡方案→ 老文档没有 ACL 记录时从 chunk metadata 临时构造。生产跑 migration 补全
* 通配符是把双刃剑→ 超管跳过所有检查。生产最小化超管数量，CI/CD 用单独 service account
权限粒度越细管理越贵→ 文档级 ACL 覆盖 80% 场景。chunk 级只在"同文档内不同密级"时才值得

10. 自测

问题 1：bob（viewer）和 alice（kb_admin）同租户。bob 能看 alice 上传的 RESTRICTED、allowed_roles=["kb_admin"] 的文档吗？

问题 2：把 visibility 也加到第 1 阶段粗过滤（Chroma where），有什么问题？

问题 3：filter_hits_by_acl 对每个 chunk 查一次 DB。60 条 chunk = 60 次查询，怎么优化？

问题 4：ROLE_PERMISSIONS 是静态字典。运营要临时给 analyst 加 kb:delete，需要改代码部署。怎么设计热更新？

问题 5（最重要）：文档的 chunk 1-5 是公开摘要，chunk 6-10 是机密细节。要实现 chunk 级 ACL，要改哪些地方？有什么代价？

答案与解析

问题 1：RESTRICTED 文档权限

bob 是 viewer → set(["viewer"]) & set(["kb_admin"]) = 空集 → 不能看。

alice 是 kb_admin → 交集非空 → 能看。

RESTRICTED 是"默认拒绝、白名单放行"，和 PUBLIC/INTERNAL 的"默认允许同租户"相反。

问题 2：粗过滤加 visibility

问题一：bob 在 RESTRICTED 文档的白名单里 → 粗过滤却把它排除了 → bob 永远搜不到。Chroma where 不支持"如果 user_id 在 JSON 数组里就保留"。

问题二：chunk 入库时 visibility=INTERNAL → 后来管理员改成 RESTRICTED → metadata 是旧的 → 粗过滤放行了本该拦截的文档。

粗过滤只做 tenant_id（几乎不变），visibility 逻辑留给第 2 阶段查最新 ACL 表。

问题 3：DB 查询优化

方案 1——批量查询：WHERE doc_id IN (id1, id2, ...) → 60 次变 1 次。效果最好。

方案 2——LRU 缓存：同一个 doc 切 10 个 chunk → 10 次变 1 次 + 9 次缓存命中。但 ACL 变更需失效。

方案 3——两者结合：先查缓存，miss 的批量查 DB。

60 次 SQLite 查询 ~30ms，对 RAG 总延迟可以忽略。换远程 DB（每次 5-10ms）则必须优化。

问题 4：热更新方案

DB 化：roles + role_permissions 两张表 → 改权限 = UPDATE 一行 → 管理员通过 API/UI 即时生效。加 60 秒缓存防频繁查询。

配置中心（生产级）：etcd/Consul 存配置 → 服务启动加载 + 监听变更 → 秒级推送所有实例。

课程用静态字典的理由：零依赖、权限少（不到30个）、变更少、代码即文档。

配置化程度和系统复杂度正相关。选择和业务变更频率匹配的方案。

问题 5：chunk 级 ACL

要改：ACL 表粒度从 doc_id → chunk_id；filter_hits_by_acl 按 chunk_id 查而非 doc_id；Ingestion 阶段写入 chunk 级 visibility；Chroma metadata 加 chunk 级 visibility。

代价：ACL 记录数爆炸（1000 文档 × 10 chunks = 10000 条）、管理复杂度（需 UI 支持按段落设权限）、ingestion 复杂（需自动识别"机密"段落）、一致性维护（文档更新后 chunk 拆分变了 → ACL 失效 → 重建）。

折中——Section 级：按"段落"设权限（摘要/详情），ACL 记录 = section 数（2-5/文档），远少于 chunk 数。

大多数企业选择"拆成多个文档"而不是 chunk 级 ACL——简单粗暴但管理成本低。

本节要点

三层安全：认证（你是谁）→ RBAC（你能做什么）→ ACL（你能看哪些数据）。每层独立可关闭，向后兼容
两阶段 ACL 过滤：向量库层 tenant_id 粗过滤 + 应用层逐条精细过滤。冗余取 3x 保证过滤后结果够用
权限粒度越细管理越贵——文档级 ACL 覆盖 80% 场景。遇到"同文档不同密级"优先考虑拆文档