Python后端权限系统03：登录之后还要继续核验——不透明令牌、过期与退出登录
Alice 在一台共享电脑上完成采购操作，点击退出，然后离开。接班同事从浏览器调试记录里找到了她刚才的 Authorization 头，把这个头复制到自己的脚本中。如果服务器仍然接受这段凭据，“退出”就只是界面清除了一个变量，Alice 的服务端访问能力并没有结束。
本课把会话从前端存储细节中拿出来，视为一个有生存期、有撤销状态、必须在每次请求时验证的服务端对象。我们选用不透明 access token，并在数据库中维护会话关系。选择它是为了把撤销、禁用与刷新教学放在同一条可观察的链路中；这并不意味着所有系统只能使用这种方案。
上一课 E02：为什么断言记录不同，而不是指定记录内容
lessons/03/answer_previous.py的完整答案先在内存中生成临时口令，验证两个独立 scrypt 记录拥有不同的盐与结果，再验证正确、错误、损坏记录三个分支。它随后调用真实登录端点，检查错误口令和未知用户都得到 401，同时确认正确口令仍能得到会话。运行python lessons/03/answer_previous.py即可执行。
本地实际运行输出如下，成功摘要不会泄露随机口令或散列：
E02 PASS: salted scrypt and uniform login rejection这份答案没有测“散列函数输出是否等于某个固定字符串”，因为盐是每次生成的。相反，它测的是系统必须稳定维持的性质。也不要用微秒计时的单次结果声称消除了账号枚举侧信道：我们检查统一对外拒绝契约，不把机器调度噪声包装成密码学证据。完整代码位于本篇末尾，与专题文件一一对应。
不透明令牌究竟不透明在哪里
成功登录返回的access_token是随机生成的持有者凭据。客户端不能从它的字符推导出用户名、部门、角色或有效期，也不应该对它进行业务解析。服务端接到Authorization: Bearer ...后计算摘要，通过摘要查找相应会话，再检查当前状态。调用方只负责原样携带，不需要知道数据库结构。
这与把 JSON 做一次 Base64 编码完全不同。编码后的用户编号仍然可以被解码或篡改，不能用作可靠认证证据。它也不同于签名 JWT：JWT 可以携带可验证的声明，而本项目的随机字符串本身不承载任何声明。第十、十一课会在 OIDC 身份令牌里介绍签名 JWT；届时要区分“上游身份断言”和“本地会话凭据”各自的受众与生命周期。
使用不透明令牌意味着每次请求依赖服务端状态查询，代价是需要可用的会话存储，收益是撤销和当前用户状态能够及时生效。若系统选择短期 JWT 来减少查询，也必须回答登出、账号禁用、权限改变后多长时间生效，是否需要撤销列表或会话版本。不能把“无状态”当作所有身份生命周期问题已经自动消失。
为什么数据库保存摘要，而不是原始令牌
服务端不需要重新取出原始 access token。客户端每次带来它，服务端计算相同的摘要后查找即可。ProcureSafe 使用 SHA-256 保存令牌摘要，原始令牌只在签发响应中交付。这样，单独拿到会话表的读取者不能直接把摘要当作 Bearer 值使用。
上一课要求对密码使用慢散列，这里却使用快速 SHA-256，看起来似乎矛盾。区别在输入的熵：人类口令通常来自可猜测分布，需要提高每次猜测成本；本项目令牌由密码学随机源生成，具有足够随机性，攻击者无法合理枚举候选空间。慢散列不能替代随机令牌生成，快速摘要也不能被倒推成“密码照样用 SHA-256 就行”。
摘要保护也有边界。若原始令牌从浏览器脚本、代理日志或错误截图里泄露，持有者仍可使用它。若攻击者能够修改会话数据库，而不只是读取摘要，也可能改变撤销状态或关联关系。我们需要精确说明每个控制抵御的场景，不能把“数据库里存的是哈希”扩展成“令牌不会被盗”。
一次带凭据的请求，要连续通过哪些检查
GET /auth/me是观察会话判断的最小入口。有效请求返回用户 id 与 username，不返回散列、会话摘要或 refresh token。会话层必须拒绝未知凭据，必须拒绝过期 access token，必须拒绝已经撤销的 family，还必须读取用户当前的 enabled 状态。某次登录时用户启用，不保证几小时后仍然启用。
本项目的 access 生存期最多六百秒，refresh family 的绝对生存期是三千六百秒。接近 family 截止时间刷新时，新 access 也不能活得比家族更久，响应中的expires_in会相应缩短。时间来自应用工厂注入的clock，默认使用真实时间，实验可使用FakeClock。这样可以把时间推进到边界外，观察旧凭据被拒绝，而不用在测试中睡眠十分钟。注入时钟改变的是时间来源，不会替代数据库、认证过程或 HTTP 路由。
还有两个很常见的边界错误。一是只在刷新时检查用户是否禁用，让禁用后的旧 access token 在剩余期限内继续有效；二是先从 token 里缓存角色，再长期不查当前成员关系，使管理员已经撤回的审批权继续存在。本项目在身份检查时核对用户状态，在租户业务请求时重新读取成员关系，把两种变化分别落在负责它们的边界上。
会话和权限不是一份数据
Alice 的 access token 能证明“当前请求代表 Alice”，不能单独证明她能访问 acme，也不能证明她在该组织中仍然是经理。身份相对稳定，组织成员关系可以快速变化。若把角色复制进长期凭据，就需要额外设计失效机制，否则发放时的历史角色会与当前数据库事实发生冲突。
采购系统中的服务层接收已认证用户，在目标租户查询 membership，再据此过滤资源与判断动作。这个设计使同一账号未来加入两个租户时，可以在每个租户拥有不同部门和角色。请求的 tenant 参数仍然不可信，但它明确选择当前组织上下文，服务端必须核对这种选择是否合法。
因此不应该写一个笼统的is_admin覆盖所有判断。平台管理员、租户管理员、采购操作员并没有天然的包含关系；管理员能改成员，并不表示管理员拥有付款资格。我们先把身份会话做可靠，第六课再展开组织关系，第七课讨论资源与动作，这样每条拒绝都能指向一项明确规则。
401 与 403 会影响用户怎样恢复
当凭据不存在、过期或被撤销时，当前身份不能成立，接口返回 401，客户端可以尝试合适的重新认证流程。若凭据有效但缺少当前组织或动作权限，接口返回 403；盲目让用户再登录一次通常不会增加权限，只会造成循环跳转。
具体资源不可见时可以返回 404，目的是减少对目标存在性的披露。若申请可见，但当前角色不能付款，返回 403 就是恰当的动作拒绝。若角色正确、资源可见，但申请还没有进入允许付款的状态，返回 409，告诉调用者自己依据的业务前提已经不成立。四种结果分别对应身份、授权、可见性与冲突，不应混为一个通用失败码。
客户端可以据此做明确恢复：401 引导恢复会话；403 说明当前操作不被允许；404 显示目标不可访问；409 重新加载当前资源并让用户重新确认。用户体验不是安全逻辑的对立面，清楚的错误契约可以减少前端自行猜测与不恰当重试。
退出登录撤销哪一段生命周期
POST /auth/logout用当前 Bearer 凭据识别 family，然后撤销这一整个 family。它不只是删除这一个 access token，也会使同一家族内后续依赖的 refresh token 失效。否则浏览器清掉 access 后，残留的 refresh token 仍可重新生成可用访问凭据，退出的语义就不完整。
这里的 family 可以理解为一次登录建立的会话链。另一次独立登录通常创建另一个 family；撤销当前链不自动意味着“所有设备退出”。产品若需要账号级全部退出，必须明确找出并撤销该用户全部活动 family，或者引入账号会话版本。教学接口只承诺当前 family，不用模糊措辞暗示它完成了设备管理功能。
也不要假设登出一定能取消已经在数据库事务中执行的业务写入。撤销作用于之后的认证判断；一个已完成身份校验、正在提交的请求与登出的竞争关系，需要结合具体事务边界分析。对高风险操作可以增加更近的核验或再认证，但这属于额外需求，不是前端清除存储后自然得到的效果。
本课练习 E03：旧令牌还在手里，也必须失去作用
用 Alice 登录并保留原始 access 与 refresh token，只在内存中保存。先调用/auth/me断言身份正确，再调用/auth/logout断言得到 204。随后重新使用完全相同的旧 access 请求/auth/me，必须得到 401；使用同一家族的 refresh 请求刷新，也必须得到 401。
第二段实验创建一次独立登录，确认它可以访问身份端点，再把FakeClock推进到 access 期限之后，确认过期凭据同样得到 401。用两段实验分别证明“主动撤销”和“自然到期”，避免把失败原因混在一起。输出只给断言摘要，不输出登录响应内容。
下一课会完整解答 E03，并故意重放已经消费过的 refresh token。你会看到只判断一个 token 是否过期远远不够：为了识别复制与重放，会话还需要父子关系、单次消费以及能处理并发的事务。
上一课答案源码与本课完整会话模块
"""E02 完整答案：验证盐与散列性质，而非固定随机输出快照。"""frompathlibimportPathimportsecretsimportsys sys.path.insert(0,str(Path(__file__).resolve().parents[1]))fromsupportimportlogin,sandboxfromprocuresafe.passwordsimporthash_password,verify_passworddefmain():password=secrets.token_urlsafe(24)first,second=hash_password(password),hash_password(password)assertfirst!=second parts=[value.split("$")forvaluein(first,second)]assertall(row[:4]==["scrypt","16384","8","1"]forrowinparts)assertall(len(bytes.fromhex(row[4]))==16forrowinparts)assertall(len(bytes.fromhex(row[5]))==32forrowinparts)assertparts[0][4]!=parts[1][4]assertverify_password(password,first)andverify_password(password,second)assertnotverify_password(secrets.token_urlsafe(24),first)assertnotverify_password(password,"damaged-record")assertnotverify_password(password,first.replace("$16384$","$1073741824$"))withsandbox()as(client,credentials):wrong=client.post("/auth/login",json={"username":"alice","password":secrets.token_urlsafe(24)})unknown=client.post("/auth/login",json={"username":"unknown-user","password":credentials["alice"]})assertwrong.status_code==unknown.status_code==401assertwrong.json()==unknown.json()forresponsein(wrong,unknown):assertnot{"access_token","refresh_token","csrf_token"}&response.json().keys()# 负例成立后再做正例，排除“任何人都登不进来”的假安全实现。accepted=login(client,credentials,"alice")assertaccepted["token_type"]=="bearer"andaccepted["expires_in"]==600print("E02 PASS: salted scrypt and uniform login rejection")if__name__=="__main__":main()"""Opaque令牌只存hash，权限在每次请求时读取，不把角色冻结进令牌。"""fromdataclassesimportdataclassimporthashlibimportsecretsfromtypingimportCallablefrom.dbimportsnapshot,transactionfrom.errorsimportDomainErrorfrom.passwordsimporthash_password,verify_password ACCESS_TTL,REFRESH_TTL=600,3600LOCK_SECONDS,MAX_FAILURES=60,5# 未知用户也走同等scrypt计算。统一响应并不等于已解决所有时间侧信道。_DUMMY_HASH=hash_password(secrets.token_urlsafe(32))deftoken_digest(value:str)->str:returnhashlib.sha256(value.encode("utf-8")).hexdigest()@dataclass(frozen=True)classPrincipal:user_id:strusername:strfamily_id:strtoken_hash:strcsrf_hash:strdefissue_tokens(conn,user_id:str,now:float,family_id:str|None=None)->dict:"""调用者持有写事务；刷新沿用family绝对过期时间，不能无限续命。"""iffamily_idisNone:family_id=secrets.token_urlsafe(24)conn.execute("INSERT INTO families(id,user_id,expires_at) VALUES(?,?,?)",(family_id,user_id,now+REFRESH_TTL))family=conn.execute("SELECT expires_at FROM families WHERE id=?",(family_id,)).fetchone()access,refresh,csrf=(secrets.token_urlsafe(32)for_inrange(3))conn.execute("INSERT INTO access_tokens VALUES(?,?,?,?)",(token_digest(access),family_id,min(now+ACCESS_TTL,family["expires_at"]),token_digest(csrf)))conn.execute("INSERT INTO refresh_tokens(token_hash,family_id) VALUES(?,?)",(token_digest(refresh),family_id))return{"access_token":access,"refresh_token":refresh,"csrf_token":csrf,"token_type":"bearer","expires_in":int(min(ACCESS_TTL,family["expires_at"]-now))}deflogin(database:str,username:str,password:str,clock:Callable[[],float])->dict:tokens=Nonewithtransaction(database)asconn:user=conn.execute("SELECT * FROM users WHERE username=?",(username,)).fetchone()valid=verify_password(password,user["password_hash"]ifuserelse_DUMMY_HASH)now=clock()# 排队和密码计算都可能消耗时间，不能沿用请求到达时刻。ifuseranduser["enabled"]anduser["locked_until"]<=now:failures=0ifuser["locked_until"]elseuser["failed_attempts"]ifvalid:conn.execute("UPDATE users SET failed_attempts=0,locked_until=0 WHERE id=?",(user["id"],))tokens=issue_tokens(conn,user["id"],now)else:failures+=1until=now+LOCK_SECONDSiffailures>=MAX_FAILURESelse0conn.execute("UPDATE users SET failed_attempts=?,locked_until=? WHERE id=?",(failures,until,user["id"]))# 先提交失败计数再拒绝，否则抛异常会把限流记录一起回滚。iftokensisNone:raiseDomainError(401,"Invalid credentials")returntokensdefauthenticate_connection(conn,digest:str,now:float)->Principal:row=conn.execute("""SELECT u.id,u.username,a.family_id,a.csrf_hash FROM access_tokens a JOIN families f ON f.id=a.family_id JOIN users u ON u.id=f.user_id WHERE a.token_hash=? AND a.expires_at>? AND f.expires_at>? AND f.revoked=0 AND u.enabled=1""",(digest,now,now)).fetchone()ifnotrow:raiseDomainError(401,"Authentication required")returnPrincipal(row["id"],row["username"],row["family_id"],digest,row["csrf_hash"])defauthenticate(database:str,token:str,clock:Callable[[],float])->Principal:ifnottokenorlen(token)>256:raiseDomainError(401,"Authentication required")withsnapshot(database)asconn:returnauthenticate_connection(conn,token_digest(token),clock())defrefresh(database:str,token:str,clock:Callable[[],float])->dict:tokens=Nonewithtransaction(database)asconn:now=clock()# BEGIN IMMEDIATE可能等待；取锁后再判断family绝对期限。row=conn.execute("""SELECT r.family_id,r.used,f.user_id,f.expires_at,f.revoked,u.enabled FROM refresh_tokens r JOIN families f ON f.id=r.family_id JOIN users u ON u.id=f.user_id WHERE r.token_hash=?""",(token_digest(token),)).fetchone()ifrowandrow["used"]:# 重放的撤销必须提交；不能在当前事务里直接raise导致撤销回滚。conn.execute("UPDATE families SET revoked=1 WHERE id=?",(row["family_id"],))elifrowandnotrow["revoked"]androw["enabled"]androw["expires_at"]>now:conn.execute("UPDATE refresh_tokens SET used=1 WHERE token_hash=?",(token_digest(token),))tokens=issue_tokens(conn,row["user_id"],now,row["family_id"])iftokensisNone:raiseDomainError(401,"Invalid refresh token")returntokensdeflogout(database:str,principal:Principal,clock:Callable[[],float])->None:withtransaction(database)asconn:current=authenticate_connection(conn,principal.token_hash,clock())conn.execute("UPDATE families SET revoked=1 WHERE id=?",(current.family_id,))专题目录：《认证、授权与企业级权限系统：课程目录》 · 上一篇：《Python后端权限系统02：数据库泄露之后，密码还能守住什么——scrypt 与统一登录失败》 · 下一篇：《Python后端权限系统04：刷新不是续一张票——单次轮换、重放检测与事务竞争》
