From a1acc6e63ba9245f224d287c96936cbc29ff7837 Mon Sep 17 00:00:00 2001
From: bot_dev1 <bot_dev1@xayunmei.com>
Date: Wed, 5 Aug 2026 02:05:50 +0000
Subject: [PATCH] =?UTF-8?q?feat(#150):=20=E4=BC=9A=E8=AF=9D=E7=AE=A1?=
 =?UTF-8?q?=E7=90=86=20session.py=EF=BC=88HMAC=20=E7=AD=BE=E5=90=8D=20toke?=
 =?UTF-8?q?n=20+=20=E6=81=92=E5=AE=9A=E6=97=B6=E9=97=B4=E6=A0=A1=E9=AA=8C?=
 =?UTF-8?q?=EF=BC=8CHttpOnly=20cookie=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 core/auth/session.py | 97 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 97 insertions(+)
 create mode 100644 core/auth/session.py

diff --git a/core/auth/session.py b/core/auth/session.py
new file mode 100644
index 0000000..ea9151d
--- /dev/null
+++ b/core/auth/session.py
@@ -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