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

from __future__ import annotations

from controllers.openapi.auth.conditions import (
    EDITION_ENTERPRISE,
    HAS_ALLOWED_ROLES,
    HAS_RBAC,
    LOADED_APP_IS_PRIVATE,
    PATH_HAS_APP_ID,
    WEBAPP_AUTH_ENABLED,
    WEBAPP_RUN_SCOPED,
    WORKSPACE_MEMBERSHIP_REQUIRED,
    WORKSPACE_SCOPED,
)
from controllers.openapi.auth.flow import When
from controllers.openapi.auth.pipeline import AuthPipeline, PipelineRoute, PipelineRouter
from controllers.openapi.auth.prepare import (
    load_account,
    load_app,
    load_app_access_mode,
    load_tenant,
    load_tenant_from_request,
    load_workspace_role,
    resolve_external_user,
)
from controllers.openapi.auth.verify import (
    check_acl,
    check_app_api_enabled,
    check_private_app_permission,
    check_rbac_permission,
    check_scope,
    check_workspace_member,
    check_workspace_mismatch,
    check_workspace_role,
)
from enums import DeploymentEdition
from libs.oauth_bearer import TokenType

account_pipeline = AuthPipeline(
    prepare=[
        When(PATH_HAS_APP_ID, then=load_app),
        When(PATH_HAS_APP_ID, then=load_tenant),
        When(WORKSPACE_MEMBERSHIP_REQUIRED, then=load_tenant_from_request),
        load_account,
        When(WORKSPACE_SCOPED, then=load_workspace_role),
        When(PATH_HAS_APP_ID & EDITION_ENTERPRISE, then=load_app_access_mode),
    ],
    auth=[
        When(PATH_HAS_APP_ID, then=check_app_api_enabled),
        check_scope,
        When(WORKSPACE_SCOPED, then=check_workspace_member),
        When(PATH_HAS_APP_ID, then=check_workspace_mismatch),
        When(HAS_ALLOWED_ROLES, then=check_workspace_role),
        When(HAS_RBAC, then=check_rbac_permission),
        When(PATH_HAS_APP_ID & EDITION_ENTERPRISE & WEBAPP_AUTH_ENABLED & WEBAPP_RUN_SCOPED, then=check_acl),
        When(EDITION_ENTERPRISE & LOADED_APP_IS_PRIVATE & WEBAPP_RUN_SCOPED, then=check_private_app_permission),
    ],
)

external_sso_pipeline = AuthPipeline(
    prepare=[
        When(PATH_HAS_APP_ID, then=load_app),
        When(PATH_HAS_APP_ID, then=load_tenant),
        When(PATH_HAS_APP_ID, then=resolve_external_user),
        When(PATH_HAS_APP_ID, then=load_app_access_mode),
    ],
    auth=[
        When(PATH_HAS_APP_ID, then=check_app_api_enabled),
        check_scope,
        When(PATH_HAS_APP_ID & WEBAPP_AUTH_ENABLED, then=check_acl),
        When(LOADED_APP_IS_PRIVATE, then=check_private_app_permission),
    ],
)

auth_router = PipelineRouter(
    {
        TokenType.OAUTH_ACCOUNT: PipelineRoute(account_pipeline),
        TokenType.OAUTH_EXTERNAL_SSO: PipelineRoute(
            external_sso_pipeline,
            required_edition=frozenset({DeploymentEdition.ENTERPRISE}),
        ),
    }
)