Implement mailserver canary service. Closes #1086

.gitignore

@@ -66,3 +66,4 @@ Session.vim
 .undodir/*
 /.idea/
 /.vscode/
+/cmd/*/.ethereum/

Makefile

@@ -87,6 +87,10 @@ bootnode: ##@build Build discovery v5 bootnode using status-go deps
 	go build -i -o $(GOBIN)/bootnode -v -tags '$(BUILD_TAGS)' $(BUILD_FLAGS) ./cmd/bootnode/
 	@echo "Compilation done."
 
+mailserver-canary: ##@build Build mailserver canary using status-go deps
+	go build -i -o $(GOBIN)/mailserver-canary -v -tags '$(BUILD_TAGS)' $(BUILD_FLAGS) ./cmd/mailserver-canary/
+	@echo "Compilation done."
+
 statusgo-cross: statusgo-android statusgo-ios
 	@echo "Full cross compilation done."
 	@ls -ld $(GOBIN)/statusgo-*

cmd/mailserver-canary/README.md

@@ -0,0 +1,16 @@
+Canary service
+======================
+
+The mailserver canary service's goal is to provide feedback on whether a specified mailserver is responding
+correctly to historic messages request. It sends a request for 1 message in a specified chat room (defaults
+to #status) to the mailserver within a specified time window (default is last 24 hours) and succeeds if the
+mailserver responds with an acknowledgement to the request message (using the request's hash value as a
+match).
+
+## How to run it
+
+```shell
+make mailserver-canary
+
+./build/bin/mailserver-canary -log=INFO --mailserver=enode://69f72baa7f1722d111a8c9c68c39a31430e9d567695f6108f31ccb6cd8f0adff4991e7fdca8fa770e75bc8a511a87d24690cbc80e008175f40c157d6f6788d48@206.189.240.16:30504
+```

cmd/mailserver-canary/main.go

@@ -0,0 +1,314 @@
+// mailserver-canary tests whether a mailserver enode responds to a historic messages request.
+package main
+
+import (
+	"context"
+	"errors"
+	"flag"
+	"fmt"
+	stdlog "log"
+	"os"
+	"path"
+	"path/filepath"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto/sha3"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/p2p/discv5"
+	whisper "github.com/ethereum/go-ethereum/whisper/whisperv6"
+	"github.com/status-im/status-go/api"
+	"github.com/status-im/status-go/logutils"
+	"github.com/status-im/status-go/params"
+	"github.com/status-im/status-go/rpc"
+	"github.com/status-im/status-go/services/shhext"
+	"github.com/status-im/status-go/t/helpers"
+	"golang.org/x/crypto/ssh/terminal"
+)
+
+const (
+	mailboxPassword = "status-offline-inbox"
+)
+
+// All general log messages in this package should be routed through this logger.
+var logger = log.New("package", "status-go/cmd/mailserver-canary")
+
+var (
+	enodeAddr        = flag.String("mailserver", "", "mailserver enode address (e.g. enode://1da276e34126e93babf24ec88aac1a7602b4cbb2e11b0961d0ab5e989ca9c261aa7f7c1c85f15550a5f1e5a5ca2305b53b9280cf5894d5ecf7d257b173136d40@167.99.209.61:30504)")
+	publicChannel    = flag.String("channel", "status", "The public channel name to retrieve historic messages from")
+	period           = flag.Int("period", 24*60*60, "How far in the past to request messages from mailserver, in seconds")
+	minPow           = flag.Float64("shh.pow", params.WhisperMinimumPoW, "PoW for messages to be added to queue, in float format")
+	ttl              = flag.Int("shh.ttl", params.WhisperTTL, "Time to live for messages, in seconds")
+	homePath         = flag.String("home-dir", ".", "Home directory where state is stored")
+	logLevel         = flag.String("log", "INFO", `Log level, one of: "ERROR", "WARN", "INFO", "DEBUG", and "TRACE"`)
+	logFile          = flag.String("logfile", "", "Path to the log file")
+	logWithoutColors = flag.Bool("log-without-color", false, "Disables log colors")
+)
+
+func main() {
+	if enodeAddr == nil {
+		logger.Crit("No mailserver address specified", "enodeAddr", *enodeAddr)
+		os.Exit(1)
+	}
+
+	mailserverParsedNode, err := discv5.ParseNode(*enodeAddr)
+	if err != nil {
+		logger.Crit("Invalid mailserver address specified", "enodeAddr", *enodeAddr, "error", err)
+		os.Exit(1)
+	}
+
+	verifyMailserverBehavior(mailserverParsedNode)
+	logger.Info("Mailserver responded correctly", "address", enodeAddr)
+	os.Exit(0)
+}
+
+func init() {
+	flag.Parse()
+
+	colors := !(*logWithoutColors)
+	if colors {
+		colors = terminal.IsTerminal(int(os.Stdin.Fd()))
+	}
+
+	if err := logutils.OverrideRootLog(*logLevel != "", *logLevel, *logFile, colors); err != nil {
+		stdlog.Fatalf("Error initializing logger: %s", err)
+	}
+}
+
+func verifyMailserverBehavior(mailserverNode *discv5.Node) {
+	clientBackend, err := startClientNode()
+	if err != nil {
+		logger.Error("Node start failed", "error", err)
+		os.Exit(1)
+	}
+	defer func() { _ = clientBackend.StopNode() }()
+
+	clientNode := clientBackend.StatusNode()
+	clientWhisperService, err := clientNode.WhisperService()
+	if err != nil {
+		logger.Error("Could not retrieve Whisper service", "error", err)
+		os.Exit(1)
+	}
+	clientShhExtService, err := clientNode.ShhExtService()
+	if err != nil {
+		logger.Error("Could not retrieve shhext service", "error", err)
+		os.Exit(1)
+	}
+
+	// add mailserver peer to client
+	clientErrCh := helpers.WaitForPeerAsync(clientNode.Server(), *enodeAddr, p2p.PeerEventTypeAdd, 5*time.Second)
+	err = clientNode.AddPeer(*enodeAddr)
+	if err != nil {
+		logger.Error("Failed to add mailserver peer to client", "error", err)
+		os.Exit(1)
+	}
+
+	err = <-clientErrCh
+	if err != nil {
+		logger.Error("Error detected while waiting for mailserver peer to be added", "error", err)
+		os.Exit(1)
+	}
+
+	// add mailserver sym key
+	mailServerKeyID, err := clientWhisperService.AddSymKeyFromPassword(mailboxPassword)
+	if err != nil {
+		logger.Error("Error adding mailserver sym key to client peer", "error", err)
+		os.Exit(1)
+	}
+
+	mailboxPeer := mailserverNode.ID[:]
+	mailboxPeerStr := mailserverNode.ID.String()
+	err = clientWhisperService.AllowP2PMessagesFromPeer(mailboxPeer)
+	if err != nil {
+		logger.Error("Failed to allow P2P messages from peer", "error", err)
+		os.Exit(1)
+	}
+
+	clientRPCClient := clientNode.RPCClient()
+
+	// TODO: Replace chat implementation with github.com/status-im/status-go-sdk
+	_, topic, _, err := joinPublicChat(clientWhisperService, clientRPCClient, *publicChannel)
+	if err != nil {
+		logger.Error("Failed to join public chat", "error", err)
+		os.Exit(1)
+	}
+
+	// watch for envelopes to be available in filters in the client
+	envelopeAvailableWatcher := make(chan whisper.EnvelopeEvent, 1024)
+	sub := clientWhisperService.SubscribeEnvelopeEvents(envelopeAvailableWatcher)
+	defer sub.Unsubscribe()
+
+	// watch for mailserver responses in the client
+	mailServerResponseWatcher := make(chan whisper.EnvelopeEvent, 1024)
+	sub = clientWhisperService.SubscribeEnvelopeEvents(mailServerResponseWatcher)
+	defer sub.Unsubscribe()
+
+	// request messages from mailbox
+	shhextAPI := shhext.NewPublicAPI(clientShhExtService)
+	requestIDBytes, err := shhextAPI.RequestMessages(context.TODO(),
+		shhext.MessagesRequest{
+			MailServerPeer: mailboxPeerStr,
+			From:           uint32(clientWhisperService.GetCurrentTime().Add(-time.Duration(*period) * time.Second).Unix()),
+			Limit:          1,
+			Topic:          topic,
+			SymKeyID:       mailServerKeyID,
+		})
+	if err != nil {
+		logger.Error("Error requesting historic messages from mailserver", "error", err)
+		os.Exit(2)
+	}
+	requestID := common.BytesToHash(requestIDBytes)
+
+	// wait for mailserver response
+	resp, err := waitForMailServerResponse(mailServerResponseWatcher, requestID, 10*time.Second)
+	if err != nil {
+		logger.Error("Error waiting for mailserver response", "error", err)
+		os.Exit(3)
+	}
+
+	// wait for last envelope sent by the mailserver to be available for filters
+	err = waitForEnvelopeEvents(envelopeAvailableWatcher, []string{resp.LastEnvelopeHash.String()}, whisper.EventEnvelopeAvailable)
+	if err != nil {
+		logger.Error("Error waiting for envelopes to be available to client filter", "error", err)
+		os.Exit(4)
+	}
+}
+
+// makeNodeConfig parses incoming CLI options and returns node configuration object
+func makeNodeConfig() (*params.NodeConfig, error) {
+	err := error(nil)
+
+	workDir := ""
+	if path.IsAbs(*homePath) {
+		workDir = *homePath
+	} else {
+		workDir, err = filepath.Abs(filepath.Dir(os.Args[0]))
+		if err == nil {
+			workDir = path.Join(workDir, *homePath)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	nodeConfig, err := params.NewNodeConfig(path.Join(workDir, ".ethereum"), "", uint64(params.RopstenNetworkID))
+	if err != nil {
+		return nil, err
+	}
+
+	if *logLevel != "" {
+		nodeConfig.LogLevel = *logLevel
+		nodeConfig.LogEnabled = true
+	}
+
+	if *logFile != "" {
+		nodeConfig.LogFile = *logFile
+	}
+
+	nodeConfig.NoDiscovery = true
+
+	return whisperConfig(nodeConfig)
+}
+
+// whisperConfig creates node configuration object from flags
+func whisperConfig(nodeConfig *params.NodeConfig) (*params.NodeConfig, error) {
+	whisperConfig := nodeConfig.WhisperConfig
+	whisperConfig.Enabled = true
+	whisperConfig.LightClient = true
+	whisperConfig.MinimumPoW = *minPow
+	whisperConfig.TTL = *ttl
+	whisperConfig.EnableNTPSync = true
+
+	return nodeConfig, nil
+}
+
+func startClientNode() (*api.StatusBackend, error) {
+	config, err := makeNodeConfig()
+	if err != nil {
+		return nil, err
+	}
+	clientBackend := api.NewStatusBackend()
+	err = clientBackend.StartNode(config)
+	if err != nil {
+		return nil, err
+	}
+	return clientBackend, err
+}
+
+func joinPublicChat(w *whisper.Whisper, rpcClient *rpc.Client, name string) (string, whisper.TopicType, string, error) {
+	keyID, err := w.AddSymKeyFromPassword(name)
+	if err != nil {
+		return "", whisper.TopicType{}, "", err
+	}
+
+	h := sha3.NewKeccak256()
+	_, err = h.Write([]byte(name))
+	if err != nil {
+		return "", whisper.TopicType{}, "", err
+	}
+	fullTopic := h.Sum(nil)
+	topic := whisper.BytesToTopic(fullTopic)
+
+	whisperAPI := whisper.NewPublicWhisperAPI(w)
+	filterID, err := whisperAPI.NewMessageFilter(whisper.Criteria{SymKeyID: keyID, Topics: []whisper.TopicType{topic}})
+
+	return keyID, topic, filterID, err
+}
+
+func waitForMailServerResponse(events chan whisper.EnvelopeEvent, requestID common.Hash, timeout time.Duration) (*whisper.MailServerResponse, error) {
+	timeoutTimer := time.NewTimer(timeout)
+	for {
+		select {
+		case event := <-events:
+			if event.Hash == requestID {
+				resp, err := decodeMailServerResponse(event)
+				if resp != nil || err != nil {
+					timeoutTimer.Stop()
+					return resp, err
+				}
+			}
+		case <-timeoutTimer.C:
+			return nil, errors.New("timed out waiting for mailserver response")
+		}
+	}
+}
+
+func decodeMailServerResponse(event whisper.EnvelopeEvent) (*whisper.MailServerResponse, error) {
+	switch event.Event {
+	case whisper.EventMailServerRequestCompleted:
+		resp, ok := event.Data.(*whisper.MailServerResponse)
+		if !ok {
+			return nil, errors.New("failed to convert event to a *MailServerResponse")
+		}
+
+		return resp, nil
+	case whisper.EventMailServerRequestExpired:
+		return nil, errors.New("no messages available from mailserver")
+	default:
+		return nil, errors.New("unexpected event type")
+	}
+}
+
+func waitForEnvelopeEvents(events chan whisper.EnvelopeEvent, hashes []string, event whisper.EventType) error {
+	check := make(map[string]struct{})
+	for _, hash := range hashes {
+		check[hash] = struct{}{}
+	}
+
+	timeout := time.NewTimer(time.Second * 5)
+	for {
+		select {
+		case e := <-events:
+			if e.Event == event {
+				delete(check, e.Hash.String())
+				if len(check) == 0 {
+					timeout.Stop()
+					return nil
+				}
+			}
+		case <-timeout.C:
+			return fmt.Errorf("timed out while waiting for event on envelopes. event: %s", event)
+		}
+	}
+}

node/status_node.go

@@ -26,6 +26,7 @@ import (
 	"github.com/status-im/status-go/peers"
 	"github.com/status-im/status-go/rpc"
 	"github.com/status-im/status-go/services/peer"
+	"github.com/status-im/status-go/services/shhext"
 	"github.com/status-im/status-go/services/status"
 )
 
@@ -494,6 +495,19 @@ func (n *StatusNode) WhisperService() (w *whisper.Whisper, err error) {
 	return
 }
 
+// ShhExtService exposes reference to shh extension service running on top of the node
+func (n *StatusNode) ShhExtService() (s *shhext.Service, err error) {
+	n.mu.RLock()
+	defer n.mu.RUnlock()
+
+	err = n.gethService(&s)
+	if err == node.ErrServiceUnknown {
+		err = ErrServiceUnknown
+	}
+
+	return
+}
+
 // AccountManager exposes reference to node's accounts manager
 func (n *StatusNode) AccountManager() (*accounts.Manager, error) {
 	n.mu.RLock()

services/shhext/README.md

@@ -49,7 +49,7 @@ Sends a request for historic messages to a mail server.
 - `from`:`QUANTITY` - (optional) Lower bound of time range as unix timestamp, default is 24 hours back from now
 - `to`:`QUANTITY`- (optional) Upper bound of time range as unix timestamp, default is now
 - `topic`:`DATA`, 4 Bytes - Regular whisper topic
-- `symKeyID`:`DATA`- ID of a symmetric key to authenticate to mail server, derived form mail server password
+- `symKeyID`:`DATA`- ID of a symmetric key to authenticate to mail server, derived from mail server password
 
 ##### Returns
 

t/e2e/whisper/whisper_mailbox_test.go

@@ -344,7 +344,7 @@ func (s *WhisperMailboxSuite) TestRequestMessagesWithPagination() {
 	s.Require().True(client.IsNodeRunning())
 	clientRPCClient := client.StatusNode().RPCPrivateClient()
 
-	// Add mailbox to clients's peers
+	// Add mailbox to client's peers
 	errCh := helpers.WaitForPeerAsync(client.StatusNode().Server(), mailboxEnode, p2p.PeerEventTypeAdd, time.Second)
 	s.Require().NoError(client.StatusNode().AddPeer(mailboxEnode))
 	s.Require().NoError(<-errCh)