whisper: GetMessages fixed; size restriction updated

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)
+	success := api.whisper.DeleteIdentity(identity)
-	return nil
+	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)

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 {
+func waitForMessages(api *PublicWhisperAPI, id string, target int) []*WhisperMessage {
-	for i := 0; i < 64; i++ {
+	// timeout: 2 seconds
-		all := api.GetMessages(id)
+	result := make([]*WhisperMessage, 0, target)
-		if len(all) >= target {
+	for i := 0; i < 100; i++ {
-			return true
+		mail := api.GetFilterChanges(id)
+		if len(mail) > 0 {
+			for _, m := range mail {
+				result = append(result, m)
 			}
-		time.Sleep(time.Millisecond * 16)
+			if len(result) >= target {
+				break
+			}
+		}
+		time.Sleep(time.Millisecond * 20)
 	}
 
-	// timeout 1024 milliseconds
+	return result
-	return false
 }
 
 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)
+	mail := waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive first message: timeout.")
-	}
-
-	mail := api.GetFilterChanges(id)
 	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)
+	mail = waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive second message: timeout.")
-	}
-
-	mail = api.GetFilterChanges(id)
 	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)
+	mail := waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive first message: timeout.")
-	}
-
-	mail := api.GetFilterChanges(id)
 	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)
+	mail = waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive second message: timeout.")
-	}
-
-	mail = api.GetFilterChanges(id)
 	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)
+	mail := waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive first message: timeout.")
-	}
-
-	mail := api.GetFilterChanges(id)
 	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)
+	mail = waitForMessages(api, id, 1)
-	if !ok {
-		t.Fatalf("failed to receive second message: timeout.")
-	}
-
-	mail = api.GetFilterChanges(id)
 	if len(mail) != 1 {
 		t.Fatalf("failed to GetFilterChanges: got %d messages.", len(mail))
 	}

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

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
+}
 
+func (f *Filter) processEnvelope(env *Envelope) *ReceivedMessage {
+	if f.MatchEnvelope(env) {
+		msg := env.Open(f)
 		if msg != nil {
-		fs.whisper.addDecryptedMessage(msg)
+			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 {

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
 }

whisper/whisperv5/whisper.go

@@ -52,7 +52,6 @@ type Whisper struct {
 	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
 
@@ -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()
+
+	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 {
+		for _, env := range w.envelopes {
-			if filter.MatchMessage(msg) {
+			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) {
+func (s *Statistics) reset() {
-	w.poolMu.Lock()
-	defer w.poolMu.Unlock()
-
-	w.messages[msg.EnvelopeHash] = msg
-}
-
-func (s *Statistics) clear() {
 	s.memoryCleared = 0
 	s.messagesCleared = 0
 }