From 3c3ed3bcfbf9caeb2abe9f06e5348a9ad6ce6122 Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Sat, 10 Jan 2026 13:25:31 -0800 Subject: [PATCH] fix resposne api utils --- litellm/llms/bedrock/base_aws_llm.py | 44 +-- litellm/llms/bedrock/common_utils.py | 2 +- litellm/llms/bedrock/files/handler.py | 1 - .../llms/bedrock/test_bedrock_ssl_verify.py | 349 ------------------ 4 files changed, 4 insertions(+), 392 deletions(-) delete mode 100644 tests/test_litellm/llms/bedrock/test_bedrock_ssl_verify.py diff --git a/litellm/llms/bedrock/base_aws_llm.py b/litellm/llms/bedrock/base_aws_llm.py index 18e9deb53b..e9cea23ea4 100644 --- a/litellm/llms/bedrock/base_aws_llm.py +++ b/litellm/llms/bedrock/base_aws_llm.py @@ -74,41 +74,6 @@ class BaseAWSLLM: "aws_external_id", ] - def _get_ssl_verify(self): - """ - Get SSL verification setting for boto3 clients. - - This ensures that custom CA certificates are properly used for all AWS API calls, - including STS and Bedrock services. - - Returns: - Union[bool, str]: SSL verification setting - False to disable, True to enable, - or a string path to a CA bundle file - """ - import litellm - from litellm.secret_managers.main import str_to_bool - - # Check environment variable first (highest priority) - ssl_verify = os.getenv("SSL_VERIFY", litellm.ssl_verify) - - # Convert string "False"/"True" to boolean - if isinstance(ssl_verify, str): - # Check if it's a file path - if os.path.exists(ssl_verify): - return ssl_verify - # Otherwise try to convert to boolean - ssl_verify_bool = str_to_bool(ssl_verify) - if ssl_verify_bool is not None: - ssl_verify = ssl_verify_bool - - # Check SSL_CERT_FILE environment variable for custom CA bundle - if ssl_verify is True or ssl_verify == "True": - ssl_cert_file = os.getenv("SSL_CERT_FILE") - if ssl_cert_file and os.path.exists(ssl_cert_file): - return ssl_cert_file - - return ssl_verify - def get_cache_key(self, credential_args: Dict[str, Optional[str]]) -> str: """ Generate a unique cache key based on the credential arguments. @@ -604,7 +569,6 @@ class BaseAWSLLM: "sts", region_name=aws_region_name, endpoint_url=sts_endpoint, - verify=self._get_ssl_verify(), ) # https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html @@ -661,7 +625,7 @@ class BaseAWSLLM: # Create an STS client without credentials with tracer.trace("boto3.client(sts) for manual IRSA"): - sts_client = boto3.client("sts", region_name=region, verify=self._get_ssl_verify()) + sts_client = boto3.client("sts", region_name=region) # Manually assume the IRSA role with the session name verbose_logger.debug( @@ -684,7 +648,6 @@ class BaseAWSLLM: aws_access_key_id=irsa_creds["AccessKeyId"], aws_secret_access_key=irsa_creds["SecretAccessKey"], aws_session_token=irsa_creds["SessionToken"], - verify=self._get_ssl_verify(), ) # Get current caller identity for debugging @@ -723,7 +686,7 @@ class BaseAWSLLM: verbose_logger.debug("Same account role assumption, using automatic IRSA") with tracer.trace("boto3.client(sts) with automatic IRSA"): - sts_client = boto3.client("sts", region_name=region, verify=self._get_ssl_verify()) + sts_client = boto3.client("sts", region_name=region) # Get current caller identity for debugging try: @@ -846,7 +809,7 @@ class BaseAWSLLM: # This allows the web identity token to work automatically if aws_access_key_id is None and aws_secret_access_key is None: with tracer.trace("boto3.client(sts)"): - sts_client = boto3.client("sts", verify=self._get_ssl_verify()) + sts_client = boto3.client("sts") else: with tracer.trace("boto3.client(sts)"): sts_client = boto3.client( @@ -854,7 +817,6 @@ class BaseAWSLLM: aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, aws_session_token=aws_session_token, - verify=self._get_ssl_verify(), ) assume_role_params = { diff --git a/litellm/llms/bedrock/common_utils.py b/litellm/llms/bedrock/common_utils.py index f4b5de8f7c..d62a8bae42 100644 --- a/litellm/llms/bedrock/common_utils.py +++ b/litellm/llms/bedrock/common_utils.py @@ -260,7 +260,7 @@ def init_bedrock_client( status_code=401, ) - sts_client = boto3.client("sts", verify=ssl_verify) + sts_client = boto3.client("sts") # https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/sts/client/assume_role_with_web_identity.html diff --git a/litellm/llms/bedrock/files/handler.py b/litellm/llms/bedrock/files/handler.py index 0350271dc4..d6177e090d 100644 --- a/litellm/llms/bedrock/files/handler.py +++ b/litellm/llms/bedrock/files/handler.py @@ -142,7 +142,6 @@ class BedrockFilesHandler(BaseAWSLLM): aws_secret_access_key=credentials.secret_key, aws_session_token=credentials.token, region_name=aws_region_name, - verify=self._get_ssl_verify(), ) # Download file from S3 diff --git a/tests/test_litellm/llms/bedrock/test_bedrock_ssl_verify.py b/tests/test_litellm/llms/bedrock/test_bedrock_ssl_verify.py deleted file mode 100644 index 9142de295e..0000000000 --- a/tests/test_litellm/llms/bedrock/test_bedrock_ssl_verify.py +++ /dev/null @@ -1,349 +0,0 @@ -""" -Test SSL verification for AWS Bedrock boto3 clients. - -This test ensures that custom CA certificates are properly passed to all boto3 clients -(STS and Bedrock services) to support internal certificate authorities. - -Issue: https://github.com/BerriAI/litellm/issues/XXXX -User reported that SSL_CERT_FILE environment variable and ssl_verify config were not -being applied to boto3 clients, causing "certificate verify failed" errors. -""" - -import os -import sys -import tempfile -from unittest.mock import MagicMock, Mock, patch - -import pytest - -sys.path.insert(0, os.path.abspath("../..")) - -import litellm -from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM -from litellm.llms.bedrock.common_utils import init_bedrock_client - - -class TestBedrockSSLVerify: - """Test suite for SSL verification in Bedrock boto3 clients.""" - - def test_base_aws_llm_get_ssl_verify_default(self): - """Test that _get_ssl_verify returns default value when no custom config is set.""" - base_aws = BaseAWSLLM() - - # Clear any environment variables - os.environ.pop("SSL_VERIFY", None) - os.environ.pop("SSL_CERT_FILE", None) - - # Reset litellm.ssl_verify to default - litellm.ssl_verify = True - - ssl_verify = base_aws._get_ssl_verify() - assert ssl_verify is True - - def test_base_aws_llm_get_ssl_verify_false(self): - """Test that _get_ssl_verify returns False when SSL verification is disabled.""" - base_aws = BaseAWSLLM() - - # Set SSL_VERIFY to False via environment - os.environ["SSL_VERIFY"] = "False" - - ssl_verify = base_aws._get_ssl_verify() - assert ssl_verify is False - - # Clean up - os.environ.pop("SSL_VERIFY", None) - - def test_base_aws_llm_get_ssl_verify_custom_ca_bundle(self): - """Test that _get_ssl_verify returns custom CA bundle path when SSL_CERT_FILE is set.""" - base_aws = BaseAWSLLM() - - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set SSL_CERT_FILE environment variable - os.environ["SSL_CERT_FILE"] = ca_bundle_path - os.environ.pop("SSL_VERIFY", None) - litellm.ssl_verify = True - - ssl_verify = base_aws._get_ssl_verify() - assert ssl_verify == ca_bundle_path - finally: - # Clean up - os.environ.pop("SSL_CERT_FILE", None) - os.unlink(ca_bundle_path) - - def test_base_aws_llm_get_ssl_verify_litellm_config(self): - """Test that _get_ssl_verify uses litellm.ssl_verify when set.""" - base_aws = BaseAWSLLM() - - # Clear environment variables - os.environ.pop("SSL_VERIFY", None) - os.environ.pop("SSL_CERT_FILE", None) - - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set litellm.ssl_verify to custom CA bundle - litellm.ssl_verify = ca_bundle_path - - ssl_verify = base_aws._get_ssl_verify() - # When ssl_verify is a path, it should be returned directly - assert ssl_verify == ca_bundle_path - finally: - # Clean up - litellm.ssl_verify = True - os.unlink(ca_bundle_path) - - @patch("boto3.client") - def test_init_bedrock_client_passes_ssl_verify_to_sts(self, mock_boto3_client): - """Test that init_bedrock_client passes ssl_verify to STS client.""" - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set SSL_CERT_FILE environment variable - os.environ["SSL_CERT_FILE"] = ca_bundle_path - litellm.ssl_verify = True - - # Mock the STS client and Bedrock client - mock_sts_client = MagicMock() - mock_sts_response = { - "Credentials": { - "AccessKeyId": "test_access_key", - "SecretAccessKey": "test_secret_key", - "SessionToken": "test_session_token", - } - } - mock_sts_client.assume_role.return_value = mock_sts_response - - mock_bedrock_client = MagicMock() - - # Configure mock to return different clients based on service name - def side_effect(service_name=None, **kwargs): - if service_name == "sts": - return mock_sts_client - elif service_name == "bedrock-runtime": - return mock_bedrock_client - return MagicMock() - - mock_boto3_client.side_effect = side_effect - - # Call init_bedrock_client with role assumption - client = init_bedrock_client( - aws_region_name="us-west-2", - aws_access_key_id="test_key", - aws_secret_access_key="test_secret", - aws_role_name="arn:aws:iam::123456789012:role/test-role", - aws_session_name="test-session", - ) - - # Verify that boto3.client was called with verify parameter for STS - sts_calls = [ - call for call in mock_boto3_client.call_args_list - if (len(call[0]) > 0 and call[0][0] == "sts") or - ("service_name" not in call[1]) # STS calls don't use service_name kwarg - ] - - assert len(sts_calls) > 0, "STS client should have been created" - - # Check that verify parameter was passed to STS client - sts_call = sts_calls[0] - assert "verify" in sts_call[1], "verify parameter should be passed to STS client" - assert sts_call[1]["verify"] == ca_bundle_path, f"verify should be set to CA bundle path, got {sts_call[1]['verify']}" - - # Verify that boto3.client was called with verify parameter for Bedrock - bedrock_calls = [ - call for call in mock_boto3_client.call_args_list - if "service_name" in call[1] and call[1]["service_name"] == "bedrock-runtime" - ] - - assert len(bedrock_calls) > 0, "Bedrock client should have been created" - - bedrock_call = bedrock_calls[0] - assert "verify" in bedrock_call[1], "verify parameter should be passed to Bedrock client" - assert bedrock_call[1]["verify"] == ca_bundle_path, f"verify should be set to CA bundle path, got {bedrock_call[1]['verify']}" - - finally: - # Clean up - os.environ.pop("SSL_CERT_FILE", None) - os.unlink(ca_bundle_path) - - @patch("boto3.client") - def test_base_aws_llm_auth_with_role_passes_ssl_verify(self, mock_boto3_client): - """Test that _auth_with_aws_role passes ssl_verify to STS client.""" - base_aws = BaseAWSLLM() - - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set SSL_CERT_FILE environment variable - os.environ["SSL_CERT_FILE"] = ca_bundle_path - litellm.ssl_verify = True - - # Mock the STS client - mock_sts_client = MagicMock() - mock_sts_response = { - "Credentials": { - "AccessKeyId": "test_access_key", - "SecretAccessKey": "test_secret_key", - "SessionToken": "test_session_token", - "Expiration": "2025-01-10T00:00:00Z", - } - } - - # Convert Expiration to datetime - from datetime import datetime, timezone - mock_sts_response["Credentials"]["Expiration"] = datetime.now(timezone.utc) - - mock_sts_client.assume_role.return_value = mock_sts_response - mock_boto3_client.return_value = mock_sts_client - - # Call _auth_with_aws_role - credentials, ttl = base_aws._auth_with_aws_role( - aws_access_key_id="test_key", - aws_secret_access_key="test_secret", - aws_session_token=None, - aws_role_name="arn:aws:iam::123456789012:role/test-role", - aws_session_name="test-session", - ) - - # Verify that boto3.client was called with verify parameter - assert mock_boto3_client.called, "boto3.client should have been called" - - call_kwargs = mock_boto3_client.call_args[1] - assert "verify" in call_kwargs, "verify parameter should be passed to STS client" - assert call_kwargs["verify"] == ca_bundle_path, f"verify should be set to CA bundle path, got {call_kwargs['verify']}" - - finally: - # Clean up - os.environ.pop("SSL_CERT_FILE", None) - os.unlink(ca_bundle_path) - - @patch("litellm.llms.bedrock.base_aws_llm.get_secret") - @patch("boto3.client") - def test_base_aws_llm_auth_with_web_identity_passes_ssl_verify(self, mock_boto3_client, mock_get_secret): - """Test that _auth_with_web_identity_token passes ssl_verify to STS client.""" - base_aws = BaseAWSLLM() - - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set SSL_CERT_FILE environment variable - os.environ["SSL_CERT_FILE"] = ca_bundle_path - litellm.ssl_verify = True - - # Mock get_secret to return the token - mock_get_secret.return_value = "mocked_oidc_token" - - # Mock the STS client - mock_sts_client = MagicMock() - mock_sts_response = { - "Credentials": { - "AccessKeyId": "test_access_key", - "SecretAccessKey": "test_secret_key", - "SessionToken": "test_session_token", - }, - "PackedPolicySize": 100, - } - - mock_sts_client.assume_role_with_web_identity.return_value = mock_sts_response - - # Mock boto3.Session - mock_session = MagicMock() - mock_credentials = MagicMock() - mock_session.get_credentials.return_value = mock_credentials - - mock_boto3_client.return_value = mock_sts_client - - with patch("boto3.Session", return_value=mock_session): - # Call _auth_with_web_identity_token - credentials, ttl = base_aws._auth_with_web_identity_token( - aws_web_identity_token="test_token", - aws_role_name="arn:aws:iam::123456789012:role/test-role", - aws_session_name="test-session", - aws_region_name="us-west-2", - aws_sts_endpoint=None, - ) - - # Verify that boto3.client was called with verify parameter - assert mock_boto3_client.called, "boto3.client should have been called" - - call_kwargs = mock_boto3_client.call_args[1] - assert "verify" in call_kwargs, "verify parameter should be passed to STS client" - assert call_kwargs["verify"] == ca_bundle_path, f"verify should be set to CA bundle path, got {call_kwargs['verify']}" - - finally: - # Clean up - os.environ.pop("SSL_CERT_FILE", None) - os.unlink(ca_bundle_path) - - def test_ssl_verify_priority_env_over_litellm_config(self): - """Test that SSL_VERIFY environment variable takes priority over litellm.ssl_verify.""" - base_aws = BaseAWSLLM() - - # Set litellm.ssl_verify to True - litellm.ssl_verify = True - - # Set SSL_VERIFY environment variable to False - os.environ["SSL_VERIFY"] = "False" - - try: - ssl_verify = base_aws._get_ssl_verify() - assert ssl_verify is False, "Environment variable should take priority" - finally: - # Clean up - os.environ.pop("SSL_VERIFY", None) - litellm.ssl_verify = True - - def test_ssl_cert_file_priority_over_default(self): - """Test that SSL_CERT_FILE takes priority when ssl_verify is True.""" - base_aws = BaseAWSLLM() - - # Create a temporary CA bundle file - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as f: - f.write("-----BEGIN CERTIFICATE-----\n") - f.write("FAKE CERTIFICATE FOR TESTING\n") - f.write("-----END CERTIFICATE-----\n") - ca_bundle_path = f.name - - try: - # Set SSL_CERT_FILE environment variable - os.environ["SSL_CERT_FILE"] = ca_bundle_path - os.environ.pop("SSL_VERIFY", None) - litellm.ssl_verify = True - - ssl_verify = base_aws._get_ssl_verify() - assert ssl_verify == ca_bundle_path, "SSL_CERT_FILE should be used when ssl_verify is True" - finally: - # Clean up - os.environ.pop("SSL_CERT_FILE", None) - os.unlink(ca_bundle_path) - - -if __name__ == "__main__": - # Run tests - pytest.main([__file__, "-v", "-s"])