refactor: session 改用 itsdangerous,去掉手写密码学原语

为什么: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 拒绝。
This commit is contained in:
snowy-yang committed 2026-09-26 18:53:47 +08:00
1 parent 0b25f215a2
commit 2b4704183c
2 files changed
+63 -47

No files matched your search

+39 -47
View File
@@ -1,81 +1,73 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""Flask 兼容的会话 Cookie 编解码(itsdangerous 签名格式)。 """会话 Cookie 的编解码(itsdangerous 签名格式,与 Flask 会话同格式)。
Flask 默认会话即 itsdangerous 签名 Cookie(`session` cookie,格式 `payload.timestamp.hmac`, 存量兼容:本站早期由 Flask 签发会话(`session` cookie,格式
payload 为会话字典的 compact JSON;bytes/datetime/tuple 等类型会被 Flask 打上 `payload.timestamp.hmac`;盐 `cookie-session`;HMAC-SHA1;密钥派生 `hmac`),
单字符标签再包一层)。本站早期由 Flask 签发会话,为让**存量 Cookie 无需重新登录即可继续使用**, 为让**存量 Cookie 无需重新登录即可继续使用**,此处直接调用 itsdangerous——
本模块复刻同一套编解码(同格式、同密钥、同摘要算法)。 与 Flask `SecureCookieSessionInterface.get_signing_serializer` 使用同一套参数,
格式逐字节一致,不再手写复刻签名算法(手写密码学原语是负债:难审、易错、
且必须靠测试钉住与上游的字节级一致)。
密钥同源:取自 auth_system.get_secret_key()(data/secret.key)。 密钥同源:取自 auth_system.get_secret_key()(data/secret.key)。
""" """
import base64
import hashlib import hashlib
import hmac from functools import lru_cache
import json
import time
from typing import Any from typing import Any
from itsdangerous import BadData, URLSafeTimedSerializer
# 与 Flask 默认一致:SHA1 + itsdangerous 的 key derivation(HMAC(digest, salt)) # 与 Flask 默认一致:SHA1 + itsdangerous 的 key derivation(HMAC(digest, salt))
_SALT = 'cookie-session' _SALT = 'cookie-session'
_DIGEST = hashlib.sha1 _SIGNER_KWARGS: dict[str, Any] = {'key_derivation': 'hmac', 'digest_method': hashlib.sha1}
SESSION_KEY = 'site_uid' SESSION_KEY = 'site_uid'
def _derive_key(secret: str, salt: str = _SALT) -> bytes: @lru_cache(maxsize=8)
"""itsdangerous 的密钥派生:HMAC(secret, salt)。""" def _serializer(secret: str) -> URLSafeTimedSerializer:
mac = hmac.new(secret.encode('utf-8'), salt.encode('utf-8'), _DIGEST) """构造与 Flask 会话同参数的序列化器(盐 / 派生 / 摘要三项对齐)。
return mac.digest()
按密钥缓存:`loads` 位于每个请求的会话解析路径上,每次新建序列化器是
def _int_to_bytes(num: int) -> bytes: 无谓开销;密钥种类极少(通常只有 data/secret.key 一个),maxsize=8 足够。
"""itsdangerous 时间戳编码:8 字节大端并去前导零(struct.pack('>Q') + lstrip)。""" """
return num.to_bytes(8, 'big').lstrip(b'\x00') or b'\x00' return URLSafeTimedSerializer(secret, salt=_SALT, signer_kwargs=dict(_SIGNER_KWARGS))
def dumps(secret: str, session: dict[str, Any]) -> str: def dumps(secret: str, session: dict[str, Any]) -> str:
"""把会话字典序列化为 Flask/itsdangerous 兼容的签名 Cookie 值。""" """把会话字典序列化为 Flask/itsdangerous 兼容的签名 Cookie 值。"""
payload = base64.b64encode( return _serializer(secret).dumps(session)
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("=")}'
def loads(secret: str, value: str) -> dict[str, Any] | None: def loads(secret: str, value: str, max_age: int | None = None) -> dict[str, Any] | None:
"""校验签名并解出会话字典;无效/被篡改返回 None。 """校验签名并解出会话字典;无效 / 被篡改 / 超龄返回 None。
注:Flask 的会话 JSON 采用「带空格」的 compact 编码(`{"a": 1}`), ``max_age=None``(默认)不校验签发时间——与本站既有行为一致:会话有效期由
解码时直接 json.loads 即可,两种空白形式都能解析。 Cookie 自身的 max_age(登录时 7 天,见 web.routers.register._SESSION_MAX_AGE)
与库中 login_seq 顶号校验共同约束,服务端不叠加第二层独立过期判定;
传入秒数即启用 itsdangerous 的签名时间戳校验。
""" """
if not value: if not value:
return None return None
try: try:
payload, ts, sig = value.rsplit('.', 2) data = _serializer(secret).loads(value, max_age=max_age)
except ValueError: except BadData:
return None # BadSignature / SignatureExpired / BadPayload —— 一律视为无效会话
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:
return None return None
return data if isinstance(data, dict) else None return data if isinstance(data, dict) else None
def get_uid(session: dict[str, Any] | None) -> int | None: def get_uid(session: dict[str, Any] | None) -> int | None:
"""取会话中的站点用户 id(Flask 存的是 int,直接可用)。""" """取会话中的站点用户 id(Flask 存的是 int,直接可用)。
非整数一律按未登录处理:签名有效但字段畸形只可能来自密钥泄露后的伪造,
此时返回 None(当普通未登录)比抛 ValueError 触发 500 更安全。
"""
if not session: if not session:
return None return None
val = session.get(SESSION_KEY) 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
+24
View File
@@ -65,6 +65,30 @@ def test_roundtrip_selfconsistency():
assert cs.loads(SECRET, cs.dumps(SECRET, payload)) == payload 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(): def test_get_uid_none():
assert cs.get_uid(None) is None assert cs.get_uid(None) is None
assert cs.get_uid({}) is None assert cs.get_uid({}) is None