bfe39e45a9
5. 修改统一认证登录和管理员登录是通过接口形式进行,存储返回的accessToken。 6. 修改交叉评查的部分样式
33 KiB
33 KiB
角色-路由权限实现方案
版本: v1.0 日期: 2025-11-17 状态: 设计完成,待实现
目录
功能概述
业务需求
实现基于角色的前端路由权限控制,使不同角色的用户登录后看到不同的菜单和页面:
- ✅ 系统管理员 - 可访问所有路由(用户管理、角色管理、权限管理等)
- ✅ 文档管理员 - 可访问文档管理相关路由
- ✅ 文档审查员 - 可访问文档查看和评查路由
- ✅ 普通用户 - 仅可访问基础路由(首页、个人中心)
核心功能
- 路由定义管理 - 系统路由的增删改查
- 角色-路由关联 - 配置角色可访问的路由
- 用户路由查询 - 根据用户角色获取可访问路由列表
- 动态菜单生成 - 前端根据路由列表动态生成菜单
- 路由守卫 - 前端路由守卫拦截未授权访问
设计原则
1. 高可用性原则
- 缓存优先: 路由权限数据缓存到 Redis(TTL: 30分钟)
- 降级策略: Redis 不可用时直接查询数据库
- 异步刷新: 后台定时刷新缓存,避免缓存雪崩
2. 高性能原则
- 单次查询: 用户登录时一次性获取所有可访问路由
- 前端缓存: 路由数据存储在前端 LocalStorage
- 懒加载: 路由组件按需加载
3. 安全性原则
- 前后端双重验证: 前端路由守卫 + 后端 API 权限校验
- 最小权限: 默认拒绝所有访问,显式授权
- 审计日志: 记录路由访问日志
4. 可维护性原则
- 配置化: 路由通过数据库配置,无需修改代码
- 层级结构: 支持多级路由嵌套
- 元信息: 路由携带图标、标题等元信息
架构设计
整体架构
┌─────────────────────────────────────────────────────────────┐
│ 前端应用 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 路由守卫 │ │ 动态菜单 │ │ 权限指令 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓ HTTP请求
┌─────────────────────────────────────────────────────────────┐
│ FastAPI 后端 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 路由权限API │ │ 路由管理API │ │ 角色管理API │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ ┌──────────────────────────────────────────────────┐ │
│ │ RoutePermission 模块 │ │
│ │ - get_user_routes() 获取用户可访问路由 │ │
│ │ - get_role_routes() 获取角色可访问路由 │ │
│ │ - check_route_access() 检查路由访问权限 │ │
│ └──────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 数据层 │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ sys_routes │ │ role_route │ │ Redis Cache │ │
│ │ (路由定义) │ │ (角色-路由) │ │ (路由缓存) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────────────────┘
数据流向
用户登录流程
1. 用户登录
↓
2. 后端验证凭据
↓
3. 查询用户角色
↓
4. 根据角色查询可访问路由(优先从 Redis 缓存读取)
↓
5. 返回用户信息 + 路由列表
↓
6. 前端存储路由数据到 LocalStorage
↓
7. Vue Router 动态注册路由
↓
8. 根据路由生成侧边栏菜单
路由访问流程
1. 用户访问路由(如 /system/users)
↓
2. 前端路由守卫拦截
↓
3. 检查路由是否在可访问列表
↓
4. [是] 允许访问 → 加载页面组件
↓
5. [否] 拒绝访问 → 跳转 403 页面
数据库设计
表结构
1. sys_routes(系统路由表)
已存在,需要确认字段:
CREATE TABLE IF NOT EXISTS sys_routes (
id SERIAL PRIMARY KEY,
route_path VARCHAR(255) NOT NULL, -- 路由路径 /system/users
route_name VARCHAR(100) NOT NULL, -- 路由名称 UserManagement
component VARCHAR(255), -- 组件路径 views/system/Users.vue
parent_id INTEGER, -- 父路由ID(支持多级路由)
route_title VARCHAR(100), -- 路由标题(中文)
icon VARCHAR(50), -- 图标名称
sort_order INTEGER DEFAULT 0, -- 排序顺序
is_hidden BOOLEAN DEFAULT FALSE, -- 是否隐藏(隐藏的路由不显示在菜单)
is_cache BOOLEAN DEFAULT TRUE, -- 是否缓存(KeepAlive)
meta JSONB, -- 元信息(扩展字段)
status INTEGER DEFAULT 0, -- 状态 0=启用 1=禁用
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP,
CONSTRAINT fk_parent_route FOREIGN KEY (parent_id) REFERENCES sys_routes(id) ON DELETE CASCADE
);
-- 索引
CREATE INDEX idx_routes_parent_id ON sys_routes(parent_id);
CREATE INDEX idx_routes_status ON sys_routes(status);
CREATE UNIQUE INDEX idx_routes_path ON sys_routes(route_path) WHERE deleted_at IS NULL;
字段说明:
route_path: 前端路由路径,如/system/usersroute_name: 路由名称(唯一标识),如UserManagementcomponent: 组件路径,如views/system/Users.vueparent_id: 父路由ID,支持多级嵌套菜单route_title: 菜单显示的标题(中文)icon: 图标(Element Plus / Ant Design 图标名)is_hidden: 是否在菜单中隐藏(某些路由不显示在菜单,但可访问)meta: JSON 扩展字段,存储额外信息(如权限标识、面包屑等)
2. role_route(角色-路由关联表)
已存在,需要确认字段:
CREATE TABLE IF NOT EXISTS role_route (
id SERIAL PRIMARY KEY,
role_id INTEGER NOT NULL,
route_id INTEGER NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
CONSTRAINT fk_role FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
CONSTRAINT fk_route FOREIGN KEY (route_id) REFERENCES sys_routes(id) ON DELETE CASCADE,
CONSTRAINT uk_role_route UNIQUE (role_id, route_id)
);
-- 索引
CREATE INDEX idx_role_route_role_id ON role_route(role_id);
CREATE INDEX idx_role_route_route_id ON role_route(route_id);
数据示例
系统路由数据
-- 一级路由:首页
INSERT INTO sys_routes (id, route_path, route_name, component, route_title, icon, sort_order)
VALUES (1, '/dashboard', 'Dashboard', 'views/Dashboard.vue', '首页', 'el-icon-house', 1);
-- 一级路由:系统管理(父菜单)
INSERT INTO sys_routes (id, route_path, route_name, component, route_title, icon, sort_order)
VALUES (2, '/system', 'System', 'Layout', '系统管理', 'el-icon-setting', 10);
-- 二级路由:用户管理
INSERT INTO sys_routes (id, route_path, route_name, component, parent_id, route_title, icon, sort_order)
VALUES (3, '/system/users', 'SystemUsers', 'views/system/Users.vue', 2, '用户管理', 'el-icon-user', 1);
-- 二级路由:角色管理
INSERT INTO sys_routes (id, route_path, route_name, component, parent_id, route_title, icon, sort_order)
VALUES (4, '/system/roles', 'SystemRoles', 'views/system/Roles.vue', 2, '角色管理', 'el-icon-user-filled', 2);
-- 一级路由:文档管理
INSERT INTO sys_routes (id, route_path, route_name, component, route_title, icon, sort_order)
VALUES (10, '/documents', 'Documents', 'Layout', '文档管理', 'el-icon-document', 20);
-- 二级路由:文档列表
INSERT INTO sys_routes (id, route_path, route_name, component, parent_id, route_title, icon, sort_order)
VALUES (11, '/documents/list', 'DocumentList', 'views/documents/List.vue', 10, '文档列表', 'el-icon-tickets', 1);
角色-路由关联数据
-- 系统管理员 - 拥有所有路由
INSERT INTO role_route (role_id, route_id)
SELECT 1, id FROM sys_routes WHERE deleted_at IS NULL;
-- 文档管理员 - 拥有首页 + 文档管理路由
INSERT INTO role_route (role_id, route_id)
VALUES (2, 1), (2, 10), (2, 11);
-- 普通用户 - 只有首页
INSERT INTO role_route (role_id, route_id)
VALUES (8, 1);
后端实现
文件结构
app/
├── rbac/
│ ├── route_permission.py # 路由权限检查模块(新建)
│ ├── permission_checker.py # 现有权限检查模块
│ └── data_scope_injector.py # 现有数据范围注入模块
├── routes/
│ └── rbac_routes.py # RBAC 路由管理接口(新建)
核心模块:route_permission.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
路由权限检查模块
提供基于角色的路由权限检查功能
"""
from typing import List, Dict, Optional, Set
import asyncpg
from core.database import get_asyncpg_pool
from core.redis_client import get_redis_client
from core.logger import rbac_logger
import json
class RoutePermission:
"""路由权限管理器"""
# Redis 缓存键前缀
CACHE_PREFIX = "rbac:routes"
CACHE_TTL = 1800 # 30分钟
@classmethod
async def get_user_routes(
cls,
user_id: int,
use_cache: bool = True
) -> List[Dict]:
"""
获取用户可访问的路由列表
Args:
user_id: 用户ID
use_cache: 是否使用缓存
Returns:
路由列表(树形结构)
"""
# 1. 尝试从缓存获取
if use_cache:
cached = await cls._get_routes_from_cache(user_id)
if cached is not None:
rbac_logger.info(f"路由权限缓存命中: user={user_id}")
return cached
# 2. 查询用户角色
pool = await get_asyncpg_pool()
async with pool.acquire() as conn:
roles = await conn.fetch(
"""
SELECT r.id, r.role_key, r.role_name
FROM user_role ur
JOIN roles r ON ur.role_id = r.id
WHERE ur.user_id = $1
""",
user_id
)
if not roles:
rbac_logger.warning(f"用户无角色: user={user_id}")
return []
role_ids = [role['id'] for role in roles]
# 3. 查询角色可访问的路由
routes = await conn.fetch(
"""
SELECT DISTINCT
sr.id,
sr.route_path,
sr.route_name,
sr.component,
sr.parent_id,
sr.route_title,
sr.icon,
sr.sort_order,
sr.is_hidden,
sr.is_cache,
sr.meta
FROM role_route rr
JOIN sys_routes sr ON rr.route_id = sr.id
WHERE rr.role_id = ANY($1)
AND sr.status = 0
AND sr.deleted_at IS NULL
ORDER BY sr.sort_order, sr.id
""",
role_ids
)
# 4. 转换为字典列表
route_list = [dict(route) for route in routes]
# 5. 构建树形结构
route_tree = cls._build_route_tree(route_list)
# 6. 缓存结果
if use_cache:
await cls._cache_routes(user_id, route_tree)
rbac_logger.info(f"查询用户路由: user={user_id}, routes={len(route_list)}")
return route_tree
@classmethod
async def get_role_routes(
cls,
role_id: int,
use_cache: bool = True
) -> List[Dict]:
"""
获取角色可访问的路由列表
Args:
role_id: 角色ID
use_cache: 是否使用缓存
Returns:
路由列表
"""
# 缓存键
cache_key = f"{cls.CACHE_PREFIX}:role:{role_id}"
# 1. 尝试从缓存获取
if use_cache:
redis = await get_redis_client()
try:
cached = await redis.get(cache_key)
if cached:
rbac_logger.info(f"角色路由缓存命中: role={role_id}")
return json.loads(cached)
except Exception as e:
rbac_logger.warning(f"读取路由缓存失败: {e}")
# 2. 查询数据库
pool = await get_asyncpg_pool()
async with pool.acquire() as conn:
routes = await conn.fetch(
"""
SELECT
sr.id,
sr.route_path,
sr.route_name,
sr.route_title
FROM role_route rr
JOIN sys_routes sr ON rr.route_id = sr.id
WHERE rr.role_id = $1
AND sr.status = 0
AND sr.deleted_at IS NULL
ORDER BY sr.sort_order
""",
role_id
)
route_list = [dict(route) for route in routes]
# 3. 缓存结果
if use_cache:
redis = await get_redis_client()
try:
await redis.setex(
cache_key,
cls.CACHE_TTL,
json.dumps(route_list, ensure_ascii=False)
)
except Exception as e:
rbac_logger.warning(f"缓存路由失败: {e}")
return route_list
@classmethod
async def check_route_access(
cls,
user_id: int,
route_path: str
) -> bool:
"""
检查用户是否有访问指定路由的权限
Args:
user_id: 用户ID
route_path: 路由路径
Returns:
是否有权限
"""
routes = await cls.get_user_routes(user_id, use_cache=True)
# 递归检查路由树
def check_in_tree(routes_list: List[Dict]) -> bool:
for route in routes_list:
if route['route_path'] == route_path:
return True
if route.get('children'):
if check_in_tree(route['children']):
return True
return False
has_access = check_in_tree(routes)
rbac_logger.info(
f"路由访问检查: user={user_id}, route={route_path}, "
f"result={has_access}"
)
return has_access
@classmethod
def _build_route_tree(cls, routes: List[Dict]) -> List[Dict]:
"""
构建路由树形结构
Args:
routes: 路由列表(扁平)
Returns:
路由树(嵌套)
"""
# 按 parent_id 分组
route_map = {}
root_routes = []
for route in routes:
route_id = route['id']
route_map[route_id] = route
route['children'] = []
# 构建父子关系
for route in routes:
parent_id = route.get('parent_id')
if parent_id and parent_id in route_map:
route_map[parent_id]['children'].append(route)
else:
root_routes.append(route)
# 移除空 children
def remove_empty_children(routes_list):
for route in routes_list:
if not route['children']:
del route['children']
else:
remove_empty_children(route['children'])
remove_empty_children(root_routes)
return root_routes
@classmethod
async def _get_routes_from_cache(cls, user_id: int) -> Optional[List[Dict]]:
"""从缓存获取路由列表"""
cache_key = f"{cls.CACHE_PREFIX}:user:{user_id}"
redis = await get_redis_client()
try:
cached = await redis.get(cache_key)
if cached:
return json.loads(cached)
except Exception as e:
rbac_logger.warning(f"读取路由缓存失败: {e}")
return None
@classmethod
async def _cache_routes(cls, user_id: int, routes: List[Dict]):
"""缓存路由列表"""
cache_key = f"{cls.CACHE_PREFIX}:user:{user_id}"
redis = await get_redis_client()
try:
await redis.setex(
cache_key,
cls.CACHE_TTL,
json.dumps(routes, ensure_ascii=False)
)
except Exception as e:
rbac_logger.warning(f"缓存路由失败: {e}")
@classmethod
async def clear_user_routes_cache(cls, user_id: int):
"""清除用户路由缓存"""
cache_key = f"{cls.CACHE_PREFIX}:user:{user_id}"
redis = await get_redis_client()
try:
await redis.delete(cache_key)
rbac_logger.info(f"清除路由缓存: user={user_id}")
except Exception as e:
rbac_logger.warning(f"清除路由缓存失败: {e}")
@classmethod
async def clear_role_routes_cache(cls, role_id: int):
"""清除角色路由缓存"""
cache_key = f"{cls.CACHE_PREFIX}:role:{role_id}"
redis = await get_redis_client()
try:
await redis.delete(cache_key)
rbac_logger.info(f"清除角色路由缓存: role={role_id}")
except Exception as e:
rbac_logger.warning(f"清除角色路由缓存失败: {e}")
API 接口:rbac_routes.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
RBAC 路由管理接口
提供路由权限相关的 API 端点
"""
from fastapi import APIRouter, Depends, Query
from typing import List, Dict, Optional
from pydantic import BaseModel, Field
from app.rbac.route_permission import RoutePermission
from app.auth.auth import verify_token, User
from app.base_api import unified_resp
from core.logger import api_logger
router_rbac = APIRouter(prefix="/rbac", tags=["RBAC-路由权限"])
class RouteResponse(BaseModel):
"""路由响应模型"""
id: int
route_path: str
route_name: str
component: Optional[str]
parent_id: Optional[int]
route_title: str
icon: Optional[str]
sort_order: int
is_hidden: bool
is_cache: bool
meta: Optional[Dict]
children: Optional[List['RouteResponse']] = None
class UserRoutesResponse(BaseModel):
"""用户路由响应"""
user_id: int
username: str
routes: List[RouteResponse]
@router_rbac.get("/user/routes", summary="获取当前用户可访问路由")
@unified_resp
async def get_current_user_routes(
current_user: User = Depends(verify_token)
) -> Dict:
"""
获取当前用户可访问的路由列表(树形结构)
返回数据用于前端动态路由和菜单生成
"""
try:
routes = await RoutePermission.get_user_routes(
user_id=current_user.id,
use_cache=True
)
api_logger.info(f"获取用户路由: user={current_user.id}, count={len(routes)}")
return {
'user_id': current_user.id,
'username': current_user.username,
'routes': routes
}
except Exception as e:
api_logger.error(f"获取用户路由失败: {e}", exc_info=True)
raise
@router_rbac.get("/roles/{role_id}/routes", summary="获取角色可访问路由")
@unified_resp
async def get_role_routes(
role_id: int,
current_user: User = Depends(verify_token)
) -> Dict:
"""
获取指定角色可访问的路由列表
用于角色管理页面展示角色权限
"""
try:
routes = await RoutePermission.get_role_routes(
role_id=role_id,
use_cache=True
)
api_logger.info(f"获取角色路由: role={role_id}, count={len(routes)}")
return {
'role_id': role_id,
'routes': routes
}
except Exception as e:
api_logger.error(f"获取角色路由失败: {e}", exc_info=True)
raise
@router_rbac.get("/check-route", summary="检查路由访问权限")
@unified_resp
async def check_route_access(
route_path: str = Query(..., description="路由路径"),
current_user: User = Depends(verify_token)
) -> Dict:
"""
检查当前用户是否有访问指定路由的权限
前端可用此接口动态控制按钮/链接显示
"""
try:
has_access = await RoutePermission.check_route_access(
user_id=current_user.id,
route_path=route_path
)
return {
'route_path': route_path,
'has_access': has_access
}
except Exception as e:
api_logger.error(f"检查路由权限失败: {e}", exc_info=True)
raise
前端实现
1. 路由配置(Vue3 + Vue Router 4)
// router/index.js
import { createRouter, createWebHistory } from 'vue-router'
import { useUserStore } from '@/stores/user'
// 静态路由(无需权限)
const constantRoutes = [
{
path: '/login',
name: 'Login',
component: () => import('@/views/Login.vue'),
meta: { title: '登录' }
},
{
path: '/403',
name: 'Forbidden',
component: () => import('@/views/error/403.vue'),
meta: { title: '无权限' }
},
{
path: '/404',
name: 'NotFound',
component: () => import('@/views/error/404.vue'),
meta: { title: '页面不存在' }
}
]
const router = createRouter({
history: createWebHistory(),
routes: constantRoutes
})
// 全局路由守卫
router.beforeEach(async (to, from, next) => {
const userStore = useUserStore()
const token = localStorage.getItem('access_token')
// 白名单路由
const whiteList = ['/login', '/403', '/404']
if (token) {
if (to.path === '/login') {
next('/')
return
}
// 检查是否已加载动态路由
if (!userStore.hasLoadedRoutes) {
try {
// 获取用户路由
await userStore.loadUserRoutes()
// 动态添加路由
const dynamicRoutes = userStore.routes
dynamicRoutes.forEach(route => {
router.addRoute(route)
})
// 标记已加载
userStore.hasLoadedRoutes = true
// 重新导航到目标路由
next({ ...to, replace: true })
return
} catch (error) {
console.error('加载路由失败:', error)
// 清除 token,跳转登录
localStorage.removeItem('access_token')
next('/login')
return
}
}
// 检查路由权限
if (!userStore.hasRoute(to.path)) {
next('/403')
return
}
next()
} else {
// 未登录
if (whiteList.includes(to.path)) {
next()
} else {
next(`/login?redirect=${to.path}`)
}
}
})
export default router
2. Pinia Store(用户状态管理)
// stores/user.js
import { defineStore } from 'pinia'
import { ref, computed } from 'vue'
import apiClient from '@/utils/request'
export const useUserStore = defineStore('user', () => {
const userInfo = ref(null)
const routes = ref([])
const hasLoadedRoutes = ref(false)
// 加载用户路由
async function loadUserRoutes() {
try {
const response = await apiClient.get('/rbac/user/routes')
const data = response.data
userInfo.value = {
user_id: data.user_id,
username: data.username
}
// 转换后端路由为 Vue Router 格式
routes.value = convertRoutes(data.routes)
// 存储到 LocalStorage
localStorage.setItem('user_routes', JSON.stringify(data.routes))
return routes.value
} catch (error) {
console.error('加载用户路由失败:', error)
throw error
}
}
// 转换路由格式
function convertRoutes(backendRoutes) {
return backendRoutes.map(route => {
const vueRoute = {
path: route.route_path,
name: route.route_name,
component: () => import(`@/views/${route.component}`),
meta: {
title: route.route_title,
icon: route.icon,
hidden: route.is_hidden,
keepAlive: route.is_cache,
...route.meta
}
}
// 递归处理子路由
if (route.children && route.children.length > 0) {
vueRoute.children = convertRoutes(route.children)
}
return vueRoute
})
}
// 检查是否有指定路由权限
function hasRoute(path) {
const checkInRoutes = (routeList) => {
for (const route of routeList) {
if (route.path === path) {
return true
}
if (route.children && checkInRoutes(route.children)) {
return true
}
}
return false
}
return checkInRoutes(routes.value)
}
// 清除路由数据
function clearRoutes() {
routes.value = []
hasLoadedRoutes.value = false
localStorage.removeItem('user_routes')
}
return {
userInfo,
routes,
hasLoadedRoutes,
loadUserRoutes,
hasRoute,
clearRoutes
}
})
3. 动态菜单生成
<!-- components/Sidebar.vue -->
<template>
<el-menu
:default-active="activeMenu"
:collapse="isCollapse"
:unique-opened="true"
router
>
<sidebar-item
v-for="route in menuRoutes"
:key="route.path"
:item="route"
/>
</el-menu>
</template>
<script setup>
import { computed } from 'vue'
import { useRoute } from 'vue-router'
import { useUserStore } from '@/stores/user'
import SidebarItem from './SidebarItem.vue'
const route = useRoute()
const userStore = useUserStore()
// 当前激活菜单
const activeMenu = computed(() => route.path)
// 菜单路由(过滤隐藏的路由)
const menuRoutes = computed(() => {
return filterHiddenRoutes(userStore.routes)
})
// 过滤隐藏路由
function filterHiddenRoutes(routes) {
return routes
.filter(route => !route.meta?.hidden)
.map(route => {
if (route.children) {
route.children = filterHiddenRoutes(route.children)
}
return route
})
}
</script>
<!-- components/SidebarItem.vue -->
<template>
<div>
<!-- 有子菜单 -->
<el-sub-menu v-if="hasChildren" :index="item.path">
<template #title>
<el-icon v-if="item.meta?.icon">
<component :is="item.meta.icon" />
</el-icon>
<span>{{ item.meta?.title }}</span>
</template>
<sidebar-item
v-for="child in item.children"
:key="child.path"
:item="child"
/>
</el-sub-menu>
<!-- 无子菜单 -->
<el-menu-item v-else :index="item.path">
<el-icon v-if="item.meta?.icon">
<component :is="item.meta.icon" />
</el-icon>
<span>{{ item.meta?.title }}</span>
</el-menu-item>
</div>
</template>
<script setup>
import { computed } from 'vue'
const props = defineProps({
item: {
type: Object,
required: true
}
})
const hasChildren = computed(() => {
return props.item.children && props.item.children.length > 0
})
</script>
安全策略
1. 前后端双重验证
前端路由守卫(第一层防护)
↓
后端 API 权限校验(第二层防护)
↓
PostgREST 数据权限(第三层防护)
2. Token 防篡改
- JWT 签名验证
- Token 有效期控制(24小时)
- Token 刷新机制
3. 路由访问审计
记录所有路由访问日志:
# 在路由守卫中记录审计日志
await AuditLogger.log_route_access(
user_id=user_id,
route_path=route_path,
is_allowed=has_access,
ip_address=request.client.host
)
4. 最小权限原则
- 默认拒绝所有访问
- 显式授权才能访问
- 未配置的路由一律返回 403
性能优化
1. 缓存策略
| 缓存层级 | 存储位置 | TTL | 说明 |
|---|---|---|---|
| 用户路由缓存 | Redis | 30分钟 | 用户可访问路由列表 |
| 角色路由缓存 | Redis | 30分钟 | 角色可访问路由列表 |
| 前端路由缓存 | LocalStorage | 登录期间 | 前端动态路由数据 |
2. 懒加载
// 路由组件按需加载
component: () => import('@/views/system/Users.vue')
3. 数据库优化
- 索引优化:
parent_id、role_id、route_id - 查询优化:使用 JOIN 减少查询次数
- 连接池:复用数据库连接
实施步骤
阶段一:数据库准备(第1天)
- ✅ 检查
sys_routes和role_route表结构 - ✅ 添加缺失字段(如
icon、is_hidden等) - ✅ 创建索引
- ✅ 编写初始化数据 SQL(系统路由 + 角色-路由关联)
阶段二:后端实现(第2天)
- ✅ 实现
RoutePermission模块 - ✅ 实现
/rbac/user/routes接口 - ✅ 实现
/rbac/roles/{role_id}/routes接口 - ✅ 实现
/rbac/check-route接口 - ✅ 集成到现有认证流程
阶段三:前端对接文档(第2天)
- ✅ 编写接口文档
- ✅ 编写前端集成示例(Vue3)
- ✅ 编写路由守卫示例
- ✅ 编写动态菜单示例
阶段四:测试验证(第3天)
- ✅ 编写单元测试
- ✅ 编写集成测试
- ✅ 性能测试(缓存命中率、响应时间)
- ✅ 安全测试(未授权访问、Token 篡改)
测试方案
1. 单元测试
# tests/test_route_permission.py
async def test_get_user_routes():
"""测试获取用户路由"""
routes = await RoutePermission.get_user_routes(user_id=5)
assert len(routes) > 0
assert routes[0]['route_path'] == '/dashboard'
async def test_check_route_access_allowed():
"""测试路由访问权限(有权限)"""
has_access = await RoutePermission.check_route_access(
user_id=5,
route_path='/system/users'
)
assert has_access == True
async def test_check_route_access_denied():
"""测试路由访问权限(无权限)"""
has_access = await RoutePermission.check_route_access(
user_id=9,
route_path='/system/roles'
)
assert has_access == False
2. 集成测试
# tests/test_rbac_routes_api.py
async def test_get_current_user_routes_api():
"""测试获取当前用户路由接口"""
response = await client.get(
'/rbac/user/routes',
headers={'Authorization': f'Bearer {token}'}
)
assert response.status_code == 200
data = response.json()
assert 'routes' in data
assert len(data['routes']) > 0
3. 性能测试
- 缓存命中率 > 95%
- 接口响应时间 < 100ms
- 并发 1000 QPS 无压力
风险评估与应对
风险1:Redis 不可用
应对:降级策略,直接查询数据库
风险2:路由数据量大(>1000条)
应对:
- 分页加载
- 按需加载子菜单
- 前端虚拟滚动
风险3:缓存一致性
应对:
- 更新路由时清除相关缓存
- 设置合理的 TTL(30分钟)
- 提供手动刷新缓存接口
文档版本: v1.0 最后更新: 2025-11-17 维护者: DocAuditAI Team