fix(security): P0 - JWTConfig强制注入密钥,禁止无参数实例化
This commit is contained in:
@@ -13,11 +13,41 @@ from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
|
||||
class JWTConfig:
|
||||
"""JWT 配置"""
|
||||
|
||||
# 从环境变量读取,这里先用默认值
|
||||
SECRET_KEY: str = "your-secret-key-change-in-production"
|
||||
ALGORITHM: str = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 # 30 分钟
|
||||
REFRESH_TOKEN_EXPIRE_DAYS: int = 30 # 30 天
|
||||
def __init__(self, secret_key: str, algorithm: str = "HS256",
|
||||
access_token_expire_minutes: int = 30,
|
||||
refresh_token_expire_days: int = 30):
|
||||
"""
|
||||
初始化 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")
|
||||
|
||||
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(
|
||||
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:
|
||||
@@ -31,7 +61,12 @@ class JWTService:
|
||||
"""JWT 服务类"""
|
||||
|
||||
def __init__(self, config: JWTConfig = None):
|
||||
self.config = config or JWTConfig()
|
||||
if config is None:
|
||||
raise ValueError(
|
||||
"JWTService requires a JWTConfig instance. "
|
||||
"Please provide a configured JWTConfig with a valid secret_key."
|
||||
)
|
||||
self.config = config
|
||||
|
||||
def create_access_token(
|
||||
self,
|
||||
@@ -123,12 +158,7 @@ class JWTService:
|
||||
token: JWT Token 字符串
|
||||
|
||||
Returns:
|
||||
Token payload
|
||||
|
||||
Raises:
|
||||
ValueError: Token 类型不是 access
|
||||
ExpiredSignatureError: Token 已过期
|
||||
InvalidTokenError: Token 无效
|
||||
Token payload(如果解码失败返回 None)
|
||||
"""
|
||||
payload = self.verify_token(token)
|
||||
|
||||
@@ -137,7 +167,7 @@ class JWTService:
|
||||
|
||||
return payload
|
||||
|
||||
def verify_refresh_token(self, token: str) -> Dict[str, Any]:
|
||||
def verify_verify_refresh_token(self, token: str) -> Dict[str, Any]:
|
||||
"""
|
||||
验证 refresh_token
|
||||
|
||||
@@ -175,5 +205,5 @@ class JWTService:
|
||||
return None
|
||||
|
||||
|
||||
# 全局实例(生产环境应该从配置读取)
|
||||
jwt_service = JWTService()
|
||||
# 全局实例(生产环境必须从配置读取有效的 secret_key)
|
||||
# jwt_service = JWTService() # 不再允许无参数实例化
|
||||
|
||||
Reference in New Issue
Block a user