diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index 7fb10094..1b11b658 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -13,7 +13,7 @@ from sap_cloud_sdk.core.telemetry.metrics_decorator import record_metrics from sap_cloud_sdk.core.telemetry.module import Module from sap_cloud_sdk.core.telemetry.operation import Operation -from .completion import acompletion, completion +from .completion import acompletion, completion, reload_aicore_credentials from .filtering import ( AzureContentFilter, ContentFilter, @@ -30,6 +30,16 @@ logger = logging.getLogger(__name__) +# When set, the infrastructure sidecar adds the mTLS certificate transparently. +# The SDK calls the XSUAA token endpoint over plain HTTPS with only client_id. +# No client_secret or certificate material is required in the service binding. +TRANSPARENT_TLS_ENV_VAR = "AICORE_TRANSPARENT_TLS" + + +def _is_transparent_tls() -> bool: + """Return True when transparent TLS proxy mode is active.""" + return os.environ.get(TRANSPARENT_TLS_ENV_VAR, "").strip().lower() in ("1", "true", "yes") + def _get_secret( env_var_name: str, @@ -123,10 +133,15 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: File mappings based on the Kubernetes secret structure: clientid → AICORE_CLIENT_ID - clientsecret → AICORE_CLIENT_SECRET + clientsecret → AICORE_CLIENT_SECRET (skipped in transparent TLS mode) url → AICORE_AUTH_URL serviceurls (JSON with AI_API_URL) → AICORE_BASE_URL + When ``AICORE_TRANSPARENT_TLS=true`` is set, the infrastructure sidecar + adds the mTLS certificate on the SDK's behalf. In this mode the SDK omits + ``AICORE_CLIENT_SECRET`` from the environment — LiteLLM will use plain + HTTPS to the token endpoint and the sidecar will attach the certificate. + After credentials are loaded, content filtering is activated on every ``sap/*`` LiteLLM call at the configured thresholds (default: severity ``MEDIUM`` on all categories + prompt shield enabled). Override via @@ -135,11 +150,10 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: to turn filtering off at runtime, or set ``AICORE_FILTER_ENABLED=false`` to keep it off entirely. """ + transparent_tls = _is_transparent_tls() + # Load secrets client_id = _get_secret("AICORE_CLIENT_ID", "clientid", instance_name=instance_name) - client_secret = _get_secret( - "AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name - ) auth_url = _get_secret("AICORE_AUTH_URL", "url", instance_name=instance_name) base_url = _get_aicore_base_url(instance_name) resource_group = _get_secret( @@ -156,8 +170,6 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: # Set environment variables for LiteLLM if client_id: os.environ["AICORE_CLIENT_ID"] = client_id - if client_secret: - os.environ["AICORE_CLIENT_SECRET"] = client_secret if auth_url: os.environ["AICORE_AUTH_URL"] = auth_url if base_url: @@ -165,6 +177,17 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: if resource_group: os.environ["AICORE_RESOURCE_GROUP"] = resource_group + if transparent_tls: + # Remove any stale client_secret — the sidecar provides the mTLS cert. + os.environ.pop("AICORE_CLIENT_SECRET", None) + logger.info("AI Core transparent TLS mode active — client_secret not required") + else: + client_secret = _get_secret( + "AICORE_CLIENT_SECRET", "clientsecret", instance_name=instance_name + ) + if client_secret: + os.environ["AICORE_CLIENT_SECRET"] = client_secret + # Log configuration completion (excluding sensitive information) logger.info("AI Core configuration has been set successfully") @@ -177,6 +200,7 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: __all__ = [ "set_aicore_config", + "reload_aicore_credentials", "set_filtering", "disable_filtering", "completion", diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index 3c869cfe..f74bf7e1 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -14,6 +14,16 @@ 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. + Usage:: from sap_cloud_sdk.aicore import completion, ContentFilteredError @@ -39,12 +49,31 @@ from __future__ import annotations +import logging from typing import Any import litellm from .filtering.filters import _parse_input_filter_error +logger = logging.getLogger(__name__) + + +def reload_aicore_credentials() -> None: + """Re-read AI Core credentials from the mounted secret volume. + + Called automatically by :func:`completion` and :func:`acompletion` when + LiteLLM raises ``AuthenticationError`` — covers credential rotation + (client_secret or mTLS certificate) without requiring a pod restart. + + Safe to call manually if the application needs to force a reload, e.g. + after a deliberate secret rotation triggered by the operator. + """ + # Import here to avoid a circular import: completion ← __init__ ← completion + from sap_cloud_sdk.aicore import set_aicore_config + logger.info("AI Core credentials reloading after authentication failure") + set_aicore_config() + def _maybe_translate_filter_error(exc: BaseException) -> BaseException: """Return a :class:`ContentFilteredError` if ``exc`` is a wrapped @@ -60,19 +89,18 @@ def _maybe_translate_filter_error(exc: BaseException) -> BaseException: def completion(*args: Any, **kwargs: Any) -> Any: - """Wrapper around :func:`litellm.completion` that normalises filter errors. - - Forwards every argument unchanged. The only difference from calling - ``litellm.completion`` directly is that an input-filter rejection - (which litellm wraps in ``APIConnectionError``) is re-raised as - :class:`ContentFilteredError`. Output-filter rejections already - surface as :class:`ContentFilteredError` via the SDK's transport patch - and pass through unchanged. + """Wrapper around :func:`litellm.completion` that normalises filter errors + and handles credential rotation transparently. - All other exceptions surface verbatim. + 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) + except litellm.AuthenticationError: + reload_aicore_credentials() + return litellm.completion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: @@ -83,10 +111,13 @@ def completion(*args: Any, **kwargs: Any) -> Any: async def acompletion(*args: Any, **kwargs: Any) -> Any: """Async wrapper around :func:`litellm.acompletion`. - Same translation semantics as :func:`completion`. + Same translation and credential-rotation semantics as :func:`completion`. """ try: return await litellm.acompletion(*args, **kwargs) + except litellm.AuthenticationError: + reload_aicore_credentials() + return await litellm.acompletion(*args, **kwargs) except Exception as exc: translated = _maybe_translate_filter_error(exc) if translated is exc: @@ -94,4 +125,4 @@ async def acompletion(*args: Any, **kwargs: Any) -> Any: raise translated from exc -__all__ = ["completion", "acompletion"] +__all__ = ["completion", "acompletion", "reload_aicore_credentials"] diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py index 5439329c..50acb264 100644 --- a/tests/aicore/unit/test_aicore.py +++ b/tests/aicore/unit/test_aicore.py @@ -10,6 +10,7 @@ _get_secret, set_aicore_config, ) +from sap_cloud_sdk.aicore import _is_transparent_tls class TestGetSecret: @@ -710,3 +711,127 @@ def test_set_config_decorated_with_record_metrics(self): # Function should complete without errors even with decorator # The actual telemetry recording is tested in telemetry tests + + +class TestIsTransparentTls: + """Test suite for _is_transparent_tls helper.""" + + def test_returns_true_for_value_true(self): + with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}): + assert _is_transparent_tls() is True + + def test_returns_true_for_value_1(self): + with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "1"}): + assert _is_transparent_tls() is True + + def test_returns_true_for_value_yes(self): + with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "yes"}): + assert _is_transparent_tls() is True + + def test_returns_true_case_insensitive(self): + with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "TRUE"}): + assert _is_transparent_tls() is True + + def test_returns_false_when_absent(self): + with patch.dict("os.environ", {}, clear=True): + assert _is_transparent_tls() is False + + def test_returns_false_for_value_false(self): + with patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "false"}): + assert _is_transparent_tls() is False + + +class TestSetAICoreConfigTransparentTls: + """Test suite for set_aicore_config in transparent TLS mode.""" + + def _base_secrets(self): + return { + "AICORE_CLIENT_ID": "test-client-id", + "AICORE_AUTH_URL": "https://auth.example.com", + "AICORE_RESOURCE_GROUP": "default", + } + + def test_transparent_tls_does_not_set_client_secret(self): + """In transparent TLS mode, AICORE_CLIENT_SECRET must not be written to env.""" + with ( + patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret, + patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value="https://api.example.com"), + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True), + ): + mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": ( + self._base_secrets().get(name, default) + ) + + set_aicore_config() + + assert "AICORE_CLIENT_SECRET" not in os.environ + + def test_transparent_tls_removes_stale_client_secret(self): + """Any pre-existing AICORE_CLIENT_SECRET is cleared in transparent TLS mode.""" + with ( + patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret, + patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""), + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict( + "os.environ", + {"AICORE_TRANSPARENT_TLS": "true", "AICORE_CLIENT_SECRET": "stale-secret"}, + clear=True, + ), + ): + mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": ( + self._base_secrets().get(name, default) + ) + + set_aicore_config() + + assert "AICORE_CLIENT_SECRET" not in os.environ + + def test_transparent_tls_sets_other_credentials(self): + """Non-secret credentials are still set in transparent TLS mode.""" + with ( + patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret, + patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value="https://api.example.com"), + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True), + ): + mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": ( + self._base_secrets().get(name, default) + ) + + set_aicore_config() + + assert os.environ["AICORE_CLIENT_ID"] == "test-client-id" + assert os.environ["AICORE_AUTH_URL"] == "https://auth.example.com/oauth/token" + assert os.environ["AICORE_BASE_URL"] == "https://api.example.com/v2" + + def test_standard_mode_still_sets_client_secret(self): + """Regression: without transparent TLS, client_secret is still written.""" + with ( + patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret, + patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""), + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", {}, clear=True), + ): + mock_get_secret.side_effect = lambda name, file_name=None, default="", instance_name="aicore-instance": ( + {**self._base_secrets(), "AICORE_CLIENT_SECRET": "my-secret"}.get(name, default) + ) + + set_aicore_config() + + assert os.environ["AICORE_CLIENT_SECRET"] == "my-secret" + + def test_transparent_tls_does_not_call_get_secret_for_client_secret(self): + """_get_secret should not be called for clientsecret in transparent TLS mode.""" + with ( + patch("sap_cloud_sdk.aicore._get_secret") as mock_get_secret, + patch("sap_cloud_sdk.aicore._get_aicore_base_url", return_value=""), + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", {"AICORE_TRANSPARENT_TLS": "true"}, clear=True), + ): + mock_get_secret.return_value = "" + + set_aicore_config() + + called_names = [c.args[0] for c in mock_get_secret.call_args_list] + assert "AICORE_CLIENT_SECRET" not in called_names diff --git a/tests/aicore/unit/test_completion.py b/tests/aicore/unit/test_completion.py index 9857f1d4..5ea4aaf6 100644 --- a/tests/aicore/unit/test_completion.py +++ b/tests/aicore/unit/test_completion.py @@ -15,17 +15,21 @@ - :class:`ContentFilteredError` already raised by the transport patch passes through unchanged (we don't double-wrap). - ``acompletion`` exhibits the same behaviour on the async path. +- On ``AuthenticationError``, credentials are reloaded and the call is + retried once (credential rotation without pod restart). """ from __future__ import annotations import asyncio import json -from unittest.mock import patch +from unittest.mock import MagicMock, call, 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.filtering.exceptions import ContentFilteredError @@ -187,13 +191,142 @@ async def fake_acompletion(**kwargs): def test_non_filter_exception_surfaces_verbatim(self): raised = _FakeAPIConnectionError("SapException - other transport error") - async def fake_acompletion(**kwargs): + async def fake_acompletion_non_filter(**kwargs): raise raised with patch( "sap_cloud_sdk.aicore.completion.litellm.acompletion", - side_effect=fake_acompletion, + side_effect=fake_acompletion_non_filter, ): with pytest.raises(_FakeAPIConnectionError) as ei: asyncio.run(acompletion(model="sap/x", messages=[])) assert ei.value is raised + + +# --------------------------------------------------------------------------- +# reload_aicore_credentials() +# --------------------------------------------------------------------------- + + +class TestReloadAICoreCredentials: + def test_calls_set_aicore_config(self): + with patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_config: + reload_aicore_credentials() + mock_config.assert_called_once_with() + + +# --------------------------------------------------------------------------- +# Reactive reload on AuthenticationError — sync +# --------------------------------------------------------------------------- + + +class TestCompletionReactiveReload: + def test_auth_error_triggers_reload_and_retry_succeeds(self): + """On AuthenticationError, credentials reload and second call succeeds.""" + sentinel = object() + auth_err = litellm.AuthenticationError( + message="401 Unauthorized", 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") as mock_reload, + ): + result = completion(model="sap/x", messages=[]) + + assert result is sentinel + mock_reload.assert_called_once_with() + + def test_auth_error_retry_also_fails_propagates(self): + """If the retry also raises AuthenticationError, it propagates to the caller.""" + auth_err = litellm.AuthenticationError( + message="401 Unauthorized", llm_provider="sap", model="sap/x" + ) + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=auth_err), + patch("sap_cloud_sdk.aicore.set_aicore_config"), + ): + with pytest.raises(litellm.AuthenticationError): + completion(model="sap/x", messages=[]) + + def test_auth_error_reload_called_exactly_once(self): + """Reload is called exactly once — no infinite retry loop.""" + auth_err = litellm.AuthenticationError( + message="401", llm_provider="sap", model="sap/x" + ) + mock_litellm = MagicMock(side_effect=auth_err) + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.completion", mock_litellm), + patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload, + ): + with pytest.raises(litellm.AuthenticationError): + completion(model="sap/x", messages=[]) + + mock_reload.assert_called_once() + assert mock_litellm.call_count == 2 + + def test_non_auth_error_does_not_trigger_reload(self): + """Non-authentication errors do not trigger a credential reload.""" + raised = ValueError("some other error") + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.completion", side_effect=raised), + patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload, + ): + with pytest.raises(ValueError): + completion(model="sap/x", messages=[]) + + mock_reload.assert_not_called() + + +# --------------------------------------------------------------------------- +# Reactive reload on AuthenticationError — async +# --------------------------------------------------------------------------- + + +class TestACompletionReactiveReload: + def test_auth_error_triggers_reload_and_retry_succeeds(self): + sentinel = object() + auth_err = litellm.AuthenticationError( + message="401 Unauthorized", llm_provider="sap", model="sap/x" + ) + call_returns = [auth_err, sentinel] + + async def fake_acompletion(*args, **kwargs): + result = call_returns.pop(0) + if isinstance(result, Exception): + raise result + return result + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake_acompletion), + patch("sap_cloud_sdk.aicore.set_aicore_config") as mock_reload, + ): + result = asyncio.run(acompletion(model="sap/x", messages=[])) + + assert result is sentinel + mock_reload.assert_called_once_with() + + def test_auth_error_retry_also_fails_propagates(self): + auth_err = litellm.AuthenticationError( + message="401", llm_provider="sap", model="sap/x" + ) + + async def fake_acompletion(*args, **kwargs): + raise auth_err + + with ( + patch("sap_cloud_sdk.aicore.completion.litellm.acompletion", side_effect=fake_acompletion), + patch("sap_cloud_sdk.aicore.set_aicore_config"), + ): + with pytest.raises(litellm.AuthenticationError): + asyncio.run(acompletion(model="sap/x", messages=[]))