feat(worker): prefer mounted projected ServiceAccount token for NVCF/NVCT auth

src/libraries/go/worker/auth/auth.go

@@ -28,6 +28,7 @@ import (
 
 type allowInsecureOauthTokenSource struct {
 	oauth.TokenSource
+	requireTLS bool
 }
 
 func (ts allowInsecureOauthTokenSource) GetRequestMetadata(context.Context, ...string) (map[string]string, error) {
@@ -40,13 +41,19 @@ func (ts allowInsecureOauthTokenSource) GetRequestMetadata(context.Context, ...s
 	}, nil
 }
 
+// RequireTransportSecurity is false for legacy NVCF-issued tokens, which may travel over
+// in-cluster plaintext, and true for a mounted projected ServiceAccount token, which must
+// only be presented over TLS.
 func (ts allowInsecureOauthTokenSource) RequireTransportSecurity() bool {
-	return false
+	return ts.requireTLS
 }
 
-func GrpcTokenFromSource(ts oauth2.TokenSource) grpc.CallOption {
+// GrpcTokenFromSource attaches ts as per-RPC bearer credentials. requireTLS must be true
+// when ts serves a mounted projected ServiceAccount token.
+func GrpcTokenFromSource(ts oauth2.TokenSource, requireTLS bool) grpc.CallOption {
 	return grpc.PerRPCCredentials(allowInsecureOauthTokenSource{
-		oauth.TokenSource{TokenSource: ts},
+		TokenSource: oauth.TokenSource{TokenSource: ts},
+		requireTLS:  requireTLS,
 	})
 }
 

src/libraries/go/worker/auth/auth_test.go

@@ -36,7 +36,7 @@ type stubTokenSource struct {
 func (s stubTokenSource) Token() (*oauth2.Token, error) { return s.tok, s.err }
 
 func newSource(ts oauth2.TokenSource) allowInsecureOauthTokenSource {
-	return allowInsecureOauthTokenSource{oauth.TokenSource{TokenSource: ts}}
+	return allowInsecureOauthTokenSource{TokenSource: oauth.TokenSource{TokenSource: ts}}
 }
 
 func TestGetRequestMetadata(t *testing.T) {
@@ -54,10 +54,18 @@ func TestGetRequestMetadata(t *testing.T) {
 	})
 }
 
-func TestRequireTransportSecurityIsFalse(t *testing.T) {
-	// Deliberate posture: worker RPCs carry tokens over in-cluster plaintext.
-	// Pin it so it is not flipped accidentally.
-	require.False(t, newSource(stubTokenSource{}).RequireTransportSecurity())
+func TestRequireTransportSecurity(t *testing.T) {
+	t.Run("legacy NVCF-issued tokens may travel over in-cluster plaintext", func(t *testing.T) {
+		require.False(t, newSource(stubTokenSource{}).RequireTransportSecurity())
+	})
+	t.Run("mounted projected ServiceAccount tokens require TLS", func(t *testing.T) {
+		src := allowInsecureOauthTokenSource{TokenSource: oauth.TokenSource{TokenSource: stubTokenSource{}}, requireTLS: true}
+		require.True(t, src.RequireTransportSecurity())
+	})
+	t.Run("GrpcTokenFromSource propagates the flag", func(t *testing.T) {
+		require.NotNil(t, GrpcTokenFromSource(stubTokenSource{}, true))
+		require.NotNil(t, GrpcTokenFromSource(stubTokenSource{}, false))
+	})
 }
 
 func TestSettableTokenSource(t *testing.T) {

src/libraries/go/worker/nvcf/client.go

@@ -84,6 +84,9 @@ type Client struct {
 	assertionTokenPath      string
 	sharedConfigDir         string
 	clientTimeout           time.Duration
+	// delegatedToken is true when the credential is a mounted projected ServiceAccount
+	// token. NVCF then issues no replacement token and the client persists none.
+	delegatedToken bool
 
 	// for keeping nvcf region state
 	ConnectedRegions       atomic.Pointer[ConnectionRegions]
@@ -128,8 +131,16 @@ func CreateClient(nvcfFqdn string, nvcfFqdnNats *string, nvcfWorkerToken string,
 
 	tokenProvider := auth.NewSettableTokenSource(oauth2.StaticTokenSource(nvcfToken))
 
+	// Prefer a mounted projected ServiceAccount Token (PSAT) over the bootstrap token
+	// when running on a self-hosted cluster with worker identity enabled.
+	delegatedToken, err := token.SelectMountedToken(tokenProvider, nvcfFqdn, "NVCF")
+	if err != nil {
+		return nil, err
+	}
+
 	client := &Client{
-		Client: workerClient,
+		delegatedToken: delegatedToken,
+		Client:         workerClient,
 		regionalNvcfClients: map[string]pb.WorkerClient{
 			nvcfFqdn: workerClient,
 		},
@@ -504,7 +515,7 @@ func (c *Client) GetArtifacts(ctx context.Context) (*types.ArtifactsList, error)
 		var internalResources []types.Artifact
 		var internalInvalidArtifacts int
 
-		stream, err := c.Client.StreamArtifacts(ctx, &pb.ArtifactsRequest{}, auth.GrpcTokenFromSource(c.NvcfTokenProvider))
+		stream, err := c.Client.StreamArtifacts(ctx, &pb.ArtifactsRequest{}, auth.GrpcTokenFromSource(c.NvcfTokenProvider, c.delegatedToken))
 		if err != nil {
 			span.AddEvent(fmt.Sprintf("Failed to start streaming artifacts from NVCF: %s", err.Error()))
 			zap.L().Warn("failed to start streaming artifacts from NVCF", zap.Error(err))

src/libraries/go/worker/nvcf/connect.go

@@ -64,13 +64,13 @@ func (c *Client) ConnectIndefinitely(ctx context.Context) (context.Context, erro
 				}
 				utils.SleepWithContext(ctx, sleepDuration)
 			}
-		maxElapsed := time.Until(token.Expiry)
-		if maxElapsed <= 0 {
-			maxElapsed = 5 * time.Minute
-		}
-		err = backoff.Retry(func() error {
-			return c.connect(ctx)
-		}, backoff.WithContext(backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(maxElapsed)), ctx))
+			maxElapsed := time.Until(token.Expiry)
+			if maxElapsed <= 0 {
+				maxElapsed = 5 * time.Minute
+			}
+			err = backoff.Retry(func() error {
+				return c.connect(ctx)
+			}, backoff.WithContext(backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(maxElapsed)), ctx))
 			if err != nil {
 				zap.L().Error("failed to reconnect to NVCF", zap.Error(err))
 				return
@@ -91,14 +91,20 @@ func (c *Client) connect(ctx context.Context) error {
 		InstanceId:        c.instanceId,
 		FunctionId:        c.functionId,
 		FunctionVersionId: c.functionVersionId,
-	}, auth.GrpcTokenFromSource(c.NvcfTokenProvider))
+	}, auth.GrpcTokenFromSource(c.NvcfTokenProvider, c.delegatedToken))
 	if err != nil {
 		return fmt.Errorf("failed to send connect request to NVCF: %w", err)
 	}
 	if connected.ConnectedRegion == "" {
 		return fmt.Errorf("nvcf did not respond with a connected region")
 	}
 	c.updateConnectedRegions(connected.ConnectedRegion, connected.OtherRegions)
+	if c.delegatedToken {
+		// The mounted JWT stays the credential for the life of the process; NVCF issues no
+		// replacement token on this path and nothing is persisted.
+		zap.L().Info("connected to NVCF", zap.String("region", connected.ConnectedRegion), zap.Strings("secondaryRegions", connected.OtherRegions))
+		return nil
+	}
 	oauthToken := &oauth2.Token{
 		AccessToken: connected.NvcfWorkerToken,
 		Expiry:      connected.Expiration.AsTime(),

src/libraries/go/worker/nvcf/ess.go

@@ -56,7 +56,7 @@ func (c *Client) getAssertionToken(ctx context.Context) (token.Token, error) {
 		resp, err = c.Client.RequestSecretCredentials(ctx, &pb.SecretCredentialsRequest{
 			FunctionId:        c.functionId,
 			FunctionVersionId: c.functionVersionId,
-		}, auth.GrpcTokenFromSource(c.NvcfTokenProvider))
+		}, auth.GrpcTokenFromSource(c.NvcfTokenProvider, c.delegatedToken))
 		if err != nil {
 			return err
 		}

src/libraries/go/worker/nvcf/metadata.go

@@ -56,7 +56,7 @@ func (c *Client) getMetadataCredentials(ctx context.Context) (token.Token, error
 				NcaId:             c.ncaId,
 				FunctionId:        c.functionId,
 				FunctionVersionId: c.functionVersionId,
-			}, auth.GrpcTokenFromSource(c.NvcfTokenProvider))
+			}, auth.GrpcTokenFromSource(c.NvcfTokenProvider, c.delegatedToken))
 		return err
 	}, backoff.WithContext(backoff.WithMaxRetries(backoff.NewExponentialBackOff(), 10), ctx))
 

src/libraries/go/worker/nvcf/psat_client_test.go

@@ -0,0 +1,142 @@
+/*
+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 nvcf
+
+import (
+	"context"
+	"encoding/base64"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/oauth2"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	pb "github.com/NVIDIA/nvcf/src/libraries/go/worker/proto/nvcf"
+	"github.com/NVIDIA/nvcf/src/libraries/go/worker/token"
+)
+
+func fakePSAT(exp int64) string {
+	header := base64.RawURLEncoding.EncodeToString([]byte(`{"alg":"RS256","typ":"JWT"}`))
+	claims := base64.RawURLEncoding.EncodeToString([]byte(
+		`{"sub":"system:serviceaccount:inst-1:nvcf-worker","aud":["nvcf-icms:cl-1"],"exp":` +
+			strconv.FormatInt(exp, 10) + `}`))
+	return header + "." + claims + ".fakesig"
+}
+
+// mountPSAT writes a fake PSAT under a temporary allowed root and points the env var at it.
+func mountPSAT(t *testing.T) string {
+	t.Helper()
+	root, err := filepath.EvalSymlinks(t.TempDir())
+	require.NoError(t, err)
+	old := token.MountedTokenRoot
+	token.MountedTokenRoot = root + "/"
+	t.Cleanup(func() { token.MountedTokenRoot = old })
+	path := filepath.Join(root, "token")
+	jwt := fakePSAT(time.Now().Add(15 * time.Minute).Unix())
+	require.NoError(t, os.WriteFile(path, []byte(jwt), 0600))
+	t.Setenv(token.MountedTokenPathEnvKey, path)
+	return jwt
+}
+
+func TestCreateClient_NoMountedJWT_UsesBootstrap(t *testing.T) {
+	t.Setenv(token.MountedTokenPathEnvKey, filepath.Join(t.TempDir(), "absent"))
+	fqdn := startMockServer(t, &mockWorkerServer{})
+
+	client, err := CreateClient(fqdn, nil, "bootstrap-token", nil, "nca", "inst", "fn", "fnv", t.TempDir(), DefaultNvcfClientTimeout)
+	require.NoError(t, err)
+	assert.False(t, client.delegatedToken)
+	tok, err := client.NvcfTokenProvider.Token()
+	require.NoError(t, err)
+	assert.Equal(t, "bootstrap-token", tok.AccessToken)
+}
+
+func TestCreateClient_MountedJWT_RequiresHTTPS(t *testing.T) {
+	mountPSAT(t)
+	fqdn := startMockServer(t, &mockWorkerServer{}) // http://
+
+	_, err := CreateClient(fqdn, nil, "bootstrap-token", nil, "nca", "inst", "fn", "fnv", t.TempDir(), DefaultNvcfClientTimeout)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "requires TLS")
+}
+
+func TestCreateClient_MountedJWT_PreferredOverBootstrapAndCache(t *testing.T) {
+	jwt := mountPSAT(t)
+	sharedDir := t.TempDir()
+	require.NoError(t, token.CacheToken(filepath.Join(sharedDir, cachedNvcfTokenFilename),
+		&oauth2.Token{AccessToken: "cached-token", Expiry: time.Now().Add(time.Hour)}))
+
+	client, err := CreateClient("https://127.0.0.1:1", nil, "bootstrap-token", nil, "nca", "inst", "fn", "fnv", sharedDir, DefaultNvcfClientTimeout)
+	require.NoError(t, err)
+	assert.True(t, client.delegatedToken)
+	tok, err := client.NvcfTokenProvider.Token()
+	require.NoError(t, err)
+	assert.Equal(t, jwt, tok.AccessToken, "mounted JWT wins over cached and bootstrap tokens")
+}
+
+func TestCreateClient_MountedJWT_UnreadableIsAnError(t *testing.T) {
+	if os.Geteuid() == 0 {
+		t.Skip("root can read mode 0000 files")
+	}
+	mountPSAT(t)
+	require.NoError(t, os.Chmod(os.Getenv(token.MountedTokenPathEnvKey), 0000))
+
+	_, err := CreateClient("https://127.0.0.1:1", nil, "bootstrap-token", nil, "nca", "inst", "fn", "fnv", t.TempDir(), DefaultNvcfClientTimeout)
+	require.Error(t, err)
+	assert.False(t, strings.Contains(err.Error(), "no mounted JWT"), "read failures must not be treated as no token mounted")
+}
+
+// delegatedConnectClient answers ConnectOnce in-process. A real plaintext gRPC connection
+// would reject the per-RPC credentials because a mounted JWT requires TLS.
+type delegatedConnectClient struct {
+	pb.WorkerClient
+}
+
+func (delegatedConnectClient) ConnectOnce(context.Context, *pb.WorkerConnect, ...grpc.CallOption) (*pb.WorkerConnectOnceResponse, error) {
+	return &pb.WorkerConnectOnceResponse{
+		ConnectedRegion: "us-east-1",
+		NvcfWorkerToken: "", // NVCF issues no replacement token on the delegated path
+		Expiration:      timestamppb.New(time.Now().Add(15 * time.Minute)),
+	}, nil
+}
+
+func TestConnect_DelegatedToken_KeepsPSATAndPersistsNothing(t *testing.T) {
+	fqdn := startMockServer(t, &mockWorkerServer{})
+	c := newTestClient(t, fqdn)
+	c.Client = delegatedConnectClient{}
+	c.delegatedToken = true
+	before, err := c.NvcfTokenProvider.Token()
+	require.NoError(t, err)
+
+	require.NoError(t, c.connect(context.Background()))
+
+	after, err := c.NvcfTokenProvider.Token()
+	require.NoError(t, err)
+	assert.Equal(t, before.AccessToken, after.AccessToken, "PSAT source must remain installed")
+	_, statErr := os.Stat(filepath.Join(c.sharedConfigDir, cachedNvcfTokenFilename))
+	assert.True(t, os.IsNotExist(statErr), "no token may be persisted on the delegated path")
+	regions := c.ConnectedRegions.Load()
+	require.NotNil(t, regions)
+	assert.Equal(t, "us-east-1", regions.Primary)
+}

src/libraries/go/worker/nvct/artifacts.go

@@ -40,7 +40,7 @@ func (c *Client) GetArtifacts(ctx context.Context) (*types.ArtifactsList, error)
 		var err error
 		response, err = c.Client.GetArtifacts(ctx, &pb.ArtifactsRequest{
 			TaskId: c.taskId,
-		}, auth.GrpcTokenFromSource(c.NvctTokenProvider))
+		}, auth.GrpcTokenFromSource(c.NvctTokenProvider, c.delegatedToken))
 		if err != nil {
 			zap.L().Warn("failed to get artifacts from NVCT", zap.Error(err))
 		}

src/libraries/go/worker/nvct/client.go

@@ -56,6 +56,9 @@ type Client struct {
 	instanceType            string
 	clientTimeout           time.Duration
 	sharedConfigDir         string
+	// delegatedToken is true when the credential is a mounted projected ServiceAccount
+	// token. NVCT then issues no replacement token and the client persists none.
+	delegatedToken bool
 }
 
 func CreateClient(nvctFqdn string, nvctWorkerToken string, instanceId string, taskId string, instanceType string, nvctClientTimeout time.Duration, sharedConfigDir string) (*Client, error) {
@@ -100,12 +103,22 @@ func CreateClient(nvctFqdn string, nvctWorkerToken string, instanceId string, ta
 		zap.L().Info("no cached token found - using environment token")
 	}
 
+	nvctTokenProvider := auth.NewSettableTokenSource(oauth2.StaticTokenSource(nvctToken))
+
+	// Prefer a mounted projected ServiceAccount Token (PSAT) over the bootstrap token
+	// when running on a self-hosted cluster with worker identity enabled.
+	delegatedToken, err := token.SelectMountedToken(nvctTokenProvider, nvctFqdn, "NVCT")
+	if err != nil {
+		return nil, err
+	}
+
 	return &Client{
-		Client: workerClient,
+		delegatedToken: delegatedToken,
+		Client:         workerClient,
 		regionalNvctClients: map[string]pb.WorkerClient{
 			nvctFqdn: workerClient,
 		},
-		NvctTokenProvider: auth.NewSettableTokenSource(oauth2.StaticTokenSource(nvctToken)),
+		NvctTokenProvider: nvctTokenProvider,
 		instanceId:        instanceId,
 		taskId:            taskId,
 		instanceType:      instanceType,

src/libraries/go/worker/nvct/client_test.go

@@ -29,8 +29,8 @@ import (
 	"testing"
 	"time"
 
-	"go.uber.org/zap"
 	"github.com/NVIDIA/nvcf/src/libraries/go/lib/pkg/nvkit/logs"
+	"go.uber.org/zap"
 
 	"github.com/NVIDIA/nvcf/src/libraries/go/worker/proto/nvct"
 	"github.com/NVIDIA/nvcf/src/libraries/go/worker/test/testutils"
@@ -176,6 +176,7 @@ func TestSendResultMetadata(t *testing.T) {
 		ctx,
 		mockClient.Client,
 		mockClient.NvctTokenProvider,
+		false,
 	)
 	defer func() {
 		if err := mockStreamingClient.Close(); err != nil {

src/libraries/go/worker/nvct/connect.go

@@ -47,7 +47,7 @@ func (c *Client) Connect(ctx context.Context) error {
 		connectRespnse, connectErr := c.Client.Connect(ctx, &pb.ConnectRequest{
 			InstanceId: c.instanceId,
 			TaskId:     c.taskId,
-		}, auth.GrpcTokenFromSource(c.NvctTokenProvider))
+		}, auth.GrpcTokenFromSource(c.NvctTokenProvider, c.delegatedToken))
 		if connectErr != nil {
 			zap.L().Error("failed to connect to NVCT", zap.Error(connectErr))
 			return connectErr