加上登录和角色权限,并用 Vue 控制台区分用户与管理员。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-10 22:43:31 +08:00
parent 78601aae5e
commit 8b55b9cf55
36 changed files with 2947 additions and 1007 deletions
+87
View File
@@ -0,0 +1,87 @@
"""密码哈希与 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)