Merge origin/develop into codex/vault-request-authorizer-in-gw

Author: prashantkumar1982

core/capabilities/remote/executable/request/server_request.go

@@ -208,6 +208,11 @@ func (e *ServerRequest) Expired() bool {
 	return time.Since(e.createdTime) > e.requestTimeout
 }
 
+func (e *ServerRequest) Evictable(minRetention time.Duration) bool {
+	age := time.Since(e.createdTime)
+	return age > e.requestTimeout && age > minRetention
+}
+
 func (e *ServerRequest) Cancel(ctx context.Context, err types.Error, msg string) error {
 	e.mux.Lock()
 	defer e.mux.Unlock()

core/capabilities/remote/executable/request/server_request_test.go

@@ -338,6 +338,44 @@ func Test_ServerRequest_MessageValidation(t *testing.T) {
 	})
 }
 
+func Test_ServerRequest_Evictable(t *testing.T) {
+	lggr := logger.Test(t)
+	capability := TestCapability{}
+	capabilityPeerID := NewP2PPeerID(t)
+	workflowPeer := NewP2PPeerID(t)
+
+	callingDon := commoncap.DON{
+		Members: []p2ptypes.PeerID{workflowPeer},
+		ID:      1,
+		F:       0,
+	}
+
+	newRequest := func(requestTimeout time.Duration) *request.ServerRequest {
+		req, err := request.NewServerRequest(capability, types.MethodExecute, "capabilityID", 2,
+			capabilityPeerID, callingDon, "requestMessageID", &testDispatcher{}, requestTimeout, "", lggr)
+		require.NoError(t, err)
+		return req
+	}
+
+	t.Run("expired but below minimum retention", func(t *testing.T) {
+		req := newRequest(20 * time.Millisecond)
+		time.Sleep(60 * time.Millisecond)
+		assert.False(t, req.Evictable(200*time.Millisecond))
+	})
+
+	t.Run("expired and retained past minimum retention", func(t *testing.T) {
+		req := newRequest(20 * time.Millisecond)
+		time.Sleep(60 * time.Millisecond)
+		assert.True(t, req.Evictable(10*time.Millisecond))
+	})
+
+	t.Run("minimum retention elapsed but request timeout still active", func(t *testing.T) {
+		req := newRequest(200 * time.Millisecond)
+		time.Sleep(60 * time.Millisecond)
+		assert.False(t, req.Evictable(10*time.Millisecond))
+	})
+}
+
 type serverRequest interface {
 	OnMessage(ctx context.Context, msg *types.MessageBody) error
 }

core/capabilities/remote/executable/server.go

@@ -222,7 +222,7 @@ func (r *server) expireRequests() {
 	defer r.receiveLock.Unlock()
 
 	for requestID, executeReq := range r.requestIDToRequest {
-		if executeReq.request.Expired() {
+		if executeReq.request.Evictable(commoncap.DefaultExecutableRequestTimeout) {
 			ctx, cancelFn := r.stopCh.NewCtx()
 			err := executeReq.request.Cancel(ctx, types.Error_TIMEOUT, "request expired by executable server")
 			cancelFn()

core/capabilities/remote/executable/server_test.go

@@ -1000,3 +1000,101 @@ func Test_Server_Execute_WithConcurrentSetConfig(t *testing.T) {
 	expectedResponses := numWorkflowPeers * numExecuteCalls
 	require.Equal(t, expectedResponses, successCount)
 }
+
+func Test_Server_DuplicateRequestRemainsDedupedPastRequestTimeout(t *testing.T) {
+	ctx := testutils.Context(t)
+	lggr := logger.Test(t)
+
+	serverPeerID := NewP2PPeerID(t)
+	senderPeerID := NewP2PPeerID(t)
+
+	dispatcher := &noopDispatcher{}
+	server := executable.NewServer("cap_id@1.0.0", "", serverPeerID, dispatcher, lggr)
+
+	cfg := &commoncap.RemoteExecutableConfig{
+		RequestTimeout:            20 * time.Millisecond,
+		ServerMaxParallelRequests: 1,
+	}
+	capInfo := commoncap.CapabilityInfo{
+		ID:             "cap_id@1.0.0",
+		CapabilityType: commoncap.CapabilityTypeTarget,
+	}
+	localDON := commoncap.DON{
+		ID:      1,
+		Members: []p2ptypes.PeerID{serverPeerID},
+		F:       0,
+	}
+	workflowDONs := map[uint32]commoncap.DON{
+		2: {
+			ID:      2,
+			Members: []p2ptypes.PeerID{senderPeerID},
+			F:       0,
+		},
+	}
+
+	require.NoError(t, server.SetConfig(cfg, TestCapability{}, capInfo, localDON, workflowDONs, nil))
+	require.NoError(t, server.Start(ctx))
+	defer func() {
+		require.NoError(t, server.Close())
+	}()
+
+	inputs, err := values.NewMap(map[string]any{"executeValue1": "aValue1"})
+	require.NoError(t, err)
+	rawRequest, err := pb.MarshalCapabilityRequest(commoncap.CapabilityRequest{
+		Metadata: commoncap.RequestMetadata{
+			WorkflowExecutionID: "exec-1",
+		},
+		Inputs: inputs,
+	})
+	require.NoError(t, err)
+
+	msg := &remotetypes.MessageBody{
+		CapabilityId:    capInfo.ID,
+		CapabilityDonId: localDON.ID,
+		CallerDonId:     2,
+		Method:          remotetypes.MethodExecute,
+		Payload:         rawRequest,
+		MessageId:       []byte(remotetypes.MethodExecute + ":exec-1"),
+		Sender:          senderPeerID[:],
+		Receiver:        serverPeerID[:],
+	}
+
+	server.Receive(ctx, msg)
+	require.Eventually(t, func() bool { return len(dispatcher.sent) == 1 }, time.Second, 10*time.Millisecond)
+
+	time.Sleep(2 * cfg.RequestTimeout)
+	server.Receive(ctx, msg)
+
+	time.Sleep(100 * time.Millisecond)
+	require.Len(t, dispatcher.sent, 1)
+}
+
+type noopDispatcher struct {
+	services.StateMachine
+	sent []*remotetypes.MessageBody
+}
+
+func (n *noopDispatcher) Name() string { return "noopDispatcher" }
+
+func (n *noopDispatcher) Start(context.Context) error { return nil }
+
+func (n *noopDispatcher) Close() error { return nil }
+
+func (n *noopDispatcher) Ready() error { return nil }
+
+func (n *noopDispatcher) HealthReport() map[string]error { return nil }
+
+func (n *noopDispatcher) SetReceiver(string, uint32, remotetypes.Receiver) error { return nil }
+
+func (n *noopDispatcher) RemoveReceiver(string, uint32) {}
+
+func (n *noopDispatcher) SetReceiverForMethod(string, uint32, string, remotetypes.Receiver) error {
+	return nil
+}
+
+func (n *noopDispatcher) RemoveReceiverForMethod(string, uint32, string) {}
+
+func (n *noopDispatcher) Send(peerID p2ptypes.PeerID, msgBody *remotetypes.MessageBody) error {
+	n.sent = append(n.sent, msgBody)
+	return nil
+}

core/capabilities/vault/gw_handler.go

@@ -2,16 +2,13 @@ package vault
 
 import (
 	"context"
-	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"sort"
 	"strings"
 
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
-	"google.golang.org/protobuf/proto"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	vaultcommon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/actions/vault"
@@ -63,22 +60,20 @@ type GatewayHandler struct {
 	services.Service
 	eng *services.Engine
 
-	capRegistry       core.CapabilitiesRegistry
 	secretsService    vaulttypes.SecretsService
 	gatewayConnector  gatewayConnector
 	requestAuthorizer RequestAuthorizer
 	lggr              logger.Logger
 	metrics           *metrics
 }
 
-func NewGatewayHandler(capabilitiesRegistry core.CapabilitiesRegistry, secretsService vaulttypes.SecretsService, connector gatewayConnector, requestAuthorizer RequestAuthorizer, lggr logger.Logger) (*GatewayHandler, error) {
+func NewGatewayHandler(secretsService vaulttypes.SecretsService, connector gatewayConnector, requestAuthorizer RequestAuthorizer, lggr logger.Logger) (*GatewayHandler, error) {
 	metrics, err := newMetrics()
 	if err != nil {
 		return nil, fmt.Errorf("failed to create metrics: %w", err)
 	}
 
 	gh := &GatewayHandler{
-		capRegistry:       capabilitiesRegistry,
 		secretsService:    secretsService,
 		gatewayConnector:  connector,
 		requestAuthorizer: requestAuthorizer,
@@ -112,7 +107,7 @@ func (h *GatewayHandler) ID(ctx context.Context) (string, error) {
 }
 
 func (h *GatewayHandler) Methods() []string {
-	return vaulttypes.GetSupportedMethods(h.lggr)
+	return vaulttypes.Methods
 }
 
 func (h *GatewayHandler) HandleGatewayMessage(ctx context.Context, gatewayID string, req *jsonrpc.Request[json.RawMessage]) (err error) {
@@ -127,8 +122,6 @@ func (h *GatewayHandler) HandleGatewayMessage(ctx context.Context, gatewayID str
 			break
 		}
 		response = h.handleSecretsCreate(ctx, gatewayID, req, owner)
-	case vaulttypes.MethodSecretsGet:
-		response = h.handleSecretsGet(ctx, gatewayID, req)
 	case vaulttypes.MethodSecretsUpdate:
 		owner, authErr := h.authorizeAndPrefixRequest(ctx, req)
 		if authErr != nil {
@@ -256,51 +249,6 @@ func (h *GatewayHandler) handleSecretsUpdate(ctx context.Context, gatewayID stri
 	return jsonResponse
 }
 
-func (h *GatewayHandler) handleSecretsGet(ctx context.Context, gatewayID string, req *jsonrpc.Request[json.RawMessage]) *jsonrpc.Response[json.RawMessage] {
-	var request vaultcommon.GetSecretsRequest
-	if err := json.Unmarshal(*req.Params, &request); err != nil {
-		return h.errorResponse(ctx, gatewayID, req, api.UserMessageParseError, err)
-	}
-	encryptionKeys, err := h.getEncryptionKeys(ctx)
-	if err != nil {
-		return h.errorResponse(ctx, gatewayID, req, api.FatalError, err)
-	}
-	getSecretsRequest := vaultcommon.GetSecretsRequest{}
-	for _, reqItem := range request.Requests {
-		getSecretsRequest.Requests = append(getSecretsRequest.Requests, &vaultcommon.SecretRequest{
-			Id: &vaultcommon.SecretIdentifier{
-				Owner:     reqItem.Id.Owner,
-				Namespace: reqItem.Id.Namespace,
-				Key:       reqItem.Id.Key,
-			},
-			EncryptionKeys: encryptionKeys,
-		})
-	}
-	vaultCapResponse, err := h.secretsService.GetSecrets(ctx, req.ID, &getSecretsRequest)
-	if err != nil {
-		return h.errorResponse(ctx, gatewayID, req, api.FatalError, err)
-	}
-
-	vaultResponseProto := &vaultcommon.GetSecretsResponse{}
-	err = proto.Unmarshal(vaultCapResponse.Payload, vaultResponseProto)
-	if err != nil {
-		h.lggr.Errorf("Debugging: handleSecretsCreate failed to unmarshal response: %s. Payload was: %s", err.Error(), string(vaultCapResponse.Payload))
-		return h.errorResponse(ctx, gatewayID, req, api.NodeReponseEncodingError, err)
-	}
-
-	vaultAPIResponseBytes, err := json.Marshal(vaultResponseProto)
-	if err != nil {
-		return h.errorResponse(ctx, gatewayID, req, api.NodeReponseEncodingError, err)
-	}
-	vaultAPIResponseJSON := json.RawMessage(vaultAPIResponseBytes)
-	return &jsonrpc.Response[json.RawMessage]{
-		Version: jsonrpc.JsonRpcVersion,
-		ID:      req.ID,
-		Method:  req.Method,
-		Result:  &vaultAPIResponseJSON,
-	}
-}
-
 func (h *GatewayHandler) handleSecretsDelete(ctx context.Context, gatewayID string, req *jsonrpc.Request[json.RawMessage], owner string) *jsonrpc.Response[json.RawMessage] {
 	r := &vaultcommon.DeleteSecretsRequest{}
 	if err := json.Unmarshal(*req.Params, r); err != nil {
@@ -408,26 +356,6 @@ func (h *GatewayHandler) errorResponse(
 	}
 }
 
-// getEncryptionKeys retrieves the encryption keys of all members in the Workflow DON.
-func (h *GatewayHandler) getEncryptionKeys(ctx context.Context) ([]string, error) {
-	myNode, err := h.capRegistry.LocalNode(ctx)
-	if err != nil {
-		return nil, errors.New("failed to get local node from registry" + err.Error())
-	}
-
-	encryptionKeys := make([]string, 0, len(myNode.WorkflowDON.Members))
-	for _, peerID := range myNode.WorkflowDON.Members {
-		peerNode, err := h.capRegistry.NodeByPeerID(ctx, peerID)
-		if err != nil {
-			return nil, errors.New("failed to get node info for peerID: " + peerID.String() + " - " + err.Error())
-		}
-		encryptionKeys = append(encryptionKeys, hex.EncodeToString(peerNode.EncryptionPublicKey[:]))
-	}
-	// Sort the encryption keys to ensure consistent ordering across all nodes.
-	sort.Strings(encryptionKeys)
-	return encryptionKeys, nil
-}
-
 func toJSONResponse(vaultCapResponse *vaulttypes.Response, method string) (*jsonrpc.Response[json.RawMessage], error) {
 	vaultResponseBytes, err := vaultCapResponse.ToJSONRPCResult()
 	if err != nil {

core/capabilities/vault/gw_handler_test.go

@@ -11,7 +11,6 @@ import (
 
 	vaultcommon "github.com/smartcontractkit/chainlink-common/pkg/capabilities/actions/vault"
 	jsonrpc "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2"
-	core_mocks "github.com/smartcontractkit/chainlink-common/pkg/types/core/mocks"
 	vaultcap "github.com/smartcontractkit/chainlink/v2/core/capabilities/vault"
 	vaultcapmocks "github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/mocks"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/vaulttypes"
@@ -162,7 +161,6 @@ func TestGatewayHandler_HandleGatewayMessage(t *testing.T) {
 						RequestId: "test-secret",
 						Ids: []*vaultcommon.SecretIdentifier{
 							{
-
 								Key:       "Foo",
 								Namespace: "Bar",
 								Owner:     "0xAbC",
@@ -243,12 +241,11 @@ func TestGatewayHandler_HandleGatewayMessage(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			secretsService := vaulttypesmocks.NewSecretsService(t)
 			gwConnector := connector_mocks.NewGatewayConnector(t)
-			capRegistry := core_mocks.NewCapabilitiesRegistry(t)
 			requestAuthorizer := vaultcapmocks.NewRequestAuthorizer(t)
 
 			tt.setupMocks(secretsService, gwConnector, requestAuthorizer)
 
-			handler, err := vaultcap.NewGatewayHandler(capRegistry, secretsService, gwConnector, requestAuthorizer, lggr)
+			handler, err := vaultcap.NewGatewayHandler(secretsService, gwConnector, requestAuthorizer, lggr)
 			require.NoError(t, err)
 
 			err = handler.HandleGatewayMessage(ctx, "gateway-1", tt.request)
@@ -268,20 +265,19 @@ func TestGatewayHandler_Lifecycle(t *testing.T) {
 
 	secretsService := vaulttypesmocks.NewSecretsService(t)
 	gwConnector := connector_mocks.NewGatewayConnector(t)
-	capRegistry := core_mocks.NewCapabilitiesRegistry(t)
 	requestAuthorizer := vaultcapmocks.NewRequestAuthorizer(t)
 
-	handler, err := vaultcap.NewGatewayHandler(capRegistry, secretsService, gwConnector, requestAuthorizer, lggr)
+	handler, err := vaultcap.NewGatewayHandler(secretsService, gwConnector, requestAuthorizer, lggr)
 	require.NoError(t, err)
 
 	t.Run("start", func(t *testing.T) {
-		gwConnector.On("AddHandler", mock.Anything, vaulttypes.GetSupportedMethods(lggr), handler).Return(nil).Once()
+		gwConnector.On("AddHandler", mock.Anything, vaulttypes.Methods, handler).Return(nil).Once()
 		err := handler.Start(ctx)
 		require.NoError(t, err)
 	})
 
 	t.Run("close", func(t *testing.T) {
-		gwConnector.On("RemoveHandler", mock.Anything, vaulttypes.GetSupportedMethods(lggr)).Return(nil).Once()
+		gwConnector.On("RemoveHandler", mock.Anything, vaulttypes.Methods).Return(nil).Once()
 		err := handler.Close()
 		require.NoError(t, err)
 	})