feat(deployments): migrate plugin authz to @path_rule surface

The deployments plugin still declared authz through the removed get_authz_contribution() classmethod, which the routes-derived discovery pipeline ignores, leaving all 14 routes unruled (silently fenced under deny_route, a bundle-build abort under the new hard_fail default).

Declare permissions as typed PermissionSets and attach @path_rule to every route; the controller-only status PUTs require SERVICE_PRINCIPAL (the existing require_service_principal dependency stays as defense-in-depth). Permission ids are preserved exactly, so role grants are unchanged. Drop the old classmethod and assert per-route rule coverage in the startup test.

Signed-off-by: Max Dubrinsky <mdubrinsky@nvidia.com>

Author: maxdubrinsky

plugins/nemo-deployments/src/nemo_deployments_plugin/api/v2/_perms.py

@@ -0,0 +1,37 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Typed permission vocabulary for the deployments plugin's routes.
+
+Three sub-namespaces under ``deployments`` (one per entity collection). Route handlers
+reference these constants in their ``@path_rule``; the platform derives the permission
+catalog from the routes, so there is no parallel list to keep in sync. The controller-only
+status routes mint a ``status.update`` permission under the collection they project onto.
+"""
+
+from __future__ import annotations
+
+from nemo_platform_plugin.authz import PermissionSet, perm
+
+
+class DeploymentConfigPerms(PermissionSet, namespace="deployments.deployment-configs"):
+    CREATE = perm("Create deployments deployment-configs")
+    LIST = perm("List deployments deployment-configs")
+    READ = perm("Read deployments deployment-configs")
+    DELETE = perm("Delete deployments deployment-configs")
+
+
+class DeploymentPerms(PermissionSet, namespace="deployments.deployments"):
+    CREATE = perm("Create deployments deployments")
+    LIST = perm("List deployments deployments")
+    READ = perm("Read deployments deployments")
+    DELETE = perm("Delete deployments deployments")
+    STATUS_UPDATE = perm("Update deployment observed status (controller)", suffix="status.update")
+
+
+class VolumePerms(PermissionSet, namespace="deployments.volumes"):
+    CREATE = perm("Create deployments volumes")
+    LIST = perm("List deployments volumes")
+    READ = perm("Read deployments volumes")
+    DELETE = perm("Delete deployments volumes")
+    STATUS_UPDATE = perm("Update volume observed status (controller)", suffix="status.update")

plugins/nemo-deployments/src/nemo_deployments_plugin/api/v2/deployment_configs.py

@@ -8,7 +8,9 @@
 import logging
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from nemo_deployments_plugin.api.v2._perms import DeploymentConfigPerms
 from nemo_deployments_plugin.api.v2.dependencies import get_entity_client
+from nemo_deployments_plugin.authz import SCOPE
 from nemo_deployments_plugin.entities import DeploymentConfig
 from nemo_deployments_plugin.references import deployment_names_using_config
 from nemo_deployments_plugin.schema import (
@@ -23,6 +25,7 @@
     prerequisite_names,
 )
 from nemo_platform_plugin.api.filters import make_filter_obj_dep
+from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 from nemo_platform_plugin.schema import PaginationData
 
@@ -55,6 +58,7 @@ async def _list_all_deployment_configs(
 
 
 @router.post("/deployment-configs", response_model=DeploymentConfig, status_code=201, tags=["Deployment Configs"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentConfigPerms.CREATE], scopes=SCOPE.write())
 async def create_deployment_config(
     workspace: str,
     body: CreateDeploymentConfigRequest,
@@ -87,6 +91,7 @@ async def create_deployment_config(
 
 
 @router.get("/deployment-configs", response_model=DeploymentConfigPage, tags=["Deployment Configs"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentConfigPerms.LIST], scopes=SCOPE.read())
 async def list_deployment_configs(
     workspace: str,
     page: int = Query(default=1, ge=1),
@@ -109,6 +114,7 @@ async def list_deployment_configs(
 
 
 @router.get("/deployment-configs/{name}", response_model=DeploymentConfig, tags=["Deployment Configs"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentConfigPerms.READ], scopes=SCOPE.read())
 async def get_deployment_config(
     workspace: str,
     name: str,
@@ -124,6 +130,7 @@ async def get_deployment_config(
 
 
 @router.delete("/deployment-configs/{name}", status_code=204, tags=["Deployment Configs"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentConfigPerms.DELETE], scopes=SCOPE.write())
 async def delete_deployment_config(
     workspace: str,
     name: str,

plugins/nemo-deployments/src/nemo_deployments_plugin/api/v2/deployments.py

@@ -9,10 +9,13 @@
 from typing import cast
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from nemo_deployments_plugin.api.v2._perms import DeploymentPerms
 from nemo_deployments_plugin.api.v2.dependencies import get_entity_client
+from nemo_deployments_plugin.authz import SCOPE
 from nemo_deployments_plugin.entities import Deployment, DeploymentConfig, DeploymentStatus
 from nemo_deployments_plugin.schema import CreateDeploymentRequest, DeploymentFilter, DeploymentPage
 from nemo_platform_plugin.api.filters import make_filter_obj_dep
+from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 from nemo_platform_plugin.filter_ops import ComparisonOperation, FilterOperator
 from nemo_platform_plugin.schema import PaginationData
@@ -55,6 +58,7 @@ def _parse_deployment_config_ref(ref: str, default_workspace: str) -> tuple[str,
 
 
 @router.post("/deployments", response_model=Deployment, status_code=201, tags=["Deployments"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentPerms.CREATE], scopes=SCOPE.write())
 async def create_deployment(
     workspace: str,
     body: CreateDeploymentRequest,
@@ -87,6 +91,7 @@ async def create_deployment(
 
 
 @router.get("/deployments", response_model=DeploymentPage, tags=["Deployments"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentPerms.LIST], scopes=SCOPE.read())
 async def list_deployments(
     workspace: str,
     page: int = Query(default=1, ge=1),
@@ -122,6 +127,7 @@ async def list_deployments(
 
 
 @router.get("/deployments/{name}", response_model=Deployment, tags=["Deployments"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentPerms.READ], scopes=SCOPE.read())
 async def get_deployment(
     workspace: str,
     name: str,
@@ -137,6 +143,7 @@ async def get_deployment(
 
 
 @router.delete("/deployments/{name}", status_code=204, tags=["Deployments"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[DeploymentPerms.DELETE], scopes=SCOPE.write())
 async def delete_deployment(
     workspace: str,
     name: str,

plugins/nemo-deployments/src/nemo_deployments_plugin/api/v2/status.py

@@ -6,15 +6,19 @@
 from __future__ import annotations
 
 from fastapi import APIRouter, Depends, HTTPException
+from nemo_deployments_plugin.api.v2._perms import DeploymentPerms, VolumePerms
 from nemo_deployments_plugin.api.v2.dependencies import get_entity_client, require_service_principal
+from nemo_deployments_plugin.authz import SCOPE
 from nemo_deployments_plugin.entities import Deployment, Volume
 from nemo_deployments_plugin.schema import UpdateDeploymentStatusRequest, UpdateVolumeStatusRequest
+from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 
 router = APIRouter()
 
 
 @router.put("/deployments/{name}/status", response_model=Deployment, tags=["Deployment Status"])
+@path_rule(callers=[CallerKind.SERVICE_PRINCIPAL], permissions=[DeploymentPerms.STATUS_UPDATE], scopes=SCOPE.write())
 async def update_deployment_status(
     workspace: str,
     name: str,
@@ -47,6 +51,7 @@ async def update_deployment_status(
 
 
 @router.put("/volumes/{name}/status", response_model=Volume, tags=["Volume Status"])
+@path_rule(callers=[CallerKind.SERVICE_PRINCIPAL], permissions=[VolumePerms.STATUS_UPDATE], scopes=SCOPE.write())
 async def update_volume_status(
     workspace: str,
     name: str,

plugins/nemo-deployments/src/nemo_deployments_plugin/api/v2/volumes.py

@@ -6,11 +6,14 @@
 from __future__ import annotations
 
 from fastapi import APIRouter, Depends, HTTPException, Query
+from nemo_deployments_plugin.api.v2._perms import VolumePerms
 from nemo_deployments_plugin.api.v2.dependencies import get_entity_client
+from nemo_deployments_plugin.authz import SCOPE
 from nemo_deployments_plugin.entities import Volume
 from nemo_deployments_plugin.references import deployment_config_names_referencing_volume
 from nemo_deployments_plugin.schema import CreateVolumeRequest, VolumeFilter, VolumePage
 from nemo_platform_plugin.api.filters import make_filter_obj_dep
+from nemo_platform_plugin.authz import CallerKind, path_rule
 from nemo_platform_plugin.entity_client import NemoEntitiesClient, NemoEntityConflictError, NemoEntityNotFoundError
 from nemo_platform_plugin.schema import PaginationData
 
@@ -20,6 +23,7 @@
 
 
 @router.post("/volumes", response_model=Volume, status_code=201, tags=["Volumes"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[VolumePerms.CREATE], scopes=SCOPE.write())
 async def create_volume(
     workspace: str,
     body: CreateVolumeRequest,
@@ -41,6 +45,7 @@ async def create_volume(
 
 
 @router.get("/volumes", response_model=VolumePage, tags=["Volumes"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[VolumePerms.LIST], scopes=SCOPE.read())
 async def list_volumes(
     workspace: str,
     page: int = Query(default=1, ge=1),
@@ -63,6 +68,7 @@ async def list_volumes(
 
 
 @router.get("/volumes/{name}", response_model=Volume, tags=["Volumes"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[VolumePerms.READ], scopes=SCOPE.read())
 async def get_volume(
     workspace: str,
     name: str,
@@ -78,6 +84,7 @@ async def get_volume(
 
 
 @router.delete("/volumes/{name}", status_code=204, tags=["Volumes"])
+@path_rule(callers=[CallerKind.PRINCIPAL], permissions=[VolumePerms.DELETE], scopes=SCOPE.write())
 async def delete_volume(
     workspace: str,
     name: str,

plugins/nemo-deployments/src/nemo_deployments_plugin/authz.py

@@ -0,0 +1,13 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""The deployments plugin's authz scope.
+
+The route modules import :data:`SCOPE` so the plugin shares one ``AuthzScope("deployments")``.
+"""
+
+from __future__ import annotations
+
+from nemo_platform_plugin.authz import AuthzScope
+
+SCOPE = AuthzScope("deployments")

plugins/nemo-deployments/src/nemo_deployments_plugin/service.py

@@ -11,28 +11,11 @@
 from nemo_deployments_plugin.backends.registry import ExecutorRegistry, ExecutorSpec
 from nemo_deployments_plugin.config import DeploymentsConfig
 from nemo_platform import AsyncNeMoPlatform
-from nemo_platform_plugin.authz import AuthzContribution, AuthzEndpointMethod
 from nemo_platform_plugin.sdk_provider import get_async_platform_sdk
 from nemo_platform_plugin.service import NemoService, RouterSpec
 
 logger = logging.getLogger(__name__)
 
-_SERVICE_NAME = "deployments"
-_READ_SCOPES = [f"{_SERVICE_NAME}:read", "platform:read"]
-_WRITE_SCOPES = [f"{_SERVICE_NAME}:write", "platform:write"]
-
-
-def _read_method(permission: str) -> AuthzEndpointMethod:
-    return AuthzEndpointMethod(permissions=[permission], scopes=list(_READ_SCOPES))
-
-
-def _write_method(permission: str) -> AuthzEndpointMethod:
-    return AuthzEndpointMethod(permissions=[permission], scopes=list(_WRITE_SCOPES))
-
-
-def _read_methods(permission: str) -> dict[str, AuthzEndpointMethod]:
-    return {method: _read_method(permission) for method in ("get", "head")}
-
 
 class DeploymentsService(NemoService):
     """HTTP service for deployment configs, deployments, volumes, and controller status."""
@@ -49,52 +32,6 @@ def executor_registry(self) -> ExecutorRegistry:
             self._executor_registry = ExecutorRegistry.empty()
         return self._executor_registry
 
-    @classmethod
-    def get_authz_contribution(cls) -> AuthzContribution:
-        """Authorization policy for deployments plugin routes."""
-        base = f"/apis/{cls.name}/v2/workspaces/{{workspace}}"
-        permissions: dict[str, str] = {}
-        endpoints: dict[str, dict[str, AuthzEndpointMethod]] = {}
-
-        for resource, path_segment in (
-            ("deployment-configs", "deployment-configs"),
-            ("deployments", "deployments"),
-            ("volumes", "volumes"),
-        ):
-            create_perm = f"{cls.name}.{resource}.create"
-            list_perm = f"{cls.name}.{resource}.list"
-            read_perm = f"{cls.name}.{resource}.read"
-            delete_perm = f"{cls.name}.{resource}.delete"
-            permissions.update(
-                {
-                    create_perm: f"Create {cls.name} {resource}",
-                    list_perm: f"List {cls.name} {resource}",
-                    read_perm: f"Read {cls.name} {resource}",
-                    delete_perm: f"Delete {cls.name} {resource}",
-                }
-            )
-            endpoints[f"{base}/{path_segment}"] = {
-                **_read_methods(list_perm),
-                "post": _write_method(create_perm),
-            }
-            endpoints[f"{base}/{path_segment}/{{name}}"] = {
-                "delete": _write_method(delete_perm),
-                **_read_methods(read_perm),
-            }
-
-        deployment_status_perm = f"{cls.name}.deployments.status.update"
-        volume_status_perm = f"{cls.name}.volumes.status.update"
-        permissions[deployment_status_perm] = "Update deployment observed status (controller)"
-        permissions[volume_status_perm] = "Update volume observed status (controller)"
-        endpoints[f"{base}/deployments/{{name}}/status"] = {
-            "put": _write_method(deployment_status_perm),
-        }
-        endpoints[f"{base}/volumes/{{name}}/status"] = {
-            "put": _write_method(volume_status_perm),
-        }
-
-        return AuthzContribution(permissions=permissions, endpoints=endpoints)
-
     def get_routers(self) -> list[RouterSpec]:
         from nemo_deployments_plugin.api.v2 import (
             deployment_configs,

plugins/nemo-deployments/tests/unit/test_service_startup.py

@@ -5,6 +5,7 @@
 
 from fastapi.routing import APIRoute
 from nemo_deployments_plugin.service import DeploymentsService
+from nemo_platform_plugin.authz import get_path_rules
 
 
 def _mounted_paths() -> set[str]:
@@ -31,16 +32,14 @@ def test_service_name_matches_entry_point() -> None:
 
 
 def test_service_authz_covers_mounted_routes() -> None:
-    contribution = DeploymentsService.get_authz_contribution()
-    endpoint_paths = set(contribution.endpoints.keys())
-    for path in _mounted_paths():
-        assert path in endpoint_paths, f"missing authz entry for {path}"
-        route_methods = {
-            method.lower()
-            for spec in DeploymentsService().get_routers()
-            for route in spec.router.routes
-            if isinstance(route, APIRoute) and f"/apis/deployments{spec.prefix}{route.path}" == path
-            for method in route.methods or set()
-        }
-        for method in route_methods:
-            assert method in contribution.endpoints[path], f"missing authz method {method} for {path}"
+    """Every mounted route carries at least one ``@path_rule``.
+
+    Authz is derived from the routes (no separate contribution); an unruled route would be
+    treated as invalid and fenced/denied by the PDP, so coverage is the property to assert.
+    """
+    service = DeploymentsService()
+    for spec in service.get_routers():
+        for route in spec.router.routes:
+            if isinstance(route, APIRoute):
+                full = f"/apis/deployments{spec.prefix}{route.path}"
+                assert get_path_rules(route.endpoint), f"missing @path_rule for {full}"