Merge branch 'master' into fix/rest-api-errors

waku/v2/protocol/filter/filter_proto_ident_test.go

@@ -224,8 +224,8 @@ func (s *FilterTestSuite) TestIncorrectSubscribeIdentifier() {
 	// Create test context
 	s.ctx, s.ctxCancel = context.WithTimeout(context.Background(), 10*time.Second) // Test can't exceed 10 seconds
 
-	s.testTopic = "/waku/2/go/filter/test"
-	s.testContentTopic = "TopicA"
+	s.testTopic = defaultTestPubSubTopic
+	s.testContentTopic = defaultTestContentTopic
 
 	s.lightNode = s.makeWakuFilterLightNode(true, true)
 
@@ -263,8 +263,8 @@ func (s *FilterTestSuite) TestIncorrectPushIdentifier() {
 	s.ctx = ctx
 	s.ctxCancel = cancel
 
-	s.testTopic = "/waku/2/go/filter/test"
-	s.testContentTopic = "TopicA"
+	s.testTopic = defaultTestPubSubTopic
+	s.testContentTopic = defaultTestContentTopic
 
 	s.lightNode = s.makeWakuFilterLightNode(false, true)
 

waku/v2/protocol/filter/filter_push_test.go

@@ -3,6 +3,8 @@ package filter
 import (
 	"context"
 	"github.com/waku-org/go-waku/tests"
+	"github.com/waku-org/go-waku/waku/v2/protocol/relay"
+	"github.com/waku-org/go-waku/waku/v2/utils"
 	"go.uber.org/zap"
 	"strconv"
 	"time"
@@ -141,3 +143,23 @@ func (s *FilterTestSuite) TestLargePayloadsUTF8() {
 	s.Require().NoError(err)
 
 }
+
+func (s *FilterTestSuite) TestFuturePayloadEncryptionVersion() {
+
+	// Subscribe
+	s.subDetails = s.subscribe(s.testTopic, s.testContentTopic, s.fullNodeHost.ID())
+
+	message := tests.CreateWakuMessage(s.testContentTopic, utils.GetUnixEpoch(), "test_payload")
+	futureVersion := uint32(100)
+	message.Version = &futureVersion
+
+	// Should get accepted
+	_, err := s.relayNode.Publish(s.ctx, message, relay.WithPubSubTopic(s.testTopic))
+	s.Require().NoError(err)
+
+	// Should be received
+	s.waitForMsg(nil, s.subDetails[0].C)
+
+	_, err = s.lightNode.UnsubscribeAll(s.ctx)
+	s.Require().NoError(err)
+}

waku/v2/protocol/filter/filter_test.go

@@ -29,6 +29,9 @@ func TestFilterSuite(t *testing.T) {
 	suite.Run(t, new(FilterTestSuite))
 }
 
+const defaultTestPubSubTopic = "/waku/2/go/filter/test"
+const defaultTestContentTopic = "/test/10/my-app"
+
 type FilterTestSuite struct {
 	suite.Suite
 
@@ -291,8 +294,8 @@ func (s *FilterTestSuite) publishMessages(msgs []WakuMsg) {
 
 func prepareData(quantity int, topics, contentTopics, payloads bool, sg tests.StringGenerator) []WakuMsg {
 	var (
-		pubsubTopic     = "/waku/2/go/filter/test" // Has to be the same with initial s.testTopic
-		contentTopic    = "TopicA"                 // Has to be the same with initial s.testContentTopic
+		pubsubTopic     = defaultTestPubSubTopic  // Has to be the same with initial s.testTopic
+		contentTopic    = defaultTestContentTopic // Has to be the same with initial s.testContentTopic
 		payload         = "test_msg"
 		messages        []WakuMsg
 		strMaxLenght    = 4097
@@ -341,8 +344,8 @@ func (s *FilterTestSuite) SetupTest() {
 	s.ctx = ctx
 	s.ctxCancel = cancel
 
-	s.testTopic = "/waku/2/go/filter/test"
-	s.testContentTopic = "TopicA"
+	s.testTopic = defaultTestPubSubTopic
+	s.testContentTopic = defaultTestContentTopic
 
 	s.lightNode = s.makeWakuFilterLightNode(true, true)
 

waku/v2/protocol/relay/waku_relay_test.go

@@ -20,8 +20,11 @@ import (
 	"github.com/waku-org/go-waku/waku/v2/utils"
 )
 
+const defaultTestPubSubTopic = "/waku/2/go/relay/test"
+const defaultTestContentTopic = "/test/10/my-app"
+
 func TestWakuRelay(t *testing.T) {
-	testTopic := "/waku/2/go/relay/test"
+	testTopic := defaultTestPubSubTopic
 
 	port, err := tests.FindFreePort(t, "", 5)
 	require.NoError(t, err)
@@ -85,7 +88,7 @@ func createRelayNode(t *testing.T) (host.Host, *WakuRelay) {
 }
 
 func TestGossipsubScore(t *testing.T) {
-	testTopic := "/waku/2/go/relay/test"
+	testTopic := defaultTestPubSubTopic
 
 	hosts := make([]host.Host, 5)
 	relay := make([]*WakuRelay, 5)
@@ -302,3 +305,53 @@ func TestWakuRelayAutoShard(t *testing.T) {
 	require.NoError(t, err)
 
 }
+
+func TestInvalidMessagePublish(t *testing.T) {
+
+	testTopic := defaultTestPubSubTopic
+	testContentTopic := defaultTestContentTopic
+
+	port, err := tests.FindFreePort(t, "", 5)
+	require.NoError(t, err)
+
+	host, err := tests.MakeHost(context.Background(), port, rand.Reader)
+	require.NoError(t, err)
+	bcaster := NewBroadcaster(10)
+	relay := NewWakuRelay(bcaster, 0, timesource.NewDefaultClock(), prometheus.DefaultRegisterer, utils.Logger())
+	relay.SetHost(host)
+	err = relay.Start(context.Background())
+	require.NoError(t, err)
+
+	err = bcaster.Start(context.Background())
+	require.NoError(t, err)
+	defer relay.Stop()
+
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 10*time.Second)
+
+	subs, err := relay.subscribe(context.Background(), protocol.NewContentFilter(testTopic))
+	require.NoError(t, err)
+
+	// Test empty contentTopic
+	_, err = relay.Publish(ctx, tests.CreateWakuMessage("", utils.GetUnixEpoch(), "test_payload"), WithPubSubTopic(testTopic))
+	require.Error(t, err)
+
+	// Test empty payload
+	_, err = relay.Publish(ctx, tests.CreateWakuMessage(testContentTopic, utils.GetUnixEpoch(), ""), WithPubSubTopic(testTopic))
+	require.Error(t, err)
+
+	// Test empty contentTopic and payload
+	_, err = relay.Publish(ctx, tests.CreateWakuMessage("", utils.GetUnixEpoch(), ""), WithPubSubTopic(testTopic))
+	require.Error(t, err)
+
+	// Test Meta size over limit
+	message := tests.CreateWakuMessage(testContentTopic, utils.GetUnixEpoch(), "test_payload")
+	message.Meta = make([]byte, 65)
+	_, err = relay.Publish(ctx, message, WithPubSubTopic(testTopic))
+	require.Error(t, err)
+
+	err = relay.Unsubscribe(ctx, protocol.NewContentFilter(subs[0].contentFilter.PubsubTopic))
+	require.NoError(t, err)
+
+	ctxCancel()
+
+}