shaguabijia-app-server/app/core/security.py

"""JWT 签发与校验。

设计:
  - access_token : 短期(默认 2 小时),每个业务请求 Bearer 携带
  - refresh_token: 长期(默认 30 天),过期时 POST /refresh 换新的 access
两者结构一致(都是 JWT),靠 payload 里的 `typ` 字段区分,防止用 refresh 当 access 用。

为什么不用 OAuth2 + session 表?
  - 占坑 + 早期阶段,JWT 自包含、零状态,够用
  - 后续需要"主动作废"(改密码/账号被盗)时再加 jti 黑名单表
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any, Literal

import bcrypt
import jwt

from app.core.config import settings

TokenType = Literal["access", "refresh"]


class TokenError(Exception):
    """token 解析/校验失败时统一抛这个,api 层捕获后返回 401。"""


def _now() -> datetime:
    return datetime.now(timezone.utc)


def create_token(*, user_id: int, token_type: TokenType) -> tuple[str, datetime]:
    """签发 token,返回 (token_string, expires_at_utc)。"""
    now = _now()
    if token_type == "access":
        expire = now + timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
    else:
        expire = now + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS)

    payload: dict[str, Any] = {
        "sub": str(user_id),
        "typ": token_type,
        "iat": int(now.timestamp()),
        "exp": int(expire.timestamp()),
    }
    token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
    return token, expire


def decode_token(token: str, *, expected_type: TokenType | None = None) -> dict[str, Any]:
    """解析并校验 token。

    - 签名错/过期 → TokenError
    - expected_type 不匹配 → TokenError (防 refresh 当 access 用)
    """
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
    except jwt.ExpiredSignatureError as e:
        raise TokenError("token expired") from e
    except jwt.InvalidTokenError as e:
        raise TokenError(f"invalid token: {e}") from e

    if expected_type is not None and payload.get("typ") != expected_type:
        raise TokenError(f"wrong token type: want={expected_type} got={payload.get('typ')}")

    if "sub" not in payload:
        raise TokenError("token missing sub")

    return payload


def issue_token_pair(user_id: int) -> dict[str, Any]:
    """登录/刷新成功后调,一次性发回 access + refresh。

    返回字段对齐 OAuth2 习惯,前端易接:
      access_token / refresh_token / token_type / expires_in
    """
    access, access_exp = create_token(user_id=user_id, token_type="access")
    refresh, refresh_exp = create_token(user_id=user_id, token_type="refresh")
    return {
        "access_token": access,
        "refresh_token": refresh,
        "token_type": "Bearer",
        "expires_in": int((access_exp - _now()).total_seconds()),
        "refresh_expires_in": int((refresh_exp - _now()).total_seconds()),
    }


# ===================== 密码 hash(admin 后台账号用)=====================
# 用户侧是手机号+验证码登录,不存密码;仅 admin 账号用 username+password 登录。

def hash_password(plain: str) -> str:
    """bcrypt 加盐 hash,返回可入库的字符串。

    bcrypt 限制明文 ≤72 字节,超出抛 ValueError(实测 bcrypt 5.0:25 个中文=75 字节即触发)。
    统一截断到 72 字节(bcrypt 标准做法)——72 字节仍是强密码,verify 同样截断保证一致。
    """
    return bcrypt.hashpw(plain.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")


def verify_password(plain: str, hashed: str) -> bool:
    """校验明文与 bcrypt hash。明文同样截断 72 字节(与 hash_password 一致,否则超长密码永远不匹配);
    hash 串损坏/格式非法时返回 False,不抛异常。"""
    try:
        return bcrypt.checkpw(plain.encode("utf-8")[:72], hashed.encode("utf-8"))
    except (ValueError, TypeError):
        return False