From 20fe3feec520151e18f487e1978a9044391c2ed4 Mon Sep 17 00:00:00 2001 From: Nikita Kryuchkov Date: Thu, 19 Mar 2026 14:45:15 +0300 Subject: [PATCH 1/2] message/validation: move signature verification outside validation lock --- message/validation/consensus_validation.go | 43 ++-- message/validation/partial_validation.go | 30 ++- message/validation/validation.go | 12 +- message/validation/validation_lock_test.go | 224 +++++++++++++++++++++ 4 files changed, 281 insertions(+), 28 deletions(-) create mode 100644 message/validation/validation_lock_test.go diff --git a/message/validation/consensus_validation.go b/message/validation/consensus_validation.go index 49ddf21e36..c7ec995b7a 100644 --- a/message/validation/consensus_validation.go +++ b/message/validation/consensus_validation.go @@ -47,28 +47,23 @@ func (mv *messageValidator) validateConsensusMessage( return consensusMessage, err } - state := mv.validatorState(ssvMessage.GetID(), committeeInfo) - - if err := mv.validateQBFTLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, receivedAt, state); err != nil { + if err := mv.verifyConsensusMessageSignatures(signedSSVMessage); err != nil { return consensusMessage, err } - if err := mv.validateQBFTMessageByDutyLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedAt, state); err != nil { - return consensusMessage, err - } + if err := mv.withValidationLock(ssvMessage.GetID(), func() error { + state := mv.validatorState(ssvMessage.GetID(), committeeInfo) - for i := range signedSSVMessage.Signatures { - operatorID := signedSSVMessage.OperatorIDs[i] - signature := signedSSVMessage.Signatures[i] + if err := mv.validateQBFTLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, receivedAt, state); err != nil { + return err + } - if err := mv.signatureVerifier.VerifySignature(operatorID, ssvMessage, signature); err != nil { - e := ErrSignatureVerification - e.innerErr = fmt.Errorf("verify opid: %v signature: %w", operatorID, err) - return consensusMessage, e + if err := mv.validateQBFTMessageByDutyLogic(signedSSVMessage, consensusMessage, committeeInfo, receivedAt, state); err != nil { + return err } - } - if err := mv.updateConsensusState(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, state); err != nil { + return mv.updateConsensusState(signedSSVMessage, consensusMessage, committeeInfo, receivedFrom, state) + }); err != nil { return consensusMessage, err } @@ -171,6 +166,22 @@ func (mv *messageValidator) validateConsensusMessageSemantics( return nil } +func (mv *messageValidator) verifyConsensusMessageSignatures(signedSSVMessage *spectypes.SignedSSVMessage) error { + ssvMessage := signedSSVMessage.SSVMessage + for i := range signedSSVMessage.Signatures { + operatorID := signedSSVMessage.OperatorIDs[i] + signature := signedSSVMessage.Signatures[i] + + if err := mv.signatureVerifier.VerifySignature(operatorID, ssvMessage, signature); err != nil { + e := ErrSignatureVerification + e.innerErr = fmt.Errorf("verify opid: %v signature: %w", operatorID, err) + return e + } + } + + return nil +} + func (mv *messageValidator) validateQBFTLogic( signedSSVMessage *spectypes.SignedSSVMessage, consensusMessage *specqbft.Message, @@ -281,7 +292,9 @@ func (mv *messageValidator) validateQBFTMessageByDutyLogic( } msgSlot := phase0.Slot(consensusMessage.Height) + randaoMsg := false + // Rule: Message must correspond to a known beacon duty when the role requires it. if err := mv.validateBeaconDuty(signedSSVMessage.SSVMessage.GetID().GetRoleType(), msgSlot, committeeInfo.validatorIndices, randaoMsg); err != nil { return err } diff --git a/message/validation/partial_validation.go b/message/validation/partial_validation.go index 25710b94e0..19b3b54e9c 100644 --- a/message/validation/partial_validation.go +++ b/message/validation/partial_validation.go @@ -43,26 +43,37 @@ func (mv *messageValidator) validatePartialSignatureMessage( return partialSignatureMessages, err } - state := mv.validatorState(ssvMessage.GetID(), committeeInfo) - if err := mv.validatePartialSigMessagesByDutyLogic(signedSSVMessage, partialSignatureMessages, committeeInfo, receivedFrom, receivedAt, state); err != nil { + if err := mv.verifyPartialSignatureMessageSignature(signedSSVMessage); err != nil { return partialSignatureMessages, err } - signature := signedSSVMessage.Signatures[0] signer := signedSSVMessage.OperatorIDs[0] - if err := mv.signatureVerifier.VerifySignature(signer, ssvMessage, signature); err != nil { - e := ErrSignatureVerification - e.innerErr = fmt.Errorf("verify opid: %v signature: %w", signer, err) - return partialSignatureMessages, e - } + if err := mv.withValidationLock(ssvMessage.GetID(), func() error { + state := mv.validatorState(ssvMessage.GetID(), committeeInfo) + if err := mv.validatePartialSigMessagesByDutyLogic(signedSSVMessage, partialSignatureMessages, committeeInfo, receivedFrom, receivedAt, state); err != nil { + return err + } - if err := mv.updatePartialSignatureState(partialSignatureMessages, receivedFrom, state, signer, committeeInfo); err != nil { + return mv.updatePartialSignatureState(partialSignatureMessages, receivedFrom, state, signer, committeeInfo) + }); err != nil { return partialSignatureMessages, err } return partialSignatureMessages, nil } +func (mv *messageValidator) verifyPartialSignatureMessageSignature(signedSSVMessage *spectypes.SignedSSVMessage) error { + signature := signedSSVMessage.Signatures[0] + signer := signedSSVMessage.OperatorIDs[0] + if err := mv.signatureVerifier.VerifySignature(signer, signedSSVMessage.SSVMessage, signature); err != nil { + e := ErrSignatureVerification + e.innerErr = fmt.Errorf("verify opid: %v signature: %w", signer, err) + return e + } + + return nil +} + func (mv *messageValidator) validatePartialSignatureMessageSemantics( signedSSVMessage *spectypes.SignedSSVMessage, partialSignatureMessages *spectypes.PartialSignatureMessages, @@ -158,6 +169,7 @@ func (mv *messageValidator) validatePartialSigMessagesByDutyLogic( } randaoMsg := partialSignatureMessages.Type == spectypes.RandaoPartialSig + // Rule: Message must correspond to a known beacon duty when the role requires it. if err := mv.validateBeaconDuty(signedSSVMessage.SSVMessage.GetID().GetRoleType(), messageSlot, committeeInfo.validatorIndices, randaoMsg); err != nil { return err } diff --git a/message/validation/validation.go b/message/validation/validation.go index 66720375a3..5904473569 100644 --- a/message/validation/validation.go +++ b/message/validation/validation.go @@ -189,10 +189,6 @@ func (mv *messageValidator) handleSignedSSVMessage( return decodedMessage, err } - validationMu := mv.getValidationLock(signedSSVMessage.SSVMessage.GetID()) - validationMu.Lock() - defer validationMu.Unlock() - switch signedSSVMessage.SSVMessage.MsgType { case spectypes.SSVConsensusMsgType: consensusMessage, err := mv.validateConsensusMessage(signedSSVMessage, committeeInfo, receivedFrom, receivedAt) @@ -257,6 +253,14 @@ func (mv *messageValidator) getValidationLock(key spectypes.MessageID) *sync.Mut return lock } +func (mv *messageValidator) withValidationLock(key spectypes.MessageID, fn func() error) error { + validationMu := mv.getValidationLock(key) + validationMu.Lock() + defer validationMu.Unlock() + + return fn() +} + func (mv *messageValidator) getCommitteeAndValidatorIndices(msgID spectypes.MessageID) (CommitteeInfo, error) { if mv.committeeRole(msgID.GetRoleType()) { // TODO: add metrics and logs for committee role diff --git a/message/validation/validation_lock_test.go b/message/validation/validation_lock_test.go new file mode 100644 index 0000000000..9571e58233 --- /dev/null +++ b/message/validation/validation_lock_test.go @@ -0,0 +1,224 @@ +package validation + +import ( + "bytes" + "maps" + "slices" + "testing" + "time" + + "github.com/attestantio/go-eth2-client/spec" + "github.com/attestantio/go-eth2-client/spec/phase0" + libp2ptest "github.com/libp2p/go-libp2p/core/test" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" + "go.uber.org/zap/zaptest" + + spectypes "github.com/ssvlabs/ssv-spec/types" + spectestingutils "github.com/ssvlabs/ssv-spec/types/testingutils" + + "github.com/ssvlabs/ssv/network/commons" + "github.com/ssvlabs/ssv/networkconfig" + "github.com/ssvlabs/ssv/operator/duties/dutystore" + "github.com/ssvlabs/ssv/operator/storage" + ssvtypes "github.com/ssvlabs/ssv/protocol/v2/types" + registrystorage "github.com/ssvlabs/ssv/registry/storage" + "github.com/ssvlabs/ssv/registry/storage/mocks" + kv "github.com/ssvlabs/ssv/storage/badger" + "github.com/ssvlabs/ssv/storage/basedb" +) + +type observingSignatureVerifier struct { + called chan struct{} +} + +type validationLockTestEnv struct { + validator *messageValidator + committeeID spectypes.CommitteeID + committeeIdentifier spectypes.MessageID + netCfg *networkconfig.Network + ks *spectestingutils.TestKeySet +} + +func (v *observingSignatureVerifier) VerifySignature(spectypes.OperatorID, *spectypes.SSVMessage, []byte) error { + select { + case v.called <- struct{}{}: + default: + } + + return nil +} + +func newValidationLockTestEnv(t *testing.T) validationLockTestEnv { + ctrl := gomock.NewController(t) + + logger := zaptest.NewLogger(t) + db, err := kv.NewInMemory(logger, basedb.Options{}) + require.NoError(t, err) + + ns, err := storage.NewNodeStorage(networkconfig.TestNetwork.Beacon, logger, db) + require.NoError(t, err) + + netCfg := networkconfig.TestNetwork + ks := spectestingutils.Testing4SharesSet() + shares := generateShares(t, ks, ns, netCfg) + + dutyStore := dutystore.New() + validatorStore := mocks.NewMockValidatorStore(ctrl) + operators := mocks.NewMockOperators(ctrl) + + committee := slices.Collect(maps.Keys(ks.Shares)) + slices.Sort(committee) + + committeeID := shares.active.CommitteeID() + validatorStore.EXPECT().Committee(gomock.Any()).DoAndReturn(func(id spectypes.CommitteeID) (*registrystorage.Committee, bool) { + if id != committeeID { + return nil, false + } + + share1 := cloneSSVShare(t, shares.active) + share2 := cloneSSVShare(t, share1) + share2.ValidatorIndex = share1.ValidatorIndex + 1 + share3 := cloneSSVShare(t, share2) + share3.ValidatorIndex = share2.ValidatorIndex + 1 + + return ®istrystorage.Committee{ + ID: id, + Operators: committee, + Shares: []*ssvtypes.SSVShare{ + share1, + share2, + share3, + }, + Indices: []phase0.ValidatorIndex{ + share1.ValidatorIndex, + share2.ValidatorIndex, + share3.ValidatorIndex, + }, + }, true + }).AnyTimes() + + for _, id := range []spectypes.OperatorID{1, 2, 3, 4, 5} { + operators.EXPECT(). + OperatorsExist(gomock.Any(), []spectypes.OperatorID{id}). + Return(true, nil). + AnyTimes() + } + + verifier := &observingSignatureVerifier{called: make(chan struct{}, 1)} + + validator := New( + netCfg, + validatorStore, + operators, + dutyStore, + verifier, + ).(*messageValidator) + + encodedCommitteeID := append(bytes.Repeat([]byte{0}, 16), committeeID[:]...) + committeeIdentifier := spectypes.NewMsgID(netCfg.DomainType, encodedCommitteeID, spectypes.RoleCommittee) + + return validationLockTestEnv{ + validator: validator, + committeeID: committeeID, + committeeIdentifier: committeeIdentifier, + netCfg: netCfg, + ks: ks, + } +} + +func TestConsensusSignatureVerificationOutsideValidationLock(t *testing.T) { + env := newValidationLockTestEnv(t) + + slot := env.netCfg.FirstSlotAtEpoch(1) + signedSSVMessage := generateSignedMessage(env.ks, env.committeeIdentifier, slot) + topicID := commons.CommitteeTopicID(env.committeeID)[0] + peerID, err := libp2ptest.RandPeerID() + require.NoError(t, err) + + validationMu := env.validator.getValidationLock(signedSSVMessage.SSVMessage.GetID()) + validationMu.Lock() + locked := true + defer func() { + if locked { + validationMu.Unlock() + } + }() + + done := make(chan error, 1) + go func() { + _, err := env.validator.handleSignedSSVMessage(signedSSVMessage, topicID, peerID, env.netCfg.SlotStartTime(slot)) + done <- err + }() + + select { + case <-env.validator.signatureVerifier.(*observingSignatureVerifier).called: + case <-time.After(time.Second): + t.Fatal("signature verification did not start while the validation lock was held") + } + + select { + case err := <-done: + t.Fatalf("validation completed before the lock was released: %v", err) + default: + } + + validationMu.Unlock() + locked = false + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("validation did not complete after the lock was released") + } +} + +func TestPartialSignatureVerificationOutsideValidationLock(t *testing.T) { + env := newValidationLockTestEnv(t) + + slot := env.netCfg.FirstSlotAtEpoch(1) + ssvMessage := spectestingutils.SSVMsgAggregator(nil, spectestingutils.PostConsensusAggregatorMsg(env.ks.Shares[1], 1, spec.DataVersionPhase0)) + ssvMessage.MsgID = env.committeeIdentifier + signedSSVMessage := spectestingutils.SignPartialSigSSVMessage(env.ks, ssvMessage) + topicID := commons.CommitteeTopicID(env.committeeID)[0] + peerID, err := libp2ptest.RandPeerID() + require.NoError(t, err) + + validationMu := env.validator.getValidationLock(signedSSVMessage.SSVMessage.GetID()) + validationMu.Lock() + locked := true + defer func() { + if locked { + validationMu.Unlock() + } + }() + + done := make(chan error, 1) + go func() { + _, err := env.validator.handleSignedSSVMessage(signedSSVMessage, topicID, peerID, env.netCfg.SlotStartTime(slot)) + done <- err + }() + + select { + case <-env.validator.signatureVerifier.(*observingSignatureVerifier).called: + case <-time.After(time.Second): + t.Fatal("partial signature verification did not start while the validation lock was held") + } + + select { + case err := <-done: + t.Fatalf("partial validation completed before the lock was released: %v", err) + default: + } + + validationMu.Unlock() + locked = false + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("partial validation did not complete after the lock was released") + } +} From 653f8eb5e9ec5ecef80d14db5ee02be4d2d03ac1 Mon Sep 17 00:00:00 2001 From: Nikita Kryuchkov Date: Thu, 19 Mar 2026 16:09:25 +0300 Subject: [PATCH 2/2] remove redundant comments --- message/validation/consensus_validation.go | 1 - message/validation/partial_validation.go | 1 - 2 files changed, 2 deletions(-) diff --git a/message/validation/consensus_validation.go b/message/validation/consensus_validation.go index c7ec995b7a..0cd5d42bb0 100644 --- a/message/validation/consensus_validation.go +++ b/message/validation/consensus_validation.go @@ -294,7 +294,6 @@ func (mv *messageValidator) validateQBFTMessageByDutyLogic( msgSlot := phase0.Slot(consensusMessage.Height) randaoMsg := false - // Rule: Message must correspond to a known beacon duty when the role requires it. if err := mv.validateBeaconDuty(signedSSVMessage.SSVMessage.GetID().GetRoleType(), msgSlot, committeeInfo.validatorIndices, randaoMsg); err != nil { return err } diff --git a/message/validation/partial_validation.go b/message/validation/partial_validation.go index 19b3b54e9c..27a71b723c 100644 --- a/message/validation/partial_validation.go +++ b/message/validation/partial_validation.go @@ -169,7 +169,6 @@ func (mv *messageValidator) validatePartialSigMessagesByDutyLogic( } randaoMsg := partialSignatureMessages.Type == spectypes.RandaoPartialSig - // Rule: Message must correspond to a known beacon duty when the role requires it. if err := mv.validateBeaconDuty(signedSSVMessage.SSVMessage.GetID().GetRoleType(), messageSlot, committeeInfo.validatorIndices, randaoMsg); err != nil { return err }