diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py index 1e795980..68ebab50 100644 --- a/src/sap_cloud_sdk/aicore/__init__.py +++ b/src/sap_cloud_sdk/aicore/__init__.py @@ -31,6 +31,26 @@ 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" + +# Option 3 — transparent proxy routing. +# Deployer injects these; agent code is identical in all environments. +_PROXY_URL_ENV = "AICORE_PROXY_URL" +_PROXY_API_KEY_ENV = "AICORE_PROXY_API_KEY" +_DESTINATION_NAME_ENV = "AICORE_DESTINATION_NAME" + + +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, @@ -119,14 +139,25 @@ def _get_aicore_base_url(instance_name: str = "aicore-instance") -> str: def set_aicore_config(instance_name: str = "aicore-instance") -> None: """Load AI Core credentials and activate content filtering. - Loads secrets from files or environment variables and sets them as - process env vars so ``litellm`` picks them up. + Detects which routing mode is active based on environment variables: + + - ``AICORE_PROXY_URL`` set → **proxy mode**: routes all LiteLLM calls + through a LiteLLM proxy via ``litellm.api_base``; model strings are + passed verbatim. No AI Core credentials are written to the process + environment. - File mappings based on the Kubernetes secret structure: - clientid → AICORE_CLIENT_ID - clientsecret → AICORE_CLIENT_SECRET - url → AICORE_AUTH_URL - serviceurls (JSON with AI_API_URL) → AICORE_BASE_URL + - ``AICORE_DESTINATION_NAME`` set → **destination mode**: loads AI Core + credentials from a BTP Destination Service destination at startup. + The deployer only needs to inject Destination Service binding credentials; + the AI Core ``client_secret`` never needs to be in the K8s Secret. + + - Neither set → **direct mode** (existing behaviour): credentials are + loaded from a mounted K8s secret volume or environment variables. + ``AICORE_TRANSPARENT_TLS=true`` suppresses ``client_secret`` and + relies on an mTLS sidecar. + + Agent code is identical in all three modes — the deployer controls + routing by choosing which env vars to inject. After credentials are loaded, content filtering is activated on every ``sap/*`` LiteLLM call at the configured thresholds (default: severity @@ -136,29 +167,112 @@ 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. """ - # Load secrets + proxy_url = os.environ.get(_PROXY_URL_ENV, "") + destination_name = os.environ.get(_DESTINATION_NAME_ENV, "") + + if proxy_url: + _configure_proxy_mode(proxy_url) + elif destination_name: + _configure_destination_mode(destination_name) + else: + _configure_direct_mode(instance_name) + + set_filtering() + + +def _configure_proxy_mode(proxy_url: str) -> None: + """Configure LiteLLM to route calls through an external proxy. + + Sets ``litellm.api_base`` / ``litellm.api_key`` globally. + Model strings (e.g. ``sap/``) are passed verbatim — no rewrite. + No AI Core credentials are written to env. + """ + import litellm as _litellm + + api_key = os.environ.get(_PROXY_API_KEY_ENV, "") + _litellm.api_base = proxy_url + if api_key: + _litellm.api_key = api_key + logger.info("AI Core proxy mode active — routing via %s", proxy_url) + + +def _configure_destination_mode(name: str) -> None: + """Load AI Core credentials from a BTP Destination Service destination. + + Calls the Destination Service at startup to resolve the named destination + and extracts ``clientId``, ``clientSecret``, ``tokenServiceURL``, and the + AI Core ``URL`` from the destination configuration properties. These are + written to the standard ``AICORE_*`` env vars so that LiteLLM can fetch + an OAuth token from XSUAA as usual. + + Security: The deployer does NOT need to inject ``AICORE_CLIENT_SECRET`` + directly — only Destination Service binding credentials are required in + the agent environment. + + Raises ``RuntimeError`` if the destination is not found or does not + return ``clientId`` / ``clientSecret``. + """ + from sap_cloud_sdk.destination import create_client # lazy import + + client = create_client() + dest = client.get_destination(name) + + if dest is None: + raise RuntimeError( + f"AI Core destination '{name}' not found in Destination Service. " + "Check that the destination exists and the binding has access." + ) + + base_url = dest.url or "" + if base_url and not base_url.endswith("/v2"): + base_url = base_url.rstrip("/") + "/v2" + if base_url: + os.environ["AICORE_BASE_URL"] = base_url + + resource_group = dest.properties.get("resource_group", "default") + os.environ["AICORE_RESOURCE_GROUP"] = resource_group + + client_id = dest.properties.get("clientId", "") + client_secret = dest.properties.get("clientSecret", "") + token_service_url = dest.properties.get("tokenServiceURL", "") + + if not client_id or not client_secret: + raise RuntimeError( + f"Destination '{name}' did not return clientId/clientSecret. " + "Ensure the destination uses OAuth2ClientCredentials authentication " + "and the calling app has the Destination Service technical-user scope." + ) + + os.environ["AICORE_CLIENT_ID"] = client_id + os.environ["AICORE_CLIENT_SECRET"] = client_secret + + if token_service_url: + if not token_service_url.endswith("/oauth/token"): + token_service_url = token_service_url.rstrip("/") + "/oauth/token" + os.environ["AICORE_AUTH_URL"] = token_service_url + + logger.info("AI Core destination mode active — credentials loaded from '%s'", name) + + +def _configure_direct_mode(instance_name: str) -> None: + """Load AI Core credentials directly from mounted secrets or env vars.""" + transparent_tls = _is_transparent_tls() + 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( "AICORE_RESOURCE_GROUP", default="default", instance_name=instance_name ) - # Ensure AICORE_AUTH_URL has /oauth/token suffix if auth_url and not auth_url.endswith("/oauth/token"): auth_url = auth_url.rstrip("/") + "/oauth/token" if base_url and not base_url.endswith("/v2"): base_url = base_url.rstrip("/") + "/v2" - # 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: @@ -166,14 +280,17 @@ def set_aicore_config(instance_name: str = "aicore-instance") -> None: if resource_group: os.environ["AICORE_RESOURCE_GROUP"] = resource_group - # Log configuration completion (excluding sensitive information) - logger.info("AI Core configuration has been set successfully") + if transparent_tls: + 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 - # Activate content filtering for all sap/* LiteLLM model calls. - # AICORE_FILTER_ENABLED=false disables; AICORE_FILTER_* tune thresholds. - # Errors propagate — filtering misconfiguration should surface at startup - # rather than be swallowed silently. - set_filtering() + logger.info("AI Core configuration has been set successfully") def _get_secret_dir_mtime(instance_name: str = "aicore-instance") -> float: diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py index a2d72ad7..b8e6b645 100644 --- a/src/sap_cloud_sdk/aicore/completion.py +++ b/src/sap_cloud_sdk/aicore/completion.py @@ -76,9 +76,12 @@ def completion(*args: Any, **kwargs: Any) -> Any: """Wrapper around :func:`litellm.completion` that normalises filter errors and handles credential rotation transparently. - On ``AuthenticationError`` (e.g. rotated client_secret), reloads - credentials from the mounted secret volume and retries once. - All other exceptions surface verbatim after the filter-error translation. + On ``AuthenticationError`` (e.g. rotated client_secret or mTLS cert), + reloads credentials from the mounted secret volume and retries once. + + Model strings (e.g. ``sap/``) are passed verbatim to LiteLLM in all + routing modes — proxy routing is handled by ``litellm.api_base`` configured + in :func:`set_aicore_config`, not by rewriting the model name. """ try: return litellm.completion(*args, **kwargs) @@ -99,7 +102,8 @@ 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-rotation semantics as :func:`completion`. + Model strings are passed verbatim to LiteLLM in all routing modes. """ try: return await litellm.acompletion(*args, **kwargs) diff --git a/tests/aicore/unit/test_aicore.py b/tests/aicore/unit/test_aicore.py index 5439329c..b0103eba 100644 --- a/tests/aicore/unit/test_aicore.py +++ b/tests/aicore/unit/test_aicore.py @@ -4,10 +4,12 @@ import os from unittest.mock import mock_open, patch +import pytest from sap_cloud_sdk.aicore import ( _get_aicore_base_url, _get_secret, + _is_transparent_tls, set_aicore_config, ) @@ -710,3 +712,342 @@ 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 + + +# --------------------------------------------------------------------------- +# Proxy mode — set_aicore_config() with AICORE_PROXY_URL +# --------------------------------------------------------------------------- + + +class TestSetAICoreConfigProxyMode: + """set_aicore_config() routes via proxy when AICORE_PROXY_URL is present.""" + + def _base_proxy_env(self, **extra): + return {"AICORE_PROXY_URL": "https://proxy.example.com", **extra} + + def test_proxy_mode_sets_litellm_api_base(self): + import litellm + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + assert litellm.api_base == "https://proxy.example.com" + litellm.api_base = None # cleanup + + def test_proxy_mode_sets_litellm_api_key_when_virtual_key_present(self): + import litellm + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict( + "os.environ", + self._base_proxy_env(AICORE_PROXY_API_KEY="sk-virt-123"), + clear=True, + ), + ): + set_aicore_config() + assert litellm.api_key == "sk-virt-123" + litellm.api_key = None # cleanup + + def test_proxy_mode_does_not_write_aicore_credentials(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + for var in ("AICORE_CLIENT_ID", "AICORE_CLIENT_SECRET", "AICORE_AUTH_URL"): + assert var not in os.environ, f"{var} must not be written in proxy mode" + + def test_proxy_mode_takes_precedence_over_destination(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._configure_destination_mode") as mock_dest, + patch.dict( + "os.environ", + self._base_proxy_env(AICORE_DESTINATION_NAME="aicore"), + clear=True, + ), + ): + set_aicore_config() + mock_dest.assert_not_called() + + def test_proxy_mode_still_calls_set_filtering(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter, + patch.dict("os.environ", self._base_proxy_env(), clear=True), + ): + set_aicore_config() + mock_filter.assert_called_once() + + def test_direct_mode_used_when_neither_proxy_nor_destination_set(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.aicore._configure_direct_mode") as mock_direct, + patch.dict("os.environ", {}, clear=True), + ): + set_aicore_config() + mock_direct.assert_called_once() + + +# --------------------------------------------------------------------------- +# Destination mode — set_aicore_config() with AICORE_DESTINATION_NAME +# --------------------------------------------------------------------------- + + +class TestSetAICoreConfigDestinationMode: + """set_aicore_config() loads credentials from BTP Destination Service.""" + + def _mock_destination( + self, + url="https://api.ai.prod.example.com", + properties=None, + auth_tokens=None, + ): + from unittest.mock import MagicMock + dest = MagicMock() + dest.url = url + dest.properties = properties or { + "clientId": "sb-client-id", + "clientSecret": "client-secret-value", + "tokenServiceURL": "https://auth.example.com/oauth/token", + } + dest.auth_tokens = auth_tokens or [] + return dest + + def test_destination_mode_sets_base_url_with_v2_suffix(self): + dest = self._mock_destination(url="https://api.ai.prod.example.com") + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch( + "sap_cloud_sdk.destination.create_client" + ) as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2" + + def test_destination_mode_does_not_double_v2(self): + dest = self._mock_destination(url="https://api.ai.prod.example.com/v2") + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_BASE_URL"] == "https://api.ai.prod.example.com/v2" + + def test_destination_mode_sets_resource_group_from_properties(self): + dest = self._mock_destination( + properties={ + "clientId": "id", + "clientSecret": "sec", + "tokenServiceURL": "https://auth.example.com/oauth/token", + "resource_group": "production", + } + ) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_RESOURCE_GROUP"] == "production" + + def test_destination_mode_defaults_resource_group_to_default(self): + dest = self._mock_destination() # no resource_group in properties + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_RESOURCE_GROUP"] == "default" + + def test_destination_mode_sets_client_credentials(self): + dest = self._mock_destination() + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_CLIENT_ID"] == "sb-client-id" + assert os.environ["AICORE_CLIENT_SECRET"] == "client-secret-value" + + def test_destination_mode_appends_oauth_token_suffix(self): + dest = self._mock_destination( + properties={ + "clientId": "id", + "clientSecret": "sec", + "tokenServiceURL": "https://auth.example.com", # no /oauth/token + } + ) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + assert os.environ["AICORE_AUTH_URL"] == "https://auth.example.com/oauth/token" + + def test_destination_mode_raises_when_destination_not_found(self): + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "missing"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = None + with pytest.raises(RuntimeError, match="not found"): + set_aicore_config() + + def test_destination_mode_raises_when_no_client_credentials(self): + dest = self._mock_destination(properties={"resource_group": "default"}) + with ( + patch("sap_cloud_sdk.aicore.set_filtering"), + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + with pytest.raises(RuntimeError, match="clientId/clientSecret"): + set_aicore_config() + + def test_destination_mode_still_calls_set_filtering(self): + dest = self._mock_destination() + with ( + patch("sap_cloud_sdk.aicore.set_filtering") as mock_filter, + patch("sap_cloud_sdk.destination.create_client") as mock_create, + patch.dict("os.environ", {"AICORE_DESTINATION_NAME": "aicore"}, clear=True), + ): + mock_create.return_value.get_destination.return_value = dest + set_aicore_config() + mock_filter.assert_called_once()