fix(plugin): pin daemon dispatches to package identifiers

api/controllers/console/datasets/rag_pipeline/datasource_auth.py

@@ -180,7 +180,7 @@ def get(self, current_tenant_id: str, current_user: Account, provider_id: str):
             tenant_id=tenant_id,
             datasource_provider_id=datasource_provider_id,
         )
-        if not oauth_config:
+        if not oauth_config or not oauth_config.credentials:
             raise ValueError(f"No OAuth Client Config for {provider_id}")
 
         # Visibility is chosen by the user in the frontend before the redirect,
@@ -203,7 +203,10 @@ def get(self, current_tenant_id: str, current_user: Account, provider_id: str):
             plugin_id=plugin_id,
             provider=provider_name,
             credential_id=credential_id,
-            extra_data={"visibility": requested_visibility.value},
+            extra_data={
+                "visibility": requested_visibility.value,
+                "plugin_unique_identifier": oauth_config.plugin_unique_identifier,
+            },
         )
         oauth_handler = OAuthHandler()
         redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/callback"
@@ -213,7 +216,8 @@ def get(self, current_tenant_id: str, current_user: Account, provider_id: str):
             plugin_id=plugin_id,
             provider=provider_name,
             redirect_uri=redirect_uri,
-            system_credentials=oauth_config,
+            system_credentials=dict(oauth_config.credentials),
+            plugin_unique_identifier=oauth_config.plugin_unique_identifier,
         )
         # response-contract:ignore cookie-bearing Flask response
         response = make_response(dump_response(PluginOAuthAuthorizationUrlResponse, authorization_url_response))
@@ -251,8 +255,18 @@ def get(self, provider_id: str):
             tenant_id=tenant_id,
             datasource_provider_id=datasource_provider_id,
         )
-        if not oauth_client_params:
+        if not oauth_client_params or not oauth_client_params.credentials:
             raise NotFound()
+        try:
+            OAuthProxyService.validate_plugin_context(
+                context,
+                plugin_id=plugin_id,
+                provider=datasource_provider_id.provider_name,
+                plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
+            )
+        except ValueError as error:
+            raise Forbidden(str(error)) from error
+
         redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/datasource/callback"
         oauth_handler = OAuthHandler()
         oauth_response = oauth_handler.get_credentials(
@@ -261,7 +275,8 @@ def get(self, provider_id: str):
             plugin_id=plugin_id,
             provider=datasource_provider_id.provider_name,
             redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
+            system_credentials=dict(oauth_client_params.credentials),
+            plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
             request=request,
         )
         credential_id: str | None = context.get("credential_id")

api/controllers/console/workspace/tool_providers.py

@@ -1156,7 +1156,10 @@ def get(self, tenant_id: str, user: Account, provider: str):
             tenant_id=tenant_id,
             plugin_id=plugin_id,
             provider=provider_name,
-            extra_data={"visibility": requested_visibility.value},
+            extra_data={
+                "visibility": requested_visibility.value,
+                "plugin_unique_identifier": oauth_client_params.plugin_unique_identifier,
+            },
         )
         redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
         authorization_url_response = oauth_handler.get_authorization_url(
@@ -1165,7 +1168,8 @@ def get(self, tenant_id: str, user: Account, provider: str):
             plugin_id=plugin_id,
             provider=provider_name,
             redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
+            system_credentials=dict(oauth_client_params.credentials),
+            plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
         )
         # response-contract:ignore cookie-bearing Flask response
         response = make_response(dump_response(PluginOAuthAuthorizationUrlResponse, authorization_url_response))
@@ -1203,14 +1207,25 @@ def get(self, provider: str):
         if oauth_client_params is None:
             raise Forbidden("no oauth available client config found for this tool provider")
 
+        try:
+            OAuthProxyService.validate_plugin_context(
+                context,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
+            )
+        except ValueError as error:
+            raise Forbidden(str(error)) from error
+
         redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
         credentials_response = oauth_handler.get_credentials(
             tenant_id=tenant_id,
             user_id=user_id,
             plugin_id=plugin_id,
             provider=provider_name,
             redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
+            system_credentials=dict(oauth_client_params.credentials),
+            plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
             request=request,
         )
 

api/controllers/console/workspace/trigger_providers.py

@@ -601,6 +601,7 @@ def get(self, tenant_id: str, user: Account, provider: str):
                 provider=provider_name,
                 extra_data={
                     "subscription_builder_id": subscription_builder.id,
+                    "plugin_unique_identifier": oauth_client_params.plugin_unique_identifier,
                 },
             )
 
@@ -614,7 +615,8 @@ def get(self, tenant_id: str, user: Account, provider: str):
                 plugin_id=plugin_id,
                 provider=provider_name,
                 redirect_uri=redirect_uri,
-                system_credentials=oauth_client_params,
+                system_credentials=dict(oauth_client_params.credentials),
+                plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
             )
 
             response = make_response(
@@ -672,6 +674,16 @@ def get(self, provider: str):
         if oauth_client_params is None:
             raise Forbidden("No OAuth client configuration found for this trigger provider")
 
+        try:
+            OAuthProxyService.validate_plugin_context(
+                context,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
+            )
+        except ValueError as error:
+            raise Forbidden(str(error)) from error
+
         # Get OAuth credentials from callback
         oauth_handler = OAuthHandler()
         redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
@@ -682,7 +694,8 @@ def get(self, provider: str):
             plugin_id=plugin_id,
             provider=provider_name,
             redirect_uri=redirect_uri,
-            system_credentials=oauth_client_params,
+            system_credentials=dict(oauth_client_params.credentials),
+            plugin_unique_identifier=oauth_client_params.plugin_unique_identifier,
             request=request,
         )
 

api/core/agent/strategy/plugin.py

@@ -18,10 +18,13 @@ class PluginAgentStrategy(BaseAgentStrategy):
     declaration: AgentStrategyEntity
     meta_version: str | None = None
 
-    def __init__(self, tenant_id: str, declaration: AgentStrategyEntity, meta_version: str | None):
+    def __init__(
+        self, tenant_id: str, declaration: AgentStrategyEntity, meta_version: str | None, plugin_unique_identifier: str
+    ):
         self.tenant_id = tenant_id
         self.declaration = declaration
         self.meta_version = meta_version
+        self.plugin_unique_identifier = plugin_unique_identifier
 
     @override
     def get_parameters(self) -> Sequence[AgentStrategyParameter]:
@@ -48,7 +51,7 @@ def _invoke(
         """
         Invoke the agent strategy.
         """
-        manager = PluginAgentClient()
+        manager = PluginAgentClient(plugin_unique_identifier=self.plugin_unique_identifier)
 
         initialized_params = self.initialize_parameters(params)
         params = convert_parameters_to_plugin_format(initialized_params)

api/core/datasource/__base/datasource_provider.py

@@ -11,6 +11,7 @@
 class DatasourcePluginProviderController(ABC):
     entity: DatasourceProviderEntityWithPlugin
     tenant_id: str
+    plugin_unique_identifier: str
 
     def __init__(self, entity: DatasourceProviderEntityWithPlugin, tenant_id: str) -> None:
         self.entity = entity
@@ -29,7 +30,7 @@ def _validate_credentials(self, user_id: str, credentials: dict[str, Any]) -> No
         """
         validate the credentials of the provider
         """
-        manager = PluginToolManager()
+        manager = PluginToolManager(plugin_unique_identifier=self.plugin_unique_identifier)
         if not manager.validate_datasource_credentials(
             tenant_id=self.tenant_id,
             user_id=user_id,

api/core/datasource/online_document/online_document_plugin.py

@@ -37,7 +37,7 @@ def get_online_document_pages(
         datasource_parameters: dict[str, Any],
         provider_type: str,
     ) -> Generator[OnlineDocumentPagesMessage]:
-        manager = PluginDatasourceManager()
+        manager = PluginDatasourceManager(plugin_unique_identifier=self.plugin_unique_identifier)
 
         return manager.get_online_document_pages(
             tenant_id=self.tenant_id,
@@ -55,7 +55,7 @@ def get_online_document_page_content(
         datasource_parameters: GetOnlineDocumentPageContentRequest,
         provider_type: str,
     ) -> Generator[DatasourceMessage]:
-        manager = PluginDatasourceManager()
+        manager = PluginDatasourceManager(plugin_unique_identifier=self.plugin_unique_identifier)
 
         return manager.get_online_document_page_content(
             tenant_id=self.tenant_id,

api/core/datasource/online_drive/online_drive_plugin.py

@@ -38,7 +38,7 @@ def online_drive_browse_files(
         request: OnlineDriveBrowseFilesRequest,
         provider_type: str,
     ) -> Generator[OnlineDriveBrowseFilesResponse]:
-        manager = PluginDatasourceManager()
+        manager = PluginDatasourceManager(plugin_unique_identifier=self.plugin_unique_identifier)
 
         return manager.online_drive_browse_files(
             tenant_id=self.tenant_id,
@@ -56,7 +56,7 @@ def online_drive_download_file(
         request: OnlineDriveDownloadFileRequest,
         provider_type: str,
     ) -> Generator[DatasourceMessage]:
-        manager = PluginDatasourceManager()
+        manager = PluginDatasourceManager(plugin_unique_identifier=self.plugin_unique_identifier)
 
         return manager.online_drive_download_file(
             tenant_id=self.tenant_id,

api/core/datasource/website_crawl/website_crawl_plugin.py

@@ -35,7 +35,7 @@ def get_website_crawl(
         datasource_parameters: Mapping[str, Any],
         provider_type: str,
     ) -> Generator[WebsiteCrawlMessage]:
-        manager = PluginDatasourceManager()
+        manager = PluginDatasourceManager(plugin_unique_identifier=self.plugin_unique_identifier)
 
         return manager.get_website_crawl(
             tenant_id=self.tenant_id,

api/core/plugin/impl/base.py

@@ -118,6 +118,11 @@ def _normalize_plugin_daemon_response_for_type(json_response: Any, type_: type[o
 
 
 class BasePluginClient:
+    def __init__(self, *, plugin_unique_identifier: str | None = None) -> None:
+        if plugin_unique_identifier == "":
+            raise ValueError("Plugin unique identifier must not be empty")
+        self._plugin_unique_identifier = plugin_unique_identifier
+
     def _request(
         self,
         method: str,
@@ -180,6 +185,10 @@ def _prepare_request(
         prepared_headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY
         prepared_headers.setdefault("Accept-Encoding", "gzip, deflate, br")
 
+        from services.plugin.dispatch_identity import prepare_dispatch_identity
+
+        prepare_dispatch_identity(path, prepared_headers, self._plugin_unique_identifier)
+
         # Inject traceparent header for distributed tracing
         self._inject_trace_headers(prepared_headers)
 

api/core/plugin/impl/datasource.py

@@ -17,6 +17,7 @@
 from core.plugin.impl.base import BasePluginClient
 from core.schemas.resolver import resolve_dify_schema_refs
 from models.provider_ids import DatasourceProviderID, GenericProviderID
+from services.plugin.dispatch_identity import plugin_dispatch_headers
 from services.tools.tools_transform_service import ToolTransformService
 
 
@@ -313,7 +314,14 @@ def online_drive_download_file(
         yield from response
 
     def validate_provider_credentials(
-        self, tenant_id: str, user_id: str, provider: str, plugin_id: str, credentials: dict[str, Any]
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        plugin_id: str,
+        credentials: dict[str, Any],
+        *,
+        plugin_unique_identifier: str | None = None,
     ) -> bool:
         """
         validate the credentials of the provider
@@ -331,10 +339,7 @@ def validate_provider_credentials(
                     "credentials": credentials,
                 },
             },
-            headers={
-                "X-Plugin-ID": plugin_id,
-                "Content-Type": "application/json",
-            },
+            headers=plugin_dispatch_headers(plugin_id, plugin_unique_identifier),
         )
 
         for resp in response:

api/core/plugin/impl/dynamic_select.py

@@ -4,6 +4,7 @@
 from core.plugin.entities.plugin_daemon import PluginDynamicSelectOptionsResponse
 from core.plugin.impl.base import BasePluginClient
 from models.provider_ids import GenericProviderID
+from services.plugin.dispatch_identity import plugin_dispatch_headers
 
 
 class DynamicSelectClient(BasePluginClient):
@@ -17,6 +18,8 @@ def fetch_dynamic_select_options(
         credentials: Mapping[str, Any],
         credential_type: str,
         parameter: str,
+        *,
+        plugin_unique_identifier: str | None = None,
     ) -> PluginDynamicSelectOptionsResponse:
         """
         Fetch dynamic select options for a plugin parameter.
@@ -35,10 +38,7 @@ def fetch_dynamic_select_options(
                     "parameter": parameter,
                 },
             },
-            headers={
-                "X-Plugin-ID": plugin_id,
-                "Content-Type": "application/json",
-            },
+            headers=plugin_dispatch_headers(plugin_id, plugin_unique_identifier),
         )
 
         for options in response: