1 项目背景业务场景「云帆科技」的 RAGFlow 平台经过两个月的试运行已从最初的 HR 一个部门扩展到财务、法务、研发、市场共五个部门。每个部门的知识库中包含了各自的制度文档、项目资料和业务数据。随着用户增多安全问题逐渐浮出水面HR 总监发现研发部的同事竟然能在 RAGFlow 中搜索到《2024年薪资调整方案》——这份文档标记为机密只有 HR 部门经理以上才能查看。调查发现原因是所有部门的数据集没有做权限隔离所有人共享同一个管理员账号。与此同时CIO 提出公司很快要扩展到子公司需要租户级别的隔离——子公司的数据和母公司数据完全隔离就像两套独立的 RAGFlow。IT 安全部门也加入了讨论提出了 API Token 泄露的风险、默认密码未修改的问题、以及 MinIO 对象存储如果配置不当可能被公开访问的风险。痛点权限和安全管理缺失的典型事故水平越权用户 A 可以通过修改 API 请求中的dataset_id参数访问用户 B 的数据集——因为后端没有做数据集归属用户的校验。垂直越权普通用户通过 API 调用了管理员才能用的删除用户接口——因为缺少角色级别校验。Token 泄露开发把 API Token 写死在测试脚本里提交到了 Git任何能访问代码仓库的人都能操作 RAGFlow。默认密码未改adminragflow.io / ragflow的默认账号在生产环境仍在用等于大门敞开。对象存储裸露MinIO 未设置访问策略任何人知道 bucket 名就能直接下载所有原文件。没有权限控制的安全风险链 攻击者获得一个普通用户账号 ↓ 通过修改 API 请求参数遍历所有 dataset_id ↓ 成功读取其他部门的机密知识库 ↓ 导出所有文档原文件MinIO 公开访问 ↓ 数据泄露、合规事故、CEO 问责2 项目设计小胖紧张兮兮地关上门“大师出大事了刚才我发现我能搜到 HR 部门上个月的薪资调整方案我就是一个普通研发按理说不应该看到这玩意啊。这是 Bug 还是配置问题”大师放下手中的咖啡表情严肃“这是个严重的安全问题。RAGFlow 支持权限管理但如果你没有配置默认情况下确实没有做数据集粒度的访问控制。我们先搞清楚 RAGFlow 的权限模型是怎么设计的。”小胖“那到底有哪些权限层级我是说租户、用户、角色……这些都是啥关系”大师“RAGFlow 的权限模型是三层结构”权限模型三层结构 第一层租户Tenant - 最大的隔离单位 - 一个租户 一个独立的数据空间 - 租户A的用户完全看不到租户B的任何数据 - 适用于母公司 vs 子公司、不同客户 第二层用户User - 租户下的登录账号 - 每个用户属于一个租户 - 一个用户可以被授予多个角色 - 适用场景部门员工、管理员 第三层角色 权限Role Permission - 预定义角色admin管理员、normal普通用户 - 管理员可以创建用户、管理所有数据集、系统设置 - 普通用户只能访问自己被授权的数据集 - 数据集级别的权限读检索问答、写上传文档、管理删除技术映射租户 写字楼里的不同公司各用各的楼层互不干扰用户 公司员工角色 职位高管能进所有门实习生只能进自己工位。小白放下笔“那具体到技术实现RAGFlow 是用什么做鉴权的JWT 还是 SessionAPI Token 又是怎么回事”大师“RAGFlow 用了三层鉴权体系各有分工”RAGFlow 鉴权体系 1. Session Cookie浏览器登录 → 用户通过网页登录 /api/v1/login → 服务端生成 Session返回 Set-Cookie → 后续请求自动携带 Cookie → 优点浏览器自动管理无需手动传 Header → 缺点不支持 API 自动化过期快30分钟 2. JWT Token前后端分离鉴权 → 登录时返回 access_token refresh_token → 请求头Authorization: Bearer access_token → access_token 短期有效30分钟-2小时 → refresh_token 用于续期 → 优点无状态、可扩展 → 缺点需要自行管理刷新逻辑 3. API Token服务间鉴权 → 在控制台手动生成长期有效 → 请求头Authorization: Bearer api_token → 适用于CI/CD Pipeline、后端服务、自动化脚本 → 优点简单、长期有效 → 缺点泄露风险高需妥善保管技术映射Session 酒店房卡入住给退房交JWT 健身房年卡有有效期可续API Token 家门钥匙长期有效但绝不外借。小胖“那源码层面是怎么实现鉴权检查的每个 API 都要写一遍校验代码吗”大师“当然不是。用的是 Python 装饰器模式。核心鉴权逻辑在api/apps/__init__.py中”# 源码概念RAGFlow 鉴权装饰器简化版# 文件: api/apps/__init__.pyfromfunctoolsimportwrapsfromflaskimportrequest,gdeflogin_required(f):统一的鉴权装饰器 - 所有需要登录的 API 都加这个wraps(f)asyncdefdecorated(*args,**kwargs):# 尝试三种鉴权方式userNone# 方式1: Session Cookieuserget_user_from_session(request)# 方式2: JWT Token (Authorization: Bearer xxx)ifnotuser:userget_user_from_jwt(request.headers.get(Authorization))# 方式3: API Tokenifnotuser:userget_user_from_api_token(request.headers.get(Authorization))ifnotuser:return{code:401,message:Unauthorized},401g.current_useruser g.current_tenantuser.tenantreturnawaitf(*args,**kwargs)returndecorateddefdataset_permission_required(permissionread):数据集级别的权限检查defdecorator(f):wraps(f)asyncdefdecorated(*args,**kwargs):dataset_idkwargs.get(dataset_id)userg.current_user# 管理员直接通过ifuser.roleadmin:returnawaitf(*args,**kwargs)# 普通用户检查是否被授权访问此数据集ifnothas_permission(user.id,dataset_id,permission):return{code:403,message:Forbidden},403returnawaitf(*args,**kwargs)returndecoratedreturndecorator小白“那多租户是怎么实现的数据库里怎么区分数据属于哪个租户”大师“核心是数据库中的tenant_id字段。几乎所有核心表都有tenant_id列查询时自动带过滤条件”-- 每个表中的 tenant_id 列实现数据隔离SELECT*FROMdocumentWHEREtenant_idtenant_001;-- API 层面的效果-- 租户 A 的用户永远查不到租户 B 的文档-- 等效于每个 SQL 查询后面都自动加了 WHERE tenant_id 当前用户租户ID# 源码概念Peewee ORM 中的租户隔离# 文件: api/db/db_models.pyclassTenantAwareModel(BaseModel):所有需要租户隔离的表都继承此模型tenant_idCharField(max_length64,indexTrue)classmethoddefget_query_for_tenant(cls,tenant_id):获取当前租户可访问的查询returncls.select().where(cls.tenant_idtenant_id)classDocument(TenantAwareModel):nameCharField(max_length255)statusCharField(max_length32)# ...小胖“那 MinIO 对象存储怎么隔离万一 MinIO 的 bucket 被公开了文件不就全泄露了”大师“MinIO 的隔离也依赖租户。文件存储路径是{tenant_id}/{dataset_id}/{document_id}/{filename}。权限控制两重保障一是 MinIO 本身的访问凭证Access Key / Secret Key不能公开二是 API Server 作为网关用户通过 API 下载文件时先做权限校验通过后才生成预签名 URL。”小白“日常安全运营应该做哪些事总不能等出了事再补救。”大师“给你一张安全运维清单”检查项频率操作修改默认密码部署时修改 admin 账号密码API Token 轮换每月生成新 Token废弃旧 Token审计日志检查每周检查异常登录、越权访问记录数据库备份每天mysqldump 异地存储MinIO 访问策略部署时禁用匿名访问配置 Bucket Policy前端依赖漏洞扫描每次发版npm audit后端依赖漏洞扫描每次发版pip-audit/ Safety check代码仓库密钥扫描每次提交git-secrets / TruffleHog防止 Token 泄露3 项目实战环境准备目标在 RAGFlow 中创建两个租户每个租户创建多个用户验证数据集隔离和越权拦截。前提RAGFlow 已部署使用管理员账号登录。分步实现步骤1创建多租户和用户目标创建云帆科技和云帆子公司两个租户各自创建用户。# 使用管理员 TokenTOKENragflow-admin-xxxxxxxx# 1. 创建第二个租户curl-XPOST http://localhost:8080/api/v1/tenants\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ name: 云帆科技-子公司, description: 独立运营的子公司租户 }# 响应: {code: 0, data: {id: tenant_sub_001}}# 2. 在父公司租户下创建普通用户curl-XPOST http://localhost:8080/api/v1/users\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ email: hr_adminyunfan.com, password: SecurePass2024!, nickname: HR管理员, role: normal, tenant_id: tenant_main_001 }# 3. 在子公司租户下创建用户curl-XPOST http://localhost:8080/api/v1/users\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ email: sub_adminyunfan-sub.com, password: SecurePass2024!, nickname: 子公司管理员, role: admin, tenant_id: tenant_sub_001 }步骤2配置数据集权限目标将一个数据集授权给特定用户。# 1. 创建 HR 机密数据集HR_DS$(curl-s-XPOST http://localhost:8080/api/v1/datasets\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{name: HR-机密-薪资调整2024, description: 仅HR部门经理可访问})HR_DS_ID$(echo$HR_DS|jq-r.data.id)# 2. 将数据集授权给 hr_manager 用户curl-XPOST http://localhost:8080/api/v1/datasets/$HR_DS_ID/permissions\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ user_id: user_hr_manager_001, permission: read }# 3. 列出数据集的权限列表curlhttp://localhost:8080/api/v1/datasets/$HR_DS_ID/permissions\-HAuthorization: Bearer$TOKEN\|jq.步骤3验证租户隔离目标确认租户 A 的用户无法访问租户 B 的数据。# verify_tenant_isolation.py - 验证租户隔离fromragflowimportRAGFlow# 租户A的用户登录rag_tenant_aRAGFlow(emailhr_adminyunfan.com,passwordSecurePass2024!,base_urlhttp://localhost:8080/api/v1)# 租户A可以列出自己的数据集datasets_arag_tenant_a.list_datasets()print(f租户A的数据集数:{len(datasets_a)})# 尝试用租户A的Token访问租户B的数据集ID已知属于B的IDtry:# 直接用租户A的鉴权访问租户B的数据集resultrag_tenant_a.get_dataset(tenant_b_dataset_id)print([FAIL] 越权访问成功——存在安全漏洞)exceptExceptionase:print(f[PASS] 越权被拦截:{e})# 租户B的用户登录rag_tenant_bRAGFlow(emailsub_adminyunfan-sub.com,passwordSecurePass2024!,base_urlhttp://localhost:8080/api/v1)# 租户B也无法看到租户A的数据datasets_brag_tenant_b.list_datasets()print(f租户B的数据集数:{len(datasets_b)})步骤4API Token 安全管理目标实现 API Token 的安全生成、存储和使用流程。# secure_token_management.py - 安全的Token管理importosimportjsonfromdatetimeimportdatetime,timedeltaclassTokenManager:生产级 Token 管理最佳实践staticmethoddefgenerate_and_store():生成新 Token 并安全存储绝不在代码中硬编码# 方式1环境变量推荐tokenos.environ.get(RAGFLOW_API_TOKEN)ifnottoken:raiseValueError(请设置环境变量 RAGFLOW_API_TOKEN)# 方式2加密配置文件适合自动化# 使用 vault/secrets manager 读取# token vault_client.read(secret/ragflow/api_token)# 方式3从 CI/CD Secret 变量注入# token ${{ secrets.RAGFLOW_API_TOKEN }}returntokenstaticmethoddefrotate_token(rag_client,old_token_id):定期轮换 Token# 1. 生成新 Tokennew_tokenrag_client.create_api_token(namefauto-rotated-{datetime.now():%Y%m%d})# 2. 更新环境变量/密钥存储# update_vault(secret/ragflow/api_token, new_token.token)# 3. 等待新 Token 生效灰度切换# 建议保留旧 Token 24小时确保所有服务切换完毕# 4. 废弃旧 Tokenrag_client.revoke_api_token(old_token_id)print(fToken 轮换完成。新 Token ID:{new_token.id})staticmethoddefaudit_token_usage(rag_client):审计 Token 使用情况tokensrag_client.list_api_tokens()fortintokens:age_days(datetime.now()-t.created_at).days status⚠ 建议轮换ifage_days90else✓ 正常print(f Token:{t.name}(创建{age_days}天前) -{status})步骤5安全加固检查清单目标对 RAGFlow 部署环境进行安全检查。# security_hardening.sh - 安全加固脚本#!/bin/bashecho RAGFlow 安全加固检查 # 1. 检查默认密码是否已修改ADMIN_CHECK$(curl-s-o/dev/null-w%{http_code}\-XPOST http://localhost:8080/api/v1/login\-HContent-Type: application/json\-d{email:adminragflow.io,password:ragflow})if[$ADMIN_CHECK200];thenecho[FAIL] 默认管理员密码未修改elseecho[PASS] 默认管理员密码已修改fi# 2. 检查 MinIO 是否禁用匿名访问MINIO_ANON$(curl-s-o/dev/null-w%{http_code}\http://localhost:9001/minio/health/live)if[$MINIO_ANON200];thenecho[WARN] MinIO 健康检查端点公开可访问fi# 3. 检查数据库端口是否暴露到公网DB_PORT_CHECK$(dockerinspect ragflow-mysql-1|jq-r.[0].NetworkSettings.Ports[3306/tcp])ifecho$DB_PORT_CHECK|grep-q0.0.0.0;thenecho[FAIL] MySQL 端口暴露在公网elseecho[PASS] MySQL 端口未暴露在公网fi# 4. 检查 .env 文件中是否含明文密码ifgrep-qpassword.*.*ragflowdocker/.env2/dev/null;thenecho[FAIL] .env 文件可能包含默认密码fi# 5. 检查是否使用了 HTTPS生产环境ifcurl-s-o/dev/null-w%{http_code}https://localhost:8080|grep-q000;thenecho[WARN] 未启用 HTTPS数据传输未加密fiecho 检查完成 测试验证# test_security.py - 安全功能验证测试importpytestfromragflowimportRAGFlowclassTestTenantIsolation:deftest_cross_tenant_access_denied(self):验证跨租户访问被拒绝user_aRAGFlow(emailuser_atenant_a.com,passwordxxx,base_urlBASE)user_bRAGFlow(emailuser_btenant_b.com,passwordxxx,base_urlBASE)# user_a 创建一个数据集dsuser_a.create_dataset(nameTenantA-Confidential)# user_b 尝试用 ds.id 访问应该403withpytest.raises(Exception,match403|Forbidden|not found):user_b.get_dataset(ds.id)# 清理user_a.delete_dataset(ds.id)deftest_user_cannot_access_unpermitted_dataset(self):验证用户不能访问未授权的数据集adminRAGFlow(api_keyADMIN_KEY,base_urlBASE)normal_userRAGFlow(emailnormaltest.com,passwordxxx,base_urlBASE)# 管理员创建数据集但不授权给 normal_userdsadmin.create_dataset(nameRestricted-DS)# normal_user 尝试访问应该403或404withpytest.raises(Exception):normal_user.get_dataset(ds.id)admin.delete_dataset(ds.id)classTestTokenSecurity:deftest_expired_token_rejected(self):验证过期 Token 被拒绝# 使用一个已知过期的 Tokeninvalid_ragRAGFlow(api_keyexpired-token-xxxx,base_urlBASE)withpytest.raises(Exception,match401|Unauthorized):invalid_rag.list_datasets()deftest_revoked_token_rejected(self):验证被撤销的 Token 被拒绝adminRAGFlow(api_keyADMIN_KEY,base_urlBASE)# 创建临时 Tokentemp_tokenadmin.create_api_token(nametemp-test-token)# 立即撤销admin.revoke_api_token(temp_token.id)# 尝试使用已撤销的 Tokenrevoked_ragRAGFlow(api_keytemp_token.token,base_urlBASE)withpytest.raises(Exception,match401|Unauthorized|revoked):revoked_rag.list_datasets()完整代码清单Git 仓库https://github.com/infiniflow/ragflow路径说明api/apps/__init__.py鉴权装饰器、租户上下文、Blueprint 注册api/apps/auth/登录/注册/Token 刷新api/db/db_models.py数据库模型User、Tenant、APITokenapi/db/services/user_service.py用户管理服务层api/db/services/tenant_service.py租户管理服务层4 项目总结优点 缺点维度RAGFlow 权限Keycloak 自定义Auth0自建 RBAC多租户支持★★★ 内置★★★ 强大★★★ 标准化★★☆ 需开发数据集级权限★★☆ 基本★☆☆ 需自定义映射★☆☆ 需自定义★★★ 灵活部署复杂度★★★ 开箱即用★★☆ 独立部署★★☆ 云端依赖★☆☆ 开发量大Token 管理★★☆ 基本 CRUD★★★ 完整生命周期★★★ 企业级★★★ 灵活审计日志★☆☆ 基础★★★ 完整★★★ 完整★★★ 自定义适用场景多部门协作不同部门共享一个 RAGFlow 实例但数据集互相隔离。多客户 SaaS为每个客户创建独立租户客户之间数据完全隔离。分级授权管理员、数据集管理者、只读用户三种角色各司其职。合规审计有权限管理和 Token 生命周期的企业需要可审计的数据访问记录。API 集成安全第三方系统通过 API Token 调用Token 具有可管理的生命周期。不适用场景字段级权限如果需要对文档中的某些段落设权限如薪资文档中基本工资字段可见、绩效系数不可见RAGFlow 不支持。复杂的审批工作流如用户请求访问数据集 → 审批 → 自动授权需要额外开发。注意事项默认管理员密码是安全第一道关部署后第一步必须是修改adminragflow.io的密码。API Token 不可找回生成 Token 时只会展示一次完整 Token 值。之后只能看到 Token 的前后缀——如果丢失只能重新生成。Session 超时风险浏览器长时间不操作后 Session 过期需重新登录。对自动化测试脚本建议用 API Token。MinIO 访问策略务必锁定不要依赖隐藏路径做安全——MinIO 必须配置deny anonymous access策略。数据库直连风险所有租户数据在同一数据库的不同行中通过tenant_id区分。数据库管理员具有绝对访问权限——需要独立管控。常见踩坑经验故障现象根因解决方法用户登录后看不到任何数据集新用户未被授权任何数据集管理员在控制台→用户管理→编辑用户→授权数据集API 调用始终返回 401用了 API Token 但 Token 错误或已过期确认 Token 完整复制检查是否被撤销两个租户的用户能看到对方数据未启用租户隔离或 API 未校验 tenant_id升级 RAGFlow 版本确认tenant_id列存在Token 提交到 Git 被泄露开发把 Token 写死在代码中立即在控制台撤销该 Token生成新 Token改用环境变量MinIO 链接能直接下载文件未配置 Bucket Policy设置 MinIO 为私有访问禁用匿名读取思考题公司有 500 个用户和 100 个数据集如果每个用户都要手动配置对每个数据集的访问权限管理员的工作量不可接受。请设计一个基于属性的权限模型ABAC——根据用户部门属性自动决定其可访问的数据集范围无需逐一手动授权。假设攻击者获取了一个普通用户的 API Token但由于 RAGFlow 缺少 API 调用频率限制攻击者在一小时内发起了 10 万次问答请求消耗大量 LLM 费用拖垮服务。请设计一套 API 限流和异常检测方案保护服务不被滥用。答案提示见第14章末尾或附录 D。延伸阅读与资源10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析403 白屏。