Archived
snowy-yang
1973d6a0e8
为什么：签发侧 Cookie 的 max_age 是 7 天，但那只是**浏览器**行为——被保存、 复制或从日志里捞出来的旧 Cookie 仍带着有效签名，而解析侧 loads 从不校验签发 时间（max_age=None），等于「登录态永不过期」。登出/顶号靠递增 login_seq 兜底， 但拿不到库的一方（如泄露的 Cookie）无从失效。 改法： - core/session.py 新增 COOKIE_NAME / SESSION_MAX_AGE，作为会话名与有效期的 唯一来源。此前 'session' 在 web/deps、web/routers/auth、web/routers/register 各写一遍，7 天在两处各写一遍——三处同名常量各改各的，正是漂移温床。 - web/deps._parse_cookie 传入 max_age=SESSION_MAX_AGE，过期判定落在服务端。 - 两个签发侧改引用共享常量（保留模块级别名，对外名字不变）。 - loads() 的签名与语义**不变**（max_age=None 仍不校验）：它是通用工具，策略 属于调用点，这样 test_session 里「None 接受旧时间戳」的语义继续成立。 行为变更（本轮唯一一处）：签发时间超过 7 天的会话 Cookie 现在按未登录处理。 正常浏览器用户无感——浏览器本来就按 max_age 丢弃它；受影响的是手工保存、 跨 7 天复用的旧 Cookie，正是要堵的那类。 验证： - 新增 tests/test_deps.py::test_expired_cookie_rejected：用 monkeypatch.context() 把时钟拨回 SESSION_MAX_AGE 之前签发（itsdangerous 时间戳取自 dumps 时刻）， 再以当前时间解析 → 必须拒绝；同时断言有效期内不受影响。用 context 而非 undo()，后者会连带撤销 site fixture 的设置。 - 变异验证：把 max_age 参数摘掉后**恰好**该用例变红、其余 10 例全绿——证明测试 精确盯着这一处，不是碰巧通过。 - ruff 全过 · pyright 0/0 · pytest 214 passed（213→214）。golden 快照与 test_download_billing（真实登录拿 Cookie 打 API）均通过，端到端链路未受影响。
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""FastAPI 依赖：会话解析、当前用户、登录校验。
|
|
|
|
与 auth_system 的语义对齐（共用同一 Cookie、同一站点库）：
|
|
- 会话 Cookie 经 core.session 解码（Flask 兼容签名），密钥同源
|
|
- 过期：签发时间超过 core.session.SESSION_MAX_AGE 视为失效（服务端把关，
|
|
不依赖浏览器是否还留着 Cookie）
|
|
- 封禁账号（status 0/2）：按未登录处理
|
|
- 顶号校验：会话内的 `site_uid_seq` 必须与库中 `login_seq` 一致，否则视为失效
|
|
- 剩余次数/账本预扣属付费流程，由 web.billing 承担（此处只做身份解析）
|
|
"""
|
|
from dataclasses import dataclass
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import Cookie, Depends
|
|
|
|
from core import repo
|
|
from core import session as sess
|
|
from web.errors import AppError
|
|
|
|
_SESSION_COOKIE = sess.COOKIE_NAME
|
|
|
|
|
|
def secret_key() -> str:
|
|
"""会话签名密钥（密钥来源：data/secret.key）。"""
|
|
import auth_system
|
|
return auth_system.get_secret_key()
|
|
|
|
|
|
@dataclass
|
|
class SiteUser:
|
|
id: int
|
|
username: str
|
|
role: str
|
|
credits: int
|
|
|
|
@property
|
|
def is_admin(self) -> bool:
|
|
return self.role == 'admin'
|
|
|
|
@property
|
|
def unlimited(self) -> bool:
|
|
return self.is_admin or self.credits == -1
|
|
|
|
|
|
def _parse_cookie(session_cookie: str | None) -> dict[str, Any] | None:
|
|
# 服务端叠加签发时间校验（SESSION_MAX_AGE）：Cookie 自身的 max_age 只约束
|
|
# 浏览器，被存下来的旧 Cookie 仍能通过签名校验，故过期必须由服务端把关。
|
|
return sess.loads(secret_key(), session_cookie or '', max_age=sess.SESSION_MAX_AGE)
|
|
|
|
|
|
async def current_user(
|
|
session_cookie: Annotated[str | None, Cookie(alias=_SESSION_COOKIE)] = None,
|
|
) -> SiteUser | None:
|
|
"""当前登录用户（未登录/失效返回 None）。"""
|
|
data = _parse_cookie(session_cookie)
|
|
uid = sess.get_uid(data)
|
|
if not uid:
|
|
return None
|
|
row = await repo.get_user_by_id(uid)
|
|
if row is None:
|
|
return None
|
|
status = row.get('status', 1)
|
|
if status in (0, 2): # 禁用/封禁：按未登录处理
|
|
return None
|
|
# 顶号校验：库有 login_seq 时会话版本必须严格一致
|
|
db_seq = int(row.get('login_seq') or 0)
|
|
sess_seq = int((data or {}).get(sess.SESSION_KEY + '_seq') or 0)
|
|
if db_seq and sess_seq != db_seq:
|
|
return None
|
|
return SiteUser(
|
|
id=int(row['id']),
|
|
username=row['username'],
|
|
role=row['role'],
|
|
credits=int(row.get('duration_seconds') or 0),
|
|
)
|
|
|
|
|
|
async def require_login(
|
|
user: Annotated[SiteUser | None, Depends(current_user)],
|
|
) -> SiteUser:
|
|
"""登录闸：未登录抛 401，已登录返回用户。"""
|
|
if user is None:
|
|
raise AppError('请先登录后再使用解析功能', 401, error_code='unauthorized')
|
|
return user
|
|
|
|
|
|
async def require_admin(
|
|
user: Annotated[SiteUser | None, Depends(current_user)],
|
|
) -> SiteUser:
|
|
"""管理员闸：非管理员抛 403。"""
|
|
if user is None or not user.is_admin:
|
|
raise AppError('需要管理员权限', 403, error_code='forbidden')
|
|
return user
|