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()),
    }


def create_bind_ticket(
    *, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None
) -> str:
    """微信登录未命中 openid 时,签发短时"待绑手机"令牌,承载 openid + 微信昵称头像。

    typ='wechat_bind'、sub=openid;有效期 settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES 分钟。
    与 access/refresh 用同一 JWT_SECRET_KEY 签名,靠 typ 区分,decode_bind_ticket 校验 typ。
    """
    now = _now()
    expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
    payload: dict[str, Any] = {
        "sub": openid,
        "typ": "wechat_bind",
        "wnk": wechat_nickname,
        "wav": wechat_avatar_url,
        "iat": int(now.timestamp()),
        "exp": int(expire.timestamp()),
    }
    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)


def decode_bind_ticket(token: str) -> dict[str, Any]:
    """解析"待绑手机"令牌,校验签名/过期/类型。失败抛 TokenError。

    返回 {'openid': str, 'wnk': str|None, 'wav': str|None}。
    """
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
    except jwt.ExpiredSignatureError as e:
        raise TokenError("bind ticket expired") from e
    except jwt.InvalidTokenError as e:
        raise TokenError(f"invalid bind ticket: {e}") from e
    if payload.get("typ") != "wechat_bind":
        raise TokenError(f"wrong token type: want=wechat_bind got={payload.get('typ')}")
    if "sub" not in payload:
        raise TokenError("bind ticket missing sub")
    return {"openid": payload["sub"], "wnk": payload.get("wnk"), "wav": payload.get("wav")}


def create_conflict_ticket(
    *, openid: str, wechat_nickname: str | None, wechat_avatar_url: str | None, phone: str
) -> str:
    """手机号占用时签发的短时"冲突处理"令牌。

    比 bind_ticket 多编码 **已验证的手机号 phone** —— 换绑/继续绑定只认它,证明"这对
    openid/手机号刚在绑号时验证通过",免用户重输验证码,又堵住"拿自己 openid + 任意手机号
    去夺号"的接管漏洞。typ='wechat_conflict';有效期复用 WECHAT_BIND_TICKET_EXPIRE_MINUTES。
    """
    now = _now()
    expire = now + timedelta(minutes=settings.WECHAT_BIND_TICKET_EXPIRE_MINUTES)
    payload: dict[str, Any] = {
        "sub": openid,
        "typ": "wechat_conflict",
        "wnk": wechat_nickname,
        "wav": wechat_avatar_url,
        "phn": phone,
        "iat": int(now.timestamp()),
        "exp": int(expire.timestamp()),
    }
    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)


def decode_conflict_ticket(token: str) -> dict[str, Any]:
    """解析"冲突处理"令牌,校验签名/过期/类型。失败抛 TokenError。

    返回 {'openid': str, 'wnk': str|None, 'wav': str|None, 'phone': str}。
    """
    try:
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
    except jwt.ExpiredSignatureError as e:
        raise TokenError("conflict ticket expired") from e
    except jwt.InvalidTokenError as e:
        raise TokenError(f"invalid conflict ticket: {e}") from e
    if payload.get("typ") != "wechat_conflict":
        raise TokenError(f"wrong token type: want=wechat_conflict got={payload.get('typ')}")
    if "sub" not in payload or "phn" not in payload:
        raise TokenError("conflict ticket missing sub/phn")
    return {
        "openid": payload["sub"],
        "wnk": payload.get("wnk"),
        "wav": payload.get("wav"),
        "phone": payload["phn"],
    }


# ===================== 密码 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