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():clock=FakeClock()withsandbox(clock=clock)as(client,credentials):tokens=login(client,credentials,"alice")old_headers={"Authorization":"Bearer "+tokens["access_token"]}before=client.get("/auth/me",headers=old_headers)assertbefore.status_code==200andbefore.json()["username"]=="alice"assertclient.post("/auth/logout",headers=old_headers).status_code==204# 不清空本地变量：假设窃取者仍有旧字节，服务端仍必须拒绝。assertclient.get("/auth/me",headers=old_headers).status_code==401assertclient.post("/auth/refresh",json={"refresh_token":tokens["refresh_token"]}).status_code==401independent=login(client,credentials,"alice")current_headers={"Authorization":"Bearer "+independent["access_token"]}assertclient.get("/auth/me",headers=current_headers).status_code==200clock.advance(independent["expires_in"]+1)assertclient.get("/auth/me",headers=current_headers).status_code==401print("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:"""每次工作单元持有独立连接，避免线程共享连接掩盖并发问题。"""conn=sqlite3.connect(database,isolation_level=None,timeout=10)conn.row_factory=sqlite3.Row conn.execute("PRAGMA foreign_keys=ON")conn.execute("PRAGMA busy_timeout=10000")returnconndefinitialize(database:str)->None:conn=connect(database)try:conn.execute("PRAGMA journal_mode=WAL")conn.executescript(SCHEMA)finally:conn.close()@contextmanagerdeftransaction(database:str):"""先抢写锁再读取状态，让并发刷新和审批看到已提交的最新状态。"""conn=connect(database)try:conn.execute("BEGIN IMMEDIATE")yieldconn conn.commit()exceptBaseException:conn.rollback()raisefinally:conn.close()@contextmanagerdefsnapshot(database:str):"""同一读事务内读取会话、成员与资源，避免一次请求拼接不同快照。"""conn=connect(database)try:conn.execute("BEGIN")yieldconnfinally:conn.rollback()conn.close()

"""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后端权限系统03：登录之后还要继续核验——不透明令牌、过期与退出登录》 · 下一篇：《Python后端权限系统05：Cookie 会自己跟着请求走——Origin、CSRF 与浏览器凭据边界》