多租户隔离：把租户边界收在一个地方

多租户系统最怕的不是写不出查询，而是某一天发现有个接口忘了加 org_id。这类漏洞不会在测试里报错，只会在某次演示里被人多看一页数据。

一、先明确一件事：租户边界不是业务逻辑

刚开始写的时候，很容易把"这个用户属于哪个组织"当成业务规则的一部分，散落在每个接口里：

def list_drawings(org_id, user, db):
    if user.role not in ("admin", "member"):
        raise HTTPException(403)
    return db.query(Drawing).filter(Drawing.org_id == org_id).all()

这段代码单看没问题，问题是它出现了几十次。只要有一处 org_id 从请求参数里取而不是从当前用户身上取，隔离就断了——攻击者改一下 query string 就能读到别人的数据。

后面我把这件事拆成三层，全部收进一个地方：

二、身份层：org_id 只能来自服务端解出来的上下文

关键点是：不要让 org_id 出现在任何用户可控的输入里。路径参数里的 org_id、请求体里的 org_id、Header 里的 X-Org-Id，都视为不可信。

用一个请求级上下文承载它，业务函数只读不写：

@dataclass(frozen=True)
class CurrentUser:
    id: str
    org_id: str | None
    role: str
    account_type: str

def get_current_user(token = Depends(parse_token), db = Depends(get_db)) -> CurrentUser:
    payload = decode(token)
    user = db.get(User, payload["sub"])
    if not user or not user.is_active:
        raise Unauthorized()
    return CurrentUser(user.id, user.org_id, user.role, user.account_type)

frozen=True 是刻意的：任何函数拿到它都改不了。要"切换组织"只能重新签发令牌。

三、查询层：让遗漏变成不可能，而不是靠自觉

SQLAlchemy 2.0 给了两个很好用的东西：with_loader_criteria 和会话事件。我选的是前者，因为它同时管住了 select() 和 join：

@event.listens_for(Session, "do_orm_execute")
def _add_tenant_filter(state):
    if not state.is_select:
        return
    org_id = current_org_id()          # 从上下文取，取不到就不加（平台端）
    if not org_id:
        return
    state.statement = state.statement.options(
        with_loader_criteria(
            TenantMixin,
            lambda cls: cls.org_id == org_id,
            include_aliases=True,
        )
    )

凡是继承 TenantMixin 的模型，都会被自动补上条件，包括子查询和别名。include_aliases=True 很关键，否则 join 出来的别名表会漏掉。

这一步之后，业务代码里的 filter(Drawing.org_id == org_id) 就成了冗余保险，可以留着，但不再承担唯一的防线。

四、失败层：跨租户一律 404，不要 403

403 会告诉对方"这条记录存在，只是你无权"。在多租户场景里，这本身就是信息泄露——用 ID 枚举一遍就能摸出别人的数据规模。

所以判断顺序应该是：先在当前租户范围内查，查不到就当它不存在。查询层已经自动加了条件，天然就是 404，反而省了写判断。

如果一个"权限不足"的场景返回了 403，要先问一句：这条记录的存在性本身，算不算敏感信息？

五、五个真实漏过的口子

1. 批量更新与删除。update().where(id.in_(...)) 不走 ORM 查询钩子，得单独补条件。最好的办法是禁用裸 update，统一走"先查后改"。
2. 聚合与计数。select(func.count()) 如果手写表名而不是模型，钩子抓不到。表名一旦写死成字符串，就等于绕过了整个机制。
3. 导出与报表。导出接口常常为了性能自己拼 SQL，最容易漏。我后来强制导出复用列表接口的查询构造函数。
4. 后台任务。Celery worker 里没有请求上下文，current_org_id() 取不到值，过滤条件就静默失效。必须把 org_id 作为任务参数显式传入，并在任务开头重建上下文。
5. 缓存键。缓存 key 忘了拼 org_id，A 组织的响应会被 B 组织读到。这类 bug 只在第二个组织进来之后才出现，最容易上线后才被发现。

六、测试：跨租户用例必须成为固定项目

我给自己的规矩是：每新增一张带租户字段的表，就必须配一条跨租户测试。写法很朴素——造两个组织，各一条数据，用 A 的令牌去读 B 的 ID，断言必须 404：

def test_cross_tenant_404(client, seed_two_orgs):
    a, b, b_doc = seed_two_orgs
    r = client.get(f"/api/drawings/{b_doc.id}", headers=a.headers)
    assert r.status_code == 404

这条测试跑起来只要几十毫秒，但它拦住的是最难在线上发现的一类问题。

七、小结

多租户隔离的难点不在技术，在于"不要相信任何一次自觉"。只要还有一处需要人手写 org_id，就一定会有下一处忘记写。

把边界压到依赖注入里、把过滤压到会话钩子里、把失败统一成 404，剩下的工作才是真正的业务逻辑。这个顺序反了，后面每加一个功能都在还债。