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])
|