Dynamic welcome message (#439)

* Add dynamic welcome message to p2p handshake

* Add '/welcome-msg' to debug api

* Add missing license headers

* Rename to WelcomeMessageSynced to Get

* Add atomic.value instead of struct

* Simplify mocks, refactor tests

* Add request struct and length check

* Extract DebugService p2p interface

* Add check

* Remove MaxBytesReader

* Remove unused constant

* Remove out of scope test

* Set max request size

* Refactor test

pkg/debugapi/debugapi.go

@@ -32,7 +32,7 @@ type server struct {
 
 type Options struct {
 	Overlay        swarm.Address
-	P2P            p2p.Service
+	P2P            p2p.DebugService
 	Pingpong       pingpong.Interface
 	TopologyDriver topology.PeerAdder
 	Storer         storage.Storer

pkg/debugapi/export_test.go

@@ -13,4 +13,6 @@ type (
 	PinnedChunk              = pinnedChunk
 	ListPinnedChunksResponse = listPinnedChunksResponse
 	TagResponse              = tagResponse
+	WelcomeMessageRequest    = welcomeMessageRequest
+	WelcomeMessageResponse   = welcomeMessageResponse
 )

pkg/debugapi/router.go

@@ -86,6 +86,13 @@ func (s *server) setupRouting() {
 	router.Handle("/topology", jsonhttp.MethodHandler{
 		"GET": http.HandlerFunc(s.topologyHandler),
 	})
+	router.Handle("/welcome-message", jsonhttp.MethodHandler{
+		"GET": http.HandlerFunc(s.getWelcomeMessageHandler),
+		"POST": web.ChainHandlers(
+			jsonhttp.NewMaxBodyBytesHandler(welcomeMessageMaxRequestSize),
+			web.FinalHandlerFunc(s.setWelcomeMessageHandler),
+		),
+	})
 
 	baseRouter.Handle("/", web.ChainHandlers(
 		logging.NewHTTPAccessLogHandler(s.Logger, logrus.InfoLevel, "debug api access"),

pkg/debugapi/welcome_message.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package debugapi
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/ethersphere/bee/pkg/jsonhttp"
+)
+
+const welcomeMessageMaxRequestSize = 512
+
+type welcomeMessageRequest struct {
+	WelcomeMesssage string `json:"welcome_message"`
+}
+
+type welcomeMessageResponse struct {
+	WelcomeMesssage string `json:"welcome_message"`
+}
+
+func (s *server) getWelcomeMessageHandler(w http.ResponseWriter, r *http.Request) {
+	val := s.P2P.GetWelcomeMessage()
+	jsonhttp.OK(w, welcomeMessageResponse{
+		WelcomeMesssage: val,
+	})
+}
+
+func (s *server) setWelcomeMessageHandler(w http.ResponseWriter, r *http.Request) {
+	var data welcomeMessageRequest
+	err := json.NewDecoder(r.Body).Decode(&data)
+	if err != nil {
+		s.Logger.Debugf("debugapi: welcome message: failed to read request: %v", err)
+		jsonhttp.BadRequest(w, err)
+		return
+	}
+
+	if err := s.P2P.SetWelcomeMessage(data.WelcomeMesssage); err != nil {
+		s.Logger.Debugf("debugapi: welcome message: failed to set: %v", err)
+		s.Logger.Errorf("Failed to set welcome message")
+		jsonhttp.InternalServerError(w, err)
+		return
+	}
+	jsonhttp.OK(w, nil)
+}

pkg/debugapi/welcome_message_test.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package debugapi_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"testing"
+
+	"github.com/ethersphere/bee/pkg/debugapi"
+	"github.com/ethersphere/bee/pkg/jsonhttp"
+	"github.com/ethersphere/bee/pkg/jsonhttp/jsonhttptest"
+	"github.com/ethersphere/bee/pkg/p2p/mock"
+)
+
+func TestGetWelcomeMessage(t *testing.T) {
+	const DefaultTestWelcomeMessage = "Hello World!"
+
+	srv := newTestServer(t, testServerOptions{
+		P2P: mock.New(mock.WithGetWelcomeMessageFunc(func() string {
+			return DefaultTestWelcomeMessage
+		}))})
+
+	jsonhttptest.ResponseDirect(t, srv.Client, http.MethodGet, "/welcome-message", nil, http.StatusOK, debugapi.WelcomeMessageResponse{
+		WelcomeMesssage: DefaultTestWelcomeMessage,
+	})
+}
+
+func TestSetWelcomeMessage(t *testing.T) {
+	testCases := []struct {
+		desc        string
+		message     string
+		wantFail    bool
+		wantStatus  int
+		wantMessage string
+	}{
+		{
+			desc:       "OK",
+			message:    "Changed value",
+			wantStatus: http.StatusOK,
+		},
+		{
+			desc:       "OK - welcome message empty",
+			message:    "",
+			wantStatus: http.StatusOK,
+		},
+		{
+			desc:     "fails - request entity too large",
+			wantFail: true,
+			message: `zZZbzbzbzbBzBBZbBbZbbbBzzzZBZBbzzBBBbBzBzzZbbBzBBzBBbZz
+			bZZZBBbbZbbZzBbzBbzbZBZzBZZbZzZzZzbbZZBZzzbBZBzZzzBBzZZzzZbZZZzbbbzz
+			bBzZZBbBZBzZzBZBzbzBBbzBBzbzzzBbBbZzZBZBZzBZZbbZZBZZBzZzBZbzZBzZbBzZ
+			bbbBbbZzZbzbZzZzbzzzbzzbzZZzbbzbBZZbBbBZBBZzZzzbBBBBBZbZzBzzBbzBbbbz
+			BBzbbZBbzbzBZbzzBzbZBzzbzbbbBZBzBZzBZbzBzZzBZZZBzZZBzBZZzbzZbzzZzBBz
+			ZZzbZzzZZZBZBBbZZbZzBBBzbzZZbbZZBZZBBBbBZzZbZBZBBBzzZBbbbbzBzbbzBBBz
+			bZBBbZzBbZZBzbBbZZBzBzBzBBbzzzZBbzbZBbzBbZzbbBZBBbbZbBBbbBZbzbZzbBzB
+			bBbbZZbzZzbbBbzZbZZZZbzzZZbBzZZbZzZzzBzbZZ`, // 513 characters
+			wantStatus: http.StatusRequestEntityTooLarge,
+		},
+	}
+	testURL := "/welcome-message"
+
+	srv := newTestServer(t, testServerOptions{
+		P2P: mock.New(),
+	})
+
+	for _, tC := range testCases {
+		t.Run(tC.desc, func(t *testing.T) {
+			if tC.wantMessage == "" {
+				tC.wantMessage = http.StatusText(tC.wantStatus)
+			}
+			data, _ := json.Marshal(debugapi.WelcomeMessageRequest{
+				WelcomeMesssage: tC.message,
+			})
+			body := bytes.NewReader(data)
+			wantResponse := jsonhttp.StatusResponse{
+				Message: tC.wantMessage,
+				Code:    tC.wantStatus,
+			}
+			jsonhttptest.ResponseDirect(t, srv.Client, http.MethodPost, testURL, body, tC.wantStatus, wantResponse)
+			if !tC.wantFail {
+				got := srv.P2PMock.GetWelcomeMessage()
+				if got != tC.message {
+					t.Fatalf("could not set dynamic welcome message: want %s, got %s", tC.message, got)
+				}
+			}
+		})
+	}
+}
+
+func TestSetWelcomeMessageInternalServerError(t *testing.T) {
+	testMessage := "NO CHANCE BYE"
+	testError := errors.New("Could not set value")
+	testURL := "/welcome-message"
+
+	srv := newTestServer(t, testServerOptions{
+		P2P: mock.New(mock.WithSetWelcomeMessageFunc(func(string) error {
+			return testError
+		})),
+	})
+
+	data, _ := json.Marshal(debugapi.WelcomeMessageRequest{
+		WelcomeMesssage: testMessage,
+	})
+	body := bytes.NewReader(data)
+	t.Run("internal server error - failed to store", func(t *testing.T) {
+		wantCode := http.StatusInternalServerError
+		wantResp := jsonhttp.StatusResponse{
+			Message: testError.Error(),
+			Code:    wantCode,
+		}
+		jsonhttptest.ResponseDirect(t, srv.Client, http.MethodPost, testURL, body, wantCode, wantResp)
+	})
+
+}

pkg/p2p/libp2p/internal/handshake/handshake.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ethersphere/bee/pkg/bzz"
@@ -64,7 +65,7 @@ type Service struct {
 	overlay               swarm.Address
 	lightNode             bool
 	networkID             uint64
-	welcomeMessage        string
+	welcomeMessage        atomic.Value
 	receivedHandshakes    map[libp2ppeer.ID]struct{}
 	receivedHandshakesMu  sync.Mutex
 	logger                logging.Logger
@@ -84,17 +85,19 @@ func New(signer crypto.Signer, advertisableAddresser AdvertisableAddressResolver
 		return nil, ErrWelcomeMessageLength
 	}
 
-	return &Service{
+	svc := &Service{
 		signer:                signer,
 		advertisableAddresser: advertisableAddresser,
 		overlay:               overlay,
 		networkID:             networkID,
 		lightNode:             lighNode,
-		welcomeMessage:        welcomeMessage,
 		receivedHandshakes:    make(map[libp2ppeer.ID]struct{}),
 		logger:                logger,
 		Notifiee:              new(network.NoopNotifiee),
-	}, nil
+	}
+	svc.welcomeMessage.Store(welcomeMessage)
+
+	return svc, nil
 }
 
 // Handshake initiates a handshake with a peer.
@@ -146,6 +149,8 @@ func (s *Service) Handshake(stream p2p.Stream, peerMultiaddr ma.Multiaddr, peerI
 		return nil, err
 	}
 
+	// Synced read:
+	welcomeMessage := s.GetWelcomeMessage()
 	if err := w.WriteMsgWithTimeout(messageTimeout, &pb.Ack{
 		Address: &pb.BzzAddress{
 			Underlay:  advertisableUnderlayBytes,
@@ -154,7 +159,7 @@ func (s *Service) Handshake(stream p2p.Stream, peerMultiaddr ma.Multiaddr, peerI
 		},
 		NetworkID:      s.networkID,
 		Light:          s.lightNode,
-		WelcomeMessage: s.welcomeMessage,
+		WelcomeMessage: welcomeMessage,
 	}); err != nil {
 		return nil, fmt.Errorf("write ack message: %w", err)
 	}
@@ -216,6 +221,8 @@ func (s *Service) Handle(stream p2p.Stream, remoteMultiaddr ma.Multiaddr, remote
 		return nil, err
 	}
 
+	welcomeMessage := s.GetWelcomeMessage()
+
 	if err := w.WriteMsgWithTimeout(messageTimeout, &pb.SynAck{
 		Syn: &pb.Syn{
 			ObservedUnderlay: fullRemoteMABytes,
@@ -228,7 +235,7 @@ func (s *Service) Handle(stream p2p.Stream, remoteMultiaddr ma.Multiaddr, remote
 			},
 			NetworkID:      s.networkID,
 			Light:          s.lightNode,
-			WelcomeMessage: s.welcomeMessage,
+			WelcomeMessage: welcomeMessage,
 		},
 	}); err != nil {
 		return nil, fmt.Errorf("write synack message: %w", err)
@@ -259,6 +266,20 @@ func (s *Service) Disconnected(_ network.Network, c network.Conn) {
 	delete(s.receivedHandshakes, c.RemotePeer())
 }
 
+// SetWelcomeMessage sets the new handshake welcome message.
+func (s *Service) SetWelcomeMessage(msg string) (err error) {
+	if len(msg) > MaxWelcomeMessageLength {
+		return ErrWelcomeMessageLength
+	}
+	s.welcomeMessage.Store(msg)
+	return nil
+}
+
+// GetWelcomeMessage returns the the current handshake welcome message.
+func (s *Service) GetWelcomeMessage() string {
+	return s.welcomeMessage.Load().(string)
+}
+
 func buildFullMA(addr ma.Multiaddr, peerID libp2ppeer.ID) (ma.Multiaddr, error) {
 	return ma.NewMultiaddr(fmt.Sprintf("%s/p2p/%s", addr.String(), peerID.Pretty()))
 }

pkg/p2p/libp2p/internal/handshake/handshake_test.go

@@ -165,6 +165,29 @@ func TestHandshake(t *testing.T) {
 		}
 	})
 
+	t.Run("Handshake - dynamic welcome message too long", func(t *testing.T) {
+		const LongMessage = "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Morbi consectetur urna ut lorem sollicitudin posuere. Donec sagittis laoreet sapien."
+
+		expectedErr := handshake.ErrWelcomeMessageLength
+		err := handshakeService.SetWelcomeMessage(LongMessage)
+		if err == nil || err.Error() != expectedErr.Error() {
+			t.Fatal("expected:", expectedErr, "got:", err)
+		}
+	})
+
+	t.Run("Handshake - set welcome message", func(t *testing.T) {
+		const TestMessage = "Hi im the new test message"
+
+		err := handshakeService.SetWelcomeMessage(TestMessage)
+		if err != nil {
+			t.Fatal("Got error:", err)
+		}
+		got := handshakeService.GetWelcomeMessage()
+		if got != TestMessage {
+			t.Fatal("expected:", TestMessage, ", got:", got)
+		}
+	})
+
 	t.Run("Handshake - Syn write error", func(t *testing.T) {
 		testErr := errors.New("test error")
 		expectedErr := fmt.Errorf("write syn message: %w", testErr)

pkg/p2p/libp2p/libp2p.go

@@ -40,6 +40,7 @@ import (
 
 var (
 	_ p2p.Service      = (*Service)(nil)
+	_ p2p.DebugService = (*Service)(nil)
 )
 
 type Service struct {
@@ -503,3 +504,13 @@ func (s *Service) Close() error {
 	}
 	return s.host.Close()
 }
+
+// SetWelcomeMessage sets the welcome message for the handshake protocol.
+func (s *Service) SetWelcomeMessage(val string) error {
+	return s.handshakeService.SetWelcomeMessage(val)
+}
+
+// GetWelcomeMessage returns the value of the welcome message.
+func (s *Service) GetWelcomeMessage() string {
+	return s.handshakeService.GetWelcomeMessage()
+}

pkg/p2p/libp2p/welcome_message_test.go

+// Copyright 2020 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+package libp2p_test
+
+import (
+	"testing"
+
+	"github.com/ethersphere/bee/pkg/p2p/libp2p"
+	"github.com/ethersphere/bee/pkg/p2p/libp2p/internal/handshake"
+)
+
+func TestDynamicWelcomeMessage(t *testing.T) {
+	const TestWelcomeMessage = "Hello World!"
+	svc, _ := newService(t, 1, libp2p.Options{WelcomeMessage: TestWelcomeMessage})
+
+	t.Run("Get current message - OK", func(t *testing.T) {
+		got := svc.GetWelcomeMessage()
+		if got != TestWelcomeMessage {
+			t.Fatalf("expected %s, got %s", TestWelcomeMessage, got)
+		}
+	})
+
+	t.Run("Set new message", func(t *testing.T) {
+		t.Run("OK", func(t *testing.T) {
+			const testMessage = "I'm the new message!"
+
+			err := svc.SetWelcomeMessage(testMessage)
+			if err != nil {
+				t.Fatal("got error:", err)
+			}
+			got := svc.GetWelcomeMessage()
+			if got != testMessage {
+				t.Fatalf("expected: %s. got %s", testMessage, got)
+			}
+		})
+		t.Run("fails - message too long", func(t *testing.T) {
+			const testMessage = `Lorem ipsum dolor sit amet, consectetur adipiscing elit.
+			Maecenas eu aliquam enim. Nulla tincidunt arcu nec nulla condimentum nullam sodales` // 141 characters
+
+			want := handshake.ErrWelcomeMessageLength
+			got := svc.SetWelcomeMessage(testMessage)
+			if got != want {
+				t.Fatalf("wrong error: want %v, got %v", want, got)
+			}
+		})
+
+	})
+}

pkg/p2p/mock/mock.go

@@ -16,6 +16,7 @@ import (
 	ma "github.com/multiformats/go-multiaddr"
 )
 
+// Service is the mock of a P2P Service
 type Service struct {
 	addProtocolFunc       func(p2p.ProtocolSpec) error
 	connectFunc           func(ctx context.Context, addr ma.Multiaddr) (address *bzz.Address, err error)
@@ -23,45 +24,69 @@ type Service struct {
 	peersFunc             func() []p2p.Peer
 	setNotifierFunc       func(topology.Notifier)
 	addressesFunc         func() ([]ma.Multiaddr, error)
+	setWelcomeMessageFunc func(string) error
+	getWelcomeMessageFunc func() string
+	welcomeMessage        string
 	notifyCalled          int32
 }
 
+// WithAddProtocolFunc sets the mock implementation of the AddProtocol function
 func WithAddProtocolFunc(f func(p2p.ProtocolSpec) error) Option {
 	return optionFunc(func(s *Service) {
 		s.addProtocolFunc = f
 	})
 }
 
+// WithConnectFunc sets the mock implementation of the Connect function
 func WithConnectFunc(f func(ctx context.Context, addr ma.Multiaddr) (address *bzz.Address, err error)) Option {
 	return optionFunc(func(s *Service) {
 		s.connectFunc = f
 	})
 }
 
+// WithDisconnectFunc sets the mock implementation of the Disconnect function
 func WithDisconnectFunc(f func(overlay swarm.Address) error) Option {
 	return optionFunc(func(s *Service) {
 		s.disconnectFunc = f
 	})
 }
 
+// WithPeersFunc sets the mock implementation of the Peers function
 func WithPeersFunc(f func() []p2p.Peer) Option {
 	return optionFunc(func(s *Service) {
 		s.peersFunc = f
 	})
 }
 
+// WithSetNotifierFunc sets the mock implementation of the SetNotifier function
 func WithSetNotifierFunc(f func(topology.Notifier)) Option {
 	return optionFunc(func(s *Service) {
 		s.setNotifierFunc = f
 	})
 }
 
+// WithAddressesFunc sets the mock implementation of the Adresses function
 func WithAddressesFunc(f func() ([]ma.Multiaddr, error)) Option {
 	return optionFunc(func(s *Service) {
 		s.addressesFunc = f
 	})
 }
 
+// WithGetWelcomeMessageFunc sets the mock implementation of the GetWelcomeMessage function
+func WithGetWelcomeMessageFunc(f func() string) Option {
+	return optionFunc(func(s *Service) {
+		s.getWelcomeMessageFunc = f
+	})
+}
+
+// WithSetWelcomeMessageFunc sets the mock implementation of the SetWelcomeMessage function
+func WithSetWelcomeMessageFunc(f func(string) error) Option {
+	return optionFunc(func(s *Service) {
+		s.setWelcomeMessageFunc = f
+	})
+}
+
+// New will create a new mock P2P Service with the given options
 func New(opts ...Option) *Service {
 	s := new(Service)
 	for _, o := range opts {
@@ -126,6 +151,21 @@ func (s *Service) ConnectNotifyCalls() int32 {
 	return c
 }
 
+func (s *Service) SetWelcomeMessage(val string) error {
+	if s.setWelcomeMessageFunc != nil {
+		return s.setWelcomeMessageFunc(val)
+	}
+	s.welcomeMessage = val
+	return nil
+}
+
+func (s *Service) GetWelcomeMessage() string {
+	if s.getWelcomeMessageFunc != nil {
+		return s.getWelcomeMessageFunc()
+	}
+	return s.welcomeMessage
+}
+
 type Option interface {
 	apply(*Service)
 }

pkg/p2p/p2p.go

@@ -28,6 +28,13 @@ type Service interface {
 	Addresses() ([]ma.Multiaddr, error)
 }
 
+// DebugService extends the Service with method used for debugging.
+type DebugService interface {
+	Service
+	SetWelcomeMessage(val string) error
+	GetWelcomeMessage() string
+}
+
 // Streamer is able to create a new Stream.
 type Streamer interface {
 	NewStream(ctx context.Context, address swarm.Address, h Headers, protocol, version, stream string) (Stream, error)