whisper: bloom filter logic introduced

whisper/whisperv6/api.go

@@ -113,15 +113,32 @@ func (api *PublicWhisperAPI) Info(ctx context.Context) Info {
 // SetMaxMessageSize sets the maximum message size that is accepted.
 // Upper limit is defined by MaxMessageSize.
 func (api *PublicWhisperAPI) SetMaxMessageSize(ctx context.Context, size uint32) (bool, error) {
-	return true, api.w.SetMaxMessageSize(size)
+	err := api.w.SetMaxMessageSize(size)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
-// SetMinPow sets the minimum PoW for a message before it is accepted.
+// SetMinPow sets the minimum PoW, and notifies the peers.
 func (api *PublicWhisperAPI) SetMinPoW(ctx context.Context, pow float64) (bool, error) {
-	return true, api.w.SetMinimumPoW(pow)
+	err := api.w.SetMinimumPoW(pow)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
 }
 
-// MarkTrustedPeer marks a peer trusted. , which will allow it to send historic (expired) messages.
+// SetBloomFilter sets the new value of bloom filter, and notifies the peers.
+func (api *PublicWhisperAPI) SetBloomFilter(ctx context.Context, bloom hexutil.Bytes) (bool, error) {
+	err := api.w.SetBloomFilter(bloom)
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
+// MarkTrustedPeer marks a peer trusted, which will allow it to send historic (expired) messages.
 // Note: This function is not adding new nodes, the node needs to exists as a peer.
 func (api *PublicWhisperAPI) MarkTrustedPeer(ctx context.Context, enode string) (bool, error) {
 	n, err := discover.ParseNode(enode)

whisper/whisperv6/doc.go

@@ -35,7 +35,6 @@ import (
 )
 
 const (
-	EnvelopeVersion    = uint64(0)
 	ProtocolVersion    = uint64(6)
 	ProtocolVersionStr = "6.0"
 	ProtocolName       = "shh"
@@ -57,6 +56,7 @@ const (
 	aesKeyLength    = 32
 	AESNonceLength  = 12
 	keyIdSize       = 32
+	bloomFilterSize = 64
 
 	MaxMessageSize        = uint32(10 * 1024 * 1024) // maximum accepted size of a message.
 	DefaultMaxMessageSize = uint32(1024 * 1024)

whisper/whisperv6/envelope.go

@@ -43,8 +43,10 @@ type Envelope struct {
 	Nonce  uint64
 
 	pow float64 // Message-specific PoW as described in the Whisper specification.
+
+	// the following variables should not be accessed directly, use the corresponding function instead: Hash(), Bloom()
 	hash  common.Hash // Cached hash of the envelope to avoid rehashing every time.
-	// Don't access hash directly, use Hash() function instead.
+	bloom []byte
 }
 
 // size returns the size of envelope as it is sent (i.e. public fields only)
@@ -227,3 +229,30 @@ func (e *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) {
 	}
 	return msg
 }
+
+// Bloom maps 4-bytes Topic into 64-byte bloom filter with 3 bits set (at most).
+func (e *Envelope) Bloom() []byte {
+	if e.bloom == nil {
+		e.bloom = TopicToBloom(e.Topic)
+	}
+	return e.bloom
+}
+
+func TopicToBloom(topic TopicType) []byte {
+	var powers = [...]byte{1, 2, 4, 8, 16, 32, 64, 128}
+	b := make([]byte, bloomFilterSize)
+	var index [3]int
+	for j := 0; j < 3; j++ {
+		index[j] = int(topic[j])
+		if (topic[3] & powers[j]) != 0 {
+			index[j] += 256
+		}
+	}
+
+	for j := 0; j < 3; j++ {
+		byteIndex := index[j] / 8
+		bitIndex := index[j] % 8
+		b[byteIndex] = powers[bitIndex]
+	}
+	return b
+}

whisper/whisperv6/peer.go

@@ -36,6 +36,7 @@ type Peer struct {
 
 	trusted        bool
 	powRequirement float64
+	bloomFilter    []byte
 
 	known *set.Set // Messages already known by the peer to avoid wasting bandwidth
 
@@ -156,7 +157,7 @@ func (p *Peer) broadcast() error {
 	envelopes := p.host.Envelopes()
 	bundle := make([]*Envelope, 0, len(envelopes))
 	for _, envelope := range envelopes {
-		if !p.marked(envelope) && envelope.PoW() >= p.powRequirement {
+		if !p.marked(envelope) && envelope.PoW() >= p.powRequirement && p.bloomMatch(envelope) {
 			bundle = append(bundle, envelope)
 		}
 	}
@@ -186,3 +187,16 @@ func (p *Peer) notifyAboutPowRequirementChange(pow float64) error {
 	i := math.Float64bits(pow)
 	return p2p.Send(p.ws, powRequirementCode, i)
 }
+
+func (p *Peer) notifyAboutBloomFilterChange(bloom []byte) error {
+	return p2p.Send(p.ws, bloomFilterExCode, bloom)
+}
+
+func (p *Peer) bloomMatch(env *Envelope) bool {
+	if p.bloomFilter == nil {
+		// no filter - full node, accepts all envelops
+		return true
+	}
+
+	return bloomFilterMatch(p.bloomFilter, env.Bloom())
+}

whisper/whisperv6/whisper.go

@@ -51,6 +51,7 @@ const (
 	minPowIdx      = iota // Minimal PoW required by the whisper node
 	maxMsgSizeIdx  = iota // Maximal message length allowed by the whisper node
 	overflowIdx    = iota // Indicator of message queue overflow
+	bloomFilterIdx = iota // Bloom filter for topics of interest for this node
 )
 
 // Whisper represents a dark communication interface through the Ethereum
@@ -131,6 +132,11 @@ func (w *Whisper) MinPow() float64 {
 	return val.(float64)
 }
 
+func (w *Whisper) BloomFilter() []byte {
+	val, _ := w.settings.Load(bloomFilterIdx)
+	return val.([]byte)
+}
+
 // MaxMessageSize returns the maximum accepted message size.
 func (w *Whisper) MaxMessageSize() uint32 {
 	val, _ := w.settings.Load(maxMsgSizeIdx)
@@ -180,6 +186,23 @@ func (w *Whisper) SetMaxMessageSize(size uint32) error {
 	return nil
 }
 
+// SetBloomFilter sets the new bloom filter
+func (w *Whisper) SetBloomFilter(bloom []byte) error {
+	if len(bloom) != bloomFilterSize {
+		return fmt.Errorf("invalid bloom filter size: %d", len(bloom))
+	}
+
+	w.notifyPeersAboutBloomFilterChange(bloom)
+
+	go func() {
+		// allow some time before all the peers have processed the notification
+		time.Sleep(time.Duration(w.reactionAllowance) * time.Second)
+		w.settings.Store(bloomFilterIdx, bloom)
+	}()
+
+	return nil
+}
+
 // SetMinimumPoW sets the minimal PoW required by this node
 func (w *Whisper) SetMinimumPoW(val float64) error {
 	if val < 0.0 {
@@ -203,17 +226,14 @@ func (w *Whisper) SetMinimumPowTest(val float64) {
 	w.settings.Store(minPowIdx, val)
 }
 
+// SetBloomFilterTest sets the Bloom Filter in test environment
+func (w *Whisper) SetBloomFilterTest(bloom []byte) {
+	w.notifyPeersAboutBloomFilterChange(bloom)
+	w.settings.Store(minPowIdx, bloom)
+}
+
 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()
-
+	arr := w.getPeers()
 	for _, p := range arr {
 		err := p.notifyAboutPowRequirementChange(pow)
 		if err != nil {
@@ -221,11 +241,37 @@ func (w *Whisper) notifyPeersAboutPowRequirementChange(pow float64) {
 			err = p.notifyAboutPowRequirementChange(pow)
 		}
 		if err != nil {
-			log.Warn("oversized message received", "peer", p.ID(), "error", err)
+			log.Warn("failed to notify peer about new pow requirement", "peer", p.ID(), "error", err)
 		}
 	}
 }
 
+func (w *Whisper) notifyPeersAboutBloomFilterChange(bloom []byte) {
+	arr := w.getPeers()
+	for _, p := range arr {
+		err := p.notifyAboutBloomFilterChange(bloom)
+		if err != nil {
+			// allow one retry
+			err = p.notifyAboutBloomFilterChange(bloom)
+		}
+		if err != nil {
+			log.Warn("failed to notify peer about new bloom filter", "peer", p.ID(), "error", err)
+		}
+	}
+}
+
+func (w *Whisper) getPeers() []*Peer {
+	arr := make([]*Peer, len(w.peers))
+	i := 0
+	w.peerMu.Lock()
+	for p := range w.peers {
+		arr[i] = p
+		i++
+	}
+	w.peerMu.Unlock()
+	return arr
+}
+
 // getPeer retrieves peer by ID
 func (w *Whisper) getPeer(peerID []byte) (*Peer, error) {
 	w.peerMu.Lock()
@@ -592,7 +638,21 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error {
 			}
 			p.powRequirement = f
 		case bloomFilterExCode:
-			// to be implemented
+			var bloom []byte
+			err := packet.Decode(&bloom)
+			if err == nil && len(bloom) != bloomFilterSize {
+				err = fmt.Errorf("wrong bloom filter size %d", len(bloom))
+			}
+
+			if err != nil {
+				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) {
+				p.bloomFilter = nil
+			} else {
+				p.bloomFilter = bloom
+			}
 		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
@@ -659,7 +719,11 @@ func (wh *Whisper) add(envelope *Envelope) (bool, 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())
+		//return false, fmt.Errorf("envelope with low PoW received: PoW=%f, hash=[%v]", envelope.PoW(), envelope.Hash().Hex())
+	}
+
+	if !bloomFilterMatch(wh.BloomFilter(), envelope.Bloom()) {
+		return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v]", envelope.Hash().Hex())
 	}
 
 	hash := envelope.Hash()
@@ -897,3 +961,24 @@ func GenerateRandomID() (id string, err error) {
 	id = common.Bytes2Hex(buf)
 	return id, err
 }
+
+func isFulNode(bloom []byte) bool {
+	for _, b := range bloom {
+		if b != 255 {
+			return false
+		}
+	}
+	return true
+}
+
+func bloomFilterMatch(filter, sample []byte) bool {
+	for i := 0; i < bloomFilterSize; i++ {
+		f := filter[i]
+		s := sample[i]
+		if ((f | s) ^ f) != 0 {
+			return false
+		}
+	}
+
+	return true
+}