From 0737cc7c138bb7914c0951af0dac864beed58499 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Mon, 3 Nov 2025 13:56:00 -0800 Subject: [PATCH] [Feat] s3 logger, add support for ssl_verify when using minio logger (#16211) * fixes s3_v2 verify * test_s3_verify_false_async_client * fix * ruff fixes --- litellm/integrations/arize/_utils.py | 4 +- .../langfuse/langfuse_otel_attributes.py | 4 +- .../base_otel_llm_obs_attributes.py | 2 +- litellm/integrations/s3_v2.py | 30 ++- .../count_tokens/handler.py | 1 - litellm/proxy/proxy_config.yaml | 24 ++- tests/test_litellm/integrations/test_s3_v2.py | 182 ++++++++++++++++++ 7 files changed, 222 insertions(+), 25 deletions(-) diff --git a/litellm/integrations/arize/_utils.py b/litellm/integrations/arize/_utils.py index 1359f279ee9..f045cba5701 100644 --- a/litellm/integrations/arize/_utils.py +++ b/litellm/integrations/arize/_utils.py @@ -1,6 +1,5 @@ import json -from abc import ABC -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Type, Union +from typing import TYPE_CHECKING, Any, Dict, Optional, Type from typing_extensions import override @@ -101,7 +100,6 @@ def set_attributes( SpanAttributes, ToolCallAttributes, ) - from litellm.litellm_core_utils.safe_json_dumps import safe_dumps try: optional_params = kwargs.get("optional_params", {}) diff --git a/litellm/integrations/langfuse/langfuse_otel_attributes.py b/litellm/integrations/langfuse/langfuse_otel_attributes.py index f14412ad53b..fb4a0a6a36c 100644 --- a/litellm/integrations/langfuse/langfuse_otel_attributes.py +++ b/litellm/integrations/langfuse/langfuse_otel_attributes.py @@ -5,13 +5,11 @@ Relevant Issue: https://github.com/BerriAI/litellm/issues/13764 """ import json -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union +from typing import TYPE_CHECKING, Any, Dict, Optional, Union -from numpy import isin from pydantic import BaseModel from typing_extensions import override -import litellm from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes import ( BaseLLMObsOTELAttributes, safe_set_attribute, diff --git a/litellm/integrations/opentelemetry_utils/base_otel_llm_obs_attributes.py b/litellm/integrations/opentelemetry_utils/base_otel_llm_obs_attributes.py index 8d7b2d681ab..f74da8231f3 100644 --- a/litellm/integrations/opentelemetry_utils/base_otel_llm_obs_attributes.py +++ b/litellm/integrations/opentelemetry_utils/base_otel_llm_obs_attributes.py @@ -1,5 +1,5 @@ from abc import ABC -from typing import TYPE_CHECKING, Any, Dict, List, Union +from typing import TYPE_CHECKING, Any, Dict, Union if TYPE_CHECKING: from opentelemetry.trace import Span diff --git a/litellm/integrations/s3_v2.py b/litellm/integrations/s3_v2.py index cc44450e737..37fe4fbb81c 100644 --- a/litellm/integrations/s3_v2.py +++ b/litellm/integrations/s3_v2.py @@ -57,12 +57,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): f"in init s3 logger - s3_callback_params {litellm.s3_callback_params}" ) - # IMPORTANT: We use a concurrent limit of 1 to upload to s3 - # Files should get uploaded BUT they should not impact latency of LLM calling logic - self.async_httpx_client = get_async_httpx_client( - llm_provider=httpxSpecialProvider.LoggingCallback, - ) - + # Initialize S3 params first to get the correct s3_verify value self._init_s3_params( s3_bucket_name=s3_bucket_name, s3_region_name=s3_region_name, @@ -85,6 +80,16 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): ) verbose_logger.debug(f"s3 logger using endpoint url {s3_endpoint_url}") + # IMPORTANT + # Create httpx client AFTER _init_s3_params so we have the correct s3_verify value + verbose_logger.debug( + f"s3_v2 logger creating async httpx client with s3_verify={self.s3_verify}" + ) + self.async_httpx_client = get_async_httpx_client( + llm_provider=httpxSpecialProvider.LoggingCallback, + params={"ssl_verify": self.s3_verify} + ) + asyncio.create_task(self.periodic_flush()) self.flush_lock = asyncio.Lock() @@ -147,9 +152,11 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): litellm.s3_callback_params.get("s3_api_version") or s3_api_version ) self.s3_use_ssl = ( - litellm.s3_callback_params.get("s3_use_ssl", True) or s3_use_ssl + litellm.s3_callback_params.get("s3_use_ssl", True) if litellm.s3_callback_params.get("s3_use_ssl") is not None else s3_use_ssl + ) + self.s3_verify = ( + litellm.s3_callback_params.get("s3_verify") if litellm.s3_callback_params.get("s3_verify") is not None else s3_verify ) - self.s3_verify = litellm.s3_callback_params.get("s3_verify") or s3_verify self.s3_endpoint_url = ( litellm.s3_callback_params.get("s3_endpoint_url") or s3_endpoint_url ) @@ -279,6 +286,9 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): verbose_logger.debug( f"s3_v2 logger - uploading data to s3 - {batch_logging_element.s3_object_key}" ) + verbose_logger.debug( + f"s3_v2 logger - s3_verify setting: {self.s3_verify}" + ) # Prepare the URL url = f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{batch_logging_element.s3_object_key}" @@ -477,7 +487,9 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM): # Prepare the signed headers signed_headers = dict(aws_request.headers.items()) - httpx_client = _get_httpx_client() + httpx_client = _get_httpx_client( + params={"ssl_verify": self.s3_verify} if self.s3_verify is not None else None + ) # Make the request response = httpx_client.put(url, data=json_string, headers=signed_headers) response.raise_for_status() diff --git a/litellm/llms/vertex_ai/vertex_ai_partner_models/count_tokens/handler.py b/litellm/llms/vertex_ai/vertex_ai_partner_models/count_tokens/handler.py index 0f073d02269..da76b12c371 100644 --- a/litellm/llms/vertex_ai/vertex_ai_partner_models/count_tokens/handler.py +++ b/litellm/llms/vertex_ai/vertex_ai_partner_models/count_tokens/handler.py @@ -9,7 +9,6 @@ from typing import Any, Dict, Optional from litellm.llms.custom_httpx.http_handler import get_async_httpx_client from litellm.llms.vertex_ai.vertex_llm_base import VertexBase -from litellm.types.llms.vertex_ai import VertexPartnerProvider class VertexAIPartnerModelsTokenCounter(VertexBase): diff --git a/litellm/proxy/proxy_config.yaml b/litellm/proxy/proxy_config.yaml index 08f014dd275..40229f4c5d4 100644 --- a/litellm/proxy/proxy_config.yaml +++ b/litellm/proxy/proxy_config.yaml @@ -23,12 +23,20 @@ search_tools: api_key: os.environ/EXA_API_KEY - -# for /files endpoints -files_settings: - - custom_llm_provider: openai - api_key: os.environ/OPENAI_API_KEY - - litellm_settings: - callbacks: ["datadog"] \ No newline at end of file + # Comprehensive logging settings + store_audit_logs: true + verbose: true + log_level: "DEBUG" # Options: DEBUG, INFO, WARNING, ERROR + success_callback: ["s3_v2"] + s3_callback_params: + s3_endpoint_url: "https://localhost:443" # Replace with your Minio server URL and port + s3_aws_access_key_id: "minioadmin" + s3_aws_secret_access_key: "minioadmin" + s3_region_name: "minio" # This can be any value for Minio + s3_bucket_name: "litellm-test" # Replace with your bucket name + s3_use_ssl: False + s3_verify: False + cache: True + cache_params: + type: local \ No newline at end of file diff --git a/tests/test_litellm/integrations/test_s3_v2.py b/tests/test_litellm/integrations/test_s3_v2.py index 12ec8af1ef4..417153e24cc 100644 --- a/tests/test_litellm/integrations/test_s3_v2.py +++ b/tests/test_litellm/integrations/test_s3_v2.py @@ -30,6 +30,7 @@ class TestS3V2UnitTests: def test_s3_v2_endpoint_url(self, mock_periodic_flush, mock_create_task): """testing s3 endpoint url""" from unittest.mock import AsyncMock, MagicMock + from litellm.types.integrations.s3_v2 import s3BatchLoggingElement # Mock periodic_flush and create_task to prevent async task creation during init @@ -266,6 +267,187 @@ async def test_strip_base64_mixed_nested_objects(): assert stripped["messages"][0]["extra"]["trace_id"] == "123" +@pytest.mark.asyncio +async def test_s3_verify_false_handling(): + """ + Test that s3_verify=False is properly handled and not treated as None. + + This is a regression test for the bug where s3_verify=False was being + ignored because 'False or s3_verify' would evaluate to s3_verify (None). + """ + from unittest.mock import AsyncMock, patch + + import litellm + + # Set up s3_callback_params with s3_verify=False + litellm.s3_callback_params = { + "s3_bucket_name": "test-bucket", + "s3_endpoint_url": "https://localhost:443", + "s3_aws_access_key_id": "minioadmin", + "s3_aws_secret_access_key": "minioadmin", + "s3_region_name": "us-east-1", + "s3_verify": False, # This should NOT be ignored + "s3_use_ssl": False, # This should also NOT be ignored + } + + with patch('asyncio.create_task'): + with patch('litellm.integrations.s3_v2.get_async_httpx_client') as mock_get_client: + mock_client = AsyncMock() + mock_get_client.return_value = mock_client + + # Create logger + logger = S3Logger() + + # Verify s3_verify is False, not None + assert logger.s3_verify is False, f"Expected s3_verify=False, got {logger.s3_verify}" + assert logger.s3_use_ssl is False, f"Expected s3_use_ssl=False, got {logger.s3_use_ssl}" + + # Verify that get_async_httpx_client was called with ssl_verify=False + mock_get_client.assert_called_once() + call_kwargs = mock_get_client.call_args.kwargs + assert 'params' in call_kwargs, "params should be passed to get_async_httpx_client" + assert call_kwargs['params'] == {'ssl_verify': False}, f"Expected ssl_verify=False in params, got {call_kwargs.get('params')}" + + # Clean up + litellm.s3_callback_params = None + + +@pytest.mark.asyncio +async def test_s3_verify_none_handling(): + """ + Test that s3_verify=None uses default behavior. + """ + from unittest.mock import AsyncMock, patch + + import litellm + + # Set up s3_callback_params without s3_verify + litellm.s3_callback_params = { + "s3_bucket_name": "test-bucket", + "s3_aws_access_key_id": "test-key", + "s3_aws_secret_access_key": "test-secret", + "s3_region_name": "us-east-1", + } + + with patch('asyncio.create_task'): + with patch('litellm.integrations.s3_v2.get_async_httpx_client') as mock_get_client: + mock_client = AsyncMock() + mock_get_client.return_value = mock_client + + # Create logger without explicit s3_verify + logger = S3Logger() + + # Verify s3_verify is None (default) + assert logger.s3_verify is None, f"Expected s3_verify=None, got {logger.s3_verify}" + + # Verify that get_async_httpx_client was called + mock_get_client.assert_called_once() + call_kwargs = mock_get_client.call_args.kwargs + # When s3_verify is None, params={'ssl_verify': None} which is fine - uses default behavior + # The important thing is it's not False + if 'params' in call_kwargs and call_kwargs['params'] is not None: + assert call_kwargs['params'].get('ssl_verify') is None + # Either params is None or params={'ssl_verify': None} is acceptable + + # Clean up + litellm.s3_callback_params = None + + +@pytest.mark.asyncio +async def test_s3_verify_false_creates_httpx_client_with_verify_false(): + """ + Test that when s3_verify=False, the actual httpx client has verify=False. + + This validates that ssl_verify=False flows through to the httpx.AsyncClient. + """ + from unittest.mock import patch + + import litellm + + # Set up s3_callback_params with s3_verify=False + litellm.s3_callback_params = { + "s3_bucket_name": "test-bucket", + "s3_endpoint_url": "https://localhost:443", + "s3_aws_access_key_id": "minioadmin", + "s3_aws_secret_access_key": "minioadmin", + "s3_region_name": "us-east-1", + "s3_verify": False, + } + + with patch('asyncio.create_task'): + # Create logger - this creates the httpx client + logger = S3Logger() + + # Verify the logger has s3_verify=False + assert logger.s3_verify is False + + # Check the actual httpx client has verify=False + # The async_httpx_client.client is the actual httpx.AsyncClient + if hasattr(logger.async_httpx_client, 'client'): + httpx_client = logger.async_httpx_client.client + # Check the _verify attribute (httpx internal) + if hasattr(httpx_client, '_verify'): + assert httpx_client._verify is False, f"Expected httpx client _verify=False, got {httpx_client._verify}" + + # Clean up + litellm.s3_callback_params = None + + +@pytest.mark.asyncio +async def test_s3_verify_false_async_client(): + """ + Test that the async httpx client respects s3_verify=False. + """ + from unittest.mock import AsyncMock, MagicMock, patch + + import litellm + from litellm.types.integrations.s3_v2 import s3BatchLoggingElement + + # Set up s3_callback_params with s3_verify=False + litellm.s3_callback_params = { + "s3_bucket_name": "test-bucket", + "s3_endpoint_url": "https://localhost:443", + "s3_aws_access_key_id": "minioadmin", + "s3_aws_secret_access_key": "minioadmin", + "s3_region_name": "us-east-1", + "s3_verify": False, + } + + with patch('asyncio.create_task'): + logger = S3Logger() + + # Verify s3_verify is False + assert logger.s3_verify is False + + # Create test element + test_element = s3BatchLoggingElement( + s3_object_key="2025-11-03/test-key.json", + payload={"test": "data"}, + s3_object_download_filename="test-file.json" + ) + + # Mock the async httpx client's put method + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.raise_for_status = MagicMock() + logger.async_httpx_client.put = AsyncMock(return_value=mock_response) + + # Call async upload + await logger.async_upload_data_to_s3(test_element) + + # Verify put was called + assert logger.async_httpx_client.put.called + + # Check that the async httpx client was created with verify=False + if hasattr(logger.async_httpx_client, 'client'): + httpx_client = logger.async_httpx_client.client + if hasattr(httpx_client, '_verify'): + assert httpx_client._verify is False, f"Expected async httpx client _verify=False, got {httpx_client._verify}" + + # Clean up + litellm.s3_callback_params = None + + @pytest.mark.asyncio async def test_strip_base64_recursive_redaction(): logger = S3Logger(s3_strip_base64_files=True)