feat(#150): 会话管理 session.py(HMAC 签名 token + 恒定时间校验,HttpOnly cookie)

This commit is contained in:
2026-08-05 02:05:50 +00:00
parent b5b29d28d7
commit a1acc6e63b
+97
View File
@@ -0,0 +1,97 @@
# -*- coding: utf-8 -*-
"""④-1 会话管理:登录 token 签发/校验 —— issue #150 / PRD 8.2。
无状态会话令牌(self-contained token),服务端无需存 session 表即可校验,
适合配置台这种轻量场景:
- 令牌格式:`<user_id>.<expire_epoch>.<hmac_sha256_sig>`(base64url 友好);
- 签名密钥来自部署环境(`IAOP_AUTH_SECRET`),未配置时随机生成(进程内有效,
重启失效——生产必须配置固定密钥以支持多副本);
- 校验恒定时间(`hmac.compare_digest`),过期/签名不符一律判无效;
- `SESSION_COOKIE` 为会话 cookie 名(`iaop_session`,HttpOnly + SameSite=Lax)。
PRD 8.2 验收口径:未携带有效 token 的请求,写操作一律拒绝(见 auth_api.py)。
"""
from __future__ import annotations
import base64
import hmac
import hashlib
import os
import time
from dataclasses import dataclass
from typing import Optional, Tuple
SESSION_COOKIE = "iaop_session"
DEFAULT_TTL = 8 * 3600 # 8 小时
HASH_BYTES = 32
def _b64url(b: bytes) -> str:
return base64.urlsafe_b64encode(b).rstrip(b"=").decode("ascii")
def _b64url_decode(s: str) -> bytes:
pad = "=" * (-len(s) % 4)
return base64.urlsafe_b64decode(s + pad)
def _secret() -> bytes:
"""会话签名密钥。优先环境变量,否则进程内随机(重启失效)。"""
env = os.environ.get("IAOP_AUTH_SECRET")
if env:
return env.encode("utf-8")
if not hasattr(_secret, "_fallback"):
_secret._fallback = os.urandom(32) # type: ignore[attr-defined]
return _secret._fallback # type: ignore[attr-defined]
def _sign(payload: bytes) -> str:
sig = hmac.new(_secret(), payload, hashlib.sha256).digest()
return _b64url(sig)
@dataclass(frozen=True)
class Session:
user_id: int
expire_at: int
@property
def expired(self) -> bool:
return time.time() >= self.expire_at
def issue_token(user_id: int, *, ttl: int = DEFAULT_TTL) -> str:
"""为 user_id 签发会话 token。"""
if not isinstance(user_id, int) or user_id <= 0:
raise ValueError("user_id must be a positive int")
expire_at = int(time.time()) + ttl
payload = "%d.%d" % (user_id, expire_at)
sig = _sign(payload.encode("ascii"))
return payload + "." + sig
def parse_token(token: str) -> Optional[Session]:
"""校验 token,成功返回 Session,失败(格式/签名/过期)返回 None。"""
if not isinstance(token, str):
return None
parts = token.split(".")
if len(parts) != 3:
return None
uid_s, exp_s, sig = parts
payload = (uid_s + "." + exp_s).encode("ascii")
expected = _sign(payload)
# 恒定时间比较签名
if not hmac.compare_digest(expected, sig):
return None
try:
uid = int(uid_s)
expire_at = int(exp_s)
except ValueError:
return None
if uid <= 0:
return None
sess = Session(user_id=uid, expire_at=expire_at)
if sess.expired:
return None
return sess