harness_zbt/backend/app/core/security.py

"""JWT（JSON Web Token）的签发与校验。

access token 有效期较短，用于访问接口；refresh token 有效期较长，只用于换新令牌。
"""

import hashlib
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

import jwt

from app.core.config import Settings
from app.core.errors import AppError
from app.core.identifiers import new_ulid


@dataclass(frozen=True, slots=True)
class TokenPair:
    """一次登录签发的一对令牌及其有效期。"""

    access_token: str
    refresh_token: str
    access_expires_in: int
    refresh_expires_in: int
    refresh_jti: str


class JwtService:
    """封装 PyJWT，统一项目中的令牌字段、密钥和校验规则。"""

    def __init__(self, settings: Settings) -> None:
        self._settings = settings

    def issue_pair(
        self,
        *,
        subject_id: str,
        subject_type: str,
        audience: str,
        session_id: str,
        now: datetime | None = None,
    ) -> TokenPair:
        """为一个用户会话签发 access token 和 refresh token。"""
        issued_at = now or datetime.now(UTC)
        access_seconds = self._settings.access_token_minutes * 60
        refresh_seconds = self._settings.refresh_token_days * 24 * 60 * 60
        refresh_jti = new_ulid()
        # 两种令牌共有的标准/自定义声明；sub 是用户，sid 是服务端会话。
        common = {
            "sub": subject_id,
            "subject_type": subject_type,
            "aud": audience,
            "iss": "zhibaotong-s1",
            "sid": session_id,
            "iat": issued_at,
        }
        access = jwt.encode(
            {
                # ** 会把 common 字典的键值展开并合并到新字典。
                **common,
                "typ": "access",
                "exp": issued_at + timedelta(seconds=access_seconds),
            },
            self._settings.jwt_access_secret,
            algorithm="HS256",
        )
        refresh = jwt.encode(
            {
                **common,
                "typ": "refresh",
                "jti": refresh_jti,
                "exp": issued_at + timedelta(seconds=refresh_seconds),
            },
            self._settings.jwt_refresh_secret,
            algorithm="HS256",
        )
        return TokenPair(
            access_token=access,
            refresh_token=refresh,
            access_expires_in=access_seconds,
            refresh_expires_in=refresh_seconds,
            refresh_jti=refresh_jti,
        )

    def decode_access(self, token: str, *, audience: str) -> dict[str, Any]:
        """校验并解码 access token。"""
        return self._decode(
            token,
            secret=self._settings.jwt_access_secret,
            audience=audience,
            token_type="access",
        )

    def decode_refresh(self, token: str, *, audience: str) -> dict[str, Any]:
        """校验并解码 refresh token。"""
        return self._decode(
            token,
            secret=self._settings.jwt_refresh_secret,
            audience=audience,
            token_type="refresh",
        )

    @staticmethod
    def hash_token_identifier(value: str) -> str:
        return hashlib.sha256(value.encode("utf-8")).hexdigest()

    @staticmethod
    def _decode(
        token: str,
        *,
        secret: str,
        audience: str,
        token_type: str,
    ) -> dict[str, Any]:
        """执行公共解码逻辑，并把第三方库异常转换成本项目的业务异常。"""
        try:
            payload = jwt.decode(
                token,
                secret,
                algorithms=["HS256"],
                audience=audience,
                issuer="zhibaotong-s1",
            )
        except jwt.PyJWTError as exc:
            raise AppError("AUTH_REQUIRED", "登录凭证无效或已过期", 401) from exc
        if payload.get("typ") != token_type:
            raise AppError("AUTH_REQUIRED", "登录凭证类型无效", 401)
        return payload