feat(auth): add upstream header auth provider for gateway deployments (#5563)

# What does this PR do?

Adds a new `upstream_header` authentication provider that extracts user
identity from headers injected by an upstream gateway (Authorino, Istio,
or any reverse proxy that handles auth). This enables Llama Stack to
participate in gateway-fronted deployments without redundant token
validation, while still scoping all storage queries to the authenticated
user via the existing `AuthorizedSqlStore` pipeline.

The `AuthProvider` base class gains a `requires_http_bearer` property
(default `True`) so the `AuthenticationMiddleware` can skip Bearer token
extraction for providers that read identity from other sources. The
`authenticated_client_id` falls back to the principal when no token is
present, preserving quota tracking.

Configuration:
```yaml
server:
  auth:
    provider_config:
      type: upstream_header
      principal_header: x-auth-user-id        # required
      attributes_header: x-auth-attributes     # optional, JSON dict
```

## Test Plan

17 unit tests covering provider behavior and middleware integration:

```bash
uv run pytest tests/unit/server/test_auth_upstream_header.py -v
```

```
tests/unit/server/test_auth_upstream_header.py::test_valid_upstream_header_auth PASSED
tests/unit/server/test_auth_upstream_header.py::test_valid_upstream_header_auth_principal_only PASSED
tests/unit/server/test_auth_upstream_header.py::test_missing_principal_header PASSED
tests/unit/server/test_auth_upstream_header.py::test_invalid_attributes_json PASSED
tests/unit/server/test_auth_upstream_header.py::test_attributes_not_object PASSED
tests/unit/server/test_auth_upstream_header.py::test_no_bearer_token_required PASSED
tests/unit/server/test_auth_upstream_header.py::test_bearer_token_ignored PASSED
tests/unit/server/test_auth_upstream_header.py::test_no_attributes_header_configured PASSED
tests/unit/server/test_auth_upstream_header.py::test_case_insensitive_headers PASSED
tests/unit/server/test_auth_upstream_header.py::test_attributes_string_values_normalized PASSED
tests/unit/server/test_auth_upstream_header.py::test_error_message_includes_header_name PASSED
tests/unit/server/test_auth_upstream_header.py::test_authenticated_client_id_uses_principal PASSED
tests/unit/server/test_auth_upstream_header.py::test_provider_requires_http_bearer_false PASSED
tests/unit/server/test_auth_upstream_header.py::test_provider_validate_token_extracts_principal PASSED
tests/unit/server/test_auth_upstream_header.py::test_provider_validate_token_extracts_attributes PASSED
tests/unit/server/test_auth_upstream_header.py::test_provider_validate_token_missing_principal PASSED
tests/unit/server/test_auth_upstream_header.py::test_provider_validate_token_none_scope PASSED
```

All 35 existing auth tests continue to pass.

Signed-off-by: Sébastien Han <seb@redhat.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: leseb

src/llama_stack/core/datatypes.py

@@ -213,6 +213,7 @@ class AuthProviderType(StrEnum):
     GITHUB_TOKEN = "github_token"
     CUSTOM = "custom"
     KUBERNETES = "kubernetes"
+    UPSTREAM_HEADER = "upstream_header"
 
 
 class OAuth2TokenAuthConfig(BaseModel):
@@ -320,8 +321,30 @@ def validate_claims_mapping(cls, v):
         return v
 
 
+class UpstreamHeaderAuthConfig(BaseModel):
+    """Configuration for upstream header authentication.
+
+    Used when an upstream gateway (Authorino, Istio, or any reverse proxy) handles
+    authentication and injects user identity into request headers. Llama Stack trusts
+    these headers and extracts the principal and optional attributes from them.
+    """
+
+    type: Literal[AuthProviderType.UPSTREAM_HEADER] = AuthProviderType.UPSTREAM_HEADER
+    principal_header: str = Field(
+        description="HTTP header containing the authenticated user's identity (e.g. x-auth-user-id)",
+    )
+    attributes_header: str | None = Field(
+        default=None,
+        description="HTTP header containing JSON-encoded user attributes for access control (e.g. x-auth-attributes)",
+    )
+
+
 AuthProviderConfig = Annotated[
-    OAuth2TokenAuthConfig | GitHubTokenAuthConfig | CustomAuthConfig | KubernetesAuthProviderConfig,
+    OAuth2TokenAuthConfig
+    | GitHubTokenAuthConfig
+    | CustomAuthConfig
+    | KubernetesAuthProviderConfig
+    | UpstreamHeaderAuthConfig,
     Field(discriminator="type"),
 ]
 

src/llama_stack/core/server/auth.py

@@ -120,17 +120,20 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> Any:
                 return await self.app(scope, receive, send)
 
             # Handle authentication
-            headers = dict(scope.get("headers", []))
-            auth_header = headers.get(b"authorization", b"").decode()
+            if self.auth_provider.requires_http_bearer:
+                headers = dict(scope.get("headers", []))
+                auth_header = headers.get(b"authorization", b"").decode()
 
-            if not auth_header:
-                error_msg = self.auth_provider.get_auth_error_message(scope)
-                return await self._send_auth_error(send, error_msg)
+                if not auth_header:
+                    error_msg = self.auth_provider.get_auth_error_message(scope)
+                    return await self._send_auth_error(send, error_msg)
 
-            if not auth_header.startswith("Bearer "):
-                return await self._send_auth_error(send, "Invalid Authorization header format")
+                if not auth_header.startswith("Bearer "):
+                    return await self._send_auth_error(send, "Invalid Authorization header format")
 
-            token = auth_header.split("Bearer ", 1)[1]
+                token = auth_header.split("Bearer ", 1)[1]
+            else:
+                token = ""
 
             # Validate token and get access attributes
             try:
@@ -147,7 +150,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> Any:
 
             # Store the client ID in the request scope so that downstream middleware (like QuotaMiddleware)
             # can identify the requester and enforce per-client rate limits.
-            scope["authenticated_client_id"] = token
+            scope["authenticated_client_id"] = token or validation_result.principal
 
             # Store attributes in request scope
             scope["principal"] = validation_result.principal

src/llama_stack/core/server/auth_providers.py

@@ -20,6 +20,7 @@
     GitHubTokenAuthConfig,
     KubernetesAuthProviderConfig,
     OAuth2TokenAuthConfig,
+    UpstreamHeaderAuthConfig,
     User,
 )
 from llama_stack.log import get_logger
@@ -60,6 +61,15 @@ class AuthRequest(BaseModel):
 class AuthProvider(ABC):
     """Abstract base class for authentication providers."""
 
+    @property
+    def requires_http_bearer(self) -> bool:
+        """Whether this provider requires a Bearer token from the Authorization header.
+
+        Providers that extract identity from other sources (e.g. gateway-injected
+        headers) should override this to return False.
+        """
+        return True
+
     @abstractmethod
     async def validate_token(self, token: str, scope: Scope | None = None) -> User:
         """Validate a token and return access attributes."""
@@ -502,6 +512,70 @@ async def close(self) -> None:
         pass
 
 
+class UpstreamHeaderAuthProvider(AuthProvider):
+    """Authentication provider that extracts identity from upstream gateway headers.
+
+    Used when an upstream gateway (Authorino, Istio, or any reverse proxy) handles
+    authentication and injects user identity into request headers. This provider
+    trusts the headers and performs no token validation or outbound calls.
+    """
+
+    def __init__(self, config: UpstreamHeaderAuthConfig) -> None:
+        self.config = config
+
+    @property
+    def requires_http_bearer(self) -> bool:
+        return False
+
+    async def validate_token(self, token: str, scope: Scope | None = None) -> User:
+        if scope is None:
+            raise ValueError("Missing required authentication header: " + self.config.principal_header)
+
+        headers = dict(scope.get("headers", []))
+
+        # HTTP headers are case-insensitive; ASGI stores them as lowercase bytes
+        principal_key = self.config.principal_header.lower().encode()
+        principal_value = headers.get(principal_key)
+
+        if not principal_value:
+            raise ValueError("Missing required authentication header: " + self.config.principal_header)
+
+        principal = principal_value.decode()
+
+        attributes: dict[str, list[str]] | None = None
+        if self.config.attributes_header:
+            attributes_key = self.config.attributes_header.lower().encode()
+            attributes_value = headers.get(attributes_key)
+            if attributes_value:
+                import json
+
+                try:
+                    parsed = json.loads(attributes_value.decode())
+                except (json.JSONDecodeError, UnicodeDecodeError) as e:
+                    raise ValueError("Failed to parse authentication attributes header: invalid JSON") from e
+
+                if not isinstance(parsed, dict):
+                    raise ValueError("Failed to parse authentication attributes header: expected JSON object")
+
+                # Normalize values to list[str] to match the User.attributes type
+                attributes = {}
+                for k, v in parsed.items():
+                    if isinstance(v, list):
+                        attributes[k] = [str(item) for item in v]
+                    elif isinstance(v, str):
+                        attributes[k] = [v]
+                    else:
+                        attributes[k] = [str(v)]
+
+        return User(principal=principal, attributes=attributes)
+
+    async def close(self) -> None:
+        pass
+
+    def get_auth_error_message(self, scope: Scope | None = None) -> str:
+        return f"Authentication required. Upstream gateway must set the {self.config.principal_header} header"
+
+
 def create_auth_provider(config: AuthenticationConfig) -> AuthProvider:
     """Factory function to create the appropriate auth provider."""
     provider_config = config.provider_config
@@ -514,5 +588,7 @@ def create_auth_provider(config: AuthenticationConfig) -> AuthProvider:
         return GitHubTokenAuthProvider(provider_config)
     elif isinstance(provider_config, KubernetesAuthProviderConfig):
         return KubernetesAuthProvider(provider_config)
+    elif isinstance(provider_config, UpstreamHeaderAuthConfig):
+        return UpstreamHeaderAuthProvider(provider_config)
     else:
         raise ValueError(f"Unknown authentication provider config type: {type(provider_config)}")