97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
"""
|
||
用户认证模块。
|
||
|
||
职责：
|
||
- 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 |