Files
leaudit-platform-frontend/auth_doc/角色-路由权限实现方案.md
2025-11-18 11:06:24 +08:00

33 KiB
Raw Permalink Blame History

角色-路由权限实现方案

版本: 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(系统路由表)

已存在,需要确认字段:

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(角色-路由关联表)

已存在,需要确认字段:

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_idrole_idroute_id
  • 查询优化:使用 JOIN 减少查询次数
  • 连接池:复用数据库连接

实施步骤

阶段一:数据库准备(第1天)

  1. 检查 sys_routesrole_route 表结构
  2. 添加缺失字段(如 iconis_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. 单元测试

# 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 无压力

风险评估与应对

风险1Redis 不可用

应对:降级策略,直接查询数据库

风险2:路由数据量大(>1000条)

应对

  • 分页加载
  • 按需加载子菜单
  • 前端虚拟滚动

风险3:缓存一致性

应对

  • 更新路由时清除相关缓存
  • 设置合理的 TTL30分钟)
  • 提供手动刷新缓存接口

文档版本: v1.0 最后更新: 2025-11-17 维护者: DocAuditAI Team