diff --git a/api/.importlinter b/api/.importlinter index d9796d56504..9a0005a5703 100644 --- a/api/.importlinter +++ b/api/.importlinter @@ -171,6 +171,21 @@ forbidden_modules = sqlalchemy werkzeug +[importlinter:contract:network-access-group-service-boundary] +name = Network access group application service is framework and persistence neutral +type = forbidden +source_modules = + services.network_access_group_service +forbidden_modules = + configs + controllers + extensions + flask + models + repositories + sqlalchemy + werkzeug + [importlinter:contract:init-validation-service-boundary] name = Initialization validation application service is framework and persistence neutral type = forbidden diff --git a/api/controllers/console/workspace/network_access_group.py b/api/controllers/console/workspace/network_access_group.py index 1922a67327b..6d5ba151c52 100644 --- a/api/controllers/console/workspace/network_access_group.py +++ b/api/controllers/console/workspace/network_access_group.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Literal, Self +from typing import Any, Self from uuid import UUID from flask_restx import Resource @@ -8,34 +8,24 @@ from werkzeug.exceptions import BadGateway, BadRequest, Conflict, Forbidden, Not from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models from controllers.console import console_ns -from controllers.console.app.wraps import get_app_model -from controllers.console.wraps import ( - account_initialization_required, - cloud_edition_billing_paid_plan_required, - is_cloud_edition_billing_paid_plan, - model_validate, - only_edition_cloud, - setup_required, - with_current_tenant_id, - with_current_user, -) +from controllers.console.app.error import AppNotFoundError +from controllers.console.flask_admission import console_account_admission +from controllers.console.wraps import cloud_edition_billing_paid_plan_required, model_validate +from enums import DeploymentEdition +from extensions.ext_application_services import application_services from fields.base import ResponseModel from libs.helper import dump_response -from libs.login import login_required -from models import Account, App, AppMode, TenantAccountRole -from services.billing_service import BillingService, NetworkAccessGroupUpstreamError -from services.network_access_group_service import NetworkAccessGroupService - -NetworkAccessPoint = Literal["webapp", "service_api", "mcp", "trigger"] - -_ACCESS_POINTS_BY_APP_MODE: dict[AppMode, tuple[NetworkAccessPoint, ...]] = { - AppMode.WORKFLOW: ("webapp", "service_api", "mcp", "trigger"), - AppMode.ADVANCED_CHAT: ("webapp", "service_api", "mcp"), - AppMode.CHAT: ("webapp", "service_api", "mcp"), - AppMode.COMPLETION: ("webapp", "service_api", "mcp"), - AppMode.AGENT_CHAT: ("webapp", "service_api", "mcp"), - AppMode.AGENT: ("webapp", "service_api"), -} +from machinery.context import RequestContext +from services.network_access_group_service import ( + NetworkAccessGroupAccessDeniedError, + NetworkAccessGroupAppNotFoundError, + NetworkAccessGroupEntitlementUnavailableError, + NetworkAccessGroupError, + NetworkAccessGroupUnsupportedAccessPointsError, + NetworkAccessGroupUnsupportedAppModeError, + NetworkAccessGroupUpstreamError, + NetworkAccessPoint, +) class NetworkAccessGroupCreatePayload(BaseModel): @@ -184,59 +174,6 @@ register_response_schema_models( ) -def _available_access_points(app_model: App) -> list[NetworkAccessPoint]: - app_mode = app_model.mode if isinstance(app_model.mode, AppMode) else AppMode.value_of(str(app_model.mode)) - available = _ACCESS_POINTS_BY_APP_MODE.get(app_mode) - if available is None: - raise BadRequest(f"Network access control is not supported for app mode '{app_mode.value}'.") - return list(available) - - -def _validate_app_access_points( - requested: list[NetworkAccessPoint], - available: list[NetworkAccessPoint], -) -> None: - unsupported = sorted(set(requested).difference(available)) - if unsupported: - raise BadRequest(f"Unsupported access points for this app: {', '.join(unsupported)}.") - - -def _effective_entitlement(tenant_id: str, upstream_entitled: object) -> bool: - return bool(upstream_entitled) and is_cloud_edition_billing_paid_plan(tenant_id) - - -def _normalize_app_binding_defaults(payload: dict) -> None: - """Materialize ProtoJSON defaults so the Console response stays stable.""" - - payload.setdefault("binding", None) - binding = payload.get("binding") - if not isinstance(binding, dict): - return - binding.setdefault("enabled", False) - if "access_points" not in binding and "accessPoints" not in binding: - binding["access_points"] = [] - - -def _limit_binding_access_points(payload: dict, available: list[NetworkAccessPoint]) -> None: - """Hide stale scopes which are not real access points for the current App mode.""" - - binding = payload.get("binding") - if not isinstance(binding, dict): - return - raw_access_points = binding.get("access_points", binding.get("accessPoints", [])) - if not isinstance(raw_access_points, list): - return - binding["access_points"] = [access_point for access_point in raw_access_points if access_point in available] - binding.pop("accessPoints", None) - - -def _ensure_workspace_admin_or_owner(current_user: Account) -> None: - """Authorize management against the persisted current-workspace membership.""" - - if not TenantAccountRole.is_privileged_role(current_user.current_role): - raise Forbidden("Only workspace owners and administrators can manage network access groups.") - - def _translate_upstream_error(exc: NetworkAccessGroupUpstreamError) -> Exception: if exc.reason == "INVALID_SECRET_KEY": return ServiceUnavailable("Network access group service authentication failed.") @@ -260,7 +197,23 @@ def _translate_upstream_error(exc: NetworkAccessGroupUpstreamError) -> Exception return BadGateway("Unexpected response from the network access group service.") -def _serialize_response(response_model: type[ResponseModel], payload: dict): +def _translate_service_error(exc: NetworkAccessGroupError) -> Exception: + if isinstance(exc, NetworkAccessGroupUpstreamError): + return _translate_upstream_error(exc) + if isinstance(exc, NetworkAccessGroupAccessDeniedError): + return Forbidden("Only workspace owners and administrators can manage network access groups.") + if isinstance(exc, NetworkAccessGroupEntitlementUnavailableError): + return ServiceUnavailable("Billing entitlement is temporarily unavailable.") + if isinstance(exc, NetworkAccessGroupAppNotFoundError): + return AppNotFoundError() + if isinstance(exc, NetworkAccessGroupUnsupportedAppModeError): + return BadRequest(f"Network access control is not supported for app mode '{exc.app_mode}'.") + if isinstance(exc, NetworkAccessGroupUnsupportedAccessPointsError): + return BadRequest(f"Unsupported access points for this app: {', '.join(exc.access_points)}.") + raise TypeError(f"Unsupported network access group error: {type(exc).__name__}") + + +def _serialize_response(response_model: type[ResponseModel], payload: dict[str, Any]) -> dict[str, Any]: try: return dump_response(response_model, payload) except ValidationError as exc: @@ -274,20 +227,12 @@ class CurrentWorkspaceNetworkAccessGroupsApi(Resource): "Workspace network access groups retrieved successfully", console_ns.models[NetworkAccessGroupListResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud - @with_current_user - @with_current_tenant_id - def get(self, current_tenant_id: str, current_user: Account): - _ensure_workspace_admin_or_owner(current_user) + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) + def get(self, request_context: RequestContext) -> dict[str, Any]: try: - payload = BillingService.list_network_access_groups(current_tenant_id, current_user.id) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - payload["entitled"] = _effective_entitlement(current_tenant_id, payload.get("entitled")) - NetworkAccessGroupService.enrich_app_references(payload, current_tenant_id) + payload = application_services().network_access_groups.list_groups(request_context) + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(NetworkAccessGroupListResponse, payload) @console_ns.expect(console_ns.models[NetworkAccessGroupCreatePayload.__name__]) @@ -296,32 +241,23 @@ class CurrentWorkspaceNetworkAccessGroupsApi(Resource): "Workspace network access group created successfully", console_ns.models[NetworkAccessGroupMutationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) @cloud_edition_billing_paid_plan_required - @with_current_user - @with_current_tenant_id @model_validate(NetworkAccessGroupCreatePayload) def post( self, req_data: NetworkAccessGroupCreatePayload, - current_tenant_id: str, - current_user: Account, - ): - _ensure_workspace_admin_or_owner(current_user) + request_context: RequestContext, + ) -> tuple[dict[str, Any], int]: try: - payload = BillingService.create_network_access_group( - current_tenant_id, + payload = application_services().network_access_groups.create_group( + request_context, name=req_data.name, description=req_data.description, allowed_cidrs=req_data.allowed_cidrs, - actor_account_id=current_user.id, ) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - NetworkAccessGroupService.enrich_app_references(payload, current_tenant_id) + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(NetworkAccessGroupMutationResponse, payload), 201 @@ -332,19 +268,15 @@ class CurrentWorkspaceNetworkAccessGroupApi(Resource): "Workspace network access group retrieved successfully", console_ns.models[NetworkAccessGroupMutationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud - @with_current_user - @with_current_tenant_id - def get(self, current_tenant_id: str, current_user: Account, group_id: UUID): - _ensure_workspace_admin_or_owner(current_user) + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) + def get(self, request_context: RequestContext, group_id: UUID) -> dict[str, Any]: try: - payload = BillingService.get_network_access_group(current_tenant_id, str(group_id), current_user.id) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - NetworkAccessGroupService.enrich_app_references(payload, current_tenant_id) + payload = application_services().network_access_groups.get_group( + request_context, + group_id=str(group_id), + ) + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(NetworkAccessGroupMutationResponse, payload) @console_ns.expect(console_ns.models[NetworkAccessGroupUpdatePayload.__name__]) @@ -353,35 +285,26 @@ class CurrentWorkspaceNetworkAccessGroupApi(Resource): "Workspace network access group updated successfully", console_ns.models[NetworkAccessGroupMutationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) @cloud_edition_billing_paid_plan_required - @with_current_user - @with_current_tenant_id @model_validate(NetworkAccessGroupUpdatePayload) def put( self, req_data: NetworkAccessGroupUpdatePayload, - current_tenant_id: str, - current_user: Account, + request_context: RequestContext, group_id: UUID, - ): - _ensure_workspace_admin_or_owner(current_user) + ) -> dict[str, Any]: try: - payload = BillingService.update_network_access_group( - current_tenant_id, - str(group_id), + payload = application_services().network_access_groups.update_group( + request_context, + group_id=str(group_id), name=req_data.name, description=req_data.description, allowed_cidrs=req_data.allowed_cidrs, expected_version=req_data.expected_version, - actor_account_id=current_user.id, ) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - NetworkAccessGroupService.enrich_app_references(payload, current_tenant_id) + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(NetworkAccessGroupMutationResponse, payload) @console_ns.doc(params=query_params_from_model(NetworkAccessGroupDeleteQuery)) @@ -390,31 +313,23 @@ class CurrentWorkspaceNetworkAccessGroupApi(Resource): "Workspace network access group deleted successfully", console_ns.models[NetworkAccessGroupDeleteResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) @cloud_edition_billing_paid_plan_required - @with_current_user - @with_current_tenant_id @model_validate(NetworkAccessGroupDeleteQuery) def delete( self, req_data: NetworkAccessGroupDeleteQuery, - current_tenant_id: str, - current_user: Account, + request_context: RequestContext, group_id: UUID, - ): - _ensure_workspace_admin_or_owner(current_user) + ) -> dict[str, Any]: try: - payload = BillingService.delete_network_access_group( - current_tenant_id, - str(group_id), + payload = application_services().network_access_groups.delete_group( + request_context, + group_id=str(group_id), expected_version=req_data.expected_version, - actor_account_id=current_user.id, ) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(NetworkAccessGroupDeleteResponse, payload) @@ -425,28 +340,15 @@ class AppNetworkAccessGroupApi(Resource): "App network access group binding retrieved successfully", console_ns.models[AppNetworkAccessGroupResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud - @with_current_user - @with_current_tenant_id - @get_app_model - def get(self, current_tenant_id: str, current_user: Account, app_model: App): - _ensure_workspace_admin_or_owner(current_user) - available_access_points = _available_access_points(app_model) + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) + def get(self, request_context: RequestContext, app_id: UUID) -> dict[str, Any]: try: - payload = BillingService.get_app_network_access_group( - current_tenant_id, - str(app_model.id), - current_user.id, + payload = application_services().network_access_groups.get_app_binding( + request_context, + app_id=str(app_id), ) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - _normalize_app_binding_defaults(payload) - _limit_binding_access_points(payload, available_access_points) - payload["entitled"] = _effective_entitlement(current_tenant_id, payload.get("entitled")) - payload["available_access_points"] = available_access_points + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(AppNetworkAccessGroupResponse, payload) @console_ns.expect(console_ns.models[AppNetworkAccessGroupUpdatePayload.__name__]) @@ -455,38 +357,24 @@ class AppNetworkAccessGroupApi(Resource): "App network access group binding updated successfully", console_ns.models[AppNetworkAccessGroupMutationResponse.__name__], ) - @setup_required - @login_required - @account_initialization_required - @only_edition_cloud + @console_account_admission(editions=frozenset({DeploymentEdition.CLOUD})) @cloud_edition_billing_paid_plan_required - @with_current_user - @with_current_tenant_id - @get_app_model @model_validate(AppNetworkAccessGroupUpdatePayload) def put( self, req_data: AppNetworkAccessGroupUpdatePayload, - current_tenant_id: str, - current_user: Account, - app_model: App, - ): - _ensure_workspace_admin_or_owner(current_user) - available_access_points = _available_access_points(app_model) - _validate_app_access_points(req_data.access_points, available_access_points) + request_context: RequestContext, + app_id: UUID, + ) -> dict[str, Any]: try: - payload = BillingService.update_app_network_access_group( - current_tenant_id, - str(app_model.id), + payload = application_services().network_access_groups.update_app_binding( + request_context, + app_id=str(app_id), enabled=req_data.enabled, group_id=str(req_data.group_id) if req_data.group_id is not None else None, access_points=list(req_data.access_points), expected_version=req_data.expected_version, - actor_account_id=current_user.id, ) - except NetworkAccessGroupUpstreamError as exc: - raise _translate_upstream_error(exc) from exc - _normalize_app_binding_defaults(payload) - _limit_binding_access_points(payload, available_access_points) - payload["available_access_points"] = available_access_points + except NetworkAccessGroupError as exc: + raise _translate_service_error(exc) from exc return _serialize_response(AppNetworkAccessGroupMutationResponse, payload) diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py index 07aa6edc21e..e0099f82171 100644 --- a/api/extensions/ext_application_services.py +++ b/api/extensions/ext_application_services.py @@ -50,6 +50,7 @@ from repositories.file_grant_repository import FileGrantRepository from repositories.human_input_file_upload_repository import SQLAlchemyHumanInputFileUploadRepository from repositories.installation_state_repository import InstallationStateRepository from repositories.message_file_preview_repository import MessageFilePreviewQueryRepository +from repositories.network_access_group_repository import SQLAlchemyNetworkAccessGroupAppRepository from repositories.oauth_access_token_repository import SQLAlchemyOAuthAccessTokenRepository from repositories.oauth_server_repository import RedisOAuthServerTokenRepository, SQLAlchemyOAuthServerRepository from repositories.plugin_file_upload_repository import SQLAlchemyPluginFileUploadOwnerRepository @@ -163,6 +164,11 @@ from services.human_input_file_upload_service import HumanInputFileUploadService from services.init_validation_service import InitValidationService from services.inner_mail_service import InnerMailService from services.message_file_preview_service import MessageFilePreviewService +from services.network_access_group_gateway import ( + BillingNetworkAccessGroupEntitlementGateway, + NetworkAccessGroupGateway, +) +from services.network_access_group_service import NetworkAccessGroupService from services.notification_gateway import BillingNotificationGateway from services.notification_service import NotificationService from services.notion_data_source_gateway import NotionDataSourceGateway @@ -278,6 +284,7 @@ class ApplicationServices: oauth_server: OAuthServerService init_validation: InitValidationService notifications: NotificationService + network_access_groups: NetworkAccessGroupService step_by_step_tour: StepByStepTourService partner_tenant_bindings: PartnerTenantBindingService recommended_app_queries: RecommendedAppQueryService @@ -696,6 +703,16 @@ def build_application_services( notifications=NotificationService( notifications=BillingNotificationGateway(), ), + network_access_groups=NetworkAccessGroupService( + control_plane=NetworkAccessGroupGateway( + base_url=dify_config.NETWORK_ACCESS_API_URL, + fallback_base_url=BillingService.base_url, + secret_key=BillingService.secret_key, + ), + apps=SQLAlchemyNetworkAccessGroupAppRepository(session_factory=database_client), + memberships=workspace_query_repository, + entitlement=BillingNetworkAccessGroupEntitlementGateway(), + ), step_by_step_tour=StepByStepTourService( accounts=accounts, states=SQLAlchemyStepByStepTourStateRepository(session_factory=database_client), diff --git a/api/repositories/network_access_group_repository.py b/api/repositories/network_access_group_repository.py new file mode 100644 index 00000000000..942b7987307 --- /dev/null +++ b/api/repositories/network_access_group_repository.py @@ -0,0 +1,76 @@ +"""SQLAlchemy read repository for network access group App projections.""" + +from collections.abc import Sequence +from typing import override + +from sqlalchemy import select +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session, sessionmaker + +from models import App +from models.agent import AgentScope +from services.network_access_group_service import ( + NetworkAccessGroupAppQuery, + NetworkAccessGroupAppQueryError, + NetworkAccessGroupAppRecord, +) + + +class SQLAlchemyNetworkAccessGroupAppRepository(NetworkAccessGroupAppQuery): + def __init__(self, *, session_factory: sessionmaker[Session]) -> None: + self._session_factory = session_factory + + @override + def get_manageable_app(self, *, workspace_id: str, app_id: str) -> NetworkAccessGroupAppRecord | None: + try: + with self._session_factory() as session: + app = session.scalar( + select(App) + .where( + App.id == app_id, + App.tenant_id == workspace_id, + App.status == "normal", + ) + .limit(1) + ) + if app is None: + return None + binding = app.agent_app_binding_with_session(session=session, include_archived=True) + if binding is not None and binding.scope == AgentScope.WORKFLOW_ONLY: + return None + return self._to_record(app) + except SQLAlchemyError as exc: + raise NetworkAccessGroupAppQueryError from exc + + @override + def list_apps( + self, + *, + workspace_id: str, + app_ids: Sequence[str], + ) -> tuple[NetworkAccessGroupAppRecord, ...]: + if not app_ids: + return () + try: + with self._session_factory() as session: + apps = session.scalars( + select(App).where( + App.tenant_id == workspace_id, + App.id.in_(app_ids), + App.status == "normal", + ) + ).all() + return tuple(self._to_record(app) for app in apps) + except SQLAlchemyError as exc: + raise NetworkAccessGroupAppQueryError from exc + + @staticmethod + def _to_record(app: App) -> NetworkAccessGroupAppRecord: + return NetworkAccessGroupAppRecord( + id=str(app.id), + mode=app.mode.value, + name=app.name, + icon=app.icon, + icon_type=app.icon_type.value if app.icon_type is not None else None, + icon_background=app.icon_background, + ) diff --git a/api/repositories/workspace_query_repository.py b/api/repositories/workspace_query_repository.py index 8e83d158c47..f31af9c7acd 100644 --- a/api/repositories/workspace_query_repository.py +++ b/api/repositories/workspace_query_repository.py @@ -58,6 +58,15 @@ class WorkspaceQueryRepository( with self._session_factory() as session: return tuple(session.scalars(stmt).all()) + def get_role_for_account(self, *, workspace_id: str, account_id: str) -> str | None: + stmt = select(TenantAccountJoin.role).where( + TenantAccountJoin.tenant_id == workspace_id, + TenantAccountJoin.account_id == account_id, + ) + with self._session_factory() as session: + role = session.scalar(stmt) + return role.value if role is not None else None + @override def list_account_access_workspaces(self, account_id: str) -> tuple[AccountWorkspaceSnapshot, ...]: """List every membership for the OpenAPI account identity response. diff --git a/api/services/billing_service.py b/api/services/billing_service.py index 79cdc89df5f..c16f5b5142c 100644 --- a/api/services/billing_service.py +++ b/api/services/billing_service.py @@ -10,7 +10,6 @@ from tenacity import retry, retry_if_exception_type, stop_before_delay, wait_fix from typing_extensions import deprecated from werkzeug.exceptions import InternalServerError -from configs import dify_config from core.helper.http_client_pooling import get_pooled_http_client from enums import CloudPlan from extensions.ext_redis import redis_client @@ -216,15 +215,6 @@ class DismissNotificationDict(TypedDict): success: bool -class NetworkAccessGroupUpstreamError(Exception): - """Failure returned by the SaaS network-access-group control plane.""" - - def __init__(self, status_code: int, reason: str | None = None): - self.status_code = status_code - self.reason = reason - super().__init__(f"network access group upstream returned HTTP {status_code}") - - class BillingService: base_url = os.environ.get("BILLING_API_URL", "BILLING_API_URL") quota_base_url = os.environ.get("BILLING_QUOTA_API_URL") or base_url @@ -443,221 +433,6 @@ class BillingService: params = {"tenant_id": tenant_id, "feature_key": feature_key} return cls._send_request("GET", "/billing/tenant_feature_plan/usage", params=params) - @classmethod - def list_network_access_groups(cls, tenant_id: str, actor_account_id: str) -> dict[str, Any]: - """List the reusable network access groups owned by one workspace.""" - - return cls._send_network_access_group_request( - "GET", - f"/tenants/{tenant_id}/network-access-groups", - params={"actor_account_id": actor_account_id}, - ) - - @classmethod - def create_network_access_group( - cls, - tenant_id: str, - *, - name: str, - description: str, - allowed_cidrs: list[str], - actor_account_id: str, - ) -> dict[str, Any]: - """Create a reusable group; the Console BFF injects the authenticated actor.""" - - return cls._send_network_access_group_request( - "POST", - f"/tenants/{tenant_id}/network-access-groups", - payload_json={ - "name": name, - "description": description, - "allowed_cidrs": allowed_cidrs, - "actor_account_id": actor_account_id, - }, - ) - - @classmethod - def get_network_access_group( - cls, - tenant_id: str, - group_id: str, - actor_account_id: str, - ) -> dict[str, Any]: - """Fetch one reusable group scoped to its owning workspace.""" - - return cls._send_network_access_group_request( - "GET", - f"/tenants/{tenant_id}/network-access-groups/{group_id}", - params={"actor_account_id": actor_account_id}, - ) - - @classmethod - def update_network_access_group( - cls, - tenant_id: str, - group_id: str, - *, - name: str, - description: str, - allowed_cidrs: list[str], - expected_version: int, - actor_account_id: str, - ) -> dict[str, Any]: - """Replace a reusable group using optimistic concurrency control.""" - - return cls._send_network_access_group_request( - "PUT", - f"/tenants/{tenant_id}/network-access-groups/{group_id}", - payload_json={ - "name": name, - "description": description, - "allowed_cidrs": allowed_cidrs, - "expected_version": expected_version, - "actor_account_id": actor_account_id, - }, - ) - - @classmethod - def delete_network_access_group( - cls, - tenant_id: str, - group_id: str, - *, - expected_version: int, - actor_account_id: str, - ) -> dict[str, Any]: - """Delete a reusable group using optimistic concurrency control.""" - - return cls._send_network_access_group_request( - "DELETE", - f"/tenants/{tenant_id}/network-access-groups/{group_id}", - params={ - "expected_version": expected_version, - "actor_account_id": actor_account_id, - }, - ) - - @classmethod - def get_app_network_access_group( - cls, - tenant_id: str, - app_id: str, - actor_account_id: str, - ) -> dict[str, Any]: - """Fetch the reusable group currently assigned to one tenant-scoped app.""" - - return cls._send_network_access_group_request( - "GET", - f"/tenants/{tenant_id}/apps/{app_id}/network-access-group", - params={"actor_account_id": actor_account_id}, - ) - - @classmethod - def update_app_network_access_group( - cls, - tenant_id: str, - app_id: str, - *, - enabled: bool, - group_id: str | None, - access_points: list[str], - expected_version: int, - actor_account_id: str, - ) -> dict[str, Any]: - """Atomically replace one tenant-scoped App's complete network access configuration.""" - - return cls._send_network_access_group_request( - "PUT", - f"/tenants/{tenant_id}/apps/{app_id}/network-access-group", - payload_json={ - "enabled": enabled, - "group_id": group_id, - "access_points": access_points, - "expected_version": expected_version, - "actor_account_id": actor_account_id, - }, - ) - - @classmethod - def cleanup_app_network_access_group_binding(cls, tenant_id: str, app_id: str) -> dict[str, Any]: - """Remove an App binding through the internal App lifecycle endpoint. - - This must only be called by the post-commit App deletion task. Account - or workspace-member deletion does not own App bindings and must not use - this operation. - """ - - return cls._send_network_access_group_request( - "DELETE", - f"/tenants/{tenant_id}/apps/{app_id}/network-access-group-binding", - ) - - @classmethod - def _send_network_access_group_request( - cls, - method: Literal["GET", "POST", "PUT", "DELETE"], - endpoint: str, - *, - payload_json: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - ) -> dict[str, Any]: - """Call the SaaS group API while preserving its status for the Console boundary.""" - - try: - response = cls._send_network_access_group_http_request( - method, - endpoint, - payload_json=payload_json, - params=params, - ) - except httpx.RequestError as exc: - raise NetworkAccessGroupUpstreamError(httpx.codes.SERVICE_UNAVAILABLE) from exc - - if response.status_code not in (httpx.codes.OK, httpx.codes.CREATED): - reason = None - try: - error_payload = response.json() - if isinstance(error_payload, dict) and isinstance(error_payload.get("reason"), str): - reason = error_payload["reason"] - except (json.JSONDecodeError, TypeError, ValueError): - pass - raise NetworkAccessGroupUpstreamError(response.status_code, reason) - - try: - payload = response.json() - except (json.JSONDecodeError, TypeError, ValueError) as exc: - raise NetworkAccessGroupUpstreamError(httpx.codes.BAD_GATEWAY) from exc - - if not isinstance(payload, dict): - raise NetworkAccessGroupUpstreamError(httpx.codes.BAD_GATEWAY) - return payload - - @classmethod - @retry( - wait=wait_fixed(2), - stop=stop_before_delay(10), - retry=retry_if_exception_type(httpx.RequestError), - reraise=True, - ) - def _send_network_access_group_http_request( - cls, - method: Literal["GET", "POST", "PUT", "DELETE"], - endpoint: str, - *, - payload_json: dict[str, Any] | None = None, - params: dict[str, Any] | None = None, - ) -> httpx.Response: - headers = {"Content-Type": "application/json", "Billing-Api-Secret-Key": cls.secret_key} - network_access_base_url = dify_config.NETWORK_ACCESS_API_URL.strip() or cls.base_url - return _http_client.request( - method, - f"{network_access_base_url.rstrip('/')}{endpoint}", - json=payload_json, - params=params, - headers=headers, - follow_redirects=True, - ) - @classmethod def _send_quota_request( cls, method: Literal["GET", "POST", "DELETE", "PUT"], endpoint: str, json=None, params=None diff --git a/api/services/network_access_group_gateway.py b/api/services/network_access_group_gateway.py new file mode 100644 index 00000000000..af45b25e3fb --- /dev/null +++ b/api/services/network_access_group_gateway.py @@ -0,0 +1,235 @@ +"""HTTP and Billing adapters used by the network access group service.""" + +import json +from typing import Any, Literal, override + +import httpx +from pydantic import ValidationError +from tenacity import retry, retry_if_exception_type, stop_before_delay, wait_fixed + +from core.helper.http_client_pooling import get_pooled_http_client +from enums import CloudPlan +from services.billing_service import BillingService, _BillingHTTPStatusError +from services.network_access_group_service import ( + NetworkAccessGroupControlPlane, + NetworkAccessGroupEntitlement, + NetworkAccessGroupEntitlementUnavailableError, + NetworkAccessGroupUpstreamError, +) + +_http_client: httpx.Client = get_pooled_http_client( + "network-access-group:default", + lambda: httpx.Client( + timeout=httpx.Timeout(30.0, connect=5.0), + limits=httpx.Limits(max_keepalive_connections=50, max_connections=100), + ), +) + + +class NetworkAccessGroupGateway(NetworkAccessGroupControlPlane): + def __init__( + self, + *, + base_url: str, + fallback_base_url: str, + secret_key: str, + http_client: httpx.Client = _http_client, + ) -> None: + self._base_url = base_url.strip() or fallback_base_url + self._secret_key = secret_key + self._http_client = http_client + + @override + def list_groups(self, workspace_id: str, actor_account_id: str) -> dict[str, Any]: + return self._send_request( + "GET", + f"/tenants/{workspace_id}/network-access-groups", + params={"actor_account_id": actor_account_id}, + ) + + @override + def create_group( + self, + workspace_id: str, + *, + name: str, + description: str, + allowed_cidrs: list[str], + actor_account_id: str, + ) -> dict[str, Any]: + return self._send_request( + "POST", + f"/tenants/{workspace_id}/network-access-groups", + payload_json={ + "name": name, + "description": description, + "allowed_cidrs": allowed_cidrs, + "actor_account_id": actor_account_id, + }, + ) + + @override + def get_group(self, workspace_id: str, group_id: str, actor_account_id: str) -> dict[str, Any]: + return self._send_request( + "GET", + f"/tenants/{workspace_id}/network-access-groups/{group_id}", + params={"actor_account_id": actor_account_id}, + ) + + @override + def update_group( + self, + workspace_id: str, + group_id: str, + *, + name: str, + description: str, + allowed_cidrs: list[str], + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: + return self._send_request( + "PUT", + f"/tenants/{workspace_id}/network-access-groups/{group_id}", + payload_json={ + "name": name, + "description": description, + "allowed_cidrs": allowed_cidrs, + "expected_version": expected_version, + "actor_account_id": actor_account_id, + }, + ) + + @override + def delete_group( + self, + workspace_id: str, + group_id: str, + *, + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: + return self._send_request( + "DELETE", + f"/tenants/{workspace_id}/network-access-groups/{group_id}", + params={ + "expected_version": expected_version, + "actor_account_id": actor_account_id, + }, + ) + + @override + def get_app_binding(self, workspace_id: str, app_id: str, actor_account_id: str) -> dict[str, Any]: + return self._send_request( + "GET", + f"/tenants/{workspace_id}/apps/{app_id}/network-access-group", + params={"actor_account_id": actor_account_id}, + ) + + @override + def update_app_binding( + self, + workspace_id: str, + app_id: str, + *, + enabled: bool, + group_id: str | None, + access_points: list[str], + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: + return self._send_request( + "PUT", + f"/tenants/{workspace_id}/apps/{app_id}/network-access-group", + payload_json={ + "enabled": enabled, + "group_id": group_id, + "access_points": access_points, + "expected_version": expected_version, + "actor_account_id": actor_account_id, + }, + ) + + @override + def cleanup_app_binding(self, workspace_id: str, app_id: str) -> dict[str, Any]: + return self._send_request( + "DELETE", + f"/tenants/{workspace_id}/apps/{app_id}/network-access-group-binding", + ) + + def _send_request( + self, + method: Literal["GET", "POST", "PUT", "DELETE"], + endpoint: str, + *, + payload_json: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + try: + response = self._send_http_request( + method, + endpoint, + payload_json=payload_json, + params=params, + ) + except httpx.RequestError as exc: + raise NetworkAccessGroupUpstreamError(httpx.codes.SERVICE_UNAVAILABLE) from exc + + if response.status_code not in (httpx.codes.OK, httpx.codes.CREATED): + reason = None + try: + error_payload = response.json() + if isinstance(error_payload, dict) and isinstance(error_payload.get("reason"), str): + reason = error_payload["reason"] + except (json.JSONDecodeError, TypeError, ValueError): + pass + raise NetworkAccessGroupUpstreamError(response.status_code, reason) + + try: + payload = response.json() + except (json.JSONDecodeError, TypeError, ValueError) as exc: + raise NetworkAccessGroupUpstreamError(httpx.codes.BAD_GATEWAY) from exc + + if not isinstance(payload, dict): + raise NetworkAccessGroupUpstreamError(httpx.codes.BAD_GATEWAY) + return payload + + @retry( + wait=wait_fixed(2), + stop=stop_before_delay(10), + retry=retry_if_exception_type(httpx.RequestError), + reraise=True, + ) + def _send_http_request( + self, + method: Literal["GET", "POST", "PUT", "DELETE"], + endpoint: str, + *, + payload_json: dict[str, Any] | None = None, + params: dict[str, Any] | None = None, + ) -> httpx.Response: + headers = {"Content-Type": "application/json", "Billing-Api-Secret-Key": self._secret_key} + return self._http_client.request( + method, + f"{self._base_url.rstrip('/')}{endpoint}", + json=payload_json, + params=params, + headers=headers, + follow_redirects=True, + ) + + +class BillingNetworkAccessGroupEntitlementGateway(NetworkAccessGroupEntitlement): + @override + def is_paid_plan(self, workspace_id: str) -> bool: + try: + billing_info = BillingService.get_info(workspace_id, exclude_vector_space=True) + except ( + httpx.RequestError, + json.JSONDecodeError, + UnicodeDecodeError, + ValidationError, + _BillingHTTPStatusError, + ) as exc: + raise NetworkAccessGroupEntitlementUnavailableError from exc + return billing_info["subscription"]["plan"] in (CloudPlan.PROFESSIONAL, CloudPlan.TEAM) diff --git a/api/services/network_access_group_service.py b/api/services/network_access_group_service.py index dcbe2e89086..48e28585afa 100644 --- a/api/services/network_access_group_service.py +++ b/api/services/network_access_group_service.py @@ -1,23 +1,345 @@ +"""Application service for workspace network access groups.""" + import logging -from typing import Any +from collections.abc import Sequence +from typing import Any, Literal, NamedTuple, Protocol -from sqlalchemy import select -from sqlalchemy.exc import SQLAlchemyError - -from extensions.ext_database import db -from models import App +from machinery.context import RequestContext logger = logging.getLogger(__name__) +NetworkAccessPoint = Literal["webapp", "service_api", "mcp", "trigger"] + +_ACCESS_POINTS_BY_APP_MODE: dict[str, tuple[NetworkAccessPoint, ...]] = { + "workflow": ("webapp", "service_api", "mcp", "trigger"), + "advanced-chat": ("webapp", "service_api", "mcp"), + "chat": ("webapp", "service_api", "mcp"), + "completion": ("webapp", "service_api", "mcp"), + "agent-chat": ("webapp", "service_api", "mcp"), + "agent": ("webapp", "service_api"), +} + + +class NetworkAccessGroupError(Exception): + """Expected network access group use-case failure.""" + + +class NetworkAccessGroupUpstreamError(NetworkAccessGroupError): + """Failure returned by the network-access control plane.""" + + def __init__(self, status_code: int, reason: str | None = None) -> None: + self.status_code = status_code + self.reason = reason + super().__init__(f"network access group upstream returned HTTP {status_code}") + + +class NetworkAccessGroupAccessDeniedError(NetworkAccessGroupError): + """The actor is not a persisted owner or administrator of the workspace.""" + + +class NetworkAccessGroupEntitlementUnavailableError(NetworkAccessGroupError): + """The workspace plan could not be resolved.""" + + +class NetworkAccessGroupAppNotFoundError(NetworkAccessGroupError): + """The tenant-scoped App does not exist or is not manageable.""" + + +class NetworkAccessGroupUnsupportedAppModeError(NetworkAccessGroupError): + def __init__(self, app_mode: str) -> None: + self.app_mode = app_mode + super().__init__(f"network access control is not supported for app mode {app_mode!r}") + + +class NetworkAccessGroupUnsupportedAccessPointsError(NetworkAccessGroupError): + def __init__(self, access_points: Sequence[str]) -> None: + self.access_points = tuple(access_points) + super().__init__(f"unsupported app access points: {', '.join(self.access_points)}") + + +class NetworkAccessGroupAppQueryError(Exception): + """The local App read model could not be queried.""" + + +class NetworkAccessGroupAppRecord(NamedTuple): + id: str + mode: str + name: str + icon: str | None + icon_type: str | None + icon_background: str | None + + +class NetworkAccessGroupAppQuery(Protocol): + def get_manageable_app(self, *, workspace_id: str, app_id: str) -> NetworkAccessGroupAppRecord | None: ... + + def list_apps( + self, + *, + workspace_id: str, + app_ids: Sequence[str], + ) -> Sequence[NetworkAccessGroupAppRecord]: ... + + +class WorkspaceMembershipRoleQuery(Protocol): + def get_role_for_account(self, *, workspace_id: str, account_id: str) -> str | None: ... + + +class NetworkAccessGroupEntitlement(Protocol): + def is_paid_plan(self, workspace_id: str) -> bool: ... + + +class NetworkAccessGroupControlPlane(Protocol): + def list_groups(self, workspace_id: str, actor_account_id: str) -> dict[str, Any]: ... + + def create_group( + self, + workspace_id: str, + *, + name: str, + description: str, + allowed_cidrs: list[str], + actor_account_id: str, + ) -> dict[str, Any]: ... + + def get_group(self, workspace_id: str, group_id: str, actor_account_id: str) -> dict[str, Any]: ... + + def update_group( + self, + workspace_id: str, + group_id: str, + *, + name: str, + description: str, + allowed_cidrs: list[str], + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: ... + + def delete_group( + self, + workspace_id: str, + group_id: str, + *, + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: ... + + def get_app_binding(self, workspace_id: str, app_id: str, actor_account_id: str) -> dict[str, Any]: ... + + def update_app_binding( + self, + workspace_id: str, + app_id: str, + *, + enabled: bool, + group_id: str | None, + access_points: list[str], + expected_version: int, + actor_account_id: str, + ) -> dict[str, Any]: ... + + def cleanup_app_binding(self, workspace_id: str, app_id: str) -> dict[str, Any]: ... + class NetworkAccessGroupService: - """Dify-side orchestration for SaaS network access group responses.""" + def __init__( + self, + *, + control_plane: NetworkAccessGroupControlPlane, + apps: NetworkAccessGroupAppQuery, + memberships: WorkspaceMembershipRoleQuery, + entitlement: NetworkAccessGroupEntitlement, + ) -> None: + self._control_plane = control_plane + self._apps = apps + self._memberships = memberships + self._entitlement = entitlement + + def list_groups(self, context: RequestContext) -> dict[str, Any]: + self._ensure_workspace_admin_or_owner(context) + payload = self._control_plane.list_groups(context.active_workspace_id, context.account_id) + payload["entitled"] = self._effective_entitlement( + context.active_workspace_id, + payload.get("entitled"), + ) + return self._enrich_app_references(payload, context.active_workspace_id) + + def create_group( + self, + context: RequestContext, + *, + name: str, + description: str, + allowed_cidrs: list[str], + ) -> dict[str, Any]: + self._ensure_workspace_admin_or_owner(context) + payload = self._control_plane.create_group( + context.active_workspace_id, + name=name, + description=description, + allowed_cidrs=allowed_cidrs, + actor_account_id=context.account_id, + ) + return self._enrich_app_references(payload, context.active_workspace_id) + + def get_group(self, context: RequestContext, *, group_id: str) -> dict[str, Any]: + self._ensure_workspace_admin_or_owner(context) + payload = self._control_plane.get_group( + context.active_workspace_id, + group_id, + context.account_id, + ) + return self._enrich_app_references(payload, context.active_workspace_id) + + def update_group( + self, + context: RequestContext, + *, + group_id: str, + name: str, + description: str, + allowed_cidrs: list[str], + expected_version: int, + ) -> dict[str, Any]: + self._ensure_workspace_admin_or_owner(context) + payload = self._control_plane.update_group( + context.active_workspace_id, + group_id, + name=name, + description=description, + allowed_cidrs=allowed_cidrs, + expected_version=expected_version, + actor_account_id=context.account_id, + ) + return self._enrich_app_references(payload, context.active_workspace_id) + + def delete_group( + self, + context: RequestContext, + *, + group_id: str, + expected_version: int, + ) -> dict[str, Any]: + self._ensure_workspace_admin_or_owner(context) + return self._control_plane.delete_group( + context.active_workspace_id, + group_id, + expected_version=expected_version, + actor_account_id=context.account_id, + ) + + def get_app_binding(self, context: RequestContext, *, app_id: str) -> dict[str, Any]: + app = self._get_manageable_app(context.active_workspace_id, app_id) + self._ensure_workspace_admin_or_owner(context) + available_access_points = self._available_access_points(app.mode) + payload = self._control_plane.get_app_binding( + context.active_workspace_id, + app.id, + context.account_id, + ) + self._normalize_app_binding_defaults(payload) + self._limit_binding_access_points(payload, available_access_points) + payload["entitled"] = self._effective_entitlement( + context.active_workspace_id, + payload.get("entitled"), + ) + payload["available_access_points"] = available_access_points + return payload + + def update_app_binding( + self, + context: RequestContext, + *, + app_id: str, + enabled: bool, + group_id: str | None, + access_points: list[NetworkAccessPoint], + expected_version: int, + ) -> dict[str, Any]: + app = self._get_manageable_app(context.active_workspace_id, app_id) + self._ensure_workspace_admin_or_owner(context) + available_access_points = self._available_access_points(app.mode) + self._validate_app_access_points(access_points, available_access_points) + payload = self._control_plane.update_app_binding( + context.active_workspace_id, + app.id, + enabled=enabled, + group_id=group_id, + access_points=list(access_points), + expected_version=expected_version, + actor_account_id=context.account_id, + ) + self._normalize_app_binding_defaults(payload) + self._limit_binding_access_points(payload, available_access_points) + payload["available_access_points"] = available_access_points + return payload + + def cleanup_app_binding(self, *, workspace_id: str, app_id: str) -> dict[str, Any]: + return self._control_plane.cleanup_app_binding(workspace_id, app_id) + + def _ensure_workspace_admin_or_owner(self, context: RequestContext) -> None: + role = self._memberships.get_role_for_account( + workspace_id=context.active_workspace_id, + account_id=context.account_id, + ) + if role not in {"owner", "admin"}: + raise NetworkAccessGroupAccessDeniedError + + def _effective_entitlement(self, workspace_id: str, upstream_entitled: object) -> bool: + return bool(upstream_entitled) and self._entitlement.is_paid_plan(workspace_id) + + def _get_manageable_app(self, workspace_id: str, app_id: str) -> NetworkAccessGroupAppRecord: + app = self._apps.get_manageable_app(workspace_id=workspace_id, app_id=app_id) + if app is None: + raise NetworkAccessGroupAppNotFoundError + return app @staticmethod - def enrich_app_references( + def _available_access_points(app_mode: str) -> list[NetworkAccessPoint]: + available = _ACCESS_POINTS_BY_APP_MODE.get(app_mode) + if available is None: + raise NetworkAccessGroupUnsupportedAppModeError(app_mode) + return list(available) + + @staticmethod + def _validate_app_access_points( + requested: Sequence[NetworkAccessPoint], + available: Sequence[NetworkAccessPoint], + ) -> None: + unsupported = sorted(set(requested).difference(available)) + if unsupported: + raise NetworkAccessGroupUnsupportedAccessPointsError(unsupported) + + @staticmethod + def _normalize_app_binding_defaults(payload: dict[str, Any]) -> None: + """Materialize ProtoJSON defaults so the Console response stays stable.""" + + payload.setdefault("binding", None) + binding = payload.get("binding") + if not isinstance(binding, dict): + return + binding.setdefault("enabled", False) + if "access_points" not in binding and "accessPoints" not in binding: + binding["access_points"] = [] + + @staticmethod + def _limit_binding_access_points( payload: dict[str, Any], - tenant_id: str, - ) -> dict[str, Any]: + available: Sequence[NetworkAccessPoint], + ) -> None: + """Hide stale scopes which are not real access points for the current App mode.""" + + binding = payload.get("binding") + if not isinstance(binding, dict): + return + raw_access_points = binding.get("access_points", binding.get("accessPoints", [])) + if not isinstance(raw_access_points, list): + return + binding["access_points"] = [access_point for access_point in raw_access_points if access_point in available] + binding.pop("accessPoints", None) + + def _enrich_app_references(self, payload: dict[str, Any], workspace_id: str) -> dict[str, Any]: """Attach tenant-scoped App display metadata to one or more group payloads.""" raw_groups: list[object] = [] @@ -56,27 +378,25 @@ class NetworkAccessGroupService: return payload try: - app_models = db.session.scalars( - select(App).where( - App.tenant_id == tenant_id, - App.id.in_(app_ids), - App.status == "normal", - ) - ).all() - except SQLAlchemyError: - logger.exception("Failed to enrich network access Policy App references", extra={"tenant_id": tenant_id}) + app_records = self._apps.list_apps(workspace_id=workspace_id, app_ids=tuple(app_ids)) + except NetworkAccessGroupAppQueryError: + logger.exception( + "Failed to enrich network access group App references", + extra={"workspace_id": workspace_id}, + ) for group in groups: group["apps"] = [] return payload + apps_by_id = { - str(app.id): { - "id": str(app.id), + app.id: { + "id": app.id, "name": app.name, "icon": app.icon, - "icon_type": app.icon_type.value if app.icon_type is not None else None, + "icon_type": app.icon_type, "icon_background": app.icon_background, } - for app in app_models + for app in app_records } for group in groups: group_app_ids = group["app_ids"] if isinstance(group["app_ids"], list) else [] diff --git a/api/tasks/remove_app_and_related_data_task.py b/api/tasks/remove_app_and_related_data_task.py index 52212131124..40b6f70b481 100644 --- a/api/tasks/remove_app_and_related_data_task.py +++ b/api/tasks/remove_app_and_related_data_task.py @@ -52,7 +52,6 @@ from models.workflow import ( ) from repositories.factory import DifyAPIRepositoryFactory from services.api_token_service import ApiTokenCache -from services.billing_service import BillingService logger = logging.getLogger(__name__) @@ -76,7 +75,14 @@ def cleanup_app_network_access_group_binding_task(self, tenant_id: str, app_id: if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD: return try: - BillingService.cleanup_app_network_access_group_binding(tenant_id, app_id) + # Import lazily because AppService imports this task module while the + # application-service composition root imports AppService dependencies. + from extensions.ext_application_services import application_services + + application_services().network_access_groups.cleanup_app_binding( + workspace_id=tenant_id, + app_id=app_id, + ) except Exception as e: countdown = _network_access_binding_cleanup_retry_delay(self.request.retries) logger.exception( diff --git a/api/tests/unit_tests/controllers/console/workspace/test_network_access_group.py b/api/tests/unit_tests/controllers/console/workspace/test_network_access_group.py index 62b6ab93a2d..1b50ad3bf15 100644 --- a/api/tests/unit_tests/controllers/console/workspace/test_network_access_group.py +++ b/api/tests/unit_tests/controllers/console/workspace/test_network_access_group.py @@ -1,16 +1,17 @@ -from collections.abc import Iterator +from collections.abc import Callable, Generator +from contextlib import contextmanager from datetime import UTC, datetime from inspect import unwrap from types import SimpleNamespace -from typing import cast from unittest.mock import MagicMock, patch from uuid import UUID import pytest +from flask import Flask from pydantic import ValidationError -from sqlalchemy.exc import SQLAlchemyError from werkzeug.exceptions import BadGateway, BadRequest, Conflict, Forbidden, HTTPException, NotFound, ServiceUnavailable +from controllers.console.app.error import AppNotFoundError from controllers.console.workspace.network_access_group import ( AppNetworkAccessGroupApi, AppNetworkAccessGroupUpdatePayload, @@ -19,14 +20,19 @@ from controllers.console.workspace.network_access_group import ( NetworkAccessGroupCreatePayload, NetworkAccessGroupDeleteQuery, NetworkAccessGroupUpdatePayload, - _available_access_points, - _effective_entitlement, - _limit_binding_access_points, + _translate_service_error, _translate_upstream_error, ) -from models import App, AppMode, TenantAccountRole -from services.billing_service import BillingService, NetworkAccessGroupUpstreamError -from services.network_access_group_service import NetworkAccessGroupService +from machinery.context import RequestContext +from services.network_access_group_service import ( + NetworkAccessGroupAccessDeniedError, + NetworkAccessGroupAppNotFoundError, + NetworkAccessGroupEntitlementUnavailableError, + NetworkAccessGroupError, + NetworkAccessGroupUnsupportedAccessPointsError, + NetworkAccessGroupUnsupportedAppModeError, + NetworkAccessGroupUpstreamError, +) TENANT_ID = "11111111-1111-4111-8111-111111111111" ACCOUNT_ID = "22222222-2222-4222-8222-222222222222" @@ -35,15 +41,20 @@ GROUP_ID = "44444444-4444-4444-8444-444444444444" BINDING_ID = "55555555-5555-4555-8555-555555555555" -def _current_user(role: TenantAccountRole = TenantAccountRole.OWNER) -> SimpleNamespace: - return SimpleNamespace(id=ACCOUNT_ID, current_role=role) +def _request_context() -> RequestContext: + return RequestContext( + request_id="request-id", + trace_id=None, + account_id=ACCOUNT_ID, + active_workspace_id=TENANT_ID, + ) -@pytest.fixture(autouse=True) -def paid_entitlement() -> Iterator[None]: +@contextmanager +def _application_services(service: MagicMock) -> Generator[None]: with patch( - "controllers.console.workspace.network_access_group._effective_entitlement", - side_effect=lambda _tenant_id, upstream: bool(upstream), + "controllers.console.workspace.network_access_group.application_services", + return_value=SimpleNamespace(network_access_groups=service), ): yield @@ -58,7 +69,7 @@ def _group_payload(*, app_ids: list[str] | None = None) -> dict[str, object]: "allowedCidrs": ["203.0.113.7/32"], "usedByCount": len(app_ids), "usedByAppIds": app_ids, - # ProtoJSON represents int64 values as strings and timestamps as RFC 3339. + "apps": [], "version": "2", "updatedByAccountId": ACCOUNT_ID, "createdAt": datetime(2026, 8, 21, tzinfo=UTC).isoformat(), @@ -86,211 +97,135 @@ def _binding_payload( } -def test_list_authorizes_and_forwards_tenant_and_actor() -> None: +def test_list_forwards_request_context_and_serializes_response() -> None: + service = MagicMock() + service.list_groups.return_value = { + "tenantId": TENANT_ID, + "entitled": True, + "groups": [_group_payload()], + } api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.get) - current_user = _current_user() - upstream_payload = {"tenantId": TENANT_ID, "entitled": True, "groups": [_group_payload()]} - with ( - patch.object(BillingService, "list_network_access_groups", return_value=upstream_payload) as list_groups, - ): - result = method(api, current_tenant_id=TENANT_ID, current_user=current_user) + with _application_services(service): + result = unwrap(api.get)(api, request_context=_request_context()) - list_groups.assert_called_once_with(TENANT_ID, ACCOUNT_ID) + service.list_groups.assert_called_once_with(_request_context()) assert result["tenant_id"] == TENANT_ID - assert result["groups"][0]["name"] == "Office network" assert result["groups"][0]["version"] == 2 - assert result["groups"][0]["used_by_count"] == 0 - assert result["groups"][0]["app_ids"] == [] - assert result["groups"][0]["apps"] == [] assert result["groups"][0]["updated_at"] == "2026-08-21T00:00:00Z" -def test_create_injects_actor_and_returns_created_contract() -> None: - api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.post) - current_user = _current_user() +def test_create_forwards_payload_and_returns_201() -> None: + service = MagicMock() + service.create_group.return_value = {"group": _group_payload()} request_payload = NetworkAccessGroupCreatePayload( name="Office network", description="Reusable office egress addresses", allowed_cidrs=["203.0.113.7/32"], ) + api = CurrentWorkspaceNetworkAccessGroupsApi() - with ( - patch.object( - BillingService, - "create_network_access_group", - return_value={"group": _group_payload()}, - ) as create_group, - ): - body, status = method( + with _application_services(service): + body, status = unwrap(api.post)( api, req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=current_user, + request_context=_request_context(), ) - create_group.assert_called_once_with( - TENANT_ID, + service.create_group.assert_called_once_with( + _request_context(), name="Office network", description="Reusable office egress addresses", allowed_cidrs=["203.0.113.7/32"], - actor_account_id=ACCOUNT_ID, ) assert status == 201 assert body["group"]["id"] == GROUP_ID -def test_update_group_forwards_path_id_and_expected_version() -> None: - api = CurrentWorkspaceNetworkAccessGroupApi() - method = unwrap(api.put) - current_user = _current_user() - request_payload = NetworkAccessGroupUpdatePayload( +def test_group_detail_update_and_delete_forward_path_and_query_values() -> None: + service = MagicMock() + service.get_group.return_value = {"group": _group_payload()} + service.update_group.return_value = {"group": _group_payload()} + service.delete_group.return_value = {"deleted": True} + update_payload = NetworkAccessGroupUpdatePayload( name="Office network", description="Updated", allowed_cidrs=["203.0.113.0/24"], expected_version=1, ) + api = CurrentWorkspaceNetworkAccessGroupApi() - with ( - patch.object( - BillingService, - "update_network_access_group", - return_value={"group": _group_payload()}, - ) as update_group, - ): - result = method( + with _application_services(service): + get_result = unwrap(api.get)(api, request_context=_request_context(), group_id=UUID(GROUP_ID)) + update_result = unwrap(api.put)( api, - req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=current_user, + req_data=update_payload, + request_context=_request_context(), group_id=UUID(GROUP_ID), ) - - update_group.assert_called_once_with( - TENANT_ID, - GROUP_ID, - name="Office network", - description="Updated", - allowed_cidrs=["203.0.113.0/24"], - expected_version=1, - actor_account_id=ACCOUNT_ID, - ) - assert result["group"]["version"] == 2 - - -def test_delete_group_reads_expected_version_from_query_and_injects_actor() -> None: - api = CurrentWorkspaceNetworkAccessGroupApi() - method = unwrap(api.delete) - current_user = _current_user() - - with ( - patch.object( - BillingService, - "delete_network_access_group", - return_value={"deleted": True}, - ) as delete_group, - ): - result = method( + delete_result = unwrap(api.delete)( api, req_data=NetworkAccessGroupDeleteQuery(expected_version=2), - current_tenant_id=TENANT_ID, - current_user=current_user, + request_context=_request_context(), group_id=UUID(GROUP_ID), ) - delete_group.assert_called_once_with( - TENANT_ID, - GROUP_ID, - expected_version=2, - actor_account_id=ACCOUNT_ID, + service.get_group.assert_called_once_with(_request_context(), group_id=GROUP_ID) + service.update_group.assert_called_once_with( + _request_context(), + group_id=GROUP_ID, + name="Office network", + description="Updated", + allowed_cidrs=["203.0.113.0/24"], + expected_version=1, ) - assert result == {"deleted": True} + service.delete_group.assert_called_once_with(_request_context(), group_id=GROUP_ID, expected_version=2) + assert get_result["group"]["id"] == GROUP_ID + assert update_result["group"]["version"] == 2 + assert delete_result == {"deleted": True} -def test_app_get_forwards_tenant_scoped_app_and_supports_unbound_response() -> None: - api = AppNetworkAccessGroupApi() - method = unwrap(api.get) - current_user = _current_user() - app_model = SimpleNamespace(id=UUID(APP_ID), tenant_id=TENANT_ID, mode=AppMode.CHAT) - upstream_payload = {"tenantId": TENANT_ID, "appId": APP_ID, "entitled": True} - - with ( - patch.object( - BillingService, - "get_app_network_access_group", - return_value=upstream_payload, - ) as get_binding, - ): - result = method( - api, - current_tenant_id=TENANT_ID, - current_user=current_user, - app_model=app_model, - ) - - get_binding.assert_called_once_with(TENANT_ID, APP_ID, ACCOUNT_ID) - assert result["binding"] is None - assert result["available_access_points"] == ["webapp", "service_api", "mcp"] - - -@pytest.mark.parametrize("enabled", [True, False]) -def test_app_put_atomically_saves_enabled_group_and_access_points(enabled: bool) -> None: - api = AppNetworkAccessGroupApi() - method = unwrap(api.put) - current_user = _current_user() - app_model = SimpleNamespace(id=UUID(APP_ID), tenant_id=TENANT_ID, mode=AppMode.CHAT) +def test_app_get_and_put_forward_app_id_without_orm_models() -> None: + service = MagicMock() + service.get_app_binding.return_value = { + "tenantId": TENANT_ID, + "appId": APP_ID, + "entitled": True, + "available_access_points": ["webapp", "service_api", "mcp"], + "binding": None, + } + service.update_app_binding.return_value = { + "binding": _binding_payload(), + "available_access_points": ["webapp", "service_api", "mcp"], + } request_payload = AppNetworkAccessGroupUpdatePayload( - enabled=enabled, + enabled=True, group_id=GROUP_ID, access_points=["webapp", "service_api"], expected_version=2, ) + api = AppNetworkAccessGroupApi() - with ( - patch.object( - BillingService, - "update_app_network_access_group", - return_value={"binding": _binding_payload(enabled=enabled)}, - ) as update_binding, - ): - result = method( + with _application_services(service): + get_result = unwrap(api.get)(api, request_context=_request_context(), app_id=UUID(APP_ID)) + put_result = unwrap(api.put)( api, req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=current_user, - app_model=app_model, + request_context=_request_context(), + app_id=UUID(APP_ID), ) - update_binding.assert_called_once_with( - TENANT_ID, - APP_ID, - enabled=enabled, + service.get_app_binding.assert_called_once_with(_request_context(), app_id=APP_ID) + service.update_app_binding.assert_called_once_with( + _request_context(), + app_id=APP_ID, + enabled=True, group_id=GROUP_ID, access_points=["webapp", "service_api"], expected_version=2, - actor_account_id=ACCOUNT_ID, ) - assert result["binding"]["enabled"] is enabled - assert result["binding"]["group_id"] == GROUP_ID - assert result["binding"]["access_points"] == ["webapp", "service_api"] - assert result["binding"]["version"] == 3 - assert result["available_access_points"] == ["webapp", "service_api", "mcp"] - - -def test_list_rejects_non_privileged_workspace_member_before_upstream_call() -> None: - api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.get) - current_user = _current_user(TenantAccountRole.NORMAL) - - with ( - patch.object(BillingService, "list_network_access_groups") as list_groups, - pytest.raises(Forbidden), - ): - method(api, current_tenant_id=TENANT_ID, current_user=current_user) - - list_groups.assert_not_called() + assert get_result["binding"] is None + assert put_result["binding"]["version"] == 3 @pytest.mark.parametrize( @@ -325,124 +260,88 @@ def test_conflict_error_mapping_preserves_safe_actionable_reason(reason: str, ex def test_internal_secret_error_is_not_reported_as_tenant_input_failure() -> None: - error = _translate_upstream_error(NetworkAccessGroupUpstreamError(400, "INVALID_SECRET_KEY")) - assert isinstance(error, ServiceUnavailable) - - -def test_list_maps_invalid_upstream_contract_to_bad_gateway() -> None: - api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.get) - current_user = _current_user() - - with ( - patch.object( - BillingService, - "list_network_access_groups", - return_value={"tenant_id": TENANT_ID, "entitled": True, "groups": [{"allowed_cidrs": []}]}, - ), - pytest.raises(BadGateway), - ): - method(api, current_tenant_id=TENANT_ID, current_user=current_user) - - -def test_response_contract_accepts_protojson_omitted_optional_group_fields() -> None: - api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.get) - current_user = _current_user() - group = _group_payload() - group.pop("description") - group.pop("usedByAppIds") - group.pop("usedByCount") - - with ( - patch.object( - BillingService, - "list_network_access_groups", - return_value={"tenantId": TENANT_ID, "entitled": True, "groups": [group]}, - ), - ): - result = method(api, current_tenant_id=TENANT_ID, current_user=current_user) - - assert result["groups"][0]["description"] == "" - assert result["groups"][0]["allowed_cidrs"] == ["203.0.113.7/32"] - assert result["groups"][0]["app_ids"] == [] - assert result["groups"][0]["used_by_count"] == 0 - - -def test_binding_response_materializes_protojson_omitted_disabled_defaults() -> None: - api = AppNetworkAccessGroupApi() - method = unwrap(api.put) - current_user = _current_user() - app_model = SimpleNamespace(id=UUID(APP_ID), tenant_id=TENANT_ID, mode=AppMode.CHAT) - request_payload = AppNetworkAccessGroupUpdatePayload( - enabled=False, - group_id=None, - access_points=[], - expected_version=2, + assert isinstance( + _translate_upstream_error(NetworkAccessGroupUpstreamError(400, "INVALID_SECRET_KEY")), + ServiceUnavailable, ) - binding = _binding_payload(enabled=False, group_id=None, access_points=[]) - binding.pop("groupId") - binding.pop("enabled") - binding.pop("accessPoints") + + +@pytest.mark.parametrize( + ("service_error", "expected_exception", "expected_message"), + [ + (NetworkAccessGroupAccessDeniedError(), Forbidden, "owners and administrators"), + (NetworkAccessGroupEntitlementUnavailableError(), ServiceUnavailable, "temporarily unavailable"), + (NetworkAccessGroupAppNotFoundError(), AppNotFoundError, "not found"), + (NetworkAccessGroupUnsupportedAppModeError("channel"), BadRequest, "channel"), + (NetworkAccessGroupUnsupportedAccessPointsError(["trigger"]), BadRequest, "trigger"), + ], +) +def test_application_error_mapping( + service_error: NetworkAccessGroupError, + expected_exception: type[HTTPException], + expected_message: str, +) -> None: + error = _translate_service_error(service_error) + assert isinstance(error, expected_exception) + assert error.description is not None + assert expected_message in error.description.lower() + + +@pytest.mark.parametrize( + "mutation", + [ + CurrentWorkspaceNetworkAccessGroupsApi.post, + CurrentWorkspaceNetworkAccessGroupApi.put, + CurrentWorkspaceNetworkAccessGroupApi.delete, + AppNetworkAccessGroupApi.put, + ], +) +def test_mutation_paid_plan_admission_precedes_payload_validation(mutation: Callable[..., object]) -> None: + app = Flask(__name__) + paid_plan_wrapper = unwrap(mutation, stop=lambda candidate: candidate is not mutation) with ( - patch.object( - BillingService, - "update_app_network_access_group", - return_value={"binding": binding}, + app.test_request_context(json={"name": "", "description": "", "allowed_cidrs": []}), + patch( + "controllers.console.wraps.current_account_with_tenant", + return_value=(object(), TENANT_ID), ), + patch( + "controllers.console.wraps.is_cloud_edition_billing_paid_plan", + return_value=False, + ) as is_paid_plan, + pytest.raises(HTTPException) as exc_info, ): - result = method( - api, - req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=current_user, - app_model=app_model, + paid_plan_wrapper(object()) + + assert exc_info.value.code == 403 + is_paid_plan.assert_called_once_with(TENANT_ID) + + +def test_controller_maps_service_error_before_serialization() -> None: + service = MagicMock() + service.list_groups.side_effect = NetworkAccessGroupAccessDeniedError + + with _application_services(service), pytest.raises(Forbidden, match="owners and administrators"): + unwrap(CurrentWorkspaceNetworkAccessGroupsApi().get)( + CurrentWorkspaceNetworkAccessGroupsApi(), + request_context=_request_context(), ) - assert result["binding"]["group_id"] is None - assert result["binding"]["enabled"] is False - assert result["binding"]["access_points"] == [] - -def test_binding_response_accepts_policy_id_alias() -> None: - api = AppNetworkAccessGroupApi() - method = unwrap(api.put) - app_model = SimpleNamespace(id=UUID(APP_ID), tenant_id=TENANT_ID, mode=AppMode.CHAT) - request_payload = AppNetworkAccessGroupUpdatePayload( - enabled=True, - group_id=GROUP_ID, - access_points=["webapp"], - expected_version=2, - ) - binding = _binding_payload(access_points=["webapp"]) - binding["policyId"] = binding.pop("groupId") - - with patch.object(BillingService, "update_app_network_access_group", return_value={"binding": binding}): - result = method( - api, - req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=_current_user(), - app_model=app_model, - ) - - assert result["binding"]["group_id"] == GROUP_ID - - -def test_binding_response_hides_scopes_not_available_for_current_app_mode() -> None: - payload = { - "binding": _binding_payload( - enabled=True, - group_id=GROUP_ID, - access_points=["webapp", "service_api", "mcp", "trigger"], - ) +def test_invalid_service_response_is_bad_gateway() -> None: + service = MagicMock() + service.list_groups.return_value = { + "tenant_id": TENANT_ID, + "entitled": True, + "groups": [{"allowed_cidrs": list[str]()}], } - _limit_binding_access_points(payload, ["webapp", "service_api"]) - - assert payload["binding"]["access_points"] == ["webapp", "service_api"] - assert "accessPoints" not in payload["binding"] + with _application_services(service), pytest.raises(BadGateway): + unwrap(CurrentWorkspaceNetworkAccessGroupsApi().get)( + CurrentWorkspaceNetworkAccessGroupsApi(), + request_context=_request_context(), + ) @pytest.mark.parametrize( @@ -458,67 +357,6 @@ def test_create_payload_contract_rejects_invalid_shapes(payload: dict[str, objec NetworkAccessGroupCreatePayload.model_validate(payload) -def test_binding_payload_rejects_invalid_group_id_and_version() -> None: - with pytest.raises(ValidationError): - AppNetworkAccessGroupUpdatePayload.model_validate( - { - "enabled": True, - "group_id": "not-a-uuid", - "access_points": ["webapp"], - "expected_version": -1, - } - ) - - -@pytest.mark.parametrize( - ("app_mode", "expected"), - [ - (AppMode.WORKFLOW, ["webapp", "service_api", "mcp", "trigger"]), - (AppMode.ADVANCED_CHAT, ["webapp", "service_api", "mcp"]), - (AppMode.CHAT, ["webapp", "service_api", "mcp"]), - (AppMode.COMPLETION, ["webapp", "service_api", "mcp"]), - (AppMode.AGENT_CHAT, ["webapp", "service_api", "mcp"]), - (AppMode.AGENT, ["webapp", "service_api"]), - ], -) -def test_available_access_points_are_derived_from_app_mode(app_mode: AppMode, expected: list[str]) -> None: - app_model = SimpleNamespace(mode=app_mode) - - assert _available_access_points(cast(App, app_model)) == expected - - -@pytest.mark.parametrize("app_mode", [AppMode.CHANNEL, AppMode.RAG_PIPELINE]) -def test_unsupported_app_modes_are_rejected(app_mode: AppMode) -> None: - with pytest.raises(BadRequest, match="not supported"): - _available_access_points(cast(App, SimpleNamespace(mode=app_mode))) - - -def test_app_put_rejects_access_point_not_available_for_mode_before_upstream_call() -> None: - api = AppNetworkAccessGroupApi() - method = unwrap(api.put) - app_model = SimpleNamespace(id=UUID(APP_ID), tenant_id=TENANT_ID, mode=AppMode.AGENT) - request_payload = AppNetworkAccessGroupUpdatePayload( - enabled=True, - group_id=GROUP_ID, - access_points=["webapp", "mcp"], - expected_version=1, - ) - - with ( - patch.object(BillingService, "update_app_network_access_group") as update_binding, - pytest.raises(BadRequest, match="mcp"), - ): - method( - api, - req_data=request_payload, - current_tenant_id=TENANT_ID, - current_user=_current_user(), - app_model=app_model, - ) - - update_binding.assert_not_called() - - @pytest.mark.parametrize( "payload", [ @@ -550,83 +388,3 @@ def test_app_config_accepts_policy_id_alias_and_disabled_draft() -> None: assert str(payload.group_id) == GROUP_ID assert payload.access_points == ["webapp", "service_api"] - - -def test_group_responses_are_enriched_with_tenant_scoped_app_metadata() -> None: - app_id_2 = "66666666-6666-4666-8666-666666666666" - payload = { - "group": _group_payload(app_ids=[APP_ID, app_id_2]), - } - payload["group"].pop("usedByCount") - app = SimpleNamespace( - id=APP_ID, - name="Customer support", - icon="robot", - icon_type=SimpleNamespace(value="emoji"), - icon_background="#FFFFFF", - ) - db_mock = MagicMock() - db_mock.session.scalars.return_value.all.return_value = [app] - - with patch("services.network_access_group_service.db", db_mock): - result = NetworkAccessGroupService.enrich_app_references(payload, TENANT_ID) - - assert result["group"]["apps"] == [ - { - "id": APP_ID, - "name": "Customer support", - "icon": "robot", - "icon_type": "emoji", - "icon_background": "#FFFFFF", - } - ] - assert result["group"]["app_ids"] == [APP_ID, app_id_2] - assert result["group"]["used_by_count"] == 2 - db_mock.session.scalars.assert_called_once() - - -def test_group_app_enrichment_degrades_to_ids_when_core_database_is_unavailable() -> None: - payload = {"group": _group_payload(app_ids=[APP_ID])} - db_mock = MagicMock() - db_mock.session.scalars.side_effect = SQLAlchemyError("database unavailable") - - with patch("services.network_access_group_service.db", db_mock): - result = NetworkAccessGroupService.enrich_app_references(payload, TENANT_ID) - - assert result["group"]["app_ids"] == [APP_ID] - assert result["group"]["used_by_count"] == 1 - assert result["group"]["apps"] == [] - - -def test_sandbox_read_returns_not_entitled_for_upgrade_state() -> None: - api = CurrentWorkspaceNetworkAccessGroupsApi() - method = unwrap(api.get) - upstream_payload: dict[str, object] = { - "tenantId": TENANT_ID, - "entitled": True, - "groups": list[object](), - } - - with ( - patch.object(BillingService, "list_network_access_groups", return_value=upstream_payload), - patch("controllers.console.workspace.network_access_group._effective_entitlement", return_value=False), - ): - result = method(api, current_tenant_id=TENANT_ID, current_user=_current_user()) - - assert result["entitled"] is False - - -@pytest.mark.parametrize( - ("upstream_entitled", "paid_plan", "expected"), - [(True, True, True), (False, True, False), (True, False, False)], -) -def test_effective_entitlement_requires_rollout_and_paid_plan( - upstream_entitled: bool, - paid_plan: bool, - expected: bool, -) -> None: - with patch( - "controllers.console.workspace.network_access_group.is_cloud_edition_billing_paid_plan", - return_value=paid_plan, - ): - assert _effective_entitlement(TENANT_ID, upstream_entitled) is expected diff --git a/api/tests/unit_tests/extensions/test_ext_application_services.py b/api/tests/unit_tests/extensions/test_ext_application_services.py index df115b908bf..11ce08b1634 100644 --- a/api/tests/unit_tests/extensions/test_ext_application_services.py +++ b/api/tests/unit_tests/extensions/test_ext_application_services.py @@ -34,6 +34,7 @@ from repositories.app_statistic_query_repository import AppStatisticQueryReposit from repositories.app_tracing_config_repository import SQLAlchemyAppTracingConfigRepository from repositories.human_input_file_upload_repository import SQLAlchemyHumanInputFileUploadRepository from repositories.message_file_preview_repository import MessageFilePreviewQueryRepository +from repositories.network_access_group_repository import SQLAlchemyNetworkAccessGroupAppRepository from repositories.plugin_file_upload_repository import SQLAlchemyPluginFileUploadOwnerRepository from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository from repositories.upload_file_delivery_repository import UploadFileDeliveryQueryRepository @@ -76,6 +77,11 @@ from services.file_service import FileService from services.human_input_file_upload_service import HumanInputFileUploadService from services.init_validation_service import InvalidInitializationPasswordError from services.message_file_preview_service import MessageFilePreviewService +from services.network_access_group_gateway import ( + BillingNetworkAccessGroupEntitlementGateway, + NetworkAccessGroupGateway, +) +from services.network_access_group_service import NetworkAccessGroupService from services.partner_tenant_binding_service import PartnerTenantBindingService from services.plugin_file_upload_gateway import ToolFilePluginUploadGateway from services.plugin_file_upload_service import PluginFileUploadService @@ -479,6 +485,24 @@ def test_build_application_services_wires_billing_service( sync_partner_tenants_bindings.assert_called_once_with("account-1", "partner-key", "click-1") +def test_build_application_services_wires_network_access_group_boundary( + sqlite_session_factory: sessionmaker[Session], +) -> None: + services = ext_application_services.build_application_services( + database_client=sqlite_session_factory, + deployment_edition=DeploymentEdition.CLOUD, + initialization_password="", + redis=MagicMock(spec=RedisClientWrapper), + ) + + network_access_groups = services.network_access_groups + assert isinstance(network_access_groups, NetworkAccessGroupService) + assert isinstance(network_access_groups._control_plane, NetworkAccessGroupGateway) + assert isinstance(network_access_groups._apps, SQLAlchemyNetworkAccessGroupAppRepository) + assert network_access_groups._apps._session_factory is sqlite_session_factory + assert isinstance(network_access_groups._entitlement, BillingNetworkAccessGroupEntitlementGateway) + + def test_build_application_services_wires_compliance_downloads( sqlite_session_factory: sessionmaker[Session], ) -> None: diff --git a/api/tests/unit_tests/repositories/test_network_access_group_repository.py b/api/tests/unit_tests/repositories/test_network_access_group_repository.py new file mode 100644 index 00000000000..49ce7a0fa7c --- /dev/null +++ b/api/tests/unit_tests/repositories/test_network_access_group_repository.py @@ -0,0 +1,169 @@ +from unittest.mock import MagicMock + +import pytest +from sqlalchemy import text +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session, sessionmaker + +from models.agent import Agent, AgentScope, AgentSource, AgentStatus +from models.model import App, AppMode, IconType +from repositories.network_access_group_repository import SQLAlchemyNetworkAccessGroupAppRepository +from services.network_access_group_service import NetworkAccessGroupAppQueryError, NetworkAccessGroupAppRecord + +_WORKSPACE_ID = "11111111-1111-1111-1111-111111111111" +_OTHER_WORKSPACE_ID = "22222222-2222-2222-2222-222222222222" +_APP_ID = "33333333-3333-3333-3333-333333333333" +_OTHER_APP_ID = "44444444-4444-4444-4444-444444444444" +_DISABLED_APP_ID = "55555555-5555-5555-5555-555555555555" +_UNREQUESTED_APP_ID = "66666666-6666-6666-6666-666666666666" + + +def _app( + app_id: str, + *, + workspace_id: str = _WORKSPACE_ID, + mode: AppMode = AppMode.WORKFLOW, +) -> App: + return App( + id=app_id, + tenant_id=workspace_id, + name=f"App {app_id}", + description="", + mode=mode, + icon_type=IconType.EMOJI, + icon="robot", + icon_background="#ffffff", + enable_site=True, + enable_api=True, + ) + + +def _repository(session_factory: sessionmaker[Session]) -> SQLAlchemyNetworkAccessGroupAppRepository: + return SQLAlchemyNetworkAccessGroupAppRepository(session_factory=session_factory) + + +def test_get_manageable_app_returns_detached_projection_and_enforces_workspace_and_status_scope( + sqlite_session_factory: sessionmaker[Session], +) -> None: + with sqlite_session_factory() as session: + session.add_all( + [ + _app(_APP_ID), + _app(_OTHER_APP_ID, workspace_id=_OTHER_WORKSPACE_ID), + _app(_DISABLED_APP_ID), + ] + ) + session.commit() + session.execute( + text("UPDATE apps SET status = 'disabled' WHERE id = :app_id"), + {"app_id": _DISABLED_APP_ID}, + ) + session.commit() + + repository = _repository(sqlite_session_factory) + + result = repository.get_manageable_app(workspace_id=_WORKSPACE_ID, app_id=_APP_ID) + + assert result == NetworkAccessGroupAppRecord( + id=_APP_ID, + mode=AppMode.WORKFLOW.value, + name=f"App {_APP_ID}", + icon="robot", + icon_type=IconType.EMOJI.value, + icon_background="#ffffff", + ) + assert repository.get_manageable_app(workspace_id=_OTHER_WORKSPACE_ID, app_id=_APP_ID) is None + assert repository.get_manageable_app(workspace_id=_WORKSPACE_ID, app_id=_OTHER_APP_ID) is None + assert repository.get_manageable_app(workspace_id=_WORKSPACE_ID, app_id=_DISABLED_APP_ID) is None + assert repository.get_manageable_app(workspace_id=_WORKSPACE_ID, app_id=_UNREQUESTED_APP_ID) is None + + +def test_list_apps_returns_only_normal_apps_in_the_requested_workspace( + sqlite_session_factory: sessionmaker[Session], +) -> None: + with sqlite_session_factory() as session: + session.add_all( + [ + _app(_APP_ID), + _app(_OTHER_APP_ID, workspace_id=_OTHER_WORKSPACE_ID), + _app(_DISABLED_APP_ID), + _app(_UNREQUESTED_APP_ID), + ] + ) + session.commit() + session.execute( + text("UPDATE apps SET status = 'disabled' WHERE id = :app_id"), + {"app_id": _DISABLED_APP_ID}, + ) + session.commit() + + result = _repository(sqlite_session_factory).list_apps( + workspace_id=_WORKSPACE_ID, + app_ids=[_APP_ID, _OTHER_APP_ID, _DISABLED_APP_ID], + ) + + assert result == ( + NetworkAccessGroupAppRecord( + id=_APP_ID, + mode=AppMode.WORKFLOW.value, + name=f"App {_APP_ID}", + icon="robot", + icon_type=IconType.EMOJI.value, + icon_background="#ffffff", + ), + ) + + +@pytest.mark.parametrize( + ("scope", "source", "status", "expected_visible"), + [ + (AgentScope.WORKFLOW_ONLY, AgentSource.WORKFLOW, AgentStatus.ACTIVE, False), + (AgentScope.WORKFLOW_ONLY, AgentSource.WORKFLOW, AgentStatus.ARCHIVED, False), + (AgentScope.ROSTER, AgentSource.AGENT_APP, AgentStatus.ACTIVE, True), + ], +) +def test_get_manageable_app_applies_agent_scope_and_lifecycle_visibility( + sqlite_session_factory: sessionmaker[Session], + scope: AgentScope, + source: AgentSource, + status: AgentStatus, + expected_visible: bool, +) -> None: + app = _app(_APP_ID, mode=AppMode.AGENT) + with sqlite_session_factory() as session: + session.add_all( + [ + app, + Agent( + id="77777777-7777-7777-7777-777777777777", + tenant_id=_WORKSPACE_ID, + name="Bound agent", + description="", + role="", + scope=scope, + source=source, + status=status, + app_id=app.id if scope == AgentScope.ROSTER else "88888888-8888-8888-8888-888888888888", + backing_app_id=app.id if scope == AgentScope.WORKFLOW_ONLY else None, + workflow_id=("99999999-9999-9999-9999-999999999999" if scope == AgentScope.WORKFLOW_ONLY else None), + workflow_node_id="agent-node" if scope == AgentScope.WORKFLOW_ONLY else None, + ), + ] + ) + session.commit() + + result = _repository(sqlite_session_factory).get_manageable_app( + workspace_id=_WORKSPACE_ID, + app_id=_APP_ID, + ) + + assert (result is not None) is expected_visible + + +def test_list_apps_maps_database_failures_to_query_error() -> None: + session_factory = MagicMock() + session = session_factory.return_value.__enter__.return_value + session.scalars.side_effect = SQLAlchemyError("database unavailable") + + with pytest.raises(NetworkAccessGroupAppQueryError): + _repository(session_factory).list_apps(workspace_id=_WORKSPACE_ID, app_ids=[_APP_ID]) diff --git a/api/tests/unit_tests/repositories/test_workspace_query_repository.py b/api/tests/unit_tests/repositories/test_workspace_query_repository.py new file mode 100644 index 00000000000..6ab368049af --- /dev/null +++ b/api/tests/unit_tests/repositories/test_workspace_query_repository.py @@ -0,0 +1,58 @@ +from sqlalchemy.orm import Session, sessionmaker + +from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole +from repositories.workspace_query_repository import WorkspaceQueryRepository + +_WORKSPACE_ID = "11111111-1111-1111-1111-111111111111" +_OTHER_WORKSPACE_ID = "22222222-2222-2222-2222-222222222222" +_OWNER_ID = "33333333-3333-3333-3333-333333333333" +_ADMIN_ID = "44444444-4444-4444-4444-444444444444" +_OTHER_WORKSPACE_ACCOUNT_ID = "55555555-5555-5555-5555-555555555555" + + +def _account(account_id: str) -> Account: + account = Account(name=f"Account {account_id}", email=f"{account_id}@example.com") + account.id = account_id + return account + + +def test_get_role_for_account_is_scoped_by_workspace_and_account( + sqlite_session_factory: sessionmaker[Session], +) -> None: + workspace = Tenant(name="Workspace") + workspace.id = _WORKSPACE_ID + other_workspace = Tenant(name="Other workspace") + other_workspace.id = _OTHER_WORKSPACE_ID + with sqlite_session_factory() as session: + session.add_all( + [ + workspace, + other_workspace, + _account(_OWNER_ID), + _account(_ADMIN_ID), + _account(_OTHER_WORKSPACE_ACCOUNT_ID), + TenantAccountJoin( + tenant_id=_WORKSPACE_ID, + account_id=_OWNER_ID, + role=TenantAccountRole.OWNER, + ), + TenantAccountJoin( + tenant_id=_WORKSPACE_ID, + account_id=_ADMIN_ID, + role=TenantAccountRole.ADMIN, + ), + TenantAccountJoin( + tenant_id=_OTHER_WORKSPACE_ID, + account_id=_OTHER_WORKSPACE_ACCOUNT_ID, + role=TenantAccountRole.OWNER, + ), + ] + ) + session.commit() + + repository = WorkspaceQueryRepository(sqlite_session_factory) + + assert repository.get_role_for_account(workspace_id=_WORKSPACE_ID, account_id=_OWNER_ID) == "owner" + assert repository.get_role_for_account(workspace_id=_WORKSPACE_ID, account_id=_ADMIN_ID) == "admin" + assert repository.get_role_for_account(workspace_id=_WORKSPACE_ID, account_id="missing-account") is None + assert repository.get_role_for_account(workspace_id=_WORKSPACE_ID, account_id=_OTHER_WORKSPACE_ACCOUNT_ID) is None diff --git a/api/tests/unit_tests/services/test_network_access_group_billing_service.py b/api/tests/unit_tests/services/test_network_access_group_gateway.py similarity index 54% rename from api/tests/unit_tests/services/test_network_access_group_billing_service.py rename to api/tests/unit_tests/services/test_network_access_group_gateway.py index c034f21904d..90366728938 100644 --- a/api/tests/unit_tests/services/test_network_access_group_billing_service.py +++ b/api/tests/unit_tests/services/test_network_access_group_gateway.py @@ -1,10 +1,18 @@ -from collections.abc import Callable, Iterator from unittest.mock import MagicMock, call, patch import httpx import pytest -from services.billing_service import BillingService, NetworkAccessGroupUpstreamError +from enums import CloudPlan +from services.billing_service import _BillingHTTPStatusError +from services.network_access_group_gateway import ( + BillingNetworkAccessGroupEntitlementGateway, + NetworkAccessGroupGateway, +) +from services.network_access_group_service import ( + NetworkAccessGroupEntitlementUnavailableError, + NetworkAccessGroupUpstreamError, +) TENANT_ID = "11111111-1111-4111-8111-111111111111" ACCOUNT_ID = "22222222-2222-4222-8222-222222222222" @@ -13,30 +21,30 @@ GROUP_ID = "44444444-4444-4444-8444-444444444444" HEADERS = {"Content-Type": "application/json", "Billing-Api-Secret-Key": "test-secret"} -@pytest.fixture -def billing_config(config_overrides: Callable[..., None]) -> Iterator[None]: - config_overrides(NETWORK_ACCESS_API_URL="") - with ( - patch.object(BillingService, "base_url", "https://billing.internal/v1"), - patch.object(BillingService, "secret_key", "test-secret"), - ): - yield +def _gateway(*, base_url: str = "https://network-access.internal/v1/") -> tuple[NetworkAccessGroupGateway, MagicMock]: + http_client = MagicMock(spec=httpx.Client) + return ( + NetworkAccessGroupGateway( + base_url=base_url, + fallback_base_url="https://billing.internal/v1", + secret_key="test-secret", + http_client=http_client, + ), + http_client, + ) -@pytest.mark.usefixtures("billing_config") -def test_network_access_requests_use_independent_api_url_when_configured( - config_overrides: Callable[..., None], -) -> None: +def test_network_access_requests_use_independent_api_url_when_configured() -> None: + gateway, http_client = _gateway() response = MagicMock(status_code=httpx.codes.OK) response.json.return_value = {"tenant_id": TENANT_ID, "entitled": True, "groups": list[object]()} + http_client.request.return_value = response - config_overrides(NETWORK_ACCESS_API_URL="https://saas.internal/v1/") - with patch("services.billing_service._http_client.request", return_value=response) as request: - BillingService.list_network_access_groups(TENANT_ID, ACCOUNT_ID) + gateway.list_groups(TENANT_ID, ACCOUNT_ID) - request.assert_called_once_with( + http_client.request.assert_called_once_with( "GET", - f"https://saas.internal/v1/tenants/{TENANT_ID}/network-access-groups", + f"https://network-access.internal/v1/tenants/{TENANT_ID}/network-access-groups", json=None, params={"actor_account_id": ACCOUNT_ID}, headers=HEADERS, @@ -44,15 +52,15 @@ def test_network_access_requests_use_independent_api_url_when_configured( ) -@pytest.mark.usefixtures("billing_config") -def test_network_access_requests_fall_back_to_billing_api_url_at_request_time() -> None: +def test_network_access_requests_fall_back_to_billing_api_url() -> None: + gateway, http_client = _gateway(base_url="") response = MagicMock(status_code=httpx.codes.OK) response.json.return_value = {"tenant_id": TENANT_ID, "entitled": True, "groups": list[object]()} + http_client.request.return_value = response - with patch("services.billing_service._http_client.request", return_value=response) as request: - BillingService.list_network_access_groups(TENANT_ID, ACCOUNT_ID) + gateway.list_groups(TENANT_ID, ACCOUNT_ID) - request.assert_called_once_with( + http_client.request.assert_called_once_with( "GET", f"https://billing.internal/v1/tenants/{TENANT_ID}/network-access-groups", json=None, @@ -62,40 +70,21 @@ def test_network_access_requests_fall_back_to_billing_api_url_at_request_time() ) -@pytest.mark.usefixtures("billing_config") -def test_subscription_requests_keep_using_billing_api_url(config_overrides: Callable[..., None]) -> None: - response = MagicMock(status_code=httpx.codes.OK) - response.json.return_value = {"subscription": {"plan": "professional"}} - - config_overrides(NETWORK_ACCESS_API_URL="https://saas.internal/v1") - with patch("services.billing_service._http_client.request", return_value=response) as request: - BillingService._send_request("GET", "/subscription/info", params={"tenant_id": TENANT_ID}) - - request.assert_called_once_with( - "GET", - "https://billing.internal/v1/subscription/info", - json=None, - params={"tenant_id": TENANT_ID}, - headers=HEADERS, - follow_redirects=True, - ) - - -@pytest.mark.usefixtures("billing_config") def test_group_list_and_item_reads_use_authenticated_saas_endpoints() -> None: + gateway, http_client = _gateway(base_url="") response = MagicMock(status_code=httpx.codes.OK) response.json.side_effect = [ {"tenant_id": TENANT_ID, "entitled": True, "groups": list[object]()}, {"group": {"id": GROUP_ID}}, ] + http_client.request.return_value = response - with patch("services.billing_service._http_client.request", return_value=response) as request: - list_result = BillingService.list_network_access_groups(TENANT_ID, ACCOUNT_ID) - item_result = BillingService.get_network_access_group(TENANT_ID, GROUP_ID, ACCOUNT_ID) + list_result = gateway.list_groups(TENANT_ID, ACCOUNT_ID) + item_result = gateway.get_group(TENANT_ID, GROUP_ID, ACCOUNT_ID) assert list_result["groups"] == [] assert item_result["group"]["id"] == GROUP_ID - assert request.call_args_list == [ + assert http_client.request.call_args_list == [ call( "GET", f"https://billing.internal/v1/tenants/{TENANT_ID}/network-access-groups", @@ -115,40 +104,40 @@ def test_group_list_and_item_reads_use_authenticated_saas_endpoints() -> None: ] -@pytest.mark.usefixtures("billing_config") def test_group_mutations_inject_actor_and_use_expected_version() -> None: + gateway, http_client = _gateway(base_url="") response = MagicMock(status_code=httpx.codes.OK) response.json.side_effect = [ {"group": {"id": GROUP_ID, "version": "1"}}, {"group": {"id": GROUP_ID, "version": "2"}}, {"deleted": True}, ] + http_client.request.return_value = response - with patch("services.billing_service._http_client.request", return_value=response) as request: - BillingService.create_network_access_group( - TENANT_ID, - name="Office", - description="Office egress", - allowed_cidrs=["203.0.113.7/32"], - actor_account_id=ACCOUNT_ID, - ) - BillingService.update_network_access_group( - TENANT_ID, - GROUP_ID, - name="Office", - description="Updated office egress", - allowed_cidrs=["203.0.113.0/24"], - expected_version=1, - actor_account_id=ACCOUNT_ID, - ) - BillingService.delete_network_access_group( - TENANT_ID, - GROUP_ID, - expected_version=2, - actor_account_id=ACCOUNT_ID, - ) + gateway.create_group( + TENANT_ID, + name="Office", + description="Office egress", + allowed_cidrs=["203.0.113.7/32"], + actor_account_id=ACCOUNT_ID, + ) + gateway.update_group( + TENANT_ID, + GROUP_ID, + name="Office", + description="Updated office egress", + allowed_cidrs=["203.0.113.0/24"], + expected_version=1, + actor_account_id=ACCOUNT_ID, + ) + gateway.delete_group( + TENANT_ID, + GROUP_ID, + expected_version=2, + actor_account_id=ACCOUNT_ID, + ) - assert request.call_args_list == [ + assert http_client.request.call_args_list == [ call( "POST", f"https://billing.internal/v1/tenants/{TENANT_ID}/network-access-groups", @@ -187,8 +176,8 @@ def test_group_mutations_inject_actor_and_use_expected_version() -> None: ] -@pytest.mark.usefixtures("billing_config") def test_app_config_read_and_atomic_write_use_tenant_and_app_path() -> None: + gateway, http_client = _gateway(base_url="") response = MagicMock(status_code=httpx.codes.OK) response.json.side_effect = [ {"tenant_id": TENANT_ID, "app_id": APP_ID, "entitled": True, "binding": None}, @@ -211,30 +200,30 @@ def test_app_config_read_and_atomic_write_use_tenant_and_app_path() -> None: } }, ] + http_client.request.return_value = response - with patch("services.billing_service._http_client.request", return_value=response) as request: - BillingService.get_app_network_access_group(TENANT_ID, APP_ID, ACCOUNT_ID) - BillingService.update_app_network_access_group( - TENANT_ID, - APP_ID, - enabled=True, - group_id=GROUP_ID, - access_points=["webapp", "service_api"], - expected_version=0, - actor_account_id=ACCOUNT_ID, - ) - BillingService.update_app_network_access_group( - TENANT_ID, - APP_ID, - enabled=False, - group_id=GROUP_ID, - access_points=["webapp", "service_api"], - expected_version=1, - actor_account_id=ACCOUNT_ID, - ) + gateway.get_app_binding(TENANT_ID, APP_ID, ACCOUNT_ID) + gateway.update_app_binding( + TENANT_ID, + APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["webapp", "service_api"], + expected_version=0, + actor_account_id=ACCOUNT_ID, + ) + gateway.update_app_binding( + TENANT_ID, + APP_ID, + enabled=False, + group_id=GROUP_ID, + access_points=["webapp", "service_api"], + expected_version=1, + actor_account_id=ACCOUNT_ID, + ) endpoint = f"https://billing.internal/v1/tenants/{TENANT_ID}/apps/{APP_ID}/network-access-group" - assert request.call_args_list == [ + assert http_client.request.call_args_list == [ call( "GET", endpoint, @@ -274,16 +263,16 @@ def test_app_config_read_and_atomic_write_use_tenant_and_app_path() -> None: ] -@pytest.mark.usefixtures("billing_config") def test_app_lifecycle_cleanup_uses_internal_secret_endpoint() -> None: + gateway, http_client = _gateway(base_url="") response = MagicMock(status_code=httpx.codes.OK) response.json.return_value = {"deleted": True} + http_client.request.return_value = response - with patch("services.billing_service._http_client.request", return_value=response) as request: - result = BillingService.cleanup_app_network_access_group_binding(TENANT_ID, APP_ID) + result = gateway.cleanup_app_binding(TENANT_ID, APP_ID) assert result == {"deleted": True} - request.assert_called_once_with( + http_client.request.assert_called_once_with( "DELETE", f"https://billing.internal/v1/tenants/{TENANT_ID}/apps/{APP_ID}/network-access-group-binding", json=None, @@ -295,51 +284,112 @@ def test_app_lifecycle_cleanup_uses_internal_secret_endpoint() -> None: @pytest.mark.parametrize("status_code", [400, 403, 404, 409, 500, 503]) def test_network_access_group_request_preserves_upstream_status(status_code: int) -> None: + gateway, _ = _gateway() response = MagicMock(status_code=status_code) with ( - patch.object(BillingService, "_send_network_access_group_http_request", return_value=response), + patch.object(gateway, "_send_http_request", return_value=response), pytest.raises(NetworkAccessGroupUpstreamError) as exc_info, ): - BillingService._send_network_access_group_request("GET", "/groups") + gateway._send_request("GET", "/groups") assert exc_info.value.status_code == status_code -def test_network_access_group_request_preserves_whitelisted_upstream_reason() -> None: +def test_network_access_group_request_preserves_upstream_reason() -> None: + gateway, _ = _gateway() response = MagicMock(status_code=httpx.codes.CONFLICT) response.json.return_value = {"code": 409, "reason": "NETWORK_ACCESS_VERSION_CONFLICT", "message": "internal"} with ( - patch.object(BillingService, "_send_network_access_group_http_request", return_value=response), + patch.object(gateway, "_send_http_request", return_value=response), pytest.raises(NetworkAccessGroupUpstreamError) as exc_info, ): - BillingService._send_network_access_group_request("DELETE", "/groups/id") + gateway._send_request("DELETE", "/groups/id") assert exc_info.value.status_code == httpx.codes.CONFLICT assert exc_info.value.reason == "NETWORK_ACCESS_VERSION_CONFLICT" def test_network_access_group_request_maps_transport_failure_to_service_unavailable() -> None: - request = httpx.Request("GET", "https://billing.internal/v1/groups") + gateway, _ = _gateway() + request = httpx.Request("GET", "https://network-access.internal/v1/groups") with ( patch.object( - BillingService, - "_send_network_access_group_http_request", + gateway, + "_send_http_request", side_effect=httpx.ConnectError("unavailable", request=request), ), pytest.raises(NetworkAccessGroupUpstreamError) as exc_info, ): - BillingService._send_network_access_group_request("GET", "/groups") + gateway._send_request("GET", "/groups") assert exc_info.value.status_code == httpx.codes.SERVICE_UNAVAILABLE def test_network_access_group_request_rejects_non_object_response() -> None: + gateway, _ = _gateway() response = MagicMock(status_code=httpx.codes.OK) response.json.return_value = list[object]() with ( - patch.object(BillingService, "_send_network_access_group_http_request", return_value=response), + patch.object(gateway, "_send_http_request", return_value=response), pytest.raises(NetworkAccessGroupUpstreamError) as exc_info, ): - BillingService._send_network_access_group_request("GET", "/groups") + gateway._send_request("GET", "/groups") assert exc_info.value.status_code == httpx.codes.BAD_GATEWAY + + +@pytest.mark.parametrize("plan", [CloudPlan.PROFESSIONAL, CloudPlan.TEAM]) +def test_billing_entitlement_gateway_accepts_paid_plans(plan: CloudPlan) -> None: + gateway = BillingNetworkAccessGroupEntitlementGateway() + with patch( + "services.network_access_group_gateway.BillingService.get_info", + return_value={"subscription": {"plan": plan}}, + ) as get_info: + assert gateway.is_paid_plan(TENANT_ID) is True + + get_info.assert_called_once_with(TENANT_ID, exclude_vector_space=True) + + +def test_billing_entitlement_gateway_rejects_unpaid_plan() -> None: + gateway = BillingNetworkAccessGroupEntitlementGateway() + with patch( + "services.network_access_group_gateway.BillingService.get_info", + return_value={"subscription": {"plan": CloudPlan.SANDBOX}}, + ): + assert gateway.is_paid_plan(TENANT_ID) is False + + +def test_billing_entitlement_gateway_maps_unavailable_billing_service() -> None: + gateway = BillingNetworkAccessGroupEntitlementGateway() + with ( + patch( + "services.network_access_group_gateway.BillingService.get_info", + side_effect=httpx.ConnectError("unavailable"), + ), + pytest.raises(NetworkAccessGroupEntitlementUnavailableError), + ): + gateway.is_paid_plan(TENANT_ID) + + +def test_billing_entitlement_gateway_maps_billing_http_status_error() -> None: + gateway = BillingNetworkAccessGroupEntitlementGateway() + with ( + patch( + "services.network_access_group_gateway.BillingService.get_info", + side_effect=_BillingHTTPStatusError("unavailable", 503), + ), + pytest.raises(NetworkAccessGroupEntitlementUnavailableError), + ): + gateway.is_paid_plan(TENANT_ID) + + +def test_billing_entitlement_gateway_does_not_mask_unexpected_value_error() -> None: + gateway = BillingNetworkAccessGroupEntitlementGateway() + with ( + patch( + "services.network_access_group_gateway.BillingService.get_info", + side_effect=ValueError("programming error"), + ), + pytest.raises(ValueError, match="programming error"), + ): + gateway.is_paid_plan(TENANT_ID) diff --git a/api/tests/unit_tests/services/test_network_access_group_service.py b/api/tests/unit_tests/services/test_network_access_group_service.py new file mode 100644 index 00000000000..2e4a948cd1a --- /dev/null +++ b/api/tests/unit_tests/services/test_network_access_group_service.py @@ -0,0 +1,444 @@ +from collections.abc import Callable +from typing import NamedTuple +from unittest.mock import MagicMock, create_autospec + +import pytest + +from machinery.context import RequestContext +from services.network_access_group_service import ( + NetworkAccessGroupAccessDeniedError, + NetworkAccessGroupAppNotFoundError, + NetworkAccessGroupAppQuery, + NetworkAccessGroupAppQueryError, + NetworkAccessGroupAppRecord, + NetworkAccessGroupControlPlane, + NetworkAccessGroupEntitlement, + NetworkAccessGroupService, + NetworkAccessGroupUnsupportedAccessPointsError, + NetworkAccessGroupUnsupportedAppModeError, + WorkspaceMembershipRoleQuery, +) + +WORKSPACE_ID = "workspace-1" +ACCOUNT_ID = "account-1" +APP_ID = "app-1" +GROUP_ID = "group-1" + + +class _Harness(NamedTuple): + service: NetworkAccessGroupService + control_plane: MagicMock + apps: MagicMock + memberships: MagicMock + entitlement: MagicMock + + +def _context() -> RequestContext: + return RequestContext( + request_id="request-1", + trace_id=None, + account_id=ACCOUNT_ID, + active_workspace_id=WORKSPACE_ID, + ) + + +def _app(*, app_id: str = APP_ID, mode: str = "workflow", name: str = "App") -> NetworkAccessGroupAppRecord: + return NetworkAccessGroupAppRecord( + id=app_id, + mode=mode, + name=name, + icon=f"{app_id}.png", + icon_type="image", + icon_background="#FFFFFF", + ) + + +def _harness(*, role: str | None = "owner", paid: bool = True, app_mode: str = "workflow") -> _Harness: + control_plane = create_autospec(NetworkAccessGroupControlPlane, instance=True, spec_set=True) + apps = create_autospec(NetworkAccessGroupAppQuery, instance=True, spec_set=True) + memberships = create_autospec(WorkspaceMembershipRoleQuery, instance=True, spec_set=True) + entitlement = create_autospec(NetworkAccessGroupEntitlement, instance=True, spec_set=True) + memberships.get_role_for_account.return_value = role + entitlement.is_paid_plan.return_value = paid + apps.get_manageable_app.return_value = _app(mode=app_mode) + return _Harness( + service=NetworkAccessGroupService( + control_plane=control_plane, + apps=apps, + memberships=memberships, + entitlement=entitlement, + ), + control_plane=control_plane, + apps=apps, + memberships=memberships, + entitlement=entitlement, + ) + + +@pytest.mark.parametrize("role", ["normal", None]) +def test_reads_reject_non_privileged_persisted_roles_before_control_plane(role: str | None) -> None: + harness = _harness(role=role) + + with pytest.raises(NetworkAccessGroupAccessDeniedError): + harness.service.list_groups(_context()) + + harness.memberships.get_role_for_account.assert_called_once_with( + workspace_id=WORKSPACE_ID, + account_id=ACCOUNT_ID, + ) + assert harness.control_plane.method_calls == [] + + +Mutation = Callable[[NetworkAccessGroupService, RequestContext], object] + + +@pytest.mark.parametrize( + "mutation", + [ + pytest.param( + lambda service, context: service.create_group( + context, + name="Office", + description="Office network", + allowed_cidrs=["203.0.113.0/24"], + ), + id="create-group", + ), + pytest.param( + lambda service, context: service.update_group( + context, + group_id=GROUP_ID, + name="Office", + description="Office network", + allowed_cidrs=["203.0.113.0/24"], + expected_version=1, + ), + id="update-group", + ), + pytest.param( + lambda service, context: service.delete_group( + context, + group_id=GROUP_ID, + expected_version=1, + ), + id="delete-group", + ), + pytest.param( + lambda service, context: service.update_app_binding( + context, + app_id=APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["webapp"], + expected_version=1, + ), + id="update-app-binding", + ), + ], +) +@pytest.mark.parametrize("role", ["normal", None]) +def test_mutations_reject_non_privileged_persisted_roles_before_control_plane( + role: str | None, + mutation: Mutation, +) -> None: + harness = _harness(role=role) + + with pytest.raises(NetworkAccessGroupAccessDeniedError): + mutation(harness.service, _context()) + + harness.memberships.get_role_for_account.assert_called_once_with( + workspace_id=WORKSPACE_ID, + account_id=ACCOUNT_ID, + ) + assert harness.control_plane.method_calls == [] + + +@pytest.mark.parametrize("role", ["owner", "admin"]) +def test_reads_allow_persisted_owner_and_admin(role: str) -> None: + harness = _harness(role=role) + harness.control_plane.list_groups.return_value = {"entitled": True, "groups": list[object]()} + + harness.service.list_groups(_context()) + + harness.control_plane.list_groups.assert_called_once_with(WORKSPACE_ID, ACCOUNT_ID) + + +def test_mutation_does_not_repeat_http_paid_plan_admission() -> None: + harness = _harness() + harness.control_plane.create_group.return_value = {"group": {"id": GROUP_ID}} + + harness.service.create_group( + _context(), + name="Office", + description="Office network", + allowed_cidrs=["203.0.113.0/24"], + ) + + harness.entitlement.is_paid_plan.assert_not_called() + + +def test_public_use_cases_forward_workspace_actor_and_request_parameters() -> None: + harness = _harness() + harness.control_plane.list_groups.return_value = {"entitled": True, "groups": list[object]()} + harness.control_plane.create_group.return_value = {"group": {"id": GROUP_ID}} + harness.control_plane.get_group.return_value = {"group": {"id": GROUP_ID}} + harness.control_plane.update_group.return_value = {"group": {"id": GROUP_ID}} + harness.control_plane.delete_group.return_value = {"deleted": True} + harness.control_plane.get_app_binding.return_value = {"entitled": True, "binding": None} + harness.control_plane.update_app_binding.return_value = dict[str, object](binding=None) + context = _context() + + harness.service.list_groups(context) + harness.service.create_group( + context, + name="Office", + description="Office network", + allowed_cidrs=["203.0.113.0/24"], + ) + harness.service.get_group(context, group_id=GROUP_ID) + harness.service.update_group( + context, + group_id=GROUP_ID, + name="Office v2", + description="Updated network", + allowed_cidrs=["203.0.113.7/32"], + expected_version=2, + ) + harness.service.delete_group(context, group_id=GROUP_ID, expected_version=3) + harness.service.get_app_binding(context, app_id=APP_ID) + harness.service.update_app_binding( + context, + app_id=APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["webapp", "service_api"], + expected_version=4, + ) + + harness.control_plane.list_groups.assert_called_once_with(WORKSPACE_ID, ACCOUNT_ID) + harness.control_plane.create_group.assert_called_once_with( + WORKSPACE_ID, + name="Office", + description="Office network", + allowed_cidrs=["203.0.113.0/24"], + actor_account_id=ACCOUNT_ID, + ) + harness.control_plane.get_group.assert_called_once_with(WORKSPACE_ID, GROUP_ID, ACCOUNT_ID) + harness.control_plane.update_group.assert_called_once_with( + WORKSPACE_ID, + GROUP_ID, + name="Office v2", + description="Updated network", + allowed_cidrs=["203.0.113.7/32"], + expected_version=2, + actor_account_id=ACCOUNT_ID, + ) + harness.control_plane.delete_group.assert_called_once_with( + WORKSPACE_ID, + GROUP_ID, + expected_version=3, + actor_account_id=ACCOUNT_ID, + ) + harness.control_plane.get_app_binding.assert_called_once_with(WORKSPACE_ID, APP_ID, ACCOUNT_ID) + harness.control_plane.update_app_binding.assert_called_once_with( + WORKSPACE_ID, + APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["webapp", "service_api"], + expected_version=4, + actor_account_id=ACCOUNT_ID, + ) + + +def test_cleanup_app_binding_forwards_without_console_admission() -> None: + harness = _harness() + harness.control_plane.cleanup_app_binding.return_value = {"deleted": True} + + result = harness.service.cleanup_app_binding(workspace_id=WORKSPACE_ID, app_id=APP_ID) + + assert result == {"deleted": True} + harness.control_plane.cleanup_app_binding.assert_called_once_with(WORKSPACE_ID, APP_ID) + harness.memberships.get_role_for_account.assert_not_called() + harness.entitlement.is_paid_plan.assert_not_called() + + +@pytest.mark.parametrize( + ("upstream_entitled", "paid", "expected"), + [ + (False, False, False), + (False, True, False), + (True, False, False), + (True, True, True), + ], +) +def test_list_effective_entitlement_requires_both_sources( + upstream_entitled: bool, + paid: bool, + expected: bool, +) -> None: + harness = _harness(paid=paid) + harness.control_plane.list_groups.return_value = { + "entitled": upstream_entitled, + "groups": list[object](), + } + + result = harness.service.list_groups(_context()) + + assert result["entitled"] is expected + + +@pytest.mark.parametrize( + ("mode", "available_access_points"), + [ + ("workflow", ["webapp", "service_api", "mcp", "trigger"]), + ("advanced-chat", ["webapp", "service_api", "mcp"]), + ("chat", ["webapp", "service_api", "mcp"]), + ("completion", ["webapp", "service_api", "mcp"]), + ("agent-chat", ["webapp", "service_api", "mcp"]), + ("agent", ["webapp", "service_api"]), + ], +) +def test_app_binding_reports_access_points_for_each_supported_mode( + mode: str, + available_access_points: list[str], +) -> None: + harness = _harness(app_mode=mode) + harness.control_plane.get_app_binding.return_value = {"entitled": True, "binding": None} + + result = harness.service.get_app_binding(_context(), app_id=APP_ID) + + assert result["available_access_points"] == available_access_points + + +def test_app_binding_rejects_unsupported_app_mode_before_control_plane() -> None: + harness = _harness(app_mode="site") + + with pytest.raises(NetworkAccessGroupUnsupportedAppModeError, match="'site'"): + harness.service.get_app_binding(_context(), app_id=APP_ID) + + harness.control_plane.get_app_binding.assert_not_called() + + +def test_app_binding_rejects_missing_or_inaccessible_app_before_control_plane() -> None: + harness = _harness() + harness.apps.get_manageable_app.return_value = None + + with pytest.raises(NetworkAccessGroupAppNotFoundError): + harness.service.get_app_binding(_context(), app_id=APP_ID) + + harness.apps.get_manageable_app.assert_called_once_with(workspace_id=WORKSPACE_ID, app_id=APP_ID) + harness.memberships.get_role_for_account.assert_not_called() + harness.control_plane.get_app_binding.assert_not_called() + + +def test_update_app_binding_rejects_missing_or_inaccessible_app_before_control_plane() -> None: + harness = _harness() + harness.apps.get_manageable_app.return_value = None + + with pytest.raises(NetworkAccessGroupAppNotFoundError): + harness.service.update_app_binding( + _context(), + app_id=APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["webapp"], + expected_version=1, + ) + + harness.apps.get_manageable_app.assert_called_once_with(workspace_id=WORKSPACE_ID, app_id=APP_ID) + harness.memberships.get_role_for_account.assert_not_called() + harness.control_plane.update_app_binding.assert_not_called() + + +def test_update_app_binding_rejects_access_points_not_supported_by_mode() -> None: + harness = _harness(app_mode="agent") + + with pytest.raises(NetworkAccessGroupUnsupportedAccessPointsError) as exc_info: + harness.service.update_app_binding( + _context(), + app_id=APP_ID, + enabled=True, + group_id=GROUP_ID, + access_points=["trigger", "mcp"], + expected_version=1, + ) + + assert exc_info.value.access_points == ("mcp", "trigger") + harness.control_plane.update_app_binding.assert_not_called() + + +def test_app_binding_materializes_protojson_defaults() -> None: + harness = _harness(app_mode="chat") + harness.control_plane.get_app_binding.return_value = { + "entitled": True, + "binding": dict[str, object](), + } + + result = harness.service.get_app_binding(_context(), app_id=APP_ID) + + assert result["binding"] == {"enabled": False, "access_points": []} + + +def test_app_binding_normalizes_camel_case_and_filters_stale_access_points() -> None: + harness = _harness(app_mode="chat") + harness.control_plane.get_app_binding.return_value = { + "entitled": True, + "binding": {"enabled": True, "accessPoints": ["webapp", "trigger", "mcp"]}, + } + + result = harness.service.get_app_binding(_context(), app_id=APP_ID) + + assert result["binding"] == {"enabled": True, "access_points": ["webapp", "mcp"]} + + +def test_group_payloads_are_enriched_with_tenant_scoped_app_metadata() -> None: + harness = _harness() + harness.control_plane.list_groups.return_value = { + "entitled": True, + "groups": [ + { + "id": "group-1", + "usedByAppIds": ["app-2", "missing-app", "app-1"], + "usedByCount": "3", + }, + {"id": "group-2", "app_ids": ["app-1"]}, + ], + } + harness.apps.list_apps.return_value = [ + _app(app_id="app-1", name="One"), + _app(app_id="app-2", name="Two"), + ] + + result = harness.service.list_groups(_context()) + + groups = result["groups"] + assert groups[0]["app_ids"] == ["app-2", "missing-app", "app-1"] + assert groups[0]["used_by_count"] == "3" + assert [app["id"] for app in groups[0]["apps"]] == ["app-2", "app-1"] + assert groups[1]["used_by_count"] == 1 + assert groups[1]["apps"] == [ + { + "id": "app-1", + "name": "One", + "icon": "app-1.png", + "icon_type": "image", + "icon_background": "#FFFFFF", + } + ] + call_kwargs = harness.apps.list_apps.call_args.kwargs + assert call_kwargs["workspace_id"] == WORKSPACE_ID + assert set(call_kwargs["app_ids"]) == {"app-1", "app-2", "missing-app"} + + +def test_app_query_failure_degrades_group_enrichment_to_empty_apps() -> None: + harness = _harness() + harness.control_plane.get_group.return_value = { + "group": {"id": GROUP_ID, "used_by_app_ids": [APP_ID], "used_by_count": 1} + } + harness.apps.list_apps.side_effect = NetworkAccessGroupAppQueryError + + result = harness.service.get_group(_context(), group_id=GROUP_ID) + + assert result["group"]["app_ids"] == [APP_ID] + assert result["group"]["apps"] == [] diff --git a/api/tests/unit_tests/tasks/test_remove_app_and_related_data_task.py b/api/tests/unit_tests/tasks/test_remove_app_and_related_data_task.py index e489c0598c5..30e6568407e 100644 --- a/api/tests/unit_tests/tasks/test_remove_app_and_related_data_task.py +++ b/api/tests/unit_tests/tasks/test_remove_app_and_related_data_task.py @@ -1,5 +1,6 @@ import logging from datetime import UTC, datetime +from types import SimpleNamespace from unittest.mock import MagicMock, call, patch from uuid import uuid4 @@ -14,7 +15,7 @@ from models import AppStar from models.agent import WorkflowAgentBindingType, WorkflowAgentNodeBinding from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom from models.workflow import WorkflowArchiveLog -from services.billing_service import NetworkAccessGroupUpstreamError +from services.network_access_group_service import NetworkAccessGroupUpstreamError from tasks.remove_app_and_related_data_task import ( _delete_app_stars, _delete_app_workflow_archive_logs, @@ -152,17 +153,20 @@ def test_community_app_cleanup_does_not_schedule_binding_cleanup(monkeypatch: py def test_independent_binding_cleanup_task_succeeds(monkeypatch: pytest.MonkeyPatch) -> None: - cleanup = MagicMock(return_value={"deleted": True}) + service = MagicMock() + service.cleanup_app_binding.return_value = {"deleted": True} apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.CLOUD) - monkeypatch.setattr( - remove_app_task_module.BillingService, - "cleanup_app_network_access_group_binding", - cleanup, - ) - remove_app_task_module.cleanup_app_network_access_group_binding_task.run(tenant_id="tenant-1", app_id="app-1") + with patch( + "extensions.ext_application_services.application_services", + return_value=SimpleNamespace(network_access_groups=service), + ): + remove_app_task_module.cleanup_app_network_access_group_binding_task.run( + tenant_id="tenant-1", + app_id="app-1", + ) - cleanup.assert_called_once_with("tenant-1", "app-1") + service.cleanup_app_binding.assert_called_once_with(workspace_id="tenant-1", app_id="app-1") assert remove_app_task_module.cleanup_app_network_access_group_binding_task.max_retries == 24 @@ -172,16 +176,22 @@ def test_independent_binding_cleanup_task_retries_with_exponential_backoff( upstream_error = NetworkAccessGroupUpstreamError(409, "NETWORK_ACCESS_APP_STILL_EXISTS") retry_error = RuntimeError("retry scheduled") retry = MagicMock(side_effect=retry_error) + service = MagicMock() + service.cleanup_app_binding.side_effect = upstream_error apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.CLOUD) - monkeypatch.setattr( - remove_app_task_module.BillingService, - "cleanup_app_network_access_group_binding", - MagicMock(side_effect=upstream_error), - ) monkeypatch.setattr(remove_app_task_module.cleanup_app_network_access_group_binding_task, "retry", retry) - with pytest.raises(RuntimeError, match="retry scheduled"): - remove_app_task_module.cleanup_app_network_access_group_binding_task.run(tenant_id="tenant-1", app_id="app-1") + with ( + patch( + "extensions.ext_application_services.application_services", + return_value=SimpleNamespace(network_access_groups=service), + ), + pytest.raises(RuntimeError, match="retry scheduled"), + ): + remove_app_task_module.cleanup_app_network_access_group_binding_task.run( + tenant_id="tenant-1", + app_id="app-1", + ) retry.assert_called_once_with(exc=upstream_error, countdown=15) assert remove_app_task_module._network_access_binding_cleanup_retry_delay(5) == 480