NVIDIA · rapids-bot · Sep 1, 2026 · Aug 31, 2026 · Aug 31, 2026 · Aug 31, 2026
@@ -141,15 +141,29 @@ The automatic memory wrapper agent supports several configuration parameters:
 
 ### Multi-Tenant Memory Isolation
 
-User ID is automatically extracted at runtime for memory isolation via:
-1. `SessionManager.session(user_id=...)` - For production with custom auth middleware (recommended)
-2. `X-User-ID` HTTP header - Illustrative/testing only; assumes a trusted upstream proxy authenticates the request and injects this header
-3. Console front end `user_id` - Defaults to `"nat_run_user_id"` for `nat run`
+The automatic memory wrapper reads only the identity resolved by the runtime session. Resolve identity through
+authenticated front-end credentials, `SessionManager.session(user_id=...)`, or the console front end `user_id` (which
+defaults to `"nat_run_user_id"` for `nat run`). Memory operations fail closed when no identity is available.
+
+For local testing or an isolated deployment behind an authenticating reverse proxy, you can explicitly opt in to a
+trusted upstream identity header:
+
+```yaml
+general:
+  front_end:
+    _type: fastapi
+    identity_header: X-User-ID
+```
+
+Do not enable this setting unless clients cannot reach `nat serve` directly, the proxy authenticates every request and
+overwrites the header, the toolkit port is not published outside the trusted backend network, and every container on
+that network is trusted. A client-supplied identity header is not authentication.
 
 Conversation-aware memory backends can also use `conversation_id` to isolate separate conversations for the same user.
 For `nat run`, pass `--conversation_id` when testing independent memory conversations from the CLI.
 
-> Never treat a client-supplied `X-User-ID` header as authentication.
+When configured, the header is authoritative for HTTP and WebSocket requests. Missing, empty, and repeated values are
+rejected, and other credentials cannot override it.
 
 For detailed configuration and usage examples, refer to the `examples/agents/auto_memory_wrapper/README.md` guide.
 

@@ -163,11 +163,14 @@ The wrapper automatically:
 The automatic memory wrapper agent provides multi-tenant support through runtime user ID extraction. Configure user IDs
 through the front end or session runtime, not the `auto_memory_agent` workflow block.
 
-### User ID Extraction Priority
+### User ID Resolution
 
-1. **`SessionManager.session(user_id=...)`** - For production with custom auth middleware (recommended)
-2. **`X-User-ID` HTTP header** - Illustrative/testing only; assumes a hypothetical trusted upstream proxy authenticates the request and injects the header
-3. **Console front end `user_id`** - Defaults to `"nat_run_user_id"` for `nat run`
+The wrapper reads only the user ID resolved by the runtime session. It does not read request headers directly and does
+not use a shared fallback identity. If a memory operation is enabled and the runtime has not resolved an identity, the
+request fails instead of sharing memory across unauthenticated users.
+
+Resolve identity with authenticated front-end credentials, `SessionManager.session(user_id=...)`, or the console front
+end `user_id` (which defaults to `"nat_run_user_id"` for `nat run`).
 
 Conversation-aware memory backends can also use `conversation_id` to isolate separate conversations for the same user.
 For Zep Cloud, if no conversation ID is supplied, the integration uses a deterministic per-user default thread.
@@ -196,9 +199,27 @@ async def handle_request(request):
     return result
 ```
 
-### Testing: X-User-ID Header
+### Local Testing or a Trusted Upstream Identity Header
+
+The FastAPI front end can explicitly trust one upstream identity header:
+
+```yaml
+general:
+  front_end:
+    _type: fastapi
+    identity_header: X-User-ID
+```
+
+Use this setting only for local testing or behind an authenticating reverse proxy. It is secure only when all of the following are true:
+
+- Clients cannot connect to `nat serve` directly; only the trusted proxy can reach its listening port.
+- The proxy authenticates every request and overwrites `X-User-ID` with the authenticated principal. It must not preserve, append, or forward a client-supplied value.
+- `nat serve` is network isolated in such a way that all incoming network traffic is from the proxy.
+
+When configured, the header is authoritative for HTTP and WebSocket requests. A missing, empty, or repeated header is rejected. Other connection credentials and WebSocket auth messages cannot replace it. The setting cannot be combined
+with `accepted_identity_credentials` or `identity_authentication`.
 
-For quick testing without custom middleware:
+For local testing, this can be simulated with:
 
 ```bash
 curl -X POST http://localhost:8000/chat \
@@ -327,14 +348,13 @@ workflow:
 
 ## Important Notes
 
-1. **User ID is runtime/front-end scoped** - Set via `SessionManager.session(user_id=...)`, `X-User-ID`, or
-   `nat run --user_id`. When no runtime identity or header is available, the wrapper falls back to `"default_user"`.
-   Use this fallback for development and testing only. Production deployments must require an authenticated runtime
-   identity.
+1. **User ID is runtime/front-end scoped** - Set through authenticated front-end identity resolution,
+   `SessionManager.session(user_id=...)`, an explicitly configured trusted `identity_header`, or `nat run --user_id`.
+   Memory operations fail when no identity is available.
 2. **Memory backends are interchangeable** - Works with any implementation of `MemoryEditor` interface
 3. **No memory tools needed** - The wrapped agent does not need explicit memory tools configured
 4. **Transparent to inner agent** - The wrapped agent is unaware of memory operations
-5. **X-User-ID** - We used this header for illustrative purposes only. Do not rely on it for authentication in production.
+5. **Trusted identity headers are opt-in** - Never enable `identity_header` on a server that untrusted clients can reach directly.
 
 ---
 

@@ -112,6 +112,19 @@ general:
 
 Each JWT provider requires `issuer_url`, `jwks_uri`, and `audience`. It can also require `scopes` and configure `timeout` and `leeway`. Add another named provider and select it in `identity_authentication` to accept JWT tokens from another issuer. The issuer claim selects the matching provider; an unknown issuer or a failed signature, time, audience, or scope check returns an authentication error and does not restore workflow state. If `identity_authentication` is omitted, JWT tokens retain the existing decode-only behavior. Configuring `identity_authentication` while excluding `jwt` from `accepted_identity_credentials` is invalid.
 
+For local testing or a deployment where an authenticating reverse proxy is the only service that can reach `nat serve`, the FastAPI front end can instead trust an upstream identity header:
+
+```yaml
+general:
+  front_end:
+    _type: fastapi
+    identity_header: X-User-ID
+```
+
+This setting is secure only if the proxy authenticates every connection, overwrites any client-supplied value, and the server is in an isolated network environment. When enabled, the header is authoritative: the connection is rejected if it is missing, empty, or repeated,
+and an `auth_message` cannot replace the resolved identity. `identity_header` cannot be combined with
+`accepted_identity_credentials` or `identity_authentication`. Do not use this technique if `nat serve` is reachable by untrusted clients.
+
 ## Auth Message
 This message allows clients to authenticate over a WebSocket connection when header-based or
 cookie-based authentication is not feasible (e.g., browser WebSocket APIs that do not support custom headers).

@@ -130,11 +130,11 @@ See `config_zep.yml` for comprehensive parameter examples.
 User ID is extracted at runtime for memory isolation. Configure it through the front end or session runtime, not the
 `auto_memory_agent` workflow block.
 
-### User ID Extraction Priority
+### User ID Resolution
 
-1. **`SessionManager.session(user_id=...)`** - For production with custom auth middleware (recommended)
-2. **`X-User-ID` HTTP header** - Illustrative/testing only; assumes a hypothetical trusted upstream proxy authenticates the request and injects the header
-3. **Console front end `user_id`** - Defaults to `"nat_run_user_id"` for `nat run`
+The wrapper reads only the identity resolved by the runtime session. Use authenticated front-end credentials,
+`SessionManager.session(user_id=...)`, or the console front end `user_id` (which defaults to `"nat_run_user_id"` for
+`nat run`). Memory operations fail closed when no identity is available.
 
 Conversation-aware memory backends can also use `conversation_id` to isolate separate conversations for the same user.
 For Zep Cloud, if no conversation ID is supplied, the integration uses a deterministic per-user default thread.
@@ -163,9 +163,23 @@ async def handle_request(request):
     return result
 ```
 
-### Testing: X-User-ID Header
+### Local Testing or a Trusted Upstream Identity Header
 
-For quick testing without custom middleware:
+To explicitly trust an identity header, configure the FastAPI front end:
+
+```yaml
+general:
+  front_end:
+    _type: fastapi
+    identity_header: X-User-ID
+```
+
+Use this only for local testing or when `nat serve` is isolated behind an authenticating reverse proxy. Clients must
+not be able to reach the toolkit service directly. The proxy must overwrite the header after authentication, the
+toolkit port must not be published outside the trusted backend network, and every container on that network must be
+trusted.
+
+After configuring that trust boundary:
 
 ```bash
 curl -X POST http://localhost:8000/chat \
@@ -175,7 +189,8 @@ curl -X POST http://localhost:8000/chat \
   -d '{"messages": [{"role": "user", "content": "Hello!"}]}'
 ```
 
-The example usage of the `X-User-ID` header is for illustrative purposes only; do not accept this header directly from untrusted clients.
+Never accept this header directly from untrusted clients. When `identity_header` is configured, it is authoritative;
+missing, empty, or repeated values are rejected and other credentials cannot override it.
 
 
 ### Local Development: Console User and Conversation IDs
@@ -216,9 +231,11 @@ workflow:
 
 ## Important Notes
 
-1. **User ID is runtime/front-end scoped** - Set via `SessionManager.session(user_id=...)`, `X-User-ID`, or `nat run --user_id`
+1. **User ID is runtime/front-end scoped** - Set through authenticated front-end identity resolution,
+   `SessionManager.session(user_id=...)`, an explicitly configured trusted `identity_header`, or `nat run --user_id`
 2. **Memory backends are interchangeable** - Works with any implementation of `MemoryEditor` interface
-3. `X-User-ID` HTTP header - Illustrative/testing only; assumes a trusted upstream proxy authenticates the request and injects this header
+3. **Trusted headers require network enforcement** - Never expose a trusted identity-header deployment directly to
+   untrusted clients or containers
 
 ## Examples
 

@@ -23,7 +23,6 @@
 from nat.data_models.authentication import AuthResult
 from nat.data_models.authentication import BasicAuthCred
 from nat.data_models.authentication import BearerTokenCred
-from nat.runtime.session import SESSION_COOKIE_NAME
 
 
 class HTTPBasicAuthProvider(AuthProviderBase):
@@ -46,13 +45,10 @@ async def authenticate(self, user_id: str | None = None, **kwargs) -> AuthResult
 
         context = Context.get()
 
-        if user_id is None and hasattr(context, "metadata") and hasattr(
-                context.metadata, "cookies") and context.metadata.cookies is not None:
-            session_id = context.metadata.cookies.get(SESSION_COOKIE_NAME, None)
-            if not session_id:
-                raise RuntimeError("Authentication failed. No session ID found. Cannot identify user.")
-
-            user_id = session_id
+        if user_id is None:
+            user_id = context.user_id
+        if not user_id:
+            raise RuntimeError("Authentication failed. No resolved user identity is available.")
 
         if user_id and user_id in self._authenticated_tokens:
             return self._authenticated_tokens[user_id]

@@ -31,7 +31,6 @@
 from nat.data_models.authentication import AuthFlowType
 from nat.data_models.authentication import AuthResult
 from nat.data_models.authentication import BearerTokenCred
-from nat.runtime.session import SESSION_COOKIE_NAME
 
 logger = logging.getLogger(__name__)
 
@@ -91,13 +90,10 @@ def _set_custom_auth_callback(self,
 
     async def authenticate(self, user_id: str | None = None, **kwargs) -> AuthResult:
         context = Context.get()
-        if user_id is None and hasattr(context, "metadata") and hasattr(
-                context.metadata, "cookies") and context.metadata.cookies is not None:
-            session_id = context.metadata.cookies.get(SESSION_COOKIE_NAME, None)
-            if not session_id:
-                raise RuntimeError("Authentication failed. No session ID found. Cannot identify user.")
-
-            user_id = session_id
+        if user_id is None:
+            user_id = context.user_id
+        if not user_id:
+            raise RuntimeError("Authentication failed. No resolved user identity is available.")
 
         if user_id:
             # Try to retrieve from token storage

@@ -173,6 +173,13 @@ def _from_session_cookie(cls, cookie: str) -> "UserInfo":
     def _from_api_key(cls, api_key: str) -> "UserInfo":
         return cls(api_key=SecretStr(api_key))
 
+    @classmethod
+    def _from_identity_header(cls, header_name: str, header_value: str) -> "UserInfo":
+        """Create a user from an identity asserted by a trusted upstream proxy."""
+        instance: UserInfo = cls()
+        instance._set_user_id(f"trusted-header:{header_name.lower()}\x1f{header_value}")
+        return instance
+
     @classmethod
     def _from_jwt(cls, jwt_info: JwtUserInfo, *, issuer_scoped: bool = False) -> "UserInfo":
         identity: str | None = jwt_info.identity_claim

@@ -15,6 +15,7 @@
 
 import logging
 import os
+import re
 import sys
 import typing
 from datetime import datetime
@@ -295,8 +296,32 @@ class CrossOriginResourceSharing(BaseModel):
                      "If omitted, JWT claims retain the existing decode-only behavior."),
     )
 
+    identity_header: str | None = Field(
+        default=None,
+        description=("Name of an HTTP header containing an identity asserted by a trusted upstream proxy. "
+                     "This is disabled by default and must only be enabled when untrusted clients cannot reach "
+                     "the server directly and the proxy removes any client-supplied value before setting it."),
+    )
+
+    @field_validator("identity_header")
+    @classmethod
+    def validate_identity_header(cls, identity_header: str | None) -> str | None:
+        if identity_header is None:
+            return None
+        identity_header = identity_header.strip()
+        if not identity_header:
+            raise ValueError("identity_header must not be empty")
+        if re.fullmatch(r"[!#$%&'*+\-.^_`|~0-9A-Za-z]+", identity_header) is None:
+            raise ValueError("identity_header must be a valid HTTP header name")
+        return identity_header
+
     @model_validator(mode="after")
     def validate_jwt_identity_policy(self) -> typing.Self:
+        if self.identity_header is not None:
+            if self.accepted_identity_credentials is not None:
+                raise ValueError("identity_header cannot be combined with accepted_identity_credentials")
+            if self.identity_authentication:
+                raise ValueError("identity_header cannot be combined with identity_authentication")
         if (self.identity_authentication and self.accepted_identity_credentials is not None
                 and "jwt" not in self.accepted_identity_credentials):
             raise ValueError(

@@ -27,12 +27,14 @@
 from fastapi import Response
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
+from starlette.responses import JSONResponse
 
 from nat.builder.evaluator import EvaluatorInfo
 from nat.builder.workflow_builder import WorkflowBuilder
 from nat.builder.workflow_builder import WorkflowEvalBuilderBase
 from nat.data_models.config import Config
 from nat.runtime.session import SessionManager
+from nat.runtime.user_manager import IdentityHeaderError
 from nat.utils.log_utils import setup_logging
 
 from .auth_flow_handlers.http_flow_handler import HTTPAuthenticationFlowHandler
@@ -126,6 +128,10 @@ async def lifespan(starting_app: FastAPI):
 
         nat_app = FastAPI(lifespan=lifespan)
 
+        @nat_app.exception_handler(IdentityHeaderError)
+        async def identity_header_error_handler(_request: Request, exc: IdentityHeaderError) -> JSONResponse:
+            return JSONResponse(status_code=401, content={"detail": str(exc)})
+
         # Configure app CORS.
         self.set_cors_config(nat_app)
 

@@ -66,6 +66,7 @@
 from nat.front_ends.fastapi.step_adaptor import StepAdaptor
 from nat.runtime.session import SessionManager
 from nat.runtime.user_manager import IdentityCredentialNotAcceptedError
+from nat.runtime.user_manager import IdentityHeaderError
 from nat.runtime.user_manager import JwtVerificationError
 from nat.runtime.user_manager import UserManager
 
@@ -96,13 +97,15 @@ def __init__(
         worker: "FastApiFrontEndPluginWorker",
         accepted_identity_credentials: typing.Collection[IdentityCredentialType] | None = None,
         jwt_validators: typing.Mapping[str, BearerTokenValidator] | None = None,
+        identity_header: str | None = None,
     ):
         self._socket: WebSocket = socket
         self._session_manager: SessionManager = session_manager
         self._step_adaptor: StepAdaptor = step_adaptor
         self._worker: FastApiFrontEndPluginWorker = worker
         self._accepted_identity_credentials = accepted_identity_credentials
         self._jwt_validators = jwt_validators
+        self._identity_header = identity_header
 
         self._message_validator: MessageValidator = MessageValidator()
         self._running_workflow_task: asyncio.Task | None = None
@@ -192,8 +195,9 @@ async def __aenter__(self) -> "WebSocketMessageHandler":
                 self._socket,
                 accepted_identity_credentials=self._accepted_identity_credentials,
                 jwt_validators=self._jwt_validators,
+                identity_header=self._identity_header,
             )
-        except (IdentityCredentialNotAcceptedError, JwtVerificationError) as exc:
+        except (IdentityCredentialNotAcceptedError, IdentityHeaderError, JwtVerificationError) as exc:
             self._connection_rejected = True
             response = WebSocketAuthResponseMessage(
                 status=AuthMessageStatus.ERROR,
@@ -313,6 +317,18 @@ async def _process_auth_message(self, message: WebSocketAuthMessage) -> None:
                 self._flow_handler.set_oauth_mode(message.payload.mode)
             return
 
+        if self._identity_header is not None:
+            response = WebSocketAuthResponseMessage(
+                status=AuthMessageStatus.ERROR,
+                payload=Error(
+                    code=ErrorTypes.USER_AUTH_ERROR,
+                    message="Authentication failed",
+                    details="WebSocket auth messages cannot replace an identity asserted by a trusted header",
+                ),
+            )
+            await self._socket.send_json(response.model_dump())
+            return
+
         identity_resolved = False
         try:
             user_info: UserInfo = await UserManager.from_auth_payload_with_verification(

@@ -114,6 +114,7 @@ async def _websocket_endpoint(websocket: WebSocket):
                 worker,
                 accepted_identity_credentials=worker.front_end_config.accepted_identity_credentials,
                 jwt_validators=jwt_validators,
+                identity_header=worker.front_end_config.identity_header,
         ) as handler:
             origin = websocket.headers.get("origin")
             allowed_origins = worker.front_end_config.cors.allow_origins or []