Initial commit: MyBlog full stack blog

backend/auth.py

@@ -0,0 +1,97 @@
+"""
+用户认证模块。
+
+职责：
+- bcrypt 密码哈希与校验（禁止明文保存密码）
+- JWT 令牌的生成与解析（载荷包含用户 id、角色、过期时间）
+- FastAPI 依赖 get_current_user：从请求头解析当前登录用户
+"""
+
+import os
+from datetime import datetime, timedelta, timezone
+
+import bcrypt
+import jwt
+from dotenv import load_dotenv
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from sqlalchemy.orm import Session
+
+from .database import PROJECT_ROOT, get_db
+from .models import User
+
+# 加载项目根目录下的 .env，确保 JWT_SECRET 可用（重复调用无副作用）
+load_dotenv(PROJECT_ROOT / ".env")
+
+# JWT 签名密钥：必须在 .env 中配置。
+# 缺失时直接报错（fail-fast），禁止回退到硬编码默认密钥，避免生产环境使用弱密钥。
+JWT_SECRET = os.getenv("JWT_SECRET", "").strip()
+if not JWT_SECRET:
+    raise RuntimeError(
+        "缺少 JWT_SECRET 环境变量：请在项目根目录 .env 中配置强随机密钥后重试"
+    )
+
+JWT_ALGORITHM = "HS256"
+
+# 令牌默认有效期：7 天，可通过 .env 的 JWT_EXPIRE_MINUTES 覆盖
+JWT_EXPIRE_MINUTES = int(os.getenv("JWT_EXPIRE_MINUTES") or str(7 * 24 * 60))
+
+# HTTP Bearer 令牌认证方案（auto_error=False：缺失令牌时不自动报错，便于统一错误返回格式）
+bearer_scheme = HTTPBearer(auto_error=False)
+
+
+def hash_password(password: str) -> str:
+    """使用 bcrypt 生成密码哈希，返回可直接存入数据库的字符串。"""
+    salt = bcrypt.gensalt()
+    return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    """校验明文密码与 bcrypt 哈希是否匹配。"""
+    return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
+
+
+def create_access_token(user: User) -> str:
+    """生成 JWT 令牌：载荷包含用户 id、角色与过期时间。"""
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": str(user.id),                                     # 用户 id（subject）
+        "role": user.role,
+        "ver": user.token_version,                                   # 令牌版本（重置密码后旧令牌失效）
+        "iat": now,                                              # 签发时间
+        "exp": now + timedelta(minutes=JWT_EXPIRE_MINUTES),      # 过期时间
+    }
+    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
+
+
+def decode_token(token: str) -> dict:
+    """解析并校验 JWT 令牌，返回载荷；令牌无效或已过期时抛出 401。"""
+    try:
+        return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
+    except jwt.PyJWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="登录状态无效或已过期，请重新登录",
+        )
+
+
+def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
+    db: Session = Depends(get_db),
+) -> User:
+    """FastAPI 依赖：从 Authorization: Bearer <token> 解析并返回当前用户。"""
+    if credentials is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="未登录，请先登录")
+    payload = decode_token(credentials.credentials)
+    user_id = payload.get("sub")
+    try:
+        user_id = int(user_id)
+    except (TypeError, ValueError):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="令牌无效")
+    user = db.query(User).filter(User.id == user_id).first()
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在")
+    # 令牌版本校验：重置密码后 token_version 自增，旧令牌立即失效（吊销机制）
+    if (payload.get("ver") or 0) != (user.token_version or 0):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="登录状态已失效，请重新登录")
+    return user