Dify 中的 Bearer Token 与 API-Key 鉴权方式

本文使用 Dify v0.10.2 版本，在 Dify 中包括 Bearer Token 与 API-Key 鉴权这 2 种方式。console（URL 前缀/console/api）和 web（URL 前缀/api）蓝图使用的是 Bearer Token 鉴权方式，而 service_api（URL 前缀/v1）蓝图使用的是 API-Key 鉴权方式。console 蓝图通过 login_required 装饰器[1]实现，web 蓝图通过继承 WebApiResource，即 validate_jwt_token 装饰器实现，而 service_api 通过 validate_app_token 装饰器实现。

一.service_api 蓝图与 API-Key 鉴权

API-Key 与 Bearer Token 鉴权区别，如下所示：

1.service_api 蓝图

简单理解，只有这一个地方的 API 使用的是 API-Key 鉴权方式，其它地方均使用 earer Token 鉴权方式。

2.validate_app_token 装饰器

该装饰器 validate_app_token 用于验证应用程序的 API 令牌，检查相关的应用和租户状态，并根据需要提取用户信息，为后续的视图函数提供必要的上下文。

（1）函数定义和参数

def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optional[FetchUserArg] = None):

• 定义一个装饰器 validate_app_token，接受一个可选的视图函数 view 和一个关键字参数 fetch_user_arg。
• fetch_user_arg 是一个可选参数，用于指定如何从请求中获取用户 ID。

（2）装饰器内部定义

def decorator(view_func):

• 定义内部函数 decorator，它接受要装饰的视图函数 view_func。

（3）装饰的视图函数

@wraps(view_func)
def decorated_view(*args, **kwargs):

• 使用 functools.wraps 保留原始视图函数的元数据。
• 定义内部函数 decorated_view，接受任意位置和关键字参数。

（4）获取和验证 API 令牌

api_token = validate_and_get_api_token("app")

• 调用函数 validate_and_get_api_token 验证并获取 API 令牌，类型为 “app”。

（5）查询应用模型

app_model = db.session.query(App).filter(App.id == api_token.app_id).first()
if not app_model:
    raise Forbidden("The app no longer exists.")

• 查询数据库，获取与 api_token 中的 app_id 关联的应用模型 app_model。
• 如果未找到该应用模型，抛出 Forbidden 异常，说明应用不存在。

（6）检查应用状态

if app_model.status != "normal":
    raise Forbidden("The app's status is abnormal.")

• 检查应用的状态是否为正常。如果不是，则抛出 Forbidden 异常，说明应用状态异常。

if not app_model.enable_api:
    raise Forbidden("The app's API service has been disabled.")

• 检查应用的 API 服务是否启用。如果禁用，则抛出 Forbidden 异常。

（7）查询租户模型

tenant = db.session.query(Tenant).filter(Tenant.id == app_model.tenant_id).first()
if tenant.status == TenantStatus.ARCHIVE:
    raise Forbidden("The workspace's status is archived.")

• 查询与 app_model 相关联的租户模型 tenant，并检查租户的状态。
• 如果租户状态为归档（ARCHIVE），则抛出 Forbidden 异常。

（8）将应用模型传递给视图函数

kwargs["app_model"] = app_model

• 将 app_model 添加到 kwargs 中，以便传递给被装饰的视图函数。

（9）获取用户 ID

if fetch_user_arg:
    if fetch_user_arg.fetch_from == WhereisUserArg.QUERY:
        user_id = request.args.get("user")
    elif fetch_user_arg.fetch_from == WhereisUserArg.JSON:
        user_id = request.get_json().get("user")
    elif fetch_user_arg.fetch_from == WhereisUserArg.FORM:
        user_id = request.form.get("user")
    else:
        # use default-user
        user_id = None

• 如果 fetch_user_arg 被提供，根据其 fetch_from 属性从不同来源获取用户 ID：
  • 如果是查询参数（QUERY），从请求参数中获取。
  • 如果是 JSON 体（JSON），从请求 JSON 中获取。
  • 如果是表单数据（FORM），从请求表单中获取。
  • 如果没有指定来源，则 user_id 设置为 None。

（10）检查用户 ID 的有效性

if not user_id and fetch_user_arg.required:
    raise ValueError("Arg user must be provided.")

• 如果 user_id 不存在且 fetch_user_arg 指定了该参数是必需的，则抛出 ValueError。

if user_id:
    user_id = str(user_id)

如果 user_id 存在，则将其转换为字符串。

（11）创建或更新用户信息

kwargs["end_user"] = create_or_update_end_user_for_user_id(app_model, user_id)

• 调用函数 create_or_update_end_user_for_user_id，根据 app_model 和 user_id 创建或更新用户信息，并将结果存入 kwargs 中。

（12）返回视图函数的结果

return view_func(*args, **kwargs)

• 调用被装饰的视图函数，传递原始参数和 kwargs，并返回结果。

（13）装饰器的返回

return decorated_view

• 返回装饰后的视图函数 decorated_view。

（14）检查视图函数参数

if view is None:
    return decorator
else:
    return decorator(view)

• 如果 view 参数为 None，则返回 decorator 函数；否则将视图函数 view 传递给 decorator 并返回。

二.console 蓝图与 Bearer Token 鉴权（用户登录操作）

源码位置：dify\api\controllers\console\auth\login.py

该接口返回结果示例，如下所示：

{
    "result": "success",
    "data": {
        "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiZWJiZjI1YmUtZWM4NS00ZDY4LTlkZDItNWJjMmVlMjg1NWU0IiwiZXhwIjoxNzMwNjg5Mjk4LCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.EPfEaeoDY8NVeS7TDC95-B0rvFYjH9BVqiFY9IwYcVQ",
        "refresh_token": "32a914bb7d7dac9b1b7b5ba98d6584007acbcba4638374d8e0171877f5c9eee34eae400917aadc433d7801181367579dd72894ae8f512aaba00929ab98bf509f"
    }
}

1.AccountService.authenticate()方法

account = AccountService.authenticate(args[“email”], args[“password”])该方法实现了基于邮箱和密码的账户认证流程，包括账户查询、状态检查、密码哈希、首次登录处理和状态更新。如果所有条件都满足，最后将返回一个有效的 Account 对象。

@staticmethod
def authenticate(email: str, password: str, invite_token: Optional[str] = None) -> Account:
    <em>"""authenticate account with email and password"""</em>

<em>    </em>account = Account.query.filter_by(email=email).first()
    if not account:
        raise AccountNotFoundError()

    if account.status == AccountStatus.BANNED.value:
        raise AccountLoginError("Account is banned.")

    if password and invite_token and account.password is None:
        # if invite_token is valid, set password and password_salt
        salt = secrets.token_bytes(16)
        base64_salt = base64.b64encode(salt).decode()
        password_hashed = hash_password(password, salt)
        base64_password_hashed = base64.b64encode(password_hashed).decode()
        account.password = base64_password_hashed
        account.password_salt = base64_salt

    if account.password is None or not compare_password(password, account.password, account.password_salt):
        raise AccountPasswordError("Invalid email or password.")

    if account.status == AccountStatus.PENDING.value:
        account.status = AccountStatus.ACTIVE.value
        account.initialized_at = datetime.now(timezone.utc).replace(tzinfo=None)

    db.session.commit()

    return account

生成 16 字节的随机盐值，并将其编码为 Base64 字符串，以便后续用于密码哈希。对用户提供的密码进行哈希处理，使用生成的盐值，然后将哈希结果编码为 Base64 字符串。将哈希后的密码和盐值存储到账户对象中。

2. AccountService.login()方法

token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))

@staticmethod
def login(account: Account, *, ip_address: Optional[str] = None) -> TokenPair:
    if ip_address:
        AccountService.update_login_info(account=account, ip_address=ip_address)

    if account.status == AccountStatus.PENDING.value:
        account.status = AccountStatus.ACTIVE.value
        db.session.commit()

    access_token = AccountService.get_account_jwt_token(account=account)
    refresh_token = _generate_refresh_token()

    AccountService._store_refresh_token(refresh_token, account.id)

    return TokenPair(access_token=access_token, refresh_token=refresh_token)

（1）AccountService.get_account_jwt_token(account=account)

该方法主要是生成 jwt token 的过程，如下所示：

@staticmethod
def get_account_jwt_token(account: Account) -> str:
    exp_dt = datetime.now(timezone.utc) + timedelta(minutes=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES)
    exp = int(exp_dt.timestamp())
    payload = {
        "user_id": account.id,
        "exp": exp,
        "iss": dify_config.EDITION,
        "sub": "Console API Passport",
    }

    token = PassportService().issue(payload)  # jwt.encode(payload, self.sk, algorithm="HS256")
    return token

（2）_generate_refresh_token()

该方法主要是生成 refresh token 的过程，如下所示：

def _generate_refresh_token(length: int = 64):
    token = secrets.token_hex(length)  #  返回一个随机的文本字符串，使用十六进制表示。
    return token

（3）AccountService._store_refresh_token(refresh_token, account.id)

@staticmethod
def _store_refresh_token(refresh_token: str, account_id: str) -> None:
    redis_client.setex(AccountService._get_refresh_token_key(refresh_token), REFRESH_TOKEN_EXPIRY, account_id)
    redis_client.setex(
        AccountService._get_account_refresh_token_key(account_id), REFRESH_TOKEN_EXPIRY, refresh_token
    )

该方法的作用是将刷新令牌和账户 ID 以键值对的形式存储到 Redis 中，并设置过期时间，以便后续的身份验证和管理。当用户登录或需要刷新令牌时，可以快速查找这些信息。

3.AccountService.reset_login_error_rate_limit()方法

AccountService.reset_login_error_rate_limit(args[“email”])

@staticmethod
def reset_login_error_rate_limit(email: str):
    key = f"login_error_rate_limit:{email}"
    redis_client.delete(key)

该方法用于重置指定邮箱的登录错误次数限制，通过删除与该邮箱相关的 Redis 键。

三.console 蓝图与 Bearer Token 鉴权（用户登出操作）

1./console/api/logout 接口

http://localhost:5001/console/api/logout

Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyX2lkIjoiZWJiZjI1YmUtZWM4NS00ZDY4LTlkZDItNWJjMmVlMjg1NWU0IiwiZXhwIjoxNzMwNjg1NTcyLCJpc3MiOiJTRUxGX0hPU1RFRCIsInN1YiI6IkNvbnNvbGUgQVBJIFBhc3Nwb3J0In0.2e_5wPvf8GU9JARduOiIc06loBlQv0mgUrfkbMceeSg

2.AccountService.logout()

源码位置：dify\api\controllers\console\auth\login.py

AccountService.logout(account=account)该方法用于用户登出操作，首先尝试从 Redis 中获取用户的刷新令牌，如果存在，则删除该令牌，以实现用户登出的效果。

@staticmethod
def logout(*, account: Account) -> None:
    refresh_token = redis_client.get(AccountService._get_account_refresh_token_key(account.id))
    if refresh_token:
        AccountService._delete_refresh_token(refresh_token.decode("utf-8"), account.id)

使用 redis_client 从 Redis 数据库中获取与给定 account（账户）相关的刷新令牌（refresh token）。_get_account_refresh_token_key(account.id) 是一个静态方法，用于生成存储刷新令牌的键，account.id 是账户的唯一标识符。

如果获取到了刷新令牌，调用 AccountService 的 _delete_refresh_token 方法来删除该令牌。refresh_token.decode("utf-8") 将字节串解码为字符串，传入方法以执行删除操作，同时传入 account.id 以确保删除的是对应账户的令牌。

四.Web 蓝图与 Bearer Token 鉴权

1.Web 蓝图

简单理解，将应用发布后的运行 Web 页面调用接口 URL 前缀为/api，如下所示：

2.validate_jwt_token 装饰器

这里面的类基本都继承自 WebApiResource 类，而 WebApiResource 类实现，如下所示：

class WebApiResource(Resource):
    method_decorators = [validate_jwt_token]

validate_jwt_token 装饰器用于在调用视图函数之前验证 JWT。它会解码 JWT，提取出相关的用户信息，并将其传递给视图函数。如下所示：

def validate_jwt_token(view=None):
    def decorator(view):
        @wraps(view)
        def decorated(*args, **kwargs):
            app_model, end_user = decode_jwt_token()

            return view(app_model, end_user, *args, **kwargs)

        return decorated

    if view:
        return decorator(view)
    return decorator

参考文献

[1] Dify 中的 @login_required 装饰器：https://z0yrmerhgi8.feishu.cn/wiki/GaSqwbQRUimJjrkQV6pcV7TKnMb

NLP工程化(星球号)
欢迎加入我的知识星球，长按或扫码下方二维码。

星球号主要针对公众号提供增值服务，内容包括但不限于：源码解析及答疑(Dify等)，书籍|报告下载(电子资料)，代码|数据下载(公众号付费资料)。