From a02cf9a0ad3c0daf78158cdff78ebb281c933ef4 Mon Sep 17 00:00:00 2001 From: Vlad Date: Tue, 28 Feb 2017 00:02:35 +0100 Subject: [PATCH] whisper: GetMessages fixed; size restriction updated --- whisper/whisperv5/api.go | 10 ++--- whisper/whisperv5/api_test.go | 74 +++++++++++++---------------------- whisper/whisperv5/doc.go | 2 +- whisper/whisperv5/filter.go | 30 ++++++++++---- whisper/whisperv5/message.go | 4 ++ whisper/whisperv5/whisper.go | 42 ++++++++++---------- 6 files changed, 81 insertions(+), 81 deletions(-) diff --git a/whisper/whisperv5/api.go b/whisper/whisperv5/api.go index 9b43f7b706..2e69089502 100644 --- a/whisper/whisperv5/api.go +++ b/whisper/whisperv5/api.go @@ -102,12 +102,12 @@ func (api *PublicWhisperAPI) HasIdentity(identity string) (bool, error) { } // DeleteIdentity deletes the specifies key if it exists. -func (api *PublicWhisperAPI) DeleteIdentity(identity string) error { +func (api *PublicWhisperAPI) DeleteIdentity(identity string) (bool, error) { if api.whisper == nil { - return whisperOffLineErr + return false, whisperOffLineErr } - api.whisper.DeleteIdentity(identity) - return nil + success := api.whisper.DeleteIdentity(identity) + return success, nil } // NewIdentity generates a new cryptographic identity for the client, and injects @@ -352,7 +352,7 @@ func (api *PublicWhisperAPI) Post(args PostArgs) error { log.Error(fmt.Sprintf(err.Error())) return err } - if len(envelope.Data) > MaxMessageLength { + if envelope.size() > MaxMessageLength { info := "Post: message is too big" log.Error(fmt.Sprintf(info)) return errors.New(info) diff --git a/whisper/whisperv5/api_test.go b/whisper/whisperv5/api_test.go index ea0a2c40bc..cde84e991a 100644 --- a/whisper/whisperv5/api_test.go +++ b/whisper/whisperv5/api_test.go @@ -55,10 +55,13 @@ func TestBasic(t *testing.T) { t.Fatalf("failed initial HasIdentity: false positive.") } - err = api.DeleteIdentity(id) + success, err := api.DeleteIdentity(id) if err != nil { t.Fatalf("failed DeleteIdentity: %s.", err) } + if success { + t.Fatalf("deleted non-existing identity: false positive.") + } pub, err := api.NewIdentity() if err != nil { @@ -76,10 +79,13 @@ func TestBasic(t *testing.T) { t.Fatalf("failed HasIdentity: false negative.") } - err = api.DeleteIdentity(pub) + success, err = api.DeleteIdentity(pub) if err != nil { t.Fatalf("failed to delete second identity: %s.", err) } + if !success { + t.Fatalf("failed to delete second identity.") + } exist, err = api.HasIdentity(pub) if err != nil { @@ -257,17 +263,23 @@ func TestUnmarshalPostArgs(t *testing.T) { } } -func waitForMessage(api *PublicWhisperAPI, id string, target int) bool { - for i := 0; i < 64; i++ { - all := api.GetMessages(id) - if len(all) >= target { - return true +func waitForMessages(api *PublicWhisperAPI, id string, target int) []*WhisperMessage { + // timeout: 2 seconds + result := make([]*WhisperMessage, 0, target) + for i := 0; i < 100; i++ { + mail := api.GetFilterChanges(id) + if len(mail) > 0 { + for _, m := range mail { + result = append(result, m) + } + if len(result) >= target { + break + } } - time.Sleep(time.Millisecond * 16) + time.Sleep(time.Millisecond * 20) } - // timeout 1024 milliseconds - return false + return result } func TestIntegrationAsym(t *testing.T) { @@ -334,12 +346,7 @@ func TestIntegrationAsym(t *testing.T) { t.Errorf("failed to post message: %s.", err) } - ok := waitForMessage(api, id, 1) - if !ok { - t.Fatalf("failed to receive first message: timeout.") - } - - mail := api.GetFilterChanges(id) + mail := waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed to GetFilterChanges: got %d messages.", len(mail)) } @@ -356,12 +363,7 @@ func TestIntegrationAsym(t *testing.T) { t.Fatalf("failed to post next message: %s.", err) } - ok = waitForMessage(api, id, 2) - if !ok { - t.Fatalf("failed to receive second message: timeout.") - } - - mail = api.GetFilterChanges(id) + mail = waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed to GetFilterChanges: got %d messages.", len(mail)) } @@ -434,12 +436,7 @@ func TestIntegrationSym(t *testing.T) { t.Fatalf("failed to post first message: %s.", err) } - ok := waitForMessage(api, id, 1) - if !ok { - t.Fatalf("failed to receive first message: timeout.") - } - - mail := api.GetFilterChanges(id) + mail := waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed GetFilterChanges: got %d messages.", len(mail)) } @@ -456,12 +453,7 @@ func TestIntegrationSym(t *testing.T) { t.Fatalf("failed to post second message: %s.", err) } - ok = waitForMessage(api, id, 2) - if !ok { - t.Fatalf("failed to receive second message: timeout.") - } - - mail = api.GetFilterChanges(id) + mail = waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed second GetFilterChanges: got %d messages.", len(mail)) } @@ -534,12 +526,7 @@ func TestIntegrationSymWithFilter(t *testing.T) { t.Fatalf("failed to post message: %s.", err) } - ok := waitForMessage(api, id, 1) - if !ok { - t.Fatalf("failed to receive first message: timeout.") - } - - mail := api.GetFilterChanges(id) + mail := waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed to GetFilterChanges: got %d messages.", len(mail)) } @@ -556,12 +543,7 @@ func TestIntegrationSymWithFilter(t *testing.T) { t.Fatalf("failed to post next message: %s.", err) } - ok = waitForMessage(api, id, 2) - if !ok { - t.Fatalf("failed to receive second message: timeout.") - } - - mail = api.GetFilterChanges(id) + mail = waitForMessages(api, id, 1) if len(mail) != 1 { t.Fatalf("failed to GetFilterChanges: got %d messages.", len(mail)) } diff --git a/whisper/whisperv5/doc.go b/whisper/whisperv5/doc.go index 70c7008a73..9e533010b0 100644 --- a/whisper/whisperv5/doc.go +++ b/whisper/whisperv5/doc.go @@ -55,7 +55,7 @@ const ( saltLength = 12 AESNonceMaxLength = 12 - MaxMessageLength = 0x0FFFFF // todo: remove this restriction after testing. this should be regulated by PoW. + MaxMessageLength = 1024 * 1024 MinimumPoW = 10.0 // todo: review after testing. padSizeLimitLower = 128 // it can not be less - we don't want to reveal the absence of signature diff --git a/whisper/whisperv5/filter.go b/whisper/whisperv5/filter.go index ffa5ae9463..40c2ebbc1e 100644 --- a/whisper/whisperv5/filter.go +++ b/whisper/whisperv5/filter.go @@ -102,11 +102,16 @@ func (fs *Filters) Get(id string) *Filter { } func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { - fs.mutex.RLock() + var j int var msg *ReceivedMessage - for j, watcher := range fs.watchers { + + fs.mutex.RLock() + defer fs.mutex.RUnlock() + + for _, watcher := range fs.watchers { + j++ if p2pMessage && !watcher.AcceptP2P { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: p2p messages are not allowed", env.Hash(), j)) + log.Trace(fmt.Sprintf("msg [%x], filter [%d]: p2p messages are not allowed", env.Hash(), j)) continue } @@ -118,10 +123,10 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { if match { msg = env.Open(watcher) if msg == nil { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: failed to open", env.Hash(), j)) + log.Trace(fmt.Sprintf("msg [%x], filter [%d]: failed to open", env.Hash(), j)) } } else { - log.Trace(fmt.Sprintf("msg [%x], filter [%s]: does not match", env.Hash(), j)) + log.Trace(fmt.Sprintf("msg [%x], filter [%d]: does not match", env.Hash(), j)) } } @@ -129,11 +134,20 @@ func (fs *Filters) NotifyWatchers(env *Envelope, p2pMessage bool) { watcher.Trigger(msg) } } - fs.mutex.RUnlock() // we need to unlock before calling addDecryptedMessage +} - if msg != nil { - fs.whisper.addDecryptedMessage(msg) +func (f *Filter) processEnvelope(env *Envelope) *ReceivedMessage { + if f.MatchEnvelope(env) { + msg := env.Open(f) + if msg != nil { + return msg + } else { + log.Trace(fmt.Sprintf("processing msg [%x]: failed to open", env.Hash())) + } + } else { + log.Trace(fmt.Sprintf("processing msg [%x]: does not match", env.Hash())) } + return nil } func (f *Filter) expectsAsymmetricEncryption() bool { diff --git a/whisper/whisperv5/message.go b/whisper/whisperv5/message.go index 9677f278e5..995cfebbef 100644 --- a/whisper/whisperv5/message.go +++ b/whisper/whisperv5/message.go @@ -258,6 +258,10 @@ func (msg *SentMessage) Wrap(options *MessageParams) (envelope *Envelope, err er if err != nil { return nil, err } + if envelope.size() > MaxMessageLength { + log.Error(fmt.Sprintf("Envelope size must not exceed %d bytes", MaxMessageLength)) + return nil, errors.New("Oversized message") + } return envelope, nil } diff --git a/whisper/whisperv5/whisper.go b/whisper/whisperv5/whisper.go index 5062f7b6bc..c4f2b237d5 100644 --- a/whisper/whisperv5/whisper.go +++ b/whisper/whisperv5/whisper.go @@ -51,10 +51,9 @@ type Whisper struct { symKeys map[string][]byte keyMu sync.RWMutex - envelopes map[common.Hash]*Envelope // Pool of envelopes currently tracked by this node - messages map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages, which are not expired yet - expirations map[uint32]*set.SetNonTS // Message expiration pool - poolMu sync.RWMutex // Mutex to sync the message and expiration pools + envelopes map[common.Hash]*Envelope // Pool of envelopes currently tracked by this node + expirations map[uint32]*set.SetNonTS // Message expiration pool + 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 @@ -78,7 +77,6 @@ func New() *Whisper { privateKeys: make(map[string]*ecdsa.PrivateKey), symKeys: make(map[string][]byte), envelopes: make(map[common.Hash]*Envelope), - messages: make(map[common.Hash]*ReceivedMessage), expirations: make(map[uint32]*set.SetNonTS), peers: make(map[*Peer]struct{}), messageQueue: make(chan *Envelope, messageQueueLimit), @@ -188,10 +186,15 @@ func (w *Whisper) NewIdentity() *ecdsa.PrivateKey { } // DeleteIdentity deletes the specified key if it exists. -func (w *Whisper) DeleteIdentity(key string) { +func (w *Whisper) DeleteIdentity(key string) bool { w.keyMu.Lock() defer w.keyMu.Unlock() - delete(w.privateKeys, key) + + if w.privateKeys[key] != nil { + delete(w.privateKeys, key) + return true + } + return false } // HasIdentity checks if the the whisper node is configured with the private key @@ -355,6 +358,9 @@ func (wh *Whisper) runMessageLoop(p *Peer, rw p2p.MsgReadWriter) error { if err != nil { return err } + if packet.Size > MaxMessageLength { + return fmt.Errorf("oversized message received") + } switch packet.Code { case statusCode: @@ -435,7 +441,7 @@ func (wh *Whisper) add(envelope *Envelope) (bool, error) { } } - if len(envelope.Data) > MaxMessageLength { + if envelope.size() > MaxMessageLength { return false, fmt.Errorf("huge messages are not allowed [%x]", envelope.Hash()) } @@ -558,7 +564,7 @@ func (w *Whisper) expire() { w.poolMu.Lock() defer w.poolMu.Unlock() - w.stats.clear() + w.stats.reset() now := uint32(time.Now().Unix()) for expiry, hashSet := range w.expirations { if expiry < now { @@ -570,7 +576,6 @@ func (w *Whisper) expire() { w.stats.memoryCleared += sz w.stats.totalMemoryUsed -= sz delete(w.envelopes, v.(common.Hash)) - delete(w.messages, v.(common.Hash)) return true }) w.expirations[expiry].Clear() @@ -596,15 +601,17 @@ func (w *Whisper) Envelopes() []*Envelope { return all } -// Messages retrieves all the decrypted messages matching a filter id. +// Messages iterates through all currently floating envelopes +// and retrieves all the messages, that this filter could decrypt. func (w *Whisper) Messages(id string) []*ReceivedMessage { result := make([]*ReceivedMessage, 0) w.poolMu.RLock() defer w.poolMu.RUnlock() if filter := w.filters.Get(id); filter != nil { - for _, msg := range w.messages { - if filter.MatchMessage(msg) { + for _, env := range w.envelopes { + msg := filter.processEnvelope(env) + if msg != nil { result = append(result, msg) } } @@ -620,14 +627,7 @@ func (w *Whisper) isEnvelopeCached(hash common.Hash) bool { return exist } -func (w *Whisper) addDecryptedMessage(msg *ReceivedMessage) { - w.poolMu.Lock() - defer w.poolMu.Unlock() - - w.messages[msg.EnvelopeHash] = msg -} - -func (s *Statistics) clear() { +func (s *Statistics) reset() { s.memoryCleared = 0 s.messagesCleared = 0 }