Python后端AI专题28多租户隔离不是加一个 tenant_id三层权限防线在表里加tenant_id只是数据建模隔离要靠每条入口、每次资源查询和每条检索 SQL 都使用它。KnowFlow 的防线分三层Token 建立身份Repository/Service 约束资源归属pgvector 在 Top K 前过滤任一层遗漏都可能泄露。本篇最终完整模块knowledge_bases.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsfromcollections.abcimportSequencefromuuidimportUUIDfromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionfromapp.modelsimportKnowledgeBaseclassKnowledgeBaseRepository:All public resource operations require tenant identity explicitly.asyncdefget(self,tenant_id:UUID,knowledge_base_id:UUID,*,session:AsyncSession)-KnowledgeBase|None:statementselect(KnowledgeBase).where(KnowledgeBase.tenant_idtenant_id,KnowledgeBase.idknowledge_base_id,)returnawaitsession.scalar(statement)asyncdeflist(self,tenant_id:UUID,*,session:AsyncSession)-Sequence[KnowledgeBase]:statement(select(KnowledgeBase).where(KnowledgeBase.tenant_idtenant_id).order_by(KnowledgeBase.created_at,KnowledgeBase.id))returnlist((awaitsession.scalars(statement)).all())asyncdefcreate(self,tenant_id:UUID,name:str,description:str,*,session:AsyncSession,)-KnowledgeBase:knowledge_baseKnowledgeBase(tenant_idtenant_id,namename.strip(),descriptiondescription.strip())session.add(knowledge_base)awaitsession.flush()returnknowledge_base重复投递练习答案新增测试让 ready job 再执行一次并用 CountingStorage/Embedding 调用数证明它连原文件都不再读取firstawaitindexer.run(job.id)calls_after_firstembedder.calls reads_after_firststorage.get_calls secondawaitindexer.run(job.id)recordsawaitvectors.records_for(tenant_idtenant,knowledge_base_idkb)assertfirst.statussecond.statusreadyassertembedder.callscalls_after_firstassertstorage.get_callsreads_after_firstassertlen(records)first.total_chunksIndexer 入口增加jobawaitself.jobs.get(job_id)ifjob.statusready:returnjob当前 ingestion 套件4 passed in 0.38s。显式 reindex 不应偷偷复用 ready job而应创建带新 pipeline fingerprint 的版本化 job这样旧索引仍可服务和回滚。第一层Token 只提供身份不提供最终授权JWT 只保存sub/tid/jti/iat/exp/type不放知识库清单payload{sub:str(identity.user_id),tid:str(identity.tenant_id),jti:str(identity.token_id),iat:issued,exp:issuedexpires_delta,type:access,}签名防篡改过期时间限制寿命资源是否仍存在、用户是否仍 active要由数据库实时判断。客户端请求体没有tenant_id字段路由从已验证 Token 取得避免“传谁的 tenant 就看谁的数据”。密码使用带随机盐的 scrypt验证用hmac.compare_digest。日志不记录密码、JWT 或完整 Prompt。第二层Repository 签名强迫携带 tenant知识库 Repository 的完整模块from__future__importannotationsfromcollections.abcimportSequencefromuuidimportUUIDfromsqlalchemyimportselectfromsqlalchemy.ext.asyncioimportAsyncSessionfromapp.modelsimportKnowledgeBaseclassKnowledgeBaseRepository:All public resource operations require tenant identity explicitly.asyncdefget(self,tenant_id:UUID,knowledge_base_id:UUID,*,session:AsyncSession,)-KnowledgeBase|None:statementselect(KnowledgeBase).where(KnowledgeBase.tenant_idtenant_id,KnowledgeBase.idknowledge_base_id,)returnawaitsession.scalar(statement)asyncdeflist(self,tenant_id:UUID,*,session:AsyncSession)-Sequence[KnowledgeBase]:statement(select(KnowledgeBase).where(KnowledgeBase.tenant_idtenant_id).order_by(KnowledgeBase.created_at,KnowledgeBase.id))returnlist((awaitsession.scalars(statement)).all())asyncdefcreate(self,tenant_id:UUID,name:str,description:str,*,session:AsyncSession,)-KnowledgeBase:itemKnowledgeBase(tenant_idtenant_id,namename.strip(),descriptiondescription.strip(),)session.add(item)awaitsession.flush()returnitem单测用inspect.signature规定所有公开方法的第一个业务参数必须是 tenant_id。它不能证明 SQL 一定正确却能防止新增方法忘记设计隔离入口SQL/E2E 另行验证。第三层向量排名前过滤第 21 篇已经让DocumentChunk.tenant_id、knowledge_base 和 ready 状态进入 WHERE。不能因为 API 已鉴权就删除这一层内部调用、后台任务、以后新增路由都可能绕过原入口。安全需要纵深防御。为什么统一返回 404Alice 请求 Bob 的 UUID如果返回 403攻击者知道资源存在如果不存在返回 404、越权返回 404外部无法区分。响应中也不能包含 Bob 名称。已有跨租户知识库与会话 E2E 均检查这一点。404 不是万能安全日志和审计仍应记录“鉴权主体请求了不属于它的 ID”用于攻击检测只是不对客户端泄露。数据库约束能做什么、不能做什么复合索引(tenant_id, knowledge_base_id)提升过滤效率外键保证引用存在但普通外键无法自动保证 chunk 的 tenant 与 document 的 tenant 一致。可进一步使用 PostgreSQL RLS 或复合外键增强。当前课程把应用查询与测试作为主防线并明确 RLS 是可继续加强的生产项。本篇练习同租户跨用户与跨租户矩阵画一个 2×2 验收矩阵同租户/不同租户 × 本人会话/他人会话。为每格写预期 HTTP 状态再为“管理员可读同租户会话”设计最小 RBAC 变更说明角色应从哪里读取、Repository WHERE 如何变化、为什么不能只相信 JWT 中长期不变的roleadmin。下一篇会给出矩阵和 RBAC 答案并处理另一类共享状态并发请求下配额、幂等键、缓存和 Token 成本如何一起设计。