KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA

03 · 落地形态：函数内判定、依赖注入与注解 — keel 龙骨

同一套判定逻辑，写在三个地方，可靠性完全不同。这一章用配套项目的三份实现做对照：guards.py 里的函数内判定、api.py 里的依赖注入、guards.require_perm 装饰器。结论先给：依赖注入是 Web 入口的正解，装饰器是非 Web 入口的正解，函数内判定只适合兜底。而"中间件"这种形态在真实项目里常被证伪——本章给出一个实测过的反例。

同一套判定逻辑，写在三个地方，可靠性完全不同。这一章用配套项目的三份实现做对照：guards.py 里的函数内判定、api.py 里的依赖注入、guards.require_perm 装饰器。结论先给：依赖注入是 Web 入口的正解，装饰器是非 Web 入口的正解，函数内判定只适合兜底。而"中间件"这种形态在真实项目里常被证伪——本章给出一个实测过的反例。

一、现场：同一个团队，两种写法，差一个接口

一个项目里同时存在两种鉴权写法：

# 写法一（老接口，团队 A 维护）
@router.delete("/users/{id}")
async def delete_user(id: str):
    if not has_perm(current_user, "system:user:delete"):
        raise HTTPException(403)
    ...

# 写法二（新接口，团队 B 维护）
@router.delete("/agents/{id}")
async def delete_agent(id: str, _=Depends(require_perm("agent:delete"))):
    ...

半年后出了漏洞：新加的 delete_agent 团队在实现时把 _=Depends(...) 删掉了（"这个接口内部会判"），
而内部那个函数是团队 A 写的、判的是另一个权限点。没有任何测试发现，因为测试也不测权限。

这一章要回答的正是：怎样让"漏写"这件事从"靠人记得"变成"结构上不可能"。

二、概念边界：三种形态的差别是"声明放在哪"

关键判断标准只有一个：声明与被保护的代码，在同一次阅读里能不能一起看到。
函数内判定的失败模式是"这段代码看起来在处理业务，没意识到它还负责鉴权"；
依赖注入的声明就在路由上，任何人读路由定义都能看到"这个接口要什么权限"。

三、一次完整运行：三份实现与它们的实测差异

配套项目 project/guards.py 里三种形态都在，判定内核只有一份（都调用 rbac.has_permission）：

# 形态 1：函数内判定
def delete_user_via_inline_check(conn, target_user_id):
    require_permission(conn, "system:user:delete")     # ← 权限声明混在业务里
    one(conn, "SELECT 1 FROM app_user WHERE id = ?", (target_user_id,))
    return f"deleted:{target_user_id}"

# 形态 2：依赖注入（api.py 里用）
@app.delete("/users/{target_user_id}")
async def delete_user(target_user_id: str, identity: Identity = Depends(needs("system:user:delete"))):
    return {"deleted": target_user_id, "by_member": identity.member_id}   # 函数体里没有鉴权代码

# 形态 3：装饰器（脚本/定时任务/消费者用）
@require_perm("system:user:delete")
def run_batch_delete(...):
    ...

形态 2 的函数体里一行鉴权代码都没有——这不是因为它不安全，而是因为权限声明已经在路由上，
依赖注入保证了它一定被执行。实测（python api_probe.py）：

  运营角色删用户（越权）   DELETE /users/u_bob   -> 403  missing_permission:system:user:delete

四、为什么"重复判一次"不是更安全

配套项目最初在 list_users 端点里既写了 Depends(needs("system:user:list"))，
又在函数体里调了一次 require_permission(conn, "system:user:list")，注释写着"幂等的第二道"。

这行代码被删掉了。理由是：

• 它不提供任何额外保证（依赖已经保证了同一件事）；
• 它制造了第二处会漂移的副本——将来有人把依赖改成 system:user:read，
  忘了改函数体里那行，就出现"依赖放行、函数体拒绝"或反之的诡异行为；
• 排查问题时你会去查两个地方，而只有一个是真相来源。

判据：如果两次判定是同一个决策，那就只留一个；如果是两个不同的决策
（例如"能不能看列表"和"能不能导出"），那就应该用 needs("a", "b") 在声明处一次表达。

五、一个实测过的反例：为什么不用中间件

很多项目的鉴权写在中间件里（middleware/authorize.py），看起来最统一。
但这种写法有一个结构性缺陷：中间件运行时拿不到路由上声明的元数据。

设想你要用中间件实现"路由 → 所需权限"的映射，只能这样：

# 中间件路线：另维护一张 URL → 权限 的硬编码表
ROUTE_PERMISSION = {
    ("DELETE", "/api/users/{id}"): "system:user:delete",
    ("POST", "/api/users/export"):  "system:user:export",
    # ... 每加一个接口就要在这里补一行
}

async def auth_middleware(request, call_next):
    key = (request.method, normalize(request.url.path))   # ← 还得处理路径参数归一化
    need = ROUTE_PERMISSION.get(key)
    ...

而依赖注入的写法是：

@router.delete("/users/{target_user_id}")
async def delete_user(target_user_id: str, _=Depends(needs("system:user:delete"))):
    ...

区别不在于代码量，而在于"新增接口时会不会忘记"。 中间件路线里，新接口如果不改那张表，
鉴权就静默失效；依赖注入路线里，不写 Depends 就没有权限声明，而"这个接口没有权限声明"
在 code review 时是显眼的（旁边就是路由定义本身）。

配套项目没有实现中间件版本，因为它的失败模式无法用"加个测试"兜住——
它需要的是一个人记住去改另一张表。真实项目里这类"授权路由表"经常与真实路由表不同步，
最后变成一张谁都不敢删的历史包袱。

这不是纸上推演：不少团队的项目里都存在一个被注释掉的鉴权中间件文件，和一张没人敢删的
ROUTE_PERMISSION 常量表——它们常常是同一次改造留下的两份残留物。要判断"该保留哪一份"，
判据是权限声明是否与路由定义在同一次阅读里：同处一行才可能不漏，分成两张表就必然要靠人同步。

六、一个真实的坑：ContextVar 不跨线程池

配套项目在接线时踩到了这个坑，值得完整记录——它是"鉴权代码写对了但行为不对"的典型。

身份放在 ContextVar 里（这是正确选择，见下文），FastAPI 的端点最初写成同步函数：

@app.get("/resources/{resource_id}")
def read_resource(resource_id: str, ...):        # ← 同步端点
    actual = can_access(conn, resource_id, level)  # 里面读 current_identity()

实测报错：

401 no_identity_in_context

原因：FastAPI 会把同步的依赖和端点丢进线程池执行，而 ContextVar 不跨线程池传递。
于是依赖里 set_identity() 写进的是那个工作线程的 ContextVar，
端点里 current_identity() 读的是请求协程的 ContextVar，拿到 None。

改成 async def 后正常。但随后又踩第二次——依赖函数本身还是同步的：

def current_identity(x_user_id: str = Header(...)):    # ← 同步依赖，同样进线程池
    identity = resolve_identity(conn, ...)
    set_identity(identity)          # 写进了工作线程的 ContextVar
    return identity

端点已经是 async 了，但依赖还是同步的，set_identity 依然发生在另一个线程。
两个函数都必须 async def。修正后的实测输出（python api_probe.py 全绿）：

  无凭证访问 /me              GET    /me                                -> 401  missing_user
  非该租户成员                 GET    /me                                -> 403  not_a_member_of_tenant
  运营角色读用户列表              GET    /users                             -> 200  {'visible_users': 1}
  运营角色删用户（越权）            DELETE /users/u_bob                       -> 403  missing_permission:system:user:delete
  管理员读用户列表               GET    /users                             -> 200  {'visible_users': 3}
  同一账号换租户                GET    /me                                -> 200  {'user_id': 'u_alice', 'member_id': 'm_alice_globex', ...}
  读他人资源 use 级            GET    /resources/r_agent_x               -> 200  {... 'actual_level': 'use'}
  读他人资源 observe 级        GET    /resources/r_agent_x?level=observe -> 403  access_level_denied:observe<use
  读不存在的资源                GET    /resources/r_nope                  -> 404  resource_not_found
  跨租户读资源                 GET    /resources/r_flow_y                -> 404  resource_not_found

为什么用 ContextVar 而不是把身份当参数层层传：

• 参数传递会污染每一个函数签名（而签名是"这个函数需要什么"的最好文档）；
• ContextVar 对 async 与线程都安全，且不会在并发请求之间串味
  （普通全局变量会——这是"用户 A 看到用户 B 数据"的经典成因之一）。

代价就是本节这个坑：一旦有代码在线程边界之外（线程池、Celery worker、其他进程）执行，
它读不到请求上下文。这时必须显式传递（04 章讲跨进程时的身份传递）。

七、误判澄清

八、生产环境怎样替换

九、练习与验收

练习 1：在你的项目里找出所有"函数体第一行是鉴权"的接口，统计数量；估算如果新增 20 个接口，按这种写法需要改多少处。

练习 2：在配套项目里把 current_identity 改回同步 def（端点保持 async），复现 401，并解释为什么这次端点是对的、依赖是错的。

练习 3：用 Depends 表达"需要 A 且 B 两个权限"的依赖，再想一个"需要 A 或 B"的场景，说明后者为什么不该用 needs("A", "B")（提示：06 章的 checker 链）。

验收点：不看资料，说出三种落地形态的差别是"声明放在哪"；解释中间件路线的失败模式；复现并解释 ContextVar 跨线程池的问题；说明"重复判定"为什么零收益甚至有害。

现在能解释什么：三种鉴权落地形态的可靠性差异与判断标准；中间件为什么在真实项目里常被证伪；ContextVar 的适用边界与跨线程/跨进程的传递要求。

下一步：第 04 章进入多租户——搞清楚 user_id、租户内身份 id、租户 id 这三级 id 各自的分工。

进入 keel 阅读