feat(nvca): provision worker identity for container function pods

src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel

@@ -22,6 +22,7 @@ go_library(
         "tokenfetcher.go",
         "translate_workload.go",
         "transport_tls.go",
+        "worker_identity.go",
     ],
     embedsrcs = ["karta/dynamo/nvidia.com_dynamographdeployment_v1alpha1.yaml"],
     importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/internal/miniservice",

src/compute-plane-services/nvca/internal/miniservice/controller.go

@@ -126,6 +126,12 @@ type ControllerOptions struct {
 	// profiling enabled. Shared with BackendK8sCache.
 	NsightProfilingAllowlist *profiling.Allowlist
 
+	// WorkerIdentityEnabled gates projected ServiceAccount token provisioning for
+	// MiniService utils pods (self-hosted PSAT mode).
+	WorkerIdentityEnabled bool
+	// ClusterID is the NVCF cluster identifier, used to set the PSAT token audience.
+	ClusterID string
+
 	// Internal use.
 	cacheDir string
 }

src/compute-plane-services/nvca/internal/miniservice/reconcile.go

@@ -844,6 +844,13 @@ func (r *Reconciler) doInstall(ctx context.Context,
 		return reconcile.Result{}, reconcile.TerminalError(err)
 	}
 
+	if r.WorkerIdentityEnabled {
+		if err := ensureWorkerIdentity(ctx, r.Client, ms.Spec.Namespace); err != nil {
+			return reconcile.Result{}, fmt.Errorf("ensure worker identity for MiniService %s: %w", ms.Name, err)
+		}
+		injectWorkerTokenVolume(utilsPod, r.ClusterID)
+	}
+
 	infraObjs = append(infraObjs, utilsPod)
 
 	if r.FeatureFlagFetcher.IsAttributeEnabled(featureflag.AttrNVLinkOptimized) {

src/compute-plane-services/nvca/internal/miniservice/worker_identity.go

@@ -0,0 +1,128 @@
+/*
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mscontroller
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const (
+	// miniserviceWorkerSAName is the ServiceAccount name for MiniService worker identity.
+	// Each MiniService gets its own namespace, so a fixed name is sufficient.
+	miniserviceWorkerSAName = "nvcf-worker"
+	// miniserviceWorkerTokenVolumeName is the projected SAT volume name injected into the utils pod.
+	miniserviceWorkerTokenVolumeName = "nvcf-worker-token"
+	// miniserviceWorkerTokenMountPath is where the projected SAT is mounted inside the container.
+	miniserviceWorkerTokenMountPath = "/var/run/secrets/tokens"
+	// miniserviceWorkerTokenFilePath is the full path to the projected token file.
+	miniserviceWorkerTokenFilePath = miniserviceWorkerTokenMountPath + "/token"
+	// miniserviceWorkerTokenFilePathEnvKey is the env var pointing at the token file.
+	miniserviceWorkerTokenFilePathEnvKey = "NVCF_TOKEN_FILE_PATH"
+	// miniserviceWorkerIdentitySourceEnvKey indicates the active identity mechanism.
+	miniserviceWorkerIdentitySourceEnvKey = "NVCF_IDENTITY_SOURCE"
+	// miniserviceWorkerIdentitySourcePSAT is the value written when PSAT is the mechanism.
+	miniserviceWorkerIdentitySourcePSAT = "psat"
+)
+
+// miniserviceWorkerTokenExpirationSeconds is the requested SAT lifetime.
+var miniserviceWorkerTokenExpirationSeconds int64 = 900
+
+// ensureWorkerIdentity creates the worker ServiceAccount, Role, and RoleBinding in namespace.
+// It is idempotent: existing objects are not modified.
+func ensureWorkerIdentity(ctx context.Context, c client.Client, namespace string) error {
+	sa := &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{Name: miniserviceWorkerSAName, Namespace: namespace},
+	}
+	if err := c.Create(ctx, sa); err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("create worker ServiceAccount %s/%s: %w", namespace, miniserviceWorkerSAName, err)
+	}
+
+	role := &rbacv1.Role{
+		ObjectMeta: metav1.ObjectMeta{Name: miniserviceWorkerSAName, Namespace: namespace},
+		Rules:      nil,
+	}
+	if err := c.Create(ctx, role); err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("create worker Role %s/%s: %w", namespace, miniserviceWorkerSAName, err)
+	}
+
+	rb := &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{Name: miniserviceWorkerSAName, Namespace: namespace},
+		RoleRef: rbacv1.RoleRef{
+			APIGroup: "rbac.authorization.k8s.io",
+			Kind:     "Role",
+			Name:     miniserviceWorkerSAName,
+		},
+		Subjects: []rbacv1.Subject{
+			{Kind: "ServiceAccount", Name: miniserviceWorkerSAName, Namespace: namespace},
+		},
+	}
+	if err := c.Create(ctx, rb); err != nil && !apierrors.IsAlreadyExists(err) {
+		return fmt.Errorf("create worker RoleBinding %s/%s: %w", namespace, miniserviceWorkerSAName, err)
+	}
+
+	return nil
+}
+
+// injectWorkerTokenVolume assigns the worker ServiceAccount to pod, adds the projected SAT
+// volume, and injects worker identity env vars into all non-init containers.
+// The token audience is "nvcf-icms:<clusterID>" with a 900-second expiry.
+// The volume is mounted read-only at /var/run/secrets/tokens in all non-init containers.
+func injectWorkerTokenVolume(pod *corev1.Pod, clusterID string) {
+	pod.Spec.ServiceAccountName = miniserviceWorkerSAName
+	audience := "nvcf-icms:" + clusterID
+
+	volume := corev1.Volume{
+		Name: miniserviceWorkerTokenVolumeName,
+		VolumeSource: corev1.VolumeSource{
+			Projected: &corev1.ProjectedVolumeSource{
+				Sources: []corev1.VolumeProjection{
+					{
+						ServiceAccountToken: &corev1.ServiceAccountTokenProjection{
+							Audience:          audience,
+							ExpirationSeconds: &miniserviceWorkerTokenExpirationSeconds,
+							Path:              "token",
+						},
+					},
+				},
+			},
+		},
+	}
+	pod.Spec.Volumes = append(pod.Spec.Volumes, volume)
+
+	mount := corev1.VolumeMount{
+		Name:      miniserviceWorkerTokenVolumeName,
+		MountPath: miniserviceWorkerTokenMountPath,
+		ReadOnly:  true,
+	}
+	envVars := []corev1.EnvVar{
+		{Name: miniserviceWorkerTokenFilePathEnvKey, Value: miniserviceWorkerTokenFilePath},
+		{Name: miniserviceWorkerIdentitySourceEnvKey, Value: miniserviceWorkerIdentitySourcePSAT},
+	}
+
+	for i := range pod.Spec.Containers {
+		pod.Spec.Containers[i].VolumeMounts = append(pod.Spec.Containers[i].VolumeMounts, mount)
+		pod.Spec.Containers[i].Env = append(pod.Spec.Containers[i].Env, envVars...)
+	}
+}

src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "queue_manager.go",
         "transport_tls.go",
         "validator_summary_reconciler.go",
+        "worker_identity.go",
         "workloadwatcher.go",
     ],
     importpath = "github.com/NVIDIA/nvcf/src/compute-plane-services/nvca/pkg/nvca",
@@ -179,6 +180,7 @@ go_test(
         "queue_manager_test.go",
         "transport_tls_test.go",
         "validator_summary_reconciler_test.go",
+        "worker_identity_test.go",
         "workloadwatcher_test.go",
     ],
     data = glob(["testdata/**"]),

src/compute-plane-services/nvca/pkg/nvca/agent_manager.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/core"
 	cmnhttp "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/http"
+	nvcaconfig "github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/types/nvca/config"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -200,6 +201,9 @@ func startControllerManagerForAgent(
 			CustomAnnotations:          a.backendk8scache.customAnnotations,
 			Kartas:                     kartas,
 			NsightProfilingAllowlist:   a.backendk8scache.nsightProfilingAllowlist,
+			WorkerIdentityEnabled: a.FeatureFlagFetcher.IsFeatureFlagEnabled(featureflag.SelfHosted) &&
+				a.AgentOptions.Config.Authz.ClusterIssuedTokenSource == nvcaconfig.ClusterIssuedTokenSourcePSAT,
+			ClusterID: a.ClusterID,
 		},
 	); err != nil {
 		log.WithError(err).Error("Failed to create miniservice controller")

src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go

@@ -225,6 +225,7 @@ type BackendK8sCache struct {
 	// Environment variable overrides for workloads
 	functionEnvOverrides map[string]string
 	taskEnvOverrides     map[string]string
+
 }
 
 // BackendK8sCacheBuilder builds Backendk8sCache and start related edge K8s

src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go

@@ -1022,6 +1022,12 @@ func (c K8sComputeBackend) CreatePodArtifactInstances(ctx context.Context, pod *
 		setTerminationGracePeriodIfNotSet(pod)
 		k8sutil.ApplyCustomAnnotations(pod, c.bk8s.customAnnotations)
 
+		// Container function and task pods use the legacy NVCF-issued worker token
+		// (NVCF_WORKER_TOKEN env var). PSAT-based delegated worker identity is not
+		// provisioned here. When container workloads are migrated to the MiniService
+		// controller, they will inherit PSAT provisioning automatically via the
+		// MiniService reconciler.
+
 		if _, err := c.clients.K8s.CoreV1().Pods(pod.Namespace).Create(ctx, pod, metav1.CreateOptions{}); err != nil {
 			if !apierrors.IsAlreadyExists(err) {
 				return nil, fmt.Errorf("failed to create instance for Request %v/%v, err: %v", req.Namespace, req.Name, err)

src/compute-plane-services/nvca/pkg/types/types.go

@@ -253,6 +253,20 @@ const (
 	ErrorSourceTaskContainer = "task_container"
 )
 
+// WorkerIdentifier identifies a single worker pod by name and UID.
+type WorkerIdentifier struct {
+	Name string `json:"name"`
+	UID  string `json:"uid"`
+}
+
+// WorkerAuth carries the worker identity set that ICMS stores for a given instance.
+// Populated by NVCA when oidcClusterIdentity is enabled.
+type WorkerAuth struct {
+	Sub               string             `json:"sub"`
+	SAuid             string             `json:"saUid,omitempty"`
+	WorkerIdentifiers []WorkerIdentifier `json:"workerIdentifiers"`
+}
+
 type ICMSInstanceStatusUpdateRequest struct {
 	Status           ICMSRequestStatus        `json:"status,omitempty"`
 	InstanceState    ICMSInstanceState        `json:"instanceState,omitempty"`
@@ -263,6 +277,7 @@ type ICMSInstanceStatusUpdateRequest struct {
 	SystemFailure    string                   `json:"systemFailure,omitempty"`
 	MessageBatchID   string                   `json:"messageBatchId,omitempty"`
 	InstanceIPs      []string                 `json:"instanceIps,omitempty"`
+	WorkerAuth       *WorkerAuth              `json:"workerAuth,omitempty"`
 }
 
 type ICMSRequestUpdateInfo struct {

src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook.go

@@ -143,6 +143,10 @@ func (v *helmMiniServiceValWebhookHandler) validateUpdate(ctx context.Context, _
 func (v *helmMiniServiceValWebhookHandler) validate(ctx context.Context, obj client.Object) (warnings admission.Warnings, err error) {
 	var errs []error
 
+	// REQ-220: workload pods must never run as a worker ServiceAccount regardless of
+	// AllowWorkloadKubernetesAPIAccess, to prevent Helm charts from forging worker tokens.
+	errs = append(errs, validateWorkerSARestriction(obj)...)
+
 	if shouldEnforceResourceLimits(v.fff, obj) {
 		warns, verrs := v.validateResourceLimits(ctx, obj)
 		warnings = append(warnings, warns...)
@@ -152,6 +156,32 @@ func (v *helmMiniServiceValWebhookHandler) validate(ctx context.Context, obj cli
 	return warnings, errors.Join(errs...)
 }
 
+// validateWorkerSARestriction rejects any pod-bearing resource that specifies a worker
+// ServiceAccount (name prefix "nvcf-worker-") as its service account.
+func validateWorkerSARestriction(obj client.Object) []error {
+	var ps *corev1.PodSpec
+	switch t := obj.(type) {
+	case *corev1.Pod:
+		ps = &t.Spec
+	case *appsv1.Deployment:
+		ps = &t.Spec.Template.Spec
+	case *appsv1.ReplicaSet:
+		ps = &t.Spec.Template.Spec
+	case *appsv1.StatefulSet:
+		ps = &t.Spec.Template.Spec
+	case *batchv1.Job:
+		ps = &t.Spec.Template.Spec
+	case *batchv1.CronJob:
+		ps = &t.Spec.JobTemplate.Spec.Template.Spec
+	default:
+		return nil
+	}
+	if strings.HasPrefix(ps.ServiceAccountName, "nvcf-worker-") {
+		return []error{fmt.Errorf("workload pods may not use worker ServiceAccounts (prefix \"nvcf-worker-\")")}
+	}
+	return nil
+}
+
 func (v *helmMiniServiceValWebhookHandler) validateResourceLimits(ctx context.Context, obj client.Object) (
 	warnings admission.Warnings,
 	errs []error,

src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook_test.go

@@ -662,6 +662,73 @@ func TestValidateResourceLimitsVariousObjects(t *testing.T) {
 	}
 }
 
+func TestValidateWorkerSARestriction(t *testing.T) {
+	workerSA := "nvcf-worker-inst-001"
+	regularSA := "helm-instance-permissions"
+
+	tests := []struct {
+		name    string
+		obj     client.Object
+		wantErr bool
+	}{
+		{
+			name: "pod with worker SA is rejected",
+			obj: &corev1.Pod{
+				Spec: corev1.PodSpec{ServiceAccountName: workerSA},
+			},
+			wantErr: true,
+		},
+		{
+			name: "pod with regular SA is allowed",
+			obj: &corev1.Pod{
+				Spec: corev1.PodSpec{ServiceAccountName: regularSA},
+			},
+			wantErr: false,
+		},
+		{
+			name: "deployment with worker SA is rejected",
+			obj: &appsv1.Deployment{
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{ServiceAccountName: workerSA},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "job with worker SA is rejected",
+			obj: &batchv1.Job{
+				Spec: batchv1.JobSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{ServiceAccountName: workerSA},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "pod with empty SA is allowed",
+			obj: &corev1.Pod{
+				Spec: corev1.PodSpec{},
+			},
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			errs := validateWorkerSARestriction(tt.obj)
+			if tt.wantErr {
+				require.NotEmpty(t, errs, "expected validation error")
+				assert.Contains(t, errs[0].Error(), "nvcf-worker-")
+			} else {
+				assert.Empty(t, errs)
+			}
+		})
+	}
+}
+
 func TestValidateContainerLimits_DisallowedResource(t *testing.T) {
 	badResName := corev1.ResourceName("example.com/foo")
 	pod := corev1.Pod{