feat(aicore): clear AICORE_CLIENT_SECRET after token acquisition (AFSDK-4291)

src/sap_cloud_sdk/aicore/completion.py

@@ -14,15 +14,20 @@
 re-raising as :class:`ContentFilteredError` so callers can rely on a
 single exception type for "filter blocked you."
 
-Credential rotation handling
-----------------------------
-When a credential (client_secret or mTLS certificate) is rotated while the
-pod is running, LiteLLM's cached token becomes invalid and the next token
-refresh attempt raises ``litellm.AuthenticationError``. The wrappers
-intercept this error, reload credentials from the mounted secret volume via
-:func:`reload_aicore_credentials`, and retry the call once. The caller is
-unaffected — rotation is transparent. If the retry also fails, the
-``AuthenticationError`` propagates normally.
+Credential handling
+-------------------
+CLIENT_SECRET minimisation (AFSDK-4291):
+After the first successful LiteLLM call, ``AICORE_CLIENT_SECRET`` is removed
+from ``os.environ``. LiteLLM has already captured the secret inside its token
+creator closure at that point and no longer needs the env var. This minimises
+the window of exposure to child processes and container introspection.
+
+Credential rotation (reactive reload):
+When a credential is rotated while the pod is running, LiteLLM's cached token
+becomes invalid and the next token refresh raises ``litellm.AuthenticationError``.
+The wrappers intercept this, reload credentials from the mounted secret volume
+via :func:`reload_aicore_credentials`, and retry once. The secret is cleared
+again after the retry succeeds.
 
 Usage::
 
@@ -50,6 +55,8 @@
 from __future__ import annotations
 
 import logging
+import os
+import threading
 from typing import Any
 
 import litellm
@@ -58,6 +65,36 @@
 
 logger = logging.getLogger(__name__)
 
+# Tracks whether AICORE_CLIENT_SECRET has already been cleared after the first
+# successful LiteLLM call. Reset when credentials are reloaded so the secret
+# is cleared again after the retry succeeds.
+_secret_lock = threading.Lock()
+_secret_cleared = False
+
+
+def _clear_client_secret() -> None:
+    """Remove AICORE_CLIENT_SECRET from env after LiteLLM has cached the token.
+
+    Safe to call multiple times — subsequent calls are no-ops once cleared.
+    No-op in transparent TLS mode (secret was never written).
+    """
+    global _secret_cleared
+    with _secret_lock:
+        if not _secret_cleared:
+            if os.environ.pop("AICORE_CLIENT_SECRET", None) is not None:
+                logger.info(
+                    "AICORE_CLIENT_SECRET cleared from environment "
+                    "after token acquisition (AFSDK-4291)"
+                )
+            _secret_cleared = True
+
+
+def _reset_secret_cleared() -> None:
+    """Allow _clear_client_secret() to fire again after a credential reload."""
+    global _secret_cleared
+    with _secret_lock:
+        _secret_cleared = False
+
 
 def reload_aicore_credentials() -> None:
     """Re-read AI Core credentials from the mounted secret volume.
@@ -71,6 +108,7 @@ def reload_aicore_credentials() -> None:
     """
     # Import here to avoid a circular import: completion ← __init__ ← completion
     from sap_cloud_sdk.aicore import set_aicore_config
+    _reset_secret_cleared()
     logger.info("AI Core credentials reloading after authentication failure")
     set_aicore_config()
 
@@ -92,15 +130,22 @@ def completion(*args: Any, **kwargs: Any) -> Any:
     """Wrapper around :func:`litellm.completion` that normalises filter errors
     and handles credential rotation transparently.
 
+    After the first successful call, ``AICORE_CLIENT_SECRET`` is removed from
+    ``os.environ`` — LiteLLM has captured it in its token creator closure and
+    no longer needs the env var (AFSDK-4291).
+
     On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert),
     reloads credentials from the mounted secret volume and retries once.
-    All other exceptions surface verbatim after the filter-error translation.
     """
     try:
-        return litellm.completion(*args, **kwargs)
+        result = litellm.completion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except litellm.AuthenticationError:
         reload_aicore_credentials()
-        return litellm.completion(*args, **kwargs)
+        result = litellm.completion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:
@@ -111,13 +156,17 @@ def completion(*args: Any, **kwargs: Any) -> Any:
 async def acompletion(*args: Any, **kwargs: Any) -> Any:
     """Async wrapper around :func:`litellm.acompletion`.
 
-    Same translation and credential-rotation semantics as :func:`completion`.
+    Same credential-minimisation and rotation semantics as :func:`completion`.
     """
     try:
-        return await litellm.acompletion(*args, **kwargs)
+        result = await litellm.acompletion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except litellm.AuthenticationError:
         reload_aicore_credentials()
-        return await litellm.acompletion(*args, **kwargs)
+        result = await litellm.acompletion(*args, **kwargs)
+        _clear_client_secret()
+        return result
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
         if translated is exc:

tests/aicore/unit/test_completion.py

@@ -23,13 +23,18 @@
 
 import asyncio
 import json
-from unittest.mock import MagicMock, call, patch
+import os
+from unittest.mock import MagicMock, patch
 
 import litellm
 import pytest
 
 from sap_cloud_sdk.aicore import acompletion, completion
-from sap_cloud_sdk.aicore.completion import reload_aicore_credentials
+from sap_cloud_sdk.aicore.completion import (
+    reload_aicore_credentials,
+    _clear_client_secret,
+    _reset_secret_cleared,
+)
 from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
 
 
@@ -214,6 +219,114 @@ def test_calls_set_aicore_config(self):
             reload_aicore_credentials()
         mock_config.assert_called_once_with()
 
+    def test_resets_secret_cleared_flag(self):
+        """After reload, _clear_client_secret() must be able to clear the secret again."""
+        # Simulate: secret was cleared once already
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "old"}):
+            _clear_client_secret()
+        # Flag is now True — a second clear would be a no-op
+        with patch("sap_cloud_sdk.aicore.set_aicore_config"):
+            reload_aicore_credentials()
+        # After reload the flag is reset — clearing works again
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "new"}):
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+
+# ---------------------------------------------------------------------------
+# CLIENT_SECRET minimisation — _clear_client_secret()
+# ---------------------------------------------------------------------------
+
+
+class TestClearClientSecret:
+    def setup_method(self):
+        _reset_secret_cleared()
+
+    def test_removes_secret_from_env(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_noop_when_secret_absent(self):
+        env = {}
+        with patch.dict("os.environ", env, clear=True):
+            _clear_client_secret()  # must not raise
+
+    def test_idempotent_second_call_is_noop(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            os.environ["AICORE_CLIENT_SECRET"] = "restored"
+            _clear_client_secret()
+            # second call must not remove the restored value
+            assert os.environ.get("AICORE_CLIENT_SECRET") == "restored"
+
+    def test_reset_allows_clear_again(self):
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            _clear_client_secret()
+            _reset_secret_cleared()
+            os.environ["AICORE_CLIENT_SECRET"] = "new-secret"
+            _clear_client_secret()
+            assert "AICORE_CLIENT_SECRET" not in os.environ
+
+
+# ---------------------------------------------------------------------------
+# completion() clears secret on success
+# ---------------------------------------------------------------------------
+
+
+class TestCompletionClearsSecret:
+    def setup_method(self):
+        _reset_secret_cleared()
+
+    def test_secret_cleared_after_successful_call(self):
+        sentinel = object()
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", return_value=sentinel),
+            patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}),
+        ):
+            result = completion(model="sap/x", messages=[])
+        assert result is sentinel
+        assert "AICORE_CLIENT_SECRET" not in os.environ
+
+    def test_secret_not_cleared_on_filter_error(self):
+        """Filter errors are not successful calls — secret stays until next success."""
+        from sap_cloud_sdk.aicore.filtering.exceptions import ContentFilteredError
+        raised = ContentFilteredError(direction="input", details={}, request_id="r")
+        secret_present_after = {}
+        with patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}):
+            with pytest.raises(ContentFilteredError):
+                with patch(
+                    "sap_cloud_sdk.aicore.completion.litellm.completion",
+                    side_effect=raised,
+                ):
+                    completion(model="sap/x", messages=[])
+            secret_present_after["value"] = os.environ.get("AICORE_CLIENT_SECRET")
+        assert secret_present_after["value"] == "s3cr3t"
+
+    def test_secret_cleared_after_auth_error_and_retry(self):
+        """After reload + successful retry, secret must be cleared."""
+        sentinel = object()
+        auth_err = litellm.AuthenticationError(
+            message="401", llm_provider="sap", model="sap/x"
+        )
+        call_returns = [auth_err, sentinel]
+
+        def fake_completion(*args, **kwargs):
+            result = call_returns.pop(0)
+            if isinstance(result, Exception):
+                raise result
+            return result
+
+        with (
+            patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=fake_completion),
+            patch("sap_cloud_sdk.aicore.set_aicore_config"),
+            patch.dict("os.environ", {"AICORE_CLIENT_SECRET": "s3cr3t"}),
+        ):
+            result = completion(model="sap/x", messages=[])
+
+        assert result is sentinel
+        assert "AICORE_CLIENT_SECRET" not in os.environ
+
 
 # ---------------------------------------------------------------------------
 # Reactive reload on AuthenticationError — sync