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..062f3ba62c 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) @@ -68,8 +68,8 @@ const ( expirationCycle = time.Second transmissionCycle = 300 * time.Millisecond - DefaultTTL = 50 // seconds - SynchAllowance = 10 // seconds + DefaultTTL = 50 // seconds + DefaultSyncAllowance = 10 // seconds EnvelopeHeaderLength = 20 ) 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..b21f07e031 100644 --- a/whisper/whisperv6/peer.go +++ 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()) +} diff --git a/whisper/whisperv6/whisper.go b/whisper/whisperv6/whisper.go index 492591486b..9ffd7faa24 100644 --- a/whisper/whisperv6/whisper.go +++ b/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 + 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 @@ -91,15 +94,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{}), - reactionAllowance: SynchAllowance, + 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{}), + 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) - w.settings.Store(minPowIdx, val) + time.Sleep(time.Duration(w.syncAllowance) * time.Second) + 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)) - 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 +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() { - log.Debug("envelope with low PoW dropped", "PoW", envelope.PoW(), "hash", envelope.Hash().Hex()) - return false, nil // drop envelope without error for now + // 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 +}