diff --git a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel b/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel index c1b28ec60..21d47342f 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/BUILD.bazel +++ b/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", diff --git a/src/compute-plane-services/nvca/internal/miniservice/controller.go b/src/compute-plane-services/nvca/internal/miniservice/controller.go index 1e744691b..4467b0a53 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/controller.go +++ b/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 } diff --git a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go b/src/compute-plane-services/nvca/internal/miniservice/reconcile.go index df5796538..6a92f4c37 100644 --- a/src/compute-plane-services/nvca/internal/miniservice/reconcile.go +++ b/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) { diff --git a/src/compute-plane-services/nvca/internal/miniservice/worker_identity.go b/src/compute-plane-services/nvca/internal/miniservice/worker_identity.go new file mode 100644 index 000000000..73cbd6775 --- /dev/null +++ b/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:" 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...) + } +} diff --git a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel b/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel index 2c6d647e4..0ec2c5076 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/BUILD.bazel +++ b/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/**"]), diff --git a/src/compute-plane-services/nvca/pkg/nvca/agent_manager.go b/src/compute-plane-services/nvca/pkg/nvca/agent_manager.go index 477bbc9c9..86170272f 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/agent_manager.go +++ b/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") diff --git a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go b/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go index 7a6ffd961..e748e5042 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/backendk8scache.go +++ b/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 diff --git a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go b/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go index f0d18f43a..51c1965d5 100644 --- a/src/compute-plane-services/nvca/pkg/nvca/k8scomputebackend.go +++ b/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) diff --git a/src/compute-plane-services/nvca/pkg/types/types.go b/src/compute-plane-services/nvca/pkg/types/types.go index 4a81ee52c..45242cafd 100644 --- a/src/compute-plane-services/nvca/pkg/types/types.go +++ b/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 { diff --git a/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook.go b/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook.go index eb69bf207..f551dbb08 100644 --- a/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook.go +++ b/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, diff --git a/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook_test.go b/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook_test.go index 4290134d7..14421b89f 100644 --- a/src/compute-plane-services/nvca/pkg/webhook/helm_mini_service_validate_webhook_test.go +++ b/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{