feat: passthrough safety provider for forwarding to downstream /v1/moderations (#5004)

# What does this PR do?

Right now, external providers can't access per-request headers like API
tokens because NeedsRequestProviderData lives in llama_stack.core, not
in the public llama-stack-api package. This adds a remote::passthrough
safety provider that demonstrates the forward_headers pattern — a
deployer-controlled mapping from X-LlamaStack-Provider-Data keys to
outbound HTTP headers, so only the keys each provider needs get
forwarded to the downstream service.

The provider forwards moderation calls to any downstream HTTP service
implementing OpenAI's /v1/moderations endpoint, supporting both
run_shield (Llama Stack native) and run_moderation (OpenAI-compatible)
paths.

Hop-by-hop and framing headers (Host, Transfer-Encoding, etc.) are
blocked via a `_BLOCKED_HEADERS` frozenset to prevent SSRF and request
smuggling, and dunder keys like `__authenticated_user` never leak
downstream.

### Usage

```yaml
providers:
  - provider_id: passthrough
    provider_type: remote::passthrough
    config:
      base_url: ${env.SAFETY_SERVICE_URL}
      api_key: ${env.SAFETY_API_KEY:=}
      forward_headers:
        maas_api_token: Authorization
        tenant_id: X-Tenant-Id
        team_id: X-Team-Id
```

Clients pass per-request credentials via X-LlamaStack-Provider-Data, and
only the keys listed in forward_headers get mapped to outbound HTTP
headers on the downstream call.

Relates #4607

Will open follow up PRs for other passthrough providers

## Test Plan

```bash
uv run pytest tests/unit/providers/safety/ tests/integration/safety/test_passthrough.py -v
```

Manually tested against a local mock server. Full example:

```bash
curl -s http://localhost:8321/v1/moderations \
  -H "Content-Type: application/json" \
  -H 'X-LlamaStack-Provider-Data: {"maas_api_token": "Bearer trustai-prod-xyz", "tenant_id": "enterprise-42", "host_override": "evil.internal"}' \
  -d '{"input": "test", "model": "text-moderation-mock"}'

# headers received by mock server:
curl -s http://127.0.0.1:9999/v1/debug/last-headers | python3 -m json.tool
# {
#     "Authorization": "Bearer trustai-prod-xyz",   <-- forwarded from maas_api_token
#     "X-Tenant-Id": "enterprise-42",               <-- forwarded from tenant_id
#     "Content-Type": "application/json",
#     "Host": "127.0.0.1:9999"                      <-- NOT evil.internal (blocked)
# }
```

## Breaking changes

None

Author: skamenan7

docs/docs/providers/safety/remote_passthrough.mdx

@@ -0,0 +1,26 @@
+---
+description: "Passthrough safety provider that forwards moderation calls to a downstream HTTP service."
+sidebar_label: Remote - Passthrough
+title: remote::passthrough
+---
+
+# remote::passthrough
+
+## Description
+
+Passthrough safety provider that forwards moderation calls to a downstream HTTP service.
+
+## Configuration
+
+| Field | Type | Required | Default | Description |
+|-------|------|----------|---------|-------------|
+| `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"} |
+
+## Sample Configuration
+
+```yaml
+base_url: ${env.PASSTHROUGH_SAFETY_URL}
+api_key: ${env.PASSTHROUGH_SAFETY_API_KEY:=}
+```

src/llama_stack/providers/registry/safety.py

@@ -65,6 +65,16 @@ def available_providers() -> list[ProviderSpec]:
             config_class="llama_stack.providers.remote.safety.nvidia.NVIDIASafetyConfig",
             description="NVIDIA's safety provider for content moderation and safety filtering.",
         ),
+        RemoteProviderSpec(
+            api=Api.safety,
+            adapter_type="passthrough",
+            provider_type="remote::passthrough",
+            pip_packages=[],
+            module="llama_stack.providers.remote.safety.passthrough",
+            config_class="llama_stack.providers.remote.safety.passthrough.PassthroughSafetyConfig",
+            provider_data_validator="llama_stack.providers.remote.safety.passthrough.config.PassthroughProviderDataValidator",
+            description="Passthrough safety provider that forwards moderation calls to a downstream HTTP service.",
+        ),
         RemoteProviderSpec(
             api=Api.safety,
             adapter_type="sambanova",

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

@@ -0,0 +1,17 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from typing import Any
+
+from .config import PassthroughSafetyConfig
+
+
+async def get_adapter_impl(config: PassthroughSafetyConfig, _deps: Any) -> Any:
+    from .passthrough import PassthroughSafetyAdapter
+
+    impl = PassthroughSafetyAdapter(config)
+    await impl.initialize()
+    return impl

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

@@ -0,0 +1,85 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator
+
+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
+    model_config = ConfigDict(extra="allow")
+
+    passthrough_api_key: SecretStr | None = Field(
+        default=None,
+        description="API key for the downstream safety service",
+    )
+
+
+@json_schema_type
+class PassthroughSafetyConfig(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    base_url: HttpUrl = Field(
+        description="Base URL of the downstream safety service (e.g. https://safety.example.com/v1)",
+    )
+    api_key: SecretStr | None = Field(
+        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,
+        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"}'
+        ),
+    )
+
+    @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
+
+    @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,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        config: dict[str, Any] = {
+            "base_url": base_url,
+            "api_key": api_key,
+        }
+        if forward_headers:
+            config["forward_headers"] = forward_headers
+        return config

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

@@ -0,0 +1,245 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+import asyncio
+import uuid
+from typing import Any
+
+import httpx
+from pydantic import SecretStr
+
+from llama_stack.core.request_headers import NeedsRequestProviderData
+from llama_stack_api import (
+    GetShieldRequest,
+    ModerationObject,
+    ModerationObjectResults,
+    RunModerationRequest,
+    RunShieldRequest,
+    RunShieldResponse,
+    Safety,
+    SafetyViolation,
+    Shield,
+    ShieldsProtocolPrivate,
+    ViolationLevel,
+)
+
+from .config import PassthroughSafetyConfig
+
+
+class PassthroughSafetyAdapter(
+    Safety,
+    ShieldsProtocolPrivate,
+    NeedsRequestProviderData,
+):
+    """Forwards safety calls to a downstream service via /v1/moderations."""
+
+    shield_store: Any  # injected by framework after initialization
+
+    def __init__(self, config: PassthroughSafetyConfig) -> None:
+        self.config = config
+        self._client = httpx.AsyncClient(timeout=httpx.Timeout(30.0))
+
+    async def initialize(self) -> None:
+        pass
+
+    async def shutdown(self) -> None:
+        # shield so cancellation doesn't leak the connection
+        await asyncio.shield(self._client.aclose())
+
+    async def register_shield(self, shield: Shield) -> None:
+        pass
+
+    async def unregister_shield(self, identifier: str) -> None:
+        pass
+
+    def _get_api_key(self) -> str | None:
+        if self.config.api_key is not None:
+            value = self.config.api_key.get_secret_value()
+            if value:
+                return value
+
+        provider_data = self.get_request_provider_data()
+        if provider_data is not None and provider_data.passthrough_api_key:
+            return str(provider_data.passthrough_api_key.get_secret_value())
+
+        return 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
+
+    def _build_request_headers(self) -> dict[str, str]:
+        """Combine auth + forwarded headers for the downstream request."""
+        headers: dict[str, str] = {"Content-Type": "application/json"}
+
+        # forwarded headers go first so config api_key can't be overwritten
+        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:
+        shield = await self.shield_store.get_shield(GetShieldRequest(identifier=request.shield_id))
+        if not shield:
+            raise ValueError(f"Shield {request.shield_id} not found")
+
+        # convert messages to a single string for the moderation payload
+        texts: list[str] = []
+        for msg in request.messages:
+            content = msg.get("content", "") if isinstance(msg, dict) else getattr(msg, "content", "")
+            if isinstance(content, str):
+                texts.append(content)
+            elif isinstance(content, list):
+                # content parts - extract text parts
+                for part in content:
+                    if isinstance(part, dict) and part.get("type") == "text":
+                        texts.append(part.get("text", ""))
+                    elif isinstance(part, str):
+                        texts.append(part)
+
+        if not texts:
+            return RunShieldResponse(violation=None)
+
+        moderation_input = texts if len(texts) != 1 else texts[0]
+
+        payload = {
+            "input": moderation_input,
+            "model": shield.provider_resource_id or request.shield_id,
+        }
+
+        base_url = str(self.config.base_url).rstrip("/")
+        url = f"{base_url}/moderations"
+
+        headers = self._build_request_headers()
+
+        data = await self._post_moderation(url, payload, headers)
+        return self._parse_moderation_response(data)
+
+    async def run_moderation(self, request: RunModerationRequest) -> ModerationObject:
+        """Forward directly to downstream /v1/moderations instead of going through run_shield."""
+        inputs = request.input if isinstance(request.input, list) else [request.input]
+
+        payload: dict[str, str | list[str]] = {"input": request.input}
+        if request.model is not None:
+            payload["model"] = request.model
+
+        base_url = str(self.config.base_url).rstrip("/")
+        url = f"{base_url}/moderations"
+
+        headers = self._build_request_headers()
+
+        data = await self._post_moderation(url, payload, headers)
+
+        # parse downstream response into our ModerationObject
+        results_data = data.get("results")
+        if not isinstance(results_data, list):
+            raise RuntimeError("Downstream safety service returned malformed response (missing or invalid 'results')")
+        results: list[ModerationObjectResults] = []
+
+        for result in results_data:
+            if not isinstance(result, dict):
+                raise RuntimeError("Downstream safety service returned malformed result entry (expected object)")
+            flagged = result.get("flagged", False)
+            categories = result.get("categories") or {}
+            category_scores = result.get("category_scores") or {}
+
+            results.append(
+                ModerationObjectResults(
+                    flagged=flagged,
+                    categories=categories,
+                    category_scores=category_scores,
+                    category_applied_input_types=result.get("category_applied_input_types"),
+                    user_message=None,
+                    metadata={},
+                )
+            )
+
+        if len(results) != len(inputs):
+            raise RuntimeError(f"Downstream safety service returned {len(results)} results for {len(inputs)} inputs")
+
+        return ModerationObject(
+            id=data.get("id", f"modr-{uuid.uuid4()}"),
+            model=data.get("model", request.model or ""),
+            results=results,
+        )
+
+    async def _post_moderation(self, url: str, payload: dict[str, Any], headers: dict[str, str]) -> dict[str, Any]:
+        try:
+            response = await self._client.post(url, json=payload, headers=headers)
+            response.raise_for_status()
+        except httpx.TimeoutException as e:
+            raise RuntimeError("Failed to reach downstream safety service: request timed out") from e
+        except httpx.ConnectError as e:
+            raise RuntimeError("Failed to reach downstream safety service: connection failed") from e
+        except httpx.HTTPStatusError as e:
+            if 400 <= e.response.status_code < 500:
+                raise ValueError(
+                    f"Downstream safety service rejected the request (HTTP {e.response.status_code})"
+                ) from e
+            raise RuntimeError(f"Downstream safety service returned HTTP {e.response.status_code}") from e
+        except httpx.RequestError as e:
+            raise RuntimeError("Failed to reach downstream safety service: unexpected request error") from e
+
+        try:
+            raw = response.json()
+        except (ValueError, UnicodeDecodeError) as e:
+            raise RuntimeError(
+                f"Downstream safety service returned non-JSON response (HTTP {response.status_code})"
+            ) from e
+
+        if not isinstance(raw, dict):
+            raise RuntimeError("Downstream safety service returned invalid response (expected JSON object)")
+
+        return raw
+
+    def _parse_moderation_response(self, data: dict[str, Any]) -> RunShieldResponse:
+        """Convert a /v1/moderations JSON response into RunShieldResponse."""
+        results = data.get("results")
+        if not isinstance(results, list):
+            raise RuntimeError("Downstream safety service returned malformed response (missing or invalid 'results')")
+        if not results:
+            raise RuntimeError("Downstream safety service returned empty results")
+
+        for result in results:
+            if not isinstance(result, dict):
+                raise RuntimeError("Downstream safety service returned malformed result entry (expected object)")
+            if not result.get("flagged", False):
+                continue
+
+            categories = result.get("categories") or {}
+            flagged_categories = [cat for cat, flagged in categories.items() if flagged]
+            violation_type = flagged_categories[0] if flagged_categories else "unsafe"
+
+            return RunShieldResponse(
+                violation=SafetyViolation(
+                    violation_level=ViolationLevel.ERROR,
+                    user_message="Content was flagged by the safety service.",
+                    metadata={"violation_type": violation_type},
+                )
+            )
+
+        return RunShieldResponse(violation=None)