From ae4bfc3cfb3f1debad9dd0211950ce09038ffa90 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Tue, 21 Apr 2015 18:31:08 +0300 Subject: [PATCH] rpc, ui/qt/qwhisper, whisper, xeth: introduce complex topic filters --- rpc/api.go | 8 +- rpc/args.go | 67 +++++++++++++---- rpc/args_test.go | 2 +- ui/qt/qwhisper/whisper.go | 2 +- whisper/filter.go | 53 +++++++++++-- whisper/topic.go | 121 +++++++++++++++++++++++++++--- whisper/topic_test.go | 151 +++++++++++++++++++++++++++++++++++--- whisper/whisper.go | 22 +++--- whisper/whisper_test.go | 2 +- xeth/whisper.go | 4 +- xeth/xeth.go | 2 +- 11 files changed, 374 insertions(+), 60 deletions(-) diff --git a/rpc/api.go b/rpc/api.go index a73188f07..4da2fb17a 100644 --- a/rpc/api.go +++ b/rpc/api.go @@ -422,12 +422,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err case "shh_newIdentity": *reply = api.xeth().Whisper().NewIdentity() - // case "shh_removeIdentity": - // args := new(WhisperIdentityArgs) - // if err := json.Unmarshal(req.Params, &args); err != nil { - // return err - // } - // *reply = api.xeth().Whisper().RemoveIdentity(args.Identity) + case "shh_hasIdentity": args := new(WhisperIdentityArgs) if err := json.Unmarshal(req.Params, &args); err != nil { @@ -439,6 +434,7 @@ func (api *EthereumApi) GetRequestReply(req *RpcRequest, reply *interface{}) err return NewNotImplementedError(req.Method) case "shh_newFilter": + // Create a new filter to watch and match messages with args := new(WhisperFilterArgs) if err := json.Unmarshal(req.Params, &args); err != nil { return err diff --git a/rpc/args.go b/rpc/args.go index 7694a3d3f..8df483f08 100644 --- a/rpc/args.go +++ b/rpc/args.go @@ -1010,25 +1010,27 @@ func (args *WhisperIdentityArgs) UnmarshalJSON(b []byte) (err error) { } type WhisperFilterArgs struct { - To string `json:"to"` + To string From string - Topics []string + Topics [][]string } +// UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a +// JSON message blob into a WhisperFilterArgs structure. func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) { + // Unmarshal the JSON message and sanity check var obj []struct { - To interface{} - Topics []interface{} + To interface{} `json:"to"` + From interface{} `json:"from"` + Topics interface{} `json:"topics"` } - - if err = json.Unmarshal(b, &obj); err != nil { + if err := json.Unmarshal(b, &obj); err != nil { return NewDecodeParamError(err.Error()) } - if len(obj) < 1 { return NewInsufficientParamsError(len(obj), 1) } - + // Retrieve the simple data contents of the filter arguments if obj[0].To == nil { args.To = "" } else { @@ -1038,17 +1040,52 @@ func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) { } args.To = argstr } - - t := make([]string, len(obj[0].Topics)) - for i, j := range obj[0].Topics { - argstr, ok := j.(string) + if obj[0].From == nil { + args.From = "" + } else { + argstr, ok := obj[0].From.(string) if !ok { - return NewInvalidTypeError("topics["+string(i)+"]", "is not a string") + return NewInvalidTypeError("from", "is not a string") } - t[i] = argstr + args.From = argstr } - args.Topics = t + // Construct the nested topic array + if obj[0].Topics != nil { + // Make sure we have an actual topic array + list, ok := obj[0].Topics.([]interface{}) + if !ok { + return NewInvalidTypeError("topics", "is not an array") + } + // Iterate over each topic and handle nil, string or array + topics := make([][]string, len(list)) + for idx, field := range list { + switch value := field.(type) { + case nil: + topics[idx] = []string{""} + case string: + topics[idx] = []string{value} + + case []interface{}: + topics[idx] = make([]string, len(value)) + for i, nested := range value { + switch value := nested.(type) { + case nil: + topics[idx][i] = "" + + case string: + topics[idx][i] = value + + default: + return NewInvalidTypeError(fmt.Sprintf("topic[%d][%d]", idx, i), "is not a string") + } + } + default: + return NewInvalidTypeError(fmt.Sprintf("topic[%d]", idx), "not a string or array") + } + } + args.Topics = topics + } return nil } diff --git a/rpc/args_test.go b/rpc/args_test.go index 2f011bfd9..f5949b7a2 100644 --- a/rpc/args_test.go +++ b/rpc/args_test.go @@ -1943,7 +1943,7 @@ func TestWhisperFilterArgs(t *testing.T) { input := `[{"topics": ["0x68656c6c6f20776f726c64"], "to": "0x34ag445g3455b34"}]` expected := new(WhisperFilterArgs) expected.To = "0x34ag445g3455b34" - expected.Topics = []string{"0x68656c6c6f20776f726c64"} + expected.Topics = [][]string{[]string{"0x68656c6c6f20776f726c64"}} args := new(WhisperFilterArgs) if err := json.Unmarshal([]byte(input), &args); err != nil { diff --git a/ui/qt/qwhisper/whisper.go b/ui/qt/qwhisper/whisper.go index 50b0626f5..b7409c57f 100644 --- a/ui/qt/qwhisper/whisper.go +++ b/ui/qt/qwhisper/whisper.go @@ -106,7 +106,7 @@ func filterFromMap(opts map[string]interface{}) (f whisper.Filter) { if topicList, ok := opts["topics"].(*qml.List); ok { var topics []string topicList.Convert(&topics) - f.Topics = whisper.NewTopicsFromStrings(topics...) + f.Topics = whisper.NewTopicFilterFromStringsFlat(topics...) } return diff --git a/whisper/filter.go b/whisper/filter.go index 8fcc45afd..8a398ab76 100644 --- a/whisper/filter.go +++ b/whisper/filter.go @@ -2,12 +2,55 @@ package whisper -import "crypto/ecdsa" +import ( + "crypto/ecdsa" + + "github.com/ethereum/go-ethereum/event/filter" +) // Filter is used to subscribe to specific types of whisper messages. type Filter struct { - To *ecdsa.PublicKey // Recipient of the message - From *ecdsa.PublicKey // Sender of the message - Topics []Topic // Topics to watch messages on - Fn func(*Message) // Handler in case of a match + To *ecdsa.PublicKey // Recipient of the message + From *ecdsa.PublicKey // Sender of the message + Topics [][]Topic // Topics to filter messages with + Fn func(msg *Message) // Handler in case of a match +} + +// filterer is the internal, fully initialized filter ready to match inbound +// messages to a variety of criteria. +type filterer struct { + to string // Recipient of the message + from string // Sender of the message + matcher *topicMatcher // Topics to filter messages with + fn func(data interface{}) // Handler in case of a match +} + +// Compare checks if the specified filter matches the current one. +func (self filterer) Compare(f filter.Filter) bool { + filter := f.(filterer) + + // Check the message sender and recipient + if len(self.to) > 0 && self.to != filter.to { + return false + } + if len(self.from) > 0 && self.from != filter.from { + return false + } + // Check the topic filtering + topics := make([]Topic, len(filter.matcher.conditions)) + for i, group := range filter.matcher.conditions { + // Message should contain a single topic entry, extract + for topics[i], _ = range group { + break + } + } + if !self.matcher.Matches(topics) { + return false + } + return true +} + +// Trigger is called when a filter successfully matches an inbound message. +func (self filterer) Trigger(data interface{}) { + self.fn(data) } diff --git a/whisper/topic.go b/whisper/topic.go index a965c7cc2..b2a264e29 100644 --- a/whisper/topic.go +++ b/whisper/topic.go @@ -27,6 +27,26 @@ func NewTopics(data ...[]byte) []Topic { return topics } +// NewTopicFilter creates a 2D topic array used by whisper.Filter from binary +// data elements. +func NewTopicFilter(data ...[][]byte) [][]Topic { + filter := make([][]Topic, len(data)) + for i, condition := range data { + filter[i] = NewTopics(condition...) + } + return filter +} + +// NewTopicFilterFlat creates a 2D topic array used by whisper.Filter from flat +// binary data elements. +func NewTopicFilterFlat(data ...[]byte) [][]Topic { + filter := make([][]Topic, len(data)) + for i, element := range data { + filter[i] = []Topic{NewTopic(element)} + } + return filter +} + // NewTopicFromString creates a topic using the binary data contents of the // specified string. func NewTopicFromString(data string) Topic { @@ -43,19 +63,100 @@ func NewTopicsFromStrings(data ...string) []Topic { return topics } +// NewTopicFilterFromStrings creates a 2D topic array used by whisper.Filter +// from textual data elements. +func NewTopicFilterFromStrings(data ...[]string) [][]Topic { + filter := make([][]Topic, len(data)) + for i, condition := range data { + filter[i] = NewTopicsFromStrings(condition...) + } + return filter +} + +// NewTopicFilterFromStringsFlat creates a 2D topic array used by whisper.Filter from flat +// binary data elements. +func NewTopicFilterFromStringsFlat(data ...string) [][]Topic { + filter := make([][]Topic, len(data)) + for i, element := range data { + filter[i] = []Topic{NewTopicFromString(element)} + } + return filter +} + // String converts a topic byte array to a string representation. func (self *Topic) String() string { return string(self[:]) } -// TopicSet represents a hash set to check if a topic exists or not. -type topicSet map[string]struct{} - -// NewTopicSet creates a topic hash set from a slice of topics. -func newTopicSet(topics []Topic) topicSet { - set := make(map[string]struct{}) - for _, topic := range topics { - set[topic.String()] = struct{}{} - } - return topicSet(set) +// topicMatcher is a filter expression to verify if a list of topics contained +// in an arriving message matches some topic conditions. The topic matcher is +// built up of a list of conditions, each of which must be satisfied by the +// corresponding topic in the message. Each condition may require: a) an exact +// topic match; b) a match from a set of topics; or c) a wild-card matching all. +// +// If a message contains more topics than required by the matcher, those beyond +// the condition count are ignored and assumed to match. +// +// Consider the following sample topic matcher: +// sample := { +// {TopicA1, TopicA2, TopicA3}, +// {TopicB}, +// nil, +// {TopicD1, TopicD2} +// } +// In order for a message to pass this filter, it should enumerate at least 4 +// topics, the first any of [TopicA1, TopicA2, TopicA3], the second mandatory +// "TopicB", the third is ignored by the filter and the fourth either "TopicD1" +// or "TopicD2". If the message contains further topics, the filter will match +// them too. +type topicMatcher struct { + conditions []map[Topic]struct{} +} + +// newTopicMatcher create a topic matcher from a list of topic conditions. +func newTopicMatcher(topics ...[]Topic) *topicMatcher { + matcher := make([]map[Topic]struct{}, len(topics)) + for i, condition := range topics { + matcher[i] = make(map[Topic]struct{}) + for _, topic := range condition { + matcher[i][topic] = struct{}{} + } + } + return &topicMatcher{conditions: matcher} +} + +// newTopicMatcherFromBinary create a topic matcher from a list of binary conditions. +func newTopicMatcherFromBinary(data ...[][]byte) *topicMatcher { + topics := make([][]Topic, len(data)) + for i, condition := range data { + topics[i] = NewTopics(condition...) + } + return newTopicMatcher(topics...) +} + +// newTopicMatcherFromStrings creates a topic matcher from a list of textual +// conditions. +func newTopicMatcherFromStrings(data ...[]string) *topicMatcher { + topics := make([][]Topic, len(data)) + for i, condition := range data { + topics[i] = NewTopicsFromStrings(condition...) + } + return newTopicMatcher(topics...) +} + +// Matches checks if a list of topics matches this particular condition set. +func (self *topicMatcher) Matches(topics []Topic) bool { + // Mismatch if there aren't enough topics + if len(self.conditions) > len(topics) { + return false + } + // Check each topic condition for existence (skip wild-cards) + for i := 0; i < len(topics) && i < len(self.conditions); i++ { + if len(self.conditions[i]) > 0 { + if _, ok := self.conditions[i][topics[i]]; !ok { + return false + } + } + } + return true } diff --git a/whisper/topic_test.go b/whisper/topic_test.go index 4015079dc..22ee06096 100644 --- a/whisper/topic_test.go +++ b/whisper/topic_test.go @@ -52,16 +52,149 @@ func TestTopicCreation(t *testing.T) { } } -func TestTopicSetCreation(t *testing.T) { - topics := make([]Topic, len(topicCreationTests)) - for i, tt := range topicCreationTests { - topics[i] = NewTopic(tt.data) +var topicMatcherCreationTest = struct { + binary [][][]byte + textual [][]string + matcher []map[[4]byte]struct{} +}{ + binary: [][][]byte{ + [][]byte{}, + [][]byte{ + []byte("Topic A"), + }, + [][]byte{ + []byte("Topic B1"), + []byte("Topic B2"), + []byte("Topic B3"), + }, + }, + textual: [][]string{ + []string{}, + []string{"Topic A"}, + []string{"Topic B1", "Topic B2", "Topic B3"}, + }, + matcher: []map[[4]byte]struct{}{ + map[[4]byte]struct{}{}, + map[[4]byte]struct{}{ + [4]byte{0x25, 0xfc, 0x95, 0x66}: struct{}{}, + }, + map[[4]byte]struct{}{ + [4]byte{0x93, 0x6d, 0xec, 0x09}: struct{}{}, + [4]byte{0x25, 0x23, 0x34, 0xd3}: struct{}{}, + [4]byte{0x6b, 0xc2, 0x73, 0xd1}: struct{}{}, + }, + }, +} + +func TestTopicMatcherCreation(t *testing.T) { + test := topicMatcherCreationTest + + matcher := newTopicMatcherFromBinary(test.binary...) + for i, cond := range matcher.conditions { + for topic, _ := range cond { + if _, ok := test.matcher[i][topic]; !ok { + t.Errorf("condition %d; extra topic found: 0x%x", i, topic[:]) + } + } } - set := newTopicSet(topics) - for i, tt := range topicCreationTests { - topic := NewTopic(tt.data) - if _, ok := set[topic.String()]; !ok { - t.Errorf("topic %d: not found in set", i) + for i, cond := range test.matcher { + for topic, _ := range cond { + if _, ok := matcher.conditions[i][topic]; !ok { + t.Errorf("condition %d; topic not found: 0x%x", i, topic[:]) + } + } + } + + matcher = newTopicMatcherFromStrings(test.textual...) + for i, cond := range matcher.conditions { + for topic, _ := range cond { + if _, ok := test.matcher[i][topic]; !ok { + t.Errorf("condition %d; extra topic found: 0x%x", i, topic[:]) + } + } + } + for i, cond := range test.matcher { + for topic, _ := range cond { + if _, ok := matcher.conditions[i][topic]; !ok { + t.Errorf("condition %d; topic not found: 0x%x", i, topic[:]) + } + } + } +} + +var topicMatcherTests = []struct { + filter [][]string + topics []string + match bool +}{ + // Empty topic matcher should match everything + { + filter: [][]string{}, + topics: []string{}, + match: true, + }, + { + filter: [][]string{}, + topics: []string{"a", "b", "c"}, + match: true, + }, + // Fixed topic matcher should match strictly, but only prefix + { + filter: [][]string{[]string{"a"}, []string{"b"}}, + topics: []string{"a"}, + match: false, + }, + { + filter: [][]string{[]string{"a"}, []string{"b"}}, + topics: []string{"a", "b"}, + match: true, + }, + { + filter: [][]string{[]string{"a"}, []string{"b"}}, + topics: []string{"a", "b", "c"}, + match: true, + }, + // Multi-matcher should match any from a sub-group + { + filter: [][]string{[]string{"a1", "a2"}}, + topics: []string{"a"}, + match: false, + }, + { + filter: [][]string{[]string{"a1", "a2"}}, + topics: []string{"a1"}, + match: true, + }, + { + filter: [][]string{[]string{"a1", "a2"}}, + topics: []string{"a2"}, + match: true, + }, + // Wild-card condition should match anything + { + filter: [][]string{[]string{}, []string{"b"}}, + topics: []string{"a"}, + match: false, + }, + { + filter: [][]string{[]string{}, []string{"b"}}, + topics: []string{"a", "b"}, + match: true, + }, + { + filter: [][]string{[]string{}, []string{"b"}}, + topics: []string{"b", "b"}, + match: true, + }, +} + +func TestTopicMatcher(t *testing.T) { + for i, tt := range topicMatcherTests { + topics := NewTopicsFromStrings(tt.topics...) + + matcher := newTopicMatcherFromStrings(tt.filter...) + if match := matcher.Matches(topics); match != tt.match { + t.Errorf("test %d: match mismatch: have %v, want %v", i, match, tt.match) } } } diff --git a/whisper/whisper.go b/whisper/whisper.go index 61999f07a..5d6ee6e3b 100644 --- a/whisper/whisper.go +++ b/whisper/whisper.go @@ -118,11 +118,11 @@ func (self *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey { // Watch installs a new message handler to run in case a matching packet arrives // from the whisper network. func (self *Whisper) Watch(options Filter) int { - filter := filter.Generic{ - Str1: string(crypto.FromECDSAPub(options.To)), - Str2: string(crypto.FromECDSAPub(options.From)), - Data: newTopicSet(options.Topics), - Fn: func(data interface{}) { + filter := filterer{ + to: string(crypto.FromECDSAPub(options.To)), + from: string(crypto.FromECDSAPub(options.From)), + matcher: newTopicMatcher(options.Topics...), + fn: func(data interface{}) { options.Fn(data.(*Message)) }, } @@ -273,10 +273,14 @@ func (self *Whisper) open(envelope *Envelope) *Message { // createFilter creates a message filter to check against installed handlers. func createFilter(message *Message, topics []Topic) filter.Filter { - return filter.Generic{ - Str1: string(crypto.FromECDSAPub(message.To)), - Str2: string(crypto.FromECDSAPub(message.Recover())), - Data: newTopicSet(topics), + matcher := make([][]Topic, len(topics)) + for i, topic := range topics { + matcher[i] = []Topic{topic} + } + return filterer{ + to: string(crypto.FromECDSAPub(message.To)), + from: string(crypto.FromECDSAPub(message.Recover())), + matcher: newTopicMatcher(matcher...), } } diff --git a/whisper/whisper_test.go b/whisper/whisper_test.go index def8e68d8..8fce0e036 100644 --- a/whisper/whisper_test.go +++ b/whisper/whisper_test.go @@ -129,7 +129,7 @@ func testBroadcast(anonymous bool, t *testing.T) { dones[i] = done targets[i].Watch(Filter{ - Topics: NewTopicsFromStrings("broadcast topic"), + Topics: NewTopicFilterFromStrings([]string{"broadcast topic"}), Fn: func(msg *Message) { close(done) }, diff --git a/xeth/whisper.go b/xeth/whisper.go index 386897f39..25c4af3b1 100644 --- a/xeth/whisper.go +++ b/xeth/whisper.go @@ -67,11 +67,11 @@ func (self *Whisper) Post(payload string, to, from string, topics []string, prio // Watch installs a new message handler to run in case a matching packet arrives // from the whisper network. -func (self *Whisper) Watch(to, from string, topics []string, fn func(WhisperMessage)) int { +func (self *Whisper) Watch(to, from string, topics [][]string, fn func(WhisperMessage)) int { filter := whisper.Filter{ To: crypto.ToECDSAPub(common.FromHex(to)), From: crypto.ToECDSAPub(common.FromHex(from)), - Topics: whisper.NewTopicsFromStrings(topics...), + Topics: whisper.NewTopicFilterFromStrings(topics...), } filter.Fn = func(message *whisper.Message) { fn(NewWhisperMessage(message)) diff --git a/xeth/xeth.go b/xeth/xeth.go index 8cc32c958..ea6ae9950 100644 --- a/xeth/xeth.go +++ b/xeth/xeth.go @@ -452,7 +452,7 @@ func (self *XEth) AllLogs(earliest, latest int64, skip, max int, address []strin return filter.Find() } -func (p *XEth) NewWhisperFilter(to, from string, topics []string) int { +func (p *XEth) NewWhisperFilter(to, from string, topics [][]string) int { var id int callback := func(msg WhisperMessage) { p.messagesMut.Lock()