98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
|
# -*- 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
|