fix(runs): address review — gate header trust, guard enrichment, bound cache

- Add Config.TrustForwardedIdentityHeaders (default true). executed_by is derived
  from the proxy-forwarded, unverified JWTs only when set; turn off where the
  service may be reached without a trusted proxy.
- Enricher: reject userinfo responses whose `sub` doesn't match the caller (token
  confusion), and evict expired cache entries (lazily + swept on store) so the map
  can't grow unbounded.
- Fix the misleading "never blocks" doc — userinfo is a synchronous call on cache miss.
- created_by column: VARCHAR(255) -> TEXT (OIDC sub length is IdP-dependent).
- Clarify created_by is intentionally not in the API filter allowlist.
- Tests: actionMetadataFromModel (executed_by populate / created_by fallback /
  corrupt bytes / none) and enricher subject-mismatch rejection.

Signed-off-by: Kevin Su <pingsutw@apache.org>

Author: pingsutw

runs/config/config.go

@@ -32,6 +32,9 @@ var defaultConfig = &Config{
 		ExecutionQPS:          10.0,
 		ExecutionBurst:        20,
 	},
+	// Defaults on: the runs service is designed to sit behind an auth proxy/LB. Set
+	// false for deployments where that guarantee does not hold.
+	TrustForwardedIdentityHeaders: true,
 }
 
 var configSection = config.MustRegisterSection(configSectionKey, defaultConfig)
@@ -68,6 +71,17 @@ type Config struct {
 	// AuthMetadata configures the OAuth2 authorization-server metadata endpoint
 	// (the GetOAuth2Metadata RPC and /.well-known/oauth-authorization-server).
 	AuthMetadata AuthMetadataConfig `json:"authMetadata"`
+
+	// TrustForwardedIdentityHeaders controls whether run attribution (executed_by)
+	// is derived from the auth headers the proxy forwards (X-Amzn-Oidc-*, Authorization).
+	// Those JWTs are decoded but NOT signature-verified here — that is safe only when
+	// the service sits behind a trusted proxy/LB that validates tokens and strips any
+	// client-supplied copies of these headers. Set to false if the service can be
+	// reached directly or through an untrusted proxy, in which case executed_by is left
+	// unset rather than risk a spoofed identity. (Note: the runs service performs no
+	// authorization itself, so a direct caller can already act unauthenticated — this
+	// flag only governs whether to trust the forwarded identity for attribution.)
+	TrustForwardedIdentityHeaders bool `json:"trustForwardedIdentityHeaders" pflag:",Derive executed_by from proxy-forwarded auth headers (requires a trusted proxy)"`
 }
 
 // AuthMetadataConfig controls how the runs service serves OAuth2 authorization

runs/migrations/sql/20260618120000_add_actions_created_by.sql

@@ -1,4 +1,5 @@
 -- Add created_by to actions: the OIDC subject of the identity that created the run.
 -- Captured from the auth headers the load balancer forwards (it enforces auth),
 -- and used to populate ActionMetadata.executed_by on read.
-ALTER TABLE actions ADD COLUMN IF NOT EXISTS created_by VARCHAR(255);
+-- TEXT, not VARCHAR(n): the OIDC `sub` length is IdP-dependent and can exceed 255.
+ALTER TABLE actions ADD COLUMN IF NOT EXISTS created_by TEXT;

runs/repository/models/action.go

@@ -53,8 +53,9 @@ type Action struct {
 	RunSource string `db:"run_source" json:"run_source,omitempty"`
 
 	// CreatedBy is the OIDC subject of the identity that created this run, captured
-	// from the auth headers the load balancer forwards. Kept for querying/filtering.
-	// NULL for runs created without an authenticated identity.
+	// from the auth headers the load balancer forwards. NULL for runs created without
+	// an authenticated identity. Not exposed for API-level filtering/sorting — it is
+	// intentionally absent from ActionColumnsSet; add it there only if that's desired.
 	CreatedBy sql.NullString `db:"created_by" json:"created_by,omitempty"`
 
 	// ExecutedBy is the serialized common.EnrichedIdentity of the run's creator

runs/service/executed_by_test.go

@@ -0,0 +1,55 @@
+package service
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/flyteorg/flyte/v2/gen/go/flyteidl2/common"
+	"github.com/flyteorg/flyte/v2/runs/repository/models"
+)
+
+func mustMarshalIdentity(t *testing.T, id *common.EnrichedIdentity) []byte {
+	t.Helper()
+	b, err := proto.Marshal(id)
+	require.NoError(t, err)
+	return b
+}
+
+func fullIdentity(sub, first, last, email string) *common.EnrichedIdentity {
+	return &common.EnrichedIdentity{Principal: &common.EnrichedIdentity_User{User: &common.User{
+		Id:   &common.UserIdentifier{Subject: sub},
+		Spec: &common.UserSpec{FirstName: first, LastName: last, Email: email},
+	}}}
+}
+
+func TestActionMetadataFromModel_ExecutedBy(t *testing.T) {
+	t.Run("full identity from executed_by", func(t *testing.T) {
+		m := &models.Action{ExecutedBy: mustMarshalIdentity(t, fullIdentity("00u1", "Kevin", "Su", "kevin@union.ai"))}
+		eb := actionMetadataFromModel(m).GetExecutedBy().GetUser()
+		assert.Equal(t, "00u1", eb.GetId().GetSubject())
+		assert.Equal(t, "Kevin", eb.GetSpec().GetFirstName())
+		assert.Equal(t, "Su", eb.GetSpec().GetLastName())
+		assert.Equal(t, "kevin@union.ai", eb.GetSpec().GetEmail())
+	})
+
+	t.Run("falls back to subject-only from created_by", func(t *testing.T) {
+		m := &models.Action{}
+		m.CreatedBy.Valid, m.CreatedBy.String = true, "00u2"
+		eb := actionMetadataFromModel(m).GetExecutedBy().GetUser()
+		assert.Equal(t, "00u2", eb.GetId().GetSubject())
+		assert.Nil(t, eb.GetSpec())
+	})
+
+	t.Run("corrupt executed_by falls back to created_by", func(t *testing.T) {
+		m := &models.Action{ExecutedBy: []byte("not a valid proto\xff\xfe")}
+		m.CreatedBy.Valid, m.CreatedBy.String = true, "00u3"
+		assert.Equal(t, "00u3", actionMetadataFromModel(m).GetExecutedBy().GetUser().GetId().GetSubject())
+	})
+
+	t.Run("no identity yields nil executed_by", func(t *testing.T) {
+		assert.Nil(t, actionMetadataFromModel(&models.Action{}).GetExecutedBy())
+	})
+}

runs/service/identity_enricher.go

@@ -55,8 +55,10 @@ func newIdentityEnricher(authServerBaseURL string) *identityEnricher {
 // enrich fills any profile fields (email, first/last name) missing from base with
 // userinfo claims fetched using the access token. Fields already present on base
 // (e.g. from x-amzn-oidc-data) are authoritative and kept. userinfo is queried only
-// when the profile is incomplete and not cached. base is returned unchanged on any
-// miss or error — enrichment never blocks or fails run creation.
+// when the profile is incomplete and not cached. On a cache miss it makes a
+// synchronous userinfo call (bounded by userinfoHTTPTimeout), which adds latency to
+// run creation; on any error or timeout it returns base unchanged — enrichment is
+// best-effort and never fails run creation.
 func (e *identityEnricher) enrich(ctx context.Context, accessToken string, base *common.EnrichedIdentity) *common.EnrichedIdentity {
 	if e == nil || base.GetUser() == nil {
 		return base
@@ -77,22 +79,41 @@ func (e *identityEnricher) enrich(ctx context.Context, accessToken string, base
 		logger.Warnf(ctx, "identity enrichment: userinfo fetch failed for subject %q: %v", subject, err)
 		return base
 	}
+	// Guard against token confusion / IdP misconfiguration: never associate a profile
+	// fetched for a different subject with this run.
+	if claims.Sub != "" && claims.Sub != subject {
+		logger.Warnf(ctx, "identity enrichment: userinfo subject %q does not match caller %q; ignoring", claims.Sub, subject)
+		return base
+	}
 	e.store(subject, claims)
 	return mergeClaims(base, claims)
 }
 
 func (e *identityEnricher) cachedFor(subject string) *oidcClaims {
 	e.mu.Lock()
 	defer e.mu.Unlock()
-	if c, ok := e.cache[subject]; ok && time.Now().Before(c.expires) {
-		return c.claims
+	c, ok := e.cache[subject]
+	if !ok {
+		return nil
 	}
-	return nil
+	if time.Now().After(c.expires) {
+		// Drop the stale entry so the map does not accumulate dead keys.
+		delete(e.cache, subject)
+		return nil
+	}
+	return c.claims
 }
 
 func (e *identityEnricher) store(subject string, claims *oidcClaims) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
+	// Opportunistically evict any other expired entries to bound the map size.
+	now := time.Now()
+	for k, c := range e.cache {
+		if now.After(c.expires) {
+			delete(e.cache, k)
+		}
+	}
 	e.cache[subject] = cachedClaims{claims: claims, expires: time.Now().Add(identityCacheTTL)}
 }
 

runs/service/identity_enricher_test.go

@@ -83,6 +83,16 @@ func TestEnrich_FillsOnlyMissingFields(t *testing.T) {
 	assert.Equal(t, "kevin@union.ai", spec.GetEmail()) // header email preserved
 }
 
+func TestEnrich_RejectsSubjectMismatch(t *testing.T) {
+	// userinfo returns a different subject than the caller — must not be trusted.
+	srv, _ := newTestIdP(t, `{"sub":"someone-else","email":"evil@x.com","given_name":"Mallory"}`, http.StatusOK)
+	e := newIdentityEnricher(srv.URL)
+
+	got := e.enrich(context.Background(), "access-tok", subjectOnlyIdentity("00u2"))
+	assert.Nil(t, got.GetUser().GetSpec())
+	assert.Equal(t, "00u2", got.GetUser().GetId().GetSubject())
+}
+
 func TestEnrich_UserinfoErrorFallsBackToBase(t *testing.T) {
 	srv, _ := newTestIdP(t, `nope`, http.StatusUnauthorized)
 	e := newIdentityEnricher(srv.URL)

runs/service/run_service.go

@@ -47,6 +47,8 @@ type RunService struct {
 	dataStore       *storage.DataStore
 	abortReconciler *AbortReconciler
 	enricher        *identityEnricher
+	// trustHeaders gates deriving executed_by from proxy-forwarded auth headers.
+	trustHeaders bool
 }
 
 type actionDataClient interface {
@@ -154,6 +156,7 @@ func NewRunService(
 	dataStore *storage.DataStore,
 	reconciler *AbortReconciler,
 	authServerBaseURL string,
+	trustForwardedIdentityHeaders bool,
 ) *RunService {
 	return &RunService{
 		repo:            repo,
@@ -164,6 +167,7 @@ func NewRunService(
 		dataStore:       dataStore,
 		abortReconciler: reconciler,
 		enricher:        newIdentityEnricher(authServerBaseURL),
+		trustHeaders:    trustForwardedIdentityHeaders,
 	}
 }
 
@@ -331,8 +335,13 @@ func (s *RunService) CreateRun(
 	// Capture who created the run from the auth headers the load balancer forwards
 	// (it enforces auth upstream). nil when there is no authenticated identity. On the
 	// Bearer path the token carries only the subject, so enrich name/email via userinfo.
-	executedBy := identityFromHeaders(req.Header())
-	executedBy = s.enricher.enrich(ctx, accessTokenFromHeaders(req.Header()), executedBy)
+	// Only trust the proxy-forwarded identity headers when configured to (the JWTs are
+	// decoded, not signature-verified — see Config.TrustForwardedIdentityHeaders).
+	var executedBy *common.EnrichedIdentity
+	if s.trustHeaders {
+		executedBy = identityFromHeaders(req.Header())
+		executedBy = s.enricher.enrich(ctx, accessTokenFromHeaders(req.Header()), executedBy)
+	}
 
 	// Persist task spec and create a run model
 	run, err := s.persistRunModel(ctx, runId, taskID, taskSpec, inputPrefix, runOutputBase, runSpec, request.GetSource(), triggerName, triggerTaskName, triggerRevision, triggerType, executedBy)

runs/setup.go

@@ -93,7 +93,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
 		return abortReconciler.Run(ctx)
 	})
 
-	runsSvc := service.NewRunService(repo, actionsClient, dataProxyClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL)
+	runsSvc := service.NewRunService(repo, actionsClient, dataProxyClient, projectClient, cfg.StoragePrefix, sc.DataStore, abortReconciler, cfg.AuthMetadata.ExternalAuthServerBaseURL, cfg.TrustForwardedIdentityHeaders)
 	taskSvc := service.NewTaskService(repo, projectClient)
 
 	runsPath, runsHandler := workflowconnect.NewRunServiceHandler(runsSvc, connect.WithInterceptors(otelInterceptor))

runs/test/api/setup_test.go

@@ -113,7 +113,7 @@ func TestMain(m *testing.M) {
 	// Create RunService with a no-op actions client (points at test server; not used by watch tests)
 	endpointURL := fmt.Sprintf("http://localhost:%d", testPort)
 	actionsClient := actionsconnect.NewActionsServiceClient(http.DefaultClient, endpointURL)
-	runSvc := service.NewRunService(repo, actionsClient, nil, nil, "", nil, nil, "")
+	runSvc := service.NewRunService(repo, actionsClient, nil, nil, "", nil, nil, "", true)
 
 	// Setup HTTP server
 	mux := http.NewServeMux()