fix(agents): scope agent deployments to creator via on-behalf-of (#1000)

* fix(agents): scope agent deployments to creator via on-behalf-of

Agent deployments previously authenticated as the service:agents principal,
which resolves to the ServiceSystem role (platform-wide wildcard access).
Scope a deployment's runtime platform access to the identity that created it
by delegating via on-behalf-of, instead of granting admin-level reach.

- workload-proxy sidecar: add optional NMP_AUTH_PROXY_ON_BEHALF_OF; stamp
  X-NMP-Principal-On-Behalf-Of on forwarded requests, and strip the inbound
  OBO header so a co-located workload cannot spoof the delegated identity.
- deployments plugin: add DeploymentConfig.auth_proxy_sidecar_on_behalf_of and
  wire it into the auth-proxy sidecar env (regenerated plugin OpenAPI spec).
- agents runner: thread the deployment creator (created_by) through the
  container backend into the sidecar OBO env at deploy time.
- inference-gateway: enforce delegated workspace access on the proxy path.
  For a delegated service principal, verify the on-behalf-of user holds the
  required inference permission in the target workspace (via the PDP evaluated
  as the delegated user). Non-delegated callers are unchanged, preserving the
  existing internal service bypass. Without this, the IGW route gate takes the
  service bypass and never narrows on OBO.

The control plane (controller reconcile + DeploymentConfig/Deployment entity
writes) intentionally remains service:agents.

Adds unit + integration coverage for the sidecar OBO stamping/spoof-stripping,
the DeploymentConfig wiring, the created_by threading, and IGW delegated
allow/deny behavior.

Signed-off-by: Ben McCown <bmccown@nvidia.com>

* fix(auth): harden auth-proxy OBO header handling

Address PR review feedback on the auth-proxy sidecar:

- Strip inbound X-NMP-Principal-On-Behalf-Of-Email and
  -Groups companion headers, not just the OBO id. The platform derives the
  delegated user's effective groups/email from these headers and feeds them to
  the PDP, so a co-located workload could otherwise pair our stamped OBO id with
  attacker-chosen groups/email and be authorized as those, defeating the
  scoping. Extend the spoof-strip tests to cover both companion headers in the
  configured and unconfigured paths.
- Do not log the delegated principal id in the sidecar startup line; log a
  boolean (delegated=<bool>) instead to keep creator identifiers out of logs.

Signed-off-by: Ben McCown <bmccown@nvidia.com>

---------

Signed-off-by: Ben McCown <bmccown@nvidia.com>

Author: benmccown

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

@@ -13,6 +13,15 @@
 use (``get_platform_sdk(as_service=...)``); the proxy exists only for workloads
 that cannot set the header themselves.
 
+When ``NMP_AUTH_PROXY_ON_BEHALF_OF`` is set, the proxy additionally stamps
+``X-NMP-Principal-On-Behalf-Of`` so the platform authorizes the request as that
+delegated principal rather than granting the service principal's full
+(ServiceSystem) reach. This scopes a deployed workload's platform access to the
+identity that created it (e.g. an agent deployment acting as its creator). The
+delegated identity is baked in at deploy time and is *not* taken from the
+incoming request — the inbound principal/OBO headers are stripped so a co-located
+workload cannot spoof a different identity.
+
 Started via ``nemo services run --sidecars auth-proxy``.
 """
 
@@ -36,15 +45,28 @@
 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"
+# Optional principal id to delegate to via on-behalf-of (e.g. the workload's
+# creator). When set, the service principal acts on behalf of this identity so
+# the platform scopes access to what that principal can reach.
+AUTH_PROXY_ON_BEHALF_OF_ENVVAR = "NMP_AUTH_PROXY_ON_BEHALF_OF"
 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"
+_ON_BEHALF_OF_HEADER = "x-nmp-principal-on-behalf-of"
+# Companion metadata for the on-behalf-of principal. The platform derives the
+# delegated user's groups/email from these (Principal.from_headers -> effective_*),
+# and effective_groups/effective_email feed the PDP authorization input. We stamp
+# only the OBO id here, so any inbound companion headers are untrusted and must be
+# dropped — otherwise a colocated workload could pair our stamped OBO id with
+# attacker-chosen groups/email and be evaluated with those, defeating the scoping.
+_ON_BEHALF_OF_EMAIL_HEADER = "x-nmp-principal-on-behalf-of-email"
+_ON_BEHALF_OF_GROUPS_HEADER = "x-nmp-principal-on-behalf-of-groups"
 
 # 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;
+# - the workload's own credential / principal / on-behalf-of headers (we set the
+#   identity), so they 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(
@@ -53,6 +75,9 @@
         "content-length",
         "authorization",
         _PRINCIPAL_ID_HEADER,
+        _ON_BEHALF_OF_HEADER,
+        _ON_BEHALF_OF_EMAIL_HEADER,
+        _ON_BEHALF_OF_GROUPS_HEADER,
     }
 )
 # We stream the response, so the upstream's framing headers no longer apply.
@@ -73,8 +98,14 @@ def _upstream_base_url() -> str:
     )
 
 
-def build_app(*, base_url: str, principal: str) -> FastAPI:
-    """Build the forwarding FastAPI app for the given upstream and service principal."""
+def build_app(*, base_url: str, principal: str, on_behalf_of: str | None = None) -> FastAPI:
+    """Build the forwarding FastAPI app for the given upstream and service principal.
+
+    When *on_behalf_of* is provided, every forwarded request also carries
+    ``X-NMP-Principal-On-Behalf-Of``, delegating to that principal so the
+    platform scopes access to what it can reach rather than the service
+    principal's full ServiceSystem reach.
+    """
     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)
@@ -100,6 +131,8 @@ async def healthz() -> dict[str, str]:
     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
+        if on_behalf_of:
+            headers[_ON_BEHALF_OF_HEADER] = on_behalf_of
         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)
@@ -131,14 +164,22 @@ def run(parent_stop_signal: threading.Event | None = None) -> None:
     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")
+    on_behalf_of = os.environ.get(AUTH_PROXY_ON_BEHALF_OF_ENVVAR) or None
     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)
+    app = build_app(base_url=base_url, principal=principal, on_behalf_of=on_behalf_of)
 
     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)
+    logger.info(
+        "Starting auth-proxy sidecar on %s:%s -> %s (principal=service:%s, delegated=%s)",
+        host,
+        port,
+        base_url,
+        principal,
+        on_behalf_of is not None,
+    )
     if parent_stop_signal is None:
         server.run()
         return

packages/nmp_common/tests/auth/test_workload_proxy.py

@@ -37,6 +37,88 @@ def test_forward_stamps_service_principal_and_preserves_path() -> None:
     assert "authorization" not in {k.lower() for k in sent.headers}
 
 
+@respx.mock
+def test_forward_stamps_on_behalf_of_when_configured() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(200, json={}))
+    app = build_app(base_url=upstream, principal="agents", on_behalf_of="user:alice")
+    client = TestClient(app)
+
+    client.get("/apis/entities/v2/workspaces")
+
+    sent = route.calls.last.request
+    # Service principal clears the route gate; on-behalf-of narrows access to the creator.
+    assert sent.headers["x-nmp-principal-id"] == "service:agents"
+    assert sent.headers["x-nmp-principal-on-behalf-of"] == "user:alice"
+
+
+@respx.mock
+def test_forward_omits_on_behalf_of_when_not_configured() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(200, json={}))
+    app = build_app(base_url=upstream, principal="agents")
+    client = TestClient(app)
+
+    client.get("/apis/entities/v2/workspaces")
+
+    sent = route.calls.last.request
+    assert "x-nmp-principal-on-behalf-of" not in {k.lower() for k in sent.headers}
+
+
+@respx.mock
+def test_forward_strips_inbound_on_behalf_of_to_prevent_spoofing() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(200, json={}))
+    # The delegated identity is baked in at deploy time; a co-located workload must
+    # not be able to override it (or inject one when none is configured) via headers.
+    app = build_app(base_url=upstream, principal="agents", on_behalf_of="user:alice")
+    client = TestClient(app)
+
+    client.get(
+        "/apis/entities/v2/workspaces",
+        headers={
+            "x-nmp-principal-id": "service:platform",
+            "x-nmp-principal-on-behalf-of": "user:attacker",
+            # Companion metadata must not be smuggled onto our stamped OBO id:
+            # the platform derives effective groups/email from these and feeds
+            # them to the PDP, so attacker-chosen values would escalate.
+            "x-nmp-principal-on-behalf-of-email": "attacker@evil.test",
+            "x-nmp-principal-on-behalf-of-groups": "platform-admins",
+        },
+    )
+
+    sent = route.calls.last.request
+    sent_keys = {k.lower() for k in sent.headers}
+    assert sent.headers["x-nmp-principal-id"] == "service:agents"
+    assert sent.headers["x-nmp-principal-on-behalf-of"] == "user:alice"
+    # The inbound companion headers are dropped (we stamp only the OBO id).
+    assert "x-nmp-principal-on-behalf-of-email" not in sent_keys
+    assert "x-nmp-principal-on-behalf-of-groups" not in sent_keys
+
+
+@respx.mock
+def test_forward_strips_inbound_on_behalf_of_when_none_configured() -> None:
+    upstream = "http://nemo-platform-api:8080"
+    route = respx.get(f"{upstream}/apis/entities/v2/workspaces").mock(return_value=httpx.Response(200, json={}))
+    app = build_app(base_url=upstream, principal="agents")
+    client = TestClient(app)
+
+    client.get(
+        "/apis/entities/v2/workspaces",
+        headers={
+            "x-nmp-principal-on-behalf-of": "user:attacker",
+            "x-nmp-principal-on-behalf-of-email": "attacker@evil.test",
+            "x-nmp-principal-on-behalf-of-groups": "platform-admins",
+        },
+    )
+
+    sent = route.calls.last.request
+    sent_keys = {k.lower() for k in sent.headers}
+    assert "x-nmp-principal-on-behalf-of" not in sent_keys
+    assert "x-nmp-principal-on-behalf-of-email" not in sent_keys
+    assert "x-nmp-principal-on-behalf-of-groups" not in sent_keys
+
+
 @respx.mock
 def test_forward_normalizes_bare_principal_name() -> None:
     upstream = "http://nemo-platform-api:8080"

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

@@ -95,8 +95,16 @@ async def create_deployment(
         *,
         image: str | None = None,
         deployment_mode: DeploymentMode = "subprocess",
+        created_by: str | None = None,
     ) -> DeploymentInfo:
-        """Start the agent process; returns status="starting"."""
+        """Start the agent process; returns status="starting".
+
+        ``created_by`` is the principal id that created the deployment. When
+        platform auth is enabled, container-mode backends delegate the deployed
+        agent's platform calls to this principal (on-behalf-of) so its access is
+        scoped to what the creator can reach rather than the agents service
+        principal's full reach.
+        """
         ...
 
     @abstractmethod

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

@@ -188,6 +188,7 @@ async def _start_deployment(self, dep: AgentDeployment) -> None:
                 port=port,
                 image=dep.image or None,
                 deployment_mode=dep.deployment_mode,
+                created_by=dep.created_by,
             )
         except Exception as exc:
             logger.exception("Failed to start agent for deployment '%s'", dep.name)

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

@@ -284,6 +284,7 @@ def build_deployment_config(
     plugin_wheels_init_image: str | None = None,
     labels: dict[str, str] | None = None,
     auth_proxy_identity: str | None = None,
+    auth_proxy_on_behalf_of: str | None = None,
 ) -> DeploymentConfig:
     """Compile an agent into a long-running ``DeploymentConfig`` (Always).
 
@@ -407,6 +408,7 @@ def build_deployment_config(
             "restart_policy": "Always",
             "auth_proxy_sidecar": auth_proxy_identity is not None,
             "auth_proxy_sidecar_identity": auth_proxy_identity,
+            "auth_proxy_sidecar_on_behalf_of": auth_proxy_on_behalf_of,
         }
     )
 
@@ -433,6 +435,7 @@ async def create_deployment(
         *,
         image: str | None = None,
         deployment_mode: DeploymentMode = "docker",
+        created_by: str | None = None,
     ) -> DeploymentInfo:
         """Create DeploymentConfig + Deployment entities for the agent container."""
         del port  # Host port is allocated by the deployments executor, not agents.
@@ -472,10 +475,24 @@ async def create_deployment(
         # 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.
+        #
+        # The sidecar also delegates to the deployment's creator via on-behalf-of
+        # (when known) so the running agent's platform access is scoped to what the
+        # creator can reach — the workspace(s) they have access to — rather than the
+        # agents service principal's full (ServiceSystem) reach.
         auth_proxy_identity: str | None = None
+        auth_proxy_on_behalf_of: str | None = None
         is_fabric = _is_fabric_agent_config(config)
         if platform_auth_enabled():
             auth_proxy_identity = _AUTH_PROXY_IDENTITY
+            auth_proxy_on_behalf_of = created_by or None
+            if not auth_proxy_on_behalf_of:
+                logger.warning(
+                    "Deployment %r has no creator principal; the agent will run as the "
+                    "unscoped %s service principal without on-behalf-of delegation.",
+                    name,
+                    _AUTH_PROXY_IDENTITY,
+                )
             rewrite_target = f"http://127.0.0.1:{auth_proxy_port()}"
         else:
             rewrite_target = gateway
@@ -504,6 +521,7 @@ async def create_deployment(
             plugin_wheels_init_image=self._config.plugin_wheels_init_image,
             labels=deployment_labels,
             auth_proxy_identity=auth_proxy_identity,
+            auth_proxy_on_behalf_of=auth_proxy_on_behalf_of,
         )
         await entities.create(deployment_config)
         try:

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

@@ -219,9 +219,13 @@ async def create_deployment(
         *,
         image: str | None = None,
         deployment_mode: DeploymentMode = "subprocess",
+        created_by: str | None = None,
     ) -> DeploymentInfo:
         """Start a local deployment for NAT workflows or Platform-owned agent specs."""
-        del image, deployment_mode
+        # created_by drives on-behalf-of delegation only for container modes (via
+        # the auth-proxy sidecar). Subprocess deployments run in-process on the
+        # platform host and do not use the sidecar, so it does not apply here.
+        del image, deployment_mode, created_by
         if config.get("config_format") == NEMO_AGENTS_SPEC_CONFIG_FORMAT:
             return await self._create_fabric_deployment(workspace, name, config, port)
 

plugins/nemo-agents/tests/unit/test_runner_deployments.py

@@ -485,7 +485,12 @@ async def test_create_deployment_k8s_auth_on_requests_auth_proxy_sidecar() -> No
         patch("nemo_agents_plugin.runner.deployments_backend.auth_proxy_port", return_value=8090),
     ):
         info = await backend.create_deployment(
-            workspace="default", name="hello-dep", config=config, port=0, deployment_mode="k8s"
+            workspace="default",
+            name="hello-dep",
+            config=config,
+            port=0,
+            deployment_mode="k8s",
+            created_by="user:alice",
         )
     assert info.status == "starting"
     created_config = entities.create.await_args_list[0].args[0]
@@ -494,6 +499,9 @@ async def test_create_deployment_k8s_auth_on_requests_auth_proxy_sidecar() -> No
     # agents layer does not build the container itself.
     assert created_config.auth_proxy_sidecar is True
     assert created_config.auth_proxy_sidecar_identity == "agents"
+    # The creator principal is delegated via on-behalf-of so the running agent's
+    # platform access is scoped to what the creator can reach.
+    assert created_config.auth_proxy_sidecar_on_behalf_of == "user:alice"
     assert [c.name for c in created_config.containers] == ["agent"]
     assert [c.name for c in created_config.init_containers] == []
 
@@ -504,6 +512,32 @@ async def test_create_deployment_k8s_auth_on_requests_auth_proxy_sidecar() -> No
     )
 
 
+@pytest.mark.asyncio
+async def test_create_deployment_k8s_auth_on_without_creator_omits_on_behalf_of() -> None:
+    # When auth is on but the deployment has no known creator, the sidecar still
+    # stamps the service principal but cannot delegate — access is unscoped.
+    backend = _backend(
+        default_image="nmp-api:latest",
+        default_executor="k8s",
+        k8s_internal_base_url="http://nmp-api:8080",
+    )
+    entities = AsyncMock()
+    backend._entities = entities
+    with (
+        patch("nemo_agents_plugin.runner.deployments_backend.get_base_url", return_value="http://localhost:8080"),
+        patch("nemo_agents_plugin.runner.deployments_backend.platform_auth_enabled", return_value=True),
+        patch("nemo_agents_plugin.runner.deployments_backend.auth_proxy_port", return_value=8090),
+    ):
+        info = await backend.create_deployment(
+            workspace="default", name="hello-dep", config={}, port=0, deployment_mode="k8s"
+        )
+    assert info.status == "starting"
+    created_config = entities.create.await_args_list[0].args[0]
+    assert created_config.auth_proxy_sidecar is True
+    assert created_config.auth_proxy_sidecar_identity == "agents"
+    assert created_config.auth_proxy_sidecar_on_behalf_of is None
+
+
 @pytest.mark.asyncio
 async def test_create_deployment_k8s_auth_off_no_sidecar() -> None:
     backend = _backend(