From 3e9ef4362870443bc6bea969e1ca4ba17be6a532 Mon Sep 17 00:00:00 2001 From: Tiago Kochenborger Date: Tue, 4 Aug 2026 16:31:10 -0300 Subject: [PATCH] feat(aicore): clear AICORE_CLIENT_SECRET after first successful token acquisition After the first successful litellm.completion() call, AICORE_CLIENT_SECRET is removed from os.environ. LiteLLM has captured the secret inside its token creator closure at that point and no longer reads from the environment. Removing it minimises the exposure window to child processes and container introspection APIs (AFSDK-4291 / HASI2026203 SEC-309). The flag is reset when credentials are reloaded (credential rotation flow) so the secret is cleared again after the retry succeeds. No-op in transparent TLS mode where the secret was never written. Relates-to: AFSDK-4291 --- src/sap_cloud_sdk/aicore/completion.py | 79 +++++++++++++---- tests/aicore/unit/test_completion.py | 117 ++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 17 deletions(-) diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index f74bf7e1..a389ebf6 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/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: diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 5ea4aaf6..30c89005 100644 --- a/tests/aicore/unit/test_completion.py +++ b/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