From 2b4704183c326ea6e3ccaffb8cb9ca865f17f1bb Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 18:53:47 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20session=20=E6=94=B9=E7=94=A8=20itsd?= =?UTF-8?q?angerous=EF=BC=8C=E5=8E=BB=E6=8E=89=E6=89=8B=E5=86=99=E5=AF=86?= =?UTF-8?q?=E7=A0=81=E5=AD=A6=E5=8E=9F=E8=AF=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：core/session.py 原本手写 HMAC + 自定义 base64 签名，等于自己复刻 itsdangerous 的核心功能——密码学原语自己实现是风险点，且难与上游的安全 修复同步。 关键约束（换库的坑）：两者字节级格式并不一致——手写实现用标准 base64 且 保留 '=' 填充，itsdangerous 用 urlsafe 且去填充。若不钉住旧格式，部署瞬间 线上全部登录态会失效。 改法： - 用 URLSafeTimedSerializer，按 secret 做 lru_cache(maxsize=8)，避免每请求 重建序列化器 - loads() 增加 max_age 参数（默认 None，保持既有「不叠加服务端过期」行为） - get_uid() 顺带加固：cookie 里塞非整数不再抛 ValueError 变 500，改为返回 None - 删除 _derive_key / _int_to_bytes 及手写 HMAC/b64 验证：新增 2 例回归测试，其中 legacy cookie 样本由替换前的旧实现在固定时间戳 1700000000 下采集，钉住的是格式而非实现；另一例验证 max_age=None 接受旧 时间戳、max_age=60 拒绝。 --- core/session.py | 86 ++++++++++++++++++++----------------------- tests/test_session.py | 24 ++++++++++++ 2 files changed, 63 insertions(+), 47 deletions(-) diff --git a/core/session.py b/core/session.py index 6930005..e3c1159 100644 --- a/core/session.py +++ b/core/session.py @@ -1,81 +1,73 @@ # -*- coding: utf-8 -*- -"""Flask 兼容的会话 Cookie 编解码（itsdangerous 签名格式）。 +"""会话 Cookie 的编解码（itsdangerous 签名格式，与 Flask 会话同格式）。 -Flask 默认会话即 itsdangerous 签名 Cookie（`session` cookie，格式 `payload.timestamp.hmac`， -payload 为会话字典的 compact JSON；bytes/datetime/tuple 等类型会被 Flask 打上 -单字符标签再包一层）。本站早期由 Flask 签发会话，为让**存量 Cookie 无需重新登录即可继续使用**， -本模块复刻同一套编解码（同格式、同密钥、同摘要算法）。 +存量兼容：本站早期由 Flask 签发会话（`session` cookie，格式 +`payload.timestamp.hmac`；盐 `cookie-session`；HMAC-SHA1；密钥派生 `hmac`）， +为让**存量 Cookie 无需重新登录即可继续使用**，此处直接调用 itsdangerous—— +与 Flask `SecureCookieSessionInterface.get_signing_serializer` 使用同一套参数， +格式逐字节一致，不再手写复刻签名算法（手写密码学原语是负债：难审、易错、 +且必须靠测试钉住与上游的字节级一致）。 密钥同源：取自 auth_system.get_secret_key()（data/secret.key）。 """ -import base64 import hashlib -import hmac -import json -import time +from functools import lru_cache from typing import Any +from itsdangerous import BadData, URLSafeTimedSerializer + # 与 Flask 默认一致：SHA1 + itsdangerous 的 key derivation（HMAC(digest, salt)） _SALT = 'cookie-session' -_DIGEST = hashlib.sha1 +_SIGNER_KWARGS: dict[str, Any] = {'key_derivation': 'hmac', 'digest_method': hashlib.sha1} SESSION_KEY = 'site_uid' -def _derive_key(secret: str, salt: str = _SALT) -> bytes: - """itsdangerous 的密钥派生：HMAC(secret, salt)。""" - mac = hmac.new(secret.encode('utf-8'), salt.encode('utf-8'), _DIGEST) - return mac.digest() +@lru_cache(maxsize=8) +def _serializer(secret: str) -> URLSafeTimedSerializer: + """构造与 Flask 会话同参数的序列化器（盐 / 派生 / 摘要三项对齐）。 - -def _int_to_bytes(num: int) -> bytes: - """itsdangerous 时间戳编码：8 字节大端并去前导零（struct.pack('>Q') + lstrip）。""" - return num.to_bytes(8, 'big').lstrip(b'\x00') or b'\x00' + 按密钥缓存：`loads` 位于每个请求的会话解析路径上，每次新建序列化器是 + 无谓开销；密钥种类极少（通常只有 data/secret.key 一个），maxsize=8 足够。 + """ + return URLSafeTimedSerializer(secret, salt=_SALT, signer_kwargs=dict(_SIGNER_KWARGS)) def dumps(secret: str, session: dict[str, Any]) -> str: """把会话字典序列化为 Flask/itsdangerous 兼容的签名 Cookie 值。""" - payload = base64.b64encode( - json.dumps(session, separators=(',', ':')).encode('utf-8')).decode('ascii') - ts = base64.b64encode(_int_to_bytes(int(time.time()))).decode('ascii').rstrip('=') - value = f'{payload}.{ts}' - sig = hmac.new(_derive_key(secret), value.encode('utf-8'), _DIGEST).digest() - return f'{value}.{base64.b64encode(sig).decode("ascii").rstrip("=")}' + return _serializer(secret).dumps(session) -def loads(secret: str, value: str) -> dict[str, Any] | None: - """校验签名并解出会话字典；无效/被篡改返回 None。 +def loads(secret: str, value: str, max_age: int | None = None) -> dict[str, Any] | None: + """校验签名并解出会话字典；无效 / 被篡改 / 超龄返回 None。 - 注：Flask 的会话 JSON 采用「带空格」的 compact 编码（`{"a": 1}`）， - 解码时直接 json.loads 即可，两种空白形式都能解析。 + ``max_age=None``（默认）不校验签发时间——与本站既有行为一致：会话有效期由 + Cookie 自身的 max_age（登录时 7 天，见 web.routers.register._SESSION_MAX_AGE） + 与库中 login_seq 顶号校验共同约束，服务端不叠加第二层独立过期判定； + 传入秒数即启用 itsdangerous 的签名时间戳校验。 """ if not value: return None try: - payload, ts, sig = value.rsplit('.', 2) - except ValueError: - return None - value_no_sig = f'{payload}.{ts}' - expected = hmac.new(_derive_key(secret), value_no_sig.encode('utf-8'), _DIGEST).digest() - pad = '=' * (-len(sig) % 4) - try: - actual = base64.urlsafe_b64decode(sig + pad) - except Exception: - return None - if not hmac.compare_digest(expected, actual): - return None - pad_p = '=' * (-len(payload) % 4) - try: - raw = base64.urlsafe_b64decode(payload + pad_p) - data = json.loads(raw.decode('utf-8')) - except Exception: + data = _serializer(secret).loads(value, max_age=max_age) + except BadData: + # BadSignature / SignatureExpired / BadPayload —— 一律视为无效会话 return None return data if isinstance(data, dict) else None def get_uid(session: dict[str, Any] | None) -> int | None: - """取会话中的站点用户 id（Flask 存的是 int，直接可用）。""" + """取会话中的站点用户 id（Flask 存的是 int，直接可用）。 + + 非整数一律按未登录处理：签名有效但字段畸形只可能来自密钥泄露后的伪造， + 此时返回 None（当普通未登录）比抛 ValueError 触发 500 更安全。 + """ if not session: return None val = session.get(SESSION_KEY) - return int(val) if val else None + if not val: + return None + try: + return int(val) + except (TypeError, ValueError): + return None diff --git a/tests/test_session.py b/tests/test_session.py index a56bda1..67bccb3 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -65,6 +65,30 @@ def test_roundtrip_selfconsistency(): assert cs.loads(SECRET, cs.dumps(SECRET, payload)) == payload +def test_legacy_handwritten_cookie_still_loads(): + """旧手写实现签发的 Cookie 仍能被解析（存量会话不必重新登录）。 + + 样本由「替换 itsdangerous 之前」的 core.session.dumps 生成，时间戳固定为 + 1700000000，故字符串永久有效。它钉住的是**格式**而非实现：旧实现的 + payload 用标准 base64 且保留 `=` 填充，签名同理；itsdangerous 用 urlsafe + 且去掉填充——两侧必须继续互通，否则线上所有登录态会在部署瞬间失效。 + """ + legacy = ('eyJzaXRlX3VpZCI6MjYsInNpdGVfdWlkX3NlcSI6NX0=' + '.ZVPxAA.hW2ZBQ0BL1OhI4D4eDKFFVvqGO0') + data = cs.loads(SECRET, legacy) + assert data == {'site_uid': 26, 'site_uid_seq': 5} + assert cs.get_uid(data) == 26 + + +def test_max_age_none_accepts_old_timestamp(): + """默认不校验签发时间（max_age=None）——与既有行为一致，旧 Cookie 不过期。""" + legacy = ('eyJzaXRlX3VpZCI6MjYsInNpdGVfdWlkX3NlcSI6NX0=' + '.ZVPxAA.hW2ZBQ0BL1OhI4D4eDKFFVvqGO0') + assert cs.loads(SECRET, legacy) is not None + # 显式给出 max_age 时，超龄的旧 Cookie 必须被拒绝（供未来启用过期用） + assert cs.loads(SECRET, legacy, max_age=60) is None + + def test_get_uid_none(): assert cs.get_uid(None) is None assert cs.get_uid({}) is None