feat: add secure passthrough header forwarding with forward_headers and extra_blocked_headers

Author: skamenan7

docs/docs/providers/inference/remote_passthrough.mdx

@@ -35,10 +35,12 @@ Passthrough inference provider for connecting to any external inference service
 | `network.timeout.read` | `float \| None` | No |  | Read timeout in seconds. |
 | `network.headers` | `dict[str, str] \| None` | No |  | Additional HTTP headers to include in all requests. |
 | `base_url` | `HttpUrl \| None` | No |  | The URL for the passthrough endpoint |
+| `forward_headers` | `dict[str, str] \| None` | No |  | Mapping of X-LlamaStack-Provider-Data keys to outbound HTTP header names. Only listed keys are forwarded — all others are ignored (default-deny). Values are forwarded verbatim; include any required prefix in the client payload (e.g. 'Bearer sk-xxx' not 'sk-xxx' when targeting Authorization). Header name values should use canonical HTTP casing (e.g. 'Authorization', 'X-Tenant-ID'). Keys with a __ prefix and core security-sensitive headers (for example Host, Content-Type, Transfer-Encoding, Cookie) are rejected at config parse time. When this field is set and auth comes from forwarded headers rather than a static api_key, the caller must include the required keys in X-LlamaStack-Provider-Data on every request. Example: {"maas_api_token": "Authorization"} |
+| `extra_blocked_headers` | `list[str]` | No | [] | Additional outbound header names to block in forward_headers. Names are matched case-insensitively and added to the core blocked list. This can tighten policy but cannot unblock core security-sensitive headers. |
 
 ## Sample Configuration
 
 ```yaml
 base_url: ${env.PASSTHROUGH_URL}
-api_key: ${env.PASSTHROUGH_API_KEY}
+api_key: ${env.PASSTHROUGH_API_KEY:=}
 ```

docs/docs/providers/safety/remote_passthrough.mdx

@@ -16,7 +16,8 @@ Passthrough safety provider that forwards moderation calls to a downstream HTTP
 |-------|------|----------|---------|-------------|
 | `base_url` | `HttpUrl` | No |  | Base URL of the downstream safety service (e.g. https://safety.example.com/v1) |
 | `api_key` | `SecretStr \| None` | No |  | API key for the downstream safety service. If set, takes precedence over provider data. |
-| `forward_headers` | `dict[str, str]` | No | {} | Mapping of provider data keys to outbound HTTP header names. Only keys listed here are forwarded from X-LlamaStack-Provider-Data to the downstream service. Example: {"maas_api_token": "Authorization"} |
+| `forward_headers` | `dict[str, str] \| None` | No |  | Mapping of provider data keys to outbound HTTP header names. Only keys listed here are forwarded from X-LlamaStack-Provider-Data to the downstream service. Keys with a __ prefix and core security-sensitive headers (for example Host, Content-Type, Transfer-Encoding, Cookie) are rejected at config parse time. Example: {"maas_api_token": "Authorization"} |
+| `extra_blocked_headers` | `list[str]` | No | [] | Additional outbound header names to block in forward_headers. Names are matched case-insensitively and added to the core blocked list. This can tighten policy but cannot unblock core security-sensitive headers. |
 
 ## Sample Configuration
 

src/llama_stack/providers/remote/inference/passthrough/__init__.py

@@ -4,20 +4,29 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
-from pydantic import BaseModel, SecretStr
+from pydantic import BaseModel, ConfigDict, SecretStr
 
 from .config import PassthroughImplConfig
 
 
 class PassthroughProviderDataValidator(BaseModel):
-    passthrough_url: str
-    passthrough_api_key: SecretStr
+    # Lives here because the framework resolves provider_data_validator by module path,
+    # and the registry entry points to this package root.
+    #
+    # extra="allow" because forward_headers key names (e.g. "maas_api_token") are
+    # deployer-defined at config time — they can't be declared as typed fields.
+    # Without it, Pydantic drops them before build_forwarded_headers() can read them.
+    model_config = ConfigDict(extra="allow")
+
+    passthrough_url: str | None = None
+    passthrough_api_key: SecretStr | None = None
 
 
 async def get_adapter_impl(config: PassthroughImplConfig, _deps):
     from .passthrough import PassthroughInferenceAdapter
 
-    assert isinstance(config, PassthroughImplConfig), f"Unexpected config type: {type(config)}"
+    if not isinstance(config, PassthroughImplConfig):
+        raise ValueError(f"Unexpected config type: {type(config)}")
     impl = PassthroughInferenceAdapter(config)
     await impl.initialize()
     return impl

src/llama_stack/providers/remote/inference/passthrough/config.py

@@ -6,8 +6,9 @@
 
 from typing import Any
 
-from pydantic import Field, HttpUrl
+from pydantic import Field, HttpUrl, model_validator
 
+from llama_stack.providers.utils.forward_headers import validate_forward_headers_config
 from llama_stack.providers.utils.inference.model_registry import RemoteInferenceProviderConfig
 from llama_stack_api import json_schema_type
 
@@ -18,12 +19,50 @@ class PassthroughImplConfig(RemoteInferenceProviderConfig):
         default=None,
         description="The URL for the passthrough endpoint",
     )
+    forward_headers: dict[str, str] | None = Field(
+        default=None,
+        description=(
+            "Mapping of X-LlamaStack-Provider-Data keys to outbound HTTP header names. "
+            "Only listed keys are forwarded — all others are ignored (default-deny). "
+            "Values are forwarded verbatim; include any required prefix in the client payload "
+            "(e.g. 'Bearer sk-xxx' not 'sk-xxx' when targeting Authorization). "
+            "Header name values should use canonical HTTP casing (e.g. 'Authorization', 'X-Tenant-ID'). "
+            "Keys with a __ prefix and core security-sensitive headers (for example Host, "
+            "Content-Type, Transfer-Encoding, Cookie) are rejected at config parse time. "
+            "When this field is set and auth comes from forwarded headers rather than a static api_key, "
+            "the caller must include the required keys in X-LlamaStack-Provider-Data on every request. "
+            'Example: {"maas_api_token": "Authorization"}'
+        ),
+    )
+    extra_blocked_headers: list[str] = Field(
+        default_factory=list,
+        description=(
+            "Additional outbound header names to block in forward_headers. "
+            "Names are matched case-insensitively and added to the core blocked list. "
+            "This can tighten policy but cannot unblock core security-sensitive headers."
+        ),
+    )
+
+    @model_validator(mode="after")
+    def validate_forward_headers(self) -> "PassthroughImplConfig":
+        validate_forward_headers_config(self.forward_headers, self.extra_blocked_headers)
+        return self
 
     @classmethod
     def sample_run_config(
-        cls, base_url: HttpUrl | None = "${env.PASSTHROUGH_URL}", api_key: str = "${env.PASSTHROUGH_API_KEY}", **kwargs
+        cls,
+        base_url: HttpUrl | None = "${env.PASSTHROUGH_URL}",
+        api_key: str = "${env.PASSTHROUGH_API_KEY:=}",
+        forward_headers: dict[str, str] | None = None,
+        extra_blocked_headers: list[str] | None = None,
+        **kwargs,
     ) -> dict[str, Any]:
-        return {
+        config: dict[str, Any] = {
             "base_url": base_url,
             "api_key": api_key,
         }
+        if forward_headers:
+            config["forward_headers"] = forward_headers
+        if extra_blocked_headers:
+            config["extra_blocked_headers"] = extra_blocked_headers
+        return config

src/llama_stack/providers/remote/inference/passthrough/passthrough.py

@@ -9,6 +9,8 @@
 from openai import AsyncOpenAI
 
 from llama_stack.core.request_headers import NeedsRequestProviderData
+from llama_stack.log import get_logger
+from llama_stack.providers.utils.forward_headers import build_forwarded_headers
 from llama_stack.providers.utils.inference.stream_utils import wrap_async_stream
 from llama_stack_api import (
     Inference,
@@ -24,6 +26,8 @@
 
 from .config import PassthroughImplConfig
 
+logger = get_logger(__name__, category="inference")
+
 
 class PassthroughInferenceAdapter(NeedsRequestProviderData, Inference):
     def __init__(self, config: PassthroughImplConfig) -> None:
@@ -74,37 +78,70 @@ async def should_refresh_models(self) -> bool:
     def _get_openai_client(self) -> AsyncOpenAI:
         """Get an AsyncOpenAI client configured for the downstream server."""
         base_url = self._get_passthrough_url()
-        api_key = self._get_passthrough_api_key()
+        request_headers = self._build_request_headers()
 
+        # api_key="" means the SDK adds no Authorization header of its own;
+        # auth comes entirely from request_headers (forwarded or static api_key).
+        # This avoids the "passthrough" sentinel that would send a spurious
+        # Authorization: Bearer passthrough to every downstream, even when
+        # forward_headers only targets non-auth headers like X-Tenant-ID.
         return AsyncOpenAI(
             base_url=f"{base_url.rstrip('/')}/v1",
-            api_key=api_key,
+            api_key="",
+            default_headers=request_headers or None,
         )
 
+    def _build_request_headers(self) -> dict[str, str]:
+        """Build outbound headers: forwarded provider-data keys first, then static api_key.
+
+        Static api_key always wins over a forwarded Authorization, regardless of casing.
+        """
+        provider_data = self.get_request_provider_data()
+        headers = build_forwarded_headers(provider_data, self.config.forward_headers)
+        if self.config.forward_headers and not headers:
+            logger.warning(
+                "forward_headers is configured but no matching keys found in provider data — "
+                "outbound request may be unauthenticated"
+            )
+        api_key = self._get_passthrough_api_key_or_none(provider_data)
+        if api_key:
+            # remove any forwarded authorization variant (case-insensitive) so static key wins
+            headers = {k: v for k, v in headers.items() if k.lower() != "authorization"}
+            headers["Authorization"] = f"Bearer {api_key}"
+        return headers
+
+    def _get_passthrough_api_key_or_none(self, provider_data: object | None = None) -> str | None:
+        """Return the static or per-request API key, or None if not configured."""
+        if self.config.auth_credential is not None:
+            configured_api_key = self.config.auth_credential.get_secret_value()
+            if configured_api_key:
+                return configured_api_key
+
+        if provider_data is None:
+            provider_data = self.get_request_provider_data()
+        passthrough_api_key = getattr(provider_data, "passthrough_api_key", None)
+        if passthrough_api_key is not None:
+            if hasattr(passthrough_api_key, "get_secret_value"):
+                provider_data_api_key = passthrough_api_key.get_secret_value()
+            else:
+                provider_data_api_key = str(passthrough_api_key)
+            if provider_data_api_key:
+                return provider_data_api_key
+
+        return None
+
     def _get_passthrough_url(self) -> str:
         """Get the passthrough URL from config or provider data."""
         if self.config.base_url is not None:
             return str(self.config.base_url)
 
         provider_data = self.get_request_provider_data()
-        if provider_data is None:
+        if provider_data is None or provider_data.passthrough_url is None:
             raise ValueError(
                 'Pass url of the passthrough endpoint in the header X-LlamaStack-Provider-Data as { "passthrough_url": <your passthrough url>}'
             )
         return provider_data.passthrough_url
 
-    def _get_passthrough_api_key(self) -> str:
-        """Get the passthrough API key from config or provider data."""
-        if self.config.auth_credential is not None:
-            return self.config.auth_credential.get_secret_value()
-
-        provider_data = self.get_request_provider_data()
-        if provider_data is None:
-            raise ValueError(
-                'Pass API Key for the passthrough endpoint in the header X-LlamaStack-Provider-Data as { "passthrough_api_key": <your api key>}'
-            )
-        return provider_data.passthrough_api_key.get_secret_value()
-
     async def openai_completion(
         self,
         params: OpenAICompletionRequestWithExtraBody,

src/llama_stack/providers/remote/safety/passthrough/config.py

@@ -6,28 +6,16 @@
 
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, model_validator
 
+from llama_stack.providers.utils.forward_headers import validate_forward_headers_config
 from llama_stack_api import json_schema_type
 
-_BLOCKED_HEADERS = frozenset(
-    {
-        "host",
-        "content-type",
-        "content-length",
-        "transfer-encoding",
-        "connection",
-        "upgrade",
-        "te",
-        "trailer",
-        "cookie",
-        "set-cookie",
-    }
-)
-
 
 class PassthroughProviderDataValidator(BaseModel):
-    # allow arbitrary keys so forward_headers can access them
+    # extra="allow" because forward_headers key names (e.g. "maas_api_token") are
+    # deployer-defined at config time — they can't be declared as typed fields.
+    # Without it, Pydantic drops them before build_forwarded_headers() can read them.
     model_config = ConfigDict(extra="allow")
 
     passthrough_api_key: SecretStr | None = Field(
@@ -46,34 +34,37 @@ class PassthroughSafetyConfig(BaseModel):
         default=None,
         description="API key for the downstream safety service. If set, takes precedence over provider data.",
     )
-    forward_headers: dict[str, str] = Field(
-        default_factory=dict,
+    forward_headers: dict[str, str] | None = Field(
+        default=None,
         description=(
             "Mapping of provider data keys to outbound HTTP header names. "
-            "Only keys listed here are forwarded from X-LlamaStack-Provider-Data "
-            'to the downstream service. Example: {"maas_api_token": "Authorization"}'
+            "Only keys listed here are forwarded from X-LlamaStack-Provider-Data to the downstream service. "
+            "Keys with a __ prefix and core security-sensitive headers (for example Host, "
+            "Content-Type, Transfer-Encoding, Cookie) are rejected at config parse time. "
+            'Example: {"maas_api_token": "Authorization"}'
+        ),
+    )
+    extra_blocked_headers: list[str] = Field(
+        default_factory=list,
+        description=(
+            "Additional outbound header names to block in forward_headers. "
+            "Names are matched case-insensitively and added to the core blocked list. "
+            "This can tighten policy but cannot unblock core security-sensitive headers."
         ),
     )
 
-    @field_validator("forward_headers")
-    @classmethod
-    def validate_forward_headers(cls, v: dict[str, str]) -> dict[str, str]:
-        errors: list[str] = []
-        for provider_key, header_name in v.items():
-            if provider_key.startswith("__"):
-                errors.append(f"provider key '{provider_key}' uses reserved __ prefix")
-            if header_name.lower() in _BLOCKED_HEADERS:
-                errors.append(f"header '{header_name}' is blocked (security-sensitive)")
-        if errors:
-            raise ValueError(f"invalid forward_headers: {'; '.join(errors)}")
-        return v
+    @model_validator(mode="after")
+    def validate_forward_headers(self) -> "PassthroughSafetyConfig":
+        validate_forward_headers_config(self.forward_headers, self.extra_blocked_headers)
+        return self
 
     @classmethod
     def sample_run_config(
         cls,
         base_url: str = "${env.PASSTHROUGH_SAFETY_URL}",
         api_key: str = "${env.PASSTHROUGH_SAFETY_API_KEY:=}",
         forward_headers: dict[str, str] | None = None,
+        extra_blocked_headers: list[str] | None = None,
         **kwargs: Any,
     ) -> dict[str, Any]:
         config: dict[str, Any] = {
@@ -82,4 +73,6 @@ def sample_run_config(
         }
         if forward_headers:
             config["forward_headers"] = forward_headers
+        if extra_blocked_headers:
+            config["extra_blocked_headers"] = extra_blocked_headers
         return config

src/llama_stack/providers/remote/safety/passthrough/passthrough.py

@@ -9,9 +9,12 @@
 from typing import Any
 
 import httpx
-from pydantic import SecretStr
 
 from llama_stack.core.request_headers import NeedsRequestProviderData
+from llama_stack.log import get_logger
+from llama_stack.providers.utils.forward_headers import build_forwarded_headers
+
+logger = get_logger(__name__, category="safety")
 from llama_stack_api import (
     GetShieldRequest,
     ModerationObject,
@@ -69,37 +72,27 @@ def _get_api_key(self) -> str | None:
 
     def _build_forward_headers(self) -> dict[str, str]:
         """Build outbound headers from provider data using the forward_headers mapping."""
-        if not self.config.forward_headers:
-            return {}
-
         provider_data = self.get_request_provider_data()
-        if provider_data is None:
-            return {}
-
-        headers: dict[str, str] = {}
-        raw = provider_data.model_dump()
-        for provider_key, header_name in self.config.forward_headers.items():
-            value = raw.get(provider_key)
-            if value is not None:
-                # unwrap SecretStr so we forward the real value, not '**********'
-                if isinstance(value, SecretStr):
-                    value = value.get_secret_value()
-                # strip control chars that could enable header injection
-                sanitized = str(value).replace("\r", "").replace("\n", "")
-                headers[header_name] = sanitized
-        return headers
+        forwarded = build_forwarded_headers(provider_data, self.config.forward_headers)
+        if self.config.forward_headers and not forwarded:
+            logger.warning(
+                "forward_headers is configured but no matching keys found in provider data — "
+                "outbound request may be unauthenticated"
+            )
+        return forwarded
 
     def _build_request_headers(self) -> dict[str, str]:
-        """Combine auth + forwarded headers for the downstream request."""
-        headers: dict[str, str] = {"Content-Type": "application/json"}
+        """Combine auth + forwarded headers for the downstream request.
 
-        # forwarded headers go first so config api_key can't be overwritten
+        Forwarded headers go first; static api_key overwrites Authorization if set.
+        build_forwarded_headers() normalizes header names case-insensitively so
+        there are no duplicate Authorization variants in the forwarded dict.
+        """
+        headers: dict[str, str] = {"Content-Type": "application/json"}
         headers.update(self._build_forward_headers())
-
         api_key = self._get_api_key()
         if api_key:
             headers["Authorization"] = f"Bearer {api_key}"
-
         return headers
 
     async def run_shield(self, request: RunShieldRequest) -> RunShieldResponse: