diff --git a/apps/application/serializers/application.py b/apps/application/serializers/application.py index e0989366528..042eb4ef689 100644 --- a/apps/application/serializers/application.py +++ b/apps/application/serializers/application.py @@ -41,6 +41,7 @@ ) from common.utils.logger import maxkb_logger from common.utils.tool_code import ToolExecutor +from common.utils.url_validator import ALLOWED_CALLBACK_HOSTS, ALLOWED_DOWNLOAD_HOSTS, validate_trusted_url from django.core import validators from django.db import models, transaction from django.db.models import Q, QuerySet @@ -747,11 +748,10 @@ def insert_template_workflow(self, instance: Dict): self.is_valid(raise_exception=True) work_flow_template = instance.get("work_flow_template") download_url = work_flow_template.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) - app = ApplicationSerializer( + res = requests.get(download_url, timeout=5, allow_redirects=False) data={"user_id": self.data.get("user_id"), "workspace_id": self.data.get("workspace_id")} ).import_( { @@ -771,9 +771,9 @@ def insert_template_workflow(self, instance: Dict): ) try: download_callback_url = work_flow_template.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") return app @@ -1482,12 +1482,10 @@ def update_template_workflow(self, instance: Dict, app: Application): self.is_valid(raise_exception=True) work_flow_template = instance.get("work_flow_template") download_url = work_flow_template.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) - try: - mk_instance = restricted_loads(res.content) + res = requests.get(download_url, timeout=5, allow_redirects=False) except Exception as e: raise AppApiException(1001, _("Unsupported file format")) application = mk_instance.application @@ -1541,9 +1539,9 @@ def update_template_workflow(self, instance: Dict, app: Application): ).auth_resource_batch([t.id for t in tool_model_list]) try: download_callback_url = work_flow_template.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") diff --git a/apps/common/utils/url_validator.py b/apps/common/utils/url_validator.py new file mode 100644 index 00000000000..73c6782968c --- /dev/null +++ b/apps/common/utils/url_validator.py @@ -0,0 +1,42 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author：虎虎 +@file： url_validator.py +@date：2025/7/27 +@desc: Shared URL validation utilities to prevent SSRF (CWE-918) +""" +from urllib.parse import urlparse + +# Allowlist for template/tool/knowledge download asset URLs +ALLOWED_DOWNLOAD_HOSTS = {"apps-assets.fit2cloud.com"} + +# Allowlist for download-callback notification URLs +ALLOWED_CALLBACK_HOSTS = {"apps.fit2cloud.com"} + + +def validate_trusted_url(url, allowed_hosts): + """Return True only if *url* is a safe HTTPS URL whose hostname is an + exact (case-insensitive) match against *allowed_hosts*. + + Rejects: + - Non-string or empty values. + - Non-HTTPS schemes. + - URLs containing userinfo (user:pass@host). + - URLs with an explicit port number. + - Hostnames not present in *allowed_hosts*. + """ + if not url or not isinstance(url, str): + return False + try: + parsed = urlparse(url) + except (ValueError, TypeError): + return False + if parsed.scheme != "https": + return False + if parsed.username or parsed.password: + return False + if parsed.port is not None: + return False + hostname = (parsed.hostname or "").lower() + return hostname in allowed_hosts diff --git a/apps/knowledge/serializers/knowledge_workflow.py b/apps/knowledge/serializers/knowledge_workflow.py index 54181983265..d19d7d37be7 100644 --- a/apps/knowledge/serializers/knowledge_workflow.py +++ b/apps/knowledge/serializers/knowledge_workflow.py @@ -32,6 +32,7 @@ from common.utils.logger import maxkb_logger from common.utils.rsa_util import rsa_long_decrypt from common.utils.tool_code import ToolExecutor +from common.utils.url_validator import ALLOWED_CALLBACK_HOSTS, ALLOWED_DOWNLOAD_HOSTS, validate_trusted_url from knowledge.models import ( KnowledgeScope, Knowledge, @@ -389,10 +390,10 @@ def save_workflow(self, instance: Dict): if instance.get("work_flow_template") is not None: template_instance = instance.get("work_flow_template") download_url = template_instance.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) + res = requests.get(download_url, timeout=5, allow_redirects=False) KnowledgeWorkflowSerializer.Import( data={ "user_id": self.data.get("user_id"), @@ -403,9 +404,9 @@ def save_workflow(self, instance: Dict): try: download_callback_url = template_instance.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") @@ -636,10 +637,10 @@ def edit(self, instance: Dict): if instance.get("work_flow_template"): template_instance = instance.get("work_flow_template") download_url = template_instance.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) + res = requests.get(download_url, timeout=5, allow_redirects=False) KnowledgeWorkflowSerializer.Import( data={ "user_id": self.data.get("user_id"), @@ -650,9 +651,9 @@ def edit(self, instance: Dict): try: download_callback_url = template_instance.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): - raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): + raise AppApiException(500, _("Illegal download callback url")) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") diff --git a/apps/tools/serializers/tool.py b/apps/tools/serializers/tool.py index 727cb7cd4f3..9b648e0346e 100644 --- a/apps/tools/serializers/tool.py +++ b/apps/tools/serializers/tool.py @@ -26,6 +26,7 @@ from common.utils.logger import maxkb_logger from common.utils.rsa_util import rsa_long_decrypt, rsa_long_encrypt from common.utils.tool_code import ToolExecutor +from common.utils.url_validator import ALLOWED_CALLBACK_HOSTS, ALLOWED_DOWNLOAD_HOSTS, validate_trusted_url from django.core import validators from django.core.cache import cache from django.db import transaction @@ -476,11 +477,10 @@ def insert(self, instance, with_valid=True): if instance.get("work_flow_template") is not None: template_instance = instance.get("work_flow_template") download_url = template_instance.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) - tool = ToolSerializer.Import( + res = requests.get(download_url, timeout=5, allow_redirects=False) data={ "file": bytes_to_uploaded_file(res.content, "file.tool"), "user_id": self.data.get("user_id"), @@ -491,9 +491,9 @@ def insert(self, instance, with_valid=True): try: download_callback_url = template_instance.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") return tool @@ -1330,13 +1330,13 @@ def add(self, instance: Dict, with_valid=True): versions = instance.get("versions", []) download_url = instance.get("download_url") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 version_name = next( (version.get("name") for version in versions if version.get("downloadUrl") == download_url), ) - res = requests.get(download_url, timeout=5) + res = requests.get(download_url, timeout=5, allow_redirects=False) tool_data = RestrictedUnpickler(io.BytesIO(res.content)).load().tool tool_id = uuid.uuid7() # 如果是SKILL类型的工具，保存文件内容到file表，并将code替换为file_id @@ -1381,9 +1381,9 @@ def add(self, instance: Dict, with_valid=True): ).auth_resource(str(tool_id)) try: download_callback_url = instance.get("download_callback_url") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") return ToolModelSerializer(tool).data @@ -1400,7 +1400,7 @@ class UpdateStoreTool(serializers.Serializer): def update_tool(self, with_valid=True): if with_valid: self.is_valid(raise_exception=True) - if not self.data.get("download_url").startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(self.data.get("download_url"), ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) tool = QuerySet(Tool).filter( id=self.data.get("tool_id"), workspace_id=self.data.get("workspace_id") @@ -1415,7 +1415,7 @@ def update_tool(self, with_valid=True): if version.get("downloadUrl") == self.data.get("download_url") ), ) - res = requests.get(self.data.get("download_url"), timeout=5) + res = requests.get(self.data.get("download_url"), timeout=5, allow_redirects=False) tool_data = RestrictedUnpickler(io.BytesIO(res.content)).load().tool # 如果是SKILL类型的工具，保存文件内容到file表，并将code替换为file_id if tool_data.get("tool_type") == ToolType.SKILL: @@ -1438,7 +1438,10 @@ def update_tool(self, with_valid=True): # tool.is_active = False tool.save() try: - requests.get(self.data.get("download_callback_url"), timeout=5) + download_callback_url = self.data.get("download_callback_url") + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): + raise AppApiException(500, _("Illegal download callback url")) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}") return ToolModelSerializer(tool).data diff --git a/apps/tools/serializers/tool_workflow.py b/apps/tools/serializers/tool_workflow.py index 49aa94d72f4..a1a695abf83 100644 --- a/apps/tools/serializers/tool_workflow.py +++ b/apps/tools/serializers/tool_workflow.py @@ -37,6 +37,7 @@ from common.utils.common import bytes_to_uploaded_file, generate_uuid, restricted_loads from common.utils.logger import maxkb_logger from common.utils.tool_code import ToolExecutor +from common.utils.url_validator import ALLOWED_CALLBACK_HOSTS, ALLOWED_DOWNLOAD_HOSTS, validate_trusted_url from django.db import transaction from django.db.models import Q, QuerySet from django.http import HttpResponse @@ -353,11 +354,10 @@ def edit(self, instance: Dict): if instance.get("work_flow_template"): template_instance = instance.get("work_flow_template") download_url = template_instance.get("downloadUrl") - if not download_url.startswith("https://apps-assets.fit2cloud.com/"): + if not validate_trusted_url(download_url, ALLOWED_DOWNLOAD_HOSTS): raise AppApiException(500, _("Illegal download url")) # 查找匹配的版本名称 - res = requests.get(download_url, timeout=5) - tool = QuerySet(Tool).filter(id=self.data.get("tool_id")).first() + res = requests.get(download_url, timeout=5, allow_redirects=False) ToolSerializer.Import( data={ "user_id": self.data.get("user_id"), @@ -369,9 +369,9 @@ def edit(self, instance: Dict): try: download_callback_url = template_instance.get("downloadCallbackUrl", "") - if not download_callback_url.startswith("https://apps.fit2cloud.com/"): + if not validate_trusted_url(download_callback_url, ALLOWED_CALLBACK_HOSTS): raise AppApiException(500, _("Illegal download callback url")) - requests.get(download_callback_url, timeout=5) + requests.get(download_callback_url, timeout=5, allow_redirects=False) except Exception as e: maxkb_logger.error(f"callback appstore tool download error: {e}")