Archived
snowy-yang
2c55b55e03
会话互通（迁移关键）： - core/session.py：itsdangerous 兼容的签名 Cookie 编解码（同密钥/同 salt/同 时间戳编码），Flask 与 FastAPI 双向可读——实测 F->core 与 core->F 双向解出一致， 篡改/错密钥返回 None FastAPI 基础设施： - web/deps.py 会话→当前用户（封禁/顶号校验，口径同 auth_system.current_user） - web/errors.py AppError + 全局异常处理，404/405/400 统一为本站包体 - web/ratelimit.py 复用 core.state 同一批限流桶（跨栈共享，避免绕框架绕过限流） - core/repo.py 补 get_user_by_id（异步） 迁移路由（/api/v1/*，与 Flask 同端点逐字段对齐）： - toplist / artist_songs / comments / search，走 music_api 异步方法 - search 的服务器昵称遮蔽改为 FastAPI 侧异步实现（不再依赖 main） 契约奇偶校验：扩展至 7 项（含登录会话互通 + 未登录 401），全 OK； CDN 分片主机差异归一化（上游易变值，非契约漂移）。 验证：pytest 33/33；ruff All checks passed；pyright 0 errors； uvicorn 实跑未登录 401 包体与 Flask 一致。
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Flask 会话 Cookie 与 Starlette/FastAPI 的互操作。
|
||
|
||
Flask 默认会话即 itsdangerous 签名 Cookie（`session` cookie，格式 `payload.timestamp.hmac`，
|
||
payload 为 compact JSON、值类型化 `{"site_uid":{" b":i"4":".."}}`）。迁移期要让
|
||
「Flask 签发的会话在 FastAPI 侧可读、FastAPI 刷新后可被 Flask 读回」，
|
||
本模块给出同一套编解码，避免两栈各签一份、互相看不懂。
|
||
|
||
密钥同源：都取自 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
|