Merge pull request #27222 from BerriAI/litellm_s3AuditParams

[Feat] Decouple S3 audit-log config via s3_audit_callback_params

litellm/__init__.py

@@ -388,6 +388,7 @@ anthropic_beta_headers_url: str = os.getenv(
 suppress_debug_info = False
 dynamodb_table_name: Optional[str] = None
 s3_callback_params: Optional[Dict] = None
+s3_audit_callback_params: Optional[Dict] = None
 datadog_llm_observability_params: Optional[Union[DatadogLLMObsInitParams, Dict]] = None
 datadog_params: Optional[Union[DatadogInitParams, Dict]] = None
 aws_sqs_callback_params: Optional[Dict] = None

litellm/integrations/s3_v2.py

@@ -16,6 +16,7 @@ from litellm._logging import print_verbose, verbose_logger
 from litellm.constants import DEFAULT_S3_BATCH_SIZE, DEFAULT_S3_FLUSH_INTERVAL_SECONDS
 from litellm.integrations.s3 import get_s3_object_key
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
+from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
 from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
 from litellm.llms.custom_httpx.http_handler import (
     _get_httpx_client,
@@ -53,15 +54,25 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
         s3_strip_base64_files: bool = False,
         s3_use_key_prefix: bool = False,
         s3_use_virtual_hosted_style: bool = False,
+        s3_callback_params_override: Optional[dict] = None,
         **kwargs,
     ):
         try:
-            verbose_logger.debug(
-                f"in init s3 logger - s3_callback_params {litellm.s3_callback_params}"
-            )
+            _masker = SensitiveDataMasker()
+            if s3_callback_params_override is not None:
+                verbose_logger.debug(
+                    f"in init s3 logger (audit override) - "
+                    f"{_masker.mask_dict(dict(s3_callback_params_override))}"
+                )
+            else:
+                verbose_logger.debug(
+                    f"in init s3 logger - s3_callback_params "
+                    f"{_masker.mask_dict(dict(litellm.s3_callback_params or {}))}"
+                )
 
             # Initialize S3 params first to get the correct s3_verify value
             self._init_s3_params(
+                params_source=s3_callback_params_override,
                 s3_bucket_name=s3_bucket_name,
                 s3_region_name=s3_region_name,
                 s3_api_version=s3_api_version,
@@ -139,94 +150,85 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
         s3_strip_base64_files: bool = False,
         s3_use_key_prefix: bool = False,
         s3_use_virtual_hosted_style: bool = False,
+        params_source: Optional[dict] = None,
     ):
         """
-        Initialize the s3 params for this logging callback
+        Initialize the s3 params for this logging callback. Reads from
+        `params_source` if given (e.g. `s3_audit_callback_params` for the
+        audit-log instance), otherwise falls back to `litellm.s3_callback_params`.
+        Resolves `os.environ/X` markers into a local dict; never mutates the source.
         """
-        litellm.s3_callback_params = litellm.s3_callback_params or {}
-        # read in .env variables - example os.environ/AWS_BUCKET_NAME
-        for key, value in litellm.s3_callback_params.items():
-            if isinstance(value, str) and value.startswith("os.environ/"):
-                litellm.s3_callback_params[key] = litellm.get_secret(value)
+        if params_source is None:
+            params_source = litellm.s3_callback_params or {}
+        params: dict = {
+            key: (
+                litellm.get_secret(value)
+                if isinstance(value, str) and value.startswith("os.environ/")
+                else value
+            )
+            for key, value in params_source.items()
+        }
 
-        self.s3_bucket_name = (
-            litellm.s3_callback_params.get("s3_bucket_name") or s3_bucket_name
-        )
-        self.s3_region_name = (
-            litellm.s3_callback_params.get("s3_region_name") or s3_region_name
-        )
-        self.s3_api_version = (
-            litellm.s3_callback_params.get("s3_api_version") or s3_api_version
-        )
+        self.s3_bucket_name = params.get("s3_bucket_name") or s3_bucket_name
+        self.s3_region_name = params.get("s3_region_name") or s3_region_name
+        self.s3_api_version = params.get("s3_api_version") or s3_api_version
         self.s3_use_ssl = (
-            litellm.s3_callback_params.get("s3_use_ssl", True)
-            if litellm.s3_callback_params.get("s3_use_ssl") is not None
+            params.get("s3_use_ssl", True)
+            if 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
+            params.get("s3_verify")
+            if params.get("s3_verify") is not None
             else s3_verify
         )
-        self.s3_endpoint_url = (
-            litellm.s3_callback_params.get("s3_endpoint_url") or s3_endpoint_url
-        )
+        self.s3_endpoint_url = params.get("s3_endpoint_url") or s3_endpoint_url
         self.s3_aws_access_key_id = (
-            litellm.s3_callback_params.get("s3_aws_access_key_id")
-            or s3_aws_access_key_id
+            params.get("s3_aws_access_key_id") or s3_aws_access_key_id
         )
 
         self.s3_aws_secret_access_key = (
-            litellm.s3_callback_params.get("s3_aws_secret_access_key")
-            or s3_aws_secret_access_key
+            params.get("s3_aws_secret_access_key") or s3_aws_secret_access_key
         )
 
         self.s3_aws_session_token = (
-            litellm.s3_callback_params.get("s3_aws_session_token")
-            or s3_aws_session_token
+            params.get("s3_aws_session_token") or s3_aws_session_token
         )
 
         self.s3_aws_session_name = (
-            litellm.s3_callback_params.get("s3_aws_session_name") or s3_aws_session_name
+            params.get("s3_aws_session_name") or s3_aws_session_name
         )
 
         self.s3_aws_profile_name = (
-            litellm.s3_callback_params.get("s3_aws_profile_name") or s3_aws_profile_name
+            params.get("s3_aws_profile_name") or s3_aws_profile_name
         )
 
-        self.s3_aws_role_name = (
-            litellm.s3_callback_params.get("s3_aws_role_name") or s3_aws_role_name
-        )
+        self.s3_aws_role_name = params.get("s3_aws_role_name") or s3_aws_role_name
 
         self.s3_aws_web_identity_token = (
-            litellm.s3_callback_params.get("s3_aws_web_identity_token")
-            or s3_aws_web_identity_token
+            params.get("s3_aws_web_identity_token") or s3_aws_web_identity_token
         )
 
         self.s3_aws_sts_endpoint = (
-            litellm.s3_callback_params.get("s3_aws_sts_endpoint") or s3_aws_sts_endpoint
+            params.get("s3_aws_sts_endpoint") or s3_aws_sts_endpoint
         )
 
-        self.s3_config = litellm.s3_callback_params.get("s3_config") or s3_config
-        self.s3_path = litellm.s3_callback_params.get("s3_path") or s3_path
-        # done reading litellm.s3_callback_params
+        self.s3_config = params.get("s3_config") or s3_config
+        self.s3_path = params.get("s3_path") or s3_path
         self.s3_use_team_prefix = (
-            bool(litellm.s3_callback_params.get("s3_use_team_prefix", False))
-            or s3_use_team_prefix
+            bool(params.get("s3_use_team_prefix", False)) or s3_use_team_prefix
         )
 
         self.s3_use_key_prefix = (
-            bool(litellm.s3_callback_params.get("s3_use_key_prefix", False))
-            or s3_use_key_prefix
+            bool(params.get("s3_use_key_prefix", False)) or s3_use_key_prefix
         )
 
         self.s3_strip_base64_files = (
-            bool(litellm.s3_callback_params.get("s3_strip_base64_files", False))
-            or s3_strip_base64_files
+            bool(params.get("s3_strip_base64_files", False)) or s3_strip_base64_files
         )
 
         self.s3_use_virtual_hosted_style = (
-            bool(litellm.s3_callback_params.get("s3_use_virtual_hosted_style", False))
+            bool(params.get("s3_use_virtual_hosted_style", False))
             or s3_use_virtual_hosted_style
         )
 

litellm/proxy/management_helpers/audit_logs.py

@@ -46,25 +46,46 @@ def get_audit_log_changed_by(
 
 
 def _resolve_audit_log_callback(name: str) -> Optional[CustomLogger]:
-    """Resolve a string callback name to a CustomLogger instance, with caching."""
+    """Resolve a string callback name to a CustomLogger instance, with caching.
+
+    For "s3_v2" with `litellm.s3_audit_callback_params` set, constructs a
+    dedicated `S3Logger` so audit logs can target a different bucket than the
+    normal-log singleton served by `_init_custom_logger_compatible_class`.
+    """
     if name in _audit_log_callback_cache:
         return _audit_log_callback_cache[name]
 
-    from litellm.litellm_core_utils.litellm_logging import (
-        _init_custom_logger_compatible_class,
-    )
+    instance: Optional[CustomLogger]
+    if (
+        name == "s3_v2"
+        and getattr(litellm, "s3_audit_callback_params", None) is not None
+    ):
+        from litellm.integrations.s3_v2 import S3Logger as S3V2Logger
 
-    instance = _init_custom_logger_compatible_class(
-        logging_integration=name,  # type: ignore
-        internal_usage_cache=None,
-        llm_router=None,
-    )
+        instance = S3V2Logger(
+            s3_callback_params_override=litellm.s3_audit_callback_params
+        )
+    else:
+        from litellm.litellm_core_utils.litellm_logging import (
+            _init_custom_logger_compatible_class,
+        )
+
+        instance = _init_custom_logger_compatible_class(
+            logging_integration=name,  # type: ignore
+            internal_usage_cache=None,
+            llm_router=None,
+        )
 
     if instance is not None:
         _audit_log_callback_cache[name] = instance
     return instance
 
 
+def reset_audit_log_callback_cache() -> None:
+    """Clear cached audit-log callback instances. Call on config reload."""
+    _audit_log_callback_cache.clear()
+
+
 def _build_audit_log_payload(
     request_data: LiteLLM_AuditLogs,
 ) -> StandardAuditLogPayload:

litellm/proxy/proxy_server.py

@@ -3823,6 +3823,11 @@ class ProxyConfig:
                         f"{blue_color_code} Initialized Failure Callbacks - {litellm.failure_callback} {reset_color_code}"
                     )  # noqa
                 elif key == "audit_log_callbacks":
+                    from litellm.proxy.management_helpers.audit_logs import (
+                        reset_audit_log_callback_cache,
+                    )
+
+                    reset_audit_log_callback_cache()
                     litellm.audit_log_callbacks = []
 
                     for callback in value:
@@ -3904,6 +3909,21 @@ class ProxyConfig:
                         f"{blue_color_code} setting litellm.{key}={value}{reset_color_code}"
                     )
                     setattr(litellm, key, value)
+                    if key in {"s3_audit_callback_params", "s3_callback_params"}:
+                        from litellm.proxy.management_helpers.audit_logs import (
+                            reset_audit_log_callback_cache,
+                        )
+                        from litellm.litellm_core_utils.litellm_logging import (
+                            _in_memory_loggers,
+                        )
+                        from litellm.integrations.s3_v2 import S3Logger as S3V2Logger
+
+                        reset_audit_log_callback_cache()
+                        _in_memory_loggers[:] = [
+                            cb
+                            for cb in _in_memory_loggers
+                            if not isinstance(cb, S3V2Logger)
+                        ]
 
         ## GENERAL SERVER SETTINGS (e.g. master key,..) # do this after initializing litellm, to ensure sentry logging works for proxylogging
         general_settings = config.get("general_settings", {})

tests/logging_callback_tests/conftest.py

@@ -101,6 +101,7 @@ _SCALAR_ATTRS = (
     "redact_messages_in_exceptions",
     "redact_user_api_key_info",
     "s3_callback_params",
+    "s3_audit_callback_params",
     "datadog_params",
     "vector_store_registry",
 )
@@ -128,6 +129,7 @@ def isolate_litellm_state():
     leaking across tests within the same xdist worker.
     """
     from litellm.litellm_core_utils import litellm_logging as ll_logging
+    from litellm.proxy.management_helpers import audit_logs as ll_audit_logs
 
     # Flush cache and clear internal logger instances before test
     if hasattr(litellm, "in_memory_llm_clients_cache"):
@@ -135,6 +137,7 @@ def isolate_litellm_state():
 
     # Clear cached logger instances (LangsmithLogger, SlackAlerting, etc.)
     ll_logging._in_memory_loggers.clear()
+    ll_audit_logs._audit_log_callback_cache.clear()
 
     # Reset ALL attrs to their true defaults before the test runs.
     # This undoes any module-level mutations from test file imports.
@@ -156,6 +159,7 @@ def isolate_litellm_state():
         litellm.in_memory_llm_clients_cache.flush_cache()
 
     ll_logging._in_memory_loggers.clear()
+    ll_audit_logs._audit_log_callback_cache.clear()
 
     for attr in _LIST_ATTRS:
         if attr in _DEFAULTS:

tests/test_litellm/integrations/test_s3_v2.py

@@ -1123,3 +1123,74 @@ async def test_combined_prefix_reflects_in_s3_object_key():
     result = logger.create_s3_batch_logging_element(datetime.utcnow(), payload)
     key = result.s3_object_key
     assert "myteam/apikey/" in key, f"Expected both prefixes in key: {key}"
+
+
+# --------------------------------------------------------------
+# params_source / s3_callback_params_override (audit-log decoupling)
+# --------------------------------------------------------------
+def test_s3_callback_params_override_uses_alternate_dict():
+    """`s3_callback_params_override` makes the logger read its config from
+    the override dict instead of `litellm.s3_callback_params`."""
+    import litellm
+
+    original = litellm.s3_callback_params
+    litellm.s3_callback_params = {"s3_bucket_name": "normal-bucket"}
+    try:
+        logger = S3Logger(
+            s3_callback_params_override={
+                "s3_bucket_name": "audit-bucket",
+                "s3_path": "audit-prefix",
+                "s3_region_name": "us-west-2",
+            }
+        )
+        assert logger.s3_bucket_name == "audit-bucket"
+        assert logger.s3_path == "audit-prefix"
+        assert logger.s3_region_name == "us-west-2"
+    finally:
+        litellm.s3_callback_params = original
+
+
+def test_s3_callback_params_override_does_not_mutate_inputs(monkeypatch):
+    """Resolving `os.environ/X` markers must not mutate the override dict
+    or `litellm.s3_callback_params`."""
+    import litellm
+
+    monkeypatch.setenv("MY_AUDIT_BUCKET", "resolved-bucket")
+    override = {"s3_bucket_name": "os.environ/MY_AUDIT_BUCKET"}
+    original_global = litellm.s3_callback_params
+    litellm.s3_callback_params = {"s3_bucket_name": "os.environ/MY_AUDIT_BUCKET"}
+    try:
+        logger = S3Logger(s3_callback_params_override=override)
+        assert logger.s3_bucket_name == "resolved-bucket"
+        assert override["s3_bucket_name"] == "os.environ/MY_AUDIT_BUCKET"
+        assert (
+            litellm.s3_callback_params["s3_bucket_name"] == "os.environ/MY_AUDIT_BUCKET"
+        )
+    finally:
+        litellm.s3_callback_params = original_global
+
+
+def test_s3_callback_params_override_none_falls_back_to_global():
+    """No override → behaves exactly as today (reads `litellm.s3_callback_params`)."""
+    import litellm
+
+    original = litellm.s3_callback_params
+    litellm.s3_callback_params = {"s3_bucket_name": "from-global"}
+    try:
+        logger = S3Logger()
+        assert logger.s3_bucket_name == "from-global"
+    finally:
+        litellm.s3_callback_params = original
+
+
+def test_s3_callback_params_override_empty_dict_is_opt_in():
+    """An empty override dict skips the global entirely (env/IAM-only config)."""
+    import litellm
+
+    original = litellm.s3_callback_params
+    litellm.s3_callback_params = {"s3_bucket_name": "from-global"}
+    try:
+        logger = S3Logger(s3_callback_params_override={})
+        assert logger.s3_bucket_name is None
+    finally:
+        litellm.s3_callback_params = original

tests/test_litellm/proxy/management_helpers/test_audit_log_callbacks.py

@@ -346,3 +346,126 @@ class TestS3LoggerAuditLogEvent:
             element = logger.log_queue[0]
             assert element.s3_object_key.startswith("audit_logs/")
             assert "audit-456" in element.s3_object_key
+
+
+class TestS3AuditCallbackParamsDecoupling:
+    """`s3_audit_callback_params` should give the audit-log path its own
+    S3Logger instance, distinct from the singleton serving normal logs."""
+
+    @pytest.fixture(autouse=True)
+    def _isolate_caches_and_globals(self):
+        from litellm.litellm_core_utils import litellm_logging as ll_logging
+        from litellm.proxy.management_helpers import audit_logs as ll_audit_logs
+
+        original_s3 = litellm.s3_callback_params
+        original_audit = getattr(litellm, "s3_audit_callback_params", None)
+        ll_audit_logs._audit_log_callback_cache.clear()
+        ll_logging._in_memory_loggers.clear()
+        yield
+        litellm.s3_callback_params = original_s3
+        litellm.s3_audit_callback_params = original_audit
+        ll_audit_logs._audit_log_callback_cache.clear()
+        ll_logging._in_memory_loggers.clear()
+
+    def test_opt_in_constructs_separate_instance_with_audit_config(self):
+        """Audit config set → audit resolver returns a fresh S3Logger pointing
+        at the audit bucket, distinct from the normal-log singleton."""
+        from litellm.integrations.s3_v2 import S3Logger
+        from litellm.litellm_core_utils.litellm_logging import (
+            _init_custom_logger_compatible_class,
+        )
+        from litellm.proxy.management_helpers.audit_logs import (
+            _resolve_audit_log_callback,
+        )
+
+        litellm.s3_callback_params = {"s3_bucket_name": "normal-bucket"}
+        litellm.s3_audit_callback_params = {"s3_bucket_name": "audit-bucket"}
+
+        with patch("asyncio.create_task"):
+            audit_instance = _resolve_audit_log_callback("s3_v2")
+            normal_instance = _init_custom_logger_compatible_class(
+                logging_integration="s3_v2",
+                internal_usage_cache=None,
+                llm_router=None,
+            )
+
+        assert isinstance(audit_instance, S3Logger)
+        assert isinstance(normal_instance, S3Logger)
+        assert id(audit_instance) != id(normal_instance)
+        assert audit_instance.s3_bucket_name == "audit-bucket"
+        assert normal_instance.s3_bucket_name == "normal-bucket"
+
+    def test_opt_out_preserves_singleton_behavior(self):
+        """No `s3_audit_callback_params` → audit and normal share the singleton
+        (existing behavior, regression guard)."""
+        from litellm.integrations.s3_v2 import S3Logger
+        from litellm.litellm_core_utils.litellm_logging import (
+            _init_custom_logger_compatible_class,
+        )
+        from litellm.proxy.management_helpers.audit_logs import (
+            _resolve_audit_log_callback,
+        )
+
+        litellm.s3_callback_params = {"s3_bucket_name": "shared-bucket"}
+        litellm.s3_audit_callback_params = None
+
+        with patch("asyncio.create_task"):
+            normal_instance = _init_custom_logger_compatible_class(
+                logging_integration="s3_v2",
+                internal_usage_cache=None,
+                llm_router=None,
+            )
+            audit_instance = _resolve_audit_log_callback("s3_v2")
+
+        assert isinstance(audit_instance, S3Logger)
+        assert id(audit_instance) == id(normal_instance)
+        assert audit_instance.s3_bucket_name == "shared-bucket"
+
+    def test_empty_dict_opts_in(self):
+        """`s3_audit_callback_params = {}` is opt-in (truthy-by-presence) and
+        produces a separate instance with no bucket configured (env/IAM-only)."""
+        from litellm.integrations.s3_v2 import S3Logger
+        from litellm.litellm_core_utils.litellm_logging import (
+            _init_custom_logger_compatible_class,
+        )
+        from litellm.proxy.management_helpers.audit_logs import (
+            _resolve_audit_log_callback,
+        )
+
+        litellm.s3_callback_params = {"s3_bucket_name": "normal-bucket"}
+        litellm.s3_audit_callback_params = {}
+
+        with patch("asyncio.create_task"):
+            audit_instance = _resolve_audit_log_callback("s3_v2")
+            normal_instance = _init_custom_logger_compatible_class(
+                logging_integration="s3_v2",
+                internal_usage_cache=None,
+                llm_router=None,
+            )
+
+        assert id(audit_instance) != id(normal_instance)
+        assert audit_instance.s3_bucket_name is None
+        assert normal_instance.s3_bucket_name == "normal-bucket"
+
+    def test_reset_audit_log_callback_cache_clears_audit_instance(self):
+        """`reset_audit_log_callback_cache()` must drop the cached audit
+        instance so a config reload picks up the new params."""
+        from litellm.proxy.management_helpers.audit_logs import (
+            _audit_log_callback_cache,
+            _resolve_audit_log_callback,
+            reset_audit_log_callback_cache,
+        )
+
+        litellm.s3_audit_callback_params = {"s3_bucket_name": "first"}
+        with patch("asyncio.create_task"):
+            first = _resolve_audit_log_callback("s3_v2")
+            assert first is not None and "s3_v2" in _audit_log_callback_cache
+
+            reset_audit_log_callback_cache()
+            assert "s3_v2" not in _audit_log_callback_cache
+
+            litellm.s3_audit_callback_params = {"s3_bucket_name": "second"}
+            second = _resolve_audit_log_callback("s3_v2")
+            assert second is not None
+            assert id(second) != id(first)
+            assert second.s3_bucket_name == "second"