Python后端权限系统04刷新不是续一张票——单次轮换、重放检测与事务竞争Alice 的浏览器和一个窃取了凭据的脚本都持有同一个 refresh token。浏览器先刷新得到新 access token窃取者随后拿旧 refresh 再刷新。如果服务端只检查“这段字符串是否存在、是否过期”两边都可能得到新凭据甚至持续互相接力。这时缩短 access 的生存期没有真正结束被盗会话因为续签入口一直保持开放。本课给 refresh 加上历史它属于哪一次登录是否已经被消费谁是这次轮换的新凭据以及发现重复使用后如何撤销整条会话链。我们还要把“先查再改”的普通 Python 逻辑放进数据库事务否则两个几乎同时到达的刷新请求会把看似正确的单次消费变成两次成功。上一课 E03分别证明主动撤销与自然过期lessons/04/answer_previous.py首先登录、读取身份、退出再拿原样保存的 access 和 refresh 验证二者都不再被接受。第二段用新登录和可推进的时钟证明 access 到期独立生效。运行python lessons/04/answer_previous.py即可验证完整源码位于本篇末尾。本地实际运行得到E03 PASS: logout revokes family; access expires independently这份答案刻意没有在退出后立即清空 Python 变量。用户界面清理变量是必要体验但安全测试要假设攻击者仍保留旧字节。只有服务端在旧凭据再次出现时拒绝才能证明撤销契约成立。以同样方法测试密码重置后的会话策略、设备解绑或人员离职会比“页面看起来退出了”得到更可靠的证据。Access 与 refresh 为什么必须分工access token 直接进入业务接口因此会出现在较多请求路径中暴露面较广。refresh token 只应该发送到刷新入口用于延续会话而不应该被当作普通 Bearer 凭据访问采购业务。分工使 access 可以采用较短生存期并把会话延续操作集中到更严格的检查点。本项目 access 最多有效六百秒且不能超过所属 family 的结束时间refresh family 从创建起绝对存活三千六百秒。所谓“绝对”是说每次刷新不会把家族结束时间无限推向未来。若一小时到期以后仍要继续工作必须重新建立登录会话。滑动过期与绝对过期都是可以设计的策略但必须明确区分否则一个低频脚本也许会让本应短期的会话永不结束。这两个期限并不能互相代替。只把 refresh 设置为很短会频繁打断用户只把 access 设置为很短却允许 refresh 无限制复用仍会让被盗凭据长期存在。本课把短期访问、有限会话寿命与单次刷新组合起来各自承担不同风险。单次轮换要记住“曾经存在过”刷新成功后服务端标记旧 refresh 已消费再签发新的 access、refresh 与 CSRF 数据。客户端必须用新响应替换原来的刷新凭据。再次提交旧 refresh服务端应该知道它不是一段从未见过的随机垃圾而是一张已经被消费、现在又出现的旧票。因此不能一消费就彻底删除所有旧记录。保留消费状态以及关联 family才能把重放追溯到整条会话链。收到完全未知的 refresh只能拒绝它收到已消费且属于某个 family 的 refresh则可以进一步撤销该 family。两者的外部响应都可以是 401但内部需要采取的动作不同。“重放”并不必然说明账户主人正在攻击系统。可能是恶意复制也可能是客户端重试、两个标签页竞争或网络响应丢失。服务器看见的都是同一凭据再次出现不能仅凭请求来源猜出哪个是善意。ProcureSafe 采用严格规则旧 refresh 再次使用就撤销 family。这是清晰、容易验证的教学策略代价是客户端协调不好时会让用户重新登录。生产系统也可能设计短暂的重试容忍窗口但那需要额外回答如何绑定同一次请求、是否返回同一组新凭据、容忍期间被盗凭据能做什么、如何保留检测能力。不能简单写“允许旧 token 再用一次”然后把重放防护视为已经完成。家族撤销为什么要覆盖新 access假设初始刷新凭据叫 R0合法浏览器使用它得到 A1 与 R1。攻击者随后重放 R0。如果服务端只拒绝 R0而仍允许 A1 与 R1就无法结束可能已经分叉的会话链。攻击者也可能是先成功刷新的一方服务器没有可靠依据只撤销它认为“较旧”的一张凭据。family 把这些对象关联到同一次登录根源。发现旧 refresh 重放时系统把 family 标记撤销之后 A1 的认证查询也要检查这个标记因此即使 A1 本身尚未到期访问/auth/me仍得到 401。R1 同样不能继续刷新。这就是本课练习最关键的断言失效对象不仅是被重放的旧 refresh还有刚签出的新访问能力。撤销并不需要把所有原始 token 存起来。服务端保存摘要、家族标识、期限和消费状态即可。原始 token 仍只交给客户端。把关系状态与秘密字节分开既能够跟踪生命周期也避免数据库表直接成为可用凭据清单。两个请求之间只有一毫秒也足以打破普通判断考虑以下顺序请求甲读取 R0发现未消费请求乙也读取 R0发现未消费甲写入“已消费”并发出新 token乙也写入“已消费”并发出另一组 token。如果检查与更新不在同一个受保护事务里每一个函数单独看都先检查后修改整体仍出现双重消费。ProcureSafe 的数据库上下文transaction(database)使用BEGIN IMMEDIATE。对 SQLite 而言它在事务开始时竞争写入资格使会话消费过程不能被另一个写事务插入成两个独立的成功判断。持有事务的请求完成后后来者读取到的已经是已消费状态并执行拒绝与 family 撤销。连接还设置等待时限让短暂竞争可以等待而不是不加解释地立即暴露“数据库锁定”。等待锁还会跨越时间边界。接口把可调用的clock交给服务层事务成功取得写锁后重新采样当前时间再检查会话或家族是否过期。若把路由入口计算好的时间一直传下去请求可能在等待时已经到期拿到锁后却仍按旧时间获准执行。数据库提供一致状态重新采样提供正确的期限判断两者缺一不可。这里的重点不是背诵某条 SQLite 语句。换成 PostgreSQL 时可能采用行锁、条件更新配合受影响行数检查或者其他能在数据库中完成唯一消费的模式。判断是否正确的标准是相同的一个凭据最多被成功消费一次状态更新和新会话记录作为一个事务提交失败时不会留下半组可用凭据。一个容易写错的回滚拒绝之前已经发生了必要状态变化一般业务失败会回滚事务但刷新重放有一项必须保留的安全效果撤销 family。若代码在事务中标记撤销后直接抛出异常而通用事务上下文因此回滚接口虽然返回了 401实际数据库里的家族仍然有效。这是“响应看起来安全状态却不安全”的典型错误。审查完整会话模块时要追踪重放分支如何确保撤销被提交然后再向 HTTP 层报告认证失败。不能只搜索是否出现revoked字样也不能只断言重放请求本身得到 401。E04 会立即拿新 access 和新 refresh 再试一次正是为了验证这项副作用确实留在数据库里。同样的思路还会出现在审计拒绝事件与账户锁定上某些拒绝需要记录事实而成功业务写入应保持原子性。事务边界必须围绕实际要保留的状态设计不能用“所有异常一概回滚”代替领域判断。客户端也必须把刷新当作协调操作多标签页、多个并发业务请求同时发现 access 过期时很容易各自发出 refresh。严格轮换策略下一个请求成功另一个旧凭据重放会撤销整个 family刚拿到新 token 的请求也随之失败。服务端正确拒绝并不能弥补客户端无协调地并发刷新。实际客户端可以把一次刷新收敛到共享任务让等待中的业务请求复用同一结果跨标签页还需要适当的通信或集中会话架构。刷新发生网络超时后也不能不加区分地无限重试旧凭据因为服务器也许已经提交成功只是响应丢了。重试与轮换的组合必须在产品策略中明确约定。本课程浏览器章节不实现完整刷新代理也不暗示返回一组 JSON token 就等于完成生产 BFF。这里先把令牌端点协议与数据库生命周期讲清楚第五课再把浏览器自动携带 Cookie 的另一类风险独立分析。本课练习 E04重放后新凭据也必须失败让 Alice 登录保存 R0调用/auth/refresh消费 R0取得新的 access A1 与 refresh R1。断言刷新成功并用 A1 调用/auth/me确认它此时确实有效。然后再次提交完全相同的 R0断言 401。接下来用 A1 再次请求身份端点必须是 401再用 R1 刷新也必须是 401。最后独立登录一次确认新 family 能访问身份端点证明撤销范围是旧会话链而不是错误地禁用了整个 Alice 账号。输出只保留性质摘要。下一篇将完整给出 E04 答案并把相同的认证能力放进浏览器。届时我们会发现一个足够随机、可撤销的 Cookie 仍然可能被其他站点诱导自动发送令牌生命周期正确并不能替代请求来源与 CSRF 检查。上一课答案源码与本课完整事务、会话模块E03 完整答案保留旧凭据原字节分别验证撤销与自然到期。frompathlibimportPathimportsys sys.path.insert(0,str(Path(__file__).resolve().parents[1]))fromsupportimportFakeClock,login,sandboxdefmain():clockFakeClock()withsandbox(clockclock)as(client,credentials):tokenslogin(client,credentials,alice)old_headers{Authorization:Bearer tokens[access_token]}beforeclient.get(/auth/me,headersold_headers)assertbefore.status_code200andbefore.json()[username]aliceassertclient.post(/auth/logout,headersold_headers).status_code204# 不清空本地变量假设窃取者仍有旧字节服务端仍必须拒绝。assertclient.get(/auth/me,headersold_headers).status_code401assertclient.post(/auth/refresh,json{refresh_token:tokens[refresh_token]}).status_code401independentlogin(client,credentials,alice)current_headers{Authorization:Bearer independent[access_token]}assertclient.get(/auth/me,headerscurrent_headers).status_code200clock.advance(independent[expires_in]1)assertclient.get(/auth/me,headerscurrent_headers).status_code401print(E03 PASS: logout revokes family; access expires independently)if__name____main__:main()少量显式SQL让安全边界可见所有写事务从BEGIN IMMEDIATE开始。fromcontextlibimportcontextmanagerimportsqlite3 SCHEMA CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0,1)), platform_admin INTEGER NOT NULL DEFAULT 0 CHECK(platform_admin IN (0,1)), failed_attempts INTEGER NOT NULL DEFAULT 0, locked_until REAL NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS tenants (id TEXT PRIMARY KEY, name TEXT NOT NULL); CREATE TABLE IF NOT EXISTS memberships ( user_id TEXT NOT NULL REFERENCES users(id), tenant_id TEXT NOT NULL REFERENCES tenants(id), department TEXT NOT NULL CHECK(department IN (engineering,sales)), roles TEXT NOT NULL CHECK(json_valid(roles)), PRIMARY KEY(user_id,tenant_id) ); CREATE TABLE IF NOT EXISTS requests ( id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL, owner_id TEXT NOT NULL, department TEXT NOT NULL, title TEXT NOT NULL, amount_cents INTEGER NOT NULL CHECK(typeof(amount_cents)integer AND amount_cents BETWEEN 1 AND 100000000), status TEXT NOT NULL CHECK(status IN (submitted,approved,ordered,paid)), version INTEGER NOT NULL CHECK(version 1), FOREIGN KEY(owner_id,tenant_id) REFERENCES memberships(user_id,tenant_id) ); CREATE INDEX IF NOT EXISTS request_scope ON requests(tenant_id,department,owner_id); CREATE TABLE IF NOT EXISTS families ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL REFERENCES users(id), expires_at REAL NOT NULL, revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0,1)) ); CREATE TABLE IF NOT EXISTS access_tokens ( token_hash TEXT PRIMARY KEY, family_id TEXT NOT NULL REFERENCES families(id), expires_at REAL NOT NULL, csrf_hash TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS refresh_tokens ( token_hash TEXT PRIMARY KEY, family_id TEXT NOT NULL REFERENCES families(id), used INTEGER NOT NULL DEFAULT 0 CHECK(used IN (0,1)) ); CREATE TABLE IF NOT EXISTS audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, tenant_id TEXT NOT NULL REFERENCES tenants(id), time REAL NOT NULL, actor TEXT NOT NULL, action TEXT NOT NULL, resource_id TEXT, result TEXT NOT NULL CHECK(result IN (success,denied)), request_id TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS external_identities ( issuer TEXT NOT NULL, subject TEXT NOT NULL, user_id TEXT NOT NULL REFERENCES users(id), PRIMARY KEY(issuer,subject) ); CREATE TABLE IF NOT EXISTS oidc_states ( state_hash TEXT PRIMARY KEY, binding_hash TEXT NOT NULL, nonce TEXT NOT NULL, verifier TEXT NOT NULL, expires_at REAL NOT NULL, used INTEGER NOT NULL DEFAULT 0 ); defconnect(database:str)-sqlite3.Connection:每次工作单元持有独立连接避免线程共享连接掩盖并发问题。connsqlite3.connect(database,isolation_levelNone,timeout10)conn.row_factorysqlite3.Row conn.execute(PRAGMA foreign_keysON)conn.execute(PRAGMA busy_timeout10000)returnconndefinitialize(database:str)-None:connconnect(database)try:conn.execute(PRAGMA journal_modeWAL)conn.executescript(SCHEMA)finally:conn.close()contextmanagerdeftransaction(database:str):先抢写锁再读取状态让并发刷新和审批看到已提交的最新状态。connconnect(database)try:conn.execute(BEGIN IMMEDIATE)yieldconn conn.commit()exceptBaseException:conn.rollback()raisefinally:conn.close()contextmanagerdefsnapshot(database:str):同一读事务内读取会话、成员与资源避免一次请求拼接不同快照。connconnect(database)try:conn.execute(BEGIN)yieldconnfinally:conn.rollback()conn.close()Opaque令牌只存hash权限在每次请求时读取不把角色冻结进令牌。fromdataclassesimportdataclassimporthashlibimportsecretsfromtypingimportCallablefrom.dbimportsnapshot,transactionfrom.errorsimportDomainErrorfrom.passwordsimporthash_password,verify_password ACCESS_TTL,REFRESH_TTL600,3600LOCK_SECONDS,MAX_FAILURES60,5# 未知用户也走同等scrypt计算。统一响应并不等于已解决所有时间侧信道。_DUMMY_HASHhash_password(secrets.token_urlsafe(32))deftoken_digest(value:str)-str:returnhashlib.sha256(value.encode(utf-8)).hexdigest()dataclass(frozenTrue)classPrincipal:user_id:strusername:strfamily_id:strtoken_hash:strcsrf_hash:strdefissue_tokens(conn,user_id:str,now:float,family_id:str|NoneNone)-dict:调用者持有写事务刷新沿用family绝对过期时间不能无限续命。iffamily_idisNone:family_idsecrets.token_urlsafe(24)conn.execute(INSERT INTO families(id,user_id,expires_at) VALUES(?,?,?),(family_id,user_id,nowREFRESH_TTL))familyconn.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(nowACCESS_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:tokensNonewithtransaction(database)asconn:userconn.execute(SELECT * FROM users WHERE username?,(username,)).fetchone()validverify_password(password,user[password_hash]ifuserelse_DUMMY_HASH)nowclock()# 排队和密码计算都可能消耗时间不能沿用请求到达时刻。ifuseranduser[enabled]anduser[locked_until]now:failures0ifuser[locked_until]elseuser[failed_attempts]ifvalid:conn.execute(UPDATE users SET failed_attempts0,locked_until0 WHERE id?,(user[id],))tokensissue_tokens(conn,user[id],now)else:failures1untilnowLOCK_SECONDSiffailuresMAX_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:rowconn.execute(SELECT u.id,u.username,a.family_id,a.csrf_hash FROM access_tokens a JOIN families f ON f.ida.family_id JOIN users u ON u.idf.user_id WHERE a.token_hash? AND a.expires_at? AND f.expires_at? AND f.revoked0 AND u.enabled1,(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:tokensNonewithtransaction(database)asconn:nowclock()# BEGIN IMMEDIATE可能等待取锁后再判断family绝对期限。rowconn.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.idr.family_id JOIN users u ON u.idf.user_id WHERE r.token_hash?,(token_digest(token),)).fetchone()ifrowandrow[used]:# 重放的撤销必须提交不能在当前事务里直接raise导致撤销回滚。conn.execute(UPDATE families SET revoked1 WHERE id?,(row[family_id],))elifrowandnotrow[revoked]androw[enabled]androw[expires_at]now:conn.execute(UPDATE refresh_tokens SET used1 WHERE token_hash?,(token_digest(token),))tokensissue_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:currentauthenticate_connection(conn,principal.token_hash,clock())conn.execute(UPDATE families SET revoked1 WHERE id?,(current.family_id,))专题目录《认证、授权与企业级权限系统课程目录》 · 上一篇《Python后端权限系统03登录之后还要继续核验——不透明令牌、过期与退出登录》 · 下一篇《Python后端权限系统05Cookie 会自己跟着请求走——Origin、CSRF 与浏览器凭据边界》
