2025-05-02 23:56:59 +08:00
|
|
|
import hashlib
|
|
|
|
|
import os
|
2025-10-13 15:08:54 +08:00
|
|
|
from datetime import timedelta
|
2025-05-23 15:30:14 +08:00
|
|
|
from typing import Any
|
2025-05-02 23:56:59 +08:00
|
|
|
|
2025-09-01 22:37:03 +08:00
|
|
|
import jwt
|
|
|
|
|
|
2025-10-13 15:08:54 +08:00
|
|
|
from src.utils.datetime_utils import utc_now
|
|
|
|
|
|
2025-05-02 23:56:59 +08:00
|
|
|
# JWT配置
|
|
|
|
|
JWT_SECRET_KEY = os.environ.get("JWT_SECRET_KEY", "yuxi_know_secure_key")
|
|
|
|
|
JWT_ALGORITHM = "HS256"
|
2025-09-07 20:06:56 +08:00
|
|
|
JWT_EXPIRATION = 7 * 24 * 60 * 60 # 7天过期
|
2025-05-02 23:56:59 +08:00
|
|
|
|
2025-09-01 22:37:03 +08:00
|
|
|
|
2025-05-02 23:56:59 +08:00
|
|
|
class AuthUtils:
|
|
|
|
|
"""认证工具类"""
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
|
|
|
"""使用SHA-256哈希密码"""
|
|
|
|
|
# 生成盐
|
|
|
|
|
salt = os.urandom(32).hex()
|
|
|
|
|
# 哈希密码
|
|
|
|
|
hashed = hashlib.sha256((password + salt).encode()).hexdigest()
|
|
|
|
|
# 返回格式: "哈希值:盐"
|
|
|
|
|
return f"{hashed}:{salt}"
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
|
def verify_password(stored_password: str, provided_password: str) -> bool:
|
|
|
|
|
"""验证密码"""
|
|
|
|
|
# 分离哈希值和盐
|
|
|
|
|
if ":" not in stored_password:
|
|
|
|
|
return False
|
|
|
|
|
|
|
|
|
|
hashed, salt = stored_password.split(":")
|
|
|
|
|
|
|
|
|
|
# 使用相同的盐哈希提供的密码
|
|
|
|
|
check_hash = hashlib.sha256((provided_password + salt).encode()).hexdigest()
|
|
|
|
|
|
|
|
|
|
# 比较哈希值
|
|
|
|
|
return hashed == check_hash
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
2025-05-23 15:30:14 +08:00
|
|
|
def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str:
|
2025-05-02 23:56:59 +08:00
|
|
|
"""创建JWT访问令牌"""
|
|
|
|
|
to_encode = data.copy()
|
|
|
|
|
|
|
|
|
|
# 设置过期时间
|
|
|
|
|
if expires_delta:
|
2025-10-13 15:08:54 +08:00
|
|
|
expire = utc_now() + expires_delta
|
2025-05-02 23:56:59 +08:00
|
|
|
else:
|
2025-10-13 15:08:54 +08:00
|
|
|
expire = utc_now() + timedelta(seconds=JWT_EXPIRATION)
|
2025-05-02 23:56:59 +08:00
|
|
|
|
|
|
|
|
to_encode.update({"exp": expire})
|
|
|
|
|
|
|
|
|
|
# 编码JWT
|
|
|
|
|
encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
|
|
|
|
|
return encoded_jwt
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
2025-05-23 15:30:14 +08:00
|
|
|
def decode_token(token: str) -> dict[str, Any] | None:
|
2025-05-02 23:56:59 +08:00
|
|
|
"""解码验证JWT令牌"""
|
|
|
|
|
try:
|
|
|
|
|
payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
|
|
|
|
|
return payload
|
|
|
|
|
except jwt.PyJWTError:
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
2025-05-23 15:30:14 +08:00
|
|
|
def verify_access_token(token: str) -> dict[str, Any]:
|
2025-05-02 23:56:59 +08:00
|
|
|
"""验证访问令牌,如果无效则抛出异常"""
|
|
|
|
|
try:
|
|
|
|
|
payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
|
|
|
|
|
return payload
|
|
|
|
|
except jwt.ExpiredSignatureError:
|
|
|
|
|
raise ValueError("令牌已过期")
|
|
|
|
|
except jwt.InvalidTokenError:
|
2025-05-23 15:30:14 +08:00
|
|
|
raise ValueError("无效的令牌")
|