用户认证与JWT授权系统实现

2026-04-07 · 48 分钟阅读

在上一篇文章中，我们完成了RAG系统的深度优化和多会话支持，现在我们的知识库系统已经具备了优秀的回答质量和用户体验。但目前系统是完全开放的，任何人都可以访问和操作，这在企业环境中是不可接受的。这篇文章，我们将实现完整的用户认证与JWT授权系统，为系统添加安全防护，确保只有授权用户才能访问和使用知识库。

摘要

  • 理解JWT认证的核心原理与优势
  • 掌握FastAPI中JWT认证的实现方法
  • 实现安全的密码加密与存储
  • 开发用户注册、登录和注销接口
  • 在前端实现登录状态管理与路由守卫
  • 为系统接口添加权限验证，保护敏感操作

1. 为什么需要用户认证与JWT？

1.1 开放系统的安全风险

目前我们的系统存在以下严重的安全问题：

  • 任何人都可以上传和删除文档
  • 任何人都可以访问所有知识库内容
  • 没有用户身份标识，无法追踪操作记录
  • 无法实现基于角色的权限控制

在企业环境中，这些问题可能导致数据泄露、知识资产丢失和系统被滥用。因此，用户认证是构建生产级系统必不可少的一环。

1.2 常见的认证方式对比

认证方式 优点 缺点 适用场景
Session-Cookie 简单易用，服务器可控 有状态，扩展性差，跨域问题 传统单体应用
JWT (JSON Web Token) 无状态，扩展性好，跨域支持 无法即时注销，token大小有限 前后端分离应用，微服务架构
OAuth 2.0 安全，支持第三方登录 实现复杂 开放平台，第三方应用接入

对于我们的前后端分离知识库系统，JWT是最佳选择。它具有无状态、易于扩展、支持跨域等优点，非常适合现代Web应用。

1.3 JWT的工作原理

JWT是一种紧凑的、URL安全的方式，用于在双方之间传递声明。它由三部分组成，用点(.)分隔：

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoxLCJ1c2VybmFtZSI6ImFkbWluIiwiZXhwIjoxNzE2NzI4MDAwfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c
  • Header（头部）：指定签名算法和令牌类型
  • Payload（载荷）：包含要传递的声明（用户ID、用户名、过期时间等）
  • Signature（签名）：使用密钥对头部和载荷进行签名，防止篡改

JWT认证流程：

  1. 用户输入用户名和密码登录
  2. 服务器验证用户名和密码，生成JWT token返回给客户端
  3. 客户端将token存储在localStorage或sessionStorage中
  4. 后续请求中，客户端在Authorization头中携带token
  5. 服务器验证token的有效性，确认用户身份
  6. 服务器处理请求并返回结果

2. 后端认证系统实现

2.1 安装依赖

我们需要以下依赖来实现JWT认证：

  • python-jose：用于生成和验证JWT token
  • passlib：用于密码加密
  • python-multipart：用于处理表单数据
  • sqlalchemy：用于数据库操作
cd backend
source venv/bin/activate  # Windows 使用 venv\Scripts\activate
pip install python-jose[cryptography] passlib[bcrypt] sqlalchemy

2.2 数据库设计与模型

我们将使用SQLite作为开发环境的数据库，它不需要额外部署，非常适合开发。

创建database.py：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./knowledge_base.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

创建models.py，定义用户模型：

from sqlalchemy import Column, Integer, String, DateTime, Boolean
from database import Base
from datetime import datetime

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    full_name = Column(String)
    is_active = Column(Boolean, default=True)
    is_admin = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

创建schemas.py，定义Pydantic数据模型：

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional

class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str

class UserLogin(BaseModel):
    username: str
    password: str

class User(UserBase):
    id: int
    is_active: bool
    is_admin: bool
    created_at: datetime
    updated_at: datetime

    class Config:
        orm_mode = True

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    user_id: Optional[int] = None
    username: Optional[str] = None

2.3 安全工具函数

创建security.py，实现密码加密和JWT工具函数：

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from dotenv import load_dotenv
import os

load_dotenv()

# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT配置
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "your-secret-key-here-please-change-in-production")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30"))

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证密码是否正确"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """生成密码哈希"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """创建JWT访问令牌"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

在.env文件中添加JWT配置：

OPENAI_API_KEY=your-openai-api-key-here
CHROMA_PERSIST_DIRECTORY=./chroma_db
COHERE_API_KEY=your-cohere-api-key-here
JWT_SECRET_KEY=your-strong-secret-key-here-keep-it-safe  # 新增
ACCESS_TOKEN_EXPIRE_MINUTES=1440  # 24小时，新增

2.4 用户服务

创建services/user_service.py：

from sqlalchemy.orm import Session
from models import User
from schemas import UserCreate
from security import get_password_hash, verify_password

class UserService:
    def get_user_by_id(self, db: Session, user_id: int):
        return db.query(User).filter(User.id == user_id).first()
    
    def get_user_by_username(self, db: Session, username: str):
        return db.query(User).filter(User.username == username).first()
    
    def get_user_by_email(self, db: Session, email: str):
        return db.query(User).filter(User.email == email).first()
    
    def create_user(self, db: Session, user: UserCreate):
        hashed_password = get_password_hash(user.password)
        db_user = User(
            username=user.username,
            email=user.email,
            full_name=user.full_name,
            hashed_password=hashed_password
        )
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user
    
    def authenticate_user(self, db: Session, username: str, password: str):
        user = self.get_user_by_username(db, username)
        if not user:
            return False
        if not verify_password(password, user.hashed_password):
            return False
        return user

# 创建全局实例
user_service = UserService()

2.5 认证依赖项

创建dependencies.py，实现获取当前用户的依赖项：

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from database import SessionLocal
from security import SECRET_KEY, ALGORITHM
from schemas import TokenData
from services.user_service import user_service

security = HTTPBearer()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        token = credentials.credentials
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("user_id")
        username: str = payload.get("sub")
        if user_id is None or username is None:
            raise credentials_exception
        token_data = TokenData(user_id=user_id, username=username)
    except JWTError:
        raise credentials_exception
    
    user = user_service.get_user_by_id(db, user_id=token_data.user_id)
    if user is None:
        raise credentials_exception
    if not user.is_active:
        raise HTTPException(status_code=400, detail="用户已被禁用")
    return user

async def get_current_active_user(
    current_user = Depends(get_current_user)
):
    return current_user

async def get_current_admin_user(
    current_user = Depends(get_current_user)
):
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="需要管理员权限"
        )
    return current_user

2.6 认证接口实现

修改main.py，添加认证相关接口：

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from datetime import timedelta
from database import engine, Base
from schemas import UserCreate, UserLogin, User, Token
from services.user_service import user_service
from security import create_access_token, ACCESS_TOKEN_EXPIRE_MINUTES
from dependencies import get_db, get_current_active_user
from services.vector_service import vector_service
from services.document_service import document_service

# 创建数据库表
Base.metadata.create_all(bind=engine)

app = FastAPI(title="企业级智能知识库 API", version="0.1.0")

# 配置 CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 开发环境允许所有来源，生产环境请限制为具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ... 保留之前的请求模型 ...

# 认证接口
@app.post("/api/auth/register", response_model=User, status_code=status.HTTP_201_CREATED)
def register(user: UserCreate, db: Session = Depends(get_db)):
    db_user = user_service.get_user_by_username(db, username=user.username)
    if db_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="用户名已被注册"
        )
    
    db_user = user_service.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="邮箱已被注册"
        )
    
    return user_service.create_user(db=db, user=user)

@app.post("/api/auth/login", response_model=Token)
def login(user: UserLogin, db: Session = Depends(get_db)):
    user = user_service.authenticate_user(db, username=user.username, password=user.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"user_id": user.id, "sub": user.username},
        expires_delta=access_token_expires
    )
    
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/api/auth/me", response_model=User)
def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

# ... 为之前的接口添加权限验证 ...

@app.post("/api/documents/upload")
async def upload_document(
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_active_user)
):
    try:
        result = await document_service.upload_document(file)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/api/documents")
async def get_documents(current_user: User = Depends(get_current_active_user)):
    try:
        documents = document_service.get_all_documents()
        return {"documents": documents, "count": len(documents)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.delete("/api/documents/{document_id}")
async def delete_document(
    document_id: str,
    current_user: User = Depends(get_current_active_user)
):
    try:
        result = document_service.delete_document(document_id)
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/vector/search")
async def search_documents(
    request: SearchRequest,
    current_user: User = Depends(get_current_active_user)
):
    try:
        results = vector_service.multi_query_search(
            query=request.query,
            top_k_per_query=5,
            rerank_top_k=request.top_k
        )
        return {"results": results, "count": len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

3. 前端认证系统实现

3.1 API接口封装

修改src/api/index.ts，添加认证相关接口：

// 认证相关接口
export const authApi = {
  // 注册
  register: (data: { username: string; email: string; password: string; fullName?: string }) => 
    api.post('/auth/register', data),
  
  // 登录
  login: (data: { username: string; password: string }) => 
    api.post('/auth/login', data),
  
  // 获取当前用户信息
  getCurrentUser: () => 
    api.get('/auth/me'),
  
  // 注销
  logout: () => {
    localStorage.removeItem('token');
  }
};

// 添加请求拦截器，自动携带token
api.interceptors.request.use(
  (config) => {
    const token = localStorage.getItem('token');
    if (token) {
      config.headers.Authorization = `Bearer ${token}`;
    }
    return config;
  },
  (error) => {
    return Promise.reject(error);
  }
);

// 添加响应拦截器，处理401错误
api.interceptors.response.use(
  (response) => {
    return response;
  },
  (error) => {
    if (error.response && error.response.status === 401) {
      // 清除token并跳转到登录页
      localStorage.removeItem('token');
      window.location.href = '/login';
    }
    return Promise.reject(error);
  }
);

3.2 状态管理更新

修改src/store/index.ts，添加用户相关状态：

interface User {
  id: number;
  username: string;
  email: string;
  fullName: string | null;
  isAdmin: boolean;
}

interface AppState {
  // 用户状态
  user: User | null;
  isAuthenticated: boolean;
  isLoading: boolean;
  
  // 会话状态
  sessions: Session[];
  currentSessionId: string | null;
  messages: Message[];
  isChatLoading: boolean;
  
  // 文档状态
  documents: any[];
  isUploading: boolean;
  uploadProgress: number;
  
  // 方法
  setUser: (user: User | null) => void;
  setIsAuthenticated: (value: boolean) => void;
  setIsLoading: (value: boolean) => void;
  login: (token: string, user: User) => void;
  logout: () => void;
  
  // ... 保留之前的方法 ...
}

export const useAppStore = create<AppState>((set) => ({
  user: null,
  isAuthenticated: false,
  isLoading: true,
  
  sessions: [],
  currentSessionId: null,
  messages: [],
  isChatLoading: false,
  
  documents: [],
  isUploading: false,
  uploadProgress: 0,
  
  setUser: (user) => set({ user }),
  setIsAuthenticated: (value) => set({ isAuthenticated: value }),
  setIsLoading: (value) => set({ isLoading: value }),
  
  login: (token, user) => {
    localStorage.setItem('token', token);
    set({ 
      user, 
      isAuthenticated: true 
    });
  },
  
  logout: () => {
    localStorage.removeItem('token');
    set({ 
      user: null, 
      isAuthenticated: false,
      sessions: [],
      currentSessionId: null,
      messages: []
    });
  },
  
  // ... 保留之前的方法 ...
}));

3.3 登录与注册页面

创建src/pages/LoginPage.tsx：

import { useState } from 'react';
import { Card, Form, Input, Button, Tabs, message, Typography } from 'antd';
import { UserOutlined, LockOutlined, MailOutlined } from '@ant-design/icons';
import { useNavigate } from 'react-router-dom';
import { authApi } from '../api';
import { useAppStore } from '../store';

const { Title } = Typography;

const LoginPage = () => {
  const [activeTab, setActiveTab] = useState('login');
  const [loading, setLoading] = useState(false);
  const [loginForm] = Form.useForm();
  const [registerForm] = Form.useForm();
  
  const navigate = useNavigate();
  const { login } = useAppStore();
  
  // 处理登录
  const handleLogin = async (values: any) => {
    setLoading(true);
    try {
      const response = await authApi.login(values);
      const { access_token } = response.data;
      
      // 获取用户信息
      const userResponse = await authApi.getCurrentUser();
      login(access_token, userResponse.data);
      
      message.success('登录成功');
      navigate('/');
    } catch (error: any) {
      message.error(error.response?.data?.detail || '登录失败');
    } finally {
      setLoading(false);
    }
  };
  
  // 处理注册
  const handleRegister = async (values: any) => {
    setLoading(true);
    try {
      await authApi.register({
        username: values.username,
        email: values.email,
        password: values.password,
        fullName: values.fullName
      });
      
      message.success('注册成功，请登录');
      setActiveTab('login');
      loginForm.setFieldsValue({
        username: values.username,
        password: values.password
      });
    } catch (error: any) {
      message.error(error.response?.data?.detail || '注册失败');
    } finally {
      setLoading(false);
    }
  };
  
  return (
    <div style={{ 
      display: 'flex', 
      justifyContent: 'center', 
      alignItems: 'center', 
      minHeight: '100vh',
      background: '#f0f2f5'
    }}>
      <Card style={{ width: 400, boxShadow: '0 4px 12px rgba(0,0,0,0.1)' }}>
        <div style={{ textAlign: 'center', marginBottom: 24 }}>
          <Title level={2}>企业级智能知识库</Title>
        </div>
        
        <Tabs activeKey={activeTab} onChange={setActiveTab}>
          <Tabs.TabPane tab="登录" key="login">
            <Form
              form={loginForm}
              layout="vertical"
              onFinish={handleLogin}
              autoComplete="off"
            >
              <Form.Item
                name="username"
                label="用户名"
                rules={[{ required: true, message: '请输入用户名' }]}
              >
                <Input prefix={<UserOutlined />} placeholder="请输入用户名" />
              </Form.Item>
              
              <Form.Item
                name="password"
                label="密码"
                rules={[{ required: true, message: '请输入密码' }]}
              >
                <Input.Password prefix={<LockOutlined />} placeholder="请输入密码" />
              </Form.Item>
              
              <Form.Item>
                <Button type="primary" htmlType="submit" loading={loading} block>
                  登录
                </Button>
              </Form.Item>
            </Form>
          </Tabs.TabPane>
          
          <Tabs.TabPane tab="注册" key="register">
            <Form
              form={registerForm}
              layout="vertical"
              onFinish={handleRegister}
              autoComplete="off"
            >
              <Form.Item
                name="username"
                label="用户名"
                rules={[{ required: true, message: '请输入用户名' }]}
              >
                <Input prefix={<UserOutlined />} placeholder="请输入用户名" />
              </Form.Item>
              
              <Form.Item
                name="email"
                label="邮箱"
                rules={[
                  { required: true, message: '请输入邮箱' },
                  { type: 'email', message: '请输入有效的邮箱地址' }
                ]}
              >
                <Input prefix={<MailOutlined />} placeholder="请输入邮箱" />
              </Form.Item>
              
              <Form.Item
                name="fullName"
                label="姓名"
              >
                <Input placeholder="请输入姓名（可选）" />
              </Form.Item>
              
              <Form.Item
                name="password"
                label="密码"
                rules={[
                  { required: true, message: '请输入密码' },
                  { min: 6, message: '密码至少6个字符' }
                ]}
              >
                <Input.Password prefix={<LockOutlined />} placeholder="请输入密码" />
              </Form.Item>
              
              <Form.Item
                name="confirmPassword"
                label="确认密码"
                dependencies={['password']}
                rules={[
                  { required: true, message: '请确认密码' },
                  ({ getFieldValue }) => ({
                    validator(_, value) {
                      if (!value || getFieldValue('password') === value) {
                        return Promise.resolve();
                      }
                      return Promise.reject(new Error('两次输入的密码不一致'));
                    },
                  }),
                ]}
              >
                <Input.Password prefix={<LockOutlined />} placeholder="请确认密码" />
              </Form.Item>
              
              <Form.Item>
                <Button type="primary" htmlType="submit" loading={loading} block>
                  注册
                </Button>
              </Form.Item>
            </Form>
          </Tabs.TabPane>
        </Tabs>
      </Card>
    </div>
  );
};

export default LoginPage;

3.4 路由守卫实现

创建src/components/ProtectedRoute.tsx：

import { Navigate } from 'react-router-dom';
import { useAppStore } from '../store';
import { Spin } from 'antd';

interface ProtectedRouteProps {
  children: React.ReactNode;
}

const ProtectedRoute: React.FC<ProtectedRouteProps> = ({ children }) => {
  const { isAuthenticated, isLoading } = useAppStore();
  
  if (isLoading) {
    return (
      <div style={{ 
        display: 'flex', 
        justifyContent: 'center', 
        alignItems: 'center', 
        height: '100vh' 
      }}>
        <Spin size="large" />
      </div>
    );
  }
  
  if (!isAuthenticated) {
    return <Navigate to="/login" replace />;
  }
  
  return <>{children}</>;
};

export default ProtectedRoute;

3.5 应用初始化与路由更新

修改src/App.tsx，添加应用初始化逻辑和路由守卫：

import { useEffect } from 'react';
import { BrowserRouter as Router, Routes, Route } from 'react-router-dom';
import { Layout, Menu, Button, Dropdown, Avatar, Typography } from 'antd';
import { 
  FileOutlined, 
  MessageOutlined, 
  SettingOutlined, 
  UserOutlined,
  LogoutOutlined
} from '@ant-design/icons';
import DocumentPage from './pages/DocumentPage';
import ChatPage from './pages/ChatPage';
import SettingsPage from './pages/SettingsPage';
import LoginPage from './pages/LoginPage';
import ProtectedRoute from './components/ProtectedRoute';
import { useAppStore } from './store';
import { authApi } from './api';
import './App.css';

const { Header, Sider, Content } = Layout;
const { Text } = Typography;

function App() {
  const { user, isAuthenticated, setUser, setIsAuthenticated, setIsLoading, logout } = useAppStore();
  
  // 应用初始化：检查登录状态
  useEffect(() => {
    const initApp = async () => {
      const token = localStorage.getItem('token');
      if (token) {
        try {
          const response = await authApi.getCurrentUser();
          setUser(response.data);
          setIsAuthenticated(true);
        } catch (error) {
          localStorage.removeItem('token');
        }
      }
      setIsLoading(false);
    };
    
    initApp();
  }, [setUser, setIsAuthenticated, setIsLoading]);
  
  // 用户菜单
  const userMenuItems = [
    {
      key: 'logout',
      icon: <LogoutOutlined />,
      label: '退出登录',
      onClick: logout
    }
  ];
  
  return (
    <Router>
      <Routes>
        <Route path="/login" element={<LoginPage />} />
        
        <Route path="*" element={
          <ProtectedRoute>
            <Layout style={{ minHeight: '100vh' }}>
              <Header style={{ 
                background: '#fff', 
                padding: '0 20px', 
                boxShadow: '0 2px 8px rgba(0,0,0,0.1)',
                display: 'flex',
                justifyContent: 'space-between',
                alignItems: 'center'
              }}>
                <h1 style={{ margin: 0, fontSize: '20px', fontWeight: 'bold' }}>企业级智能知识库</h1>
                
                <Dropdown menu={{ items: userMenuItems }} placement="bottomRight">
                  <div style={{ cursor: 'pointer', display: 'flex', alignItems: 'center' }}>
                    <Avatar icon={<UserOutlined />} style={{ marginRight: 8 }} />
                    <Text>{user?.username}</Text>
                  </div>
                </Dropdown>
              </Header>
              
              <Layout>
                <Sider width={200} style={{ background: '#fff' }}>
                  <Menu
                    mode="inline"
                    defaultSelectedKeys={['documents']}
                    style={{ height: '100%', borderRight: 0 }}
                    items={[
                      {
                        key: 'documents',
                        icon: <FileOutlined />,
                        label: '文档管理',
                        path: '/',
                      },
                      {
                        key: 'chat',
                        icon: <MessageOutlined />,
                        label: '智能问答',
                        path: '/chat',
                      },
                      {
                        key: 'settings',
                        icon: <SettingOutlined />,
                        label: '系统设置',
                        path: '/settings',
                      },
                    ]}
                  />
                </Sider>
                
                <Content style={{ padding: '24px', background: '#f5f5f5' }}>
                  <Routes>
                    <Route path="/" element={<DocumentPage />} />
                    <Route path="/chat" element={<ChatPage />} />
                    <Route path="/settings" element={<SettingsPage />} />
                  </Routes>
                </Content>
              </Layout>
            </Layout>
          </ProtectedRoute>
        } />
      </Routes>
    </Router>
  );
}

export default App;

4. 测试认证系统

现在我们来测试完整的认证流程：

  1. 重启所有服务：

    • FastAPI 后端：cd backend && uvicorn main:app --reload --port 8000
    • OpenClaw 网关：cd openclaw && pnpm run start
    • React 前端：cd frontend && pnpm run dev

  2. 测试注册功能：

    • 访问 http://localhost:5173，会自动跳转到登录页
    • 切换到"注册"标签页
    • 输入用户名、邮箱、密码，点击注册
    • 注册成功后会自动切换到登录页

  3. 测试登录功能：

    • 输入刚才注册的用户名和密码
    • 点击登录，成功后会跳转到文档管理页面
    • 右上角会显示当前登录的用户名

  4. 测试权限验证：

    • 尝试上传一个文档，验证是否需要登录
    • 打开浏览器的开发者工具，删除localStorage中的token
    • 刷新页面，会自动跳转到登录页
    • 尝试直接访问 http://localhost:5173/chat，会自动跳转到登录页

  5. 测试注销功能：

    • 点击右上角的用户名，选择"退出登录"
    • 会自动跳转到登录页
    • 此时所有需要认证的接口都无法访问

5. 安全注意事项

5.1 JWT安全最佳实践

  • 使用强密钥：在生产环境中使用足够长且随机的JWT密钥
  • 设置合理的过期时间：不要设置过长的过期时间，建议1-24小时
  • 不要在Payload中存储敏感信息：JWT是Base64编码的，任何人都可以解码
  • 使用HTTPS：防止JWT在传输过程中被窃取
  • 实现刷新token机制：使用短期访问token和长期刷新token
  • 添加token黑名单：对于已注销但未过期的token，加入黑名单

5.2 密码安全

  • 永远不要存储明文密码：使用bcrypt等算法进行加密
  • 强制密码强度：要求用户使用包含大小写字母、数字和特殊字符的密码
  • 限制登录尝试次数：防止暴力破解
  • 支持双因素认证：对于重要系统，添加双因素认证

5.3 API安全

  • 输入验证：对所有用户输入进行严格验证
  • 接口限流：防止DoS攻击
  • CORS配置：在生产环境中限制允许的来源
  • 错误处理：不要在错误信息中暴露敏感信息

6. 项目结构更新

至此，我们的项目目录已经更新为：

enterprise-knowledge-base/
├── .gitignore
├── backend/
│   ├── venv/
│   ├── .env
│   ├── main.py
│   ├── requirements.txt
│   ├── uploads/
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   ├── security.py
│   ├── dependencies.py
│   ├── utils/
│   │   ├── text_splitter.py
│   │   └── document_parser.py
│   └── services/
│       ├── vector_service.py
│       ├── document_service.py
│       ├── chat_service.py
│       └── user_service.py
├── openclaw/
│   ├── node_modules/
│   ├── package.json
│   ├── .claw/
│   │   └── config.yaml
│   └── skills/
│       ├── hello-world/
│       ├── document-parser/
│       └── rag-retrieval/
└── frontend/
    ├── node_modules/
    ├── src/
    │   ├── api/
    │   │   └── index.ts
    │   ├── components/
    │   │   ├── SessionList.tsx
    │   │   └── ProtectedRoute.tsx
    │   ├── pages/
    │   │   ├── DocumentPage.tsx
    │   │   ├── ChatPage.tsx
    │   │   ├── SettingsPage.tsx
    │   │   ├── LoginPage.tsx
    │   │   └── ChatPage.css
    │   ├── store/
    │   │   └── index.ts
    │   ├── utils/
    │   ├── App.tsx
    │   ├── App.css
    │   └── main.tsx
    ├── index.html
    ├── package.json
    └── tsconfig.json

7. 小结

本篇文章我们完成了以下事情：

  • 深入理解了JWT认证的核心原理和优势
  • 在FastAPI后端实现了完整的用户认证系统
  • 设计并实现了用户数据库模型和安全的密码加密
  • 开发了用户注册、登录和获取当前用户信息接口
  • 为所有敏感接口添加了权限验证
  • 在前端实现了登录状态管理和路由守卫
  • 开发了美观的登录和注册页面
  • 测试了完整的认证流程
  • 总结了JWT和API安全的最佳实践