From 347fcb31fafe77d9d7a355440ca3d982e15937af Mon Sep 17 00:00:00 2001 From: Vlad Date: Fri, 22 Dec 2017 19:11:36 +0200 Subject: [PATCH] whisper: bloom filter logic introduced --- whisper/whisperv6/api.go | 25 ++++++-- whisper/whisperv6/doc.go | 2 +- whisper/whisperv6/envelope.go | 35 +++++++++- whisper/whisperv6/peer.go | 16 ++++- whisper/whisperv6/whisper.go | 117 +++++++++++++++++++++++++++++----- 5 files changed, 170 insertions(+), 25 deletions(-) diff --git a/whisper/whisperv6/api.go b/whisper/whisperv6/api.go index 3dddb69539..f11e8b6dca 100644 --- a/whisper/whisperv6/api.go +++ b/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) diff --git a/whisper/whisperv6/doc.go b/whisper/whisperv6/doc.go index 2a4911d65a..7d3bdbdcad 100644 --- a/whisper/whisperv6/doc.go +++ b/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) diff --git a/whisper/whisperv6/envelope.go b/whisper/whisperv6/envelope.go index 676df669bf..7f2532c2c9 100644 --- a/whisper/whisperv6/envelope.go +++ b/whisper/whisperv6/envelope.go @@ -42,9 +42,11 @@ type Envelope struct { Data []byte Nonce uint64 - pow float64 // Message-specific PoW as described in the Whisper specification. - hash common.Hash // Cached hash of the envelope to avoid rehashing every time. - // Don't access hash directly, use Hash() function instead. + 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. + 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 +} diff --git a/whisper/whisperv6/peer.go b/whisper/whisperv6/peer.go index 65e0c77b0c..7893e4a9ad 100644 --- a/whisper/whisperv6/peer.go +++ 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()) +} diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 492591486b..99f0167274 100644 --- a/whisper/whisperv6/whisper.go +++ b/whisper/whisperv6/whisper.go @@ -48,9 +48,10 @@ 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 + 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 +}