# core/security.py # 安全工具:密码哈希(PBKDF2-SHA256,标准库实现,无额外依赖)+ JWT 签发与校验 import hashlib import hmac import os from datetime import datetime, timedelta, timezone import jwt from config import settings # PBKDF2 迭代次数与哈希长度 _ITERATIONS = 120_000 _KEY_LEN = 32 # ==================== 密码哈希 ==================== def hash_password(plain_password: str) -> str: """ 生成密码哈希,格式:pbkdf2_sha256$$$ 盐值每次随机生成,同一密码两次加密结果不同 """ salt = os.urandom(16) digest = hashlib.pbkdf2_hmac( "sha256", plain_password.encode("utf-8"), salt, _ITERATIONS, dklen=_KEY_LEN ) return f"pbkdf2_sha256${_ITERATIONS}${salt.hex()}${digest.hex()}" def verify_password(plain_password: str, password_hash: str) -> bool: """校验密码:用同样的盐和迭代次数重新计算,恒定时间比较""" try: algorithm, iterations, salt_hex, hash_hex = password_hash.split("$") if algorithm != "pbkdf2_sha256": return False digest = hashlib.pbkdf2_hmac( "sha256", plain_password.encode("utf-8"), bytes.fromhex(salt_hex), int(iterations), dklen=len(bytes.fromhex(hash_hex)), ) return hmac.compare_digest(digest, bytes.fromhex(hash_hex)) except (ValueError, TypeError): return False # ==================== JWT ==================== def create_access_token(user_id: int, role: str) -> str: """签发 JWT,payload 携带用户 ID 与角色""" expire = datetime.now(timezone.utc) + timedelta( minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES ) payload = { "sub": str(user_id), # sub 建议为字符串 "role": role, "exp": expire, "iat": datetime.now(timezone.utc), } return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM) def decode_access_token(token: str) -> dict: """ 解码并校验 JWT :raises jwt.ExpiredSignatureError: token 过期 :raises jwt.InvalidTokenError: token 无效 """ return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])