xiaoxia-saas/packages/application/auth/jwt_service.py

"""JWT Token 生成、验证、解析服务"""

from datetime import UTC, datetime, timedelta
from typing import Any, Optional

import jwt
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError

from packages.domain.auth.jwt_service import JWTServicePort


class JWTConfig:
    """JWT 配置"""

    def __init__(
        self,
        secret_key: str,
        algorithm: str = "HS256",
        access_token_expire_minutes: int = 15,
        refresh_token_expire_days: int = 7,
    ):
        """
        初始化 JWT 配置

        Args:
            secret_key: JWT 签名密钥（必须从环境变量或配置注入，不允许默认值）
            algorithm: 加密算法，默认 HS256
            access_token_expire_minutes: Access Token 过期时间（分钟）
            refresh_token_expire_days: Refresh Token 过期时间（天）

        Raises:
            ValueError: 如果 secret_key 为空或包含不安全默认值
        """
        if not secret_key or secret_key.strip() == "":
            raise ValueError("JWT secret_key must be provided and cannot be empty")  # noqa: E501

        insecure_defaults = [
            "your-secret-key-change-in-production",
            "your-secret-key",
            "secret",
            "changeme",
            "password",
        ]
        if secret_key.lower() in [d.lower() for d in insecure_defaults]:
            raise ValueError(  # noqa: E501
                f"JWT secret_key '{secret_key}' is insecure. " "Please provide a strong random secret."
            )

        self.SECRET_KEY: str = secret_key
        self.ALGORITHM: str = algorithm
        self.ACCESS_TOKEN_EXPIRE_MINUTES: int = access_token_expire_minutes
        self.REFRESH_TOKEN_EXPIRE_DAYS: int = refresh_token_expire_days


class TokenType:
    """Token 类型"""

    ACCESS = "access"
    REFRESH = "refresh"


class JWTService(JWTServicePort):
    """JWT 服务类"""

    def __init__(self, config: JWTConfig = None):
        if config is None:
            raise ValueError(  # noqa: E501
                "JWTService requires a JWTConfig instance. "  # noqa: E501
                "Please provide a configured JWTConfig with a valid "  # noqa: E501
                "secret_key."
            )
        self.config = config

    def create_access_token(
        self,
        user_id: str,
        role: str = "",
        additional_claims: Optional[dict[str, Any]] = None,
    ) -> str:
        """
        创建 access_token

        Args:
            user_id: 用户 ID
            role: 用户角色（admin/user/guest）
            additional_claims: 额外的声明信息

        Returns:
            JWT Token 字符串
        """
        now = datetime.now(UTC)
        expire = now + timedelta(minutes=self.config.ACCESS_TOKEN_EXPIRE_MINUTES)  # noqa: E501

        payload = {
            "sub": user_id,  # subject (用户ID)
            "role": role,
            "type": TokenType.ACCESS,
            "iat": now,  # issued at
            "exp": expire,  # expiration time
        }

        if additional_claims:
            payload.update(additional_claims)

        return jwt.encode(payload, self.config.SECRET_KEY, algorithm=self.config.ALGORITHM)

    def create_refresh_token(self, user_id: str, session_id: str) -> str:
        """
        创建 refresh_token

        Args:
            user_id: 用户 ID
            session_id: Session ID

        Returns:
            JWT Token 字符串
        """
        now = datetime.now(UTC)
        expire = now + timedelta(days=self.config.REFRESH_TOKEN_EXPIRE_DAYS)

        payload = {
            "sub": user_id,
            "session_id": session_id,
            "type": TokenType.REFRESH,
            "iat": now,
            "exp": expire,
        }

        return jwt.encode(payload, self.config.SECRET_KEY, algorithm=self.config.ALGORITHM)

    def verify_token(self, token: str) -> dict[str, Any]:
        """
        验证 Token

        Args:
            token: JWT Token 字符串

        Returns:
            Token payload

        Raises:
            ExpiredSignatureError: Token 已过期
            InvalidTokenError: Token 无效
        """
        try:
            payload = jwt.decode(  # noqa: E501
                token,
                self.config.SECRET_KEY,
                algorithms=[self.config.ALGORITHM],
            )
            return payload
        except ExpiredSignatureError as _e:
            raise ExpiredSignatureError("Token has expired") from _e
        except InvalidTokenError as e:
            raise InvalidTokenError(f"Invalid token: {str(e)}") from e

    def verify_access_token(self, token: str) -> dict[str, Any]:
        """
        验证 access_token

        Args:
            token: JWT Token 字符串

        Returns:
            Token payload（如果解码失败返回 None）
        """
        payload = self.verify_token(token)

        if payload.get("type") != TokenType.ACCESS:
            raise ValueError("Token type must be 'access'")

        return payload

    def verify_refresh_token(self, token: str) -> dict[str, Any]:
        """
        验证 refresh_token

        Args:
            token: JWT Token 字符串

        Returns:
            Token payload

        Raises:
            ValueError: Token 类型不是 refresh
            ExpiredSignatureError: Token 已过期
            InvalidTokenError: Token 无效
        """
        payload = self.verify_token(token)

        if payload.get("type") != TokenType.REFRESH:
            raise ValueError("Token type must be 'refresh'")

        return payload


# 全局实例（生产环境必须从配置读取有效的 secret_key）
# jwt_service = JWTService()  # 不再允许无参数实例化


# Lazy singleton - created with settings on first access
_jwt_service_instance = None


def _get_jwt_service():
    global _jwt_service_instance
    if _jwt_service_instance is None:
        from app.config import settings

        kw = dict(secret_key=settings.JWT_SECRET_KEY)
        if hasattr(settings, "JWT_ALGORITHM"):
            kw["algorithm"] = settings.JWT_ALGORITHM
        if hasattr(settings, "JWT_ACCESS_TOKEN_EXPIRE_MINUTES"):
            kw["access_token_expire_minutes"] = settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
        if hasattr(settings, "JWT_REFRESH_TOKEN_EXPIRE_DAYS"):
            kw["refresh_token_expire_days"] = settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
        _jwt_service_instance = JWTService(JWTConfig(**kw))
    return _jwt_service_instance


class _JWTServiceProxy:
    def __getattr__(self, name):
        return getattr(_get_jwt_service(), name)


jwt_service = _JWTServiceProxy()