[阶段1.1-1.7] 实现完整的认证系统
后端实现: - 创建 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>
This commit is contained in:
90
backend/auth.py
Normal file
90
backend/auth.py
Normal file
@@ -0,0 +1,90 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user