scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Fix: Fix custom ca bundle support in aiohttp transport (#12281)

Browse source 
* Unify usage of get_ssl_configuration

* Fix doc
This commit is contained in:
Joost van Doorn 2025-07-04 21:48:20 +02:00 • committed by GitHub
parent 203a10a705
commit 453591ed7c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 195 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

73
docs/my-website/docs/guides/security_settings.md
View file

@ -3,12 +3,43 @@ import TabItem from '@theme/TabItem';
# SSL, HTTP Proxy Security Settings
If you're in an environment using an older TTS bundle, with an older encryption, follow this guide.
If you're in an environment using an older TTS bundle, with an older encryption, follow this guide. By default
LiteLLM uses the certifi CA bundle for SSL verification, which is compatible with most modern servers.
However, if you need to disable SSL verification or use a custom CA bundle, you can do so by following the steps below.
Be aware that environmental variables take precedence over the settings in the SDK.
LiteLLM uses HTTPX for network requests, unless otherwise specified.
LiteLLM uses HTTPX for network requests, unless otherwise specified.
## 1. Disable SSL verification
## 1. Custom CA Bundle
You can set a custom CA bundle file path using the `SSL_CERT_FILE` environmental variable or passing a string to the the ssl_verify setting.
<Tabs>
<TabItem value="sdk" label="SDK">
```python
import litellm
litellm.ssl_verify = "client.pem"
```
</TabItem>
<TabItem value="proxy" label="PROXY">
```yaml
litellm_settings:
ssl_verify: "client.pem"
```
</TabItem>
<TabItem value="env_var" label="Environment Variables">
```bash
export SSL_CERT_FILE="client.pem"
```
</TabItem>
</Tabs>
## 2. Disable SSL verification
<Tabs>
@ -35,14 +66,42 @@ export SSL_VERIFY="False"
</TabItem>
</Tabs>
## 2. Lower security settings
## 3. Lower security settings
The `ssl_security_level` allows setting a lower security level for SSL connections.
<Tabs>
<TabItem value="sdk" label="SDK">
```python
import litellm
litellm.ssl_security_level = "DEFAULT@SECLEVEL=1"
```
</TabItem>
<TabItem value="proxy" label="PROXY">
```yaml
litellm_settings:
ssl_security_level: "DEFAULT@SECLEVEL=1"
```
</TabItem>
<TabItem value="env_var" label="Environment Variables">
```bash
export SSL_SECURITY_LEVEL="DEFAULT@SECLEVEL=1"
```
</TabItem>
</Tabs>
## 4. Certificate authentication
The `SSL_CERTIFICATE` environmental variable or `ssl_certificate` attribute allows setting a client side certificate to authenticate the client to the server.
<Tabs>
<TabItem value="sdk" label="SDK">
```python
import litellm
litellm.ssl_security_level = 1
litellm.ssl_certificate = "/path/to/certificate.pem"
```
</TabItem>
@ -50,20 +109,18 @@ litellm.ssl_certificate = "/path/to/certificate.pem"
```yaml
litellm_settings:
ssl_security_level: 1
ssl_certificate: "/path/to/certificate.pem"
```
</TabItem>
<TabItem value="env_var" label="Environment Variables">
```bash
export SSL_SECURITY_LEVEL="1"
export SSL_CERTIFICATE="/path/to/certificate.pem"
```
</TabItem>
</Tabs>
## 3. Use HTTP_PROXY environment variable
## 5. Use HTTP_PROXY environment variable
Both httpx and aiohttp libraries use `urllib.request.getproxies` from environment variables. Before client initialization, you may set proxy (and optional SSL_CERT_FILE) by setting the environment variables:

1
litellm/__init__.py
View file

@ -214,6 +214,7 @@ use_litellm_proxy: bool = (
)
use_client: bool = False
ssl_verify: Union[str, bool] = True
ssl_security_level: Optional[str] = None
ssl_certificate: Optional[str] = None
disable_streaming_logging: bool = False
disable_token_counter: bool = False

124
litellm/llms/custom_httpx/http_handler.py
View file

@ -5,6 +5,7 @@ import time
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional, Union
import httpx
import certifi
from aiohttp import ClientSession, TCPConnector
from httpx import USE_CLIENT_DEFAULT, AsyncHTTPTransport, HTTPTransport
from httpx._types import RequestFiles
@ -39,6 +40,72 @@ headers = {
_DEFAULT_TIMEOUT = httpx.Timeout(timeout=5.0, connect=5.0)
def get_ssl_configuration(ssl_verify: Optional[VerifyTypes] = None) -> Union[bool, str, ssl.SSLContext]:
"""
Unified SSL configuration function that handles ssl_context and ssl_verify logic.
SSL Configuration Priority:
1. If ssl_verify is provided -> is a SSL context use the custom SSL context
2. If ssl_verify is False -> disable SSL verification (ssl=False)
3. If ssl_verify is a string -> use it as a path to CA bundle file
4. If SSL_CERT_FILE environment variable is set and exists -> use it as CA bundle file
5. Else will use default SSL context with certifi CA bundle
If ssl_security_level is set, it will apply the security level to the SSL context.
Args:
ssl_verify: SSL verification setting. Can be:
- None: Use default from environment/litellm settings
- False: Disable SSL verification
- True: Enable SSL verification
- str: Path to CA bundle file
Returns:
Union[bool, str, ssl.SSLContext]: Appropriate SSL configuration
"""
from litellm.secret_managers.main import str_to_bool
if isinstance(ssl_verify, ssl.SSLContext):
# If ssl_verify is already an SSLContext, return it directly
return ssl_verify
# Get ssl_verify from environment or litellm settings if not provided
if ssl_verify is None:
ssl_verify = os.getenv("SSL_VERIFY", litellm.ssl_verify)
ssl_verify_bool = str_to_bool(ssl_verify) if isinstance(ssl_verify, str) else ssl_verify
if ssl_verify_bool is not None:
ssl_verify = ssl_verify_bool
ssl_security_level = os.getenv("SSL_SECURITY_LEVEL", litellm.ssl_security_level)
cafile = None
if isinstance(ssl_verify, str) and os.path.exists(ssl_verify):
cafile = ssl_verify
if not cafile:
ssl_cert_file = os.getenv("SSL_CERT_FILE")
if ssl_cert_file and os.path.exists(ssl_cert_file):
cafile = ssl_cert_file
else:
cafile = certifi.where()
if ssl_verify is not False:
custom_ssl_context = ssl.create_default_context(
cafile=cafile
)
# If security level is set, apply it to the SSL context
if (
ssl_security_level
and isinstance(ssl_security_level, str)
):
# Create a custom SSL context with reduced security level
custom_ssl_context.set_ciphers(ssl_security_level)
# Use our custom SSL context instead of the original ssl_verify value
return custom_ssl_context
return ssl_verify
def mask_sensitive_info(error_message):
# Find the start of the key parameter
if isinstance(error_message, str):
@ -119,29 +186,8 @@ class AsyncHTTPHandler:
event_hooks: Optional[Mapping[str, List[Callable[..., Any]]]],
ssl_verify: Optional[VerifyTypes] = None,
) -> httpx.AsyncClient:
# SSL certificates (a.k.a CA bundle) used to verify the identity of requested hosts.
# /path/to/certificate.pem
if ssl_verify is None:
ssl_verify = os.getenv("SSL_VERIFY", litellm.ssl_verify)
ssl_security_level = os.getenv("SSL_SECURITY_LEVEL")
# If ssl_verify is not False and we need a lower security level
if (
not ssl_verify
and ssl_security_level
and isinstance(ssl_security_level, str)
):
# Create a custom SSL context with reduced security level
custom_ssl_context = ssl.create_default_context()
custom_ssl_context.set_ciphers(ssl_security_level)
# If ssl_verify is a path to a CA bundle, load it into our custom context
if isinstance(ssl_verify, str) and os.path.exists(ssl_verify):
custom_ssl_context.load_verify_locations(cafile=ssl_verify)
# Use our custom SSL context instead of the original ssl_verify value
ssl_verify = custom_ssl_context
# Get unified SSL configuration
ssl_config = get_ssl_configuration(ssl_verify)
# An SSL certificate used by the requested host to authenticate the client.
# /path/to/client.pem
@ -152,8 +198,8 @@ class AsyncHTTPHandler:
# Create a client with a connection pool
transport = AsyncHTTPHandler._create_async_transport(
ssl_context=ssl_verify if isinstance(ssl_verify, ssl.SSLContext) else None,
ssl_verify=ssl_verify if isinstance(ssl_verify, bool) else None,
ssl_context=ssl_config if isinstance(ssl_config, ssl.SSLContext) else None,
ssl_verify=ssl_config if isinstance(ssl_config, bool) else None,
)
return httpx.AsyncClient(
@ -164,7 +210,7 @@ class AsyncHTTPHandler:
max_connections=concurrent_limit,
max_keepalive_connections=concurrent_limit,
),
verify=ssl_verify,
verify=ssl_config,
cert=cert,
headers=headers,
)
@ -544,7 +590,6 @@ class AsyncHTTPHandler:
SSL Configuration Priority:
1. If ssl_context is provided -> use the custom SSL context
2. If ssl_verify is False -> disable SSL verification (ssl=False)
3. If ssl_verify is True/None -> use default SSL context with certifi CA bundle
Returns:
Dict with appropriate SSL configuration for TCPConnector
@ -559,10 +604,6 @@ class AsyncHTTPHandler:
elif ssl_verify is False:
# Priority 2: Explicitly disable SSL verification
connector_kwargs["verify_ssl"] = False
else:
# Priority 3: Use our default SSL context with certifi CA bundle
# This covers ssl_verify=True and ssl_verify=None cases
connector_kwargs["ssl"] = AsyncHTTPHandler._get_ssl_context()
return connector_kwargs
@ -577,7 +618,6 @@ class AsyncHTTPHandler:
Note: aiohttp TCPConnector ssl parameter accepts:
- SSLContext: custom SSL context
- False: disable SSL verification
- True: use default SSL verification (equivalent to ssl.create_default_context())
"""
from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport
from litellm.secret_managers.main import str_to_bool
@ -600,17 +640,6 @@ class AsyncHTTPHandler:
trust_env=trust_env,
),
)
@staticmethod
def _get_ssl_context() -> ssl.SSLContext:
"""
Get the SSL context for the AiohttpTransport
"""
import certifi
return ssl.create_default_context(
cafile=certifi.where()
)
@staticmethod
def _create_httpx_transport() -> Optional[AsyncHTTPTransport]:
@ -637,11 +666,8 @@ class HTTPHandler:
if timeout is None:
timeout = _DEFAULT_TIMEOUT
# SSL certificates (a.k.a CA bundle) used to verify the identity of requested hosts.
# /path/to/certificate.pem
if ssl_verify is None:
ssl_verify = os.getenv("SSL_VERIFY", litellm.ssl_verify)
# Get unified SSL configuration
ssl_config = get_ssl_configuration(ssl_verify)
# An SSL certificate used by the requested host to authenticate the client.
# /path/to/client.pem
@ -658,7 +684,7 @@ class HTTPHandler:
max_connections=concurrent_limit,
max_keepalive_connections=concurrent_limit,
),
verify=ssl_verify,
verify=ssl_config,
cert=cert,
headers=headers,
)

18
litellm/llms/openai/common_utils.py
View file

@ -4,6 +4,7 @@ Common helpers / utils across al OpenAI endpoints
import hashlib
import json
import ssl
from typing import Any, Dict, List, Literal, Optional, Union
import httpx
@ -15,6 +16,7 @@ from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.custom_httpx.http_handler import (
_DEFAULT_TTL_FOR_HTTPX_CLIENTS,
AsyncHTTPHandler,
get_ssl_configuration,
)
@ -196,10 +198,16 @@ class BaseOpenAILLM:
if litellm.aclient_session is not None:
return litellm.aclient_session
# Get unified SSL configuration
ssl_config = get_ssl_configuration()
return httpx.AsyncClient(
limits=httpx.Limits(max_connections=1000, max_keepalive_connections=100),
verify=litellm.ssl_verify,
transport=AsyncHTTPHandler._create_async_transport(),
verify=ssl_config,
transport=AsyncHTTPHandler._create_async_transport(
ssl_context=ssl_config if isinstance(ssl_config, ssl.SSLContext) else None,
ssl_verify=ssl_config if isinstance(ssl_config, bool) else None,
),
follow_redirects=True,
)
@ -207,8 +215,12 @@ class BaseOpenAILLM:
def _get_sync_http_client() -> Optional[httpx.Client]:
if litellm.client_session is not None:
return litellm.client_session
# Get unified SSL configuration
ssl_config = get_ssl_configuration()
return httpx.Client(
limits=httpx.Limits(max_connections=1000, max_keepalive_connections=100),
verify=litellm.ssl_verify,
verify=ssl_config,
follow_redirects=True,
)

75
tests/test_litellm/llms/custom_httpx/test_http_handler.py
View file

@ -15,34 +15,35 @@ sys.path.insert(
) # Adds the parent directory to the system path
import litellm
from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_ssl_configuration
@pytest.mark.asyncio
async def test_ssl_security_level(monkeypatch):
# Set environment variable for SSL security level
monkeypatch.setenv("SSL_SECURITY_LEVEL", "DEFAULT@SECLEVEL=1")
with patch.dict(os.environ, clear=True):
# Set environment variable for SSL security level
monkeypatch.setenv("SSL_SECURITY_LEVEL", "DEFAULT@SECLEVEL=1")
# Create async client with SSL verification disabled to isolate SSL context testing
client = AsyncHTTPHandler(ssl_verify=False)
# Create async client with SSL verification disabled to isolate SSL context testing
client = AsyncHTTPHandler()
# Get the transport (should be LiteLLMAiohttpTransport)
transport = client.client._transport
# Get the transport (should be LiteLLMAiohttpTransport)
transport = client.client._transport
# Get the aiohttp ClientSession
client_session = transport._get_valid_client_session()
# Get the aiohttp ClientSession
client_session = transport._get_valid_client_session()
# Get the connector from the session
connector = client_session.connector
# Get the connector from the session
connector = client_session.connector
# Get the SSL context from the connector
ssl_context = connector._ssl
print("ssl_context", ssl_context)
# Get the SSL context from the connector
ssl_context = connector._ssl
print("ssl_context", ssl_context)
# Verify that the SSL context exists and has the correct cipher string
assert isinstance(ssl_context, ssl.SSLContext)
# Optionally, check the ciphers string if needed
# assert "DEFAULT@SECLEVEL=1" in ssl_context.get_ciphers()
# Verify that the SSL context exists and has the correct cipher string
assert isinstance(ssl_context, ssl.SSLContext)
# Optionally, check the ciphers string if needed
# assert "DEFAULT@SECLEVEL=1" in ssl_context.get_ciphers()
@pytest.mark.asyncio
@ -151,28 +152,30 @@ async def test_aiohttp_transport_trust_env_setting(monkeypatch):
assert client_session_with_false_env._trust_env == default_trust_env
def test_get_ssl_context():
"""Test that _get_ssl_context() returns a proper SSL context with certifi CA bundle"""
with patch('ssl.create_default_context') as mock_create_context:
# Mock the return value
mock_ssl_context = MagicMock(spec=ssl.SSLContext)
mock_create_context.return_value = mock_ssl_context
# Call the static method
result = AsyncHTTPHandler._get_ssl_context()
# Verify ssl.create_default_context was called with certifi's CA file
expected_ca_file = certifi.where()
mock_create_context.assert_called_once_with(cafile=expected_ca_file)
# Verify it returns the mocked SSL context
assert result == mock_ssl_context
def test_get_ssl_configuration():
"""Test that get_ssl_configuration() returns a proper SSL context with certifi CA bundle
when no environment variables are set."""
with patch.dict(os.environ, clear=True):
with patch('ssl.create_default_context') as mock_create_context:
# Mock the return value
mock_ssl_context = MagicMock(spec=ssl.SSLContext)
mock_create_context.return_value = mock_ssl_context
# Call the static method
result = get_ssl_configuration()
# Verify ssl.create_default_context was called with certifi's CA file
expected_ca_file = certifi.where()
mock_create_context.assert_called_once_with(cafile=expected_ca_file)
# Verify it returns the mocked SSL context
assert result == mock_ssl_context
def test_get_ssl_context_integration():
def test_get_ssl_configuration_integration():
"""Integration test that _get_ssl_context() returns a working SSL context"""
# Call the static method without mocking
ssl_context = AsyncHTTPHandler._get_ssl_context()
ssl_context = get_ssl_configuration()
# Verify it returns an SSLContext instance
assert isinstance(ssl_context, ssl.SSLContext)