feat(aicore): add transparent proxy routing and BTP Destination Service mode

Implements Option 3 from the AFSDK-4306 security alignment meeting: SDK
absorbs all routing complexity so agent code is identical in all environments.
The deployer controls routing by choosing which env vars to inject.

Two new modes in set_aicore_config():

Proxy mode (AICORE_PROXY_URL set):
- Routes all LiteLLM calls through an external LiteLLM proxy
- Sets litellm.api_base / litellm.api_key globally
- Rewrites sap/<model> → litellm_proxy/<model> transparently in
  completion() and acompletion() wrappers (including on auth-error retry)
- No AI Core credentials written to the process environment
- JWT never reaches the agent process (proxy handles OAuth)

Destination mode (AICORE_DESTINATION_NAME set):
- Loads AI Core credentials at startup from a named BTP Destination Service
  destination via the existing sap_cloud_sdk.destination client
- Deployer only injects Destination Service binding — AI Core client_secret
  is never in the K8s Secret, only in BTP Destination Service
- Combined with _clear_client_secret() (PR #257), the secret is removed
  from env after the first successful LiteLLM call

Direct mode (neither set): existing behaviour unchanged, including
transparent TLS (AICORE_TRANSPARENT_TLS).

Adds 30 unit tests covering both new modes and all edge cases.

AFSDK-4306

Author: tiagoek

src/sap_cloud_sdk/aicore/__init__.py

@@ -14,7 +14,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, _set_proxy_active
 from .filtering import (
     AzureContentFilter,
     ContentFilter,
@@ -31,6 +31,22 @@
 
 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_VIRTUAL_KEY_ENV = "AICORE_PROXY_VIRTUAL_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 +135,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; ``sap/<model>`` is aliased to
+      ``litellm_proxy/<model>`` transparently. No AI Core credentials
+      are written to the process environment.
+
+    - ``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.
 
-    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
+    - 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,44 +163,131 @@ 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 and activates
+    the ``sap/`` → ``litellm_proxy/`` model alias rewrite in the
+    completion wrappers. No AI Core credentials are written to env.
+    """
+    import litellm as _litellm
+
+    virtual_key = os.environ.get(_PROXY_VIRTUAL_KEY_ENV, "")
+    _litellm.api_base = proxy_url
+    if virtual_key:
+        _litellm.api_key = virtual_key
+    _set_proxy_active(True)
+    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:
         os.environ["AICORE_BASE_URL"] = base_url
     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:

src/sap_cloud_sdk/aicore/completion.py

@@ -50,6 +50,7 @@
 from __future__ import annotations
 
 import logging
+import threading
 from typing import Any
 
 import litellm
@@ -58,6 +59,26 @@
 
 logger = logging.getLogger(__name__)
 
+# Proxy mode state — set by _configure_proxy_mode() in __init__.py.
+# When active, completion() rewrites sap/<model> → litellm_proxy/<model>.
+_proxy_lock = threading.Lock()
+_proxy_active: bool = False
+
+
+def _set_proxy_active(value: bool) -> None:
+    """Activate or deactivate proxy model aliasing (called by set_aicore_config)."""
+    global _proxy_active
+    with _proxy_lock:
+        _proxy_active = value
+
+
+def _rewrite_model_for_proxy(kwargs: dict) -> dict:
+    """Rewrite sap/<model> to litellm_proxy/<model> when proxy mode is active."""
+    model = kwargs.get("model", "")
+    if isinstance(model, str) and model.startswith("sap/"):
+        return {**kwargs, "model": "litellm_proxy/" + model[4:]}
+    return kwargs
+
 
 def _maybe_translate_filter_error(exc: BaseException) -> BaseException:
     """Return a :class:`ContentFilteredError` if ``exc`` is a wrapped
@@ -76,10 +97,16 @@ 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.
+
+    When proxy mode is active (``AICORE_PROXY_URL`` set), rewrites
+    ``sap/<model>`` to ``litellm_proxy/<model>`` transparently.
     """
+    with _proxy_lock:
+        proxy = _proxy_active
+    if proxy:
+        kwargs = _rewrite_model_for_proxy(kwargs)
     try:
         return litellm.completion(*args, **kwargs)
     except litellm.AuthenticationError:
@@ -88,6 +115,10 @@ def completion(*args: Any, **kwargs: Any) -> Any:
 
         logger.info("AI Core credentials reloading after authentication failure")
         set_aicore_config()
+        with _proxy_lock:
+            proxy = _proxy_active
+        if proxy:
+            kwargs = _rewrite_model_for_proxy(kwargs)
         return litellm.completion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)
@@ -99,15 +130,23 @@ 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 and proxy aliasing semantics as :func:`completion`.
     """
+    with _proxy_lock:
+        proxy = _proxy_active
+    if proxy:
+        kwargs = _rewrite_model_for_proxy(kwargs)
     try:
         return await litellm.acompletion(*args, **kwargs)
     except litellm.AuthenticationError:
         from sap_cloud_sdk.aicore import set_aicore_config
 
         logger.info("AI Core credentials reloading after authentication failure")
         set_aicore_config()
+        with _proxy_lock:
+            proxy = _proxy_active
+        if proxy:
+            kwargs = _rewrite_model_for_proxy(kwargs)
         return await litellm.acompletion(*args, **kwargs)
     except Exception as exc:
         translated = _maybe_translate_filter_error(exc)