dify/api/controllers/openapi/auth/conditions.py

from __future__ import annotations

from collections.abc import Callable

from controllers.openapi.auth.data import AuthData, Edition, RequestContext, current_edition
from libs.oauth_bearer import TokenType
from services.enterprise.enterprise_service import WebAppAccessMode
from services.feature_service import FeatureService

CondFn = Callable[[RequestContext, AuthData | None], bool]


class Cond:
    def __init__(self, fn: CondFn) -> None:
        self._fn = fn

    def __call__(self, ctx: RequestContext, data: AuthData | None = None) -> bool:
        return self._fn(ctx, data)

    def __and__(self, other: Cond) -> Cond:
        return Cond(lambda ctx, data: self(ctx, data) and other(ctx, data))

    def __or__(self, other: Cond) -> Cond:
        return Cond(lambda ctx, data: self(ctx, data) or other(ctx, data))

    def __invert__(self) -> Cond:
        return Cond(lambda ctx, data: not self(ctx, data))


def request_cond(fn: Callable[[RequestContext], bool]) -> Cond:
    return Cond(lambda ctx, _: fn(ctx))


def data_cond(fn: Callable[[AuthData], bool]) -> Cond:
    return Cond(lambda _, data: data is not None and fn(data))


def config_cond(fn: Callable[[], bool]) -> Cond:
    return Cond(lambda _, __: fn())


TOKEN_IS_OAUTH_ACCOUNT = request_cond(lambda ctx: ctx.token_type == TokenType.OAUTH_ACCOUNT)
TOKEN_IS_OAUTH_EXTERNAL_SSO = request_cond(lambda ctx: ctx.token_type == TokenType.OAUTH_EXTERNAL_SSO)

PATH_HAS_APP_ID = request_cond(lambda ctx: "app_id" in ctx.path_params)

EDITION_CE = config_cond(lambda: current_edition() == Edition.CE)
EDITION_EE = config_cond(lambda: current_edition() == Edition.EE)
EDITION_SAAS = config_cond(lambda: current_edition() == Edition.SAAS)

WEBAPP_AUTH_ENABLED = config_cond(lambda: FeatureService.get_system_features().webapp_auth.enabled)

LOADED_APP_IS_PRIVATE = data_cond(lambda data: data.app_access_mode == WebAppAccessMode.PRIVATE)