上一篇我们实现了用户的 CRUD 接口，但有个大问题：任何人都能调用这些接口—— 比如随便一个人就能访问/api/users获取所有用户信息，甚至删除用户。这篇我们要解决 **“接口安全”** 问题：用 JWT（JSON Web Token）实现 “登录认证”，用权限控制实现 “不同角色访问不同接口”，让系统更符合企业级要求。

一、先搞懂：JWT 到底是啥？

小白最常问：“JWT 和密码有啥区别？”—— 举个生活中的例子：

• 你去小区健身房，第一次需要用身份证（用户名 + 密码）登记，前台给你一张电子门禁卡（JWT）；
• 之后每次去，不需要再出示身份证，直接刷门禁卡就行 —— 门禁卡里面存了你的身份信息（比如 “住户小明”），还带防伪标识（别人伪造不了）。

JWT 的核心价值：

• 「一次登录，多接口复用」：登录后拿到 JWT，之后所有请求都带这个 Token，不用每次都传用户名密码；
• 「无状态」：服务器不用存用户会话，Token 本身包含了所有必要信息；
• 「安全」：Token 有签名，篡改后会失效。

JWT 的结构（3 部分用 “.” 连接）

比如一个实际的 JWT：eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6InhpYW9taW5nIiwiZXhwIjoxNzE2MjM5MjAwfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c拆成 3 部分：

1. Header（头部）：存 Token 类型（JWT）和签名算法（比如 HS256）；

   {"alg": "HS256", "typ": "JWT"}
   

2. Payload（负载）：存用户信息（比如用户 ID、用户名）和过期时间；

   {"user_id": 1, "username": "xiaoming", "exp": 1716239200}
   

3. Signature（签名）：用 Header 的算法 + 密钥（比如secret_key）对前两部分加密，防止篡改；

   HMACSHA256(
       base64UrlEncode(Header) + "." + base64UrlEncode(Payload),
       secret_key
   )
   

二、准备工作：配置 JWT 参数

先在.env文件里加 JWT 配置（避免硬编码）：

# .env文件
# 之前的数据库配置...
JWT_SECRET_KEY=your-secret-key-keep-it-safe  # JWT密钥（要复杂，比如用openssl rand -hex 32生成）
JWT_ALGORITHM=HS256  # 签名算法
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=30  # Token过期时间（30分钟）

然后在app/config.py里读取这些配置：

# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 数据库配置...
    # JWT配置
    JWT_SECRET_KEY: str
    JWT_ALGORITHM: str
    JWT_ACCESS_TOKEN_EXPIRE_MINUTES: int

    class Config:
        env_file = ".env"

settings = Settings()

三、实现 JWT 工具类：生成与验证 Token

写一个工具类app/utils/jwt_utils.py，封装 JWT 的生成和验证逻辑：

from datetime import datetime, timedelta
from jose import JWTError, jwt
from fastapi import HTTPException, status
from sqlalchemy.orm import Session

from app.config import settings
from app.models.user import User
from app.services.user_service import UserService

# 初始化UserService
user_service = UserService()

def create_access_token(data: dict) -> str:
    """生成JWT Token"""
    to_encode = data.copy()
    # 设置过期时间：当前时间 + 配置的过期分钟数
    expire = datetime.utcnow() + timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})  # 把过期时间加入Payload
    # 生成Token：用密钥和算法加密
    encoded_jwt = jwt.encode(to_encode, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
    return encoded_jwt

def verify_access_token(token: str, db: Session) -> User:
    """验证Token，返回对应的用户"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的Token或已过期",
        headers={"WWW-Authenticate": "Bearer"},  # 告诉客户端需要用Bearer Token
    )
    try:
        # 解码Token
        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
        # 从Payload中获取用户ID
        user_id: int = payload.get("user_id")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # 查询用户是否存在
    user = user_service.get_by_id(db, user_id)
    if user is None:
        raise credentials_exception
    return user

四、实现登录接口：获取 JWT Token

登录接口是唯一不需要认证的接口 —— 用户用用户名 + 密码换 Token，之后用 Token 访问其他接口。

1. 写登录的 Pydantic 模型（app/schemas/auth.py）

from pydantic import BaseModel, Field

class LoginRequest(BaseModel):
    """登录请求体模型"""
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    password: str = Field(..., min_length=6, description="密码")

class TokenResponse(BaseModel):
    """登录响应体模型"""
    access_token: str  # JWT Token
    token_type: str = "bearer"  # Token类型（固定为bearer）

2. 写登录接口（app/routes/auth_router.py）

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.db import get_db
from app.schemas.auth import LoginRequest, TokenResponse
from app.services.user_service import UserService
from app.utils.jwt_utils import create_access_token

# 创建认证路由实例
auth_router = APIRouter(prefix="/api/auth", tags=["Auth"])

# 初始化UserService
user_service = UserService()

@auth_router.post("/login", response_model=TokenResponse, status_code=status.HTTP_200_OK)
def login(login_data: LoginRequest, db: Session = Depends(get_db)):
    """登录接口：用用户名+密码换JWT Token"""
    # 1. 根据用户名查询用户
    user = user_service.get_by_fields(db, username=login_data.username)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误"
        )
    # 2. 验证密码（对比明文密码和数据库的哈希密码）
    if not user_service.verify_password(login_data.password, user.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误"
        )
    # 3. 生成JWT Token（Payload中存用户ID）
    access_token = create_access_token(data={"user_id": user.id})
    # 4. 返回Token
    return {"access_token": access_token, "token_type": "bearer"}

五、给接口加认证：只有登录用户才能访问

现在要让所有用户接口都需要认证—— 比如访问/api/users必须带 JWT Token。实现方式是依赖注入：写一个get_current_user依赖，解析请求中的 Token，获取当前用户。

1. 写认证依赖（app/dependencies/auth.py）

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.db import get_db
from app.models.user import User
from app.utils.jwt_utils import verify_access_token

# OAuth2PasswordBearer：FastAPI提供的Bearer Token验证工具
# tokenUrl：登录接口的路径（用于Swagger文档自动生成登录按钮）
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    """依赖：获取当前登录用户"""
    return verify_access_token(token, db)

2. 修改用户接口：添加认证依赖

回到app/routes/user_router.py，给每个接口添加current_user: User = Depends(get_current_user)依赖：

from app.dependencies.auth import get_current_user  # 导入认证依赖
from app.models.user import User  # 导入User模型

# 示例：修改创建用户接口
@user_router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(
    user_in: UserCreate,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)  # 添加认证依赖
):
    """创建用户接口（需要登录）"""
    return user_service.create_user(db, user_in)

# 同理，修改其他接口：
# get_user、get_users、update_user、delete_user都要加current_user依赖

3. 测试认证效果

启动服务，用 Swagger 文档测试：

1. 先调用/api/auth/login接口，输入用户名 + 密码，拿到access_token；
2. 点击 Swagger 文档右上角的 **“Authorize”** 按钮，输入Bearer {access_token}（比如Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...）；
3. 之后调用/api/users接口，就能正常返回数据 —— 如果没登录或 Token 过期，会返回 401 错误（“无效的 Token 或已过期”）。

六、实现权限控制：不同角色访问不同接口

认证解决了 “你是谁” 的问题，权限控制解决 “你能做什么” 的问题 —— 比如：

• 普通用户只能访问自己的信息（/api/users/me）；
• 管理员能访问所有用户信息（/api/users）和删除用户（/api/users/{user_id}）。

1. 写权限依赖（app/dependencies/permissions.py）

from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session

from app.dependencies.auth import get_current_user
from app.models.user import User
from app.services.user_service import UserService

# 初始化UserService
user_service = UserService()

def has_permission(required_perm: str):
    """依赖：检查当前用户是否有指定权限"""
    def dependency(
        current_user: User = Depends(get_current_user),
        db: Session = Depends(get_db)
    ):
        # 1. 获取用户的所有权限（从角色中关联）
        user_permissions = set()
        for role in current_user.roles:
            for perm in role.permissions:
                user_permissions.add(perm.code)
        # 2. 检查是否有需要的权限
        if required_perm not in user_permissions:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="没有访问权限"
            )
        return current_user
    return dependency

2. 修改接口：添加权限依赖

比如，删除用户接口只有管理员能调用（需要user:delete权限），修改user_router.py：

from app.dependencies.permissions import has_permission  # 导入权限依赖

@user_router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(
    user_id: int,
    db: Session = Depends(get_db),
    current_user: User = Depends(has_permission("user:delete"))  # 添加权限依赖
):
    """删除用户接口（只有管理员能调用）"""
    user = user_service.get_or_404(db, user_id)
    user_service.logical_delete(db, user)

3. 测试权限控制效果

1. 用管理员用户（比如之前创建的xiaoming，有user:delete权限）登录，调用/api/users/1删除接口，能成功；
2. 用普通用户（没有user:delete权限）登录，调用删除接口，会返回 403 错误（“没有访问权限”）。

七、常见问题排查

1. 登录接口返回 “用户名或密码错误”

• 原因：用户名不存在，或密码错误；
• 解决：检查用户名是否正确，密码是否哈希正确（比如verify_password方法是否正确）。

2. 接口返回 401 错误（“无效的 Token 或已过期”）

• 原因：Token 过期，或 Token 被篡改，或请求没带 Token；
• 解决：重新登录获取新 Token，或检查请求头是否带Authorization: Bearer {token}。

3. 接口返回 403 错误（“没有访问权限”）

• 原因：用户没有对应的权限；
• 解决：给用户分配对应的角色（比如给普通用户添加user:delete权限）。

八、总结：这篇我们做了什么？

1. 搞懂了 JWT 的概念（电子门禁卡），实现了 JWT 的生成与验证；
2. 实现了登录接口，用用户名 + 密码换 JWT Token；
3. 用依赖注入实现了认证（只有登录用户才能访问接口）；
4. 用权限依赖实现了权限控制（不同角色访问不同接口）；
5. 测试了每个接口，确保认证和权限控制生效。