90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""
|
||
认证工具：密码哈希 + JWT 令牌 + FastAPI 依赖注入。
|
||
"""
|
||
|
||
import os
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from fastapi import Depends, HTTPException, status
|
||
from fastapi.security import OAuth2PasswordBearer
|
||
from jose import JWTError, jwt
|
||
from pwdlib import PasswordHash
|
||
from pwdlib.hashers.argon2 import Argon2Hasher
|
||
from pwdlib.hashers.bcrypt import BcryptHasher
|
||
from sqlmodel import Session, select
|
||
|
||
from app.db import User, get_session
|
||
|
||
# Argon2 优先用于新密码，BcryptHasher 兼容旧版已有哈希
|
||
pwd_hash = PasswordHash((Argon2Hasher(), BcryptHasher()))
|
||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
|
||
|
||
ALGORITHM = "HS256"
|
||
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
||
REFRESH_TOKEN_EXPIRE_DAYS = 7
|
||
|
||
|
||
def _get_secret() -> str:
|
||
secret = os.getenv("JWT_SECRET", "")
|
||
if not secret:
|
||
raise RuntimeError("JWT_SECRET 环境变量未设置")
|
||
return secret
|
||
|
||
|
||
def hash_password(plain: str) -> str:
|
||
return pwd_hash.hash(plain)
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
return pwd_hash.verify(plain, hashed)
|
||
|
||
|
||
def create_access_token(user_id: str) -> str:
|
||
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||
return jwt.encode(
|
||
{"sub": user_id, "exp": expire, "type": "access"},
|
||
_get_secret(),
|
||
algorithm=ALGORITHM,
|
||
)
|
||
|
||
|
||
def create_refresh_token(user_id: str) -> str:
|
||
expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
|
||
return jwt.encode(
|
||
{"sub": user_id, "exp": expire, "type": "refresh"},
|
||
_get_secret(),
|
||
algorithm=ALGORITHM,
|
||
)
|
||
|
||
|
||
def decode_token(token: str, expected_type: str = "access") -> str:
|
||
"""解码 JWT，返回 user_id。无效时抛 HTTPException 401。"""
|
||
try:
|
||
payload = jwt.decode(token, _get_secret(), algorithms=[ALGORITHM])
|
||
user_id: str = payload.get("sub", "")
|
||
token_type: str = payload.get("type", "")
|
||
if not user_id or token_type != expected_type:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无效令牌")
|
||
return user_id
|
||
except JWTError:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="令牌已过期或无效")
|
||
|
||
|
||
def get_current_user(
|
||
token: str = Depends(oauth2_scheme),
|
||
session: Session = Depends(get_session),
|
||
) -> User:
|
||
"""FastAPI 依赖：从 Bearer token 解析当前用户。"""
|
||
user_id = decode_token(token, "access")
|
||
user = session.exec(select(User).where(User.id == user_id)).first()
|
||
if not user or not user.is_active:
|
||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="用户不存在或已禁用")
|
||
return user
|
||
|
||
|
||
def require_admin(user: User = Depends(get_current_user)) -> User:
|
||
"""FastAPI 依赖：要求当前用户是管理员。"""
|
||
if not user.is_admin:
|
||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="需要管理员权限")
|
||
return user
|