feat(aicore): transparent TLS mode and reactive credential reload

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,15 +170,24 @@ 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:
         os.environ["AICORE_BASE_URL"] = base_url
     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",

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,15 +111,18 @@ 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:
             raise
         raise translated from exc
 
 
-__all__ = ["completion", "acompletion"]
+__all__ = ["completion", "acompletion", "reload_aicore_credentials"]

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