dify/api/tests/unit_tests/controllers/openapi/auth/test_verify.py

import uuid
from unittest.mock import MagicMock, patch

import pytest
from flask import Flask
from werkzeug.exceptions import Forbidden, NotFound

from controllers.openapi.auth.data import AuthData
from controllers.openapi.auth.verify import (
    check_acl,
    check_app_access,
    check_app_api_enabled,
    check_private_app_permission,
    check_scope,
    check_workspace_member,
    check_workspace_mismatch,
    check_workspace_role,
)
from libs.oauth_bearer import Scope, TokenType
from models.account import Tenant, TenantAccountRole
from models.model import App
from services.enterprise.enterprise_service import WebAppAccessMode


def _data(**kwargs) -> AuthData:
    defaults: dict = {"token_type": TokenType.OAUTH_ACCOUNT, "token_hash": "hash", "scopes": frozenset({Scope.FULL})}
    defaults.update(kwargs)
    return AuthData(**defaults)


def test_check_scope_passes_when_required_is_none():
    check_scope(_data(required_scope=None))


def test_check_scope_passes_when_full_in_scopes():
    check_scope(_data(required_scope=Scope.APPS_RUN, scopes=frozenset({Scope.FULL})))


def test_check_scope_passes_when_exact_scope_present():
    check_scope(_data(required_scope=Scope.APPS_RUN, scopes=frozenset({Scope.APPS_RUN})))


def test_check_scope_raises_forbidden_when_scope_missing():
    with pytest.raises(Forbidden, match="insufficient_scope"):
        check_scope(_data(required_scope=Scope.APPS_RUN, scopes=frozenset({Scope.APPS_READ})))


def test_check_workspace_member_raises_not_found_when_no_role():
    with pytest.raises(NotFound, match="workspace not found"):
        check_workspace_member(_data(tenant_role=None))


def test_check_workspace_member_passes_when_role_present():
    check_workspace_member(_data(tenant_role=TenantAccountRole.NORMAL))


def test_check_app_access_passes_when_tenant_none():
    check_app_access(_data(tenant=None))


def test_check_app_access_passes_when_member():
    tenant = MagicMock(spec=Tenant)
    tenant.id = "t1"
    data = _data(account_id=uuid.uuid4(), tenant=tenant)
    with patch("controllers.openapi.auth.verify.TenantService.account_belongs_to_tenant", return_value=True):
        check_app_access(data)


def test_check_app_access_raises_when_not_member():
    tenant = MagicMock(spec=Tenant)
    tenant.id = "t1"
    data = _data(account_id=uuid.uuid4(), tenant=tenant)
    with patch("controllers.openapi.auth.verify.TenantService.account_belongs_to_tenant", return_value=False):
        with pytest.raises(Forbidden, match="subject_no_app_access"):
            check_app_access(data)


def test_check_acl_raises_when_app_or_mode_missing():
    with pytest.raises(Forbidden):
        check_acl(_data(app=None, app_access_mode=None))


def test_check_acl_account_allowed_for_public():
    app = MagicMock(spec=App)
    data = _data(token_type=TokenType.OAUTH_ACCOUNT, app=app, app_access_mode=WebAppAccessMode.PUBLIC)
    check_acl(data)


def test_check_acl_external_sso_blocked_for_private():
    app = MagicMock(spec=App)
    data = _data(
        token_type=TokenType.OAUTH_EXTERNAL_SSO,
        app=app,
        app_access_mode=WebAppAccessMode.PRIVATE,
    )
    with pytest.raises(Forbidden, match="subject_not_allowed_for_access_mode"):
        check_acl(data)


def test_check_acl_external_sso_allowed_for_sso_verified():
    app = MagicMock(spec=App)
    data = _data(
        token_type=TokenType.OAUTH_EXTERNAL_SSO,
        app=app,
        app_access_mode=WebAppAccessMode.SSO_VERIFIED,
    )
    check_acl(data)


def test_check_private_app_permission_raises_when_app_none():
    with pytest.raises(Forbidden):
        check_private_app_permission(_data(app=None))


def test_check_private_app_permission_raises_when_user_not_allowed():
    app = MagicMock(spec=App)
    app.id = "app-1"
    data = _data(account_id=uuid.uuid4(), app=app)
    target = "controllers.openapi.auth.verify.EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp"
    with patch(target, return_value=False):
        with pytest.raises(Forbidden, match="user_not_allowed_for_private_app"):
            check_private_app_permission(data)


def test_check_private_app_permission_passes_when_allowed():
    app = MagicMock(spec=App)
    app.id = "app-1"
    data = _data(account_id=uuid.uuid4(), app=app)
    target = "controllers.openapi.auth.verify.EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp"
    with patch(target, return_value=True):
        check_private_app_permission(data)


# --- check_workspace_mismatch ---


@pytest.fixture
def flask_app():
    return Flask(__name__)


def test_check_workspace_mismatch_passes_when_tenant_none(flask_app):
    with flask_app.test_request_context("/test"):
        check_workspace_mismatch(_data(tenant=None))


def test_check_workspace_mismatch_passes_when_ids_match(flask_app):
    tenant = MagicMock(spec=Tenant)
    tid = uuid.uuid4()
    tenant.id = tid
    with flask_app.test_request_context(f"/test?workspace_id={tid}"):
        check_workspace_mismatch(_data(tenant=tenant, path_params={}))


def test_check_workspace_mismatch_raises_422_on_mismatch(flask_app):
    from werkzeug.exceptions import UnprocessableEntity

    tenant = MagicMock(spec=Tenant)
    tenant.id = uuid.uuid4()
    other_id = uuid.uuid4()
    with flask_app.test_request_context(f"/test?workspace_id={other_id}"):
        with pytest.raises(UnprocessableEntity):
            check_workspace_mismatch(_data(tenant=tenant, path_params={}))


def test_check_workspace_mismatch_passes_when_no_request_workspace_id(flask_app):
    tenant = MagicMock(spec=Tenant)
    tenant.id = uuid.uuid4()
    with flask_app.test_request_context("/test"):
        check_workspace_mismatch(_data(tenant=tenant, path_params={}))


# --- check_workspace_role ---


def test_check_workspace_role_passes_when_allowed_roles_none():
    check_workspace_role(_data(allowed_roles=None))


def test_check_workspace_role_raises_not_found_when_not_member():
    data = _data(tenant_role=None, allowed_roles=frozenset({TenantAccountRole.ADMIN}))
    with pytest.raises(NotFound):
        check_workspace_role(data)


def test_check_workspace_role_raises_forbidden_when_wrong_role():
    data = _data(
        tenant_role=TenantAccountRole.EDITOR,
        allowed_roles=frozenset({TenantAccountRole.OWNER}),
    )
    with pytest.raises(Forbidden, match="insufficient workspace role"):
        check_workspace_role(data)


def test_check_workspace_role_passes_when_role_allowed():
    data = _data(
        tenant_role=TenantAccountRole.ADMIN,
        allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN}),
    )
    check_workspace_role(data)


# --- check_app_api_enabled ---


def test_check_app_api_enabled_passes_when_enabled():
    app = MagicMock(spec=App)
    app.enable_api = True
    check_app_api_enabled(_data(app=app))


def test_check_app_api_enabled_raises_forbidden_when_disabled():
    app = MagicMock(spec=App)
    app.enable_api = False
    with pytest.raises(Forbidden, match="service_api_disabled"):
        check_app_api_enabled(_data(app=app))


def test_check_app_api_enabled_passes_when_app_none():
    check_app_api_enabled(_data(app=None))