fix(F008): migrate KnowledgeDao + FlowDao permission filtering to ReBAC

F008 在 v2.5.0 把 service 层迁到 ReBAC，但漏掉了 2 个 DAO 层的权限过滤路径（spec 措辞只点了 judge_role_access，遗漏 find_role_access 系列 + get_role_access 列表型）。这两处的实际后果是：F007 授权 UI 写入的 ReBAC 元组对它们无效，管理员配了等于没配；多租户场景下 Root 通过 FGA shared_with 共享给 Child 的资源也走不通这条消费路径。 KnowledgeDao.judge_knowledge_permission / ajudge_knowledge_permission （被客户端日常模式 orgKbs 检索 + 工作流通用 KB 节点 user_auth=True 消费）和 FlowDao.get_user_access_online_flows（被 Assistant.get_auto_flow_info 自动选技能消费）现在都改走 LoginUser.init_login_user[_sync] + PermissionService .list_accessible_ids('can_read', ...)，admin 短路语义保留，owner / 部门管理员 / 租户管理员的隐式范围由 _finalize_accessible_ids 自动覆盖。防回归： - features/v2.5.1/release-contract.md 表 2 新增 INV-T19（DAO/Model 层禁止直读 RoleAccessDao 做权限过滤），列出 4 类例外白名单 - scripts/arch-guard.sh 加 RULE-8（PostToolUse per-file 钩子） - scripts/check-rbac-rebac-leak.sh 新增（仓库级 one-shot 守卫，CI/pre-commit 用，当前 exit 0 clean） - features/v2.5.0/008-resource-rebac-adaptation/tasks.md 末尾追加 T12 补丁记录测试：13 用例全绿（test_knowledge_dao_judge_permission_rebac.py 10 + test_flow_dao_user_access_rebac.py 3），覆盖 admin 全集 / 普通用户交集 / 无交集 / 空入参 / 用户不存在 5 类边界。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-08-29 01:22:31 +08:00 · 2026-05-07 23:00:54 +08:00
parent 07c41a2ec1
commit ef5c3d11b4
8 changed files with 585 additions and 74 deletions
@@ -12,6 +12,7 @@
 | spec.md | ✅ 已评审 | 用户确认通过 |
 | tasks.md | ✅ 已拆解 | 11 个任务，审查通过 |
 | 实现 | ✅ 已完成 | 10 / 11 完成，T09 N/A |
+| follow-up | ✅ 2026-05-07 补丁 | T12 修复 DAO 层 RBAC 漏网（`KnowledgeDao` + `FlowDao`），登记 INV-T19 + 双层守卫 |

 ---

@@ -531,3 +532,67 @@ T11(前端group_ids移除) ← T05,T06 完成后
 > 完成后，在此记录实现与 spec.md 的偏差，供后续参考。

 - **偏差 1**: _（实现时填写）_
+
+---
+
+## T12: DAO 层 RBAC 漏网修复（v2.5.1 follow-up） ✅
+
+> **状态**: ✅ 已完成（2026-05-07，作为 v2.5.1 hotfix）
+> **背景**: T03 / T05 重点改造 service 层（`KnowledgePermissionService` / `knowledge_service.py` / `knowledge_space_service.py`），但 spec.md "Replace all `RoleAccessDao.judge_role_access` 调用" 措辞不严，遗漏了同表的 `RoleAccessDao.find_role_access` / `afind_role_access` 系列以及 `RoleAccessDao.get_role_access(*, AccessType.X)` 列表型调用。漏网点位于 **DAO 层**（`/database/models/` 与 `/domain/models/`），任务列表是按 service 层组织的，跨层覆盖的盲点直到 v2.5.1 才被发现。
+
+### 漏网点
+
+| 文件 | 方法 | 消费场景 | 状态 |
+|------|------|---------|------|
+| `bisheng/knowledge/domain/models/knowledge.py:373` | `judge_knowledge_permission(user_name, knowledge_ids) -> List[Knowledge]` | 客户端日常模式 `orgKbs` 检索（`workstation_service.queryChunksFromDB`）+ 工作流通用 KB 节点（`user_auth=True` 时） | ✅ 迁移完毕 |
+| `bisheng/knowledge/domain/models/knowledge.py:412` | `ajudge_knowledge_permission(...)` 异步版 | 同上 | ✅ 迁移完毕 |
+| `bisheng/database/models/flow.py:343` | `get_user_access_online_flows(user_id, ...) -> List[Flow]` | `Assistant.get_auto_flow_info` 自动选技能 | ✅ 迁移完毕 |
+
+### 改造模板
+
+3 处统一模板（与 T03 service 层迁移同构）：
+
+1. 构造 `LoginUser.init_login_user[_sync](user_id, user_name)` —— 拿到 `is_admin` + `is_global_super`
+2. 调 `PermissionService.list_accessible_ids(relation='can_read', object_type=<X>, login_user=...)`
+3. `None` → admin 短路返回全集；否则取交集后 `aget_list_by_ids(filtered)`
+4. 清理文件级 `RoleAccessDao` / `UserRoleDao` / `AccessType` 冗余 import
+
+### 产出文件
+
+| 操作 | 文件 |
+|------|------|
+| 修改 | `src/backend/bisheng/knowledge/domain/models/knowledge.py` |
+| 修改 | `src/backend/bisheng/database/models/flow.py` |
+| 新增 | `src/backend/test/test_knowledge_dao_judge_permission_rebac.py`（10 用例）|
+| 新增 | `src/backend/test/test_flow_dao_user_access_rebac.py`（3 用例）|
+| 修改 | `features/v2.5.1/release-contract.md`（新增 INV-T19 + 变更历史条目）|
+| 修改 | `scripts/arch-guard.sh`（新增 RULE-8，per-file PostToolUse 钩子守卫）|
+| 新增 | `scripts/check-rbac-rebac-leak.sh`（仓库级 one-shot 守卫，CI/pre-commit 用）|
+
+### 单测覆盖
+
+13 用例全绿（`pytest test/test_knowledge_dao_judge_permission_rebac.py test/test_flow_dao_user_access_rebac.py`）：
+- admin 短路返全集
+- 普通用户 `accessible_ids ∩ 入参` 交集
+- 无交集返 `[]`，且不触发 DB 查询
+- 空 `knowledge_ids` 入参短路
+- `UserDao` 返回 None 短路
+- `flow_id_extra='admin'` 魔术 sentinel 在 admin 路径保留
+
+### 实际影响
+
+- **客户端日常模式 `orgKbs` 检索**：Child 用户对 Root 配置的 KB，若 Root 通过 ReBAC `shared_with: [tenant]` 共享，现在能拿到内容；之前 RBAC 路径无视 FGA 元组直接拒
+- **工作流通用 KB 节点（`user_auth=True`）/ `Assistant.get_auto_flow_info`**：同上
+- **F007 资源授权 UI**：管理员通过新版 UI 写入的 ReBAC 元组现在生效（之前写了等于没写）
+- **管理员短路语义**：保持一致（`is_admin()` 仍走 `user_role`）
+
+### 防回归
+
+- **per-file 钩子**：`scripts/arch-guard.sh` RULE-8 在 PostToolUse 时扫描 `RoleAccessDao\.(find|judge|afind|ajudge)_role_access\(|RoleAccessDao\.get_role_access\(`，命中非白名单文件即输出 VIOLATION
+- **仓库级守卫**：`scripts/check-rbac-rebac-leak.sh` 全仓扫描 + 4 类例外白名单（auth.py / `/role_access` CRUD / WEB_MENU / `permission/migration/`），CI/pre-commit 接入；当前 exit 0 clean
+- **不变量登记**：v2.5.1 release-contract.md 表 2 新增 **INV-T19**（"DAO/Model 层禁止直读 RoleAccessDao 做权限过滤"）
+
+### 备忘
+
+- F008 spec.md "Replace all `RoleAccessDao.judge_role_access` 调用" 这条措辞应升级为"Replace all `RoleAccessDao.{find,judge,afind,ajudge}_role_access` 与 `RoleAccessDao.get_role_access(*, AccessType.X)` 调用"，未来若再有同类 spec 拆任务时按此完整列表写入
+- 任务列表组织方式从纯 service 层升级为"按调用层 + 数据流"双视角拆，避免只看 service 漏掉 DAO 层（即使 DAO 不应做权限过滤，历史代码就是这么写的，迁移时必须显式覆盖）
@@ -68,6 +68,7 @@
 | INV-T16 | 单→多租户升级时所有存量 `llm_server/llm_model.tenant_id=1`（Root），随 `share_default_to_children=1` 默认值自动对新挂 Child 可见；知识库/工作流/助手对 model_id 外键不迁移（`knowledge.model_id`、`workflow.model_id`、`assistant.model_id` 等保留原值）；挂载 Child 弹窗提供"不自动分发"选项缓解敏感模型泄露风险 | LLM Server / Model, Tenant | F020 |
 | INV-T17 | 系统级 LLM 默认配置（5 类 key：`knowledge_llm` / `assistant_llm` / `evaluation_llm` / `workflow_llm` / `linsight_llm`）按租户隔离存于 `tenant_system_model_config` 表；写入 `target_tenant_id` 必须 ∈ caller's `manageable_tenant_ids`，跨 tenant 写入返 403 + 19803；自然路径下 `target_tenant_id = get_current_tenant_id()`（admin-scope 已处理）；防御层在 router 重复校验防绕过 | TenantSystemModelConfig | F022 |
 | INV-T18 | 系统级 LLM 默认配置消费侧 tenant 解析规则：**资源场景** `LLMService.get_*_llm(tenant_id=...)` 必须按 owner（Knowledge.tenant_id / Flow.tenant_id / Evaluation.tenant_id / KnowledgeFile.tenant_id）显式传入；**创建/实时场景** 资源未落库时省略 `tenant_id` 由 `_resolve_tenant_id` 走 ContextVar → Root；**Celery worker** 必须从 task payload 显式带 `tenant_id`，漏传时 fallback ROOT + warn log（监控指标 `llm_system_config_tenant_missing_total`） | LLMService, Workflow, Celery payloads | F022 |
+| INV-T19 | **资源消费侧权限过滤必须经 PermissionService（ReBAC）委托**，DAO/Model 层禁止直接调 `RoleAccessDao.{find,judge,afind,ajudge}_role_access` 或 `RoleAccessDao.get_role_access(*, AccessType.X)` 完成"用户可见 ID 列表/单资源准入"判定（适用 KNOWLEDGE / WORKFLOW / ASSISTANT_READ / GPTS_TOOL_READ / DASHBOARD 等已在 `_ACCESS_TYPE_TO_REBAC` 映射的 5 对 10 类 AccessType）；**例外白名单**：① `LoginUser.access_check / aget_user_access_resource_ids` 内对未映射 AccessType 的 legacy fallback；② `WEB_MENU` 类 RBAC（菜单权限，F008 显式排除）；③ Admin UI 的 `/role_access/*` CRUD 端点（双写 + FGA 同步）；④ `bisheng/permission/migration/` 目录内的 F006 数据迁移工具。CI grep 守卫扫描 `RoleAccessDao\.\w*role_access\(` 发现新增非白名单调用即失败 | KnowledgeDao / FlowDao / 其他资源 DAO | F008 + F022 follow-up |

 **规则**：与 v2.5.0 一致——新增 INV 先在此表追加，再写 AC；修改 INV 必须列出 Impacted Specs 清单。

@@ -170,3 +171,4 @@ class Settings(BaseSettings):
 | 2026-04-19（开发前最终排查） | **3 方向并行深度排查** —— BLOCKER 4 项 + HIGH 6 项全部修复：① **F011 §5.4 audit_log 表与 v2.5.0 冲突解决**：改 `CREATE TABLE` 为 `ALTER TABLE` 补字段（tenant_id / operator_tenant_id / action / target_type / target_id / reason / metadata），保留 v2.5.0 既有字段（system_id / event_type / object_type / object_id）兼容旧代码；明确新旧字段查询互不干扰 + `operator_tenant_id` 填值规则（Child Admin=leaf / 超管无 scope=1 / 超管有 scope=X / 系统=1）；② **F011 §5.4.2 action 清单补 7 行**（`admin.scope_switch` + `llm.server.{create,update,delete}` + `llm.model.{create,update,delete}`）落实 INV-T7；③ **F011 §5.4.3 新增 TenantDao 扩展方法定义**（`aget_children_ids_active` / `aget_non_active_ids` / `aexists`）供 F016/F019 调用；④ **F013 §5 DSL 补 `llm_server` / `llm_model` 资源类型**，解决 F020 写 `{llm_server}#viewer → tenant#shared_to#member` 元组的 "unknown object type" 阻塞；⑤ **F012 §5.4 新增 ContextVar 扩展**（`visible_tenant_ids` / `_strict_tenant_filter` / `_admin_scope_tenant_id` / `_is_management_api`）+ `strict_tenant_filter()` context manager + `get_current_tenant_id()` 优先级规则（admin_scope > JWT leaf），v2.5.0 既有函数签名保留；⑥ **F012 §5.5 新增 Middleware 注册顺序文档**（Auth → TenantContext → AdminScope → 业务路由）；⑦ **F019/F020 前端组件归属明确**：`AdminScopeSelector.tsx` 归 F020 拥有，F019 仅提供后端 API + `useAdminScope` hook；⑧ **F019 §5.4 Celery 任务加 `bypass_tenant_filter()` 包裹** 防 Celery 上下文下 ContextVar=None 导致的隔离行为未定义；⑨ **F020 §5.5 DDL 迁移补前置重名校验**（`UNIQUE(name)` → `UNIQUE(tenant_id, name)` 前先 SELECT GROUP BY 查重，存在冲突则中止迁移）；⑩ **release-contract 新增"配置类清单"章节**（MultiTenantConf / UserTenantSyncConf / SSOSyncConf / LLMConf / OpenFGAConf 扩展），明确 Pydantic Settings 模型归属，防止 `settings.xxx.yyy` 运行时 AttributeError | 11 处修改分布在 F011 / F012 / F013 / F019 / F020 / release-contract；确认 v2.5.0 `current_tenant_id` / `bypass_tenant_filter` / `Tenant.status=String(16)` / 旧 audit_log 表结构与 v2.5.1 的兼容路径；**开发可以开始** |

 | 2026-05-07 | **F024-tenant-user-mgmt-ui-realign**（新增）：把租户用户管理 UI 从 v2.5.0 F010 的"用户多租户成员"模型对齐到 F012 派生模型。① **修订 F010**：AC-3.1/3.2/3.3（添加/移除用户）废止；AC-7.x 列表改为按"主部门挂在该租户子树"派生；② **后端**：`POST /tenants/{id}/users` / `DELETE /tenants/{id}/users/{user_id}` 改 410 Gone（mirror F011 switch-tenant 模式，handler 不带 auth dep）；新 DAO `UserDepartmentDao.aget_users_by_tenant_subtree`（User → UserDepartment is_primary=1 → Department.path LIKE `<root_dept_path>%`，UserTenant 仅 LEFT JOIN 取 last_access_time，旧 `aadd_users` 写的幽灵行天然不展示）；`TenantService.aadd_users` / `aremove_user` 加 `@deprecated` + logger.warning，保留实现给内部脚本；fallback：tenant 无 root_dept_id 时退到 `Department.tenant_id == tenant_id`；③ **前端**：`TenantUserDialog.tsx` 删添加/移除按钮 + 加跳转 `/sys` Banner；`addTenantUsersApi`/`removeTenantUserApi` 标 `@deprecated` JSDoc；i18n key `tenant.membershipBanner.{title,body,cta}` 三语齐全；④ **测试**：T001 `test_tenant_users_query_source.py`（aiosqlite 集成，7 用例覆盖 AC-01~04、AC-12 幽灵行、LEFT JOIN、keyword、分页、root_dept_id fallback）；T004 `test_tenant_membership_endpoints_deprecated.py`（8 用例覆盖 410 + dependency tree 校验）；后端 55 用例全过；⑤ **零 schema 变更**（AD-03 决策 C：不打 status=legacy 标签，幽灵行靠新数据源天然隔离），回滚零数据风险；⑥ 上线节奏：单 PR 单版本一次到位；release-notes 标 BREAKING + migration `apply-edit` 端点。详见 [024-tenant-user-mgmt-ui-realign/spec.md](024-tenant-user-mgmt-ui-realign/spec.md) | Tenant 用户管理 API + Platform 前端 TenantUserDialog；不触碰 F011/F012/F013 核心模型 |
+| 2026-05-07 | **F008 follow-up: KnowledgeDao + FlowDao DAO 层 RBAC 漏网修复**（hotfix）：F008 把 service 层迁到 ReBAC 但漏掉了 2 个 DAO 层的"用户可见 ID 列表/单资源准入"过滤路径——`KnowledgeDao.judge_knowledge_permission / ajudge_knowledge_permission`（被日常模式 `orgKbs` 检索 + 工作流 QA 检索 / 通用 KB 节点消费）和 `FlowDao.get_user_access_online_flows`（被 `Assistant.get_auto_flow_info` 自动选技能消费）仍直读 `RoleAccessDao.find_role_access` + `role_id == 1` 硬编码 admin。两处统一迁移：构造 `LoginUser.init_login_user[_sync]` → 调 `PermissionService.list_accessible_ids('can_read', ...)` → 取交集（KB）或透传（Flow）。同时清理两文件 `RoleAccessDao` / `UserRoleDao` / `AccessType` 冗余 import；新增 13 条单测（`test_knowledge_dao_judge_permission_rebac.py` 10 + `test_flow_dao_user_access_rebac.py` 3）覆盖 admin 全集 / 普通用户交集 / 无交集 / 空入参 / 用户不存在 5 类边界。**新增 INV-T19**：DAO/Model 层禁止直调 `RoleAccessDao.{find,judge}_role_access`（适用 5 对 10 类已映射 AccessType）；4 类例外白名单（auth.py legacy fallback / WEB_MENU / `/role_access` CRUD / F006 迁移工具）；CI grep 守卫扫描非白名单调用即失败 | KnowledgeDao + FlowDao；新增 INV-T19；CI 守卫脚本 |
@@ -11,6 +11,7 @@
 #   RULE-5: API 层不跨模块互相导入（VIOLATION）
 #   RULE-6: 前端 store 不直接调 HTTP（WARNING）
 #   RULE-7: 硬编码敏感信息检测（WARNING）
+#   RULE-8: DAO/Model 层不得直读 RoleAccessDao 做权限过滤（INV-T19，VIOLATION）

 FILE="$1"
 [ -z "$FILE" ] && exit 0
@@ -97,4 +98,24 @@ if echo "$FILE" | grep -qE "\.(py|ts|tsx|js|json)$"; then
    fi
 fi

+# ── RULE-8：DAO/Model 层不得直读 RoleAccessDao 做权限过滤（INV-T19）──
+# F008 已把 5 对 10 类资源 AccessType 迁到 ReBAC（PermissionService）。
+# 资源消费侧若想"按用户过滤可见 ID"，必须经 PermissionService 委托。
+# 直接调 RoleAccessDao.{find,judge,afind,ajudge}_role_access 或
+# RoleAccessDao.get_role_access(*, AccessType.X) 视为漏网。
+#
+# 例外白名单（允许文件路径）：
+#   - bisheng/user/domain/services/auth.py     LoginUser legacy fallback
+#   - bisheng/user/api/user.py                 /role_access CRUD 端点
+#   - bisheng/role/domain/services/role_service.py  WEB_MENU 读取
+#   - bisheng/permission/migration/            F006 迁移工具
+#   - bisheng/database/models/role_access.py   DAO 自身定义
+if echo "$FILE" | grep -q "/bisheng/" && echo "$FILE" | grep -q "\.py$"; then
+    if ! echo "$FILE" | grep -qE "/(user/domain/services/auth\.py|user/api/user\.py|role/domain/services/role_service\.py|permission/migration/|database/models/role_access\.py)$"; then
+        if grep -qE "RoleAccessDao\.(find|judge|afind|ajudge)_role_access\(|RoleAccessDao\.get_role_access\(" "$FILE" 2>/dev/null; then
+            echo "⚠️  [arch-guard] RULE-8 VIOLATION: $(basename "$FILE") — DAO/Model 层禁止直读 RoleAccessDao 做权限过滤（INV-T19，请改走 PermissionService.list_accessible_ids/check）"
+        fi
+    fi
+fi
+
 exit 0
@@ -0,0 +1,78 @@
+#!/usr/bin/env bash
+# check-rbac-rebac-leak.sh — INV-T19 全仓守卫
+#
+# 用途：CI 步骤或 pre-commit 钩子，扫描整个后端代码库，
+# 检测 DAO/Model 层是否存在直读 `RoleAccessDao.{find|judge|afind|ajudge}_role_access`
+# 或 `RoleAccessDao.get_role_access(...)` 做权限过滤的漏网。
+#
+# arch-guard.sh 是 per-file PostToolUse 钩子（开发期同步检查）；
+# 本脚本是仓库级 one-shot 扫描，覆盖未被编辑过的存量代码。
+#
+# 退出码：
+#   0 — 无违规
+#   1 — 发现违规（CI 应失败）
+#
+# 例外白名单参考 release-contract.md INV-T19。
+
+set -uo pipefail
+
+# 仓库根（脚本一般在 scripts/ 下）
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+BACKEND="${ROOT}/src/backend/bisheng"
+
+if [ ! -d "${BACKEND}" ]; then
+    echo "ERROR: backend dir not found: ${BACKEND}" >&2
+    exit 2
+fi
+
+# 违规命中模式：
+#   1. RoleAccessDao.find_role_access(  / judge_role_access(  / afind_role_access(  / ajudge_role_access(
+#   2. RoleAccessDao.get_role_access(  （列表型，被 flow.py 历史漏网用过）
+PATTERN='RoleAccessDao\.(find|judge|afind|ajudge)_role_access\(|RoleAccessDao\.get_role_access\('
+
+# 例外白名单（允许文件路径片段）。任何匹配以下任一片段的文件都被豁免。
+ALLOWLIST=(
+    'bisheng/user/domain/services/auth.py'
+    'bisheng/user/api/user.py'
+    'bisheng/role/domain/services/role_service.py'
+    'bisheng/permission/migration/'
+    'bisheng/database/models/role_access.py'
+)
+
+# 收集所有命中行（path:line:content 形式）
+hits="$(grep -rnE "${PATTERN}" "${BACKEND}" --include='*.py' 2>/dev/null || true)"
+
+if [ -z "${hits}" ]; then
+    echo "[check-rbac-rebac-leak] OK — no RoleAccessDao consumer-side calls found."
+    exit 0
+fi
+
+# 过滤白名单
+violations=""
+while IFS= read -r line; do
+    [ -z "${line}" ] && continue
+    file="${line%%:*}"
+    skip=0
+    for allow in "${ALLOWLIST[@]}"; do
+        if echo "${file}" | grep -q "${allow}"; then
+            skip=1
+            break
+        fi
+    done
+    if [ "${skip}" -eq 0 ]; then
+        violations="${violations}${line}"$'\n'
+    fi
+done <<< "${hits}"
+
+if [ -z "${violations}" ]; then
+    echo "[check-rbac-rebac-leak] OK — all RoleAccessDao calls are within the INV-T19 allowlist."
+    exit 0
+fi
+
+echo "❌ [check-rbac-rebac-leak] INV-T19 VIOLATION — DAO/Model 层禁止直读 RoleAccessDao 做权限过滤"
+echo ""
+echo "请改走 PermissionService.list_accessible_ids / PermissionService.check（参考 KnowledgeDao.ajudge_knowledge_permission / FlowDao.get_user_access_online_flows 迁移模板）。"
+echo ""
+echo "命中行："
+printf "%s" "${violations}" | sed 's/^/  /'
+exit 1
@@ -15,8 +15,7 @@ from bisheng.common.services import telemetry_service
 from bisheng.core.database import get_sync_db_session, get_async_db_session
 from bisheng.core.logger import trace_id_var
 from bisheng.database.models.assistant import Assistant
-from bisheng.database.models.role_access import AccessType, RoleAccess, RoleAccessDao
-from bisheng.user.domain.models.user_role import UserRoleDao
+from bisheng.database.models.role_access import AccessType, RoleAccess
 from bisheng.utils import generate_uuid


@@ -347,17 +346,34 @@ class FlowDao(FlowBase):
                                     keyword: str = None,
                                     flow_ids: List[str] = None,
                                     flow_type: int = FlowType.WORKFLOW.value) -> List[Flow]:
-        user_role = UserRoleDao.get_user_roles(user_id)
-        flow_id_extra = []
-        if user_role:
-            role_ids = [role.role_id for role in user_role]
-            if 1 in role_ids:
-                # admin
-                flow_id_extra = 'admin'
-            else:
-                role_access = RoleAccessDao.get_role_access(role_ids, AccessType.WORKFLOW)
-                if role_access:
-                    flow_id_extra = [access.third_id for access in role_access]
+        """List online flows the user can read.
+
+        F008 follow-up: delegates to ReBAC via PermissionService instead of
+        the legacy role_access table. Admin still receives the magic
+        ``flow_id_extra='admin'`` sentinel that disables ID filtering inside
+        FlowDao.get_flows. Owners + dept-admin scope are picked up by
+        list_accessible_ids' implicit scope expansion.
+        """
+        from bisheng.permission.domain.services.owner_service import _run_async_safe
+        from bisheng.permission.domain.services.permission_service import PermissionService
+        from bisheng.user.domain.services.auth import LoginUser
+
+        login_user = LoginUser.init_login_user_sync(
+            user_id=user_id,
+            user_name='',
+        )
+        accessible_ids = _run_async_safe(
+            PermissionService.list_accessible_ids(
+                user_id=user_id,
+                relation='can_read',
+                object_type='workflow',
+                login_user=login_user,
+            ),
+        )
+        if accessible_ids is None:
+            flow_id_extra = 'admin'
+        else:
+            flow_id_extra = list(accessible_ids)
        return FlowDao.get_flows(user_id,
                                 flow_id_extra,
                                 keyword,
@@ -9,10 +9,8 @@ from sqlmodel.sql.expression import Select, SelectOfScalar, col

 from bisheng.common.models.base import SQLModelSerializable
 from bisheng.core.database import get_sync_db_session, get_async_db_session
-from bisheng.database.models.role_access import AccessType, RoleAccessDao
 from bisheng.knowledge.domain.models.knowledge_file import KnowledgeFile, KnowledgeFileDao
 from bisheng.user.domain.models.user import UserDao
-from bisheng.user.domain.models.user_role import UserRoleDao


 class KnowledgeTypeEnum(Enum):
@@ -374,80 +372,71 @@ class KnowledgeDao(KnowledgeBase):
    @classmethod
    def judge_knowledge_permission(cls, user_name: str,
                                   knowledge_ids: List[int]) -> List[Knowledge]:
+        """Filter knowledge_ids to those the user can read.
+
+        F008 follow-up: delegates to ReBAC via PermissionService instead of
+        the legacy role_access table. Admin still gets the full set; owners
+        and tenant-admin scope are picked up by list_accessible_ids' implicit
+        scope expansion.
        """
-        Based on username and knowledge baseIDList to get a list of knowledge bases that the user has permission to view
-        :param user_name: Username
-        :param knowledge_ids: The knowledge base uponIDVertical
-        :return: Returns a list of knowledge bases that the user has permissions
-        """
-        # get user info
+        if not knowledge_ids:
+            return []
        user_info = UserDao.get_user_by_username(user_name)
        if not user_info:
            return []

-        # Query the role the user belongs to
-        role_list = UserRoleDao.get_user_roles(user_info.user_id)
-        if not role_list:
-            return []
+        from bisheng.permission.domain.services.owner_service import _run_async_safe
+        from bisheng.permission.domain.services.permission_service import PermissionService
+        from bisheng.user.domain.services.auth import LoginUser

-        role_id_list = []
-        is_admin = False
-        for role in role_list:
-            role_id_list.append(role.role_id)
-            if role.role_id == 1:
-                is_admin = True
-        # admin User has all knowledge base permissions
-        if is_admin:
-            return KnowledgeDao.get_list_by_ids(knowledge_ids)
+        login_user = LoginUser.init_login_user_sync(
+            user_id=user_info.user_id,
+            user_name=user_name,
+        )
+        accessible_ids = _run_async_safe(
+            PermissionService.list_accessible_ids(
+                user_id=login_user.user_id,
+                relation='can_read',
+                object_type='knowledge_library',
+                login_user=login_user,
+            ),
+        )
+        if accessible_ids is None:
+            return cls.get_list_by_ids(knowledge_ids)

-        # query role List of knowledge bases with permissions
-        role_access_list = RoleAccessDao.find_role_access(role_id_list, [str(one) for one in knowledge_ids],
-                                                          AccessType.KNOWLEDGE)
-
-        user_knowledge_list = cls.get_user_knowledge(user_info.user_id,
-                                                     knowledge_id_extra=[int(access.third_id) for access in
-                                                                         role_access_list],
-                                                     filter_knowledge=knowledge_ids)
-        return user_knowledge_list
+        accessible_set = {int(x) for x in accessible_ids}
+        filtered = [kid for kid in knowledge_ids if int(kid) in accessible_set]
+        return cls.get_list_by_ids(filtered) if filtered else []

    @classmethod
    async def ajudge_knowledge_permission(cls, user_name: str,
                                          knowledge_ids: List[int]) -> List[Knowledge]:
-        """
-        By Username and Knowledge BaseIDlist, asynchronously get a list of knowledge bases that the user has permission to view
-        Args:
-            user_name:
-            knowledge_ids:
-
-        Returns:
-
-        """
-        # get user info
+        """Async variant of :meth:`judge_knowledge_permission`. Same semantics."""
+        if not knowledge_ids:
+            return []
        user_info = await UserDao.aget_user_by_username(user_name)
        if not user_info:
            return []
-        # Query the role the user belongs to
-        role_list = await UserRoleDao.aget_user_roles(user_info.user_id)
-        if not role_list:
-            return []
-        role_id_list = []
-        is_admin = False
-        for role in role_list:
-            role_id_list.append(role.role_id)
-            if role.role_id == 1:
-                is_admin = True
-        # admin User has all knowledge base permissions
-        if is_admin:
+
+        from bisheng.permission.domain.services.permission_service import PermissionService
+        from bisheng.user.domain.services.auth import LoginUser
+
+        login_user = await LoginUser.init_login_user(
+            user_id=user_info.user_id,
+            user_name=user_name,
+        )
+        accessible_ids = await PermissionService.list_accessible_ids(
+            user_id=login_user.user_id,
+            relation='can_read',
+            object_type='knowledge_library',
+            login_user=login_user,
+        )
+        if accessible_ids is None:
            return await cls.aget_list_by_ids(knowledge_ids)
-        # query role List of knowledge bases with permissions
-        role_access_list = await RoleAccessDao.afind_role_access(role_id_list, [str(one) for one in knowledge_ids],
-                                                                 AccessType.KNOWLEDGE)
-        # Query whether the knowledge base created by the user is included
-        user_knowledge_list = await cls.aget_user_knowledge(user_info.user_id,
-                                                            knowledge_id_extra=[int(access.third_id) for access in
-                                                                                role_access_list],
-                                                            filter_knowledge=knowledge_ids)
-        return user_knowledge_list
+
+        accessible_set = {int(x) for x in accessible_ids}
+        filtered = [kid for kid in knowledge_ids if int(kid) in accessible_set]
+        return await cls.aget_list_by_ids(filtered) if filtered else []

    @classmethod
    def filter_knowledge_by_ids(cls,
@@ -0,0 +1,96 @@
+"""Tests for FlowDao.get_user_access_online_flows after F008 follow-up
+migration to ReBAC.
+
+Verifies:
+  * Admin user (list_accessible_ids returns None) preserves the magic
+    sentinel flow_id_extra='admin'.
+  * Non-admin user passes the FGA-resolved id list to FlowDao.get_flows.
+  * No-access user passes [] to FlowDao.get_flows.
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from bisheng.database.models.flow import FlowDao
+
+
+def _patch_login_user_sync(monkeypatch, user_id: int = 7):
+    fake_login_user = SimpleNamespace(user_id=user_id, user_name='')
+    from bisheng.user.domain.services import auth as auth_mod
+
+    monkeypatch.setattr(
+        auth_mod.LoginUser,
+        'init_login_user_sync',
+        MagicMock(return_value=fake_login_user),
+    )
+    return fake_login_user
+
+
+def _patch_run_async_safe(monkeypatch, return_value):
+    monkeypatch.setattr(
+        'bisheng.permission.domain.services.owner_service._run_async_safe',
+        lambda coro, *args, **kwargs: (asyncio.new_event_loop().run_until_complete(coro), return_value)[1]
+        if False else _consume_and_return(coro, return_value),
+    )
+
+
+def _consume_and_return(coro, value):
+    coro.close()
+    return value
+
+
+def test_admin_passes_magic_admin_sentinel(monkeypatch):
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch, return_value=None)
+
+    captured = {}
+
+    def fake_get_flows(user_id, flow_id_extra, *args, **kwargs):
+        captured['user_id'] = user_id
+        captured['flow_id_extra'] = flow_id_extra
+        return ['flow-row']
+
+    monkeypatch.setattr(FlowDao, 'get_flows', fake_get_flows)
+
+    out = FlowDao.get_user_access_online_flows(user_id=7, page=1, limit=50)
+
+    assert out == ['flow-row']
+    assert captured['user_id'] == 7
+    assert captured['flow_id_extra'] == 'admin'
+
+
+def test_non_admin_passes_accessible_ids_list(monkeypatch):
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch, return_value=['11', '22', '33'])
+
+    captured = {}
+
+    def fake_get_flows(user_id, flow_id_extra, *args, **kwargs):
+        captured['flow_id_extra'] = flow_id_extra
+        return []
+
+    monkeypatch.setattr(FlowDao, 'get_flows', fake_get_flows)
+
+    FlowDao.get_user_access_online_flows(user_id=7)
+
+    assert captured['flow_id_extra'] == ['11', '22', '33']
+
+
+def test_non_admin_no_accessible_passes_empty_list(monkeypatch):
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch, return_value=[])
+
+    captured = {}
+
+    def fake_get_flows(user_id, flow_id_extra, *args, **kwargs):
+        captured['flow_id_extra'] = flow_id_extra
+        return []
+
+    monkeypatch.setattr(FlowDao, 'get_flows', fake_get_flows)
+
+    FlowDao.get_user_access_online_flows(user_id=7)
+
+    assert captured['flow_id_extra'] == []
@@ -0,0 +1,244 @@
+"""Tests for KnowledgeDao.judge_knowledge_permission / ajudge_knowledge_permission
+after F008 follow-up migration to ReBAC.
+
+Verifies:
+  * Empty input short-circuits to [].
+  * Unknown user short-circuits to [].
+  * Admin user (list_accessible_ids returns None) gets the full set fetched
+    by id.
+  * Non-admin user gets only the intersection of input ids and the FGA-
+    accessible id set.
+  * Non-admin user with no overlap gets [].
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from bisheng.knowledge.domain.models import knowledge as knowledge_module
+from bisheng.knowledge.domain.models.knowledge import KnowledgeDao
+
+
+def _kb(id_: int):
+    return SimpleNamespace(id=id_, name=f'kb-{id_}')
+
+
+def _patch_user_lookup_async(monkeypatch, user_id: int = 42):
+    user = SimpleNamespace(user_id=user_id, user_name='alice')
+    monkeypatch.setattr(
+        knowledge_module.UserDao,
+        'aget_user_by_username',
+        AsyncMock(return_value=user),
+    )
+    return user
+
+
+def _patch_user_lookup_sync(monkeypatch, user_id: int = 42):
+    user = SimpleNamespace(user_id=user_id, user_name='alice')
+    monkeypatch.setattr(
+        knowledge_module.UserDao,
+        'get_user_by_username',
+        MagicMock(return_value=user),
+    )
+    return user
+
+
+def _patch_login_user_async(monkeypatch, user_id: int = 42):
+    fake_login_user = SimpleNamespace(user_id=user_id, user_name='alice')
+    from bisheng.user.domain.services import auth as auth_mod
+
+    monkeypatch.setattr(
+        auth_mod.LoginUser,
+        'init_login_user',
+        AsyncMock(return_value=fake_login_user),
+    )
+    return fake_login_user
+
+
+def _patch_login_user_sync(monkeypatch, user_id: int = 42):
+    fake_login_user = SimpleNamespace(user_id=user_id, user_name='alice')
+    from bisheng.user.domain.services import auth as auth_mod
+
+    monkeypatch.setattr(
+        auth_mod.LoginUser,
+        'init_login_user_sync',
+        MagicMock(return_value=fake_login_user),
+    )
+    return fake_login_user
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Async tests
+# ──────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_ajudge_empty_input_returns_empty():
+    result = await KnowledgeDao.ajudge_knowledge_permission('alice', [])
+    assert result == []
+
+
+@pytest.mark.asyncio
+async def test_ajudge_unknown_user_returns_empty(monkeypatch):
+    monkeypatch.setattr(
+        knowledge_module.UserDao,
+        'aget_user_by_username',
+        AsyncMock(return_value=None),
+    )
+    result = await KnowledgeDao.ajudge_knowledge_permission('ghost', [1, 2])
+    assert result == []
+
+
+@pytest.mark.asyncio
+async def test_ajudge_admin_gets_full_set(monkeypatch):
+    _patch_user_lookup_async(monkeypatch)
+    _patch_login_user_async(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=None),
+    )
+    aget_list_mock = AsyncMock(return_value=[_kb(1), _kb(2), _kb(3)])
+    monkeypatch.setattr(KnowledgeDao, 'aget_list_by_ids', aget_list_mock)
+
+    result = await KnowledgeDao.ajudge_knowledge_permission('alice', [1, 2, 3])
+
+    aget_list_mock.assert_awaited_once_with([1, 2, 3])
+    assert [k.id for k in result] == [1, 2, 3]
+
+
+@pytest.mark.asyncio
+async def test_ajudge_non_admin_filters_to_accessible_intersection(monkeypatch):
+    _patch_user_lookup_async(monkeypatch)
+    _patch_login_user_async(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=['1', '3', '99']),  # 99 not in input — must be discarded
+    )
+    aget_list_mock = AsyncMock(return_value=[_kb(1), _kb(3)])
+    monkeypatch.setattr(KnowledgeDao, 'aget_list_by_ids', aget_list_mock)
+
+    result = await KnowledgeDao.ajudge_knowledge_permission('alice', [1, 2, 3])
+
+    aget_list_mock.assert_awaited_once_with([1, 3])
+    assert sorted(k.id for k in result) == [1, 3]
+
+
+@pytest.mark.asyncio
+async def test_ajudge_non_admin_no_overlap_returns_empty(monkeypatch):
+    _patch_user_lookup_async(monkeypatch)
+    _patch_login_user_async(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=['77', '88']),
+    )
+    aget_list_mock = AsyncMock(return_value=[])
+    monkeypatch.setattr(KnowledgeDao, 'aget_list_by_ids', aget_list_mock)
+
+    result = await KnowledgeDao.ajudge_knowledge_permission('alice', [1, 2, 3])
+
+    aget_list_mock.assert_not_awaited()
+    assert result == []
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Sync tests
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_judge_empty_input_returns_empty():
+    assert KnowledgeDao.judge_knowledge_permission('alice', []) == []
+
+
+def test_judge_unknown_user_returns_empty(monkeypatch):
+    monkeypatch.setattr(
+        knowledge_module.UserDao,
+        'get_user_by_username',
+        MagicMock(return_value=None),
+    )
+    assert KnowledgeDao.judge_knowledge_permission('ghost', [1, 2]) == []
+
+
+def _patch_run_async_safe(monkeypatch):
+    """Replace run_async_safe with a plain asyncio.run so we can drive the
+    sync entrypoint deterministically inside pytest's main thread."""
+    monkeypatch.setattr(
+        'bisheng.permission.domain.services.owner_service._run_async_safe',
+        lambda coro, *args, **kwargs: asyncio.new_event_loop().run_until_complete(coro),
+    )
+
+
+def test_judge_admin_gets_full_set(monkeypatch):
+    _patch_user_lookup_sync(monkeypatch)
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=None),
+    )
+    get_list_mock = MagicMock(return_value=[_kb(1), _kb(2)])
+    monkeypatch.setattr(KnowledgeDao, 'get_list_by_ids', get_list_mock)
+
+    result = KnowledgeDao.judge_knowledge_permission('alice', [1, 2])
+
+    get_list_mock.assert_called_once_with([1, 2])
+    assert [k.id for k in result] == [1, 2]
+
+
+def test_judge_non_admin_filters_to_accessible_intersection(monkeypatch):
+    _patch_user_lookup_sync(monkeypatch)
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=['2']),
+    )
+    get_list_mock = MagicMock(return_value=[_kb(2)])
+    monkeypatch.setattr(KnowledgeDao, 'get_list_by_ids', get_list_mock)
+
+    result = KnowledgeDao.judge_knowledge_permission('alice', [1, 2, 3])
+
+    get_list_mock.assert_called_once_with([2])
+    assert [k.id for k in result] == [2]
+
+
+def test_judge_non_admin_no_overlap_returns_empty(monkeypatch):
+    _patch_user_lookup_sync(monkeypatch)
+    _patch_login_user_sync(monkeypatch)
+    _patch_run_async_safe(monkeypatch)
+
+    from bisheng.permission.domain.services import permission_service as ps_mod
+
+    monkeypatch.setattr(
+        ps_mod.PermissionService,
+        'list_accessible_ids',
+        AsyncMock(return_value=['7']),
+    )
+    get_list_mock = MagicMock(return_value=[])
+    monkeypatch.setattr(KnowledgeDao, 'get_list_by_ids', get_list_mock)
+
+    result = KnowledgeDao.judge_knowledge_permission('alice', [1, 2])
+
+    get_list_mock.assert_not_called()
+    assert result == []