taiyi
34e9be8912
- core: config、security、exceptions、responses、error_codes、db、main - models: base、business（17个模型类）、system（6个模型类）、__init__ - 每个函数和类均标注作用、返回值、被调用方 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
94 lines
3.0 KiB
Python
94 lines
3.0 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
|