# -*- coding: utf-8 -*- """Flask 兼容的会话 Cookie 编解码(itsdangerous 签名格式)。 Flask 默认会话即 itsdangerous 签名 Cookie(`session` cookie,格式 `payload.timestamp.hmac`, payload 为会话字典的 compact JSON;bytes/datetime/tuple 等类型会被 Flask 打上 单字符标签再包一层)。本站早期由 Flask 签发会话,为让**存量 Cookie 无需重新登录即可继续使用**, 本模块复刻同一套编解码(同格式、同密钥、同摘要算法)。 密钥同源:取自 auth_system.get_secret_key()(data/secret.key)。 """ import base64 import hashlib import hmac import json import time from typing import Any # 与 Flask 默认一致:SHA1 + itsdangerous 的 key derivation(HMAC(digest, salt)) _SALT = 'cookie-session' _DIGEST = 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() 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' 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("=")}' def loads(secret: str, value: str) -> dict[str, Any] | None: """校验签名并解出会话字典;无效/被篡改返回 None。 注:Flask 的会话 JSON 采用「带空格」的 compact 编码(`{"a": 1}`), 解码时直接 json.loads 即可,两种空白形式都能解析。 """ 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: return None return data if isinstance(data, dict) else None def get_uid(session: dict[str, Any] | None) -> int | None: """取会话中的站点用户 id(Flask 存的是 int,直接可用)。""" if not session: return None val = session.get(SESSION_KEY) return int(val) if val else None