118 lines
3.8 KiB
Python
118 lines
3.8 KiB
Python
"""JWT 认证与密码安全模块。
|
||
|
||
提供密码哈希、JWT 令牌创建与解码功能。
|
||
采用 HMAC-SHA256 签名，不依赖第三方 JWT 库。
|
||
"""
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from backend.app.core.config import get_settings
|
||
|
||
|
||
settings = get_settings()
|
||
|
||
|
||
def hash_password(password: str) -> str:
|
||
"""对明文密码进行 SHA256 哈希。
|
||
|
||
使用 SECRET_KEY 作为盐值，生成固定长度的十六进制摘要。
|
||
|
||
Args:
|
||
password: 用户输入的明文密码。
|
||
|
||
Returns:
|
||
str: 64 位十六进制哈希字符串。
|
||
"""
|
||
secret = settings.secret_key.encode("utf-8")
|
||
return hashlib.sha256(secret + password.encode("utf-8")).hexdigest()
|
||
|
||
|
||
def verify_password(password: str, password_hash: str) -> bool:
|
||
"""验证明文密码与存储的哈希是否匹配。
|
||
|
||
使用 hmac.compare_digest 防止时序攻击。
|
||
|
||
Args:
|
||
password: 用户输入的明文密码。
|
||
password_hash: 数据库中存储的密码哈希。
|
||
|
||
Returns:
|
||
bool: 密码匹配返回 True，否则 False。
|
||
"""
|
||
return hmac.compare_digest(hash_password(password), password_hash or "")
|
||
|
||
|
||
def create_access_token(payload: dict) -> str:
|
||
"""创建 JWT 访问令牌。
|
||
|
||
将 payload 与过期时间组合，使用 base64url 编码后
|
||
通过 HMAC-SHA256 签名，返回 "body.signature" 格式的令牌。
|
||
|
||
Args:
|
||
payload: 令牌载荷，通常包含 user_id、role_code 等字段。
|
||
|
||
Returns:
|
||
str: 签名后的 JWT 令牌字符串。
|
||
"""
|
||
data = dict(payload)
|
||
expire_at = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
|
||
data["exp"] = int(expire_at.timestamp())
|
||
body = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||
body_b64 = base64.urlsafe_b64encode(body).decode("utf-8").rstrip("=")
|
||
signature = hmac.new(settings.secret_key.encode("utf-8"), body_b64.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
return f"{body_b64}.{signature}"
|
||
|
||
|
||
def decode_access_token(token: str) -> dict | None:
|
||
"""解码并验证 JWT 令牌。
|
||
|
||
验证签名是否正确、是否过期。验证失败或令牌格式错误时返回 None。
|
||
|
||
Args:
|
||
token: JWT 令牌字符串（"body.signature" 格式）。
|
||
|
||
Returns:
|
||
dict | None: 解码后的载荷字典，验证失败返回 None。
|
||
"""
|
||
try:
|
||
body_b64, signature = token.split(".", 1)
|
||
expected = hmac.new(settings.secret_key.encode("utf-8"), body_b64.encode("utf-8"), hashlib.sha256).hexdigest()
|
||
if not hmac.compare_digest(signature, expected):
|
||
return None
|
||
padded = body_b64 + "=" * (-len(body_b64) % 4)
|
||
payload = json.loads(base64.urlsafe_b64decode(padded.encode("utf-8")).decode("utf-8"))
|
||
exp = int(payload.get("exp", 0))
|
||
if exp <= int(datetime.now(timezone.utc).timestamp()):
|
||
return None
|
||
return payload
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def should_refresh_token(token: str, threshold_days: int = 30) -> bool:
|
||
"""检查 Token 是否需要续期。
|
||
|
||
当剩余有效期不足 threshold_days 天时返回 True。
|
||
|
||
Args:
|
||
token: JWT 令牌字符串。
|
||
threshold_days: 续期阈值天数，默认 30 天。
|
||
|
||
Returns:
|
||
bool: 需要续期返回 True，否则 False。
|
||
"""
|
||
try:
|
||
body_b64, _ = token.split(".", 1)
|
||
padded = body_b64 + "=" * (-len(body_b64) % 4)
|
||
payload = json.loads(base64.urlsafe_b64decode(padded.encode("utf-8")).decode("utf-8"))
|
||
exp = int(payload.get("exp", 0))
|
||
now = int(datetime.now(timezone.utc).timestamp())
|
||
remaining_days = (exp - now) / (24 * 3600)
|
||
return remaining_days < threshold_days
|
||
except Exception:
|
||
return False
|