diff --git a/network/alsp/internal/cache_test.go b/network/alsp/internal/cache_test.go index 8a7e116e3bb..b1f4d2d179f 100644 --- a/network/alsp/internal/cache_test.go +++ b/network/alsp/internal/cache_test.go @@ -813,3 +813,56 @@ func TestSpamRecordCache_ConcurrentIdentitiesAndOperations(t *testing.T) { unittest.RequireReturnsBefore(t, wg.Wait, 1*time.Second, "timed out waiting for goroutines to finish") } + +// TestSpamRecordCache_ConcurrentGetWhileAdjustingSameRecord is a regression test for two data +// races on the same record (run with -race): +// 1. Get copied the record's fields after the backend lock was released, racing the in-place +// mutation performed by a concurrent adjustment of the same record. +// 2. AdjustWithInit read the adjusted record's penalty after the lock was released, so the +// returned penalty could reflect a different (later) adjustment. +func TestSpamRecordCache_ConcurrentGetWhileAdjustingSameRecord(t *testing.T) { + cache := internal.NewSpamRecordCache(100, zerolog.Nop(), metrics.NewNoopCollector(), model.SpamRecordFactory()) + originID := unittest.IdentifierFixture() + + const adjustments = 500 + const penaltyDelta = -1.0 + + var wg sync.WaitGroup + wg.Add(2) + + // writer: repeatedly adjusts the same record, and verifies the returned penalty is the one + // produced by its own adjustment (captured inside the adjust function, under the lock). + go func() { + defer wg.Done() + for i := 0; i < adjustments; i++ { + var want float64 + penalty, err := cache.AdjustWithInit(originID, func(record *model.ProtocolSpamRecord) (*model.ProtocolSpamRecord, error) { + record.Penalty += penaltyDelta + want = record.Penalty + return record, nil + }) + require.NoError(t, err) + require.Equal(t, want, penalty, "returned penalty must match the penalty produced by this adjustment") + } + }() + + // reader: repeatedly reads the same record while it is being adjusted. + go func() { + defer wg.Done() + for i := 0; i < adjustments; i++ { + record, ok := cache.Get(originID) + if !ok { + continue // record not initialized yet + } + // the penalty only ever decreases in steps of penaltyDelta + require.LessOrEqual(t, record.Penalty, float64(0)) + require.GreaterOrEqual(t, record.Penalty, float64(adjustments)*penaltyDelta) + } + }() + + unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent adjustments and reads did not finish on time") + + record, ok := cache.Get(originID) + require.True(t, ok) + require.Equal(t, float64(adjustments)*penaltyDelta, record.Penalty) +} diff --git a/network/p2p/inspector/validation/control_message_validation_inspector_test.go b/network/p2p/inspector/validation/control_message_validation_inspector_test.go index f19ceeb3a2a..6fdcfc608ce 100644 --- a/network/p2p/inspector/validation/control_message_validation_inspector_test.go +++ b/network/p2p/inspector/validation/control_message_validation_inspector_test.go @@ -299,6 +299,48 @@ func TestControlMessageValidationInspector_truncateRPC(t *testing.T) { }) } +// TestControlMessageInspection_DoesNotMutateRpc is a regression test verifying that the +// asynchronous inspection does not mutate the inspected RPC: the RPC object remains in use by +// the libp2p pubsub layer while the inspector's worker pool processes it, so any in-place +// modification (e.g. shuffling the publish message list for sampling) is a data race. +func TestControlMessageInspection_DoesNotMutateRpc(t *testing.T) { + logCounter := atomic.NewInt64(0) + logger := hookedLogger(logCounter, zerolog.TraceLevel, worker.QueuedItemProcessedLog) + inspector, signalerCtx, cancel, consumer, rpcTracker, sporkID, idProvider, topicProviderOracle := inspectorFixture(t, func(params *validation.InspectorParams) { + params.Logger = logger + }) + defer consumer.AssertNotCalled(t, "OnInvalidControlMessageNotification") + + topic := fmt.Sprintf("%s/%s", channels.TestNetworkChannel, sporkID) + topicProviderOracle.UpdateTopics([]string{topic}) + inspector.Start(signalerCtx) + unittest.RequireComponentsReadyBefore(t, 1*time.Second, inspector) + + from := unittest.PeerIdFixture(t) + // enough messages that an accidental in-place shuffle is detected with near certainty + pubsubMsgs := unittest.GossipSubMessageFixtures(20, topic, unittest.WithFrom(from)) + rpc := unittest.P2PRPCFixture(unittest.WithPubsubMessages(pubsubMsgs...)) + rpcTracker.On("LastHighestIHaveRPCSize").Return(int64(100)).Maybe() + idProvider.On("ByPeerID", from).Return(unittest.IdentityFixture(), true) + + // snapshot the publish message list before inspection + originalPublish := append([]*pubsub_pb.Message{}, rpc.GetPublish()...) + + require.NoError(t, inspector.Inspect(from, rpc)) + require.Eventually(t, func() bool { + return logCounter.Load() == 1 + }, 30*time.Second, 100*time.Millisecond) + + // the RPC's publish message list must be exactly as it was before the inspection + require.Equal(t, len(originalPublish), len(rpc.GetPublish())) + for i, msg := range rpc.GetPublish() { + require.Same(t, originalPublish[i], msg, "publish message %d was moved or replaced by the inspection", i) + } + + cancel() + unittest.RequireCloseBefore(t, inspector.Done(), 5*time.Second, "inspector did not stop") +} + // TestControlMessageInspection_ValidRpc ensures inspector does not disseminate invalid control message notifications for a valid RPC. func TestControlMessageInspection_ValidRpc(t *testing.T) { logCounter := atomic.NewInt64(0) diff --git a/network/p2p/scoring/internal/appSpecificScoreCache_test.go b/network/p2p/scoring/internal/appSpecificScoreCache_test.go index bea5f355833..f9bdef3c3af 100644 --- a/network/p2p/scoring/internal/appSpecificScoreCache_test.go +++ b/network/p2p/scoring/internal/appSpecificScoreCache_test.go @@ -164,3 +164,45 @@ func TestAppSpecificScoreCache_Eviction(t *testing.T) { _, _, found := cache.Get(peerIds[0]) require.False(t, found, "score should not be in cache") } + +// TestAppSpecificScoreCache_ConcurrentGetWhileUpdatingSamePeer is a regression test for a data +// race on a single peer's record (run with -race): Get read the record's fields after the +// backend lock was released, racing the in-place mutation performed by a concurrent update of +// the same record. The cache now replaces records copy-on-write, keeping returned records +// immutable. +func TestAppSpecificScoreCache_ConcurrentGetWhileUpdatingSamePeer(t *testing.T) { + cache := internal.NewAppSpecificScoreCache(10, unittest.Logger(), metrics.NewNoopCollector()) + + peerID := unittest.PeerIdFixture(t) + const updates = 500 + + var wg sync.WaitGroup + wg.Add(2) + + // writer: repeatedly updates the same peer's score. + go func() { + defer wg.Done() + for i := 0; i < updates; i++ { + require.NoError(t, cache.AdjustWithInit(peerID, float64(i), time.Now())) + } + }() + + // reader: repeatedly reads the same peer's score while it is being updated. + go func() { + defer wg.Done() + for i := 0; i < updates; i++ { + score, _, found := cache.Get(peerID) + if !found { + continue // not initialized yet + } + require.GreaterOrEqual(t, score, float64(0)) + require.Less(t, score, float64(updates)) + } + }() + + unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent updates and reads did not finish on time") + + score, _, found := cache.Get(peerID) + require.True(t, found) + require.Equal(t, float64(updates-1), score) +} diff --git a/network/p2p/scoring/registry_internal_test.go b/network/p2p/scoring/registry_internal_test.go new file mode 100644 index 00000000000..cc19b5958fb --- /dev/null +++ b/network/p2p/scoring/registry_internal_test.go @@ -0,0 +1,39 @@ +package scoring + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/atomic" +) + +// TestAfterSilencePeriod verifies the state machine of the scoring registry's startup silence +// period, in particular that the silence period does not end before the registry has started. +// Regression: the start time used to be a plain time.Time (a data race with the startup worker), +// and its zero value made time.Since(zero) exceed any configured duration, spuriously ending the +// silence period if the score function was queried before startup. +func TestAfterSilencePeriod(t *testing.T) { + reg := &GossipSubAppSpecificScoreRegistry{ + silencePeriodDuration: time.Hour, + silencePeriodStartTime: atomic.NewPointer[time.Time](nil), + silencePeriodElapsed: atomic.NewBool(false), + } + + // before startup (start time not set), the silence period has not even begun + require.False(t, reg.afterSilencePeriod()) + + // silence period started, but not yet over + now := time.Now() + reg.silencePeriodStartTime.Store(&now) + require.False(t, reg.afterSilencePeriod()) + + // silence period over + past := time.Now().Add(-2 * time.Hour) + reg.silencePeriodStartTime.Store(&past) + require.True(t, reg.afterSilencePeriod()) + + // once elapsed, the silence period stays elapsed + require.True(t, reg.silencePeriodElapsed.Load()) + require.True(t, reg.afterSilencePeriod()) +} diff --git a/network/slashing/consumer_test.go b/network/slashing/consumer_test.go new file mode 100644 index 00000000000..4ffdcc7c57b --- /dev/null +++ b/network/slashing/consumer_test.go @@ -0,0 +1,96 @@ +package slashing_test + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/onflow/flow-go/model/flow" + mockmodule "github.com/onflow/flow-go/module/mock" + "github.com/onflow/flow-go/network" + "github.com/onflow/flow-go/network/channels" + "github.com/onflow/flow-go/network/message" + mocknetwork "github.com/onflow/flow-go/network/mock" + "github.com/onflow/flow-go/network/slashing" + "github.com/onflow/flow-go/utils/unittest" +) + +// newConsumerFixture returns a slashing violations consumer whose metrics and misbehavior +// report consumer tolerate any number of calls. +func newConsumerFixture(t *testing.T) *slashing.Consumer { + metrics := mockmodule.NewNetworkSecurityMetrics(t) + metrics.On("OnUnauthorizedMessage", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return().Maybe() + metrics.On("OnViolationReportSkipped").Return().Maybe() + + misbehaviorReportConsumer := mocknetwork.NewMisbehaviorReportConsumer(t) + misbehaviorReportConsumer.On("ReportMisbehaviorOnChannel", mock.Anything, mock.Anything).Return().Maybe() + + return slashing.NewSlashingViolationsConsumer(unittest.Logger(), metrics, misbehaviorReportConsumer) +} + +// violationFixture returns a violation without an identity (a violation from an unknown peer) +// and without a message type (a violation raised before the message could be decoded). +func violationFixture() *network.Violation { + return &network.Violation{ + Identity: nil, + PeerID: "peer-id", + OriginID: flow.ZeroID, + MsgType: "", // simulates a violation raised before the message type is known + Channel: channels.TestNetworkChannel, + Protocol: message.ProtocolTypeUnicast, + Err: errors.New("unauthorized"), + } +} + +// TestConsumer_DoesNotMutateViolation is a regression test verifying that the consumer never +// mutates the violation passed to it: callers may share one violation object across goroutines +// (or reuse it for multiple notifications), so an in-place default (e.g. setting MsgType to +// "unknown") would be a data race and would leak into subsequent uses. +func TestConsumer_DoesNotMutateViolation(t *testing.T) { + consumer := newConsumerFixture(t) + + violation := violationFixture() + consumer.OnUnauthorizedSenderError(violation) + + require.Empty(t, violation.MsgType, "consumer must not mutate the violation's MsgType") + require.Nil(t, violation.Identity, "consumer must not mutate the violation's Identity") +} + +// TestConsumer_ConcurrentNotifications is a regression test verifying that a single violation +// object can be reported concurrently through all consumer entry points without a data race +// (run with -race). Before the fix, logOffense wrote a default MsgType into the shared +// violation, racing the reads of concurrent notifications. +func TestConsumer_ConcurrentNotifications(t *testing.T) { + consumer := newConsumerFixture(t) + + violation := violationFixture() + + notify := []func(*network.Violation){ + consumer.OnUnauthorizedSenderError, + consumer.OnUnknownMsgTypeError, + consumer.OnInvalidMsgError, + consumer.OnSenderEjectedError, + consumer.OnUnauthorizedUnicastOnChannel, + consumer.OnUnauthorizedPublishOnChannel, + } + + workers := 4 + iterations := 50 + var wg sync.WaitGroup + wg.Add(workers) + for range workers { + go func() { + defer wg.Done() + for i := 0; i < iterations; i++ { + notify[i%len(notify)](violation) + } + }() + } + unittest.RequireReturnsBefore(t, wg.Wait, 10*time.Second, "concurrent notifications did not finish on time") + + require.Empty(t, violation.MsgType, "consumer must not mutate the violation's MsgType") +}