CVE-2026-33626

@@ -164,7 +164,7 @@ def metrics():
     #     {
     #         "name": "切换至简体中文",
     #         "url": "https://lmdeploy.readthedocs.io/en/latest",
-    #         "icon": "https://img.shields.io/badge/Doc-%E7%AE%80%E4%BD%93%E4%B8%AD%E6%96%87-blue", # noqa: #501
+    #         "icon": "https://img.shields.io/badge/Doc-%E7%AE%80%E4%BD%93%E4%B8%AD%E6%96%87-blue", # noqa: E501
     #         "type": "url",
     #     },
     # ],

@@ -57,7 +57,13 @@ def _update_torch_dtype(config: 'ModelConfig', dtype: str, device_type: str = 'a
             torch_dtype = torch_dtype if torch_dtype in ['float16', 'bfloat16'] else 'float16'
         else:
             torch_dtype = dtype
-    config.dtype = eval(f'torch.{torch_dtype}')
+
+    resolved_dtype = getattr(torch, torch_dtype, None)
+    if not isinstance(resolved_dtype, torch.dtype):
+        raise ValueError(f'Invalid torch dtype "{torch_dtype}" resolved from model config; '
+                         'expected a torch.dtype attribute on torch.')
+    config.dtype = resolved_dtype
+
     return config
 
 
@@ -629,8 +635,11 @@ def from_config(cls, hf_config: Any):
         else:
             raise TypeError(f'Unsupported quant method: {quant_method}')
 
-        if quant_dtype is not None:
-            quant_dtype = eval(f'torch.{quant_dtype}')
+        resolved_quant_dtype = getattr(torch, quant_dtype, None)
+        if not isinstance(resolved_quant_dtype, torch.dtype):
+            raise ValueError(f'Invalid quant dtype "{quant_dtype}" resolved from model config; '
+                             'expected a torch.dtype attribute on torch.')
+        quant_dtype = resolved_quant_dtype
 
         ignored_layers = quant_config.get('ignored_layers', [])
         if not ignored_layers:

@@ -1,5 +1,7 @@
 # Copyright (c) OpenMMLab. All rights reserved.
+import ipaddress
 import os
+import socket
 from pathlib import Path
 from typing import TypeVar
 from urllib.parse import ParseResult, urlparse
@@ -20,9 +22,40 @@
 }
 
 
+def _is_safe_url(url: str) -> tuple[bool, str]:
+    """Check if the URL is safe to fetch (not internal/private)."""
+    try:
+        parsed = urlparse(url)
+        if parsed.scheme not in ('http', 'https'):
+            return False, f'Unsupported scheme: {parsed.scheme}'
+
+        hostname = parsed.hostname
+        if not hostname:
+            return False, 'Could not parse hostname from URL'
+
+        # check all IPs (IPv4 + IPv6) using getaddrinfo
+        try:
+            infos = socket.getaddrinfo(hostname, None)
+        except socket.gaierror:
+            return False, 'Hostname resolution failed'
+
+        for info in infos:
+            ip = ipaddress.ip_address(info[4][0])
+            # block any IP that is not globally routable (covers private, loopback,
+            # link-local, multicast, reserved, unspecified, etc.)
+            if not ip.is_global:
+                return False, f'Blocked non-global IP detected: {ip}'
+
+        return True, 'URL is safe'
+    except Exception as e:
+        return False, f'URL validation failed: {str(e)}'
+
+
 def _load_http_url(url_spec: ParseResult, media_io: MediaIO[_M]) -> _M:
-    if url_spec.scheme not in ('http', 'https'):
-        raise ValueError(f'Unsupported URL scheme: {url_spec.scheme}')
+    url = url_spec.geturl()
+    is_safe, reason = _is_safe_url(url)
+    if not is_safe:
+        raise ValueError(f'URL is blocked for security reasons: {reason}')
 
     fetch_timeout = 10
     if isinstance(media_io, ImageMediaIO):
@@ -31,7 +64,8 @@ def _load_http_url(url_spec: ParseResult, media_io: MediaIO[_M]) -> _M:
         fetch_timeout = int(os.environ.get('LMDEPLOY_VIDEO_FETCH_TIMEOUT', 30))
 
     client = requests.Session()
-    response = client.get(url_spec.geturl(), headers=headers, timeout=fetch_timeout)
+    client.max_redirects = 3
+    response = client.get(url_spec.geturl(), headers=headers, timeout=fetch_timeout, allow_redirects=True)
     response.raise_for_status()
 
     return media_io.load_bytes(response.content)

@@ -0,0 +1,47 @@
+import socket
+from unittest.mock import MagicMock, patch
+from urllib.parse import urlparse
+
+import pytest
+
+from lmdeploy.vl.media.connection import _is_safe_url, _load_http_url
+
+
+@pytest.mark.parametrize(
+    'url,expected_safe,mock_ips',
+    [
+        ('https://github.com', True, ['140.82.112.3']),  # Public domain
+        ('http://8.8.8.8', True, ['8.8.8.8']),  # Public IPv4
+        ('ftp://example.com', False, []),  # Forbidden scheme
+        ('http://127.0.0.1', False, ['127.0.0.1']),  # IPv4 loopback
+        ('http://localhost', False, ['127.0.0.1']),  # Resolves to loopback
+        ('http://169.254.169.254', False, ['169.254.169.254']),  # Cloud metadata service
+        ('http://[::1]', False, ['::1']),  # IPv6 loopback
+        ('http://[fc00::1]', False, ['fc00::1']),  # IPv6 unique local address
+        ('http://mixed-dns.com', False, ['1.1.1.1', '10.0.0.1']),  # DNS Rebinding
+        ('http://', False, []),  # Empty host
+        ('http://invalid-host-name', False, None),  # Invalid hostname (simulate DNS failure)
+    ])
+def test_is_safe_url(url, expected_safe, mock_ips):
+    with patch('socket.getaddrinfo') as mock_gai:
+        if mock_ips is None:
+            # simulate DNS resolution failure
+            mock_gai.side_effect = socket.gaierror('Hostname resolution failed')
+        else:
+            mock_gai.return_value = [(socket.AF_INET if '.' in ip else socket.AF_INET6, None, None, None, (ip, 80))
+                                     for ip in mock_ips]
+        is_safe, _ = _is_safe_url(url)
+        assert is_safe == expected_safe
+
+
+@patch('requests.Session.get')
+@patch('lmdeploy.vl.media.connection._is_safe_url', return_value=(True, ''))
+def test_load_http_url_logic(mock_safe, mock_get):
+    media_io = MagicMock()
+    url_spec = urlparse('https://example.com/img.jpg')
+
+    # test success with allow_redirects=True
+    mock_get.return_value = MagicMock(content=b'data', status_code=200)
+    media_io.load_bytes.return_value = 'loaded'
+    assert _load_http_url(url_spec, media_io) == 'loaded'
+    assert mock_get.call_args.kwargs['allow_redirects'] is True