Merge 787054a378 into 7a59a9380e

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)
@@ -69,7 +69,7 @@ const (
 	transmissionCycle = 300 * time.Millisecond
 
 	DefaultTTL           = 50 // seconds
-	SynchAllowance = 10 // seconds
+	DefaultSyncAllowance = 10 // seconds
 
 	EnvelopeHeaderLength = 20
 )

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 // may contain nil in case of full node
 
 	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

@@ -48,9 +48,12 @@ type Statistics struct {
 }
 
 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
+	minPowIdx               = iota // Minimal PoW required by the whisper node
+	minPowToleranceIdx      = iota // Minimal PoW tolerated by the whisper node for a limited time
+	bloomFilterIdx          = iota // Bloom filter for topics of interest for this node
+	bloomFilterToleranceIdx = iota // Bloom filter tolerated by the whisper node for a limited time
 )
 
 // Whisper represents a dark communication interface through the Ethereum
@@ -76,7 +79,7 @@ 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
+	syncAllowance int // maximum time in seconds allowed to process the whisper-related messages
 
 	statsMu sync.Mutex // guard stats
 	stats   Statistics // Statistics of whisper node
@@ -99,7 +102,7 @@ func New(cfg *Config) *Whisper {
 		messageQueue:  make(chan *Envelope, messageQueueLimit),
 		p2pMsgQueue:   make(chan *Envelope, messageQueueLimit),
 		quit:          make(chan struct{}),
-		reactionAllowance: SynchAllowance,
+		syncAllowance: DefaultSyncAllowance,
 	}
 
 	whisper.filters = NewFilters(whisper)
@@ -128,9 +131,36 @@ func New(cfg *Config) *Whisper {
 
 func (w *Whisper) MinPow() float64 {
 	val, _ := w.settings.Load(minPowIdx)
+	if val == nil {
+		return DefaultMinimumPoW
+	}
 	return val.(float64)
 }
 
+func (w *Whisper) MinPowTolerance() float64 {
+	val, _ := w.settings.Load(minPowToleranceIdx)
+	if val == nil {
+		return DefaultMinimumPoW
+	}
+	return val.(float64)
+}
+
+func (w *Whisper) BloomFilter() []byte {
+	val, _ := w.settings.Load(bloomFilterIdx)
+	if val == nil {
+		return nil
+	}
+	return val.([]byte)
+}
+
+func (w *Whisper) BloomFilterTolerance() []byte {
+	val, _ := w.settings.Load(bloomFilterToleranceIdx)
+	if val == nil {
+		return nil
+	}
+	return val.([]byte)
+}
+
 // MaxMessageSize returns the maximum accepted message size.
 func (w *Whisper) MaxMessageSize() uint32 {
 	val, _ := w.settings.Load(maxMsgSizeIdx)
@@ -180,18 +210,37 @@ 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.settings.Store(bloomFilterIdx, bloom)
+	w.notifyPeersAboutBloomFilterChange(bloom)
+
+	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)
+	}()
+
+	return nil
+}
+
 // SetMinimumPoW sets the minimal PoW required by this node
 func (w *Whisper) SetMinimumPoW(val float64) error {
 	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)
+		time.Sleep(time.Duration(w.syncAllowance) * time.Second)
-		w.settings.Store(minPowIdx, val)
+		w.settings.Store(minPowToleranceIdx, val)
 	}()
 
 	return nil
@@ -199,21 +248,20 @@ func (w *Whisper) SetMinimumPoW(val float64) error {
 
 // SetMinimumPoW sets the minimal PoW in test environment
 func (w *Whisper) SetMinimumPowTest(val float64) {
-	w.notifyPeersAboutPowRequirementChange(val)
 	w.settings.Store(minPowIdx, val)
+	w.notifyPeersAboutPowRequirementChange(val)
+	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 := make([]*Peer, len(w.peers))
+	arr := w.getPeers()
-	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 {
@@ -221,11 +269,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()
@@ -459,7 +533,28 @@ func (w *Whisper) GetSymKey(id string) ([]byte, error) {
 // Subscribe installs a new message handler used for filtering, decrypting
 // and subsequent storing of incoming messages.
 func (w *Whisper) Subscribe(f *Filter) (string, error) {
-	return w.filters.Install(f)
+	s, err := w.filters.Install(f)
+	if err == nil {
+		w.updateBloomFilter(f)
+	}
+	return s, err
+}
+
+// updateBloomFilter recalculates the new value of bloom filter,
+// and informs the peers if necessary.
+func (w *Whisper) updateBloomFilter(f *Filter) {
+	aggregate := make([]byte, bloomFilterSize)
+	for _, t := range f.Topics {
+		top := BytesToTopic(t)
+		b := TopicToBloom(top)
+		aggregate = addBloom(aggregate, b)
+	}
+
+	if !bloomFilterMatch(w.BloomFilter(), aggregate) {
+		// existing bloom filter must be updated
+		aggregate = addBloom(w.BloomFilter(), aggregate)
+		w.SetBloomFilter(aggregate)
+	}
 }
 
 // GetFilter returns the filter by id.
@@ -592,7 +687,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
@@ -633,7 +742,7 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 	sent := envelope.Expiry - envelope.TTL
 
 	if sent > now {
-		if sent-SynchAllowance > now {
+		if sent-DefaultSyncAllowance > now {
 			return false, fmt.Errorf("envelope created in the future [%x]", envelope.Hash())
 		} else {
 			// recalculate PoW, adjusted for the time difference, plus one second for latency
@@ -642,7 +751,7 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) {
 	}
 
 	if envelope.Expiry < now {
-		if envelope.Expiry+SynchAllowance*2 < now {
+		if envelope.Expiry+DefaultSyncAllowance*2 < now {
 			return false, fmt.Errorf("very old message")
 		} else {
 			log.Debug("expired envelope dropped", "hash", envelope.Hash().Hex())
@@ -655,11 +764,23 @@ 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.
+		if envelope.PoW() < wh.MinPowTolerance() {
 			log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", 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 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()) {
+		// 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.
+		if !bloomFilterMatch(wh.BloomFilterTolerance(), envelope.Bloom()) {
+			return false, fmt.Errorf("envelope does not match bloom filter, hash=[%v]", envelope.Hash().Hex())
+		}
 	}
 
 	hash := envelope.Hash()
@@ -897,3 +1018,40 @@ func GenerateRandomID() (id string, err error) {
 	id = common.Bytes2Hex(buf)
 	return id, err
 }
+
+func isFulNode(bloom []byte) bool {
+	if bloom == nil {
+		return true
+	}
+	for _, b := range bloom {
+		if b != 255 {
+			return false
+		}
+	}
+	return true
+}
+
+func bloomFilterMatch(filter, sample []byte) bool {
+	if filter == nil {
+		// full node, accepts all messages
+		return true
+	}
+
+	for i := 0; i < bloomFilterSize; i++ {
+		f := filter[i]
+		s := sample[i]
+		if ((f | s) ^ f) != 0 {
+			return false
+		}
+	}
+
+	return true
+}
+
+func addBloom(a, b []byte) []byte {
+	c := make([]byte, bloomFilterSize)
+	for i := 0; i < bloomFilterSize; i++ {
+		c[i] = a[i] | b[i]
+	}
+	return c
+}