yml2213
6fc453017c
- JWT 记录 iat，exp=min(now+24h, iat+7d) - 已认证请求自动续签 httpOnly Cookie - 新增 ACCESS_TOKEN_IDLE_HOURS / ACCESS_TOKEN_MAX_HOURS
186 lines
6.3 KiB
Python
186 lines
6.3 KiB
Python
"""安全模块：密码哈希 + JWT。
|
||
|
||
后台登录采用「滑动会话」：
|
||
|
||
- ``ACCESS_TOKEN_IDLE_HOURS``：空闲多久失效，默认 24 小时；
|
||
- ``ACCESS_TOKEN_MAX_HOURS``：从首次登录起算的绝对上限，默认 7 天。
|
||
|
||
每个带有效会话的 API 请求都会续签 Cookie，但 ``iat`` 保持不变，
|
||
因此活跃用户最多也只能维持 7 天，之后必须重新登录。
|
||
"""
|
||
|
||
import os
|
||
import secrets
|
||
import warnings
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime, timedelta
|
||
|
||
import bcrypt
|
||
from jose import JWTError, jwt
|
||
|
||
# JWT 密钥：生产环境必须通过环境变量 JWT_SECRET_KEY 设置
|
||
SECRET_KEY = os.getenv("JWT_SECRET_KEY")
|
||
if not SECRET_KEY:
|
||
warnings.warn(
|
||
"未设置 JWT_SECRET_KEY 环境变量，使用临时随机密钥（重启后所有 Token 将失效）。"
|
||
"生产环境请务必设置 JWT_SECRET_KEY 环境变量。",
|
||
stacklevel=2,
|
||
)
|
||
SECRET_KEY = secrets.token_urlsafe(32)
|
||
assert SECRET_KEY is not None
|
||
SECRET_KEY_TYPED: str = SECRET_KEY
|
||
ALGORITHM = "HS256"
|
||
|
||
# 兼容旧部署：只配置了 ACCESS_TOKEN_EXPIRE_HOURS 时，把它当作空闲时长。
|
||
ACCESS_TOKEN_IDLE_HOURS = int(
|
||
os.getenv("ACCESS_TOKEN_IDLE_HOURS", os.getenv("ACCESS_TOKEN_EXPIRE_HOURS", "24"))
|
||
)
|
||
ACCESS_TOKEN_MAX_HOURS = int(os.getenv("ACCESS_TOKEN_MAX_HOURS", str(7 * 24)))
|
||
ACCESS_TOKEN_EXPIRE_HOURS = ACCESS_TOKEN_IDLE_HOURS
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AccessToken:
|
||
"""一次签发结果：Token 本体 + 会话起点 + 过期时刻。"""
|
||
|
||
token: str
|
||
issued_at: datetime
|
||
expires_at: datetime
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
pwd_bytes = password.encode("utf-8")
|
||
# bcrypt 限制72字节，截断
|
||
pwd_bytes = pwd_bytes[:72]
|
||
return bcrypt.hashpw(pwd_bytes, bcrypt.gensalt()).decode("utf-8")
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
try:
|
||
return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8"))
|
||
except Exception: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
return False
|
||
|
||
|
||
def _utcnow() -> datetime:
|
||
return datetime.now(UTC).replace(microsecond=0)
|
||
|
||
|
||
def _as_utc(value: object) -> datetime | None:
|
||
"""把 JWT 里的时间字段（datetime / 时间戳）统一转成 UTC datetime。"""
|
||
if isinstance(value, datetime):
|
||
return (
|
||
value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)
|
||
)
|
||
if isinstance(value, (int, float)):
|
||
try:
|
||
return datetime.fromtimestamp(value, UTC)
|
||
except (OverflowError, OSError, ValueError):
|
||
return None
|
||
if isinstance(value, str):
|
||
try:
|
||
return datetime.fromtimestamp(float(value), UTC)
|
||
except (OverflowError, OSError, ValueError):
|
||
return None
|
||
return None
|
||
|
||
|
||
def _session_start(payload: dict) -> datetime | None:
|
||
"""会话起点：新 Token 用 ``iat``，旧 Token 用 ``exp - 24h`` 反推。
|
||
|
||
旧 Token 没有 ``iat``。上线后第一次续签必须把绝对上限锚定到真实登录时刻，
|
||
不能从「第一次续签」重新算 7 天，否则老会话会被额外延长。
|
||
"""
|
||
issued_at = _as_utc(payload.get("iat"))
|
||
if issued_at is not None:
|
||
return issued_at
|
||
expires_at = _as_utc(payload.get("exp"))
|
||
if expires_at is None:
|
||
return None
|
||
return expires_at - timedelta(hours=ACCESS_TOKEN_IDLE_HOURS)
|
||
|
||
|
||
def _absolute_expiry(issued_at: datetime) -> datetime:
|
||
return issued_at + timedelta(hours=ACCESS_TOKEN_MAX_HOURS)
|
||
|
||
|
||
def issue_access_token(
|
||
data: dict,
|
||
*,
|
||
issued_at: datetime | None = None,
|
||
now: datetime | None = None,
|
||
) -> AccessToken:
|
||
"""签发新会话 Token：``exp = min(now + 空闲时长, 会话起点 + 7 天)``。"""
|
||
current = _utcnow() if now is None else now.replace(microsecond=0)
|
||
started = current if issued_at is None else issued_at.replace(microsecond=0)
|
||
expire = min(
|
||
current + timedelta(hours=ACCESS_TOKEN_IDLE_HOURS),
|
||
_absolute_expiry(started),
|
||
)
|
||
to_encode = data.copy()
|
||
to_encode.update({"iat": started, "exp": expire})
|
||
token = jwt.encode(to_encode, SECRET_KEY_TYPED, algorithm=ALGORITHM)
|
||
return AccessToken(token=token, issued_at=started, expires_at=expire)
|
||
|
||
|
||
def create_access_token(
|
||
data: dict,
|
||
*,
|
||
issued_at: datetime | None = None,
|
||
now: datetime | None = None,
|
||
) -> str:
|
||
"""兼容旧调用：只返回 Token 字符串。"""
|
||
return issue_access_token(data, issued_at=issued_at, now=now).token
|
||
|
||
|
||
def decode_access_token(token: str) -> dict | None:
|
||
try:
|
||
return jwt.decode(token, SECRET_KEY_TYPED, algorithms=[ALGORITHM])
|
||
except JWTError:
|
||
return None
|
||
|
||
|
||
def is_session_expired(payload: dict, *, now: datetime | None = None) -> bool:
|
||
"""同时检查空闲过期与 7 天绝对上限。"""
|
||
current = now or _utcnow()
|
||
expires_at = _as_utc(payload.get("exp"))
|
||
if expires_at is None or current >= expires_at:
|
||
return True
|
||
issued_at = _session_start(payload)
|
||
return issued_at is not None and current >= _absolute_expiry(issued_at)
|
||
|
||
|
||
def renew_access_token(
|
||
token: str, *, now: datetime | None = None
|
||
) -> AccessToken | None:
|
||
"""续签有效会话；已到 7 天绝对上限或 Token 本身无效时返回 ``None``。
|
||
|
||
续签保留原 ``iat``，只把 ``exp`` 推到 ``now + 24h``，且不超过绝对上限。
|
||
"""
|
||
payload = decode_access_token(token)
|
||
if payload is None:
|
||
return None
|
||
current = (_utcnow() if now is None else now).replace(microsecond=0)
|
||
if is_session_expired(payload, now=current):
|
||
return None
|
||
issued_at = _session_start(payload)
|
||
if issued_at is None:
|
||
return None
|
||
|
||
absolute_expire = _absolute_expiry(issued_at)
|
||
if current >= absolute_expire:
|
||
return None
|
||
old_expire = _as_utc(payload.get("exp"))
|
||
new_expire = min(
|
||
current + timedelta(hours=ACCESS_TOKEN_IDLE_HOURS), absolute_expire
|
||
)
|
||
if old_expire is not None and new_expire <= old_expire:
|
||
# 已顶到绝对上限，或本次请求没有带来有意义的延长：不重发 Cookie。
|
||
return None
|
||
|
||
claims = payload.copy()
|
||
claims.pop("exp", None)
|
||
claims.update({"iat": issued_at, "exp": new_expire})
|
||
renewed = jwt.encode(claims, SECRET_KEY_TYPED, algorithm=ALGORITHM)
|
||
return AccessToken(token=renewed, issued_at=issued_at, expires_at=new_expire)
|