From 309b3ba093a868469411ebc230214922cc833966 Mon Sep 17 00:00:00 2001 From: Vlad Date: Tue, 6 Sep 2016 22:58:58 +0200 Subject: [PATCH] whisper5: first buildable version --- whisper5/api.go | 429 ++++++++++++++++++++++--------------------- whisper5/doc.go | 24 ++- whisper5/envelope.go | 36 +++- whisper5/filter.go | 57 ++++-- whisper5/message.go | 151 +++------------ whisper5/whisper.go | 79 +++----- 6 files changed, 364 insertions(+), 412 deletions(-) diff --git a/whisper5/api.go b/whisper5/api.go index ddd17ac392..78020d5a0d 100644 --- a/whisper5/api.go +++ b/whisper5/api.go @@ -16,31 +16,24 @@ package whisper5 -// todo: this is just a stub, delete this block ASAP -type PublicWhisperAPI struct { - w *Whisper -} - -func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { - return nil -} - -/* import ( "encoding/json" + "errors" "fmt" + "strconv" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" "github.com/ethereum/go-ethereum/rpc" ) // PublicWhisperAPI provides the whisper RPC service. type PublicWhisperAPI struct { - w *Whisper - + whisper *Whisper messagesMu sync.RWMutex messages map[int]*whisperFilter } @@ -56,114 +49,127 @@ var whisperOffLineErr = new(whisperOfflineError) // NewPublicWhisperAPI create a new RPC whisper service. func NewPublicWhisperAPI(w *Whisper) *PublicWhisperAPI { - return &PublicWhisperAPI{w: w, messages: make(map[int]*whisperFilter)} + return &PublicWhisperAPI{whisper: w, messages: make(map[int]*whisperFilter)} } // Version returns the Whisper version this node offers. -func (s *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { - if s.w == nil { +func (self *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { + if self.whisper == nil { return rpc.NewHexNumber(0), whisperOffLineErr } - return rpc.NewHexNumber(s.w.Version()), nil + return rpc.NewHexNumber(self.whisper.Version()), nil } // HasIdentity checks if the the whisper node is configured with the private key // of the specified public pair. -func (s *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { - if s.w == nil { +func (self *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { + if self.whisper == nil { return false, whisperOffLineErr } - return s.w.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil + return self.whisper.HasIdentity(crypto.ToECDSAPub(common.FromHex(identity))), nil } // NewIdentity generates a new cryptographic identity for the client, and injects // it into the known identities for message decryption. -func (s *PublicWhisperAPI) NewIdentity() (string, error) { - if s.w == nil { +func (self *PublicWhisperAPI) NewIdentity() (string, error) { + if self.whisper == nil { return "", whisperOffLineErr } - identity := s.w.NewIdentity() + identity := self.whisper.NewIdentity() return common.ToHex(crypto.FromECDSAPub(&identity.PublicKey)), nil } -type NewFilterArgs struct { - To string - From string - Topics [][][]byte -} - // NewWhisperFilter creates and registers a new message filter to watch for inbound whisper messages. -func (s *PublicWhisperAPI) NewFilter(args NewFilterArgs) (*rpc.HexNumber, error) { - if s.w == nil { +func (self *PublicWhisperAPI) NewFilter(args WhisperFilterArgs) (*rpc.HexNumber, error) { + if self.whisper == nil { return nil, whisperOffLineErr } var id int filter := Filter{ - To: crypto.ToECDSAPub(common.FromHex(args.To)), - From: crypto.ToECDSAPub(common.FromHex(args.From)), - Topics: NewFilterTopics(args.Topics...), - Fn: func(message *Message) { + Src: crypto.ToECDSAPub(common.FromHex(args.From)), + Dst: crypto.ToECDSAPub(common.FromHex(args.To)), + KeySym: common.FromHex(args.KeySym), + PoW: args.PoW, + Fn: func(message *ReceivedMessage) { wmsg := NewWhisperMessage(message) - s.messagesMu.RLock() // Only read lock to the filter pool - defer s.messagesMu.RUnlock() - if s.messages[id] != nil { - s.messages[id].insert(wmsg) + self.messagesMu.RLock() // Only read lock to the filter pool + defer self.messagesMu.RUnlock() + if self.messages[id] != nil { + self.messages[id].insert(wmsg) } + // todo: review after api is ready }, } - id = s.w.Watch(filter) + if len(args.To) == 0 && len(args.KeySym) == 0 { + info := "Filter must contain at least one key" + glog.V(logger.Error).Infof(info) + return nil, errors.New(info) + } - s.messagesMu.Lock() - s.messages[id] = newWhisperFilter(id, s.w) - s.messagesMu.Unlock() + if len(args.Topics) > 0 { + for _, s := range args.Topics { + t := common.FromHex(s) + filter.Topics = append(filter.Topics, BytesToTopic(t)) + } + } else { + // if Topics are not provided, just use the default derivation function + t := DeriveTopicFromSymmetricKey(common.FromHex(args.KeySym)) + filter.Topics = append(filter.Topics, t) + } + + id = self.whisper.Watch(&filter) + + self.messagesMu.Lock() + self.messages[id] = newWhisperFilter(id, self.whisper) + self.messagesMu.Unlock() return rpc.NewHexNumber(id), nil } -// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. -func (s *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { - s.messagesMu.RLock() - defer s.messagesMu.RUnlock() - - if s.messages[filterId.Int()] != nil { - if changes := s.messages[filterId.Int()].retrieve(); changes != nil { - return changes - } - } - return returnWhisperMessages(nil) -} - // UninstallFilter disables and removes an existing filter. -func (s *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) bool { - s.messagesMu.Lock() - defer s.messagesMu.Unlock() +func (self *PublicWhisperAPI) UninstallFilter(filterId rpc.HexNumber) bool { + self.messagesMu.Lock() + defer self.messagesMu.Unlock() - if _, ok := s.messages[filterId.Int()]; ok { - delete(s.messages, filterId.Int()) + if _, ok := self.messages[filterId.Int()]; ok { + delete(self.messages, filterId.Int()) return true } return false } -// GetMessages retrieves all the known messages that match a specific filter. -func (s *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage { - // Retrieve all the cached messages matching a specific, existing filter - s.messagesMu.RLock() - defer s.messagesMu.RUnlock() +// GetFilterChanges retrieves all the new messages matched by a filter since the last retrieval. +func (self *PublicWhisperAPI) GetFilterChanges(filterId rpc.HexNumber) []WhisperMessage { + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() - var messages []*Message - if s.messages[filterId.Int()] != nil { - messages = s.messages[filterId.Int()].messages() + if self.messages[filterId.Int()] != nil { + if changes := self.messages[filterId.Int()].retrieve(); changes != nil { + return changes + } } - - return returnWhisperMessages(messages) + return toWhisperMessages(nil) } -// returnWhisperMessages converts a Whisper message to a RPC whisper message. -func returnWhisperMessages(messages []*Message) []WhisperMessage { +// GetMessages retrieves all the known messages that match a specific filter. +func (self *PublicWhisperAPI) GetMessages(filterId rpc.HexNumber) []WhisperMessage { + // Retrieve all the cached messages matching a specific, existing filter + self.messagesMu.RLock() + defer self.messagesMu.RUnlock() + + var messages []*ReceivedMessage + if self.messages[filterId.Int()] != nil { + messages = self.messages[filterId.Int()].messages() + } + + return toWhisperMessages(messages) +} + +// toWhisperMessages converts a Whisper message to a RPC whisper message. +func toWhisperMessages(messages []*ReceivedMessage) []WhisperMessage { msgs := make([]WhisperMessage, len(messages)) for i, msg := range messages { msgs[i] = NewWhisperMessage(msg) @@ -171,96 +177,131 @@ func returnWhisperMessages(messages []*Message) []WhisperMessage { return msgs } -type PostArgs struct { - From string `json:"from"` - To string `json:"to"` - Topics [][]byte `json:"topics"` - Payload string `json:"payload"` - Priority int64 `json:"priority"` - TTL int64 `json:"ttl"` -} - // Post injects a message into the whisper network for distribution. -func (s *PublicWhisperAPI) Post(args PostArgs) (bool, error) { - if s.w == nil { +func (self *PublicWhisperAPI) Post(args PostArgs) (bool, error) { + if self.whisper == nil { return false, whisperOffLineErr } // construct whisper message with transmission options - message := NewMessage(common.FromHex(args.Payload)) + message := NewSentMessage(common.FromHex(args.Payload)) options := Options{ - To: crypto.ToECDSAPub(common.FromHex(args.To)), TTL: time.Duration(args.TTL) * time.Second, - Topics: NewTopics(args.Topics...), + Dst: crypto.ToECDSAPub(common.FromHex(args.To)), + KeySym: common.FromHex(args.Key), + Topic: BytesToTopic(common.FromHex(args.Topic)), + Pad: common.FromHex(args.Padding), } // set sender identity if len(args.From) > 0 { - if key := s.w.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); key != nil { - options.From = key + if privateKey := self.whisper.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); privateKey != nil { + options.Src = privateKey } else { return false, fmt.Errorf("unknown identity to send from: %s", args.From) } } // Wrap and send the message - pow := time.Duration(args.Priority) * time.Millisecond - envelope, err := message.Wrap(pow, options) + options.Work = time.Duration(options.Work) * time.Millisecond + envelope, err := message.Wrap(options) if err != nil { return false, err } - return true, s.w.Send(envelope) + return true, self.whisper.Send(envelope) } -// WhisperMessage is the RPC representation of a whisper message to be sent. -type WhisperMessage struct { - ref *Message - - Payload string `json:"payload"` - To string `json:"to"` - From string `json:"from"` - Sent int64 `json:"sent"` +type PostArgs struct { TTL int64 `json:"ttl"` - Hash string `json:"hash"` + From string `json:"from"` + To string `json:"to"` + Key string `json:"key"` + Topic string `json:"topic"` + Padding string `json:"padding"` + Payload string `json:"payload"` + Work int64 `json:"work"` // todo: review this field usage + PoW int64 `json:"pow"` // todo: review this field usage } func (args *PostArgs) UnmarshalJSON(data []byte) (err error) { var obj struct { - From string `json:"from"` - To string `json:"to"` - Topics []string `json:"topics"` - Payload string `json:"payload"` - Priority rpc.HexNumber `json:"priority"` - TTL rpc.HexNumber `json:"ttl"` + TTL int64 `json:"ttl"` + From string `json:"from"` + To string `json:"to"` + Key string `json:"key"` + Topic string `json:"topic"` + Payload string `json:"payload"` + Padding string `json:"padding"` + Work int64 `json:"work"` + PoW int64 `json:"pow"` } if err := json.Unmarshal(data, &obj); err != nil { return err } + args.TTL = obj.TTL args.From = obj.From args.To = obj.To + args.Key = obj.Key + args.Topic = obj.Topic args.Payload = obj.Payload - args.Priority = obj.Priority.Int64() - args.TTL = obj.TTL.Int64() - - // decode topic strings - args.Topics = make([][]byte, len(obj.Topics)) - for i, topic := range obj.Topics { - args.Topics[i] = common.FromHex(topic) - } + args.Padding = obj.Padding + args.Work = obj.Work + args.PoW = obj.PoW return nil } +// WhisperMessage is the RPC representation of a whisper message. +type WhisperMessage struct { + ref *ReceivedMessage + + Payload string `json:"payload"` + Padding string `json:"padding"` + From string `json:"from"` + To string `json:"to"` + Sent int64 `json:"sent"` + TTL int64 `json:"ttl"` + PoW int64 `json:"pow"` + Hash string `json:"hash"` +} + +// NewWhisperMessage converts an internal message into an API version. +func NewWhisperMessage(message *ReceivedMessage) WhisperMessage { + return WhisperMessage{ + ref: message, + + Payload: common.ToHex(message.Payload), + Padding: common.ToHex(message.Padding), + From: common.ToHex(crypto.FromECDSAPub(message.Recover())), + To: common.ToHex(crypto.FromECDSAPub(message.Dst)), + Sent: int64(message.Sent), // todo: review format + TTL: int64(message.TTL), // todo: review format + PoW: int64(message.PoW), + Hash: common.ToHex(message.EnvelopeHash.Bytes()), + } +} + +type WhisperFilterArgs struct { + // todo: review types + To string + From string + KeySym string + PoW int + Topics []string +} + // UnmarshalJSON implements the json.Unmarshaler interface, invoked to convert a // JSON message blob into a WhisperFilterArgs structure. -func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { +func (args *WhisperFilterArgs) UnmarshalJSON(b []byte) (err error) { // Unmarshal the JSON message and sanity check var obj struct { To interface{} `json:"to"` From interface{} `json:"from"` + Key interface{} `json:"key"` + PoW interface{} `json:"pow"` Topics interface{} `json:"topics"` } if err := json.Unmarshal(b, &obj); err != nil { @@ -273,7 +314,7 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { } else { argstr, ok := obj.To.(string) if !ok { - return fmt.Errorf("to is not a string") + return fmt.Errorf("'to' is not a string") } args.To = argstr } @@ -282,11 +323,33 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { } else { argstr, ok := obj.From.(string) if !ok { - return fmt.Errorf("from is not a string") + return fmt.Errorf("'from' is not a string") } args.From = argstr } - // Construct the nested topic array + if obj.Key == nil { + args.KeySym = "" + } else { + argstr, ok := obj.Key.(string) + if !ok { + return fmt.Errorf("'key' is not a string") + } + args.KeySym = argstr + } + if obj.PoW == nil { + args.PoW = 0 + } else { + argstr, ok := obj.PoW.(string) + if !ok { + return fmt.Errorf("'pow' is not a string") + } + x, err := strconv.Atoi(argstr) + if err != nil { + return fmt.Errorf("'pow' is invalid") + } + args.PoW = x + } + // Construct the topic array if obj.Topics != nil { // Make sure we have an actual topic array list, ok := obj.Topics.([]interface{}) @@ -294,43 +357,25 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { return fmt.Errorf("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 { + topics := make([]string, len(list)) + for i, field := range list { switch value := field.(type) { case nil: - topics[idx] = []string{} - + topics[i] = "" 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 fmt.Errorf("topic[%d][%d] is not a string", idx, i) - } - } + topics[i] = value default: - return fmt.Errorf("topic[%d] not a string or array", idx) + return fmt.Errorf("topic[%d] is not a string", i) } } - topicsDecoded := make([][][]byte, len(topics)) - for i, condition := range topics { - topicsDecoded[i] = make([][]byte, len(condition)) - for j, topic := range condition { - topicsDecoded[i][j] = common.FromHex(topic) - } - } - - args.Topics = topicsDecoded + // todo: delete this block + //topicsDecoded := make([][]byte, len(topics)) + //for j, t := range topics { + // topicsDecoded[j] = common.FromHex(t) + //} + //args.Topics = topicsDecoded + args.Topics = topics } return nil } @@ -338,8 +383,8 @@ func (args *NewFilterArgs) UnmarshalJSON(b []byte) (err error) { // whisperFilter is the message cache matching a specific filter, accumulating // inbound messages until the are requested by the client. type whisperFilter struct { - id int // Filter identifier for old message retrieval - ref *Whisper // Whisper reference for old message retrieval + id int // Filter identifier for old message retrieval + whisper *Whisper // Whisper reference for old message retrieval cache []WhisperMessage // Cache of messages not yet polled skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication @@ -348,77 +393,49 @@ type whisperFilter struct { lock sync.RWMutex // Lock protecting the filter internals } +// newWhisperFilter creates a new serialized, poll based whisper topic filter. +func newWhisperFilter(id int, w *Whisper) *whisperFilter { + return &whisperFilter{ + id: id, + whisper: w, + update: time.Now(), + skip: make(map[common.Hash]struct{}), + } +} + // messages retrieves all the cached messages from the entire pool matching the // filter, resetting the filter's change buffer. -func (w *whisperFilter) messages() []*Message { - w.lock.Lock() - defer w.lock.Unlock() +func (filter *whisperFilter) messages() []*ReceivedMessage { + filter.lock.Lock() + defer filter.lock.Unlock() - w.cache = nil - w.update = time.Now() + filter.cache = nil + filter.update = time.Now() - w.skip = make(map[common.Hash]struct{}) - messages := w.ref.Messages(w.id) + filter.skip = make(map[common.Hash]struct{}) + messages := filter.whisper.Messages(filter.id) for _, message := range messages { - w.skip[message.Hash] = struct{}{} + filter.skip[message.EnvelopeHash] = struct{}{} } return messages } // insert injects a new batch of messages into the filter cache. -func (w *whisperFilter) insert(messages ...WhisperMessage) { - w.lock.Lock() - defer w.lock.Unlock() +func (filter *whisperFilter) insert(message WhisperMessage) { + filter.lock.Lock() + defer filter.lock.Unlock() - for _, message := range messages { - if _, ok := w.skip[message.ref.Hash]; !ok { - w.cache = append(w.cache, messages...) - } + if _, ok := filter.skip[message.ref.EnvelopeHash]; !ok { + filter.cache = append(filter.cache, message) } } // retrieve fetches all the cached messages from the filter. -func (w *whisperFilter) retrieve() (messages []WhisperMessage) { - w.lock.Lock() - defer w.lock.Unlock() - - messages, w.cache = w.cache, nil - w.update = time.Now() +func (filter *whisperFilter) retrieve() (messages []WhisperMessage) { + filter.lock.Lock() + defer filter.lock.Unlock() + messages, filter.cache = filter.cache, nil + filter.update = time.Now() return } - -// activity returns the last time instance when client requests were executed on -// the filter. -func (w *whisperFilter) activity() time.Time { - w.lock.RLock() - defer w.lock.RUnlock() - - return w.update -} - -// newWhisperFilter creates a new serialized, poll based whisper topic filter. -func newWhisperFilter(id int, ref *Whisper) *whisperFilter { - return &whisperFilter{ - id: id, - ref: ref, - - update: time.Now(), - skip: make(map[common.Hash]struct{}), - } -} - -// NewWhisperMessage converts an internal message into an API version. -func NewWhisperMessage(message *Message) WhisperMessage { - return WhisperMessage{ - ref: message, - - Payload: common.ToHex(message.Payload), - From: common.ToHex(crypto.FromECDSAPub(message.Recover())), - To: common.ToHex(crypto.FromECDSAPub(message.To)), - Sent: message.Sent.Unix(), - TTL: int64(message.TTL / time.Second), - Hash: common.ToHex(message.Hash.Bytes()), - } -} -*/ diff --git a/whisper5/doc.go b/whisper5/doc.go index 351b8e8aa1..a3cefa4e88 100644 --- a/whisper5/doc.go +++ b/whisper5/doc.go @@ -31,7 +31,11 @@ particularly the notion of singular endpoints. */ package whisper5 -import "time" +import ( + "time" + + "github.com/ethereum/go-ethereum/common" +) const ( statusCode = 0x00 @@ -62,3 +66,21 @@ const ( // classifications of a message, determined as the first (left) 4 bytes of the // SHA3 hash of some arbitrary data given by the original author of the message. type TopicType [4]byte + +func BytesToTopic(b []byte) (t TopicType) { + sz := 4 + if x := len(b); x < 4 { + sz = x + } + for i := 0; i < sz; i++ { + t[i] = b[i] + } + return t +} + +func HashToTopic(h common.Hash) (t TopicType) { + for i := 0; i < 4; i++ { + t[i] = h[i] + } + return t +} diff --git a/whisper5/envelope.go b/whisper5/envelope.go index b282618e9a..a35f507618 100644 --- a/whisper5/envelope.go +++ b/whisper5/envelope.go @@ -70,14 +70,14 @@ func (self *Envelope) isAsymmetric() bool { // Seal closes the envelope by spending the requested amount of time as a proof // of work on hashing the data. -func (self *Envelope) Seal(dur time.Duration) { - self.Expiry += uint32(dur.Seconds()) // adjust for the duration of Seal() execution +func (self *Envelope) Seal(work time.Duration) { + self.Expiry += uint32(work.Seconds()) // adjust for the duration of Seal() execution buf := make([]byte, 64) h := crypto.Keccak256(self.rlpWithoutNonce()) copy(buf[:32], h) - finish, bestBit := time.Now().Add(dur).UnixNano(), 0 + finish, bestBit := time.Now().Add(work).UnixNano(), 0 for nonce := uint64(0); time.Now().UnixNano() < finish; { for i := 0; i < 1024; i++ { binary.BigEndian.PutUint64(buf[56:], nonce) @@ -150,14 +150,30 @@ func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error return } -// Open tries to decrypt an envelope -func (self *Envelope) Open(watcher *Filter) *ReceivedMessage { +// Open tries to decrypt an envelope, and populates the message fields in case of success. +func (self *Envelope) Open(watcher *Filter) (msg *ReceivedMessage) { if self.isAsymmetric() { - msg, _ := self.OpenAsymmetric(watcher.KeyAsym) - return msg + msg, _ = self.OpenAsymmetric(watcher.KeyAsym) + if msg != nil { + msg.Dst = watcher.Dst + } } else if self.isSymmetric() { - msg, _ := self.OpenSymmetric(watcher.KeySym) - return msg + msg, _ = self.OpenSymmetric(watcher.KeySym) + if msg != nil { + msg.TopicKeyHash = crypto.Keccak256Hash(watcher.KeySym) + } } - return nil + + if msg != nil { + ok := msg.Validate() + if !ok { + return nil + } + msg.Topic = self.Topic + msg.PoW = self.pow + msg.TTL = self.TTL + msg.Sent = self.Expiry - self.TTL // todo: review + msg.EnvelopeHash = self.hash + } + return msg } diff --git a/whisper5/filter.go b/whisper5/filter.go index 2d3b7798ce..1f6289ca5b 100644 --- a/whisper5/filter.go +++ b/whisper5/filter.go @@ -1,19 +1,18 @@ package whisper5 import ( - "bytes" "crypto/ecdsa" -) -var empty = TopicType{0, 0, 0, 0} + "github.com/ethereum/go-ethereum/common" +) type Filter struct { Src *ecdsa.PublicKey // Sender of the message Dst *ecdsa.PublicKey // Recipient of the message KeyAsym *ecdsa.PrivateKey // Private Key of recipient - Topic TopicType // Topics to filter messages with + Topics []TopicType // Topics to filter messages with KeySym []byte // Key associated with the Topic - TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + TopicKeyHash common.Hash // The Keccak256Hash of the symmetric key PoW int // Proof of work as described in the Whisper spec Fn func(msg *ReceivedMessage) // Handler in case of a match } @@ -23,13 +22,15 @@ type Filters struct { watchers map[int]*Filter ch chan Envelope quit chan struct{} + whisper *Whisper } -func NewFilters() *Filters { +func NewFilters(w *Whisper) *Filters { return &Filters{ ch: make(chan Envelope), watchers: make(map[int]*Filter), quit: make(chan struct{}), + whisper: w, } } @@ -80,7 +81,7 @@ func (self *Filters) processEnvelope(envelope *Envelope) { } else { match = watcher.MatchEnvelope(envelope) if match { - msg = envelope.Open(watcher) // todo: fill all the fields & validate + msg = envelope.Open(watcher) } } @@ -88,18 +89,23 @@ func (self *Filters) processEnvelope(envelope *Envelope) { watcher.Trigger(msg) } } + + if msg != nil { + go self.whisper.addDecryptedMessage(msg) + } } -func (self Filter) expectsPublicKeyEncryption() bool { +func (self Filter) expectsAsymmetricEncryption() bool { return self.KeyAsym != nil } -func (self Filter) expectsTopicEncryption() bool { +func (self Filter) expectsSymmetricEncryption() bool { return self.KeySym != nil } func (self Filter) Trigger(msg *ReceivedMessage) { - go self.Fn(msg) // todo: review + // todo: save msg hash in the filter + self.Fn(msg) } func (self Filter) MatchMessage(msg *ReceivedMessage) bool { @@ -107,14 +113,23 @@ func (self Filter) MatchMessage(msg *ReceivedMessage) bool { return false } - if self.expectsPublicKeyEncryption() && msg.isAsymmetric() { + if self.Src != nil && msg.Src != self.Src { + return false + } + + if self.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() { return self.Dst == msg.Dst - } else if self.expectsTopicEncryption() && msg.isSymmetric() { + } else if self.expectsSymmetricEncryption() && msg.isSymmetricEncryption() { // we need to compare the keys (or rather thier hashes), because of // possible collision (different keys can produce the same topic). // we also need to compare the topics, because they could be arbitrary (not related to KeySym). - if self.Topic == msg.Topic && bytes.Equal(self.TopicKeyHash, msg.TopicKeyHash) { - return true + if self.TopicKeyHash == msg.TopicKeyHash { + for _, t := range self.Topics { + if t == msg.Topic { + return true + } + } + return false } } return false @@ -126,16 +141,22 @@ func (self Filter) MatchEnvelope(envelope *Envelope) bool { } encryptionMethodMatch := false - if self.expectsPublicKeyEncryption() && envelope.isAsymmetric() { + if self.expectsAsymmetricEncryption() && envelope.isAsymmetric() { encryptionMethodMatch = true - } else if self.expectsTopicEncryption() && envelope.isSymmetric() { + if self.Topics == nil { + return true // wildcard + } + } else if self.expectsSymmetricEncryption() && envelope.isSymmetric() { encryptionMethodMatch = true } if encryptionMethodMatch { - if self.Topic == empty || self.Topic == envelope.Topic { - return true + for _, t := range self.Topics { + if t == envelope.Topic { + return true + } } + return false } return false } diff --git a/whisper5/message.go b/whisper5/message.go index 93c842cddf..22066b0310 100644 --- a/whisper5/message.go +++ b/whisper5/message.go @@ -16,7 +16,7 @@ // Contains the Whisper protocol Message element. For formal details please see // the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#messages. -// todo: fix the spec link +// todo: fix the spec link, and move it to doc.go package whisper5 @@ -39,13 +39,14 @@ import ( // Options specifies the exact way a message should be wrapped into an Envelope. type Options struct { - Topic TopicType - TTL time.Duration - Src *ecdsa.PrivateKey - Dst *ecdsa.PublicKey - Key []byte // must be 32 bytes. todo: review - Salt []byte - Pad []byte + TTL time.Duration + Src *ecdsa.PrivateKey + Dst *ecdsa.PublicKey + KeySym []byte // must be 32 bytes. todo: review + Topic TopicType + Pad []byte + Work time.Duration + PoW int } // SentMessage represents an end-user data packet to transmit through the @@ -65,15 +66,21 @@ type ReceivedMessage struct { Signature []byte PoW int // Proof of work as described in the Whisper spec - Sent time.Time // Time when the message was posted into the network - TTL time.Duration // Maximum time to live allowed for the message + Sent uint32 // Time when the message was posted into the network + TTL uint32 // Maximum time to live allowed for the message Src *ecdsa.PublicKey // Message recipient (identity used to decode the message) Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) Topic TopicType - TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + TopicKeyHash common.Hash // The Keccak256Hash of the key, associated with the Topic EnvelopeHash common.Hash // Message envelope hash to act as a unique id } +func DeriveTopicFromSymmetricKey(key []byte) TopicType { + // todo: it is not secure enough, use kdf instead + hash := crypto.Keccak256Hash(key) + return HashToTopic(hash) +} + func isMessageSigned(flags byte) bool { return (flags & signatureFlag) != 0 } @@ -82,11 +89,11 @@ func isMessagePadded(flags byte) bool { return (flags & paddingFlag) != 0 } -func (self *ReceivedMessage) isSymmetric() bool { - return self.TopicKeyHash != nil +func (self *ReceivedMessage) isSymmetricEncryption() bool { + return self.TopicKeyHash != common.Hash{} } -func (self *ReceivedMessage) isAsymmetric() bool { +func (self *ReceivedMessage) isAsymmetricEncryption() bool { return self.Dst != nil } @@ -206,11 +213,10 @@ func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte // - options.From != nil && options.To == nil: signed broadcast (known sender) // - options.From == nil && options.To != nil: encrypted anonymous message // - options.From != nil && options.To != nil: encrypted signed message -func (self *SentMessage) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { +func (self *SentMessage) Wrap(options Options) (envelope *Envelope, err error) { if options.TTL == 0 { options.TTL = DefaultTTL } - //self.TTL = options.TTL // todo: review self.appendPadding(options) if options.Src != nil { if err = self.sign(options.Src); err != nil { @@ -225,15 +231,19 @@ func (self *SentMessage) Wrap(pow time.Duration, options Options) (envelope *Env var salt, nonce []byte if options.Dst != nil { err = self.encryptAsymmetric(options.Dst) - } else if options.Key != nil { - salt, nonce, err = self.encryptSymmetric(options.Key) + } else if options.KeySym != nil { + salt, nonce, err = self.encryptSymmetric(options.KeySym) } else { err = errors.New("Unable to encrypt the message: neither Dst nor Key") } if err == nil { + if (options.Topic == TopicType{}) { + options.Topic = DeriveTopicFromSymmetricKey(options.KeySym) + } + envelope = NewEnvelope(options.TTL, options.Topic, salt, nonce, self) - envelope.Seal(pow) + envelope.Seal(options.Work) // todo: use options.pow, review Seal() as well } return } @@ -305,7 +315,7 @@ func (self *ReceivedMessage) Validate() bool { } self.Payload = self.Raw[1:cur] - if self.isSymmetric() == self.isAsymmetric() { + if self.isSymmetricEncryption() == self.isAsymmetricEncryption() { return false } return true @@ -331,104 +341,3 @@ func (self *ReceivedMessage) hash() []byte { } return crypto.Keccak256(self.Raw) } - -// todo: delete this stuff -/* -// Signature returns the signature part of the raw message. -func (self *ReceivedMessage) ExtractSignature() { - if self.Signature == nil { - if sz := len(self.Raw); sz >= signatureLength+1 { - if isMessageSigned(self.Raw[0]) { - self.Signature = self.Raw[sz-signatureLength:] - } - } - } -} - -// Payload returns the payload part of the raw message. -func (self *ReceivedMessage) ExtractPayload() { - if self.Payload == nil { - end := len(self.Raw) - if isMessageSigned(self.Raw[0]) { - end -= signatureLength - if end <= 1 { - return - } - } - if isMessagePadded(self.Raw[0]) { - paddingSize := int(self.Raw[end-1]) - end -= paddingSize - if end <= 1 { - return - } - } - self.Payload = self.Raw[1:end] - } -} - -// Padding returns the padding part of the raw message -// without the last byte (which only contains the padding size). -func (self *ReceivedMessage) ExtractPadding() { - if self.Padding == nil { - end := len(self.Raw) - if isMessagePadded(self.Raw[0]) { - if isMessageSigned(self.Raw[0]) { - end -= signatureLength - if end <= 1 { - return - } - } - paddingSize := int(self.Raw[end-1]) - beg := end - paddingSize - if beg > 1 { - self.Padding = self.Raw[beg : end-1] - } - } - } -} -*/ -/* -// Signature returns the signature part of the raw message. -func (self *ReceivedMessage) ExtractSignature() []byte { - sz := len(self.Raw) - if self.isSigned() && sz >= signatureLength+1 { - return self.Raw[sz-signatureLength:] - } else { - return nil - } -} - -// Payload returns the payload part of the raw message. -func (self *ReceivedMessage) ExtractPayload() []byte { - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - if self.isPadded() { - paddingSize := int(self.Raw[end-1]) - end -= paddingSize - } - if end <= 1 { - return nil - } - return self.Raw[1:end] -} - -// Padding returns the padding part of the raw message -// without the last byte (which only contains the padding size). -func (self *ReceivedMessage) ExtractPadding() []byte { - if !self.isPadded() { - return nil - } - end := len(self.Raw) - if self.isSigned() { - end -= signatureLength - } - paddingSize := int(self.Raw[end-1]) - beg := end - paddingSize - if beg <= 1 { - return nil - } - return self.Raw[beg : end-1] -} -*/ diff --git a/whisper5/whisper.go b/whisper5/whisper.go index 3a6f1885ba..1e24a3893c 100644 --- a/whisper5/whisper.go +++ b/whisper5/whisper.go @@ -31,12 +31,6 @@ import ( set "gopkg.in/fatih/set.v0" ) -//type MessageEvent struct { -// To *ecdsa.PrivateKey -// From *ecdsa.PublicKey -// Message *Message -//} - // Whisper represents a dark communication interface through the Ethereum // network, using its very own P2P communication layer. type Whisper struct { @@ -44,10 +38,10 @@ type Whisper struct { filters *Filters privateKeys map[string]*ecdsa.PrivateKey - topicKeys map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions + //topicKeys map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions - msgs map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages // todo: rename envelopes map[common.Hash]*Envelope // Pool of messages currently tracked by this node + messages map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages expirations map[uint32]*set.SetNonTS // Message expiration pool (TODO: something lighter) poolMu sync.RWMutex // Mutex to sync the message and expiration pools @@ -60,14 +54,16 @@ type Whisper struct { // New creates a Whisper client ready to communicate through the Ethereum P2P network. func NewWhisper() *Whisper { whisper := &Whisper{ - filters: NewFilters(), + //filters: NewFilters(), privateKeys: make(map[string]*ecdsa.PrivateKey), - topicKeys: make(map[TopicType][]byte), + //topicKeys: make(map[TopicType][]byte), envelopes: make(map[common.Hash]*Envelope), + messages: make(map[common.Hash]*ReceivedMessage), expirations: make(map[uint32]*set.SetNonTS), peers: make(map[*peer]struct{}), quit: make(chan struct{}), } + whisper.filters = NewFilters(whisper) whisper.filters.Start() // p2p whisper sub protocol handler @@ -103,16 +99,15 @@ func (self *Whisper) Version() uint { return self.protocol.Version } +// todo: review. maybe we need to delete these identity-related functions, since the key moved to Filter // NewIdentity generates a new cryptographic identity for the client, and injects // it into the known identities for message decryption. func (self *Whisper) NewIdentity() *ecdsa.PrivateKey { - // todo: review key, err := crypto.GenerateKey() if err != nil { panic(err) } self.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key - return key } @@ -245,27 +240,6 @@ func (self *Whisper) postEvent(envelope *Envelope) { self.filters.Notify(envelope) } -/* -// createFilter creates a message filter to check against installed handlers. -func createFilter(message *Message, topics []TopicType) filter.Filter { - //return Filter{ - // Src: string(crypto.FromECDSAPub(message.Recover())), - // Dst: string(crypto.FromECDSAPub(message.Dst)), - // Topics: topics, - //} - - matcher := make([][]TopicType, len(topics)) - for i, topic := range topics { - matcher[i] = []TopicType{topic} - } - return filterer{ - to: string(crypto.FromECDSAPub(message.To)), - from: string(crypto.FromECDSAPub(message.Recover())), - matcher: newTopicMatcher(matcher...), - } -} -*/ - // update loops until the lifetime of the whisper node, updating its internal // state by expiring stale messages from the pool. func (self *Whisper) update() { @@ -299,6 +273,7 @@ func (self *Whisper) expire() { // Dump all expired messages and remove timestamp hashSet.Each(func(v interface{}) bool { delete(self.envelopes, v.(common.Hash)) + delete(self.messages, v.(common.Hash)) return true }) self.expirations[then].Clear() @@ -317,33 +292,25 @@ func (self *Whisper) Envelopes() []*Envelope { return all } -/* // Messages retrieves all the currently pooled messages matching a filter id. -// todo: review -//func (self *Whisper) Messages(id int) []*Message { -// messages := make([]*Message, 0) -// if filter := self.filters.Get(id); filter != nil { -// for _, envelope := range self.messages { -// if message := self.open(envelope); message != nil { -// if self.filters.Match(filter, createFilter(message, envelope.Topic)) { -// messages = append(messages, message) -// } -// } -// } -// } -// return messages -//} func (self *Whisper) Messages(id int) []*ReceivedMessage { - messages := make([]*Envelope, 0) + self.poolMu.RLock() + defer self.poolMu.RUnlock() + + result := make([]*ReceivedMessage, 0) if filter := self.filters.Get(id); filter != nil { - for _, envelope := range self.envelopes { - //if message := self.open(envelope); message != nil { - if self.filters.Match(envelope) { - messages = append(messages, envelope) + for _, msg := range self.messages { + if filter.MatchMessage(msg) { + result = append(result, msg) } - //} } } - return messages + return result } -*/ \ No newline at end of file + +func (self *Whisper) addDecryptedMessage(msg *ReceivedMessage) { + self.poolMu.Lock() + defer self.poolMu.Unlock() + + self.messages[msg.EnvelopeHash] = msg +} \ No newline at end of file