Implement upstream_inject strategy and SubjectProviderName

cmd/thv-operator/controllers/virtualmcpserver_controller.go

@@ -447,6 +447,7 @@ func (r *VirtualMCPServerReconciler) handleSpecValidationError(
 	ctxLogger := log.FromContext(ctx)
 	if applyErr := r.applyStatusUpdates(ctx, vmcp, statusManager); applyErr != nil {
 		ctxLogger.Error(applyErr, "Failed to apply status updates after spec validation error")
+		return applyErr
 	}
 	return nil
 }
@@ -651,33 +652,12 @@ func (r *VirtualMCPServerReconciler) ensureAllResources(
 ) (ctrl.Result, error) {
 	ctxLogger := log.FromContext(ctx)
 
-	// Validate secret references before creating resources
-	// This catches configuration errors early, providing faster feedback than waiting for pod startup failures
-	if err := r.validateSecretReferences(ctx, vmcp); err != nil {
-		ctxLogger.Error(err, "Secret validation failed")
-		// Set AuthConfigured condition to False
-		statusManager.SetAuthConfiguredCondition(
-			mcpv1alpha1.ConditionReasonAuthInvalid,
-			fmt.Sprintf("Authentication configuration is invalid: %v", err),
-			metav1.ConditionFalse,
-		)
-		statusManager.SetObservedGeneration(vmcp.Generation)
-		// Record event for secret validation failure
-		if r.Recorder != nil {
-			r.Recorder.Eventf(vmcp, nil, corev1.EventTypeWarning, "SecretValidationFailed", "ValidateSecrets",
-				"Secret validation failed: %v", err)
-		}
+	// Validate secret references before creating resources.
+	// This catches configuration errors early, providing faster feedback than waiting for pod startup failures.
+	if err := r.ensureAuthSecretsValid(ctx, vmcp, statusManager); err != nil {
 		return ctrl.Result{}, err
 	}
 
-	// Authentication secrets validated successfully
-	statusManager.SetAuthConfiguredCondition(
-		mcpv1alpha1.ConditionReasonAuthValid,
-		"Authentication configuration is valid",
-		metav1.ConditionTrue,
-	)
-	statusManager.SetObservedGeneration(vmcp.Generation)
-
 	// Check EmbeddingServer readiness before proceeding to Deployment.
 	// RequeueAfter provides a safety net in case the Watches() events
 	// are missed (e.g., EmbeddingServer controller not running).
@@ -731,10 +711,14 @@ func (r *VirtualMCPServerReconciler) ensureAllResources(
 	// Ensure vmcp Config ConfigMap.
 	// handleSpecValidationError converts SpecValidationError to nil (no requeue)
 	// after applying status conditions, while passing through transient errors.
-	if err := r.handleSpecValidationError(ctx, vmcp, statusManager,
-		r.ensureVmcpConfigConfigMap(ctx, vmcp, workloadNames, statusManager)); err != nil {
-		ctxLogger.Error(err, "Failed to ensure vmcp Config ConfigMap")
-		return ctrl.Result{}, err
+	if specValidationErr := r.ensureVmcpConfigConfigMap(ctx, vmcp, workloadNames, statusManager); specValidationErr != nil {
+		if err := r.handleSpecValidationError(ctx, vmcp, statusManager, specValidationErr); err != nil {
+			ctxLogger.Error(err, "Failed to ensure vmcp Config ConfigMap")
+			return ctrl.Result{}, err
+		}
+		// SpecValidationError: status applied, stop reconciliation without requeue.
+		// Do not proceed to ensureDeployment — the ConfigMap was not created/updated.
+		return ctrl.Result{}, nil
 	}
 
 	// Ensure Deployment
@@ -756,6 +740,37 @@ func (r *VirtualMCPServerReconciler) ensureAllResources(
 	return ctrl.Result{}, nil
 }
 
+// ensureAuthSecretsValid validates secret references and sets the AuthConfigured condition.
+func (r *VirtualMCPServerReconciler) ensureAuthSecretsValid(
+	ctx context.Context,
+	vmcp *mcpv1alpha1.VirtualMCPServer,
+	statusManager virtualmcpserverstatus.StatusManager,
+) error {
+	if err := r.validateSecretReferences(ctx, vmcp); err != nil {
+		ctxLogger := log.FromContext(ctx)
+		ctxLogger.Error(err, "Secret validation failed")
+		statusManager.SetAuthConfiguredCondition(
+			mcpv1alpha1.ConditionReasonAuthInvalid,
+			fmt.Sprintf("Authentication configuration is invalid: %v", err),
+			metav1.ConditionFalse,
+		)
+		statusManager.SetObservedGeneration(vmcp.Generation)
+		if r.Recorder != nil {
+			r.Recorder.Eventf(vmcp, nil, corev1.EventTypeWarning, "SecretValidationFailed", "ValidateSecrets",
+				"Secret validation failed: %v", err)
+		}
+		return err
+	}
+
+	statusManager.SetAuthConfiguredCondition(
+		mcpv1alpha1.ConditionReasonAuthValid,
+		"Authentication configuration is valid",
+		metav1.ConditionTrue,
+	)
+	statusManager.SetObservedGeneration(vmcp.Generation)
+	return nil
+}
+
 // ensureRBACResources ensures RBAC resources for VirtualMCPServer.
 // RBAC resources are created in all modes (discovered and inline) to support:
 // - Backend discovery (discovered mode only)

deploy/charts/operator-crds/files/crds/toolhive.stacklok.dev_virtualmcpservers.yaml

@@ -1449,6 +1449,12 @@ spec:
                                   items:
                                     type: string
                                   type: array
+                                subjectProviderName:
+                                  description: |-
+                                    SubjectProviderName selects which upstream provider's token to use as the
+                                    subject token. When set, the token is looked up from Identity.UpstreamTokens
+                                    instead of using Identity.Token.
+                                  type: string
                                 subjectTokenType:
                                   description: |-
                                     SubjectTokenType is the token type of the incoming subject token.
@@ -1538,6 +1544,12 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              subjectProviderName:
+                                description: |-
+                                  SubjectProviderName selects which upstream provider's token to use as the
+                                  subject token. When set, the token is looked up from Identity.UpstreamTokens
+                                  instead of using Identity.Token.
+                                type: string
                               subjectTokenType:
                                 description: |-
                                   SubjectTokenType is the token type of the incoming subject token.

deploy/charts/operator-crds/templates/toolhive.stacklok.dev_virtualmcpservers.yaml

@@ -1452,6 +1452,12 @@ spec:
                                   items:
                                     type: string
                                   type: array
+                                subjectProviderName:
+                                  description: |-
+                                    SubjectProviderName selects which upstream provider's token to use as the
+                                    subject token. When set, the token is looked up from Identity.UpstreamTokens
+                                    instead of using Identity.Token.
+                                  type: string
                                 subjectTokenType:
                                   description: |-
                                     SubjectTokenType is the token type of the incoming subject token.
@@ -1541,6 +1547,12 @@ spec:
                                 items:
                                   type: string
                                 type: array
+                              subjectProviderName:
+                                description: |-
+                                  SubjectProviderName selects which upstream provider's token to use as the
+                                  subject token. When set, the token is looked up from Identity.UpstreamTokens
+                                  instead of using Identity.Token.
+                                type: string
                               subjectTokenType:
                                 description: |-
                                   SubjectTokenType is the token type of the incoming subject token.

pkg/vmcp/auth/factory/outgoing.go

@@ -34,6 +34,7 @@ import (
 //   - "unauthenticated": Default fallback for backends without auth
 //   - "header_injection": Custom HTTP header injection
 //   - "token_exchange": RFC-8693 OAuth 2.0 token exchange
+//   - "upstream_inject": Per-upstream token injection from stored credentials
 //
 // Parameters:
 //   - ctx: Context for any initialization that requires it
@@ -67,6 +68,12 @@ func NewOutgoingAuthRegistry(
 	); err != nil {
 		return nil, err
 	}
+	if err := registry.RegisterStrategy(
+		authtypes.StrategyTypeUpstreamInject,
+		strategies.NewUpstreamInjectStrategy(),
+	); err != nil {
+		return nil, err
+	}
 
 	return registry, nil
 }

pkg/vmcp/auth/factory/outgoing_test.go

@@ -32,6 +32,7 @@ func TestNewOutgoingAuthRegistry(t *testing.T) {
 			authtypes.StrategyTypeUnauthenticated,
 			authtypes.StrategyTypeHeaderInjection,
 			authtypes.StrategyTypeTokenExchange,
+			authtypes.StrategyTypeUpstreamInject,
 		}
 
 		for _, strategyType := range strategyTypes {
@@ -162,6 +163,7 @@ func TestNewOutgoingAuthRegistry(t *testing.T) {
 			{authtypes.StrategyTypeUnauthenticated, "unauthenticated"},
 			{authtypes.StrategyTypeHeaderInjection, "header_injection"},
 			{authtypes.StrategyTypeTokenExchange, "token_exchange"},
+			{authtypes.StrategyTypeUpstreamInject, "upstream_inject"},
 		}
 
 		for _, tc := range testCases {

pkg/vmcp/auth/strategies/tokenexchange.go

@@ -119,13 +119,22 @@ func (s *TokenExchangeStrategy) Authenticate(
 		return fmt.Errorf("no identity found in context")
 	}
 
-	if identity.Token == "" {
-		return fmt.Errorf("identity has no token")
+	var subjectToken string
+	if config.SubjectProviderName != "" {
+		subjectToken = identity.UpstreamTokens[config.SubjectProviderName] // nil map safe in Go
+		if subjectToken == "" {
+			return fmt.Errorf("provider %q: %w", config.SubjectProviderName, authtypes.ErrUpstreamTokenNotFound)
+		}
+	} else {
+		if identity.Token == "" {
+			return fmt.Errorf("identity has no token")
+		}
+		subjectToken = identity.Token
 	}
 
 	// Get user-specific exchange config. This creates a fresh config instance
 	// with the current user's token. The underlying server config is cached.
-	exchangeConfig := s.createUserConfig(config, identity.Token)
+	exchangeConfig := s.createUserConfig(config, subjectToken)
 	tokenSource := exchangeConfig.TokenSource(ctx)
 
 	token, err := tokenSource.Token()
@@ -179,12 +188,13 @@ func (s *TokenExchangeStrategy) Validate(strategy *authtypes.BackendAuthStrategy
 
 // tokenExchangeConfig holds the parsed token exchange configuration.
 type tokenExchangeConfig struct {
-	TokenURL         string
-	ClientID         string
-	ClientSecret     string //nolint:gosec // G117: field legitimately holds sensitive data
-	Audience         string
-	Scopes           []string
-	SubjectTokenType string
+	TokenURL            string
+	ClientID            string
+	ClientSecret        string //nolint:gosec // G117: field legitimately holds sensitive data
+	Audience            string
+	Scopes              []string
+	SubjectTokenType    string
+	SubjectProviderName string
 }
 
 // parseClientSecret parses and validates ClientSecret or ClientSecretEnv from TokenExchangeConfig.
@@ -258,6 +268,9 @@ func (s *TokenExchangeStrategy) parseTokenExchangeConfig(strategy *authtypes.Bac
 		config.SubjectTokenType = normalized
 	}
 
+	// Optional: SubjectProviderName
+	config.SubjectProviderName = tokenExchangeCfg.SubjectProviderName
+
 	return config, nil
 }
 
@@ -337,6 +350,7 @@ func (s *TokenExchangeStrategy) createUserConfig(
 //   - audience: Target audience
 //   - scopes: Requested scopes (sorted for consistency)
 //   - subject_token_type: Type of subject token
+//   - subject_provider_name: Upstream provider for subject token selection
 //
 // Note: No user identity is included - server configs are shared across users.
 func buildCacheKey(config *tokenExchangeConfig) string {
@@ -367,12 +381,19 @@ func buildCacheKey(config *tokenExchangeConfig) string {
 		tokenType = nonePlaceholder
 	}
 
-	// Format: token_url:client_id:audience:scopes:subject_token_type
-	return fmt.Sprintf("%s:%s:%s:%s:%s",
+	// Handle subject_provider_name (empty becomes nonePlaceholder)
+	providerName := config.SubjectProviderName
+	if providerName == "" {
+		providerName = nonePlaceholder
+	}
+
+	// Format: token_url:client_id:audience:scopes:subject_token_type:subject_provider_name
+	return fmt.Sprintf("%s:%s:%s:%s:%s:%s",
 		config.TokenURL,
 		clientID,
 		audience,
 		scopesStr,
 		tokenType,
+		providerName,
 	)
 }

pkg/vmcp/auth/strategies/tokenexchange_test.go

@@ -6,6 +6,7 @@ package strategies
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -46,6 +47,15 @@ func createContextWithIdentity(subject, token string) context.Context {
 	return auth.WithIdentity(context.Background(), createTestIdentity(subject, token))
 }
 
+func createContextWithUpstreamTokens(subject, token string, upstreamTokens map[string]string) context.Context {
+	identity := &auth.Identity{
+		PrincipalInfo:  auth.PrincipalInfo{Subject: subject},
+		Token:          token,
+		UpstreamTokens: upstreamTokens,
+	}
+	return auth.WithIdentity(context.Background(), identity)
+}
+
 func createTokenExchangeStrategy(tokenURL string, opts ...func(*authtypes.TokenExchangeConfig)) *authtypes.BackendAuthStrategy {
 	cfg := &authtypes.TokenExchangeConfig{
 		TokenURL: tokenURL,
@@ -93,6 +103,7 @@ func TestTokenExchangeStrategy_Authenticate(t *testing.T) {
 		setupServer     func() *httptest.Server
 		expectError     bool
 		errorContains   string
+		checkSentinel   bool
 		checkAuthHeader func(t *testing.T, req *http.Request)
 	}{
 		{
@@ -303,6 +314,67 @@ func TestTokenExchangeStrategy_Authenticate(t *testing.T) {
 			expectError:   true,
 			errorContains: "empty access_token",
 		},
+		{
+			name: "exchanges upstream token when SubjectProviderName is set",
+			setupCtx: func() context.Context {
+				return createContextWithUpstreamTokens("upstream-user", "incoming-bearer-token",
+					map[string]string{"github": "github-upstream-token"})
+			},
+			setupServer: func() *httptest.Server {
+				return createSuccessfulTokenServer(t, "backend-token-xxx", func(t *testing.T, r *http.Request) {
+					t.Helper()
+					assert.Equal(t, "github-upstream-token", r.Form.Get("subject_token"),
+						"should use upstream token, not identity.Token")
+				})
+			},
+			strategy: func(server *httptest.Server) *authtypes.BackendAuthStrategy {
+				return createTokenExchangeStrategy(server.URL, func(cfg *authtypes.TokenExchangeConfig) {
+					cfg.SubjectProviderName = "github"
+				})
+			},
+			expectError: false,
+			checkAuthHeader: func(t *testing.T, req *http.Request) {
+				t.Helper()
+				assert.Equal(t, "Bearer backend-token-xxx", req.Header.Get("Authorization"))
+			},
+		},
+		{
+			name: "returns ErrUpstreamTokenNotFound when SubjectProviderName token is missing",
+			setupCtx: func() context.Context {
+				return createContextWithUpstreamTokens("upstream-user", "incoming-bearer-token", nil)
+			},
+			setupServer: func() *httptest.Server {
+				return httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
+					t.Error("token endpoint should not be called")
+				}))
+			},
+			strategy: func(server *httptest.Server) *authtypes.BackendAuthStrategy {
+				return createTokenExchangeStrategy(server.URL, func(cfg *authtypes.TokenExchangeConfig) {
+					cfg.SubjectProviderName = "github"
+				})
+			},
+			expectError:   true,
+			errorContains: "upstream token not found",
+			checkSentinel: true,
+		},
+		{
+			name: "uses identity.Token when SubjectProviderName is empty",
+			setupCtx: func() context.Context {
+				return createContextWithUpstreamTokens("upstream-user", "original-bearer",
+					map[string]string{"github": "upstream-tok"})
+			},
+			setupServer: func() *httptest.Server {
+				return createSuccessfulTokenServer(t, "backend-token-yyy", func(t *testing.T, r *http.Request) {
+					t.Helper()
+					assert.Equal(t, "original-bearer", r.Form.Get("subject_token"),
+						"should use identity.Token, not upstream token")
+				})
+			},
+			strategy: func(server *httptest.Server) *authtypes.BackendAuthStrategy {
+				return createTokenExchangeStrategy(server.URL)
+			},
+			expectError: false,
+		},
 	}
 
 	for _, tt := range tests {
@@ -327,6 +399,10 @@ func TestTokenExchangeStrategy_Authenticate(t *testing.T) {
 			if tt.expectError {
 				require.Error(t, err)
 				assert.Contains(t, err.Error(), tt.errorContains)
+				if tt.checkSentinel {
+					assert.True(t, errors.Is(err, authtypes.ErrUpstreamTokenNotFound),
+						"expected error to wrap ErrUpstreamTokenNotFound, got: %v", err)
+				}
 				return
 			}