whisper: Status message changed, more params added

whisper/whisperv6/peer.go

@@ -75,8 +75,12 @@ func (p *Peer) handshake() error {
 	// Send the handshake status message asynchronously
 	errc := make(chan error, 1)
 	go func() {
-		errc <- p2p.Send(p.ws, statusCode, ProtocolVersion)
+		pow := p.host.MinPow()
+		powConverted := math.Float64bits(pow)
+		bloom := p.host.BloomFilter()
+		errc <- p2p.SendItems(p.ws, statusCode, ProtocolVersion, powConverted, bloom)
 	}()
+
 	// Fetch the remote status packet and verify protocol match
 	packet, err := p.ws.ReadMsg()
 	if err != nil {
@@ -86,14 +90,42 @@ func (p *Peer) handshake() error {
 		return fmt.Errorf("peer [%x] sent packet %x before status packet", p.ID(), packet.Code)
 	}
 	s := rlp.NewStream(packet.Payload, uint64(packet.Size))
-	peerVersion, err := s.Uint()
+	_, err = s.List()
 	if err != nil {
 		return fmt.Errorf("peer [%x] sent bad status message: %v", p.ID(), err)
 	}
+	peerVersion, err := s.Uint()
+	if err != nil {
+		return fmt.Errorf("peer [%x] sent bad status message (unable to decode version): %v", p.ID(), err)
+	}
 	if peerVersion != ProtocolVersion {
 		return fmt.Errorf("peer [%x]: protocol version mismatch %d != %d", p.ID(), peerVersion, ProtocolVersion)
 	}
-	// Wait until out own status is consumed too
+
+	// only version is mandatory, subsequent parameters are optional
+	powRaw, err := s.Uint()
+	if err == nil {
+		pow := math.Float64frombits(powRaw)
+		if math.IsInf(pow, 0) || math.IsNaN(pow) || pow < 0.0 {
+			return fmt.Errorf("peer [%x] sent bad status message: invalid pow", p.ID())
+		}
+		p.powRequirement = pow
+
+		var bloom []byte
+		err = s.Decode(&bloom)
+		if err == nil {
+			sz := len(bloom)
+			if sz != bloomFilterSize && sz != 0 {
+				return fmt.Errorf("peer [%x] sent bad status message: wrong bloom filter size %d", p.ID(), sz)
+			}
+			if isFullNode(bloom) {
+				p.bloomFilter = nil
+			} else {
+				p.bloomFilter = bloom
+			}
+		}
+	}
+
 	if err := <-errc; err != nil {
 		return fmt.Errorf("peer [%x] failed to send status packet: %v", p.ID(), err)
 	}

whisper/whisperv6/peer_test.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"crypto/ecdsa"
 	"fmt"
+	mrand "math/rand"
 	"net"
 	"sync"
 	"testing"
@@ -87,6 +88,9 @@ var nodes [NumNodes]*TestNode
 var sharedKey []byte = []byte("some arbitrary data here")
 var sharedTopic TopicType = TopicType{0xF, 0x1, 0x2, 0}
 var expectedMessage []byte = []byte("per rectum ad astra")
+var masterBloomFilter []byte
+var masterPow = 0.00000001
+var round int = 1
 
 func TestSimulation(t *testing.T) {
 	// create a chain of whisper nodes,
@@ -104,8 +108,13 @@ func TestSimulation(t *testing.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
+	// check if Status message was correctly decoded
-	powReqExchange(t)
+	checkBloomFilterExchange(t)
+	checkPowExchange(t)
+
+	// send new pow and bloom exchange messages
+	resetParams(t)
+	round++
 
 	// node #1 sends one expected (decryptable) message
 	sendMsg(t, true, 1)
@@ -113,18 +122,65 @@ func TestSimulation(t *testing.T) {
 	// check if each node (except node #0) have received and decrypted exactly one message
 	checkPropagation(t, false)
 
+	for i := 1; i < NumNodes; i++ {
+		time.Sleep(20 * time.Millisecond)
+		sendMsg(t, true, i)
+	}
+
+	// check if corresponding protocol-level messages were correctly decoded
+	checkPowExchangeForNodeZero(t)
+	checkBloomFilterExchange(t)
+
 	stopServers()
 }
 
+func resetParams(t *testing.T) {
+	// change pow only for node zero
+	masterPow = 7777777.0
+	nodes[0].shh.SetMinimumPoW(masterPow)
+
+	// change bloom for all nodes
+	masterBloomFilter = TopicToBloom(sharedTopic)
+	for i := 0; i < NumNodes; i++ {
+		nodes[i].shh.SetBloomFilter(masterBloomFilter)
+	}
+}
+
+func initBloom(t *testing.T) {
+	masterBloomFilter = make([]byte, bloomFilterSize)
+	_, err := mrand.Read(masterBloomFilter)
+	if err != nil {
+		t.Fatalf("rand failed: %s.", err)
+	}
+
+	msgBloom := TopicToBloom(sharedTopic)
+	masterBloomFilter = addBloom(masterBloomFilter, msgBloom)
+	for i := 0; i < 32; i++ {
+		masterBloomFilter[i] = 0xFF
+	}
+
+	if !bloomFilterMatch(masterBloomFilter, msgBloom) {
+		t.Fatalf("bloom mismatch on initBloom.")
+	}
+}
+
 func initialize(t *testing.T) {
+	initBloom(t)
+
 	var err error
 	ip := net.IPv4(127, 0, 0, 1)
 	port0 := 30303
 
 	for i := 0; i < NumNodes; i++ {
 		var node TestNode
+		b := make([]byte, bloomFilterSize)
+		copy(b, masterBloomFilter)
 		node.shh = New(&DefaultConfig)
-		node.shh.SetMinimumPowTest(0.00000001)
+		node.shh.SetMinimumPoW(masterPow)
+		node.shh.SetBloomFilter(b)
+		if !isBloomFilterEqual(node.shh.BloomFilter(), masterBloomFilter) {
+			t.Fatalf("bloom mismatch on init.")
+		}
 		node.shh.Start(nil)
 		topics := make([]TopicType, 0)
 		topics = append(topics, sharedTopic)
@@ -206,7 +262,7 @@ func checkPropagation(t *testing.T, includingNodeZero bool) {
 		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)
+				t.Fatalf("failed to get filterId %s from node %d, round %d.", nodes[i].filerId, i, round)
 			}
 
 			mail := f.Retrieve()
@@ -332,34 +388,52 @@ func TestPeerBasic(t *testing.T) {
 	}
 }
 
-func powReqExchange(t *testing.T) {
+func checkPowExchangeForNodeZero(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 {
+				if peer.powRequirement != masterPow {
 					t.Fatalf("node %d: failed to set the new pow requirement.", i)
 				}
 			}
 		}
 	}
-
 	if cnt == 0 {
 		t.Fatalf("no matching peers found.")
 	}
 }
+
+func checkPowExchange(t *testing.T) {
+	for i, node := range nodes {
+		for peer := range node.shh.peers {
+			if peer.peer.ID() != discover.PubkeyID(&nodes[0].id.PublicKey) {
+				if peer.powRequirement != masterPow {
+					t.Fatalf("node %d: failed to exchange pow requirement in round %d; expected %f, got %f",
+						i, round, masterPow, peer.powRequirement)
+				}
+			}
+		}
+	}
+}
+
+func checkBloomFilterExchange(t *testing.T) {
+	for i, node := range nodes {
+		for peer := range node.shh.peers {
+			if !isBloomFilterEqual(peer.bloomFilter, masterBloomFilter) {
+				t.Fatalf("node %d: failed to exchange bloom filter requirement in round %d. \n%x expected \n%x got",
+					i, round, masterBloomFilter, peer.bloomFilter)
+			}
+		}
+	}
+}
+
+func isBloomFilterEqual(a, b []byte) bool {
+	for i := 0; i < bloomFilterSize; i++ {
+		if a[i] != b[i] {
+			return false
+		}
+	}
+	return true
+}

whisper/whisperv6/whisper.go

@@ -130,32 +130,32 @@ func New(cfg *Config) *Whisper {
 }
 
 func (w *Whisper) MinPow() float64 {
-	val, _ := w.settings.Load(minPowIdx)
+	val, exist := w.settings.Load(minPowIdx)
-	if val == nil {
+	if !exist || val == nil {
 		return DefaultMinimumPoW
 	}
 	return val.(float64)
 }
 
 func (w *Whisper) MinPowTolerance() float64 {
-	val, _ := w.settings.Load(minPowToleranceIdx)
+	val, exist := w.settings.Load(minPowToleranceIdx)
-	if val == nil {
+	if !exist || val == nil {
 		return DefaultMinimumPoW
 	}
 	return val.(float64)
 }
 
 func (w *Whisper) BloomFilter() []byte {
-	val, _ := w.settings.Load(bloomFilterIdx)
+	val, exist := w.settings.Load(bloomFilterIdx)
-	if val == nil {
+	if !exist || val == nil {
 		return nil
 	}
 	return val.([]byte)
 }
 
 func (w *Whisper) BloomFilterTolerance() []byte {
-	val, _ := w.settings.Load(bloomFilterToleranceIdx)
+	val, exist := w.settings.Load(bloomFilterToleranceIdx)
-	if val == nil {
+	if !exist || val == nil {
 		return nil
 	}
 	return val.([]byte)
@@ -216,13 +216,16 @@ func (w *Whisper) SetBloomFilter(bloom []byte) error {
 		return fmt.Errorf("invalid bloom filter size: %d", len(bloom))
 	}
 
-	w.settings.Store(bloomFilterIdx, bloom)
+	b := make([]byte, bloomFilterSize)
-	w.notifyPeersAboutBloomFilterChange(bloom)
+	copy(b, bloom)
+
+	w.settings.Store(bloomFilterIdx, b)
+	w.notifyPeersAboutBloomFilterChange(b)
 
 	go func() {
 		// allow some time before all the peers have processed the notification
 		time.Sleep(time.Duration(w.syncAllowance) * time.Second)
-		w.settings.Store(bloomFilterToleranceIdx, bloom)
+		w.settings.Store(bloomFilterToleranceIdx, b)
 	}()
 
 	return nil
@@ -253,13 +256,6 @@ func (w *Whisper) SetMinimumPowTest(val float64) {
 	w.settings.Store(minPowToleranceIdx, val)
 }
 
-// SetBloomFilterTest sets the Bloom Filter in test environment
-func (w *Whisper) SetBloomFilterTest(bloom []byte) {
-	w.settings.Store(bloomFilterIdx, bloom)
-	w.notifyPeersAboutBloomFilterChange(bloom)
-	w.settings.Store(bloomFilterToleranceIdx, bloom)
-}
-
 func (w *Whisper) notifyPeersAboutPowRequirementChange(pow float64) {
 	arr := w.getPeers()
 	for _, p := range arr {
@@ -697,7 +693,7 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 				log.Warn("failed to decode bloom filter exchange message, peer will be disconnected", "peer", p.peer.ID(), "err", err)
 				return errors.New("invalid bloom filter exchange message")
 			}
-			if isFulNode(bloom) {
+			if isFullNode(bloom) {
 				p.bloomFilter = nil
 			} else {
 				p.bloomFilter = bloom
@@ -765,21 +761,20 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 
 	if envelope.PoW() < wh.MinPow() {
 		// maybe the value was recently changed, and the peers did not adjust yet.
-		// some tolerance might be still allowed for a short period of adjustment time.
+		// in this case the previous value is retrieved by MinPowTolerance()
+		// for a short period of peer synchronization.
 		if envelope.PoW() < wh.MinPowTolerance() {
-			log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", envelope.Hash().Hex())
+			return false, fmt.Errorf("envelope with low PoW received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex())
-			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 received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex())
 	}
 
 	if !bloomFilterMatch(wh.BloomFilter(), envelope.Bloom()) {
 		// maybe the value was recently changed, and the peers did not adjust yet.
-		// some tolerance might be still allowed for a short period of adjustment time.
+		// in this case the previous value is retrieved by BloomFilterTolerance()
+		// for a short period of peer synchronization.
 		if !bloomFilterMatch(wh.BloomFilterTolerance(), envelope.Bloom()) {
-			return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v]", envelope.Hash().Hex())
+			return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v], bloom: \n%x \n%x \n%x",
+				envelope.Hash().Hex(), wh.BloomFilter(), envelope.Bloom(), envelope.Topic)
 		}
 	}
 
@@ -1019,7 +1014,7 @@ func GenerateRandomID() (id string, err error) {
 	return id, err
 }
 
-func isFulNode(bloom []byte) bool {
+func isFullNode(bloom []byte) bool {
 	if bloom == nil {
 		return true
 	}

whisper/whisperv6/whisper_test.go

@@ -843,3 +843,64 @@ func TestSymmetricSendKeyMismatch(t *testing.T) {
 		t.Fatalf("received a message when keys weren't matching")
 	}
 }
+
+func TestBloom(t *testing.T) {
+	topic := TopicType{0, 0, 255, 6}
+	b := TopicToBloom(topic)
+	x := make([]byte, bloomFilterSize)
+	x[0] = byte(1)
+	x[32] = byte(1)
+	x[bloomFilterSize-1] = byte(128)
+	if !bloomFilterMatch(x, b) || !bloomFilterMatch(b, x) {
+		t.Fatalf("bloom filter does not match the mask")
+	}
+
+	_, err := mrand.Read(b)
+	if err != nil {
+		t.Fatalf("math rand error")
+	}
+	_, err = mrand.Read(x)
+	if err != nil {
+		t.Fatalf("math rand error")
+	}
+	if !bloomFilterMatch(b, b) {
+		t.Fatalf("bloom filter does not match self")
+	}
+	x = addBloom(x, b)
+	if !bloomFilterMatch(x, b) {
+		t.Fatalf("bloom filter does not match combined bloom")
+	}
+	if !isFullNode(nil) {
+		t.Fatalf("isFullNode did not recognize nil as full node")
+	}
+	x[17] = 254
+	if isFullNode(x) {
+		t.Fatalf("isFullNode false positive")
+	}
+	for i := 0; i < bloomFilterSize; i++ {
+		b[i] = byte(255)
+	}
+	if !isFullNode(b) {
+		t.Fatalf("isFullNode false negative")
+	}
+	if bloomFilterMatch(x, b) {
+		t.Fatalf("bloomFilterMatch false positive")
+	}
+	if !bloomFilterMatch(b, x) {
+		t.Fatalf("bloomFilterMatch false negative")
+	}
+
+	w := New(&DefaultConfig)
+	f := w.BloomFilter()
+	if f != nil {
+		t.Fatalf("wrong bloom on creation")
+	}
+	err = w.SetBloomFilter(x)
+	if err != nil {
+		t.Fatalf("failed to set bloom filter: %s", err)
+	}
+	f = w.BloomFilter()
+	if !bloomFilterMatch(f, x) || !bloomFilterMatch(x, f) {
+		t.Fatalf("retireved wrong bloom filter")
+	}
+}