后端实现: - 创建 AuthService 包含密码加密、JWT 生成和验证 - 创建 authMiddleware 用于 API 路由保护 - 实现 auth 路由 (register, login, logout, /me) 前端实现: - 创建 Login 页面 (登录/注册标签页) - 创建 ProtectedRoute 组件用于路由保护 - 更新 App.tsx 集成路由保护 - 前端 API 客户端已包含认证方法和拦截器 验收标准已满足: - 用户可以注册和登录 - JWT Token 正确生成和验证 - 受保护的路由需要有效 Token - 未认证用户重定向到登录页面 Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
91 lines
2.4 KiB
Python
91 lines
2.4 KiB
Python
"""
|
|
Authentication helpers: scrypt password hashing, JWT signing/verification,
|
|
and the require_auth decorator used by route blueprints.
|
|
|
|
Password hashing mirrors the original Node implementation exactly:
|
|
salt (16 random bytes, hex) : scrypt(password, salt, n=16384, r=8, p=1, dklen=64, hex)
|
|
"""
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import datetime
|
|
|
|
import jwt
|
|
from flask import request, g, jsonify
|
|
from functools import wraps
|
|
|
|
from config import JWT_SECRET, JWT_EXPIRY_DAYS
|
|
|
|
|
|
class AuthError(Exception):
|
|
def __init__(self, code, message):
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
salt = os.urandom(16)
|
|
derived = hashlib.scrypt(
|
|
password.encode("utf-8"),
|
|
salt=salt,
|
|
n=16384,
|
|
r=8,
|
|
p=1,
|
|
dklen=64,
|
|
)
|
|
return f"{salt.hex()}:{derived.hex()}"
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
if not stored or ":" not in stored:
|
|
return False
|
|
salt_hex, hash_hex = stored.split(":", 1)
|
|
try:
|
|
salt = bytes.fromhex(salt_hex)
|
|
except ValueError:
|
|
return False
|
|
derived = hashlib.scrypt(
|
|
password.encode("utf-8"),
|
|
salt=salt,
|
|
n=16384,
|
|
r=8,
|
|
p=1,
|
|
dklen=64,
|
|
)
|
|
return hmac.compare_digest(derived.hex(), hash_hex)
|
|
|
|
|
|
def sign_token(user_id: str) -> str:
|
|
now = datetime.datetime.utcnow()
|
|
payload = {
|
|
"sub": user_id,
|
|
"iat": now,
|
|
"exp": now + datetime.timedelta(days=JWT_EXPIRY_DAYS),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm="HS256")
|
|
|
|
|
|
def verify_token(token: str) -> dict:
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
|
|
return {"user_id": payload["sub"]}
|
|
|
|
|
|
def require_auth(f):
|
|
@wraps(f)
|
|
def wrapper(*args, **kwargs):
|
|
auth = request.headers.get("Authorization", "")
|
|
if not auth.startswith("Bearer "):
|
|
return jsonify({"error": "missing or malformed Authorization header"}), 401
|
|
token = auth[7:].strip()
|
|
try:
|
|
data = verify_token(token)
|
|
except jwt.ExpiredSignatureError:
|
|
return jsonify({"error": "token expired"}), 401
|
|
except jwt.InvalidTokenError:
|
|
return jsonify({"error": "invalid token"}), 401
|
|
g.user_id = data["user_id"]
|
|
return f(*args, **kwargs)
|
|
|
|
return wrapper
|