企业官网建设流程全解析

1 项目背景

业务场景

「云帆科技」的 RAGFlow 平台经过两个月的试运行，已从最初的 HR 一个部门扩展到财务、法务、研发、市场共五个部门。每个部门的知识库中包含了各自的制度文档、项目资料和业务数据。随着用户增多，安全问题逐渐浮出水面：

HR 总监发现，研发部的同事竟然能在 RAGFlow 中搜索到《2024年薪资调整方案》——这份文档标记为"机密"，只有 HR 部门经理以上才能查看。调查发现，原因是所有部门的数据集没有做权限隔离，所有人共享同一个管理员账号。

与此同时，CIO 提出：公司很快要扩展到子公司，需要"租户"级别的隔离——子公司的数据和母公司数据完全隔离，就像两套独立的 RAGFlow。IT 安全部门也加入了讨论，提出了 API Token 泄露的风险、默认密码未修改的问题、以及 MinIO 对象存储如果配置不当可能被公开访问的风险。

痛点

权限和安全管理缺失的典型事故：

1. 水平越权：用户 A 可以通过修改 API 请求中的dataset_id参数访问用户 B 的数据集——因为后端没有做"数据集归属用户"的校验。
2. 垂直越权：普通用户通过 API 调用了管理员才能用的"删除用户"接口——因为缺少角色级别校验。
3. Token 泄露：开发把 API Token 写死在测试脚本里提交到了 Git，任何能访问代码仓库的人都能操作 RAGFlow。
4. 默认密码未改：admin@ragflow.io / ragflow的默认账号在生产环境仍在用，等于大门敞开。
5. 对象存储裸露：MinIO 未设置访问策略，任何人知道 bucket 名就能直接下载所有原文件。

没有权限控制的安全风险链： 攻击者获得一个普通用户账号 ↓ 通过修改 API 请求参数遍历所有 dataset_id ↓ 成功读取其他部门的机密知识库 ↓ 导出所有文档原文件（MinIO 公开访问） ↓ 数据泄露、合规事故、CEO 问责

2 项目设计

小胖：（紧张兮兮地关上门）“大师，出大事了！刚才我发现我能搜到 HR 部门上个月的薪资调整方案！我就是一个普通研发，按理说不应该看到这玩意啊。这是 Bug 还是配置问题？”

大师：（放下手中的咖啡，表情严肃）“这是个严重的安全问题。RAGFlow 支持权限管理，但如果你没有配置，默认情况下确实没有做数据集粒度的访问控制。我们先搞清楚 RAGFlow 的权限模型是怎么设计的。”

小胖：“那到底有哪些权限层级？我是说，租户、用户、角色……这些都是啥关系？”

大师：“RAGFlow 的权限模型是三层结构：”

权限模型三层结构： 第一层：租户（Tenant） - 最大的隔离单位 - 一个租户 = 一个独立的数据空间 - 租户A的用户完全看不到租户B的任何数据 - 适用于：母公司 vs 子公司、不同客户 第二层：用户（User） - 租户下的登录账号 - 每个用户属于一个租户 - 一个用户可以被授予多个角色 - 适用场景：部门员工、管理员 第三层：角色 & 权限（Role & Permission） - 预定义角色：admin（管理员）、normal（普通用户） - 管理员：可以创建用户、管理所有数据集、系统设置 - 普通用户：只能访问自己被授权的数据集 - 数据集级别的权限：读（检索问答）、写（上传文档）、管理（删除）

技术映射：租户 = 写字楼里的不同公司（各用各的楼层，互不干扰）；用户 = 公司员工；角色 = 职位（高管能进所有门，实习生只能进自己工位）。

小白：（放下笔）“那具体到技术实现，RAGFlow 是用什么做鉴权的？JWT 还是 Session？API 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):# 尝试三种鉴权方式user=None# 方式1: Session Cookieuser=get_user_from_session(request)# 方式2: JWT Token (Authorization: Bearer xxx)ifnotuser:user=get_user_from_jwt(request.headers.get("Authorization"))# 方式3: API Tokenifnotuser:user=get_user_from_api_token(request.headers.get("Authorization"))ifnotuser:return{"code":401,"message":"Unauthorized"},401g.current_user=user g.current_tenant=user.tenantreturnawaitf(*args,**kwargs)returndecorateddefdataset_permission_required(permission="read"):"""数据集级别的权限检查"""defdecorator(f):@wraps(f)asyncdefdecorated(*args,**kwargs):dataset_id=kwargs.get("dataset_id")user=g.current_user# 管理员直接通过ifuser.role=="admin":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_id='tenant_001';-- API 层面的效果：-- 租户 A 的用户永远查不到租户 B 的文档-- 等效于每个 SQL 查询后面都自动加了 WHERE tenant_id = <当前用户租户ID>

# 源码概念：Peewee ORM 中的租户隔离# 文件: api/db/db_models.pyclassTenantAwareModel(BaseModel):"""所有需要租户隔离的表都继承此模型"""tenant_id=CharField(max_length=64,index=True)@classmethoddefget_query_for_tenant(cls,tenant_id):"""获取当前租户可访问的查询"""returncls.select().where(cls.tenant_id==tenant_id)classDocument(TenantAwareModel):name=CharField(max_length=255)status=CharField(max_length=32)# ...

小胖：“那 MinIO 对象存储怎么隔离？万一 MinIO 的 bucket 被公开了，文件不就全泄露了？”

大师：“MinIO 的隔离也依赖租户。文件存储路径是{tenant_id}/{dataset_id}/{document_id}/{filename}。权限控制两重保障：一是 MinIO 本身的访问凭证（Access Key / Secret Key），不能公开；二是 API Server 作为网关，用户通过 API 下载文件时先做权限校验，通过后才生成预签名 URL。”

小白：“日常安全运营应该做哪些事？总不能等出了事再补救。”

大师：“给你一张安全运维清单：”

3 项目实战

环境准备

目标：在 RAGFlow 中创建两个租户，每个租户创建多个用户，验证数据集隔离和越权拦截。

前提：RAGFlow 已部署，使用管理员账号登录。

分步实现

步骤1：创建多租户和用户

目标：创建"云帆科技"和"云帆子公司"两个租户，各自创建用户。

# 使用管理员 TokenTOKEN="ragflow-admin-xxxxxxxx"# 1. 创建第二个租户curl-XPOST http://localhost:8080/api/v1/tenants\-H"Authorization: Bearer$TOKEN"\-H"Content-Type: application/json"\-d'{ "name": "云帆科技-子公司", "description": "独立运营的子公司租户" }'# 响应: {"code": 0, "data": {"id": "tenant_sub_001"}}# 2. 在父公司租户下创建普通用户curl-XPOST http://localhost:8080/api/v1/users\-H"Authorization: Bearer$TOKEN"\-H"Content-Type: application/json"\-d'{ "email": "hr_admin@yunfan.com", "password": "SecurePass@2024!", "nickname": "HR管理员", "role": "normal", "tenant_id": "tenant_main_001" }'# 3. 在子公司租户下创建用户curl-XPOST http://localhost:8080/api/v1/users\-H"Authorization: Bearer$TOKEN"\-H"Content-Type: application/json"\-d'{ "email": "sub_admin@yunfan-sub.com", "password": "SecurePass@2024!", "nickname": "子公司管理员", "role": "admin", "tenant_id": "tenant_sub_001" }'

步骤2：配置数据集权限

目标：将一个数据集授权给特定用户。

# 1. 创建 HR 机密数据集HR_DS=$(curl-s-XPOST http://localhost:8080/api/v1/datasets\-H"Authorization: Bearer$TOKEN"\-H"Content-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\-H"Authorization: Bearer$TOKEN"\-H"Content-Type: application/json"\-d'{ "user_id": "user_hr_manager_001", "permission": "read" }'# 3. 列出数据集的权限列表curlhttp://localhost:8080/api/v1/datasets/$HR_DS_ID/permissions\-H"Authorization: Bearer$TOKEN"\|jq'.'

步骤3：验证租户隔离

目标：确认租户 A 的用户无法访问租户 B 的数据。

# verify_tenant_isolation.py - 验证租户隔离fromragflowimportRAGFlow# 租户A的用户登录rag_tenant_a=RAGFlow(email="hr_admin@yunfan.com",password="SecurePass@2024!",base_url="http://localhost:8080/api/v1")# 租户A可以列出自己的数据集datasets_a=rag_tenant_a.list_datasets()print(f"租户A的数据集数:{len(datasets_a)}")# 尝试用租户A的Token访问租户B的数据集ID（已知属于B的ID）try:# 直接用租户A的鉴权访问租户B的数据集result=rag_tenant_a.get_dataset("tenant_b_dataset_id")print("[FAIL] 越权访问成功——存在安全漏洞！")exceptExceptionase:print(f"[PASS] 越权被拦截:{e}")# 租户B的用户登录rag_tenant_b=RAGFlow(email="sub_admin@yunfan-sub.com",password="SecurePass@2024!",base_url="http://localhost:8080/api/v1")# 租户B也无法看到租户A的数据datasets_b=rag_tenant_b.list_datasets()print(f"租户B的数据集数:{len(datasets_b)}")

步骤4：API Token 安全管理

目标：实现 API Token 的安全生成、存储和使用流程。

# secure_token_management.py - 安全的Token管理importosimportjsonfromdatetimeimportdatetime,timedeltaclassTokenManager:"""生产级 Token 管理最佳实践"""@staticmethoddefgenerate_and_store():"""生成新 Token 并安全存储（绝不在代码中硬编码）"""# 方式1：环境变量（推荐）token=os.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 }}"returntoken@staticmethoddefrotate_token(rag_client,old_token_id):"""定期轮换 Token"""# 1. 生成新 Tokennew_token=rag_client.create_api_token(name=f"auto-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(f"Token 轮换完成。新 Token ID:{new_token.id}")@staticmethoddefaudit_token_usage(rag_client):"""审计 Token 使用情况"""tokens=rag_client.list_api_tokens()fortintokens:age_days=(datetime.now()-t.created_at).days status="⚠ 建议轮换"ifage_days>90else"✓ 正常"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\-H"Content-Type: application/json"\-d'{"email":"admin@ragflow.io","password":"ragflow"}')if["$ADMIN_CHECK"="200"];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_ANON"="200"];thenecho"[WARN] MinIO 健康检查端点公开可访问"fi# 3. 检查数据库端口是否暴露到公网DB_PORT_CHECK=$(dockerinspect ragflow-mysql-1|jq-r'.[0].NetworkSettings.Ports["3306/tcp"]')ifecho"$DB_PORT_CHECK"|grep-q"0.0.0.0";thenecho"[FAIL] MySQL 端口暴露在公网！"elseecho"[PASS] MySQL 端口未暴露在公网"fi# 4. 检查 .env 文件中是否含明文密码ifgrep-q"password.*=.*ragflow"docker/.env2>/dev/null;thenecho"[FAIL] .env 文件可能包含默认密码"fi# 5. 检查是否使用了 HTTPS（生产环境）ifcurl-s-o/dev/null-w"%{http_code}""https://localhost:8080"|grep-q"000";thenecho"[WARN] 未启用 HTTPS，数据传输未加密"fiecho"=== 检查完成 ==="

测试验证

# test_security.py - 安全功能验证测试importpytestfromragflowimportRAGFlowclassTestTenantIsolation:deftest_cross_tenant_access_denied(self):"""验证跨租户访问被拒绝"""user_a=RAGFlow(email="user_a@tenant_a.com",password="xxx",base_url=BASE)user_b=RAGFlow(email="user_b@tenant_b.com",password="xxx",base_url=BASE)# user_a 创建一个数据集ds=user_a.create_dataset(name="TenantA-Confidential")# user_b 尝试用 ds.id 访问（应该403）withpytest.raises(Exception,match="403|Forbidden|not found"):user_b.get_dataset(ds.id)# 清理user_a.delete_dataset(ds.id)deftest_user_cannot_access_unpermitted_dataset(self):"""验证用户不能访问未授权的数据集"""admin=RAGFlow(api_key=ADMIN_KEY,base_url=BASE)normal_user=RAGFlow(email="normal@test.com",password="xxx",base_url=BASE)# 管理员创建数据集但不授权给 normal_userds=admin.create_dataset(name="Restricted-DS")# normal_user 尝试访问（应该403或404）withpytest.raises(Exception):normal_user.get_dataset(ds.id)admin.delete_dataset(ds.id)classTestTokenSecurity:deftest_expired_token_rejected(self):"""验证过期 Token 被拒绝"""# 使用一个已知过期的 Tokeninvalid_rag=RAGFlow(api_key="expired-token-xxxx",base_url=BASE)withpytest.raises(Exception,match="401|Unauthorized"):invalid_rag.list_datasets()deftest_revoked_token_rejected(self):"""验证被撤销的 Token 被拒绝"""admin=RAGFlow(api_key=ADMIN_KEY,base_url=BASE)# 创建临时 Tokentemp_token=admin.create_api_token(name="temp-test-token")# 立即撤销admin.revoke_api_token(temp_token.id)# 尝试使用已撤销的 Tokenrevoked_rag=RAGFlow(api_key=temp_token.token,base_url=BASE)withpytest.raises(Exception,match="401|Unauthorized|revoked"):revoked_rag.list_datasets()

完整代码清单

Git 仓库：https://github.com/infiniflow/ragflow

4 项目总结

优点 & 缺点

适用场景

1. 多部门协作：不同部门共享一个 RAGFlow 实例，但数据集互相隔离。
2. 多客户 SaaS：为每个客户创建独立租户，客户之间数据完全隔离。
3. 分级授权：管理员、数据集管理者、只读用户三种角色各司其职。
4. 合规审计：有权限管理和 Token 生命周期的企业，需要可审计的数据访问记录。
5. API 集成安全：第三方系统通过 API Token 调用，Token 具有可管理的生命周期。

不适用场景：

1. 字段级权限：如果需要对"文档中的某些段落"设权限（如薪资文档中"基本工资"字段可见、"绩效系数"不可见），RAGFlow 不支持。
2. 复杂的审批工作流：如"用户请求访问数据集 → 审批 → 自动授权"，需要额外开发。

注意事项

1. 默认管理员密码是安全第一道关：部署后第一步必须是修改admin@ragflow.io的密码。
2. API Token 不可找回：生成 Token 时只会展示一次完整 Token 值。之后只能看到 Token 的前后缀——如果丢失，只能重新生成。
3. Session 超时风险：浏览器长时间不操作后 Session 过期，需重新登录。对自动化测试脚本建议用 API Token。
4. MinIO 访问策略务必锁定：不要依赖"隐藏路径"做安全——MinIO 必须配置deny anonymous access策略。
5. 数据库直连风险：所有租户数据在同一数据库的不同行中（通过tenant_id区分）。数据库管理员具有绝对访问权限——需要独立管控。

常见踩坑经验

思考题

1. 公司有 500 个用户和 100 个数据集，如果每个用户都要手动配置对每个数据集的访问权限，管理员的工作量不可接受。请设计一个"基于属性的权限模型（ABAC）"——根据用户部门属性自动决定其可访问的数据集范围，无需逐一手动授权。

2. 假设攻击者获取了一个普通用户的 API Token，但由于 RAGFlow 缺少 API 调用频率限制，攻击者在一小时内发起了 10 万次问答请求（消耗大量 LLM 费用+拖垮服务）。请设计一套 API 限流和异常检测方案，保护服务不被滥用。

（答案提示见第14章末尾或附录 D。）

延伸阅读与资源

10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析403 白屏。