CVE-2026-10662

@@ -18,8 +18,53 @@
 from datetime import datetime
 import hashlib, hmac, base64
 import os.path as osp
+import ipaddress
+from urllib.parse import urlparse
 from contextlib import redirect_stdout, suppress
 
+# Allowed image file extensions for local path validation
+ALLOWED_IMAGE_EXTENSIONS = {
+    '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff', '.tif',
+    '.webp', '.svg', '.ico', '.heic', '.heif',
+}
+
+
+def validate_image_path(path: str) -> str | None:
+    """Validate that a local file path points to an image file.
+
+    Returns None if the path is valid, or an error message string otherwise.
+    """
+    resolved = os.path.realpath(path)
+    ext = os.path.splitext(resolved)[1].lower()
+    if ext not in ALLOWED_IMAGE_EXTENSIONS:
+        return (
+            f"Invalid image file type '{ext}'. "
+            f"Allowed types: {', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))}"
+        )
+    if not os.path.isfile(resolved):
+        return f"File not found: {resolved}"
+    return None
+
+
+def validate_url_not_internal(url: str) -> str | None:
+    """Check that a URL does not target private/loopback/link-local addresses.
+
+    Returns None if the URL is safe, or an error message string otherwise.
+    """
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        return "URL has no hostname"
+    try:
+        addr_infos = socket.getaddrinfo(hostname, parsed.port or 443)
+    except socket.gaierror:
+        return f"Could not resolve hostname: {hostname}"
+    for family, _, _, _, sockaddr in addr_infos:
+        ip = ipaddress.ip_address(sockaddr[0])
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+            return f"URL resolves to a non-public address ({ip}), request blocked"
+    return None
+
 bl_info = {
     "name": "Blender MCP",
     "author": "BlenderMCP",
@@ -2091,14 +2136,17 @@ def create_hunyuan_job_main_site(
                 if re.match(r'^https?://', image, re.IGNORECASE) is not None:
                     data["ImageUrl"] = image
                 else:
+                    err = validate_image_path(image)
+                    if err:
+                        return {"error": err}
                     try:
                         # Convert to Base64 format
-                        with open(image, "rb") as f:
+                        with open(os.path.realpath(image), "rb") as f:
                             image_base64 = base64.b64encode(f.read()).decode("ascii")
                         data["ImageBase64"] = image_base64
                     except Exception as e:
                         return {"error": f"Image encoding failed: {str(e)}"}
-            
+
             # Get signed headers
             headers, endpoint = self.get_tencent_cloud_sign_headers("POST", "/", headParams, data, service, region, secret_id, secret_key)
 
@@ -2154,11 +2202,14 @@ def create_hunyuan_job_local_site(
                         image_base64 = base64.b64encode(resImg.content).decode("ascii")
                         data["image"] = image_base64
                     except Exception as e:
-                        return {"error": f"Failed to download or encode image: {str(e)}"} 
+                        return {"error": f"Failed to download or encode image: {str(e)}"}
                 else:
+                    err = validate_image_path(image)
+                    if err:
+                        return {"error": err}
                     try:
                         # Convert to Base64 format
-                        with open(image, "rb") as f:
+                        with open(os.path.realpath(image), "rb") as f:
                             image_base64 = base64.b64encode(f.read()).decode("ascii")
                         data["image"] = image_base64
                     except Exception as e:
@@ -2249,11 +2300,16 @@ def import_generated_asset_hunyuan(self, *args, **kwargs):
     def import_generated_asset_hunyuan_ai(self, name: str , zip_file_url: str):
         if not zip_file_url:
             return {"error": "Zip file not found"}
-        
+
         # Validate URL
         if not re.match(r'^https?://', zip_file_url, re.IGNORECASE):
             return {"error": "Invalid URL format. Must start with http:// or https://"}
-        
+
+        # Block requests to private/internal networks (SSRF protection)
+        ssrf_err = validate_url_not_internal(zip_file_url)
+        if ssrf_err:
+            return {"error": ssrf_err}
+
         # Create a temporary directory
         temp_dir = tempfile.mkdtemp(prefix="tencent_obj_")
         zip_file_path = osp.join(temp_dir, "model.zip")

@@ -9,10 +9,52 @@
 from contextlib import asynccontextmanager
 from typing import AsyncIterator, Dict, Any, List
 import os
+import ipaddress
 from pathlib import Path
 import base64
 from urllib.parse import urlparse
 
+# Allowed image file extensions for local path validation
+ALLOWED_IMAGE_EXTENSIONS = {
+    '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff', '.tif',
+    '.webp', '.svg', '.ico', '.heic', '.heif',
+}
+
+
+def _validate_image_path(path: str) -> str | None:
+    """Validate that a local file path points to an image file.
+
+    Returns None if the path is valid, or an error message string otherwise.
+    """
+    resolved = os.path.realpath(path)
+    ext = os.path.splitext(resolved)[1].lower()
+    if ext not in ALLOWED_IMAGE_EXTENSIONS:
+        return (
+            f"Invalid image file type '{ext}'. "
+            f"Allowed types: {', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))}"
+        )
+    return None
+
+
+def _validate_url_not_internal(url: str) -> str | None:
+    """Check that a URL does not target private/loopback/link-local addresses.
+
+    Returns None if the URL is safe, or an error message string otherwise.
+    """
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        return "URL has no hostname"
+    try:
+        addr_infos = socket.getaddrinfo(hostname, parsed.port or 443)
+    except socket.gaierror:
+        return f"Could not resolve hostname: {hostname}"
+    for family, _, _, _, sockaddr in addr_infos:
+        ip = ipaddress.ip_address(sockaddr[0])
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+            return f"URL resolves to a non-public address ({ip}), request blocked"
+    return None
+
 # Import telemetry
 from .telemetry import record_startup, get_telemetry
 from .telemetry_decorator import telemetry_tool
@@ -1012,6 +1054,12 @@ def generate_hunyuan3d_model(
     - Returns error message if the operation fails
     """
     try:
+        # Validate local image paths to prevent arbitrary file reads
+        if input_image_url and not input_image_url.startswith(("http://", "https://")):
+            err = _validate_image_path(input_image_url)
+            if err:
+                return f"Error: {err}"
+
         blender = get_blender_connection()
         result = blender.send_command("create_hunyuan_job", {
             "text_prompt": text_prompt,
@@ -1027,7 +1075,7 @@ def generate_hunyuan3d_model(
     except Exception as e:
         logger.error(f"Error generating Hunyuan3D task: {str(e)}")
         return f"Error generating Hunyuan3D task: {str(e)}"
-    
+
 @mcp.tool()
 def poll_hunyuan_job_status(
     ctx: Context,
@@ -1073,6 +1121,12 @@ def import_generated_asset_hunyuan(
     Return if the asset has been imported successfully.
     """
     try:
+        # Block requests to private/internal networks (SSRF protection)
+        if zip_file_url:
+            ssrf_err = _validate_url_not_internal(zip_file_url)
+            if ssrf_err:
+                return f"Error: {ssrf_err}"
+
         blender = get_blender_connection()
         kwargs = {
             "name": name

@@ -18,8 +18,70 @@
 from datetime import datetime
 import hashlib, hmac, base64
 import os.path as osp
+import ipaddress
+from urllib.parse import urlparse
 from contextlib import redirect_stdout, suppress
 
+# Allowed image file extensions for local path validation
+ALLOWED_IMAGE_EXTENSIONS = {
+    '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff', '.tif',
+    '.webp', '.svg', '.ico', '.heic', '.heif',
+}
+
+# Maximum local image file size (10 MB)
+MAX_LOCAL_IMAGE_BYTES = 10 * 1024 * 1024
+
+
+def validate_image_path(path: str) -> str | None:
+    """Validate that a local file path points to an image file.
+
+    Returns None if the path is valid, or an error message string otherwise.
+    """
+    resolved = os.path.realpath(path)
+    ext = os.path.splitext(resolved)[1].lower()
+    if ext not in ALLOWED_IMAGE_EXTENSIONS:
+        return (
+            f"Invalid image file type '{ext}'. "
+            f"Allowed types: {', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))}"
+        )
+    if not os.path.isfile(resolved):
+        return f"File not found: {resolved}"
+    try:
+        size = os.path.getsize(resolved)
+    except OSError as e:
+        return f"Cannot read file size: {e}"
+    if size > MAX_LOCAL_IMAGE_BYTES:
+        return (
+            f"Image file too large ({size} bytes). "
+            f"Maximum allowed size is {MAX_LOCAL_IMAGE_BYTES} bytes"
+        )
+    return None
+
+
+def validate_url_not_internal(url: str) -> tuple[None, list[str]] | tuple[str, None]:
+    """Check that a URL does not target private/loopback/link-local addresses.
+
+    Returns (None, validated_ips) if the URL is safe, where validated_ips are
+    the resolved IP strings that should be pinned for the actual request to
+    prevent DNS rebinding (TOCTOU) attacks.
+    Returns (error_message, None) otherwise.
+    """
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        return ("URL has no hostname", None)
+    try:
+        addr_infos = socket.getaddrinfo(hostname, parsed.port or 443)
+    except socket.gaierror:
+        return (f"Could not resolve hostname: {hostname}", None)
+    validated_ips = []
+    for family, _, _, _, sockaddr in addr_infos:
+        ip = ipaddress.ip_address(sockaddr[0])
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+            return (f"URL resolves to a non-public address ({ip}), request blocked", None)
+        validated_ips.append(sockaddr[0])
+    return (None, validated_ips)
+
 bl_info = {
     "name": "Blender MCP",
     "author": "BlenderMCP",
@@ -2089,16 +2151,22 @@ def create_hunyuan_job_main_site(
             # Handling image
             if image:
                 if re.match(r'^https?://', image, re.IGNORECASE) is not None:
+                    ssrf_err, _validated_ips = validate_url_not_internal(image)
+                    if ssrf_err:
+                        return {"error": ssrf_err}
                     data["ImageUrl"] = image
                 else:
+                    err = validate_image_path(image)
+                    if err:
+                        return {"error": err}
                     try:
                         # Convert to Base64 format
-                        with open(image, "rb") as f:
+                        with open(os.path.realpath(image), "rb") as f:
                             image_base64 = base64.b64encode(f.read()).decode("ascii")
                         data["ImageBase64"] = image_base64
                     except Exception as e:
                         return {"error": f"Image encoding failed: {str(e)}"}
-            
+
             # Get signed headers
             headers, endpoint = self.get_tencent_cloud_sign_headers("POST", "/", headParams, data, service, region, secret_id, secret_key)
 
@@ -2148,17 +2216,34 @@ def create_hunyuan_job_local_site(
             # Handling image
             if image:
                 if re.match(r'^https?://', image, re.IGNORECASE) is not None:
+                    # Validate URL is not targeting internal networks
+                    ssrf_err, validated_ips = validate_url_not_internal(image)
+                    if ssrf_err:
+                        return {"error": ssrf_err}
+                    # Pin to validated IP to prevent DNS rebinding
+                    parsed_img = urlparse(image)
+                    pinned_img_url = image.replace(
+                        f"{parsed_img.scheme}://{parsed_img.hostname}",
+                        f"{parsed_img.scheme}://{validated_ips[0]}",
+                        1,
+                    )
                     try:
-                        resImg = requests.get(image)
+                        resImg = requests.get(
+                            pinned_img_url,
+                            headers={"Host": parsed_img.hostname},
+                        )
                         resImg.raise_for_status()
                         image_base64 = base64.b64encode(resImg.content).decode("ascii")
                         data["image"] = image_base64
                     except Exception as e:
-                        return {"error": f"Failed to download or encode image: {str(e)}"} 
+                        return {"error": f"Failed to download or encode image: {str(e)}"}
                 else:
+                    err = validate_image_path(image)
+                    if err:
+                        return {"error": err}
                     try:
                         # Convert to Base64 format
-                        with open(image, "rb") as f:
+                        with open(os.path.realpath(image), "rb") as f:
                             image_base64 = base64.b64encode(f.read()).decode("ascii")
                         data["image"] = image_base64
                     except Exception as e:
@@ -2249,20 +2334,34 @@ def import_generated_asset_hunyuan(self, *args, **kwargs):
     def import_generated_asset_hunyuan_ai(self, name: str , zip_file_url: str):
         if not zip_file_url:
             return {"error": "Zip file not found"}
-        
+
         # Validate URL
         if not re.match(r'^https?://', zip_file_url, re.IGNORECASE):
             return {"error": "Invalid URL format. Must start with http:// or https://"}
-        
+
+        # Block requests to private/internal networks (SSRF protection)
+        ssrf_err, validated_ips = validate_url_not_internal(zip_file_url)
+        if ssrf_err:
+            return {"error": ssrf_err}
+
+        # Pin the download to a validated IP to prevent DNS rebinding
+        parsed = urlparse(zip_file_url)
+        pinned_url = zip_file_url.replace(
+            f"{parsed.scheme}://{parsed.hostname}",
+            f"{parsed.scheme}://{validated_ips[0]}",
+            1,
+        )
+        pin_headers = {"Host": parsed.hostname}
+
         # Create a temporary directory
         temp_dir = tempfile.mkdtemp(prefix="tencent_obj_")
         zip_file_path = osp.join(temp_dir, "model.zip")
         obj_file_path = osp.join(temp_dir, "model.obj")
         mtl_file_path = osp.join(temp_dir, "model.mtl")
 
         try:
-            # Download ZIP file
-            zip_response = requests.get(zip_file_url, stream=True)
+            # Download ZIP file using pinned IP
+            zip_response = requests.get(pinned_url, headers=pin_headers, stream=True)
             zip_response.raise_for_status()
             with open(zip_file_path, "wb") as f:
                 for chunk in zip_response.iter_content(chunk_size=8192):

@@ -9,10 +9,71 @@
 from contextlib import asynccontextmanager
 from typing import AsyncIterator, Dict, Any, List
 import os
+import ipaddress
 from pathlib import Path
 import base64
 from urllib.parse import urlparse
 
+# Allowed image file extensions for local path validation
+ALLOWED_IMAGE_EXTENSIONS = {
+    '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.tiff', '.tif',
+    '.webp', '.svg', '.ico', '.heic', '.heif',
+}
+
+# Maximum local image file size (10 MB)
+MAX_LOCAL_IMAGE_BYTES = 10 * 1024 * 1024
+
+
+def _validate_image_path(path: str) -> str | None:
+    """Validate that a local file path points to an image file.
+
+    Returns None if the path is valid, or an error message string otherwise.
+    """
+    resolved = os.path.realpath(path)
+    ext = os.path.splitext(resolved)[1].lower()
+    if ext not in ALLOWED_IMAGE_EXTENSIONS:
+        return (
+            f"Invalid image file type '{ext}'. "
+            f"Allowed types: {', '.join(sorted(ALLOWED_IMAGE_EXTENSIONS))}"
+        )
+    if not os.path.isfile(resolved):
+        return f"File not found: {resolved}"
+    try:
+        size = os.path.getsize(resolved)
+    except OSError as e:
+        return f"Cannot read file size: {e}"
+    if size > MAX_LOCAL_IMAGE_BYTES:
+        return (
+            f"Image file too large ({size} bytes). "
+            f"Maximum allowed size is {MAX_LOCAL_IMAGE_BYTES} bytes"
+        )
+    return None
+
+
+def _validate_url_not_internal(url: str) -> tuple[None, list[str]] | tuple[str, None]:
+    """Check that a URL does not target private/loopback/link-local addresses.
+
+    Returns (None, validated_ips) if the URL is safe, where validated_ips are
+    the resolved IP strings that should be pinned for the actual request to
+    prevent DNS rebinding (TOCTOU) attacks.
+    Returns (error_message, None) otherwise.
+    """
+    parsed = urlparse(url)
+    hostname = parsed.hostname
+    if not hostname:
+        return ("URL has no hostname", None)
+    try:
+        addr_infos = socket.getaddrinfo(hostname, parsed.port or 443)
+    except socket.gaierror:
+        return (f"Could not resolve hostname: {hostname}", None)
+    validated_ips = []
+    for family, _, _, _, sockaddr in addr_infos:
+        ip = ipaddress.ip_address(sockaddr[0])
+        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+            return (f"URL resolves to a non-public address ({ip}), request blocked", None)
+        validated_ips.append(sockaddr[0])
+    return (None, validated_ips)
+
 # Import telemetry
 from .telemetry import record_startup, get_telemetry
 from .telemetry_decorator import telemetry_tool
@@ -1012,6 +1073,12 @@ def generate_hunyuan3d_model(
     - Returns error message if the operation fails
     """
     try:
+        # Validate local image paths to prevent arbitrary file reads
+        if input_image_url and not input_image_url.startswith(("http://", "https://")):
+            err = _validate_image_path(input_image_url)
+            if err:
+                return f"Error: {err}"
+
         blender = get_blender_connection()
         result = blender.send_command("create_hunyuan_job", {
             "text_prompt": text_prompt,
@@ -1027,7 +1094,7 @@ def generate_hunyuan3d_model(
     except Exception as e:
         logger.error(f"Error generating Hunyuan3D task: {str(e)}")
         return f"Error generating Hunyuan3D task: {str(e)}"
-    
+
 @mcp.tool()
 def poll_hunyuan_job_status(
     ctx: Context,
@@ -1073,6 +1140,12 @@ def import_generated_asset_hunyuan(
     Return if the asset has been imported successfully.
     """
     try:
+        # Block requests to private/internal networks (SSRF protection)
+        if zip_file_url:
+            ssrf_err, _validated_ips = _validate_url_not_internal(zip_file_url)
+            if ssrf_err:
+                return f"Error: {ssrf_err}"
+
         blender = get_blender_connection()
         kwargs = {
             "name": name