status-go/wakuv2/common/filter.go

// Copyright 2019 The Waku Library Authors.
//
// The Waku 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 Waku library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty off
// 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 Waku library. If not, see <http://www.gnu.org/licenses/>.
//
// This software uses the go-ethereum library, which is licensed
// under the GNU Lesser General Public Library, version 3 or any later.

package common

import (
	"crypto/ecdsa"
	"fmt"
	"sync"

	"go.uber.org/zap"
	"golang.org/x/exp/maps"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/crypto"
	"github.com/status-im/status-go/logutils"
)

// Filter represents a Waku message filter
type Filter struct {
	Src           *ecdsa.PublicKey  // Sender of the message
	KeyAsym       *ecdsa.PrivateKey // Private Key of recipient
	KeySym        []byte            // Key associated with the Topic
	PubsubTopic   string            // Pubsub topic used to filter messages with
	ContentTopics TopicSet          // ContentTopics to filter messages with
	SymKeyHash    common.Hash       // The Keccak256Hash of the symmetric key, needed for optimization
	id            string            // unique identifier

	Messages MessageStore
}

type FilterSet = map[*Filter]struct{}
type ContentTopicToFilter = map[TopicType]FilterSet
type PubsubTopicToContentTopic = map[string]ContentTopicToFilter

// Filters represents a collection of filters
type Filters struct {
	// Map of random ID to Filter
	watchers map[string]*Filter

	// Pubsub topic to use when no pubsub topic is specified on a filter
	defaultPubsubTopic string

	// map a topic to the filters that are interested in being notified when a message matches that topic
	topicMatcher PubsubTopicToContentTopic

	// list all the filters that will be notified of a new message, no matter what its topic is
	allTopicsMatcher map[*Filter]struct{}

	logger *zap.Logger

	sync.RWMutex
}

// NewFilters returns a newly created filter collection
func NewFilters(defaultPubsubTopic string, logger *zap.Logger) *Filters {
	return &Filters{
		watchers:           make(map[string]*Filter),
		topicMatcher:       make(PubsubTopicToContentTopic),
		allTopicsMatcher:   make(map[*Filter]struct{}),
		defaultPubsubTopic: defaultPubsubTopic,
		logger:             logger,
	}
}

// Install will add a new filter to the filter collection
func (fs *Filters) Install(watcher *Filter) (string, error) {
	if watcher.KeySym != nil && watcher.KeyAsym != nil {
		return "", fmt.Errorf("filters must choose between symmetric and asymmetric keys")
	}

	id, err := GenerateRandomID()
	if err != nil {
		return "", err
	}

	fs.Lock()
	defer fs.Unlock()

	if fs.watchers[id] != nil {
		return "", fmt.Errorf("failed to generate unique ID")
	}

	if watcher.expectsSymmetricEncryption() {
		watcher.SymKeyHash = crypto.Keccak256Hash(watcher.KeySym)
	}

	watcher.id = id

	fs.watchers[id] = watcher
	fs.addTopicMatcher(watcher)

	fs.logger.Debug("filters install", zap.String("id", id))
	return id, err
}

// Uninstall will remove a filter whose id has been specified from
// the filter collection
func (fs *Filters) Uninstall(id string) bool {
	fs.Lock()
	defer fs.Unlock()
	watcher := fs.watchers[id]
	if watcher != nil {
		fs.removeFromTopicMatchers(watcher)
		delete(fs.watchers, id)

		fs.logger.Debug("filters uninstall", zap.String("id", id))
		return true
	}
	return false
}

func (fs *Filters) AllTopics() []TopicType {
	var topics []TopicType
	fs.Lock()
	defer fs.Unlock()
	for _, topicsPerPubsubTopic := range fs.topicMatcher {
		for t := range topicsPerPubsubTopic {
			topics = append(topics, t)
		}
	}

	return topics
}

// addTopicMatcher adds a filter to the topic matchers.
// If the filter's Topics array is empty, it will be tried on every topic.
// Otherwise, it will be tried on the topics specified.
func (fs *Filters) addTopicMatcher(watcher *Filter) {
	if len(watcher.ContentTopics) == 0 && (watcher.PubsubTopic == fs.defaultPubsubTopic || watcher.PubsubTopic == "") {
		fs.allTopicsMatcher[watcher] = struct{}{}
	} else {
		filtersPerContentTopic, ok := fs.topicMatcher[watcher.PubsubTopic]
		if !ok {
			filtersPerContentTopic = make(ContentTopicToFilter)
		}

		for topic := range watcher.ContentTopics {
			if filtersPerContentTopic[topic] == nil {
				filtersPerContentTopic[topic] = make(FilterSet)
			}
			filtersPerContentTopic[topic][watcher] = struct{}{}
		}

		fs.topicMatcher[watcher.PubsubTopic] = filtersPerContentTopic
	}
}

// removeFromTopicMatchers removes a filter from the topic matchers
func (fs *Filters) removeFromTopicMatchers(watcher *Filter) {
	delete(fs.allTopicsMatcher, watcher)

	filtersPerContentTopic, ok := fs.topicMatcher[watcher.PubsubTopic]
	if !ok {
		return
	}

	for topic := range watcher.ContentTopics {
		delete(filtersPerContentTopic[topic], watcher)
	}

	fs.topicMatcher[watcher.PubsubTopic] = filtersPerContentTopic
}

// GetWatchersByTopic returns a slice containing the filters that
// match a specific topic
func (fs *Filters) GetWatchersByTopic(pubsubTopic string, contentTopic TopicType) []*Filter {
	res := make([]*Filter, 0, len(fs.allTopicsMatcher))
	for watcher := range fs.allTopicsMatcher {
		res = append(res, watcher)
	}

	filtersPerContentTopic, ok := fs.topicMatcher[pubsubTopic]
	if !ok {
		return res
	}

	for watcher := range filtersPerContentTopic[contentTopic] {
		res = append(res, watcher)
	}
	return res
}

// Get returns a filter from the collection with a specific ID
func (fs *Filters) Get(id string) *Filter {
	fs.RLock()
	defer fs.RUnlock()
	return fs.watchers[id]
}

func (fs *Filters) All() []*Filter {
	fs.RLock()
	defer fs.RUnlock()
	var filters []*Filter
	for _, f := range fs.watchers {
		filters = append(filters, f)
	}
	return filters
}

func (fs *Filters) GetFilters() map[string]*Filter {
	fs.RLock()
	defer fs.RUnlock()
	return maps.Clone(fs.watchers)
}

// NotifyWatchers notifies any filter that has declared interest
// for the envelope's topic.
func (fs *Filters) NotifyWatchers(recvMessage *ReceivedMessage) bool {
	var decodedMsg *ReceivedMessage

	fs.RLock()
	defer fs.RUnlock()

	var matched bool
	candidates := fs.GetWatchersByTopic(recvMessage.PubsubTopic, recvMessage.ContentTopic)

	if len(candidates) == 0 {
		logutils.ZapLogger().Debug("no filters available for this topic",
			zap.Stringer("message", recvMessage.Hash()),
			zap.String("pubsubTopic", recvMessage.PubsubTopic),
			zap.Stringer("contentTopic", &recvMessage.ContentTopic),
		)
	}

	for _, watcher := range candidates {
		// Messages are decrypted successfully only once
		if decodedMsg == nil {
			decodedMsg = recvMessage.Open(watcher)
			if decodedMsg == nil {
				logutils.ZapLogger().Debug("processing message: failed to open",
					zap.Stringer("message", recvMessage.Hash()),
					zap.String("filter", watcher.id),
				)
				continue
			}
		}

		if watcher.MatchMessage(decodedMsg) {
			matched = true
			logutils.ZapLogger().Debug("processing message: decrypted", zap.Stringer("envelopeHash", recvMessage.Hash()))
			if watcher.Src == nil || IsPubKeyEqual(decodedMsg.Src, watcher.Src) {
				watcher.Trigger(decodedMsg)
			}
		}
	}

	return matched
}

func (f *Filter) expectsAsymmetricEncryption() bool {
	return f.KeyAsym != nil
}

func (f *Filter) expectsSymmetricEncryption() bool {
	return f.KeySym != nil
}

// Trigger adds a yet-unknown message to the filter's list of
// received messages.
func (f *Filter) Trigger(msg *ReceivedMessage) {
	err := f.Messages.Add(msg)
	if err != nil {
		logutils.ZapLogger().Error("failed to add msg into the filters store",
			zap.Stringer("hash", msg.Hash()),
			zap.Error(err),
		)
	}
}

// Retrieve will return the list of all received messages associated
// to a filter.
func (f *Filter) Retrieve() []*ReceivedMessage {
	msgs, err := f.Messages.Pop()
	if err != nil {
		logutils.ZapLogger().Error("failed to retrieve messages from filter store", zap.Error(err))
		return nil
	}
	return msgs
}

// MatchMessage checks if the filter matches an already decrypted
// message (i.e. a Message that has already been handled by
// MatchEnvelope when checked by a previous filter).
// Topics are not checked here, since this is done by topic matchers.
func (f *Filter) MatchMessage(msg *ReceivedMessage) bool {
	if f.expectsAsymmetricEncryption() && msg.isAsymmetricEncryption() {
		return IsPubKeyEqual(&f.KeyAsym.PublicKey, msg.Dst)
	} else if f.expectsSymmetricEncryption() && msg.isSymmetricEncryption() {
		return f.SymKeyHash == msg.SymKeyHash
	} else if !f.expectsAsymmetricEncryption() && !f.expectsSymmetricEncryption() && !msg.isAsymmetricEncryption() && !msg.isSymmetricEncryption() {
		return true
	}
	return false
}