71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
|
# 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$<iterations>$<salt_hex>$<hash_hex>
|
|||
|
|
盐值每次随机生成，同一密码两次加密结果不同
|
|||
|
|
"""
|
|||
|
|
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])
|