Feat/plugin install scope management (#19963)

api/core/plugin/entities/plugin_daemon.py

     task_id: str = Field(description="The ID of the install task.")
 
 
-class PluginUploadResponse(BaseModel):
+class PluginVerification(BaseModel):
+    """
+    Verification of the plugin.
+    """
+
+    class AuthorizedCategory(StrEnum):
+        Langgenius = "langgenius"
+        Partner = "partner"
+        Community = "community"
+
+    authorized_category: AuthorizedCategory = Field(description="The authorized category of the plugin.")
+
+
+class PluginDecodeResponse(BaseModel):
     unique_identifier: str = Field(description="The unique identifier of the plugin.")
     manifest: PluginDeclaration
+    verification: Optional[PluginVerification] = Field(default=None, description="Basic verification information")
 
 
 class PluginOAuthAuthorizationUrlResponse(BaseModel):

api/core/plugin/impl/plugin.py

     PluginInstallationSource,
 )
 from core.plugin.entities.plugin_daemon import (
+    PluginDecodeResponse,
     PluginInstallTask,
     PluginInstallTaskStartResponse,
     PluginListResponse,
-    PluginUploadResponse,
 )
 from core.plugin.impl.base import BasePluginClient
 
         tenant_id: str,
         pkg: bytes,
         verify_signature: bool = False,
-    ) -> PluginUploadResponse:
+    ) -> PluginDecodeResponse:
         """
         Upload a plugin package and return the plugin unique identifier.
         """
         return self._request_with_plugin_daemon_response(
             "POST",
             f"plugin/{tenant_id}/management/install/upload/package",
-            PluginUploadResponse,
+            PluginDecodeResponse,
             files=body,
             data=data,
         )
             params={"plugin_unique_identifier": plugin_unique_identifier},
         )
 
+    def decode_plugin_from_identifier(self, tenant_id: str, plugin_unique_identifier: str) -> PluginDecodeResponse:
+        """
+        Decode a plugin from an identifier.
+        """
+        return self._request_with_plugin_daemon_response(
+            "GET",
+            f"plugin/{tenant_id}/management/decode/from_identifier",
+            PluginDecodeResponse,
+            data={"plugin_unique_identifier": plugin_unique_identifier},
+            headers={"Content-Type": "application/json"},
+        )
+
     def fetch_plugin_installation_by_ids(
         self, tenant_id: str, plugin_ids: Sequence[str]
     ) -> Sequence[PluginInstallation]:

api/models/model.py

 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.signature import sign_tool_file
 from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
-from services.plugin.plugin_service import PluginService
 
 if TYPE_CHECKING:
     from models.workflow import Workflow
     @property
     def deleted_tools(self) -> list:
         from core.tools.tool_manager import ToolManager
+        from services.plugin.plugin_service import PluginService
 
         # get agent mode tools
         app_model_config = self.app_model_config

api/services/errors/plugin.py

+from services.errors.base import BaseServiceError
+
+
+class PluginInstallationForbiddenError(BaseServiceError):
+    pass

api/services/feature_service.py

     allow_email_password_login: bool = False
 
 
+class PluginInstallationScope(StrEnum):
+    NONE = "none"
+    OFFICIAL_ONLY = "official_only"
+    OFFICIAL_AND_SPECIFIC_PARTNERS = "official_and_specific_partners"
+    ALL = "all"
+
+
+class PluginInstallationPermissionModel(BaseModel):
+    # Plugin installation scope – possible values:
+    #   none: prohibit all plugin installations
+    #   official_only: allow only Dify official plugins
+    #   official_and_specific_partners: allow official and specific partner plugins
+    #   all: allow installation of all plugins
+    plugin_installation_scope: PluginInstallationScope = PluginInstallationScope.ALL
+
+    # If True, restrict plugin installation to the marketplace only
+    # Equivalent to ForceEnablePluginVerification
+    restrict_to_marketplace_only: bool = False
+
+
 class FeatureModel(BaseModel):
     billing: BillingModel = BillingModel()
     education: EducationModel = EducationModel()
     license: LicenseModel = LicenseModel()
     branding: BrandingModel = BrandingModel()
     webapp_auth: WebAppAuthModel = WebAppAuthModel()
+    plugin_installation_permission: PluginInstallationPermissionModel = PluginInstallationPermissionModel()
 
 
 class FeatureService:
                 features.license.workspaces.enabled = license_info["workspaces"]["enabled"]
                 features.license.workspaces.limit = license_info["workspaces"]["limit"]
                 features.license.workspaces.size = license_info["workspaces"]["used"]
+
+        if "PluginInstallationPermission" in enterprise_info:
+            plugin_installation_info = enterprise_info["PluginInstallationPermission"]
+            features.plugin_installation_permission.plugin_installation_scope = plugin_installation_info[
+                "pluginInstallationScope"
+            ]
+            features.plugin_installation_permission.restrict_to_marketplace_only = plugin_installation_info[
+                "restrictToMarketplaceOnly"
+            ]

api/services/plugin/plugin_service.py

     PluginInstallation,
     PluginInstallationSource,
 )
-from core.plugin.entities.plugin_daemon import PluginInstallTask, PluginListResponse, PluginUploadResponse
+from core.plugin.entities.plugin_daemon import (
+    PluginDecodeResponse,
+    PluginInstallTask,
+    PluginListResponse,
+    PluginVerification,
+)
 from core.plugin.impl.asset import PluginAssetManager
 from core.plugin.impl.debugging import PluginDebuggingClient
 from core.plugin.impl.plugin import PluginInstaller
 from extensions.ext_redis import redis_client
+from services.errors.plugin import PluginInstallationForbiddenError
+from services.feature_service import FeatureService, PluginInstallationScope
 
 logger = logging.getLogger(__name__)
 
             logger.exception("failed to fetch latest plugin version")
             return result
 
+    @staticmethod
+    def _check_marketplace_only_permission():
+        """
+        Check if the marketplace only permission is enabled
+        """
+        features = FeatureService.get_system_features()
+        if features.plugin_installation_permission.restrict_to_marketplace_only:
+            raise PluginInstallationForbiddenError("Plugin installation is restricted to marketplace only")
+
+    @staticmethod
+    def _check_plugin_installation_scope(plugin_verification: Optional[PluginVerification]):
+        """
+        Check the plugin installation scope
+        """
+        features = FeatureService.get_system_features()
+
+        match features.plugin_installation_permission.plugin_installation_scope:
+            case PluginInstallationScope.OFFICIAL_ONLY:
+                if (
+                    plugin_verification is None
+                    or plugin_verification.authorized_category != PluginVerification.AuthorizedCategory.Langgenius
+                ):
+                    raise PluginInstallationForbiddenError("Plugin installation is restricted to official only")
+            case PluginInstallationScope.OFFICIAL_AND_SPECIFIC_PARTNERS:
+                if plugin_verification is None or plugin_verification.authorized_category not in [
+                    PluginVerification.AuthorizedCategory.Langgenius,
+                    PluginVerification.AuthorizedCategory.Partner,
+                ]:
+                    raise PluginInstallationForbiddenError(
+                        "Plugin installation is restricted to official and specific partners"
+                    )
+            case PluginInstallationScope.NONE:
+                raise PluginInstallationForbiddenError("Installing plugins is not allowed")
+            case PluginInstallationScope.ALL:
+                pass
+
     @staticmethod
     def get_debugging_key(tenant_id: str) -> str:
         """
         # check if plugin pkg is already downloaded
         manager = PluginInstaller()
 
+        features = FeatureService.get_system_features()
+
         try:
             manager.fetch_plugin_manifest(tenant_id, new_plugin_unique_identifier)
             # already downloaded, skip, and record install event
         except Exception:
             # plugin not installed, download and upload pkg
             pkg = download_plugin_pkg(new_plugin_unique_identifier)
-            manager.upload_pkg(tenant_id, pkg, verify_signature=False)
+            response = manager.upload_pkg(
+                tenant_id,
+                pkg,
+                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            )
+
+            # check if the plugin is available to install
+            PluginService._check_plugin_installation_scope(response.verification)
 
         return manager.upgrade_plugin(
             tenant_id,
         """
         Upgrade plugin with github
         """
+        PluginService._check_marketplace_only_permission()
         manager = PluginInstaller()
         return manager.upgrade_plugin(
             tenant_id,
         )
 
     @staticmethod
-    def upload_pkg(tenant_id: str, pkg: bytes, verify_signature: bool = False) -> PluginUploadResponse:
+    def upload_pkg(tenant_id: str, pkg: bytes, verify_signature: bool = False) -> PluginDecodeResponse:
         """
         Upload plugin package files
 
         returns: plugin_unique_identifier
         """
+        PluginService._check_marketplace_only_permission()
         manager = PluginInstaller()
-        return manager.upload_pkg(tenant_id, pkg, verify_signature)
+        features = FeatureService.get_system_features()
+        response = manager.upload_pkg(
+            tenant_id,
+            pkg,
+            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+        )
+        return response
 
     @staticmethod
     def upload_pkg_from_github(
         tenant_id: str, repo: str, version: str, package: str, verify_signature: bool = False
-    ) -> PluginUploadResponse:
+    ) -> PluginDecodeResponse:
         """
         Install plugin from github release package files,
         returns plugin_unique_identifier
         """
+        PluginService._check_marketplace_only_permission()
         pkg = download_with_size_limit(
             f"https://github.com/{repo}/releases/download/{version}/{package}", dify_config.PLUGIN_MAX_PACKAGE_SIZE
         )
+        features = FeatureService.get_system_features()
 
         manager = PluginInstaller()
-        return manager.upload_pkg(
+        response = manager.upload_pkg(
             tenant_id,
             pkg,
-            verify_signature,
+            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
         )
+        return response
 
     @staticmethod
     def upload_bundle(
         Upload a plugin bundle and return the dependencies.
         """
         manager = PluginInstaller()
+        PluginService._check_marketplace_only_permission()
         return manager.upload_bundle(tenant_id, bundle, verify_signature)
 
     @staticmethod
     def install_from_local_pkg(tenant_id: str, plugin_unique_identifiers: Sequence[str]):
+        PluginService._check_marketplace_only_permission()
+
         manager = PluginInstaller()
+
         return manager.install_from_identifiers(
             tenant_id,
             plugin_unique_identifiers,
         Install plugin from github release package files,
         returns plugin_unique_identifier
         """
+        PluginService._check_marketplace_only_permission()
+
         manager = PluginInstaller()
         return manager.install_from_identifiers(
             tenant_id,
         )
 
     @staticmethod
-    def fetch_marketplace_pkg(
-        tenant_id: str, plugin_unique_identifier: str, verify_signature: bool = False
-    ) -> PluginDeclaration:
+    def fetch_marketplace_pkg(tenant_id: str, plugin_unique_identifier: str) -> PluginDeclaration:
         """
         Fetch marketplace package
         """
         if not dify_config.MARKETPLACE_ENABLED:
             raise ValueError("marketplace is not enabled")
 
+        features = FeatureService.get_system_features()
+
         manager = PluginInstaller()
         try:
             declaration = manager.fetch_plugin_manifest(tenant_id, plugin_unique_identifier)
         except Exception:
             pkg = download_plugin_pkg(plugin_unique_identifier)
-            declaration = manager.upload_pkg(tenant_id, pkg, verify_signature).manifest
+            response = manager.upload_pkg(
+                tenant_id,
+                pkg,
+                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            )
+            # check if the plugin is available to install
+            PluginService._check_plugin_installation_scope(response.verification)
+            declaration = response.manifest
 
         return declaration
 
     @staticmethod
-    def install_from_marketplace_pkg(
-        tenant_id: str, plugin_unique_identifiers: Sequence[str], verify_signature: bool = False
-    ):
+    def install_from_marketplace_pkg(tenant_id: str, plugin_unique_identifiers: Sequence[str]):
         """
         Install plugin from marketplace package files,
         returns installation task id
 
         manager = PluginInstaller()
 
+        features = FeatureService.get_system_features()
+
         # check if already downloaded
         for plugin_unique_identifier in plugin_unique_identifiers:
             try:
                 manager.fetch_plugin_manifest(tenant_id, plugin_unique_identifier)
+                plugin_decode_response = manager.decode_plugin_from_identifier(tenant_id, plugin_unique_identifier)
+                # check if the plugin is available to install
+                PluginService._check_plugin_installation_scope(plugin_decode_response.verification)
                 # already downloaded, skip
             except Exception:
                 # plugin not installed, download and upload pkg
                 pkg = download_plugin_pkg(plugin_unique_identifier)
-                manager.upload_pkg(tenant_id, pkg, verify_signature)
+                response = manager.upload_pkg(
+                    tenant_id,
+                    pkg,
+                    verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                )
+                # check if the plugin is available to install
+                PluginService._check_plugin_installation_scope(response.verification)
 
         return manager.install_from_identifiers(
             tenant_id,