7.2 · 认证与授权（Authentication and Authorization）

复杂文档理解与引用检索 · 本章是 RAGFlow DeepWiki 中文译文的独立章节页，保留原始链接、源码锚点、模块标签和章节层级。

项目RAGFlow 章节7.2 状态全文译文模块认证、权限与安全、系统架构、配置治理、接口与服务契约

项目要点页2.5 参考项目项目章节目录RAGFlow DeepWiki 原始章节Authentication and Authorization 上一章7.1 下一章7.3

源码线索

admin/server/admin_server.py
api/apps/__init__.py
api/db/init_data.py
api/ragflow_server.py
api/settings.py
api/utils/api_utils.py
conf/service_conf.yaml
docker/service_conf.yaml.template
rag/utils/redis_conn.py
test/testcases/test_web_api/test_system_app/test_apps_init_unit.py

模块标签

认证、权限与安全
系统架构
配置治理
接口与服务契约
界面与交互

中文译文

认证与授权（中文译文）

原始 DeepWiki 页面：https://deepwiki.com/infiniflow/ragflow/7.2-authentication-and-authorization

翻译时间：2026-05-27T08:44:21.061Z

翻译模型：deepseek-chat

原文字符数：9390

项目：RAGFlow (ragflow)

---

认证与授权

概述

RAGFlow 实现了一套灵活的认证架构，可服务于不同类型的客户端：

API 密钥认证：用于 Python SDK、CLI 工具和 HTTP API 客户端。
基于会话的认证：用于 React Web 前端，通过 Quart 管理 JWT 令牌和会话。
OAuth/OIDC 认证：支持 GitHub 和可配置的 OAuth2.0/OIDC 提供商。

授权基于租户实现，每个已认证用户都关联一个 tenant_id，该 ID 控制对数据集、聊天助手、智能体和其他资源的访问。

来源：api/apps/__init__.py:129-188, api/utils/api_utils.py:135-151

认证方法

API 密钥认证

API 密钥用于程序化访问。每个 API 密钥存储在 APIToken 表中，并关联到特定的租户。这是 ragflow_sdk 和外部 HTTP 客户端的主要认证机制。

令牌校验流程

应用初始化中的 _load_user 函数负责处理：如果 JWT 解码失败，则从 Authorization 请求头中提取 API 令牌。该函数同时支持标准的 Bearer 令牌和原始访问令牌。

标题：_load_user 中的 API 令牌解析

graph TB
    Request["HTTP 请求<br/>Authorization: Bearer TOKEN"]
    ExtractToken["从请求头提取令牌<br/>split()[1]"]
    QueryDB["查询 APIToken 表<br/>APIToken.query(token=token)"]
    CheckExists{"令牌是否存在？"}
    GetTenant["按 tenant_id 查询<br/>UserService"]
    CheckUserToken{"用户是否有有效的<br/>access_token？"}
    SetUser["设置 g.user = user[0]"]
    ReturnUser["返回用户对象"]
    Fail["返回 None"]

    Request --> ExtractToken
    ExtractToken --> QueryDB
    QueryDB --> CheckExists
    CheckExists -->|是| GetTenant
    GetTenant --> CheckUserToken
    CheckUserToken -->|是| SetUser
    SetUser --> ReturnUser
    CheckExists -->|否| Fail
    CheckUserToken -->|否| Fail

实现细节见 api/apps/__init__.py:171-186：

# JWT 解码失败，尝试作为 api_token
try:
    objs = APIToken.query(token=auth_token)
    if objs:
        user = UserService.query(id=objs[0].tenant_id, status=StatusEnum.VALID.value)
        if user:
            if not user[0].access_token or not user[0].access_token.strip():
                logging.warning(f"用户 {user[0].email} 在数据库中的 access_token 为空")
                return _load_user_from_session()
            g.auth_via_api_token = True
            g.user = user[0]
            return user[0]
except Exception as e_api_token:
    logging.warning(f"从 api token 加载用户时出现异常：{e_api_token}")

来源：api/apps/__init__.py:171-188, api/db/db_models.py:26-26, api/db/services/__init__.py:27-27

基于会话的认证

Web UI 使用基于会话的认证，令牌序列化到 Authorization 请求头中，会话状态保存在 Redis 中。

用户登录流程

标题：通过 user_app.py 的 Web 登录序列

sequenceDiagram
    participant Client as "Web 客户端"
    participant UserApp as "api.apps.user_app.login"
    participant UserService as "api.db.services.user_service.UserService"
    participant SessionStore as "Redis 会话存储"

    Client->>UserApp: POST /login {email, password}
    UserApp->>UserApp: decrypt(password)
    UserApp->>UserService: query(email=email)
    UserApp->>UserService: query_user(email, password)
    UserService-->>UserApp: 用户对象

    alt 认证成功
        UserApp->>UserApp: user.access_token = get_uuid()
        UserApp->>UserApp: login_user(user)
        UserApp->>UserService: user.save()
        UserApp-->>Client: construct_response(auth=user.get_id())
    else 认证失败
        UserApp-->>Client: get_json_result(code=AUTHENTICATION_ERROR)
    end

系统使用 get_uuid() 生成唯一的 access_token，并通过 login_user 建立会话。

来源：api/apps/__init__.py:96-126, common/misc_utils.py:37-38

@login_required 装饰器

@login_required 装饰器通过检查 current_user 代理来保护端点。

# 来自 api/apps/__init__.py:194-204
def login_required(func: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        user = current_user
        if not user:
            raise QuartAuthUnauthorized()
        return await current_app.ensure_async(func)(*args, **kwargs)
    return wrapper

current_user 是一个 LocalProxy api/apps/__init__.py:191，它会调用 _load_user api/apps/__init__.py:129。该函数尝试使用 Serializer(secret_key=settings.get_secret_key()) api/apps/__init__.py:130 从 Authorization 请求头中的 JWT 签名令牌加载用户。

来源：api/apps/__init__.py:129-204

OAuth/OIDC 认证

RAGFlow 支持动态的 OAuth 2.0 和 OIDC 提供商集成，可实现企业级单点登录。

提供商架构

系统使用 service_conf.yaml 中定义的配置来启用 OAuth 提供商，例如 GitHub、OIDC 或通用 OAuth2。

标题：OAuth 配置与数据库模型映射

graph LR
    Config["conf/service_conf.yaml"]
    OAuthBlock["oauth: github/oidc/oauth2"]
    UserDB["api.db.db_models.User"]
    TenantDB["api.db.db_models.Tenant"]

    Config --> OAuthBlock
    OAuthBlock -->|成功回调| UserDB
    UserDB -->|所有权| TenantDB

初始化与超级用户

在系统启动期间，RAGFlow 可以初始化一个默认的超级用户。这由 init_superuser 处理，它会创建初始的 User、Tenant 和 UserTenant 记录，确保系统拥有一个用于管理任务的所有者角色。

来源：api/db/init_data.py:50-94, api/ragflow_server.py:125-126

授权与 current_user

current_user 代理

current_user 符号是一个 werkzeug.local.LocalProxy，它为当前请求上下文提供 User 数据库模型实例。

实体	角色	来源
`_load_user`	执行实际数据库查询和令牌校验的函数	`api/apps/__init__.py:129`
`current_user`	`_load_user` 结果的代理	`api/apps/__init__.py:191`
`g.user`	用户对象的请求全局存储	`api/apps/__init__.py:132`

租户隔离

RAGFlow 中的授权主要通过使用 tenant_id 过滤数据库查询来实施。在 Web 前端，登录页面 web/src/pages/login-next/index.tsx:65-185 为用户提供了输入凭据的界面，这些凭据随后会在后端进行验证。

标题：代码实体中的租户授权模式

graph TD
    API["Quart 路由处理器"]
    Auth["@login_required"]
    UserObj["current_user"]
    APIToken["api.db.db_models.APIToken"]
    DBQuery["UserService.query(id=APIToken.tenant_id)"]

    API --> Auth
    Auth --> UserObj
    UserObj --> APIToken
    APIToken --> DBQuery

来源：api/apps/__init__.py:171-182, web/src/pages/login-next/index.tsx:65-185

配置

认证行为通过 service_conf.yaml 和环境变量进行控制。

会话存储：配置为使用 Redis api/apps/__init__.py:78-79。
OAuth 配置：在 service_conf.yaml 的 oauth 块中定义 conf/service_conf.yaml:111-133。
密钥：settings.get_secret_key() 用于签名 JWT 和会话 Cookie api/apps/__init__.py:83。
Redis 连接：由 RedisDB 单例处理，该单例为会话存储提供后端支持 rag/utils/redis_conn.py:61-114。

来源：api/apps/__init__.py:77-85, conf/service_conf.yaml:111-145, rag/utils/redis_conn.py:61-145