From c58ecf1a0bf416a4ef7881af8b8d6925d6827034 Mon Sep 17 00:00:00 2001 From: b00ris Date: Wed, 21 Feb 2018 23:10:00 +0300 Subject: [PATCH] whisper: add matching to v6 --- whisper/whisperv5/filter.go | 16 ++- whisper/whisperv5/filter_test.go | 1 + whisper/whisperv6/filter.go | 153 +++++++++++++++------ whisper/whisperv6/filter_test.go | 228 +++++++++++++++++++------------ 4 files changed, 256 insertions(+), 142 deletions(-) diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index b38ef3bce1..d8c558db26 100644 --- a/whisper/whisperv5/filter.go +++ b/whisper/whisperv5/filter.go @@ -26,7 +26,10 @@ import ( "github.com/ethereum/go-ethereum/log" ) -const ALL_TOPICS = "" +const ( + ALL_TOPICS = "" + MAX_POOL_CAPACITY = 1000 +) type Filter struct { Src *ecdsa.PublicKey // Sender of the message @@ -43,18 +46,16 @@ type Filter struct { type Filters struct { watchers map[string]*Filter - watchersTopics map[string]map[string]struct{} - topicMatcher *topicMatcher whisper *Whisper mutex sync.RWMutex + topicMatcher *topicMatcher } func NewFilters(w *Whisper) *Filters { fs := &Filters{ watchers: make(map[string]*Filter), - watchersTopics: make(map[string]map[string]struct{}), - topicMatcher: newTopicMatcher(), whisper: w, + topicMatcher: newTopicMatcher(), } return fs } @@ -228,6 +229,9 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool { } type topicMatcher struct { + //structure - map[topic]map[filterID] + //mapping for topics + //"" topic means that the filter allows all topic values mapper map[string]map[string]struct{} mx sync.RWMutex pool sync.Pool @@ -247,7 +251,7 @@ func (fs *topicMatcher) take() []string { return fs.pool.Get().([]string) } func (fs *topicMatcher) resolve(s []string) { - if cap(s) > 1000 { + if cap(s) > MAX_POOL_CAPACITY { return } fs.pool.Put(s[:0]) diff --git a/whisper/whisperv5/filter_test.go b/whisper/whisperv5/filter_test.go index 3e2a4f54df..475884bde3 100644 --- a/whisper/whisperv5/filter_test.go +++ b/whisper/whisperv5/filter_test.go @@ -723,6 +723,7 @@ func TestVariableTopics(t *testing.T) { if err != nil { t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) } + fs := generateFilters() filterID, err := fs.Install(f) diff --git a/whisper/whisperv6/filter.go b/whisper/whisperv6/filter.go index eb0c65fa3b..aca2a297c7 100644 --- a/whisper/whisperv6/filter.go +++ b/whisper/whisperv6/filter.go @@ -26,6 +26,11 @@ import ( "github.com/ethereum/go-ethereum/log" ) +const ( + ALL_TOPICS = "" + MAX_POOL_CAPACITY = 1000 +) + // Filter represents a Whisper message filter type Filter struct { Src *ecdsa.PublicKey // Sender of the message @@ -42,16 +47,18 @@ type Filter struct { // Filters represents a collection of filters type Filters struct { - watchers map[string]*Filter - whisper *Whisper - mutex sync.RWMutex + watchers map[string]*Filter + whisper *Whisper + mutex sync.RWMutex + topicMatcher *topicMatcher } // NewFilters returns a newly created filter collection func NewFilters(w *Whisper) *Filters { return &Filters{ - watchers: make(map[string]*Filter), - whisper: w, + watchers: make(map[string]*Filter), + whisper: w, + topicMatcher: newTopicMatcher(), } } @@ -82,6 +89,7 @@ func (fs *Filters) Install(watcher *Filter) (string, error) { } fs.watchers[id] = watcher + fs.topicMatcher.addFilterToTopicsMapping(watcher, id) return id, err } @@ -92,6 +100,7 @@ func (fs *Filters) Uninstall(id string) bool { defer fs.mutex.Unlock() if fs.watchers[id] != nil { delete(fs.watchers, id) + fs.topicMatcher.removeTopicFromTopicMapping(id) return true } return false @@ -108,15 +117,22 @@ func (fs *Filters) Get(id string) *Filter { // for the envelope's topic. func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { var msg *ReceivedMessage + matchedTopics := fs.topicMatcher.take() + defer fs.topicMatcher.resolve(matchedTopics) fs.mutex.RLock() defer fs.mutex.RUnlock() - i := -1 // only used for logging info - for _, watcher := range fs.watchers { - i++ + fs.topicMatcher.matchedTopics(env.Topic, &matchedTopics) + for _, watcherID := range matchedTopics { + watcher, ok := fs.watchers[watcherID] + if !ok { + log.Trace(fmt.Sprintf("msg [%x], filter [%s]: filter not exists", env.Hash(), watcherID)) + continue + } + if p2pMessage && !watcher.AllowP2P { - log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), i)) + log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), watcherID)) continue } @@ -128,10 +144,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { if match { msg = env.Open(watcher) if msg == nil { - log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", i) + log.Trace("processing message: failed to open", "message", env.Hash().Hex(), "filter", watcherID) } } else { - log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", i) + log.Trace("processing message: does not match", "message", env.Hash().Hex(), "filter", watcherID) } } @@ -201,9 +217,9 @@ func (f *Filter) MatchMessage(msg *ReceivedMessage) bool { } if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() { - return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) && f.MatchTopic(msg.Topic) + return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst) } else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() { - return f.SymKeyHash == msg.SymKeyHash && f.MatchTopic(msg.Topic) + return f.SymKeyHash == msg.SymKeyHash } return false } @@ -216,38 +232,6 @@ func (f *Filter) MatchEnvelope(envelope *Envelope) bool { return false } - return f.MatchTopic(envelope.Topic) -} - -// MatchTopic checks that the filter captures a given topic. -func (f *Filter) MatchTopic(topic TopicType) bool { - if len(f.Topics) == 0 { - // any topic matches - return true - } - - for _, bt := range f.Topics { - if matchSingleTopic(topic, bt) { - return true - } - } - return false -} - -func matchSingleTopic(topic TopicType, bt []byte) bool { - if len(bt) > TopicLength { - bt = bt[:TopicLength] - } - - if len(bt) < TopicLength { - return false - } - - for j, b := range bt { - if topic[j] != b { - return false - } - } return true } @@ -261,3 +245,82 @@ func IsPubKeyEqual(a, b *ecdsa.PublicKey) bool { // the curve is always the same, just compare the points return a.X.Cmp(b.X) == 0 && a.Y.Cmp(b.Y) == 0 } + +type topicMatcher struct { + //structure - map[topic]map[filterID] + //mapping for topics + //"" topic means that the filter allows all topic values + mapper map[string]map[string]struct{} + mx sync.RWMutex + pool sync.Pool +} + +func newTopicMatcher() *topicMatcher { + tm := new(topicMatcher) + tm.mapper = make(map[string]map[string]struct{}) + tm.mapper[ALL_TOPICS] = make(map[string]struct{}) + tm.pool.New = func() interface{} { + return []string{} + } + return tm +} + +func (fs *topicMatcher) take() []string { + return fs.pool.Get().([]string) +} +func (fs *topicMatcher) resolve(s []string) { + if cap(s) > MAX_POOL_CAPACITY { + return + } + fs.pool.Put(s[:0]) +} + +func (fs *topicMatcher) addFilterToTopicsMapping(watcher *Filter, id string) { + fs.mx.Lock() + defer fs.mx.Unlock() + + for i := range fs.prepareTopicsMapping(watcher) { + topicMapping, ok := fs.mapper[i] + if !ok { + fs.mapper[i] = make(map[string]struct{}) + topicMapping = fs.mapper[i] + } + topicMapping[id] = struct{}{} + } +} + +func (fs *topicMatcher) removeTopicFromTopicMapping(id string) { + fs.mx.Lock() + defer fs.mx.Unlock() + for i := range fs.mapper { + delete(fs.mapper[i], id) + } +} + +func (fs *topicMatcher) prepareTopicsMapping(watcher *Filter) map[string]struct{} { + topics := make(map[string]struct{}, len(watcher.Topics)) + + if len(watcher.Topics) == 0 { + topics[ALL_TOPICS] = struct{}{} + return topics + } + + for _, topic := range watcher.Topics { + topics[common.ToHex(topic)] = struct{}{} + } + + return topics +} + +func (fs *topicMatcher) matchedTopics(topic TopicType, matched *[]string) { + fs.mx.RLock() + defer fs.mx.RUnlock() + + for i := range fs.mapper[ALL_TOPICS] { + *matched = append(*matched, i) + } + + for i := range fs.mapper[topic.String()] { + *matched = append(*matched, i) + } +} diff --git a/whisper/whisperv6/filter_test.go b/whisper/whisperv6/filter_test.go index e7230ef388..4324470a9a 100644 --- a/whisper/whisperv6/filter_test.go +++ b/whisper/whisperv6/filter_test.go @@ -83,6 +83,14 @@ func generateFilter(t *testing.T, symmetric bool) (*Filter, error) { return &f, nil } +func generateFilters() *Filters { + fs := Filters{ + watchers: make(map[string]*Filter), + topicMatcher: newTopicMatcher(), + } + return &fs +} + func generateTestCases(t *testing.T, SizeTestFilters int) []FilterTestCase { cases := make([]FilterTestCase, SizeTestFilters) for i := 0; i < SizeTestFilters; i++ { @@ -314,19 +322,8 @@ func TestMatchEnvelope(t *testing.T) { if err != nil { t.Fatalf("failed Wrap with seed %d: %s.", seed, err) } - match := fsym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope symmetric with seed %d.", seed) - } - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope asymmetric with seed %d.", seed) - } // encrypt symmetrically - i := mrand.Int() % 4 - fsym.Topics[i] = params.Topic[:] - fasym.Topics[i] = params.Topic[:] msg, err = NewSentMessage(params) if err != nil { t.Fatalf("failed to create new message with seed %d: %s.", seed, err) @@ -336,20 +333,20 @@ func TestMatchEnvelope(t *testing.T) { t.Fatalf("failed Wrap() with seed %d: %s.", seed, err) } - // symmetric + matching topic: match - match = fsym.MatchEnvelope(env) + // symmetric + match := fsym.MatchEnvelope(env) if !match { t.Fatalf("failed MatchEnvelope() symmetric with seed %d.", seed) } - // symmetric + matching topic + insufficient PoW: mismatch + // symmetric + insufficient PoW: mismatch fsym.PoW = env.PoW() + 1.0 match = fsym.MatchEnvelope(env) if match { t.Fatalf("failed MatchEnvelope(symmetric + matching topic + insufficient PoW) asymmetric with seed %d.", seed) } - // symmetric + matching topic + sufficient PoW: match + // symmetric + sufficient PoW: match fsym.PoW = env.PoW() / 2 match = fsym.MatchEnvelope(env) if !match { @@ -387,26 +384,6 @@ func TestMatchEnvelope(t *testing.T) { t.Fatalf("failed MatchEnvelope(encryption method mismatch) with seed %d.", seed) } - // asymmetric + mismatching topic: mismatch - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(asymmetric + mismatching topic) with seed %d.", seed) - } - - // asymmetric + matching topic: match - fasym.Topics[i] = fasym.Topics[i+1] - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope(asymmetric + matching topic) with seed %d.", seed) - } - - // asymmetric + filter without topic (wildcard): match - fasym.Topics = nil - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(asymmetric + filter without topic) with seed %d.", seed) - } - // asymmetric + insufficient PoW: mismatch fasym.PoW = env.PoW() + 1.0 match = fasym.MatchEnvelope(env) @@ -420,20 +397,6 @@ func TestMatchEnvelope(t *testing.T) { if !match { t.Fatalf("failed MatchEnvelope(asymmetric + sufficient PoW) with seed %d.", seed) } - - // filter without topic + envelope without topic: match - env.Topic = TopicType{} - match = fasym.MatchEnvelope(env) - if !match { - t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed) - } - - // filter with topic + envelope without topic: mismatch - fasym.Topics = fsym.Topics - match = fasym.MatchEnvelope(env) - if match { - t.Fatalf("failed MatchEnvelope(filter without topic + envelope without topic) with seed %d.", seed) - } } func TestMatchMessageSym(t *testing.T) { @@ -485,13 +448,6 @@ func TestMatchMessageSym(t *testing.T) { t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed) } - // topic mismatch - f.Topics[index][0]++ - if f.MatchMessage(msg) { - t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed) - } - f.Topics[index][0]-- - // key mismatch f.SymKeyHash[0]++ if f.MatchMessage(msg) { @@ -578,13 +534,6 @@ func TestMatchMessageAsym(t *testing.T) { t.Fatalf("failed MatchEnvelope(sufficient PoW) with seed %d.", seed) } - // topic mismatch - f.Topics[index][0]++ - if f.MatchMessage(msg) { - t.Fatalf("failed MatchEnvelope(topic mismatch) with seed %d.", seed) - } - f.Topics[index][0]-- - // key mismatch prev := *f.KeyAsym.PublicKey.X zero := *big.NewInt(0) @@ -820,53 +769,150 @@ func TestVariableTopics(t *testing.T) { t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) } + fs := generateFilters() + + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 4; i++ { env.Topic = BytesToTopic(f.Topics[i]) + //test match + matched := []string{} + fs.topicMatcher.matchedTopics(env.Topic, &matched) match = f.MatchEnvelope(env) - if !match { + if !(match && hasFilterID(matched, filterID)) { t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) } - f.Topics[i][lastTopicByte]++ + //change envelop topic + env.Topic[lastTopicByte]++ + + //false positive test match = f.MatchEnvelope(env) - if match { + + matched = matched[:0] + fs.topicMatcher.matchedTopics(env.Topic, &matched) + if match && hasFilterID(matched, filterID) { t.Fatalf("MatchEnvelope symmetric with seed %d, step %d: false positive.", seed, i) } } } -func TestMatchSingleTopic_ReturnTrue(t *testing.T) { - bt := []byte("test") - topic := BytesToTopic(bt) +func TestTopicsMapping(t *testing.T) { + InitSingleTest() - if !matchSingleTopic(topic, bt) { - t.FailNow() + const lastTopicByte = 3 + params, err := generateMessageParams() + if err != nil { + t.Fatalf("failed generateMessageParams with seed %d: %s.", seed, err) + } + msg, err := NewSentMessage(params) + if err != nil { + t.Fatalf("failed to create new message with seed %d: %s.", seed, err) + } + env, err := msg.Wrap(params) + if err != nil { + t.Fatalf("failed Wrap with seed %d: %s.", seed, err) + } + + f, err := generateFilter(t, true) + if err != nil { + t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) + } + fs := generateFilters() + + for i := 0; i < len(f.Topics); i++ { + env.Topic = BytesToTopic(f.Topics[i]) + + //test match + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) + } + matched := []string{} + fs.topicMatcher.matchedTopics(env.Topic, &matched) + if !hasFilterID(matched, filterID) { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + + //test match without filter + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } + matched = matched[:0] + fs.topicMatcher.matchedTopics(env.Topic, &matched) + if hasFilterID(matched, filterID) { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + + //test match with changed topic + f.Topics[i][lastTopicByte]++ + filterID, err = fs.Install(f) + if err != nil { + t.Fatal(err) + } + matched = matched[:0] + fs.topicMatcher.matchedTopics(env.Topic, &matched) + if hasFilterID(matched, filterID) { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed, i) + } + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } } } -func TestMatchSingleTopic_WithTail_ReturnTrue(t *testing.T) { - bt := []byte("test with tail") - topic := BytesToTopic([]byte("test")) +func TestTopicsMapping_MatchAllTopics_Success(t *testing.T) { + InitSingleTest() - if !matchSingleTopic(topic, bt) { - t.FailNow() + f, err := generateFilter(t, true) + if err != nil { + t.Fatalf("failed generateFilter with seed %d: %s.", seed, err) + } + fs := generateFilters() + + f.Topics = [][]byte{} + + filterID, err := fs.Install(f) + if err != nil { + t.Fatal(err) + } + + //generate topic + topic := TopicType{} + mrand.Read(topic[:]) + + matched := []string{} + fs.topicMatcher.matchedTopics(topic, &matched) + if !hasFilterID(matched, filterID) { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed) + } + if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; !ok { + t.Fatal("watcher mapping incorrect") + } + + //test match without filter + if !fs.Uninstall(filterID) { + t.Fatal("Failed to uninstall filter") + } + matched = matched[:0] + fs.topicMatcher.matchedTopics(topic, &matched) + if hasFilterID(matched, filterID) { + t.Fatalf("failed MatchEnvelope symmetric with seed %d, step %d.", seed) + } + + if _, ok := fs.topicMatcher.mapper[ALL_TOPICS][filterID]; ok { + t.Fatal("watcher mapping incorrect") } } -func TestMatchSingleTopic_NotEquals_ReturnFalse(t *testing.T) { - bt := []byte("tes") - topic := BytesToTopic(bt) - - if matchSingleTopic(topic, bt) { - t.FailNow() - } -} - -func TestMatchSingleTopic_InsufficientLength_ReturnFalse(t *testing.T) { - bt := []byte("test") - topic := BytesToTopic([]byte("not_equal")) - - if matchSingleTopic(topic, bt) { - t.FailNow() +func hasFilterID(matched []string, filterID string) bool { + for i := range matched { + if matched[i] == filterID { + return true + } } + return false }