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
|