diff --git a/api/commands/plugin.py b/api/commands/plugin.py index 445ecb31415..718fa60761c 100644 --- a/api/commands/plugin.py +++ b/api/commands/plugin.py @@ -188,13 +188,13 @@ def transform_datasource_credentials(environment: str): firecrawl_plugin_id = "langgenius/firecrawl_datasource" jina_plugin_id = "langgenius/jina_datasource" if environment == "online": - notion_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(notion_plugin_id) - firecrawl_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(firecrawl_plugin_id) - jina_plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(jina_plugin_id) + notion_package_identifier = plugin_migration._fetch_latest_package_identifier(notion_plugin_id) + firecrawl_package_identifier = plugin_migration._fetch_latest_package_identifier(firecrawl_plugin_id) + jina_package_identifier = plugin_migration._fetch_latest_package_identifier(jina_plugin_id) else: - notion_plugin_unique_identifier = None - firecrawl_plugin_unique_identifier = None - jina_plugin_unique_identifier = None + notion_package_identifier = None + firecrawl_package_identifier = None + jina_package_identifier = None oauth_credential_type = CredentialType.OAUTH2 api_key_credential_type = CredentialType.API_KEY @@ -219,9 +219,9 @@ def transform_datasource_credentials(environment: str): installed_plugins = installer_manager.list_plugins(tenant_id) installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins] if notion_plugin_id not in installed_plugins_ids: - if notion_plugin_unique_identifier: + if notion_package_identifier: # install notion plugin - PluginService.install_from_marketplace_pkg(tenant_id, [notion_plugin_unique_identifier]) + PluginService.install_from_marketplace_pkg(tenant_id, [notion_package_identifier]) auth_count = 0 for notion_tenant_credential in notion_tenant_credentials: auth_count += 1 @@ -279,9 +279,9 @@ def transform_datasource_credentials(environment: str): installed_plugins = installer_manager.list_plugins(tenant_id) installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins] if firecrawl_plugin_id not in installed_plugins_ids: - if firecrawl_plugin_unique_identifier: + if firecrawl_package_identifier: # install firecrawl plugin - PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_plugin_unique_identifier]) + PluginService.install_from_marketplace_pkg(tenant_id, [firecrawl_package_identifier]) auth_count = 0 for firecrawl_tenant_credential in firecrawl_tenant_credentials: @@ -343,10 +343,10 @@ def transform_datasource_credentials(environment: str): installed_plugins = installer_manager.list_plugins(tenant_id) installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins] if jina_plugin_id not in installed_plugins_ids: - if jina_plugin_unique_identifier: + if jina_package_identifier: # install jina plugin - logger.debug("Installing Jina plugin %s", jina_plugin_unique_identifier) - PluginService.install_from_marketplace_pkg(tenant_id, [jina_plugin_unique_identifier]) + logger.debug("Installing Jina plugin %s", jina_package_identifier) + PluginService.install_from_marketplace_pkg(tenant_id, [jina_package_identifier]) auth_count = 0 for jina_tenant_credential in jina_tenant_credentials: diff --git a/api/services/app_dsl_service.py b/api/services/app_dsl_service.py index 52e936bf1ee..d042ad69f88 100644 --- a/api/services/app_dsl_service.py +++ b/api/services/app_dsl_service.py @@ -39,6 +39,7 @@ from libs.datetime_utils import naive_utc_now from models import Account, App, AppMode from models.model import AppModelConfig, AppModelConfigDict, IconType from models.workflow import Workflow +from services.dsl_content import DSL_MAX_SIZE, dsl_content_size from services.dsl_version import check_version_compatibility from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus from services.errors.app import WorkflowNotFoundError @@ -51,7 +52,6 @@ logger = logging.getLogger(__name__) IMPORT_INFO_REDIS_KEY_PREFIX = "app_import_info:" CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "app_check_dependencies:" IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes -DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB CURRENT_DSL_VERSION = CURRENT_APP_DSL_VERSION @@ -131,15 +131,16 @@ class AppDslService: yaml_url = yaml_url.replace("/blob/", "/") response = remote_fetcher.make_request("GET", yaml_url.strip(), follow_redirects=True, timeout=(10, 10)) response.raise_for_status() - content = response.content.decode() + raw_content = response.content - if len(content) > DSL_MAX_SIZE: + if dsl_content_size(raw_content) > DSL_MAX_SIZE: return Import( id=import_id, status=ImportStatus.FAILED, error="File size exceeds the limit of 10MB", ) + content = raw_content.decode("utf-8") if not content: return Import( id=import_id, @@ -160,6 +161,12 @@ class AppDslService: error="yaml_content is required when import_mode is yaml-content", ) content = yaml_content + if dsl_content_size(content) > DSL_MAX_SIZE: + return Import( + id=import_id, + status=ImportStatus.FAILED, + error="File size exceeds the limit of 10MB", + ) # Process YAML content try: diff --git a/api/services/dsl_content.py b/api/services/dsl_content.py new file mode 100644 index 00000000000..3874d0546db --- /dev/null +++ b/api/services/dsl_content.py @@ -0,0 +1,9 @@ +"""Shared DSL content size and decoding rules.""" + +DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB + + +def dsl_content_size(content: str | bytes) -> int: + if isinstance(content, bytes): + return len(content) + return len(content.encode("utf-8")) diff --git a/api/services/plugin/plugin_migration.py b/api/services/plugin/plugin_migration.py index 82eeb5a7261..d6f154df812 100644 --- a/api/services/plugin/plugin_migration.py +++ b/api/services/plugin/plugin_migration.py @@ -307,9 +307,9 @@ class PluginMigration: return result @classmethod - def _fetch_plugin_unique_identifier(cls, plugin_id: str) -> str | None: + def _fetch_latest_package_identifier(cls, plugin_id: str) -> str | None: """ - Fetch plugin unique identifier using plugin id. + Fetch the latest marketplace package identifier using a plugin id. """ if not dify_config.MARKETPLACE_ENABLED: return None @@ -328,7 +328,7 @@ class PluginMigration: @classmethod def extract_unique_plugins(cls, extracted_plugins: str) -> ExtractedPluginsDict: - plugins: dict[str, str] = {} + package_identifier_by_plugin_id: dict[str, str] = {} plugin_ids = [] plugin_not_exist = [] logger.info("Extracting unique plugins from %s", extracted_plugins) @@ -341,19 +341,19 @@ class PluginMigration: def fetch_plugin(plugin_id): try: - unique_identifier = cls._fetch_plugin_unique_identifier(plugin_id) - if unique_identifier: - plugins[plugin_id] = unique_identifier + latest_package_identifier = cls._fetch_latest_package_identifier(plugin_id) + if latest_package_identifier: + package_identifier_by_plugin_id[plugin_id] = latest_package_identifier else: plugin_not_exist.append(plugin_id) except Exception: - logger.exception("Failed to fetch plugin unique identifier for %s", plugin_id) + logger.exception("Failed to fetch latest package identifier for %s", plugin_id) plugin_not_exist.append(plugin_id) with ThreadPoolExecutor(max_workers=10) as executor: list(tqdm.tqdm(executor.map(fetch_plugin, plugin_ids), total=len(plugin_ids))) - return {"plugins": plugins, "plugin_not_exist": plugin_not_exist} + return {"plugins": package_identifier_by_plugin_id, "plugin_not_exist": plugin_not_exist} @classmethod def install_plugins(cls, extracted_plugins: str, output_file: str, workers: int = 100): @@ -362,17 +362,22 @@ class PluginMigration: """ manager = PluginInstaller() - plugins = cls.extract_unique_plugins(extracted_plugins) + extracted = cls.extract_unique_plugins(extracted_plugins) + package_identifier_by_plugin_id = extracted["plugins"] not_installed = [] plugin_install_failed = [] # use a fake tenant id to install all the plugins fake_tenant_id = uuid4().hex - logger.info("Installing %s plugin instances for fake tenant %s", len(plugins["plugins"]), fake_tenant_id) + logger.info( + "Installing %s plugin instances for fake tenant %s", + len(package_identifier_by_plugin_id), + fake_tenant_id, + ) thread_pool = ThreadPoolExecutor(max_workers=workers) - response = cls.handle_plugin_instance_install(fake_tenant_id, plugins["plugins"]) + response = cls.handle_plugin_instance_install(fake_tenant_id, package_identifier_by_plugin_id) if response.get("failed"): plugin_install_failed.extend(response.get("failed", [])) @@ -384,21 +389,21 @@ class PluginMigration: # at most 64 plugins one batch for i in range(0, len(plugin_ids), 64): batch_plugin_ids = plugin_ids[i : i + 64] - batch_plugin_identifiers = [ - plugins["plugins"][plugin_id] + batch_package_identifiers = [ + package_identifier_by_plugin_id[plugin_id] for plugin_id in batch_plugin_ids - if plugin_id not in installed_plugins_ids and plugin_id in plugins["plugins"] + if plugin_id not in installed_plugins_ids and plugin_id in package_identifier_by_plugin_id ] - if batch_plugin_identifiers: + if batch_package_identifiers: manager.install_from_identifiers( tenant_id, - batch_plugin_identifiers, + batch_package_identifiers, PluginInstallationSource.Marketplace, metas=[ { - "plugin_unique_identifier": identifier, + "plugin_unique_identifier": package_identifier, } - for identifier in batch_plugin_identifiers + for package_identifier in batch_package_identifiers ], ) PluginService.invalidate_plugin_model_providers_cache(tenant_id) @@ -412,10 +417,8 @@ class PluginMigration: tenant_id = data["tenant_id"] plugin_ids = data["plugins"] plugin_not_exist: list[str] = [] - # get plugin unique identifier for plugin_id in plugin_ids: - unique_identifier = plugins.get(plugin_id) - if unique_identifier: + if plugin_id not in package_identifier_by_plugin_id: plugin_not_exist.append(plugin_id) if plugin_not_exist: @@ -459,36 +462,44 @@ class PluginMigration: """ manager = PluginInstaller() - plugins = cls.extract_unique_plugins(extracted_plugins) + extracted = cls.extract_unique_plugins(extracted_plugins) + package_identifier_by_plugin_id = extracted["plugins"] plugin_install_failed = [] # use a fake tenant id to install all the plugins fake_tenant_id = uuid4().hex - logger.info("Installing %s plugin instances for fake tenant %s", len(plugins["plugins"]), fake_tenant_id) + logger.info( + "Installing %s plugin instances for fake tenant %s", + len(package_identifier_by_plugin_id), + fake_tenant_id, + ) thread_pool = ThreadPoolExecutor(max_workers=workers) - response = cls.handle_plugin_instance_install(fake_tenant_id, plugins["plugins"]) + response = cls.handle_plugin_instance_install(fake_tenant_id, package_identifier_by_plugin_id) if response.get("failed"): plugin_install_failed.extend(response.get("failed", [])) def install( - tenant_id: str, plugin_ids: dict[str, str], total_success_tenant: int, total_failed_tenant: int + tenant_id: str, + package_identifier_by_plugin_id: dict[str, str], + total_success_tenant: int, + total_failed_tenant: int, ) -> None: - logger.info("Installing %s plugins for tenant %s", len(plugin_ids), tenant_id) + logger.info("Installing %s plugins for tenant %s", len(package_identifier_by_plugin_id), tenant_id) try: # fetch plugin already installed installed_plugins = manager.list_plugins(tenant_id) installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins] # at most 64 plugins one batch - for i in range(0, len(plugin_ids), 64): - batch_plugin_ids = list(plugin_ids.keys())[i : i + 64] - batch_plugin_identifiers = [ - plugin_ids[plugin_id] + for i in range(0, len(package_identifier_by_plugin_id), 64): + batch_plugin_ids = list(package_identifier_by_plugin_id.keys())[i : i + 64] + batch_package_identifiers = [ + package_identifier_by_plugin_id[plugin_id] for plugin_id in batch_plugin_ids - if plugin_id not in installed_plugins_ids and plugin_id in plugin_ids + if plugin_id not in installed_plugins_ids and plugin_id in package_identifier_by_plugin_id ] - PluginService.install_from_marketplace_pkg(tenant_id, batch_plugin_identifiers) + PluginService.install_from_marketplace_pkg(tenant_id, batch_package_identifiers) total_success_tenant += 1 except Exception: @@ -510,7 +521,7 @@ class PluginMigration: thread_pool.submit( install, tenant_id, - plugins.get("plugins", {}), + package_identifier_by_plugin_id, total_success_tenant, total_failed_tenant, ) @@ -542,12 +553,12 @@ class PluginMigration: @classmethod def handle_plugin_instance_install( - cls, tenant_id: str, plugin_identifiers_map: Mapping[str, str] + cls, tenant_id: str, package_identifier_by_plugin_id: Mapping[str, str] ) -> PluginInstallResultDict: """ Install plugins for a tenant. """ - if plugin_identifiers_map and not dify_config.MARKETPLACE_ENABLED: + if package_identifier_by_plugin_id and not dify_config.MARKETPLACE_ENABLED: raise ValueError( "Marketplace disabled in offline mode; cannot bulk-install plugins. " "Pre-upload plugin packages via Console first." @@ -558,17 +569,17 @@ class PluginMigration: thread_pool = ThreadPoolExecutor(max_workers=10) futures = [] - for plugin_id, plugin_identifier in plugin_identifiers_map.items(): + for plugin_id, package_identifier in package_identifier_by_plugin_id.items(): - def download_and_upload(tenant_id, plugin_id, plugin_identifier): - plugin_package = marketplace.download_plugin_pkg(plugin_identifier) + def download_and_upload(tenant_id, plugin_id, package_identifier): + plugin_package = marketplace.download_plugin_pkg(package_identifier) if not plugin_package: - raise Exception(f"Failed to download plugin {plugin_identifier}") + raise Exception(f"Failed to download plugin {package_identifier}") # upload manager.upload_pkg(tenant_id, plugin_package, verify_signature=True) - futures.append(thread_pool.submit(download_and_upload, tenant_id, plugin_id, plugin_identifier)) + futures.append(thread_pool.submit(download_and_upload, tenant_id, plugin_id, package_identifier)) # Wait for all downloads to complete for future in futures: @@ -578,33 +589,33 @@ class PluginMigration: success = [] failed = [] - reverse_map = {v: k for k, v in plugin_identifiers_map.items()} + plugin_id_by_package_identifier = {v: k for k, v in package_identifier_by_plugin_id.items()} # at most 8 plugins one batch - for i in range(0, len(plugin_identifiers_map), 8): - batch_plugin_ids = list(plugin_identifiers_map.keys())[i : i + 8] - batch_plugin_identifiers = [plugin_identifiers_map[plugin_id] for plugin_id in batch_plugin_ids] + for i in range(0, len(package_identifier_by_plugin_id), 8): + batch_plugin_ids = list(package_identifier_by_plugin_id.keys())[i : i + 8] + batch_package_identifiers = [package_identifier_by_plugin_id[plugin_id] for plugin_id in batch_plugin_ids] try: response = manager.install_from_identifiers( tenant_id=tenant_id, - identifiers=batch_plugin_identifiers, + identifiers=batch_package_identifiers, source=PluginInstallationSource.Marketplace, metas=[ { - "plugin_unique_identifier": identifier, + "plugin_unique_identifier": package_identifier, } - for identifier in batch_plugin_identifiers + for package_identifier in batch_package_identifiers ], ) PluginService.invalidate_plugin_model_providers_cache(tenant_id) except Exception: # add to failed - failed.extend(batch_plugin_identifiers) + failed.extend(batch_plugin_ids) continue if response.all_installed: - success.extend(batch_plugin_identifiers) + success.extend(batch_plugin_ids) continue task_id = response.task_id @@ -614,10 +625,13 @@ class PluginMigration: if status.status in [PluginInstallTaskStatus.Failed, PluginInstallTaskStatus.Success]: PluginService.invalidate_plugin_model_providers_cache(tenant_id) for plugin in status.plugins: + plugin_id = plugin_id_by_package_identifier.get( + plugin.plugin_unique_identifier, plugin.plugin_unique_identifier.split(":", 1)[0] + ) if plugin.status == PluginInstallTaskStatus.Success: - success.append(reverse_map[plugin.plugin_unique_identifier]) + success.append(plugin_id) else: - failed.append(reverse_map[plugin.plugin_unique_identifier]) + failed.append(plugin_id) logger.error( "Failed to install plugin %s, error: %s", plugin.plugin_unique_identifier, diff --git a/api/services/rag_pipeline/rag_pipeline_dsl_service.py b/api/services/rag_pipeline/rag_pipeline_dsl_service.py index 4f9cde37a7b..5459c3e5f1f 100644 --- a/api/services/rag_pipeline/rag_pipeline_dsl_service.py +++ b/api/services/rag_pipeline/rag_pipeline_dsl_service.py @@ -36,6 +36,7 @@ from models import Account from models.dataset import Dataset, DatasetCollectionBinding, Pipeline from models.enums import CollectionBindingType, DatasetRuntimeMode from models.workflow import Workflow, WorkflowType +from services.dsl_content import DSL_MAX_SIZE, dsl_content_size from services.dsl_version import check_version_compatibility from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus from services.entities.knowledge_entities.rag_pipeline_entities import ( @@ -50,7 +51,6 @@ logger = logging.getLogger(__name__) IMPORT_INFO_REDIS_KEY_PREFIX = "app_import_info:" CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "app_check_dependencies:" IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes -DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB CURRENT_DSL_VERSION = "0.1.0" @@ -127,15 +127,16 @@ class RagPipelineDslService: yaml_url = yaml_url.replace("/blob/", "/") response = remote_fetcher.make_request("GET", yaml_url.strip(), follow_redirects=True, timeout=(10, 10)) response.raise_for_status() - content = response.content.decode() + raw_content = response.content - if len(content) > DSL_MAX_SIZE: + if dsl_content_size(raw_content) > DSL_MAX_SIZE: return RagPipelineImportInfo( id=import_id, status=ImportStatus.FAILED, error="File size exceeds the limit of 10MB", ) + content = raw_content.decode("utf-8") if not content: return RagPipelineImportInfo( id=import_id, @@ -156,6 +157,12 @@ class RagPipelineDslService: error="yaml_content is required when import_mode is yaml-content", ) content = yaml_content + if dsl_content_size(content) > DSL_MAX_SIZE: + return RagPipelineImportInfo( + id=import_id, + status=ImportStatus.FAILED, + error="File size exceeds the limit of 10MB", + ) # Process YAML content try: diff --git a/api/services/rag_pipeline/rag_pipeline_transform_service.py b/api/services/rag_pipeline/rag_pipeline_transform_service.py index daefaa9e30e..1b922b3f7b9 100644 --- a/api/services/rag_pipeline/rag_pipeline_transform_service.py +++ b/api/services/rag_pipeline/rag_pipeline_transform_service.py @@ -269,11 +269,13 @@ class RagPipelineTransformService: installed_plugins_ids = [plugin.plugin_id for plugin in installed_plugins] dependencies = pipeline_yaml.get("dependencies", []) - need_install_plugin_unique_identifiers = [] + package_identifiers_to_install = [] for dependency in dependencies: if dependency.get("type") == "marketplace": - plugin_unique_identifier = dependency.get("value", {}).get("plugin_unique_identifier") - plugin_id = plugin_unique_identifier.split(":")[0] + package_identifier = dependency.get("value", {}).get("plugin_unique_identifier") + if not package_identifier: + continue + plugin_id = package_identifier.split(":", 1)[0] if plugin_id not in installed_plugins_ids: if not dify_config.MARKETPLACE_ENABLED: logger.warning( @@ -282,12 +284,12 @@ class RagPipelineTransformService: plugin_id, ) continue - plugin_unique_identifier = plugin_migration._fetch_plugin_unique_identifier(plugin_id) # type: ignore - if plugin_unique_identifier: - need_install_plugin_unique_identifiers.append(plugin_unique_identifier) - if need_install_plugin_unique_identifiers: - logger.debug("Installing missing pipeline plugins %s", need_install_plugin_unique_identifiers) - PluginService.install_from_marketplace_pkg(tenant_id, need_install_plugin_unique_identifiers) + latest_package_identifier = plugin_migration._fetch_latest_package_identifier(plugin_id) # type: ignore + if latest_package_identifier: + package_identifiers_to_install.append(latest_package_identifier) + if package_identifiers_to_install: + logger.debug("Installing missing pipeline plugins %s", package_identifiers_to_install) + PluginService.install_from_marketplace_pkg(tenant_id, package_identifiers_to_install) def _transform_to_empty_pipeline(self, dataset: Dataset, session: Session): pipeline = Pipeline( diff --git a/api/services/snippet_dsl_service.py b/api/services/snippet_dsl_service.py index 19e0dabab95..ae1cd7f7e31 100644 --- a/api/services/snippet_dsl_service.py +++ b/api/services/snippet_dsl_service.py @@ -3,12 +3,10 @@ import logging import uuid from collections.abc import Mapping from datetime import UTC, datetime -from enum import StrEnum from urllib.parse import urlparse import yaml -from packaging import version -from pydantic import BaseModel, Field +from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.orm import Session @@ -20,6 +18,9 @@ from graphon.model_runtime.utils.encoders import jsonable_encoder from models import Account from models.snippet import CustomizedSnippet, SnippetType from models.workflow import Workflow +from services.dsl_content import DSL_MAX_SIZE, dsl_content_size +from services.dsl_version import check_version_compatibility +from services.entities.dsl_entities import CheckDependenciesResult, ImportMode, ImportStatus from services.plugin.dependencies_analysis import DependenciesAnalysisService from services.snippet_service import SNIPPET_FORBIDDEN_NODE_TYPES, SnippetService @@ -28,22 +29,9 @@ logger = logging.getLogger(__name__) IMPORT_INFO_REDIS_KEY_PREFIX = "snippet_import_info:" CHECK_DEPENDENCIES_REDIS_KEY_PREFIX = "snippet_check_dependencies:" IMPORT_INFO_REDIS_EXPIRY = 10 * 60 # 10 minutes -DSL_MAX_SIZE = 10 * 1024 * 1024 # 10MB CURRENT_DSL_VERSION = "0.1.0" -class ImportMode(StrEnum): - YAML_CONTENT = "yaml-content" - YAML_URL = "yaml-url" - - -class ImportStatus(StrEnum): - COMPLETED = "completed" - COMPLETED_WITH_WARNINGS = "completed-with-warnings" - PENDING = "pending" - FAILED = "failed" - - class SnippetImportInfo(BaseModel): id: str status: ImportStatus @@ -53,32 +41,9 @@ class SnippetImportInfo(BaseModel): error: str = "" -class CheckDependenciesResult(BaseModel): - leaked_dependencies: list[PluginDependency] = Field(default_factory=list) - - def _check_version_compatibility(imported_version: str) -> ImportStatus: - """Determine import status based on version comparison""" - try: - current_ver = version.parse(CURRENT_DSL_VERSION) - imported_ver = version.parse(imported_version) - except version.InvalidVersion: - return ImportStatus.FAILED - - # If imported version is newer than current, always return PENDING - if imported_ver > current_ver: - return ImportStatus.PENDING - - # If imported version is older than current's major, return PENDING - if imported_ver.major < current_ver.major: - return ImportStatus.PENDING - - # If imported version is older than current's minor, return COMPLETED_WITH_WARNINGS - if imported_ver.minor < current_ver.minor: - return ImportStatus.COMPLETED_WITH_WARNINGS - - # If imported version equals or is older than current's micro, return COMPLETED - return ImportStatus.COMPLETED + """Determine import status based on version comparison.""" + return check_version_compatibility(imported_version, CURRENT_DSL_VERSION) class SnippetPendingData(BaseModel): @@ -145,13 +110,14 @@ class SnippetDslService: status=ImportStatus.FAILED, error=f"Failed to fetch YAML from URL: {response.status_code}", ) - content = response.text - if len(content) > DSL_MAX_SIZE: + raw_content = response.content + if dsl_content_size(raw_content) > DSL_MAX_SIZE: return SnippetImportInfo( id=import_id, status=ImportStatus.FAILED, error=f"YAML content size exceeds maximum limit of {DSL_MAX_SIZE} bytes", ) + content = raw_content.decode("utf-8") except Exception as e: logger.exception("Failed to fetch YAML from URL") return SnippetImportInfo( @@ -167,7 +133,7 @@ class SnippetDslService: error="yaml_content is required when import_mode is yaml-content", ) content = yaml_content - if len(content) > DSL_MAX_SIZE: + if dsl_content_size(content) > DSL_MAX_SIZE: return SnippetImportInfo( id=import_id, status=ImportStatus.FAILED, diff --git a/api/tests/unit_tests/services/plugin/test_plugin_migration.py b/api/tests/unit_tests/services/plugin/test_plugin_migration.py index 8f730d4ed31..fa10d775aa6 100644 --- a/api/tests/unit_tests/services/plugin/test_plugin_migration.py +++ b/api/tests/unit_tests/services/plugin/test_plugin_migration.py @@ -1,3 +1,4 @@ +import json from unittest.mock import MagicMock, patch import pytest @@ -8,17 +9,17 @@ from services.plugin.plugin_migration import PluginMigration MIGRATION_MODULE = "services.plugin.plugin_migration" -def test_fetch_plugin_unique_identifier_returns_none_when_disabled(mocker: MockerFixture) -> None: +def test_fetch_latest_package_identifier_returns_none_when_disabled(mocker: MockerFixture) -> None: mocker.patch("services.plugin.plugin_migration.dify_config.MARKETPLACE_ENABLED", False) batch_fetch = mocker.patch("services.plugin.plugin_migration.marketplace.batch_fetch_plugin_manifests") - result = PluginMigration._fetch_plugin_unique_identifier("langgenius/openai") + result = PluginMigration._fetch_latest_package_identifier("langgenius/openai") assert result is None batch_fetch.assert_not_called() -def test_fetch_plugin_unique_identifier_calls_marketplace_when_enabled(mocker: MockerFixture) -> None: +def test_fetch_latest_package_identifier_calls_marketplace_when_enabled(mocker: MockerFixture) -> None: mocker.patch("services.plugin.plugin_migration.dify_config.MARKETPLACE_ENABLED", True) manifest = mocker.MagicMock() manifest.latest_package_identifier = "langgenius/openai:1.0.0@abc" @@ -27,7 +28,7 @@ def test_fetch_plugin_unique_identifier_calls_marketplace_when_enabled(mocker: M return_value=[manifest], ) - result = PluginMigration._fetch_plugin_unique_identifier("langgenius/openai") + result = PluginMigration._fetch_latest_package_identifier("langgenius/openai") assert result == "langgenius/openai:1.0.0@abc" @@ -75,7 +76,27 @@ class TestHandlePluginInstanceInstall: mock_marketplace.download_plugin_pkg.assert_called_once() invalidate_cache.assert_called_once_with("tenant1") - assert "success" in result or "failed" in result + assert result["success"] == ["langgenius/openai"] + assert result["failed"] == [] + + def test_reports_failed_plugin_ids_when_install_batch_raises(self) -> None: + with ( + patch(f"{MIGRATION_MODULE}.dify_config") as mock_cfg, + patch(f"{MIGRATION_MODULE}.marketplace") as mock_marketplace, + patch(f"{MIGRATION_MODULE}.PluginInstaller") as mock_installer_cls, + ): + mock_cfg.MARKETPLACE_ENABLED = True + mock_marketplace.download_plugin_pkg.return_value = b"pkg_data" + mock_installer = MagicMock() + mock_installer_cls.return_value = mock_installer + mock_installer.install_from_identifiers.side_effect = RuntimeError("install failed") + + result = PluginMigration.handle_plugin_instance_install( + "tenant1", {"langgenius/openai": "langgenius/openai:1.0.0@abc"} + ) + + assert result["success"] == [] + assert result["failed"] == ["langgenius/openai"] def test_install_plugins_invalidates_cache_after_direct_tenant_install(self, tmp_path) -> None: extracted_plugins = tmp_path / "plugins.jsonl" @@ -102,3 +123,30 @@ class TestHandlePluginInstanceInstall: mock_installer.install_from_identifiers.assert_called_once() invalidate_cache.assert_called_once_with("tenant1") + + def test_install_plugins_reports_missing_plugin_ids(self, tmp_path) -> None: + extracted_plugins = tmp_path / "plugins.jsonl" + output_file = tmp_path / "output.json" + extracted_plugins.write_text('{"tenant_id":"tenant1","plugins":["langgenius/openai","langgenius/missing"]}\n') + + with ( + patch( + f"{MIGRATION_MODULE}.PluginMigration.extract_unique_plugins", + return_value={ + "plugins": {"langgenius/openai": "langgenius/openai:1.0.0@abc"}, + "plugin_not_exist": ["langgenius/missing"], + }, + ), + patch(f"{MIGRATION_MODULE}.PluginMigration.handle_plugin_instance_install", return_value={}), + patch(f"{MIGRATION_MODULE}.PluginInstaller") as mock_installer_cls, + patch(f"{MIGRATION_MODULE}.PluginService.invalidate_plugin_model_providers_cache"), + ): + mock_installer = MagicMock() + mock_installer.list_plugins.return_value = [] + mock_installer_cls.return_value = mock_installer + + PluginMigration.install_plugins(str(extracted_plugins), str(output_file), workers=1) + + assert json.loads(output_file.read_text())["not_installed"] == [ + {"tenant_id": "tenant1", "plugin_not_exist": ["langgenius/missing"]} + ] diff --git a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py index 5bb577ba3dc..93884d07c5c 100644 --- a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py +++ b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_dsl_service.py @@ -643,6 +643,19 @@ def test_import_rag_pipeline_yaml_content_requires_mapping() -> None: assert "content must be a mapping" in result.error +def test_import_rag_pipeline_rejects_oversized_yaml_content_by_bytes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("services.rag_pipeline.rag_pipeline_dsl_service.DSL_MAX_SIZE", 1) + service = RagPipelineDslService(session=Mock()) + account = Mock(current_tenant_id="t1") + + result = service.import_rag_pipeline(account=account, import_mode="yaml-content", yaml_content="é") + + assert result.status == ImportStatus.FAILED + assert "10MB" in result.error + + def test_confirm_import_returns_failed_when_pending_data_is_invalid_type(mocker: MockerFixture) -> None: mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.redis_client.get", return_value=object()) service = RagPipelineDslService(session=Mock()) @@ -901,6 +914,46 @@ def test_import_rag_pipeline_url_size_exceeds_limit(mocker: MockerFixture) -> No assert "10MB" in result.error +def test_import_rag_pipeline_rejects_oversized_yaml_url_bytes_before_decode( + mocker: MockerFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("services.rag_pipeline.rag_pipeline_dsl_service.DSL_MAX_SIZE", 1) + response = Mock() + response.raise_for_status.return_value = None + response.content = b"\xff\xff" + mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.remote_fetcher.make_request", return_value=response) + service = RagPipelineDslService(session=Mock()) + account = Mock(current_tenant_id="t1") + + result = service.import_rag_pipeline( + account=account, + import_mode="yaml-url", + yaml_url="https://example.com/pipeline.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "10MB" in result.error + + +def test_import_rag_pipeline_returns_decode_error_for_invalid_yaml_url_bytes(mocker: MockerFixture) -> None: + response = Mock() + response.raise_for_status.return_value = None + response.content = b"\xff" + mocker.patch("services.rag_pipeline.rag_pipeline_dsl_service.remote_fetcher.make_request", return_value=response) + service = RagPipelineDslService(session=Mock()) + account = Mock(current_tenant_id="t1") + + result = service.import_rag_pipeline( + account=account, + import_mode="yaml-url", + yaml_url="https://example.com/pipeline.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "utf-8" in result.error + + def test_import_rag_pipeline_fails_when_rag_pipeline_data_missing() -> None: service = RagPipelineDslService(session=Mock()) account = Mock(current_tenant_id="t1") diff --git a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_transform_service.py b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_transform_service.py index 21df81d3ea8..cee8e55f8cc 100644 --- a/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_transform_service.py +++ b/api/tests/unit_tests/services/rag_pipeline/test_rag_pipeline_transform_service.py @@ -89,7 +89,7 @@ def test_deal_dependencies_installs_missing_marketplace_plugins(mocker: MockerFi installer_cls.return_value.list_plugins.return_value = [SimpleNamespace(plugin_id="installed-plugin")] migration_cls = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginMigration") - migration_cls.return_value._fetch_plugin_unique_identifier.return_value = "missing-plugin:1.0.0" + migration_cls.return_value._fetch_latest_package_identifier.return_value = "missing-plugin:1.0.0" install_mock = mocker.patch( "services.rag_pipeline.rag_pipeline_transform_service.PluginService.install_from_marketplace_pkg" @@ -518,7 +518,7 @@ def test_deal_dependencies_installs_when_enabled(mocker: MockerFixture) -> None: installer = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginInstaller").return_value installer.list_plugins.return_value = [] migration = mocker.patch("services.rag_pipeline.rag_pipeline_transform_service.PluginMigration").return_value - migration._fetch_plugin_unique_identifier.return_value = "langgenius/openai:1.0.0@abc" + migration._fetch_latest_package_identifier.return_value = "langgenius/openai:1.0.0@abc" install_call = mocker.patch( "services.rag_pipeline.rag_pipeline_transform_service.PluginService.install_from_marketplace_pkg" ) diff --git a/api/tests/unit_tests/services/test_app_dsl_service.py b/api/tests/unit_tests/services/test_app_dsl_service.py new file mode 100644 index 00000000000..64236ea5a90 --- /dev/null +++ b/api/tests/unit_tests/services/test_app_dsl_service.py @@ -0,0 +1,53 @@ +from types import SimpleNamespace +from unittest.mock import Mock + +from services.app_dsl_service import AppDslService, ImportStatus + + +def test_import_app_rejects_oversized_yaml_content_by_bytes(monkeypatch) -> None: + monkeypatch.setattr("services.app_dsl_service.DSL_MAX_SIZE", 1) + service = AppDslService(session=SimpleNamespace()) + + result = service.import_app( + account=SimpleNamespace(current_tenant_id="tenant-1"), + import_mode="yaml-content", + yaml_content="é", + ) + + assert result.status == ImportStatus.FAILED + assert "10MB" in result.error + + +def test_import_app_rejects_oversized_yaml_url_bytes_before_decode(monkeypatch) -> None: + monkeypatch.setattr("services.app_dsl_service.DSL_MAX_SIZE", 1) + response = Mock() + response.raise_for_status.return_value = None + response.content = b"\xff\xff" + monkeypatch.setattr("services.app_dsl_service.remote_fetcher.make_request", Mock(return_value=response)) + service = AppDslService(session=SimpleNamespace()) + + result = service.import_app( + account=SimpleNamespace(current_tenant_id="tenant-1"), + import_mode="yaml-url", + yaml_url="https://example.com/app.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "10MB" in result.error + + +def test_import_app_returns_decode_error_for_invalid_yaml_url_bytes(monkeypatch) -> None: + response = Mock() + response.raise_for_status.return_value = None + response.content = b"\xff" + monkeypatch.setattr("services.app_dsl_service.remote_fetcher.make_request", Mock(return_value=response)) + service = AppDslService(session=SimpleNamespace()) + + result = service.import_app( + account=SimpleNamespace(current_tenant_id="tenant-1"), + import_mode="yaml-url", + yaml_url="https://example.com/app.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "utf-8" in result.error diff --git a/api/tests/unit_tests/services/test_snippet_dsl_service.py b/api/tests/unit_tests/services/test_snippet_dsl_service.py index c155d3f8330..b234d9f91fe 100644 --- a/api/tests/unit_tests/services/test_snippet_dsl_service.py +++ b/api/tests/unit_tests/services/test_snippet_dsl_service.py @@ -95,7 +95,7 @@ def test_import_snippet_rejects_oversized_yaml_url_content(monkeypatch: pytest.M monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 3) monkeypatch.setattr( "services.snippet_dsl_service.ssrf_proxy.get", - Mock(return_value=SimpleNamespace(status_code=200, text="too large")), + Mock(return_value=SimpleNamespace(status_code=200, content=b"too large")), ) result = service.import_snippet( @@ -108,6 +108,43 @@ def test_import_snippet_rejects_oversized_yaml_url_content(monkeypatch: pytest.M assert "YAML content size exceeds maximum limit" in result.error +def test_import_snippet_rejects_oversized_yaml_url_bytes_before_decode(monkeypatch: pytest.MonkeyPatch) -> None: + service = SnippetDslService(session=SimpleNamespace()) + monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 1) + monkeypatch.setattr( + "services.snippet_dsl_service.ssrf_proxy.get", + Mock(return_value=SimpleNamespace(status_code=200, content=b"\xff\xff")), + ) + + result = service.import_snippet( + account=SimpleNamespace(current_tenant_id="tenant-1"), + import_mode=ImportMode.YAML_URL.value, + yaml_url="https://example.com/snippet.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "YAML content size exceeds maximum limit" in result.error + + +def test_import_snippet_returns_decode_error_for_invalid_yaml_url_bytes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = SnippetDslService(session=SimpleNamespace()) + monkeypatch.setattr( + "services.snippet_dsl_service.ssrf_proxy.get", + Mock(return_value=SimpleNamespace(status_code=200, content=b"\xff")), + ) + + result = service.import_snippet( + account=SimpleNamespace(current_tenant_id="tenant-1"), + import_mode=ImportMode.YAML_URL.value, + yaml_url="https://example.com/snippet.yaml", + ) + + assert result.status == ImportStatus.FAILED + assert "utf-8" in result.error + + def test_import_snippet_returns_failed_when_yaml_url_fetch_raises(monkeypatch: pytest.MonkeyPatch) -> None: service = SnippetDslService(session=SimpleNamespace()) monkeypatch.setattr( @@ -127,12 +164,12 @@ def test_import_snippet_returns_failed_when_yaml_url_fetch_raises(monkeypatch: p def test_import_snippet_rejects_oversized_yaml_content(monkeypatch: pytest.MonkeyPatch) -> None: service = SnippetDslService(session=SimpleNamespace()) - monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 3) + monkeypatch.setattr("services.snippet_dsl_service.DSL_MAX_SIZE", 1) result = service.import_snippet( account=SimpleNamespace(current_tenant_id="tenant-1"), import_mode=ImportMode.YAML_CONTENT.value, - yaml_content="too large", + yaml_content="é", ) assert result.status == ImportStatus.FAILED