feat(agents): authenticate deployed agents via service-principal sidecar

packages/nemo_platform_plugin/src/nemo_platform_plugin/auth.py

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Auth helpers exposed to plugins.
+
+Plugins must not import ``nmp_common`` directly. This module wraps the pieces of
+the platform auth configuration that plugins need. The underlying auth config
+lives in ``nmp_common`` (only present in the platform process image), so it is
+imported lazily and failures degrade to "disabled" rather than raising in
+environments without it.
+"""
+
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def platform_auth_enabled() -> bool:
+    """Return whether platform authentication is enabled.
+
+    Returns ``False`` on any failure to resolve the auth config. The realistic
+    failure is ``ImportError``: ``nmp_common`` ships only in the platform process
+    image, so when this package is used standalone (outside the platform) there
+    is no auth config and "disabled" is the correct answer.
+
+    Other failures are effectively unreachable in the context that matters here
+    (the deployment controller, which runs *inside* the platform image): a
+    missing config file resolves to defaults (``enabled=False``) rather than
+    raising, and a malformed/invalid config file would have already crashed the
+    platform service at startup before any deployment is reconciled. The config
+    read is cached from that successful startup load. We therefore accept the
+    narrow, largely theoretical fail-open window rather than propagate and block
+    deployments on a transient/unexpected error.
+    """
+    try:
+        from nmp.common.config import get_auth_config
+
+        return bool(get_auth_config().enabled)
+    except Exception:
+        logger.debug("Could not resolve auth config; assuming auth disabled", exc_info=True)
+        return False

packages/nmp_common/src/nmp/common/auth/workload_proxy/main.py

@@ -0,0 +1,162 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Service-principal auth-proxy sidecar.
+
+Runs inside a deployed workload's pod as a loopback forwarder. A co-located
+workload whose HTTP client we do not control (e.g. a NAT agent calling the
+Inference Gateway) points its platform base URL at this proxy
+(``http://127.0.0.1:<port>``) and sends no credentials of its own. The proxy
+stamps a service-principal identity header (``X-NMP-Principal-Id: service:<name>``)
+on every forwarded request, which the platform authorizes via the ServiceSystem
+role. This is the same static service-identity the platform's own SDK clients
+use (``get_platform_sdk(as_service=...)``); the proxy exists only for workloads
+that cannot set the header themselves.
+
+Started via ``nemo services run --sidecars auth-proxy``.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import httpx
+import uvicorn
+from fastapi import FastAPI, Request
+from fastapi.responses import StreamingResponse
+
+logger = logging.getLogger(__name__)
+
+# Loopback host + port the proxy listens on. The workload targets this address.
+AUTH_PROXY_HOST_ENVVAR = "NMP_AUTH_PROXY_HOST"
+AUTH_PROXY_PORT_ENVVAR = "NMP_AUTH_PROXY_PORT"
+# Service-principal name stamped on forwarded requests (e.g. "agents").
+AUTH_PROXY_PRINCIPAL_ENVVAR = "NMP_AUTH_PROXY_PRINCIPAL"
+DEFAULT_AUTH_PROXY_HOST = "127.0.0.1"
+DEFAULT_AUTH_PROXY_PORT = 8090
+
+_READ_TIMEOUT_ENVVAR = "NMP_AUTH_PROXY_READ_TIMEOUT"
+_PRINCIPAL_ID_HEADER = "x-nmp-principal-id"
+
+# Minimal request-header sanitization. We only drop what would be actively wrong:
+# - the workload's own credential / principal header (we set the identity), so it
+#   can't be spoofed or conflict with what we stamp;
+# - host and content-length, which httpx recomputes for the upstream request
+#   (a stale value corrupts routing / the body).
+_STRIP_REQUEST_HEADERS = frozenset(
+    {
+        "host",
+        "content-length",
+        "authorization",
+        _PRINCIPAL_ID_HEADER,
+    }
+)
+# We stream the response, so the upstream's framing headers no longer apply.
+_STRIP_RESPONSE_HEADERS = frozenset(
+    {
+        "content-length",
+        "transfer-encoding",
+    }
+)
+
+
+def _upstream_base_url() -> str:
+    """Return the platform base URL to forward to (env override or platform config)."""
+    from nemo_platform_plugin.config import get_platform_config
+
+    return (os.environ.get("NEMO_BASE_URL") or os.environ.get("NMP_BASE_URL") or get_platform_config().base_url).rstrip(
+        "/"
+    )
+
+
+def build_app(*, base_url: str, principal: str) -> FastAPI:
+    """Build the forwarding FastAPI app for the given upstream and service principal."""
+    principal_id = principal if principal.startswith("service:") else f"service:{principal}"
+    read_timeout = float(os.environ.get(_READ_TIMEOUT_ENVVAR, "300"))
+    timeout = httpx.Timeout(connect=10.0, read=read_timeout, write=60.0, pool=10.0)
+    client = httpx.AsyncClient(base_url=base_url, timeout=timeout, follow_redirects=False)
+
+    @asynccontextmanager
+    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
+        try:
+            yield
+        finally:
+            await client.aclose()
+
+    app = FastAPI(title="nmp-auth-proxy", lifespan=lifespan)
+
+    @app.get("/healthz")
+    async def healthz() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.api_route(
+        "/{path:path}",
+        methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"],
+    )
+    async def forward(request: Request, path: str) -> StreamingResponse:
+        headers = {k: v for k, v in request.headers.items() if k.lower() not in _STRIP_REQUEST_HEADERS}
+        headers[_PRINCIPAL_ID_HEADER] = principal_id
+        url = httpx.URL(path="/" + path, query=request.url.query.encode("utf-8"))
+        body = await request.body()
+        upstream = client.build_request(request.method, url, headers=headers, content=body)
+        response = await client.send(upstream, stream=True)
+
+        async def _body() -> AsyncIterator[bytes]:
+            # The finally runs on normal completion, exception, and client
+            # disconnect (Starlette closes the generator), so this is the only
+            # cleanup the response needs.
+            try:
+                async for chunk in response.aiter_raw():
+                    yield chunk
+            finally:
+                await response.aclose()
+
+        resp_headers = {k: v for k, v in response.headers.items() if k.lower() not in _STRIP_RESPONSE_HEADERS}
+        return StreamingResponse(
+            _body(),
+            status_code=response.status_code,
+            headers=resp_headers,
+        )
+
+    return app
+
+
+def run(parent_stop_signal: threading.Event | None = None) -> None:
+    """Sidecar entrypoint. Serves the loopback auth-proxy until stopped."""
+    base_url = _upstream_base_url()
+    principal = os.environ.get(AUTH_PROXY_PRINCIPAL_ENVVAR)
+    if not principal:
+        raise RuntimeError(f"{AUTH_PROXY_PRINCIPAL_ENVVAR} is required for the auth-proxy sidecar")
+    host = os.environ.get(AUTH_PROXY_HOST_ENVVAR, DEFAULT_AUTH_PROXY_HOST)
+    port = int(os.environ.get(AUTH_PROXY_PORT_ENVVAR, str(DEFAULT_AUTH_PROXY_PORT)))
+    app = build_app(base_url=base_url, principal=principal)
+
+    config = uvicorn.Config(app, host=host, port=port, log_level="info", access_log=False)
+    server = uvicorn.Server(config)
+
+    logger.info("Starting auth-proxy sidecar on %s:%s -> %s (principal=service:%s)", host, port, base_url, principal)
+    if parent_stop_signal is None:
+        server.run()
+        return
+
+    thread = threading.Thread(target=server.run, name="auth-proxy-uvicorn", daemon=True)
+    thread.start()
+    try:
+        while not parent_stop_signal.is_set():
+            parent_stop_signal.wait(timeout=1)
+    finally:
+        server.should_exit = True
+        thread.join(timeout=10)
+        logger.info("auth-proxy sidecar stopped")
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO").upper(),
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+    run()

packages/nmp_common/tests/auth/test_workload_proxy.py

@@ -0,0 +1,89 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the service-principal auth-proxy sidecar forwarder."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import httpx
+import respx
+from fastapi.testclient import TestClient
+from nmp.common.auth.workload_proxy.main import build_app
+
+
+@respx.mock
+def test_forward_stamps_service_principal_and_preserves_path() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.post(f"{upstream}/apis/inference-gateway/v2/workspaces/default/openai/-/v1/chat/completions").mock(
+        return_value=httpx.Response(200, json={"ok": True})
+    )
+    app = build_app(base_url=upstream, principal="agents")
+    client = TestClient(app)
+
+    resp = client.post(
+        "/apis/inference-gateway/v2/workspaces/default/openai/-/v1/chat/completions",
+        json={"model": "m", "messages": []},
+        headers={"authorization": "Bearer not-used"},
+    )
+
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True}
+    assert route.called
+    sent = route.calls.last.request
+    # The proxy sets the service-principal identity and drops the placeholder auth.
+    assert sent.headers["x-nmp-principal-id"] == "service:agents"
+    assert "authorization" not in {k.lower() for k in sent.headers}
+
+
+@respx.mock
+def test_forward_normalizes_bare_principal_name() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(200, json={}))
+    # Already-prefixed principal is passed through unchanged.
+    app = build_app(base_url=upstream, principal="service:models")
+    client = TestClient(app)
+    client.get("/apis/entities/v2/workspaces")
+    assert route.calls.last.request.headers["x-nmp-principal-id"] == "service:models"
+
+
+@respx.mock
+def test_forward_passes_through_upstream_status() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(403, json={"detail": "no"}))
+    app = build_app(base_url=upstream, principal="agents")
+    client = TestClient(app)
+
+    resp = client.get("/apis/entities/v2/workspaces")
+    assert resp.status_code == 403
+
+
+def test_healthz_does_not_require_upstream() -> None:
+    app = build_app(base_url="http://nemo-platform-api:8080", principal="agents")
+    client = TestClient(app)
+    resp = client.get("/healthz")
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok"}
+
+
+def test_lifespan_closes_upstream_client() -> None:
+    # Entering TestClient as a context manager runs the lifespan; the shared
+    # httpx client's connection pool must be closed on shutdown.
+    created: list[httpx.AsyncClient] = []
+    real_async_client = httpx.AsyncClient
+
+    def _tracking_client(*args, **kwargs) -> httpx.AsyncClient:
+        client = real_async_client(*args, **kwargs)
+        created.append(client)
+        return client
+
+    with patch("nmp.common.auth.workload_proxy.main.httpx.AsyncClient", side_effect=_tracking_client):
+        app = build_app(base_url="http://nemo-platform-api:8080", principal="agents")
+
+    assert len(created) == 1
+    upstream_client = created[0]
+    with TestClient(app) as test_client:
+        assert test_client.get("/healthz").status_code == 200
+        assert upstream_client.is_closed is False
+    assert upstream_client.is_closed is True

packages/nmp_platform_runner/src/nmp/platform_runner/registry.py

@@ -37,6 +37,7 @@
 
 AVAILABLE_SIDECARS: dict[str, str] = {
     "adapters": "nmp.core.models.sidecars.adapters.main:run",
+    "auth-proxy": "nmp.common.auth.workload_proxy.main:run",
 }
 
 SERVICE_SIDECAR_DEPENDENCIES: dict[str, set[str]] = {

plugins/nemo-agents/src/nemo_agents_plugin/runner/deployments_backend.py

@@ -30,6 +30,7 @@
 )
 from nemo_agents_plugin.runner.backend import DeploymentInfo, ExternalLog, LogLocation, RunnerBackend
 from nemo_agents_plugin.utils import get_base_url, get_internal_base_url
+from nemo_deployments_plugin.auth_proxy import auth_proxy_port
 from nemo_deployments_plugin.entities import (
     ConfigFile,
     Container,
@@ -42,6 +43,7 @@
     VolumeMount,
 )
 from nemo_platform.resources.entities import AsyncEntitiesResource
+from nemo_platform_plugin.auth import platform_auth_enabled
 from nemo_platform_plugin.config import LOOPBACK_ADDRESSES
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityNotFoundError
 from nemo_platform_plugin.sdk_provider import get_async_platform_sdk
@@ -53,6 +55,8 @@
 _PLUGIN_WHEELS_VOLUME = "plugin-wheels"
 _PLUGIN_WHEELS_MOUNT = "/opt/nemo/plugin-wheels"
 _NAT_CONFIG_ENV = "NAT_CONFIG_PATH"
+_AUTH_PROXY_IDENTITY = "agents"
+
 
 # On delete, wait up to this long for the deployments controller to tear down the
 # container and remove the Deployment entity before we drop the DeploymentConfig.
@@ -199,6 +203,7 @@ def build_deployment_config(
     mode: DeploymentMode,
     plugin_wheels_init_image: str | None = None,
     labels: dict[str, str] | None = None,
+    auth_proxy_identity: str | None = None,
 ) -> DeploymentConfig:
     """Compile an agent into a long-running ``DeploymentConfig`` (Always).
 
@@ -295,6 +300,8 @@ def build_deployment_config(
         }
     )
 
+    # Request the auth-proxy sidecar via the DeploymentConfig flags; the
+    # deployments plugin compiles and injects it (and no-ops when auth is off).
     return DeploymentConfig(
         name=name,
         workspace=workspace,
@@ -307,6 +314,8 @@ def build_deployment_config(
                 ConfigFile(path=config_mount_path, content=nat_yaml),
             ],
             "restart_policy": "Always",
+            "auth_proxy_sidecar": auth_proxy_identity is not None,
+            "auth_proxy_sidecar_identity": auth_proxy_identity,
         }
     )
 
@@ -366,7 +375,19 @@ async def create_deployment(
         except UnreachableGatewayURLError as exc:
             logger.error("Refusing to deploy agent %r: %s", name, exc)
             return DeploymentInfo(name=name, status="failed", error=str(exc))
-        config = rewrite_config_base_urls(config, gateway)
+
+        # When platform auth is enabled, the agent carries no platform credential,
+        # so route its inference calls through a loopback auth-proxy sidecar (the
+        # deployments plugin compiles the sidecar from the auth_proxy flags). The
+        # agent targets the sidecar on localhost; the sidecar forwards to the
+        # platform with a service-principal identity header.
+        auth_proxy_identity: str | None = None
+        if platform_auth_enabled():
+            auth_proxy_identity = _AUTH_PROXY_IDENTITY
+            config = rewrite_config_base_urls(config, f"http://127.0.0.1:{auth_proxy_port()}")
+        else:
+            config = rewrite_config_base_urls(config, gateway)
+
         deployment_config = build_deployment_config(
             name=name,
             workspace=workspace,
@@ -380,6 +401,7 @@ async def create_deployment(
                 "nemo.agents/deployment": name,
                 "nemo.agents/mode": deployment_mode,
             },
+            auth_proxy_identity=auth_proxy_identity,
         )
         await entities.create(deployment_config)
         try: