feat: plugin-contributed authorization policy merge

packages/nemo_platform_plugin/AGENTS.md

@@ -26,6 +26,7 @@ Before writing any plugin code, load the relevant skill. Skills contain exact im
 ## Python Conventions
 
 - **No `__init__.py` files**: The plugin package directory does not require `__init__.py`. Do not add one.
+- **Concrete type hints**: Prefer concrete type hints over string-based ones. Do not import types under `TYPE_CHECKING` when they are runtime-available in the same package — use regular imports instead.
 - **Package path**: `src/nemo_<plugin_name>/` where `plugin_name` matches the CLI entry-point key (e.g., `src/nemo_my_plugin/` for plugin `"my-plugin"`).
 - **Build system**: Always use hatchling with `packages = ["src/nemo_my_plugin"]` — use the exact package directory name.
 - **Run tests**: `uv run pytest`

packages/nemo_platform_plugin/src/nemo_platform_plugin/authz.py

@@ -0,0 +1,141 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Authorization policy contributions for NeMo Platform plugins.
+
+Plugins declare API routes and permissions so the auth service can authorize
+requests without hand-editing ``static-authz.yaml`` for every new surface.
+
+Contributions are merged at runtime when the OPA bundle is built, and can be
+materialized into ``static-authz.yaml`` via ``auth-tools sync-plugins``.
+
+Example (customization job collection)::
+
+    from nemo_platform_plugin.authz import AuthzContribution, authz_for_workspace_job_collection
+
+    class AutomodelContributor:
+        ...
+        def get_authz_contribution(self) -> AuthzContribution:
+            return authz_for_workspace_job_collection(
+                api_area="customization",
+                collection_suffix="/automodel/jobs",
+                permission_prefix="customization.automodel.jobs",
+                include_healthz=True,
+                healthz_suffix="/automodel/healthz",
+            )
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass(frozen=True)
+class AuthzEndpointMethod:
+    """One HTTP method binding for an API route."""
+
+    permissions: list[str]
+    scopes: list[str] | None = None
+
+
+@dataclass
+class AuthzContribution:
+    """Authorization data contributed by a plugin."""
+
+    permissions: dict[str, str] = field(default_factory=dict)
+    """Flat registry entries: ``permission_id`` → human-readable description."""
+
+    endpoints: dict[str, dict[str, AuthzEndpointMethod]] = field(default_factory=dict)
+    """Full API paths (``/apis/...``) → lower-case HTTP method → spec."""
+
+    role_permissions: dict[str, list[str]] = field(default_factory=dict)
+    """Optional explicit role → permission grants (merged with defaults)."""
+
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize for :func:`nmp.common.auth.authz_merge.merge_authz_contributions`."""
+        return {
+            "permissions": dict(self.permissions),
+            "endpoints": {
+                path: {
+                    method: {
+                        "permissions": spec.permissions,
+                        **({"scopes": spec.scopes} if spec.scopes is not None else {}),
+                    }
+                    for method, spec in methods.items()
+                }
+                for path, methods in self.endpoints.items()
+            },
+            "role_permissions": {role: list(perms) for role, perms in self.role_permissions.items()},
+        }
+
+
+def _scopes_for(api_area: str, write: bool) -> list[str]:
+    verb = "write" if write else "read"
+    return [f"{api_area}:{verb}", f"platform:{verb}"]
+
+
+def _job_collection_permissions(permission_prefix: str) -> dict[str, str]:
+    return {
+        f"{permission_prefix}.create": f"Create {permission_prefix} jobs",
+        f"{permission_prefix}.list": f"List {permission_prefix} jobs",
+        f"{permission_prefix}.read": f"Read {permission_prefix} jobs",
+        f"{permission_prefix}.delete": f"Delete {permission_prefix} jobs",
+    }
+
+
+def authz_for_workspace_job_collection(
+    api_area: str,
+    collection_suffix: str,
+    permission_prefix: str,
+    include_healthz: bool = False,
+    healthz_suffix: str | None = None,
+) -> AuthzContribution:
+    """Build authz for standard CORE job routes under ``/apis/<area>/v2/workspaces/{workspace}...``.
+
+    Args:
+        api_area: URL segment after ``/apis/`` (e.g. ``customization``, ``safe-synthesizer``).
+        collection_suffix: Path after workspace (e.g. ``/automodel/jobs`` or ``/jobs``).
+        permission_prefix: Dot-separated permission namespace (e.g. ``customization.automodel.jobs``).
+        include_healthz: When true, register GET healthz with empty permissions (authenticated only).
+        healthz_suffix: Defaults to ``{first segment of collection_suffix}/healthz`` when omitted.
+    """
+    if not collection_suffix.startswith("/"):
+        raise ValueError("collection_suffix must start with '/'")
+    base = f"/apis/{api_area}/v2/workspaces/{{workspace}}{collection_suffix}"
+    perms = _job_collection_permissions(permission_prefix)
+    prefix = permission_prefix
+    endpoints: dict[str, dict[str, AuthzEndpointMethod]] = {
+        base: {
+            "post": AuthzEndpointMethod(
+                permissions=[f"{prefix}.create"],
+                scopes=_scopes_for(api_area, write=True),
+            ),
+            "get": AuthzEndpointMethod(
+                permissions=[f"{prefix}.list"],
+                scopes=_scopes_for(api_area, write=False),
+            ),
+        },
+        f"{base}/{{name}}": {
+            "get": AuthzEndpointMethod(
+                permissions=[f"{prefix}.read"],
+                scopes=_scopes_for(api_area, write=False),
+            ),
+            "delete": AuthzEndpointMethod(
+                permissions=[f"{prefix}.delete"],
+                scopes=_scopes_for(api_area, write=True),
+            ),
+        },
+    }
+    if include_healthz:
+        if healthz_suffix is None:
+            first = collection_suffix.strip("/").split("/")[0]
+            healthz_suffix = f"/{first}/healthz"
+        if not healthz_suffix.startswith("/"):
+            healthz_suffix = f"/{healthz_suffix}"
+        health_path = f"/apis/{api_area}/v2/workspaces/{{workspace}}{healthz_suffix}"
+        endpoints[health_path] = {
+            "get": AuthzEndpointMethod(permissions=[], scopes=[]),
+        }
+
+    return AuthzContribution(permissions=perms, endpoints=endpoints)

packages/nemo_platform_plugin/src/nemo_platform_plugin/authz_discovery.py

@@ -0,0 +1,154 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Discover plugin authorization contributions for policy merge."""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from functools import cache
+from typing import Any, Callable
+
+from nemo_platform_plugin.authz import AuthzContribution, AuthzEndpointMethod
+
+logger = logging.getLogger(__name__)
+
+AUTHZ_GROUP = "nemo.authz"
+
+AuthzContributor = Callable[[], AuthzContribution] | type[Any]
+
+
+def _load_authz_contribution(loaded: AuthzContributor, source: str) -> AuthzContribution | None:
+    try:
+        if isinstance(loaded, type):
+            if hasattr(loaded, "get_authz_contribution"):
+                result = _invoke_get_authz_contribution(loaded)
+            else:
+                instance = loaded()
+                result = _invoke_get_authz_contribution(instance)
+        elif callable(loaded):
+            result = loaded()
+        else:
+            logger.warning("Authz entry %s is not callable or a class — skipping", source)
+            return None
+    except Exception:
+        logger.warning("Failed to load authz contribution from %s — skipping", source, exc_info=True)
+        return None
+
+    if result is None:
+        return None
+    if isinstance(result, AuthzContribution):
+        return result
+    if isinstance(result, dict):
+        return AuthzContribution(
+            permissions=result.get("permissions") or {},
+            endpoints={
+                path: {method: _method_from_dict(spec) for method, spec in methods.items() if isinstance(spec, dict)}
+                for path, methods in (result.get("endpoints") or {}).items()
+                if isinstance(methods, dict)
+            },
+            role_permissions=result.get("role_permissions") or {},
+        )
+    logger.warning("Authz contribution from %s has unexpected type %r — skipping", source, type(result))
+    return None
+
+
+def _invoke_get_authz_contribution(item: Any) -> AuthzContribution | dict[str, Any] | None:
+    """Call ``get_authz_contribution`` on a service class or contributor instance."""
+    getter = getattr(item, "get_authz_contribution", None)
+    if not callable(getter):
+        return None
+    if isinstance(item, type):
+        # discover_services() yields classes — must be @classmethod on NemoService.
+        return getter()
+    return getter()
+
+
+def _method_from_dict(spec: dict[str, Any]) -> AuthzEndpointMethod:
+    return AuthzEndpointMethod(
+        permissions=list(spec.get("permissions") or []),
+        scopes=list(spec["scopes"]) if spec.get("scopes") is not None else None,
+    )
+
+
+def _collect_from_plugin_surface(
+    items: dict[str, Any],
+    surface: str,
+) -> list[AuthzContribution]:
+    contributions: list[AuthzContribution] = []
+    for key, item in items.items():
+        if not hasattr(item, "get_authz_contribution"):
+            continue
+        if isinstance(item, type):
+            method = inspect.getattr_static(item, "get_authz_contribution", None)
+            if method is None or not isinstance(method, classmethod):
+                # Only classmethods are valid on NemoService subclasses (no instance).
+                continue
+        try:
+            result = _invoke_get_authz_contribution(item)
+        except TypeError as exc:
+            logger.warning(
+                "Authz on %s %r must be a @classmethod (discover_services loads classes): %s",
+                surface,
+                key,
+                exc,
+            )
+            continue
+        except Exception:
+            logger.warning(
+                "Failed to get authz contribution from %s %r — skipping",
+                surface,
+                key,
+                exc_info=True,
+            )
+            continue
+        if result is None:
+            continue
+        if isinstance(result, AuthzContribution):
+            contributions.append(result)
+        elif isinstance(result, dict):
+            loaded = _load_authz_contribution(lambda: result, source=f"{surface}:{key}")
+            if loaded is not None:
+                contributions.append(loaded)
+        else:
+            logger.warning(
+                "Authz contribution from %s %r has unexpected type %r — skipping",
+                surface,
+                key,
+                type(result),
+            )
+    return contributions
+
+
+@cache
+def discover_authz_contributions() -> list[AuthzContribution]:
+    """Collect authz contributions from entry points and plugin surfaces.
+
+    Sources (in order):
+
+    1. ``nemo.authz`` entry points (callable or class)
+    2. ``nemo.services`` classes implementing :meth:`get_authz_contribution`
+    """
+    from nemo_platform_plugin.discovery import discover_entry_points, discover_services
+
+    contributions: list[AuthzContribution] = []
+
+    for ep_name, ep in discover_entry_points(AUTHZ_GROUP).items():
+        try:
+            loaded = ep.load()
+            contrib = _load_authz_contribution(loaded, source=f"nemo.authz:{ep_name}")
+            if contrib is not None:
+                contributions.append(contrib)
+                logger.debug("Loaded authz contribution from nemo.authz:%s", ep_name)
+        except Exception:
+            logger.warning("Failed to load nemo.authz entry %r — skipping", ep_name, exc_info=True)
+
+    contributions.extend(_collect_from_plugin_surface(discover_services(), surface="nemo.services"))
+
+    return contributions
+
+
+def discover_authz_contribution_dicts() -> list[dict[str, Any]]:
+    """Return contributions as dicts for :func:`nmp.common.auth.authz_merge.merge_authz_contributions`."""
+    return [c.to_dict() for c in discover_authz_contributions()]

packages/nemo_platform_plugin/src/nemo_platform_plugin/commands.py

@@ -585,6 +585,7 @@ def _do_submit() -> Any:
                 workspace=workspace,
                 profile=profile,
                 options=merged_options or None,
+                headers=_resolve_submit_auth_headers(typer_ctx) or None,
             )
 
         renderer: CLIRenderer | None = None
@@ -1077,6 +1078,22 @@ async def _invoke_function_locally(
     typer.echo(_format_value_for_stdout(awaited))
 
 
+def _resolve_submit_auth_headers(typer_ctx: typer.Context) -> dict[str, str]:
+    """Bearer (and other) default headers from the active CLI context."""
+    state = typer_ctx.obj
+    if state is None or not hasattr(state, "get_sdk_context"):
+        return {}
+    try:
+        ctx = state.get_sdk_context()
+        client_config = ctx.user.get_client_config()
+        headers = client_config.get("default_headers")
+        if isinstance(headers, dict):
+            return {str(k): str(v) for k, v in headers.items()}
+    except Exception:
+        return {}
+    return {}
+
+
 # ---- submit ------------------------------------------------------ #
 
 
@@ -1119,7 +1136,7 @@ def _submit(typer_ctx: typer.Context, **kwargs: object) -> None:
             cluster=cluster,
             workspace=workspace,
         )
-        headers: dict[str, str] = {}
+        headers = _resolve_submit_auth_headers(typer_ctx)
         if request_id is not None:
             headers["X-Request-ID"] = request_id
 

packages/nemo_platform_plugin/src/nemo_platform_plugin/discovery.py

@@ -23,6 +23,7 @@
 ``nemo.executors``             → :func:`discover_executors`             — ``Executor`` class
 ``nemo.inference_middleware``  → :func:`discover_inference_middleware`  — :class:`~nemo_platform_plugin.inference_middleware.NemoInferenceMiddleware` subclass  (typed, IGW instantiates)
 ``nemo.seed``                  → :func:`discover_seed_jobs`             — :class:`~nemo_platform_plugin.seed.NemoSeedJob` subclass  (typed, platform instantiates)
+``nemo.authz``                 → :func:`~nemo_platform_plugin.authz_discovery.discover_authz_contributions` — policy endpoints/permissions (merged at runtime and via ``auth-tools sync-plugins``)
 
 Wrappers for surfaces whose types are not yet defined in this package return
 ``dict[str, Any]`` — callers cast as needed.
@@ -74,6 +75,7 @@
     "nemo.executors",
     "nemo.inference_middleware",
     "nemo.seed",
+    "nemo.authz",
 )
 
 # Surface groups whose entry-point keys are dot-separated as
@@ -96,6 +98,7 @@
     "nemo.executors": "NEMO_PLUGIN_EXECUTORS_ALLOWLIST",
     "nemo.inference_middleware": "NEMO_PLUGIN_INFERENCE_MIDDLEWARE_ALLOWLIST",
     "nemo.seed": "NEMO_PLUGIN_SEED_ALLOWLIST",
+    "nemo.authz": "NEMO_PLUGIN_AUTHZ_ALLOWLIST",
 }