From 0970826a34741307916aa74ef417477ef25852ed Mon Sep 17 00:00:00 2001 From: Sebastion Date: Sat, 18 Jul 2026 19:45:09 +0100 Subject: [PATCH] fix(ssrf): strict URL allowlist validation for template workflow fetches (CWE-918) Replace startswith() checks on downloadUrl / downloadCallbackUrl with a parsed-URL allowlist that requires https, an exact hostname match, and rejects embedded userinfo/ports which could otherwise smuggle a different origin (e.g. https://apps-assets.fit2cloud.com.attacker.tld/ or https://attacker.tld@apps-assets.fit2cloud.com/). Also disable HTTP redirect following on the outbound requests.get() calls so a trusted host cannot be used as an open redirector to internal targets. --- apps/application/serializers/application.py | 45 +++++++++++++++++---- 1 file changed, 37 insertions(+), 8 deletions(-) diff --git a/apps/application/serializers/application.py b/apps/application/serializers/application.py index 6989d7a9c66..216cc43a571 100644 --- a/apps/application/serializers/application.py +++ b/apps/application/serializers/application.py @@ -16,6 +16,7 @@ import re import tempfile import zipfile +from urllib.parse import urlparse from functools import reduce from typing import Dict, List @@ -264,6 +265,34 @@ def update_form_knowledge_fields(workflow): update_form_knowledge_fields(node.get("properties", {}).get("node_data", {}).get("loop_body") or {}) + +_ALLOWED_TEMPLATE_DOWNLOAD_HOSTS = {"apps-assets.fit2cloud.com"} +_ALLOWED_TEMPLATE_CALLBACK_HOSTS = {"apps.fit2cloud.com"} + + +def _validate_trusted_url(url, allowed_hosts): + """ + Validate a URL against an allowlist of trusted hosts to prevent SSRF. + Requires https scheme, an exact host match against ``allowed_hosts``, and + rejects URLs carrying userinfo or explicit ports which could be abused to + smuggle a different origin past a naive startswith() check. + """ + 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 + + class MKInstance: def __init__(self, application: dict, function_lib_list: List[dict], version: str, tool_list: List[dict]): self.application = application @@ -747,10 +776,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_TEMPLATE_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) app = ApplicationSerializer( data={"user_id": self.data.get("user_id"), "workspace_id": self.data.get("workspace_id")} ).import_( @@ -771,9 +800,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_TEMPLATE_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,10 +1511,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_TEMPLATE_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) try: mk_instance = restricted_loads(res.content) except Exception as e: @@ -1541,9 +1570,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_TEMPLATE_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}")