Fix security vuls

nemo_deploy/multimodal/image_url_validator.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+# Copyright (c) 2026, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+# Ranges that must never be reachable via a request-controlled image URL.
+# 169.254.0.0/16 is the cloud IMDS range (AWS/GCP/Azure 169.254.169.254) —
+# the primary SSRF target in cloud deployments.
+_BLOCKED_NETWORKS = [
+    ipaddress.ip_network("127.0.0.0/8"),  # loopback — server's own local services
+    ipaddress.ip_network("10.0.0.0/8"),  # RFC 1918 private
+    ipaddress.ip_network("172.16.0.0/12"),  # RFC 1918 private
+    ipaddress.ip_network("192.168.0.0/16"),  # RFC 1918 private
+    ipaddress.ip_network("169.254.0.0/16"),  # link-local / cloud IMDS
+    ipaddress.ip_network("::1/128"),  # IPv6 loopback
+    ipaddress.ip_network("fc00::/7"),  # IPv6 unique-local
+]
+
+
+def validate_image_url(url: str) -> None:
+    """Raise ValueError if url is not a safe http/https URL.
+
+    Rejects file://, non-http(s) schemes, and URLs that resolve to
+    private/link-local/loopback ranges (SSRF guard).
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError(f"Unsupported image URL scheme '{parsed.scheme}'. Only http and https are allowed.")
+    hostname = parsed.hostname
+    if not hostname:
+        raise ValueError("Image URL has no hostname.")
+    try:
+        resolved_ip = ipaddress.ip_address(socket.gethostbyname(hostname))
+    except (socket.gaierror, ValueError) as exc:
+        raise ValueError(f"Cannot resolve image URL hostname '{hostname}': {exc}") from exc
+    for net in _BLOCKED_NETWORKS:
+        if resolved_ip in net:
+            raise ValueError(
+                f"Image URL resolves to a blocked address ({resolved_ip}). "
+                "Private, loopback, and link-local addresses are not allowed."
+            )

nemo_deploy/multimodal/megatron_multimodal_deployable.py

@@ -173,6 +173,13 @@ def process_image_input(self, image_source):
         if isinstance(self.inference_wrapped_model, QwenVLInferenceWrapper):
             from qwen_vl_utils import process_vision_info
 
+            from nemo_deploy.multimodal.image_url_validator import validate_image_url
+
+            # data: URIs are inline base64 and never trigger a network request.
+            # All other values are treated as URLs and must pass the SSRF guard.
+            if not image_source.startswith("data:"):
+                validate_image_url(image_source)
+
             messages = [
                 {
                     "role": "user",

nemo_deploy/multimodal/query_multimodal.py

@@ -100,6 +100,9 @@ def setup_media(self, input_media):
                 raise UnavailableError(MISSING_PIL_MSG)
 
             if input_media.startswith("http") or input_media.startswith("https"):
+                from nemo_deploy.multimodal.image_url_validator import validate_image_url
+
+                validate_image_url(input_media)
                 response = requests.get(input_media, timeout=5)
                 media = Image.open(BytesIO(response.content)).convert("RGB")
             else:

tests/unit_tests/deploy/test_image_url_validator.py

@@ -0,0 +1,110 @@
+# Copyright (c) 2025, NVIDIA CORPORATION.  All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import importlib.util
+import pathlib
+import socket
+from unittest.mock import patch
+
+import pytest
+
+# Load the validator directly by file path so we don't trigger nemo_deploy/__init__.py
+# (which requires torch/triton). The module itself is pure stdlib.
+_validator_path = pathlib.Path(__file__).resolve().parents[3] / "nemo_deploy" / "multimodal" / "image_url_validator.py"
+_spec = importlib.util.spec_from_file_location("image_url_validator", _validator_path)
+_mod = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_mod)
+validate_image_url = _mod.validate_image_url
+
+
+def _mock_resolve(ip_str):
+    """Return a patch for socket.gethostbyname that always resolves to ip_str."""
+    return patch.object(_mod.socket, "gethostbyname", return_value=ip_str)
+
+
+class TestBlockedSchemes:
+    def test_file_scheme_rejected(self):
+        with pytest.raises(ValueError, match="scheme"):
+            validate_image_url("file:///etc/passwd")
+
+    def test_ftp_scheme_rejected(self):
+        with pytest.raises(ValueError, match="scheme"):
+            validate_image_url("ftp://example.com/img.png")
+
+    def test_no_scheme_rejected(self):
+        with pytest.raises(ValueError, match="scheme"):
+            validate_image_url("example.com/img.png")
+
+
+class TestBlockedRanges:
+    def test_loopback_ipv4_rejected(self):
+        with _mock_resolve("127.0.0.1"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://localhost/img.jpg")
+
+    def test_loopback_other_subnet_rejected(self):
+        with _mock_resolve("127.1.2.3"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://internal.local/img.jpg")
+
+    def test_cloud_imds_rejected(self):
+        # 169.254.169.254 is the AWS/GCP/Azure metadata service
+        with _mock_resolve("169.254.169.254"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://169.254.169.254/latest/meta-data/")
+
+    def test_link_local_rejected(self):
+        with _mock_resolve("169.254.0.1"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://169.254.0.1/img.jpg")
+
+    def test_rfc1918_10_rejected(self):
+        with _mock_resolve("10.0.0.1"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://internal.corp/img.jpg")
+
+    def test_rfc1918_172_rejected(self):
+        with _mock_resolve("172.16.0.1"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://internal.corp/img.jpg")
+
+    def test_rfc1918_192_rejected(self):
+        with _mock_resolve("192.168.1.1"):
+            with pytest.raises(ValueError, match="blocked address"):
+                validate_image_url("http://192.168.1.1/img.jpg")
+
+
+class TestAllowedUrls:
+    def test_public_https_allowed(self):
+        with _mock_resolve("93.184.216.34"):  # example.com
+            validate_image_url("https://example.com/image.jpg")  # must not raise
+
+    def test_public_http_allowed(self):
+        with _mock_resolve("1.2.3.4"):
+            validate_image_url("http://cdn.example.com/image.png")  # must not raise
+
+
+class TestNoHostname:
+    def test_url_without_hostname_rejected(self):
+        with pytest.raises(ValueError, match="hostname"):
+            validate_image_url("http:///image.jpg")
+
+    def test_dns_failure_rejected(self):
+        with patch.object(
+            _mod.socket,
+            "gethostbyname",
+            side_effect=socket.gaierror("Name or service not known"),
+        ):
+            with pytest.raises(ValueError, match="Cannot resolve"):
+                validate_image_url("http://nonexistent.invalid/img.jpg")