dekun
88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
"""密码哈希与 JWT。密钥优先用环境变量，否则写入 app_settings，重启后仍然有效。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import secrets
|
|
from base64 import urlsafe_b64decode, urlsafe_b64encode
|
|
from datetime import timedelta
|
|
|
|
import jwt
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.clock import utcnow
|
|
from app.config import settings
|
|
from app.errors import AppError
|
|
from app.models import AppSetting, User
|
|
|
|
_secret: str | None = None
|
|
_ITERATIONS = 200_000
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = secrets.token_bytes(16)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, _ITERATIONS)
|
|
return f"pbkdf2_sha256${_ITERATIONS}${_b64(salt)}${_b64(digest)}"
|
|
|
|
|
|
def verify_password(password: str, stored: str | None) -> bool:
|
|
if not stored:
|
|
return False
|
|
try:
|
|
algo, rounds, salt_b64, digest_b64 = stored.split("$", 3)
|
|
if algo != "pbkdf2_sha256":
|
|
return False
|
|
got = hashlib.pbkdf2_hmac("sha256", password.encode(), _unb64(salt_b64), int(rounds))
|
|
return hmac.compare_digest(got, _unb64(digest_b64))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def ensure_jwt_secret(session: Session) -> None:
|
|
global _secret
|
|
if settings.jwt_secret:
|
|
_secret = settings.jwt_secret
|
|
return
|
|
row = session.get(AppSetting, "jwt_secret")
|
|
if row is None:
|
|
row = AppSetting(key="jwt_secret", value=secrets.token_hex(32))
|
|
session.add(row)
|
|
session.flush()
|
|
_secret = row.value
|
|
|
|
|
|
def create_access_token(user: User) -> str:
|
|
payload = {
|
|
"sub": str(user.id),
|
|
"role": user.role.value,
|
|
"exp": utcnow() + timedelta(minutes=settings.jwt_expire_minutes),
|
|
}
|
|
return jwt.encode(payload, _require_secret(), algorithm="HS256")
|
|
|
|
|
|
def read_user_id(token: str) -> int:
|
|
try:
|
|
payload = jwt.decode(token, _require_secret(), algorithms=["HS256"])
|
|
except jwt.PyJWTError as exc:
|
|
raise AppError("登录已失效", 401) from exc
|
|
subject = payload.get("sub")
|
|
if not subject:
|
|
raise AppError("登录已失效", 401)
|
|
return int(subject)
|
|
|
|
|
|
def _require_secret() -> str:
|
|
if not _secret:
|
|
raise AppError("认证尚未初始化", 500)
|
|
return _secret
|
|
|
|
|
|
def _b64(raw: bytes) -> str:
|
|
return urlsafe_b64encode(raw).decode().rstrip("=")
|
|
|
|
|
|
def _unb64(text: str) -> bytes:
|
|
padding = "=" * (-len(text) % 4)
|
|
return urlsafe_b64decode(text + padding)
|