fix(aiohttp/ssrf): add allow_requests_to_internal_ips opt-out and fix asyncio deprecation

Adds litellm.allow_requests_to_internal_ips flag (default False) so self-hosted
/ on-prem deployments pointing api_base at an internal address (e.g. Ollama,
vLLM) can opt out of SSRF protection rather than being unconditionally blocked.
Both _assert_not_private_url and _SSRFGuardResolver respect the flag.

Also replaces the deprecated asyncio.get_event_loop() call in BaseLLMAIOHTTPHandler.__del__
with asyncio.get_running_loop(), falling back to a fresh event loop — eliminating
the DeprecationWarning raised in Python 3.10+.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

litellm/__init__.py

@@ -456,6 +456,11 @@ force_ipv4: bool = (
     False  # when True, litellm will force ipv4 for all LLM requests. Some users have seen httpx ConnectionError when using ipv6.
 )
 network_mock: bool = False  # When True, use mock transport — no real network calls
+allow_requests_to_internal_ips: bool = (
+    False  # When True, disables SSRF protection for private/reserved IPs.
+    # Only set this for self-hosted or on-prem deployments where api_base
+    # intentionally points at an internal address (e.g. local Ollama, vLLM).
+)
 
 ####### STOP SEQUENCE LIMIT #######
 disable_stop_sequence_limit: bool = False  # when True, stop sequence limit is disabled

litellm/llms/custom_httpx/aiohttp_handler.py

@@ -63,7 +63,12 @@ def _assert_not_private_url(url: str) -> None:
 
     Validates all DNS answers, not just the first, to prevent A-record rotation attacks.
     Used as a fast-fail guard on the sync path (httpx) and as defence-in-depth on async.
+
+    Set ``litellm.allow_requests_to_internal_ips = True`` to disable this check
+    for self-hosted / on-prem deployments where api_base is an internal address.
     """
+    if litellm.allow_requests_to_internal_ips:
+        return
     parsed = urlparse(url)
     hostname = parsed.hostname
     if not hostname:
@@ -104,17 +109,18 @@ class _SSRFGuardResolver(AbstractResolver):
             )
         except socket.gaierror:
             raise  # Propagate so aiohttp wraps it in a ClientConnectorError
-        for info in infos:
-            raw_ip = info[4][0]
-            try:
-                addr = ipaddress.ip_address(raw_ip)
-            except ValueError:
-                continue
-            if _is_blocked_address(addr):
-                raise ValueError(
-                    f"Host '{host}' resolves to a private/reserved IP address "
-                    f"({raw_ip}) which is not allowed (SSRF protection)"
-                )
+        if not litellm.allow_requests_to_internal_ips:
+            for info in infos:
+                raw_ip = info[4][0]
+                try:
+                    addr = ipaddress.ip_address(raw_ip)
+                except ValueError:
+                    continue
+                if _is_blocked_address(addr):
+                    raise ValueError(
+                        f"Host '{host}' resolves to a private/reserved IP address "
+                        f"({raw_ip}) which is not allowed (SSRF protection)"
+                    )
         return [
             {
                 "hostname": host,
@@ -250,26 +256,17 @@ class BaseLLMAIOHTTPHandler:
             and self._owns_session
         ):
             try:
-                import asyncio
-
                 try:
-                    loop = asyncio.get_event_loop()
-                    if loop.is_running():
-                        # Event loop is running - schedule cleanup task
-                        asyncio.create_task(self.close())
-                    else:
-                        # Event loop exists but not running - run cleanup
-                        loop.run_until_complete(self.close())
+                    loop = asyncio.get_running_loop()
+                    loop.create_task(self.close())
                 except RuntimeError:
-                    # No event loop available - create one for cleanup
+                    # No running loop — run cleanup in a temporary one.
                     loop = asyncio.new_event_loop()
-                    asyncio.set_event_loop(loop)
                     try:
                         loop.run_until_complete(self.close())
                     finally:
                         loop.close()
             except Exception:
-                # Silently ignore errors during __del__ to avoid issues
                 pass
 
     async def _make_common_async_call(

tests/test_litellm/llms/test_aiohttp_ssrf_protection.py

@@ -3,6 +3,7 @@ import ipaddress
 import pytest
 from unittest.mock import patch
 
+import litellm
 from litellm.llms.custom_httpx.aiohttp_handler import (
     _SSRFGuardResolver,
     _assert_not_private_url,
@@ -90,6 +91,61 @@ class TestAiohttpSSRFProtection:
             _assert_not_private_url("https://example.com/")  # should not raise
 
 
+class TestAllowInternalIpsOptOut:
+    """allow_requests_to_internal_ips=True disables SSRF protection for on-prem use."""
+
+    def setup_method(self):
+        self._original = litellm.allow_requests_to_internal_ips
+
+    def teardown_method(self):
+        litellm.allow_requests_to_internal_ips = self._original
+
+    def test_private_ip_allowed_when_flag_set(self):
+        litellm.allow_requests_to_internal_ips = True
+        _assert_not_private_url("http://10.0.0.1/internal")  # must not raise
+
+    def test_localhost_allowed_when_flag_set(self):
+        litellm.allow_requests_to_internal_ips = True
+        _assert_not_private_url("http://127.0.0.1:11434/api/chat")  # Ollama local
+
+    def test_aws_metadata_allowed_when_flag_set(self):
+        litellm.allow_requests_to_internal_ips = True
+        _assert_not_private_url("http://169.254.169.254/latest/meta-data/")
+
+    def test_flag_false_still_blocks(self):
+        litellm.allow_requests_to_internal_ips = False
+        with pytest.raises(ValueError, match="private/reserved"):
+            _assert_not_private_url("http://10.0.0.1/internal")
+
+    @pytest.mark.asyncio
+    async def test_resolver_allows_private_when_flag_set(self):
+        litellm.allow_requests_to_internal_ips = True
+        resolver = _SSRFGuardResolver()
+        mock_infos = [(2, 1, 6, "", ("10.0.0.1", 443))]
+
+        async def run():
+            loop = asyncio.get_running_loop()
+            with patch.object(loop, "getaddrinfo", return_value=mock_infos):
+                result = await resolver.resolve("internal.corp", 443)
+            assert result[0]["host"] == "10.0.0.1"
+
+        await run()
+
+    @pytest.mark.asyncio
+    async def test_resolver_blocks_private_when_flag_false(self):
+        litellm.allow_requests_to_internal_ips = False
+        resolver = _SSRFGuardResolver()
+        mock_infos = [(2, 1, 6, "", ("10.0.0.1", 443))]
+
+        async def run():
+            loop = asyncio.get_running_loop()
+            with patch.object(loop, "getaddrinfo", return_value=mock_infos):
+                with pytest.raises(ValueError, match="private/reserved"):
+                    await resolver.resolve("evil.internal", 443)
+
+        await run()
+
+
 class TestSSRFGuardOnRequestMethods:
     """Verify _assert_not_private_url is actually called in the request paths."""