whisper/whisperv6: PoW requirement (#15701)

New Whisper-level message introduced (PoW requirement),
corresponding logic added, plus some tests.

5 files changed, 185 insertions, 54 deletions

diff --git a/whisper/whisperv6/doc.go b/whisper/whisperv6/doc.go
index 64925ba48..2a4911d65 100644
--- a/whisper/whisperv6/doc.go
+++ b/whisper/whisperv6/doc.go
@@ -40,10 +40,13 @@ const (
 	ProtocolVersionStr = "6.0"
 	ProtocolName       = "shh"
 
-	statusCode           = 0 // used by whisper protocol
-	messagesCode         = 1 // normal whisper message
-	p2pCode              = 2 // peer-to-peer message (to be consumed by the peer, but not forwarded any further)
-	p2pRequestCode       = 3 // peer-to-peer message, used by Dapp protocol
+	// whisper protocol message codes, according to EIP-627
+	statusCode           = 0   // used by whisper protocol
+	messagesCode         = 1   // normal whisper message
+	powRequirementCode   = 2   // PoW requirement
+	bloomFilterExCode    = 3   // bloom filter exchange
+	p2pRequestCode       = 126 // peer-to-peer message, used by Dapp protocol
+	p2pMessageCode       = 127 // peer-to-peer message (to be consumed by the peer, but not forwarded any further)
 	NumberOfMessageCodes = 128
 
 	paddingMask   = byte(3)
diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go
index ffc39505e..65e0c77b0 100644
--- a/whisper/whisperv6/peer.go
+++ b/whisper/whisperv6/peer.go
@@ -18,6 +18,7 @@ package whisperv6
 
 import (
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -29,10 +30,12 @@ import (
 
 // peer represents a whisper protocol peer connection.
 type Peer struct {
-	host    *Whisper
-	peer    *p2p.Peer
-	ws      p2p.MsgReadWriter
-	trusted bool
+	host *Whisper
+	peer *p2p.Peer
+	ws   p2p.MsgReadWriter
+
+	trusted        bool
+	powRequirement float64
 
 	known *set.Set // Messages already known by the peer to avoid wasting bandwidth
 
@@ -42,12 +45,13 @@ type Peer struct {
 // newPeer creates a new whisper peer object, but does not run the handshake itself.
 func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *Peer {
 	return &Peer{
-		host:    host,
-		peer:    remote,
-		ws:      rw,
-		trusted: false,
-		known:   set.New(),
-		quit:    make(chan struct{}),
+		host:           host,
+		peer:           remote,
+		ws:             rw,
+		trusted:        false,
+		powRequirement: 0.0,
+		known:          set.New(),
+		quit:           make(chan struct{}),
 	}
 }
 
@@ -152,7 +156,7 @@ func (p *Peer) broadcast() error {
 	envelopes := p.host.Envelopes()
 	bundle := make([]*Envelope, 0, len(envelopes))
 	for _, envelope := range envelopes {
-		if !p.marked(envelope) {
+		if !p.marked(envelope) && envelope.PoW() >= p.powRequirement {
 			bundle = append(bundle, envelope)
 		}
 	}
@@ -177,3 +181,8 @@ func (p *Peer) ID() []byte {
 	id := p.peer.ID()
 	return id[:]
 }
+
+func (p *Peer) notifyAboutPowRequirementChange(pow float64) error {
+	i := math.Float64bits(pow)
+	return p2p.Send(p.ws, powRequirementCode, i)
+}
diff --git a/whisper/whisperv6/peer_test.go b/whisper/whisperv6/peer_test.go
index 39a4ab198..599a479be 100644
--- a/whisper/whisperv6/peer_test.go
+++ b/whisper/whisperv6/peer_test.go
@@ -88,21 +88,31 @@ var sharedKey []byte = []byte("some arbitrary data here")
 var sharedTopic TopicType = TopicType{0xF, 0x1, 0x2, 0}
 var expectedMessage []byte = []byte("per rectum ad astra")
 
-// This test does the following:
-// 1. creates a chain of whisper nodes,
-// 2. installs the filters with shared (predefined) parameters,
-// 3. each node sends a number of random (undecryptable) messages,
-// 4. first node sends one expected (decryptable) message,
-// 5. checks if each node have received and decrypted exactly one message.
 func TestSimulation(t *testing.T) {
+	// create a chain of whisper nodes,
+	// installs the filters with shared (predefined) parameters
 	initialize(t)
 
+	// each node sends a number of random (undecryptable) messages
 	for i := 0; i < NumNodes; i++ {
 		sendMsg(t, false, i)
 	}
 
+	// node #0 sends one expected (decryptable) message
 	sendMsg(t, true, 0)
-	checkPropagation(t)
+
+	// check if each node have received and decrypted exactly one message
+	checkPropagation(t, true)
+
+	// send protocol-level messages (powRequirementCode) and check the new PoW requirement values
+	powReqExchange(t)
+
+	// node #1 sends one expected (decryptable) message
+	sendMsg(t, true, 1)
+
+	// check if each node (except node #0) have received and decrypted exactly one message
+	checkPropagation(t, false)
+
 	stopServers()
 }
 
@@ -114,7 +124,7 @@ func initialize(t *testing.T) {
 	for i := 0; i < NumNodes; i++ {
 		var node TestNode
 		node.shh = New(&DefaultConfig)
-		node.shh.SetMinimumPoW(0.00000001)
+		node.shh.SetMinimumPowTest(0.00000001)
 		node.shh.Start(nil)
 		topics := make([]TopicType, 0)
 		topics = append(topics, sharedTopic)
@@ -154,13 +164,18 @@ func initialize(t *testing.T) {
 			},
 		}
 
-		err = node.server.Start()
-		if err != nil {
-			t.Fatalf("failed to start server %d.", i)
-		}
-
 		nodes[i] = &node
 	}
+
+	for i := 1; i < NumNodes; i++ {
+		go nodes[i].server.Start()
+	}
+
+	// we need to wait until the first node actually starts
+	err = nodes[0].server.Start()
+	if err != nil {
+		t.Fatalf("failed to start the fisrt server.")
+	}
 }
 
 func stopServers() {
@@ -174,18 +189,21 @@ func stopServers() {
 	}
 }
 
-func checkPropagation(t *testing.T) {
+func checkPropagation(t *testing.T, includingNodeZero bool) {
 	if t.Failed() {
 		return
 	}
 
-	const cycle = 100
-	const iterations = 100
+	const cycle = 50
+	const iterations = 200
 
-	for j := 0; j < iterations; j++ {
-		time.Sleep(cycle * time.Millisecond)
+	first := 0
+	if !includingNodeZero {
+		first = 1
+	}
 
-		for i := 0; i < NumNodes; i++ {
+	for j := 0; j < iterations; j++ {
+		for i := first; i < NumNodes; i++ {
 			f := nodes[i].shh.GetFilter(nodes[i].filerId)
 			if f == nil {
 				t.Fatalf("failed to get filterId %s from node %d.", nodes[i].filerId, i)
@@ -200,9 +218,18 @@ func checkPropagation(t *testing.T) {
 				return
 			}
 		}
+
+		time.Sleep(cycle * time.Millisecond)
 	}
 
 	t.Fatalf("Test was not complete: timeout %d seconds.", iterations*cycle/1000)
+
+	if !includingNodeZero {
+		f := nodes[0].shh.GetFilter(nodes[0].filerId)
+		if f != nil {
+			t.Fatalf("node zero received a message with low PoW.")
+		}
+	}
 }
 
 func validateMail(t *testing.T, index int, mail []*ReceivedMessage) bool {
@@ -304,3 +331,35 @@ func TestPeerBasic(t *testing.T) {
 		t.Fatalf("failed mark with seed %d.", seed)
 	}
 }
+
+func powReqExchange(t *testing.T) {
+	for i, node := range nodes {
+		for peer := range node.shh.peers {
+			if peer.powRequirement > 1000.0 {
+				t.Fatalf("node %d: one of the peers' pow requirement is too big (%f).", i, peer.powRequirement)
+			}
+		}
+	}
+
+	const pow float64 = 7777777.0
+	nodes[0].shh.SetMinimumPoW(pow)
+
+	// wait until all the messages are delivered
+	time.Sleep(64 * time.Millisecond)
+
+	cnt := 0
+	for i, node := range nodes {
+		for peer := range node.shh.peers {
+			if peer.peer.ID() == discover.PubkeyID(&nodes[0].id.PublicKey) {
+				cnt++
+				if peer.powRequirement != pow {
+					t.Fatalf("node %d: failed to set the new pow requirement.", i)
+				}
+			}
+		}
+	}
+
+	if cnt == 0 {
+		t.Fatalf("no matching peers found.")
+	}
+}
diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go
index 2cc1e64f5..492591486 100644
--- a/whisper/whisperv6/whisper.go
+++ b/whisper/whisperv6/whisper.go
@@ -22,6 +22,7 @@ import (
 	crand "crypto/rand"
 	"crypto/sha256"
 	"fmt"
+	"math"
 	"runtime"
 	"sync"
 	"time"
@@ -30,6 +31,7 @@ import (
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
+	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/syndtr/goleveldb/leveldb/errors"
 	"golang.org/x/crypto/pbkdf2"
@@ -74,6 +76,8 @@ type Whisper struct {
 
 	settings syncmap.Map // holds configuration settings that can be dynamically changed
 
+	reactionAllowance int // maximum time in seconds allowed to process the whisper-related messages
+
 	statsMu sync.Mutex // guard stats
 	stats   Statistics // Statistics of whisper node
 
@@ -87,14 +91,15 @@ func New(cfg *Config) *Whisper {
 	}
 
 	whisper := &Whisper{
-		privateKeys:  make(map[string]*ecdsa.PrivateKey),
-		symKeys:      make(map[string][]byte),
-		envelopes:    make(map[common.Hash]*Envelope),
-		expirations:  make(map[uint32]*set.SetNonTS),
-		peers:        make(map[*Peer]struct{}),
-		messageQueue: make(chan *Envelope, messageQueueLimit),
-		p2pMsgQueue:  make(chan *Envelope, messageQueueLimit),
-		quit:         make(chan struct{}),
+		privateKeys:       make(map[string]*ecdsa.PrivateKey),
+		symKeys:           make(map[string][]byte),
+		envelopes:         make(map[common.Hash]*Envelope),
+		expirations:       make(map[uint32]*set.SetNonTS),
+		peers:             make(map[*Peer]struct{}),
+		messageQueue:      make(chan *Envelope, messageQueueLimit),
+		p2pMsgQueue:       make(chan *Envelope, messageQueueLimit),
+		quit:              make(chan struct{}),
+		reactionAllowance: SynchAllowance,
 	}
 
 	whisper.filters = NewFilters(whisper)
@@ -177,13 +182,50 @@ func (w *Whisper) SetMaxMessageSize(size uint32) error {
 
 // SetMinimumPoW sets the minimal PoW required by this node
 func (w *Whisper) SetMinimumPoW(val float64) error {
-	if val <= 0.0 {
+	if val < 0.0 {
 		return fmt.Errorf("invalid PoW: %f", val)
 	}
-	w.settings.Store(minPowIdx, val)
+
+	w.notifyPeersAboutPowRequirementChange(val)
+
+	go func() {
+		// allow some time before all the peers have processed the notification
+		time.Sleep(time.Duration(w.reactionAllowance) * time.Second)
+		w.settings.Store(minPowIdx, val)
+	}()
+
 	return nil
 }
 
+// SetMinimumPoW sets the minimal PoW in test environment
+func (w *Whisper) SetMinimumPowTest(val float64) {
+	w.notifyPeersAboutPowRequirementChange(val)
+	w.settings.Store(minPowIdx, val)
+}
+
+func (w *Whisper) notifyPeersAboutPowRequirementChange(pow float64) {
+	arr := make([]*Peer, len(w.peers))
+	i := 0
+
+	w.peerMu.Lock()
+	for p := range w.peers {
+		arr[i] = p
+		i++
+	}
+	w.peerMu.Unlock()
+
+	for _, p := range arr {
+		err := p.notifyAboutPowRequirementChange(pow)
+		if err != nil {
+			// allow one retry
+			err = p.notifyAboutPowRequirementChange(pow)
+		}
+		if err != nil {
+			log.Warn("oversized message received", "peer", p.ID(), "error", err)
+		}
+	}
+}
+
 // getPeer retrieves peer by ID
 func (w *Whisper) getPeer(peerID []byte) (*Peer, error) {
 	w.peerMu.Lock()
@@ -233,7 +275,7 @@ func (w *Whisper) SendP2PMessage(peerID []byte, envelope *Envelope) error {
 
 // SendP2PDirect sends a peer-to-peer message to a specific peer.
 func (w *Whisper) SendP2PDirect(peer *Peer, envelope *Envelope) error {
-	return p2p.Send(peer.ws, p2pCode, envelope)
+	return p2p.Send(peer.ws, p2pMessageCode, envelope)
 }
 
 // NewKeyPair generates a new cryptographic identity for the client, and injects
@@ -536,7 +578,22 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 			if trouble {
 				return errors.New("invalid envelope")
 			}
-		case p2pCode:
+		case powRequirementCode:
+			s := rlp.NewStream(packet.Payload, uint64(packet.Size))
+			i, err := s.Uint()
+			if err != nil {
+				log.Warn("failed to decode powRequirementCode message, peer will be disconnected", "peer", p.peer.ID(), "err", err)
+				return errors.New("invalid powRequirementCode message")
+			}
+			f := math.Float64frombits(i)
+			if math.IsInf(f, 0) || math.IsNaN(f) || f < 0.0 {
+				log.Warn("invalid value in powRequirementCode message, peer will be disconnected", "peer", p.peer.ID(), "err", err)
+				return errors.New("invalid value in powRequirementCode message")
+			}
+			p.powRequirement = f
+		case bloomFilterExCode:
+			// to be implemented
+		case p2pMessageCode:
 			// peer-to-peer message, sent directly to peer bypassing PoW checks, etc.
 			// this message is not supposed to be forwarded to other peers, and
 			// therefore might not satisfy the PoW, expiry and other requirements.
@@ -599,7 +656,10 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 
 	if envelope.PoW() < wh.MinPow() {
 		log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", envelope.Hash().Hex())
-		return false, nil // drop envelope without error
+		return false, nil // drop envelope without error for now
+
+		// once the status message includes the PoW requirement, an error should be returned here:
+		//return false, fmt.Errorf("envelope with low PoW dropped: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex())
 	}
 
 	hash := envelope.Hash()
diff --git a/whisper/whisperv6/whisper_test.go b/whisper/whisperv6/whisper_test.go
index c8f3a9ed7..b391a1161 100644
--- a/whisper/whisperv6/whisper_test.go
+++ b/whisper/whisperv6/whisper_test.go
@@ -472,8 +472,8 @@ func TestExpiry(t *testing.T) {
 	InitSingleTest()
 
 	w := New(&DefaultConfig)
-	w.SetMinimumPoW(0.0000001)
-	defer w.SetMinimumPoW(DefaultMinimumPoW)
+	w.SetMinimumPowTest(0.0000001)
+	defer w.SetMinimumPowTest(DefaultMinimumPoW)
 	w.Start(nil)
 	defer w.Stop()
 
@@ -529,7 +529,7 @@ func TestCustomization(t *testing.T) {
 	InitSingleTest()
 
 	w := New(&DefaultConfig)
-	defer w.SetMinimumPoW(DefaultMinimumPoW)
+	defer w.SetMinimumPowTest(DefaultMinimumPoW)
 	defer w.SetMaxMessageSize(DefaultMaxMessageSize)
 	w.Start(nil)
 	defer w.Stop()
@@ -563,7 +563,7 @@ func TestCustomization(t *testing.T) {
 		t.Fatalf("successfully sent envelope with PoW %.06f, false positive (seed %d).", env.PoW(), seed)
 	}
 
-	w.SetMinimumPoW(smallPoW / 2)
+	w.SetMinimumPowTest(smallPoW / 2)
 	err = w.Send(env)
 	if err != nil {
 		t.Fatalf("failed to send envelope with seed %d: %s.", seed, err)
@@ -625,7 +625,7 @@ func TestSymmetricSendCycle(t *testing.T) {
 	InitSingleTest()
 
 	w := New(&DefaultConfig)
-	defer w.SetMinimumPoW(DefaultMinimumPoW)
+	defer w.SetMinimumPowTest(DefaultMinimumPoW)
 	defer w.SetMaxMessageSize(DefaultMaxMessageSize)
 	w.Start(nil)
 	defer w.Stop()
@@ -714,7 +714,7 @@ func TestSymmetricSendWithoutAKey(t *testing.T) {
 	InitSingleTest()
 
 	w := New(&DefaultConfig)
-	defer w.SetMinimumPoW(DefaultMinimumPoW)
+	defer w.SetMinimumPowTest(DefaultMinimumPoW)
 	defer w.SetMaxMessageSize(DefaultMaxMessageSize)
 	w.Start(nil)
 	defer w.Stop()
@@ -782,7 +782,7 @@ func TestSymmetricSendKeyMismatch(t *testing.T) {
 	InitSingleTest()
 
 	w := New(&DefaultConfig)
-	defer w.SetMinimumPoW(DefaultMinimumPoW)
+	defer w.SetMinimumPowTest(DefaultMinimumPoW)
 	defer w.SetMaxMessageSize(DefaultMaxMessageSize)
 	w.Start(nil)
 	defer w.Stop()