fix(security): close remaining SSRF (CWE-918) call sites via shared URL validator

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}")
 

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

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}")
 

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

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}")