74 lines
1.9 KiB
Python
74 lines
1.9 KiB
Python
"""服务端 Session 管理（MVP: 内存 dict + TTL）。
|
|
|
|
- Session token: secrets.token_urlsafe(32)
|
|
- Cookie: HttpOnly, SameSite=Lax, 生产环境 Secure
|
|
- 默认过期: 7 天
|
|
|
|
未来切换 Redis: 改 _store 为 Redis 客户端，接口不变。
|
|
"""
|
|
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from app.core.config import get_settings
|
|
|
|
# 内存存储：{token: SessionEntry}
|
|
_store: dict[str, "SessionEntry"] = {}
|
|
|
|
SESSION_COOKIE_NAME = "session_id"
|
|
SESSION_TTL_SECONDS = 7 * 24 * 3600 # 7 天
|
|
|
|
|
|
@dataclass
|
|
class SessionEntry:
|
|
user_id: str
|
|
created_at: float
|
|
last_accessed: float
|
|
|
|
|
|
def create_session(user_id: str) -> str:
|
|
"""创建 session，返回 token（写入 Cookie）。"""
|
|
_cleanup_expired()
|
|
token = secrets.token_urlsafe(32)
|
|
now = time.time()
|
|
_store[token] = SessionEntry(user_id=user_id, created_at=now, last_accessed=now)
|
|
return token
|
|
|
|
|
|
def get_session_user(token: str) -> str | None:
|
|
"""根据 token 获取 user_id；不存在或过期返回 None。"""
|
|
entry = _store.get(token)
|
|
if entry is None:
|
|
return None
|
|
now = time.time()
|
|
if now - entry.last_accessed > SESSION_TTL_SECONDS:
|
|
_store.pop(token, None)
|
|
return None
|
|
entry.last_accessed = now
|
|
return entry.user_id
|
|
|
|
|
|
def delete_session(token: str) -> None:
|
|
_store.pop(token, None)
|
|
|
|
|
|
def _cleanup_expired() -> None:
|
|
"""惰性清理过期 session。"""
|
|
now = time.time()
|
|
expired = [t for t, e in _store.items() if now - e.last_accessed > SESSION_TTL_SECONDS]
|
|
for t in expired:
|
|
_store.pop(t, None)
|
|
|
|
|
|
def get_cookie_params() -> dict:
|
|
"""构建 Set-Cookie 参数。"""
|
|
settings = get_settings()
|
|
return {
|
|
"key": SESSION_COOKIE_NAME,
|
|
"httponly": True,
|
|
"samesite": "lax",
|
|
"secure": settings.use_secure_cookie,
|
|
"max_age": SESSION_TTL_SECONDS,
|
|
"path": "/",
|
|
} |