From baef2464172a797f995e78a72c89faa4b8caf015 Mon Sep 17 00:00:00 2001 From: Vlad Date: Thu, 1 Sep 2016 18:54:25 +0200 Subject: [PATCH] whisper5: added new files --- whisper5/api.go | 424 +++++++++++++++++++++++++++++++++++++++++++ whisper5/envelope.go | 54 +++--- whisper5/filter.go | 141 ++++++++++++++ whisper5/message.go | 384 ++++++++++++++++++++++++--------------- whisper5/peer.go | 175 ++++++++++++++++++ whisper5/topic.go | 38 ++++ whisper5/whisper.go | 349 +++++++++++++++++++++++++++++++++++ 7 files changed, 1395 insertions(+), 170 deletions(-) create mode 100644 whisper5/api.go create mode 100644 whisper5/filter.go create mode 100644 whisper5/peer.go create mode 100644 whisper5/topic.go create mode 100644 whisper5/whisper.go diff --git a/whisper5/api.go b/whisper5/api.go new file mode 100644 index 0000000000..ddd17ac392 --- /dev/null +++ b/whisper5/api.go @@ -0,0 +1,424 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +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" + "fmt" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/rpc" +) + +// PublicWhisperAPI provides the whisper RPC service. +type PublicWhisperAPI struct { + w *Whisper + + messagesMu sync.RWMutex + messages map[int]*whisperFilter +} + +type whisperOfflineError struct{} + +func (e *whisperOfflineError) Error() string { + return "whisper is offline" +} + +// whisperOffLineErr is returned when the node doesn't offer the shh service. +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)} +} + +// Version returns the Whisper version this node offers. +func (s *PublicWhisperAPI) Version() (*rpc.HexNumber, error) { + if s.w == nil { + return rpc.NewHexNumber(0), whisperOffLineErr + } + return rpc.NewHexNumber(s.w.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 { + return false, whisperOffLineErr + } + return s.w.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 { + return "", whisperOffLineErr + } + + identity := s.w.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 { + 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) { + 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) + } + }, + } + + id = s.w.Watch(filter) + + s.messagesMu.Lock() + s.messages[id] = newWhisperFilter(id, s.w) + s.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() + + if _, ok := s.messages[filterId.Int()]; ok { + delete(s.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() + + var messages []*Message + if s.messages[filterId.Int()] != nil { + messages = s.messages[filterId.Int()].messages() + } + + return returnWhisperMessages(messages) +} + +// returnWhisperMessages converts a Whisper message to a RPC whisper message. +func returnWhisperMessages(messages []*Message) []WhisperMessage { + msgs := make([]WhisperMessage, len(messages)) + for i, msg := range messages { + msgs[i] = NewWhisperMessage(msg) + } + 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 { + return false, whisperOffLineErr + } + + // construct whisper message with transmission options + message := NewMessage(common.FromHex(args.Payload)) + options := Options{ + To: crypto.ToECDSAPub(common.FromHex(args.To)), + TTL: time.Duration(args.TTL) * time.Second, + Topics: NewTopics(args.Topics...), + } + + // set sender identity + if len(args.From) > 0 { + if key := s.w.GetIdentity(crypto.ToECDSAPub(common.FromHex(args.From))); key != nil { + options.From = key + } 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) + if err != nil { + return false, err + } + + return true, s.w.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"` + TTL int64 `json:"ttl"` + Hash string `json:"hash"` +} + +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"` + } + + if err := json.Unmarshal(data, &obj); err != nil { + return err + } + + args.From = obj.From + args.To = obj.To + 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) + } + + return nil +} + +// 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) { + // Unmarshal the JSON message and sanity check + var obj struct { + To interface{} `json:"to"` + From interface{} `json:"from"` + Topics interface{} `json:"topics"` + } + if err := json.Unmarshal(b, &obj); err != nil { + return err + } + + // Retrieve the simple data contents of the filter arguments + if obj.To == nil { + args.To = "" + } else { + argstr, ok := obj.To.(string) + if !ok { + return fmt.Errorf("to is not a string") + } + args.To = argstr + } + if obj.From == nil { + args.From = "" + } else { + argstr, ok := obj.From.(string) + if !ok { + return fmt.Errorf("from is not a string") + } + args.From = argstr + } + // Construct the nested topic array + if obj.Topics != nil { + // Make sure we have an actual topic array + list, ok := obj.Topics.([]interface{}) + if !ok { + 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 { + 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 fmt.Errorf("topic[%d][%d] is not a string", idx, i) + } + } + default: + return fmt.Errorf("topic[%d] not a string or array", idx) + } + } + + 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 + } + return nil +} + +// 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 + + cache []WhisperMessage // Cache of messages not yet polled + skip map[common.Hash]struct{} // List of retrieved messages to avoid duplication + update time.Time // Time of the last message query + + lock sync.RWMutex // Lock protecting the filter internals +} + +// 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() + + w.cache = nil + w.update = time.Now() + + w.skip = make(map[common.Hash]struct{}) + messages := w.ref.Messages(w.id) + for _, message := range messages { + w.skip[message.Hash] = 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() + + for _, message := range messages { + if _, ok := w.skip[message.ref.Hash]; !ok { + w.cache = append(w.cache, messages...) + } + } +} + +// 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() + + 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/envelope.go b/whisper5/envelope.go index dda1fbd6db..b282618e9a 100644 --- a/whisper5/envelope.go +++ b/whisper5/envelope.go @@ -43,11 +43,12 @@ type Envelope struct { EnvNonce uint64 hash common.Hash // Cached hash of the envelope to avoid rehashing every time + pow int // Message-specific PoW as described in the Whisper specification } // NewEnvelope wraps a Whisper message with expiration and destination data // included into an envelope for network forwarding. -func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byte, msg *Message) *Envelope { +func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byte, msg *SentMessage) *Envelope { return &Envelope{ Expiry: uint32(time.Now().Add(ttl).Unix()), TTL: uint32(ttl.Seconds()), @@ -59,16 +60,24 @@ func NewEnvelope(ttl time.Duration, topic TopicType, salt []byte, aesNonce []byt } } +func (self *Envelope) isSymmetric() bool { + return self.AESNonce != nil +} + +func (self *Envelope) isAsymmetric() bool { + return !self.isSymmetric() +} + // Seal closes the envelope by spending the requested amount of time as a proof // of work on hashing the data. -func (self *Envelope) Seal(pow time.Duration) { - self.Expiry += uint32(pow.Seconds()) // adjust for the duration of Seal() execution +func (self *Envelope) Seal(dur time.Duration) { + self.Expiry += uint32(dur.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(pow).UnixNano(), 0 + finish, bestBit := time.Now().Add(dur).UnixNano(), 0 for nonce := uint64(0); time.Now().UnixNano() < finish; { for i := 0; i < 1024; i++ { binary.BigEndian.PutUint64(buf[56:], nonce) @@ -107,7 +116,8 @@ func (self *Envelope) DecodeRLP(s *rlp.Stream) error { // The decoding of Envelope uses the struct fields but also needs // to compute the hash of the whole RLP-encoded envelope. This // type has the same structure as Envelope but is not an - // rlp.Decoder (does not implement DecodeRLP() function). + // rlp.Decoder (does not implement DecodeRLP function). + // Only public members will be encoded. type rlpenv Envelope if err := rlp.DecodeBytes(raw, (*rlpenv)(self)); err != nil { return err @@ -117,39 +127,37 @@ func (self *Envelope) DecodeRLP(s *rlp.Stream) error { } // OpenAsymmetric tries to decrypt an envelope, potentially encrypted with a particular key. -func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*Message, error) { - message := &Message{ - Raw: self.Data, - //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), - //TTL: time.Duration(self.TTL) * time.Second, - //Hash: self.Hash(), - } - +func (self *Envelope) OpenAsymmetric(key *ecdsa.PrivateKey) (*ReceivedMessage, error) { + message := &ReceivedMessage{Raw: self.Data} err := message.decryptAsymmetric(key) switch err { case nil: return message, nil - case ecies.ErrInvalidPublicKey: // addressed to somebody else return nil, err - default: return nil, fmt.Errorf("unable to open envelope, decrypt failed: %v", err) } } // OpenSymmetric tries to decrypt an envelope, potentially encrypted with a particular key. -func (self *Envelope) OpenSymmetric(key []byte) (msg *Message, err error) { - msg = &Message{ - Raw: self.Data, - //Sent: time.Unix(int64(self.Expiry-self.TTL), 0), - //TTL: time.Duration(self.TTL) * time.Second, - //Hash: self.Hash(), - } - +func (self *Envelope) OpenSymmetric(key []byte) (msg *ReceivedMessage, err error) { + msg = &ReceivedMessage{Raw: self.Data} err = msg.decryptSymmetric(key, self.Salt, self.AESNonce) if err != nil { msg = nil } return } + +// Open tries to decrypt an envelope +func (self *Envelope) Open(watcher *Filter) *ReceivedMessage { + if self.isAsymmetric() { + msg, _ := self.OpenAsymmetric(watcher.KeyAsym) + return msg + } else if self.isSymmetric() { + msg, _ := self.OpenSymmetric(watcher.KeySym) + return msg + } + return nil +} diff --git a/whisper5/filter.go b/whisper5/filter.go new file mode 100644 index 0000000000..2d3b7798ce --- /dev/null +++ b/whisper5/filter.go @@ -0,0 +1,141 @@ +package whisper5 + +import ( + "bytes" + "crypto/ecdsa" +) + +var empty = TopicType{0, 0, 0, 0} + +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 + KeySym []byte // Key associated with the Topic + TopicKeyHash []byte // The Keccak256Hash of the key, associated with the Topic + PoW int // Proof of work as described in the Whisper spec + Fn func(msg *ReceivedMessage) // Handler in case of a match +} + +type Filters struct { + id int + watchers map[int]*Filter + ch chan Envelope + quit chan struct{} +} + +func NewFilters() *Filters { + return &Filters{ + ch: make(chan Envelope), + watchers: make(map[int]*Filter), + quit: make(chan struct{}), + } +} + +func (self *Filters) Start() { + go self.loop() +} + +func (self *Filters) Stop() { + close(self.quit) +} + +func (self *Filters) Notify(env *Envelope) { + self.ch <- *env +} + +func (self *Filters) Install(watcher *Filter) int { + self.watchers[self.id] = watcher + ret := self.id + self.id++ + return ret +} + +func (self *Filters) Uninstall(id int) { + delete(self.watchers, id) +} + +func (self *Filters) Get(i int) *Filter { + return self.watchers[i] +} + +func (self *Filters) loop() { + for { + select { + case <-self.quit: + return + case envelope := <-self.ch: + self.processEnvelope(&envelope) + } + } +} + +func (self *Filters) processEnvelope(envelope *Envelope) { + var msg *ReceivedMessage + for _, watcher := range self.watchers { + match := false + if msg != nil { + match = watcher.MatchMessage(msg) + } else { + match = watcher.MatchEnvelope(envelope) + if match { + msg = envelope.Open(watcher) // todo: fill all the fields & validate + } + } + + if match && msg != nil { + watcher.Trigger(msg) + } + } +} + +func (self Filter) expectsPublicKeyEncryption() bool { + return self.KeyAsym != nil +} + +func (self Filter) expectsTopicEncryption() bool { + return self.KeySym != nil +} + +func (self Filter) Trigger(msg *ReceivedMessage) { + go self.Fn(msg) // todo: review +} + +func (self Filter) MatchMessage(msg *ReceivedMessage) bool { + if self.PoW > 0 && msg.PoW < self.PoW { + return false + } + + if self.expectsPublicKeyEncryption() && msg.isAsymmetric() { + return self.Dst == msg.Dst + } else if self.expectsTopicEncryption() && msg.isSymmetric() { + // 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 + } + } + return false +} + +func (self Filter) MatchEnvelope(envelope *Envelope) bool { + if self.PoW > 0 && envelope.pow < self.PoW { + return false + } + + encryptionMethodMatch := false + if self.expectsPublicKeyEncryption() && envelope.isAsymmetric() { + encryptionMethodMatch = true + } else if self.expectsTopicEncryption() && envelope.isSymmetric() { + encryptionMethodMatch = true + } + + if encryptionMethodMatch { + if self.Topic == empty || self.Topic == envelope.Topic { + return true + } + } + return false +} diff --git a/whisper5/message.go b/whisper5/message.go index fbec82e3f4..93c842cddf 100644 --- a/whisper5/message.go +++ b/whisper5/message.go @@ -30,6 +30,7 @@ import ( "crypto/ecdsa" "crypto/sha256" + "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" @@ -47,83 +48,50 @@ type Options struct { Pad []byte } -// Message represents an end-user data packet to transmit through the Whisper -// protocol. These are wrapped into Envelopes that need not be understood by -// intermediate nodes, just forwarded. -type Message struct { - //Flags byte // first bit: signature presence, second: padding presence - //Padding []byte // the first byte contains it's size - //Payload []byte // todo: delete all this - //Signature []byte +// SentMessage represents an end-user data packet to transmit through the +// Whisper protocol. These are wrapped into Envelopes that need not be +// understood by intermediate nodes, just forwarded. +type SentMessage struct { + Raw []byte +} +// ReceivedMessage represents a data packet to be received through the +// Whisper protocol. +type ReceivedMessage struct { Raw []byte - // todo: following are the fields, extracted from the Raw field of received msg (not transmitted) - //Sent time.Time // Time when the message was posted into the network - //TTL time.Duration // Maximum time to live allowed for the message - // - Dst *ecdsa.PublicKey // Message recipient (identity used to decode the message) - //Hash common.Hash // Message envelope hash to act as a unique id + Payload []byte + Padding []byte + 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 + 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 + EnvelopeHash common.Hash // Message envelope hash to act as a unique id } -func (self *Message) flags() byte { - return self.Raw[0] +func isMessageSigned(flags byte) bool { + return (flags & signatureFlag) != 0 } -func (self *Message) isSigned() bool { - return (self.Raw[0] & signatureFlag) != 0 +func isMessagePadded(flags byte) bool { + return (flags & paddingFlag) != 0 } -func (self *Message) isPadded() bool { - return (self.Raw[0] & paddingFlag) != 0 +func (self *ReceivedMessage) isSymmetric() bool { + return self.TopicKeyHash != nil } -// Signature returns the signature part of the raw message. -func (self *Message) Signature() []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 *Message) Payload() []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 *Message) Padding() []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] +func (self *ReceivedMessage) isAsymmetric() bool { + return self.Dst != nil } // NewMessage creates and initializes a non-signed, non-encrypted Whisper message. -func NewMessage(payload []byte) *Message { +func NewSentMessage(payload []byte) *SentMessage { // Construct an initial flag set: no signature, no padding, other bits random buf := make([]byte, 1) crand.Read(buf) @@ -132,7 +100,7 @@ func NewMessage(payload []byte) *Message { flags &= ^signatureFlag flags &= ^paddingFlag - msg := Message{} //Message{Sent: time.Now()} // todo: review + msg := SentMessage{} //Message{Sent: time.Now()} // todo: review msg.Raw = make([]byte, 1, len(payload)+signatureLength+maxPadLength) msg.Raw[0] = flags msg.Raw = append(msg.Raw, payload...) @@ -141,12 +109,12 @@ func NewMessage(payload []byte) *Message { // appendPadding appends the pseudorandom padding bytes and sets the padding flag. // The last byte contains the size of padding (thus, its size must not exceed 256). -func (self *Message) appendPadding(options Options) { - if self.isSigned() { +func (self *SentMessage) appendPadding(options Options) { + if isMessageSigned(self.Raw[0]) { // this should not happen, but no reason to panic glog.V(logger.Error).Infof("Trying to pad a message which was already signed") return - } else if self.isPadded() { + } else if isMessagePadded(self.Raw[0]) { // this should not happen, but no reason to panic glog.V(logger.Error).Infof("Trying to pad a message which was already padded") return @@ -172,13 +140,14 @@ func (self *Message) appendPadding(options Options) { // sign calculates and sets the cryptographic signature for the message, // also setting the sign flag. -func (self *Message) sign(key *ecdsa.PrivateKey) (err error) { - if self.isSigned() { +func (self *SentMessage) sign(key *ecdsa.PrivateKey) (err error) { + if isMessageSigned(self.Raw[0]) { // this should not happen, but no reason to panic glog.V(logger.Error).Infof("Trying to sign a message which was already signed") return } - signature, err := crypto.Sign(self.hash(), key) + hash := crypto.Keccak256(self.Raw) + signature, err := crypto.Sign(hash, key) if err != nil { self.Raw = append(self.Raw, signature...) self.Raw[0] |= signatureFlag @@ -186,24 +155,8 @@ func (self *Message) sign(key *ecdsa.PrivateKey) (err error) { return } -// Recover retrieves the public key of the message signer. -func (self *Message) Recover() *ecdsa.PublicKey { - defer func() { recover() }() // in case of invalid signature - - signature := self.Signature() - if signature == nil { - return nil - } - pub, err := crypto.SigToPub(self.hash(), signature) - if err != nil { - glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) - return nil - } - return pub -} - // encryptAsymmetric encrypts a message with a public key. -func (self *Message) encryptAsymmetric(key *ecdsa.PublicKey) error { +func (self *SentMessage) encryptAsymmetric(key *ecdsa.PublicKey) error { encrypted, err := crypto.Encrypt(key, self.Raw) if err == nil { self.Raw = encrypted @@ -211,26 +164,9 @@ func (self *Message) encryptAsymmetric(key *ecdsa.PublicKey) error { return err } -// decryptAsymmetric decrypts an encrypted payload with a private key. -func (self *Message) decryptAsymmetric(key *ecdsa.PrivateKey) error { - decrypted, err := crypto.Decrypt(key, self.Raw) - if err == nil { - self.Raw = decrypted - } - return err -} - // encryptSymmetric encrypts a message with a topic key, using AES-GCM-256. // nonce size should be 12 bytes (see cipher.gcmStandardNonceSize). -func (self *Message) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) { - // todo: delete this block - // The key argument should be the AES-256 key, 32 bytes - //if len(key) != aesKeyLength { - // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) - // err = errors.New("Wrong size of AES key") - // return - //} - +func (self *SentMessage) encryptSymmetric(key []byte) (salt []byte, nonce []byte, err error) { salt = make([]byte, saltLength) _, err = crand.Read(salt) if err != nil { @@ -258,47 +194,6 @@ func (self *Message) encryptSymmetric(key []byte) (salt []byte, nonce []byte, er return } -// decryptSymmetric decrypts a message with a topic key, using AES-GCM-256. -// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize). -func (self *Message) decryptSymmetric(key []byte, salt []byte, nonce []byte) error { - // todo: delete this block - // The key argument should be the AES-256 key, 32 bytes - //if len(key) != aesKeyLength { - // glog.V(logger.Error).Infof("AES key size must be %d bytes", aesKeyLength) - // return errors.New("Wrong size of AES key") - //} - - derivedKey := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) - - block, err := aes.NewCipher(derivedKey) - if err != nil { - return err - } - aesgcm, err := cipher.NewGCM(block) - if err != nil { - return err - } - if len(nonce) != aesgcm.NonceSize() { - glog.V(logger.Error).Infof("AES nonce size must be %d bytes", aesgcm.NonceSize()) - return errors.New("Wrong AES nonce size") - } - decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil) - if err != nil { - return err - } - self.Raw = decrypted - return nil -} - -// hash calculates the SHA3 checksum of the message flags and payload. -func (self *Message) hash() []byte { - if self.isSigned() { - sz := len(self.Raw) - signatureLength - return crypto.Keccak256(self.Raw[:sz]) - } - return crypto.Keccak256(self.Raw) -} - // Wrap bundles the message into an Envelope to transmit over the network. // // pow (Proof Of Work) controls how much time to spend on hashing the message, @@ -311,7 +206,7 @@ func (self *Message) hash() []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 *Message) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { +func (self *SentMessage) Wrap(pow time.Duration, options Options) (envelope *Envelope, err error) { if options.TTL == 0 { options.TTL = DefaultTTL } @@ -342,3 +237,198 @@ func (self *Message) Wrap(pow time.Duration, options Options) (envelope *Envelop } return } + +/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + +// decryptSymmetric decrypts a message with a topic key, using AES-GCM-256. +// nonce size should be 12 bytes (see cipher.gcmStandardNonceSize). +func (self *ReceivedMessage) decryptSymmetric(key []byte, salt []byte, nonce []byte) error { + derivedKey := pbkdf2.Key(key, salt, kdfIterations, aesKeyLength, sha256.New) + + block, err := aes.NewCipher(derivedKey) + if err != nil { + return err + } + aesgcm, err := cipher.NewGCM(block) + if err != nil { + return err + } + if len(nonce) != aesgcm.NonceSize() { + glog.V(logger.Error).Infof("AES nonce size must be %d bytes", aesgcm.NonceSize()) + return errors.New("Wrong AES nonce size") + } + decrypted, err := aesgcm.Open(nil, nonce, self.Raw, nil) + if err != nil { + return err + } + self.Raw = decrypted + return nil +} + +// decryptAsymmetric decrypts an encrypted payload with a private key. +func (self *ReceivedMessage) decryptAsymmetric(key *ecdsa.PrivateKey) error { + decrypted, err := crypto.Decrypt(key, self.Raw) + if err == nil { + self.Raw = decrypted + } + return err +} + +// Validate checks the validity and extracts the fields in case of success +func (self *ReceivedMessage) Validate() bool { + sz := len(self.Raw) + cur := sz + if sz < 1 { + return false + } + + if isMessageSigned(self.Raw[0]) { + cur -= signatureLength + if cur <= 1 { + return false + } + self.Signature = self.Raw[cur:] + self.Src = self.Recover() + if self.Src == nil { + return false + } + } + + if isMessagePadded(self.Raw[0]) { + paddingSize := int(self.Raw[cur-1]) + beg := cur - paddingSize + if beg <= 1 { + return false + } + self.Padding = self.Raw[beg : cur-1] + cur = beg + } + + self.Payload = self.Raw[1:cur] + if self.isSymmetric() == self.isAsymmetric() { + return false + } + return true +} + +// Recover retrieves the public key of the message signer. +func (self *ReceivedMessage) Recover() *ecdsa.PublicKey { + defer func() { recover() }() // in case of invalid signature + + pub, err := crypto.SigToPub(self.hash(), self.Signature) + if err != nil { + glog.V(logger.Error).Infof("Could not get public key from signature: %v", err) + return nil + } + return pub +} + +// hash calculates the SHA3 checksum of the message flags, payload and padding. +func (self *ReceivedMessage) hash() []byte { + if isMessageSigned(self.Raw[0]) { + sz := len(self.Raw) - signatureLength + return crypto.Keccak256(self.Raw[:sz]) + } + 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/peer.go b/whisper5/peer.go new file mode 100644 index 0000000000..434ecd62f9 --- /dev/null +++ b/whisper5/peer.go @@ -0,0 +1,175 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package whisper5 + +import ( + "fmt" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/logger/glog" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" + set "gopkg.in/fatih/set.v0" +) + +// peer represents a whisper protocol peer connection. +type peer struct { + host *Whisper + peer *p2p.Peer + ws p2p.MsgReadWriter + + known *set.Set // Messages already known by the peer to avoid wasting bandwidth + + quit chan struct{} +} + +// newPeer creates a new whisper peer object, but does not run the handshake itself. +func newPeer(host *Whisper, remote *p2p.Peer, rw p2p.MsgReadWriter) *peer { + return &peer{ + host: host, + peer: remote, + ws: rw, + known: set.New(), + quit: make(chan struct{}), + } +} + +// start initiates the peer updater, periodically broadcasting the whisper packets +// into the network. +func (self *peer) start() { + go self.update() + glog.V(logger.Debug).Infof("%v: whisper started", self.peer) +} + +// stop terminates the peer updater, stopping message forwarding to it. +func (self *peer) stop() { + close(self.quit) + glog.V(logger.Debug).Infof("%v: whisper stopped", self.peer) +} + +// handshake sends the protocol initiation status message to the remote peer and +// verifies the remote status too. +func (self *peer) handshake() error { + // Send the handshake status message asynchronously + errc := make(chan error, 1) + go func() { + errc <- p2p.SendItems(self.ws, statusCode, protocolVersion) + }() + // Fetch the remote status packet and verify protocol match + packet, err := self.ws.ReadMsg() + if err != nil { + return err + } + if packet.Code != statusCode { + return fmt.Errorf("peer sent %x before status packet", packet.Code) + } + s := rlp.NewStream(packet.Payload, uint64(packet.Size)) + if _, err := s.List(); err != nil { + return fmt.Errorf("bad status message: %v", err) + } + peerVersion, err := s.Uint() + if err != nil { + return fmt.Errorf("bad status message: %v", err) + } + if peerVersion != protocolVersion { + return fmt.Errorf("protocol version mismatch %d != %d", peerVersion, protocolVersion) + } + // Wait until out own status is consumed too + if err := <-errc; err != nil { + return fmt.Errorf("failed to send status packet: %v", err) + } + return nil +} + +// update executes periodic operations on the peer, including message transmission +// and expiration. +func (self *peer) update() { + // Start the tickers for the updates + expire := time.NewTicker(expirationCycle) + transmit := time.NewTicker(transmissionCycle) + + // Loop and transmit until termination is requested + for { + select { + case <-expire.C: + self.expire() + + case <-transmit.C: + if err := self.broadcast(); err != nil { + glog.V(logger.Info).Infof("%v: broadcast failed: %v", self.peer, err) + return + } + + case <-self.quit: + return + } + } +} + +// mark marks an envelope known to the peer so that it won't be sent back. +func (self *peer) mark(envelope *Envelope) { + self.known.Add(envelope.Hash()) +} + +// marked checks if an envelope is already known to the remote peer. +func (self *peer) marked(envelope *Envelope) bool { + return self.known.Has(envelope.Hash()) +} + +// expire iterates over all the known envelopes in the host and removes all +// expired (unknown) ones from the known list. +func (self *peer) expire() { + // Assemble the list of available envelopes + available := set.NewNonTS() + for _, envelope := range self.host.Envelopes() { + available.Add(envelope.Hash()) + } + // Cross reference availability with known status + unmark := make(map[common.Hash]struct{}) + self.known.Each(func(v interface{}) bool { + if !available.Has(v.(common.Hash)) { + unmark[v.(common.Hash)] = struct{}{} + } + return true + }) + // Dump all known but unavailable + for hash, _ := range unmark { + self.known.Remove(hash) + } +} + +// broadcast iterates over the collection of envelopes and transmits yet unknown +// ones over the network. +func (self *peer) broadcast() error { + // Fetch the envelopes and collect the unknown ones + envelopes := self.host.Envelopes() + transmit := make([]*Envelope, 0, len(envelopes)) + for _, envelope := range envelopes { + if !self.marked(envelope) { + transmit = append(transmit, envelope) + self.mark(envelope) + } + } + // Transmit the unknown batch (potentially empty) + if err := p2p.Send(self.ws, messagesCode, transmit); err != nil { + return err + } + glog.V(logger.Detail).Infoln(self.peer, "broadcasted", len(transmit), "message(s)") + return nil +} diff --git a/whisper5/topic.go b/whisper5/topic.go new file mode 100644 index 0000000000..94ebbc0c6a --- /dev/null +++ b/whisper5/topic.go @@ -0,0 +1,38 @@ +// Copyright 2015 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Contains the Whisper protocol Topic element. For formal details please see +// the specs at https://github.com/ethereum/wiki/wiki/Whisper-PoC-1-Protocol-Spec#topics. + +package whisper5 + +import "github.com/ethereum/go-ethereum/crypto" + +func NewTopic(data []byte) TopicType { + prefix := [4]byte{} + copy(prefix[:], crypto.Keccak256(data)[:4]) + return TopicType(prefix) +} + +// NewTopicFromString creates a topic using the binary data contents of the specified string. +func NewTopicFromString(data string) TopicType { + return NewTopic([]byte(data)) +} + +// String converts a topic byte array to a string representation. +func (self *TopicType) String() string { + return string(self[:]) +} diff --git a/whisper5/whisper.go b/whisper5/whisper.go new file mode 100644 index 0000000000..3a6f1885ba --- /dev/null +++ b/whisper5/whisper.go @@ -0,0 +1,349 @@ +// Copyright 2014 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package whisper5 + +import ( + "crypto/ecdsa" + "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/p2p" + "github.com/ethereum/go-ethereum/rpc" + + 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 { + protocol p2p.Protocol + filters *Filters + + privateKeys map[string]*ecdsa.PrivateKey + 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 + expirations map[uint32]*set.SetNonTS // Message expiration pool (TODO: something lighter) + poolMu sync.RWMutex // Mutex to sync the message and expiration pools + + peers map[*peer]struct{} // Set of currently active peers + peerMu sync.RWMutex // Mutex to sync the active peer set + + quit chan struct{} +} + +// New creates a Whisper client ready to communicate through the Ethereum P2P network. +func NewWhisper() *Whisper { + whisper := &Whisper{ + filters: NewFilters(), + privateKeys: make(map[string]*ecdsa.PrivateKey), + topicKeys: make(map[TopicType][]byte), + envelopes: make(map[common.Hash]*Envelope), + expirations: make(map[uint32]*set.SetNonTS), + peers: make(map[*peer]struct{}), + quit: make(chan struct{}), + } + whisper.filters.Start() + + // p2p whisper sub protocol handler + whisper.protocol = p2p.Protocol{ + Name: protocolName, + Version: uint(protocolVersion), + Length: 2, + Run: whisper.handlePeer, + } + + return whisper +} + +// APIs returns the RPC descriptors the Whisper implementation offers +func (s *Whisper) APIs() []rpc.API { + return []rpc.API{ + { + Namespace: protocolName, + Version: protocolVersionStr, + Service: NewPublicWhisperAPI(s), + Public: true, + }, + } +} + +// Protocols returns the whisper sub-protocols ran by this particular client. +func (self *Whisper) Protocols() []p2p.Protocol { + return []p2p.Protocol{self.protocol} +} + +// Version returns the whisper sub-protocols version number. +func (self *Whisper) Version() uint { + return self.protocol.Version +} + +// 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 +} + +// HasIdentity checks if the the whisper node is configured with the private key +// of the specified public pair. +func (self *Whisper) HasIdentity(key *ecdsa.PublicKey) bool { + return self.privateKeys[string(crypto.FromECDSAPub(key))] != nil +} + +// GetIdentity retrieves the private key of the specified public identity. +func (self *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey { + return self.privateKeys[string(crypto.FromECDSAPub(key))] +} + +// Watch installs a new message handler to run in case a matching packet arrives +// from the whisper network. +func (self *Whisper) Watch(f *Filter) int { + return self.filters.Install(f) +} + +// Unwatch removes an installed message handler. +func (self *Whisper) Unwatch(id int) { + self.filters.Uninstall(id) +} + +// Send injects a message into the whisper send queue, to be distributed in the +// network in the coming cycles. +func (self *Whisper) Send(envelope *Envelope) error { + return self.add(envelope) +} + +// Start implements node.Service, starting the background data propagation thread +// of the Whisper protocol. +func (self *Whisper) Start(*p2p.Server) error { + glog.V(logger.Info).Infoln("Whisper started") + go self.update() + return nil +} + +// Stop implements node.Service, stopping the background data propagation thread +// of the Whisper protocol. +func (self *Whisper) Stop() error { + close(self.quit) + glog.V(logger.Info).Infoln("Whisper stopped") + return nil +} + +// handlePeer is called by the underlying P2P layer when the whisper sub-protocol +// connection is negotiated. +func (self *Whisper) handlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + // Create the new peer and start tracking it + whisperPeer := newPeer(self, peer, rw) + + self.peerMu.Lock() + self.peers[whisperPeer] = struct{}{} + self.peerMu.Unlock() + + defer func() { + self.peerMu.Lock() + delete(self.peers, whisperPeer) + self.peerMu.Unlock() + }() + + // Run the peer handshake and state updates + if err := whisperPeer.handshake(); err != nil { + return err + } + whisperPeer.start() + defer whisperPeer.stop() + + // Read and process inbound messages directly to merge into client-global state + for { + // Fetch the next packet and decode the contained envelopes + packet, err := rw.ReadMsg() + if err != nil { + return err + } + var envelopes []*Envelope + if err := packet.Decode(&envelopes); err != nil { + glog.V(logger.Info).Infof("%v: failed to decode envelope: %v", peer, err) + continue + } + // Inject all envelopes into the internal pool + for _, envelope := range envelopes { + if err := self.add(envelope); err != nil { + // TODO Punish peer here. Invalid envelope. + glog.V(logger.Debug).Infof("%v: failed to pool envelope: %v", peer, err) + } + whisperPeer.mark(envelope) + } + } +} + +// add inserts a new envelope into the message pool to be distributed within the +// whisper network. It also inserts the envelope into the expiration pool at the +// appropriate time-stamp. +func (self *Whisper) add(envelope *Envelope) error { + self.poolMu.Lock() + defer self.poolMu.Unlock() + + // short circuit when a received envelope has already expired + if envelope.Expiry < uint32(time.Now().Unix()) { + return nil + } + + // Insert the message into the tracked pool + hash := envelope.Hash() + if _, ok := self.envelopes[hash]; ok { + glog.V(logger.Detail).Infof("whisper envelope already cached: %x\n", envelope) + return nil + } + self.envelopes[hash] = envelope + + // Insert the message into the expiration pool for later removal + if self.expirations[envelope.Expiry] == nil { + self.expirations[envelope.Expiry] = set.NewNonTS() + } + if !self.expirations[envelope.Expiry].Has(hash) { + self.expirations[envelope.Expiry].Add(hash) + + // Notify the local node of a message arrival + go self.postEvent(envelope) + } + glog.V(logger.Detail).Infof("cached whisper envelope %x\n", envelope) + return nil +} + +// postEvent delivers the message to the watchers. +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() { + // Start a ticker to check for expirations + expire := time.NewTicker(expirationCycle) + + // Repeat updates until termination is requested + for { + select { + case <-expire.C: + self.expire() + + case <-self.quit: + return + } + } +} + +// expire iterates over all the expiration timestamps, removing all stale +// messages from the pools. +func (self *Whisper) expire() { + self.poolMu.Lock() + defer self.poolMu.Unlock() + + now := uint32(time.Now().Unix()) + for then, hashSet := range self.expirations { + // Short circuit if a future time + if then > now { + continue + } + // Dump all expired messages and remove timestamp + hashSet.Each(func(v interface{}) bool { + delete(self.envelopes, v.(common.Hash)) + return true + }) + self.expirations[then].Clear() + } +} + +// envelopes retrieves all the messages currently pooled by the node. +func (self *Whisper) Envelopes() []*Envelope { + self.poolMu.RLock() + defer self.poolMu.RUnlock() + + all := make([]*Envelope, 0, len(self.envelopes)) + for _, envelope := range self.envelopes { + all = append(all, 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) + 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) + } + //} + } + } + return messages +} +*/ \ No newline at end of file