vault: integrate linking service identity resolution (#21715)

* vault: link org ids via org resolver

* vault: verify workflow owner org linkage

* vault: log org identity linking decisions

* vault: use get secrets identity fields

* vault: resolve get secrets identity from metadata

* vault: clean up linker lint issues

* chore: run gomodtidy

* vault: gate linker behind org id owner flag

* vault: tighten get secrets owner checks

* vault: fail closed without request identity linker

* vault: inline secret owner validation switch
Replace secretOwnerValidationTarget struct with a direct switch
statement for clearer, more readable owner validation logic.

* forward org id through workflow request metadata

* apply gomodtidy updates

* gate workflow org id forwarding and simplify vault execute

* simplify vault execute and clean workflow lint

* fix workflow v2 secrets integration test

* fix workflow org resolver test setup

* simplify workflow org resolution lifecycle

* gate workflow org identity propagation

* fix workflow v2 gate-aware tests

* refine vault linking and workflow org gating

* Address workflow review comments

Author: prashantkumar1982

core/capabilities/vault/capability.go

@@ -16,6 +16,7 @@ import (
 	vaultcommon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/actions/vault"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities/consensus/requests"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	"github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
@@ -31,6 +32,7 @@ type Capability struct {
 	handler              *requests.Handler[*vaulttypes.Request, *vaulttypes.Response]
 	capabilitiesRegistry core.CapabilitiesRegistry
 	publicKey            *LazyPublicKey
+	linker               *OrgIDToWorkflowOwnerLinker
 	*RequestValidator
 }
 
@@ -69,6 +71,9 @@ func (s *Capability) Close() error {
 	if lerr := s.MaxRequestBatchSizeLimiter.Close(); lerr != nil {
 		err = errors.Join(err, fmt.Errorf("error closing request batch size limiter: %w", lerr))
 	}
+	if lerr := s.linker.Close(); lerr != nil {
+		err = errors.Join(err, fmt.Errorf("error closing org_id linker: %w", lerr))
+	}
 
 	return err
 }
@@ -110,13 +115,13 @@ func (s *Capability) Execute(ctx context.Context, request capabilities.Capabilit
 		return capabilities.CapabilityResponse{}, errors.New("no secret request specified in request")
 	}
 
-	normalizedWorkflowOwner := normalizeOwner(request.Metadata.WorkflowOwner)
 	for idx, req := range r.Requests {
 		if req == nil { // defensive: protobuf strips nil elements, but guard against in-process callers
+			s.lggr.Errorw("get secrets request contains nil secret request", "index", idx)
 			return capabilities.CapabilityResponse{}, fmt.Errorf("nil secret request at index %d", idx)
 		}
-
-		if req.Id != nil && normalizeOwner(req.Id.Owner) != normalizedWorkflowOwner {
+		if req.Id != nil && normalizeOwner(req.Id.Owner) != normalizeOwner(request.Metadata.WorkflowOwner) {
+			s.lggr.Errorw("get secrets request owner mismatch", "index", idx, "secretOwner", req.Id.Owner, "workflowOwner", request.Metadata.WorkflowOwner)
 			return capabilities.CapabilityResponse{}, fmt.Errorf("secret identifier owner %q does not match workflow owner %q at index %d", req.Id.Owner, request.Metadata.WorkflowOwner, idx)
 		}
 	}
@@ -163,6 +168,12 @@ func (s *Capability) CreateSecrets(ctx context.Context, request *vaultcommon.Cre
 		s.lggr.Debugf("RequestId: [%s] failed validation checks: %s", request.RequestId, err.Error())
 		return nil, err
 	}
+	resolvedIdentity, err := s.resolveRequestIdentity(ctx, request.OrgId, request.WorkflowOwner)
+	if err != nil {
+		return nil, err
+	}
+	request.OrgId = resolvedIdentity.OrgID
+	request.WorkflowOwner = resolvedIdentity.WorkflowOwner
 	return s.handleRequest(ctx, request.RequestId, request)
 }
 
@@ -173,6 +184,12 @@ func (s *Capability) UpdateSecrets(ctx context.Context, request *vaultcommon.Upd
 		s.lggr.Debugf("RequestId: [%s] failed validation checks: %s", request.RequestId, err.Error())
 		return nil, err
 	}
+	resolvedIdentity, err := s.resolveRequestIdentity(ctx, request.OrgId, request.WorkflowOwner)
+	if err != nil {
+		return nil, err
+	}
+	request.OrgId = resolvedIdentity.OrgID
+	request.WorkflowOwner = resolvedIdentity.WorkflowOwner
 	return s.handleRequest(ctx, request.RequestId, request)
 }
 
@@ -183,6 +200,12 @@ func (s *Capability) DeleteSecrets(ctx context.Context, request *vaultcommon.Del
 		s.lggr.Debugf("Request: [%s] failed validation checks: %s", request.String(), err.Error())
 		return nil, err
 	}
+	resolvedIdentity, err := s.resolveRequestIdentity(ctx, request.OrgId, request.WorkflowOwner)
+	if err != nil {
+		return nil, err
+	}
+	request.OrgId = resolvedIdentity.OrgID
+	request.WorkflowOwner = resolvedIdentity.WorkflowOwner
 	return s.handleRequest(ctx, request.RequestId, request)
 }
 
@@ -204,6 +227,12 @@ func (s *Capability) ListSecretIdentifiers(ctx context.Context, request *vaultco
 		s.lggr.Debugf("Request: [%s] failed validation checks: %s", request.String(), err.Error())
 		return nil, err
 	}
+	resolvedIdentity, err := s.resolveRequestIdentity(ctx, request.OrgId, request.WorkflowOwner)
+	if err != nil {
+		return nil, err
+	}
+	request.OrgId = resolvedIdentity.OrgID
+	request.WorkflowOwner = resolvedIdentity.WorkflowOwner
 	return s.handleRequest(ctx, request.RequestId, request)
 }
 
@@ -256,19 +285,37 @@ func (s *Capability) handleRequest(ctx context.Context, requestID string, reques
 	}
 }
 
+// resolveRequestIdentity validates and normalizes the org/workflow-owner pair that the vault plugin consumes.
+func (s *Capability) resolveRequestIdentity(ctx context.Context, orgID string, workflowOwner string) (LinkedVaultRequestIdentity, error) {
+	s.lggr.Debugw("resolving request identity", "orgID", orgID, "workflowOwner", workflowOwner)
+	linked, err := s.linker.Link(ctx, orgID, workflowOwner)
+	if err != nil {
+		s.lggr.Errorw("failed to resolve request identity", "orgID", orgID, "workflowOwner", workflowOwner, "err", err)
+		return LinkedVaultRequestIdentity{}, err
+	}
+	s.lggr.Debugw("resolved request identity", "orgID", linked.OrgID, "workflowOwner", linked.WorkflowOwner)
+
+	return linked, nil
+}
+
 func NewCapability(
 	lggr logger.Logger,
 	clock clockwork.Clock,
 	expiresAfter time.Duration,
 	handler *requests.Handler[*vaulttypes.Request, *vaulttypes.Response],
 	capabilitiesRegistry core.CapabilitiesRegistry,
 	publicKey *LazyPublicKey,
+	orgResolver orgresolver.OrgResolver,
 	limitsFactory limits.Factory,
 ) (*Capability, error) {
 	limiter, err := limits.MakeBoundLimiter(limitsFactory, cresettings.Default.VaultRequestBatchSizeLimit)
 	if err != nil {
 		return nil, fmt.Errorf("could not create request batch size limiter: %w", err)
 	}
+	linker, err := NewOrgIDToWorkflowOwnerLinker(orgResolver, limitsFactory)
+	if err != nil {
+		return nil, err
+	}
 	ciphertextLimiter, err := limits.MakeUpperBoundLimiter(limitsFactory, cresettings.Default.VaultCiphertextSizeLimit)
 	if err != nil {
 		return nil, fmt.Errorf("could not create ciphertext size limiter: %w", err)
@@ -280,6 +327,7 @@ func NewCapability(
 		handler:              handler,
 		capabilitiesRegistry: capabilitiesRegistry,
 		publicKey:            publicKey,
+		linker:               linker,
 		RequestValidator:     NewRequestValidator(limiter, ciphertextLimiter),
 	}, nil
 }