bfe39e45a9
5. 修改统一认证登录和管理员登录是通过接口形式进行,存储返回的accessToken。 6. 修改交叉评查的部分样式
1181 lines
33 KiB
Markdown
1181 lines
33 KiB
Markdown
# 角色-路由权限实现方案
|
||
|
||
**版本**: v1.0
|
||
**日期**: 2025-11-17
|
||
**状态**: 设计完成,待实现
|
||
|
||
---
|
||
|
||
## 目录
|
||
|
||
1. [功能概述](#功能概述)
|
||
2. [设计原则](#设计原则)
|
||
3. [架构设计](#架构设计)
|
||
4. [数据库设计](#数据库设计)
|
||
5. [后端实现](#后端实现)
|
||
6. [前端实现](#前端实现)
|
||
7. [安全策略](#安全策略)
|
||
8. [性能优化](#性能优化)
|
||
9. [实施步骤](#实施步骤)
|
||
10. [测试方案](#测试方案)
|
||
|
||
---
|
||
|
||
## 功能概述
|
||
|
||
### 业务需求
|
||
|
||
实现**基于角色的前端路由权限控制**,使不同角色的用户登录后看到不同的菜单和页面:
|
||
|
||
- ✅ **系统管理员** - 可访问所有路由(用户管理、角色管理、权限管理等)
|
||
- ✅ **文档管理员** - 可访问文档管理相关路由
|
||
- ✅ **文档审查员** - 可访问文档查看和评查路由
|
||
- ✅ **普通用户** - 仅可访问基础路由(首页、个人中心)
|
||
|
||
### 核心功能
|
||
|
||
1. **路由定义管理** - 系统路由的增删改查
|
||
2. **角色-路由关联** - 配置角色可访问的路由
|
||
3. **用户路由查询** - 根据用户角色获取可访问路由列表
|
||
4. **动态菜单生成** - 前端根据路由列表动态生成菜单
|
||
5. **路由守卫** - 前端路由守卫拦截未授权访问
|
||
|
||
---
|
||
|
||
## 设计原则
|
||
|
||
### 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(系统路由表)
|
||
|
||
已存在,需要确认字段:
|
||
|
||
```sql
|
||
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/users`
|
||
- `route_name`: 路由名称(唯一标识),如 `UserManagement`
|
||
- `component`: 组件路径,如 `views/system/Users.vue`
|
||
- `parent_id`: 父路由ID,支持多级嵌套菜单
|
||
- `route_title`: 菜单显示的标题(中文)
|
||
- `icon`: 图标(Element Plus / Ant Design 图标名)
|
||
- `is_hidden`: 是否在菜单中隐藏(某些路由不显示在菜单,但可访问)
|
||
- `meta`: JSON 扩展字段,存储额外信息(如权限标识、面包屑等)
|
||
|
||
#### 2. role_route(角色-路由关联表)
|
||
|
||
已存在,需要确认字段:
|
||
|
||
```sql
|
||
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);
|
||
```
|
||
|
||
### 数据示例
|
||
|
||
#### 系统路由数据
|
||
|
||
```sql
|
||
-- 一级路由:首页
|
||
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);
|
||
```
|
||
|
||
#### 角色-路由关联数据
|
||
|
||
```sql
|
||
-- 系统管理员 - 拥有所有路由
|
||
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
|
||
|
||
```python
|
||
#!/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
|
||
|
||
```python
|
||
#!/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)
|
||
|
||
```javascript
|
||
// 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(用户状态管理)
|
||
|
||
```javascript
|
||
// 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. 动态菜单生成
|
||
|
||
```vue
|
||
<!-- 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>
|
||
```
|
||
|
||
```vue
|
||
<!-- 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. 路由访问审计
|
||
|
||
记录所有路由访问日志:
|
||
|
||
```python
|
||
# 在路由守卫中记录审计日志
|
||
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. 懒加载
|
||
|
||
```javascript
|
||
// 路由组件按需加载
|
||
component: () => import('@/views/system/Users.vue')
|
||
```
|
||
|
||
### 3. 数据库优化
|
||
|
||
- 索引优化:`parent_id`、`role_id`、`route_id`
|
||
- 查询优化:使用 JOIN 减少查询次数
|
||
- 连接池:复用数据库连接
|
||
|
||
---
|
||
|
||
## 实施步骤
|
||
|
||
### 阶段一:数据库准备(第1天)
|
||
|
||
1. ✅ 检查 `sys_routes` 和 `role_route` 表结构
|
||
2. ✅ 添加缺失字段(如 `icon`、`is_hidden` 等)
|
||
3. ✅ 创建索引
|
||
4. ✅ 编写初始化数据 SQL(系统路由 + 角色-路由关联)
|
||
|
||
### 阶段二:后端实现(第2天)
|
||
|
||
1. ✅ 实现 `RoutePermission` 模块
|
||
2. ✅ 实现 `/rbac/user/routes` 接口
|
||
3. ✅ 实现 `/rbac/roles/{role_id}/routes` 接口
|
||
4. ✅ 实现 `/rbac/check-route` 接口
|
||
5. ✅ 集成到现有认证流程
|
||
|
||
### 阶段三:前端对接文档(第2天)
|
||
|
||
1. ✅ 编写接口文档
|
||
2. ✅ 编写前端集成示例(Vue3)
|
||
3. ✅ 编写路由守卫示例
|
||
4. ✅ 编写动态菜单示例
|
||
|
||
### 阶段四:测试验证(第3天)
|
||
|
||
1. ✅ 编写单元测试
|
||
2. ✅ 编写集成测试
|
||
3. ✅ 性能测试(缓存命中率、响应时间)
|
||
4. ✅ 安全测试(未授权访问、Token 篡改)
|
||
|
||
---
|
||
|
||
## 测试方案
|
||
|
||
### 1. 单元测试
|
||
|
||
```python
|
||
# 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. 集成测试
|
||
|
||
```python
|
||
# 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
|