package protocol import ( "context" "crypto/ecdsa" "database/sql" "encoding/hex" "encoding/json" "math/rand" "sync" "time" "github.com/pkg/errors" "go.uber.org/zap" "github.com/golang/protobuf/proto" "github.com/status-im/status-go/eth-node/crypto" "github.com/status-im/status-go/eth-node/types" enstypes "github.com/status-im/status-go/eth-node/types/ens" "github.com/status-im/status-go/protocol/encryption" "github.com/status-im/status-go/protocol/encryption/multidevice" "github.com/status-im/status-go/protocol/encryption/sharedsecret" "github.com/status-im/status-go/protocol/identity/alias" "github.com/status-im/status-go/protocol/identity/identicon" "github.com/status-im/status-go/protocol/protobuf" "github.com/status-im/status-go/protocol/sqlite" transport "github.com/status-im/status-go/protocol/transport/whisper" v1protocol "github.com/status-im/status-go/protocol/v1" ) const PubKeyStringLength = 132 var ( ErrChatIDEmpty = errors.New("chat ID is empty") ErrNotImplemented = errors.New("not implemented") errChatNotFound = errors.New("chat not found") ) // Messenger is a entity managing chats and messages. // It acts as a bridge between the application and encryption // layers. // It needs to expose an interface to manage installations // because installations are managed by the user. // Similarly, it needs to expose an interface to manage // mailservers because they can also be managed by the user. type Messenger struct { node types.Node identity *ecdsa.PrivateKey persistence *sqlitePersistence transport *transport.WhisperServiceTransport encryptor *encryption.Protocol processor *messageProcessor logger *zap.Logger featureFlags featureFlags messagesPersistenceEnabled bool shutdownTasks []func() error systemMessagesTranslations map[protobuf.MembershipUpdateEvent_EventType]string allChats map[string]*Chat allContacts map[string]*Contact mutex sync.Mutex } type RawResponse struct { Filter *transport.Filter `json:"filter"` Messages []*v1protocol.StatusMessage `json:"messages"` } type MessengerResponse struct { Chats []*Chat `json:"chats,omitEmpty"` Messages []*Message `json:"messages,omitEmpty"` Contacts []*Contact `json:"contacts,omitEmpty"` // Raw unprocessed messages RawMessages []*RawResponse `json:"rawMessages,omitEmpty"` } func (m *MessengerResponse) IsEmpty() bool { return len(m.Chats) == 0 && len(m.Messages) == 0 && len(m.Contacts) == 0 && len(m.RawMessages) == 0 } type featureFlags struct { // datasync indicates whether direct messages should be sent exclusively // using datasync, breaking change for non-v1 clients. Public messages // are not impacted datasync bool } type dbConfig struct { dbPath string dbKey string } type config struct { onNewInstallationsHandler func([]*multidevice.Installation) // This needs to be exposed until we move here mailserver logic // as otherwise the client is not notified of a new filter and // won't be pulling messages from mailservers until it reloads the chats/filters onNegotiatedFilters func([]*transport.Filter) // DEPRECATED: no need to expose it onSendContactCodeHandler func(*encryption.ProtocolMessageSpec) // systemMessagesTranslations holds translations for system-messages systemMessagesTranslations map[protobuf.MembershipUpdateEvent_EventType]string // Config for the envelopes monitor envelopesMonitorConfig *transport.EnvelopesMonitorConfig messagesPersistenceEnabled bool featureFlags featureFlags // A path to a database or a database instance is required. // The database instance has a higher priority. dbConfig dbConfig db *sql.DB logger *zap.Logger } type Option func(*config) error func WithOnNewInstallationsHandler(h func([]*multidevice.Installation)) Option { return func(c *config) error { c.onNewInstallationsHandler = h return nil } } func WithSystemMessagesTranslations(t map[protobuf.MembershipUpdateEvent_EventType]string) Option { return func(c *config) error { c.systemMessagesTranslations = t return nil } } func WithOnNegotiatedFilters(h func([]*transport.Filter)) Option { return func(c *config) error { c.onNegotiatedFilters = h return nil } } func WithCustomLogger(logger *zap.Logger) Option { return func(c *config) error { c.logger = logger return nil } } func WithMessagesPersistenceEnabled() Option { return func(c *config) error { c.messagesPersistenceEnabled = true return nil } } func WithDatabaseConfig(dbPath, dbKey string) Option { return func(c *config) error { c.dbConfig = dbConfig{dbPath: dbPath, dbKey: dbKey} return nil } } func WithDatabase(db *sql.DB) Option { return func(c *config) error { c.db = db return nil } } func WithDatasync() func(c *config) error { return func(c *config) error { c.featureFlags.datasync = true return nil } } func WithEnvelopesMonitorConfig(emc *transport.EnvelopesMonitorConfig) Option { return func(c *config) error { c.envelopesMonitorConfig = emc return nil } } func NewMessenger( identity *ecdsa.PrivateKey, node types.Node, installationID string, opts ...Option, ) (*Messenger, error) { var messenger *Messenger shh, err := node.GetWhisper(nil) if err != nil { return nil, err } c := config{} for _, opt := range opts { if err := opt(&c); err != nil { return nil, err } } logger := c.logger if c.logger == nil { var err error if logger, err = zap.NewDevelopment(); err != nil { return nil, errors.Wrap(err, "failed to create a logger") } } // Set default config fields. if c.onNewInstallationsHandler == nil { c.onNewInstallationsHandler = func(installations []*multidevice.Installation) { sugar := logger.Sugar().With("site", "onNewInstallationsHandler") for _, installation := range installations { sugar.Infow( "received a new installation", "identity", installation.Identity, "id", installation.ID) } } } onNewSharedSecretHandler := func(secrets []*sharedsecret.Secret) { filters, err := messenger.handleSharedSecrets(secrets) if err != nil { slogger := logger.With(zap.String("site", "onNewSharedSecretHandler")) slogger.Warn("failed to process secrets", zap.Error(err)) } if c.onNegotiatedFilters != nil { c.onNegotiatedFilters(filters) } } if c.onSendContactCodeHandler == nil { c.onSendContactCodeHandler = func(messageSpec *encryption.ProtocolMessageSpec) { slogger := logger.With(zap.String("site", "onSendContactCodeHandler")) slogger.Debug("received a SendContactCode request") newMessage, err := messageSpecToWhisper(messageSpec) if err != nil { slogger.Warn("failed to convert spec to Whisper message", zap.Error(err)) return } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() chatName := transport.ContactCodeTopic(&messenger.identity.PublicKey) _, err = messenger.transport.SendPublic(ctx, newMessage, chatName) if err != nil { slogger.Warn("failed to send a contact code", zap.Error(err)) } } } if c.systemMessagesTranslations == nil { c.systemMessagesTranslations = defaultSystemMessagesTranslations } // Configure the database. database := c.db if c.db == nil && c.dbConfig == (dbConfig{}) { return nil, errors.New("database instance or database path needs to be provided") } if c.db == nil { logger.Info("opening a database", zap.String("dbPath", c.dbConfig.dbPath)) var err error database, err = sqlite.Open(c.dbConfig.dbPath, c.dbConfig.dbKey) if err != nil { return nil, errors.Wrap(err, "failed to initialize database from the db config") } } // Apply migrations for all components. err = sqlite.Migrate(database) if err != nil { return nil, errors.Wrap(err, "failed to apply migrations") } // Initialize transport layer. t, err := transport.NewWhisperServiceTransport( shh, identity, database, nil, c.envelopesMonitorConfig, logger, ) if err != nil { return nil, errors.Wrap(err, "failed to create a WhisperServiceTransport") } // Initialize encryption layer. encryptionProtocol := encryption.New( database, installationID, c.onNewInstallationsHandler, onNewSharedSecretHandler, c.onSendContactCodeHandler, logger, ) processor, err := newMessageProcessor( identity, database, encryptionProtocol, t, logger, c.featureFlags, ) if err != nil { return nil, errors.Wrap(err, "failed to create messageProcessor") } messenger = &Messenger{ node: node, identity: identity, persistence: &sqlitePersistence{db: database}, transport: t, encryptor: encryptionProtocol, processor: processor, featureFlags: c.featureFlags, systemMessagesTranslations: c.systemMessagesTranslations, allChats: make(map[string]*Chat), allContacts: make(map[string]*Contact), messagesPersistenceEnabled: c.messagesPersistenceEnabled, shutdownTasks: []func() error{ database.Close, t.Reset, t.Stop, func() error { processor.Stop(); return nil }, // Currently this often fails, seems like it's safe to ignore them // https://github.com/uber-go/zap/issues/328 func() error { _ = logger.Sync; return nil }, }, logger: logger, } // Start all services immediately. // TODO: consider removing identity as an argument to Start(). if err := encryptionProtocol.Start(identity); err != nil { return nil, err } logger.Debug("messages persistence", zap.Bool("enabled", c.messagesPersistenceEnabled)) return messenger, nil } // Init analyzes chats and contacts in order to setup filters // which are responsible for retrieving messages. func (m *Messenger) Init() error { m.mutex.Lock() defer m.mutex.Unlock() // Seed the for color generation rand.Seed(time.Now().Unix()) logger := m.logger.With(zap.String("site", "Init")) var ( publicChatIDs []string publicKeys []*ecdsa.PublicKey ) // Get chat IDs and public keys from the existing chats. // TODO: Get only active chats by the query. chats, err := m.persistence.Chats() if err != nil { return err } for _, chat := range chats { m.allChats[chat.ID] = chat if !chat.Active { continue } switch chat.ChatType { case ChatTypePublic: publicChatIDs = append(publicChatIDs, chat.ID) case ChatTypeOneToOne: pk, err := chat.PublicKey() if err != nil { return err } publicKeys = append(publicKeys, pk) case ChatTypePrivateGroupChat: for _, member := range chat.Members { publicKey, err := member.PublicKey() if err != nil { return errors.Wrapf(err, "invalid public key for member %s in chat %s", member.ID, chat.Name) } publicKeys = append(publicKeys, publicKey) } default: return errors.New("invalid chat type") } } // Get chat IDs and public keys from the contacts. contacts, err := m.persistence.Contacts() if err != nil { return err } for _, contact := range contacts { m.allContacts[contact.ID] = contact // We only need filters for contacts added by us and not blocked. if !contact.IsAdded() || contact.IsBlocked() { continue } publicKey, err := contact.PublicKey() if err != nil { logger.Error("failed to get contact's public key", zap.Error(err)) continue } publicKeys = append(publicKeys, publicKey) } _, err = m.transport.InitFilters(publicChatIDs, publicKeys) return err } // Shutdown takes care of ensuring a clean shutdown of Messenger func (m *Messenger) Shutdown() (err error) { for _, task := range m.shutdownTasks { if tErr := task(); tErr != nil { if err == nil { // First error appeared. err = tErr } else { // We return all errors. They will be concatenated in the order of occurrence, // however, they will also be returned as a single error. err = errors.Wrap(err, tErr.Error()) } } } return } func (m *Messenger) handleSharedSecrets(secrets []*sharedsecret.Secret) ([]*transport.Filter, error) { logger := m.logger.With(zap.String("site", "handleSharedSecrets")) var result []*transport.Filter for _, secret := range secrets { logger.Debug("received shared secret", zap.Binary("identity", crypto.FromECDSAPub(secret.Identity))) fSecret := types.NegotiatedSecret{ PublicKey: secret.Identity, Key: secret.Key, } filter, err := m.transport.ProcessNegotiatedSecret(fSecret) if err != nil { return nil, err } result = append(result, filter) } return result, nil } func (m *Messenger) EnableInstallation(id string) error { return m.encryptor.EnableInstallation(&m.identity.PublicKey, id) } func (m *Messenger) DisableInstallation(id string) error { return m.encryptor.DisableInstallation(&m.identity.PublicKey, id) } func (m *Messenger) Installations() ([]*multidevice.Installation, error) { return m.encryptor.GetOurInstallations(&m.identity.PublicKey) } func (m *Messenger) SetInstallationMetadata(id string, data *multidevice.InstallationMetadata) error { return m.encryptor.SetInstallationMetadata(&m.identity.PublicKey, id, data) } // NOT IMPLEMENTED func (m *Messenger) SelectMailserver(id string) error { return ErrNotImplemented } // NOT IMPLEMENTED func (m *Messenger) AddMailserver(enode string) error { return ErrNotImplemented } // NOT IMPLEMENTED func (m *Messenger) RemoveMailserver(id string) error { return ErrNotImplemented } // NOT IMPLEMENTED func (m *Messenger) Mailservers() ([]string, error) { return nil, ErrNotImplemented } func (m *Messenger) Join(chat Chat) error { switch chat.ChatType { case ChatTypeOneToOne: pk, err := chat.PublicKey() if err != nil { return err } return m.transport.JoinPrivate(pk) case ChatTypePrivateGroupChat: members, err := chat.MembersAsPublicKeys() if err != nil { return err } return m.transport.JoinGroup(members) case ChatTypePublic: return m.transport.JoinPublic(chat.Name) default: return errors.New("chat is neither public nor private") } } // This is not accurate, it should not leave transport on removal of chat/group // only once there is no more: Group chat with that member, one-to-one chat, contact added by us func (m *Messenger) Leave(chat Chat) error { switch chat.ChatType { case ChatTypeOneToOne: pk, err := chat.PublicKey() if err != nil { return err } return m.transport.LeavePrivate(pk) case ChatTypePrivateGroupChat: members, err := chat.MembersAsPublicKeys() if err != nil { return err } return m.transport.LeaveGroup(members) case ChatTypePublic: return m.transport.LeavePublic(chat.Name) default: return errors.New("chat is neither public nor private") } } func (m *Messenger) CreateGroupChatWithMembers(ctx context.Context, name string, members []string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse logger := m.logger.With(zap.String("site", "CreateGroupChatWithMembers")) logger.Info("Creating group chat", zap.String("name", name), zap.Any("members", members)) chat := createGroupChat() group, err := v1protocol.NewGroupWithCreator(name, m.identity) if err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) // Add members event := v1protocol.NewMembersAddedEvent(members, group.NextClockValue()) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } recipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, recipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) response.Chats = []*Chat{&chat} response.Messages = buildSystemMessages(chat.MembershipUpdates, m.systemMessagesTranslations) return &response, m.saveChat(&chat) } func (m *Messenger) RemoveMemberFromGroupChat(ctx context.Context, chatID string, member string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse logger := m.logger.With(zap.String("site", "RemoveMemberFromGroupChat")) logger.Info("Removing member form group chat", zap.String("chatID", chatID), zap.String("member", member)) chat, ok := m.allChats[chatID] if !ok { return nil, errors.New("can't find chat") } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } // We save the initial recipients as we want to send updates to also // the members kicked out oldRecipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } // Remove member event := v1protocol.NewMemberRemovedEvent(member, group.NextClockValue()) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, oldRecipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) response.Chats = []*Chat{chat} response.Messages = buildSystemMessages(chat.MembershipUpdates, m.systemMessagesTranslations) return &response, m.saveChat(chat) } func (m *Messenger) AddMembersToGroupChat(ctx context.Context, chatID string, members []string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse logger := m.logger.With(zap.String("site", "AddMembersFromGroupChat")) logger.Info("Adding members form group chat", zap.String("chatID", chatID), zap.Any("members", members)) chat, ok := m.allChats[chatID] if !ok { return nil, errors.New("can't find chat") } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } // Add members event := v1protocol.NewMembersAddedEvent(members, group.NextClockValue()) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } recipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, recipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) response.Chats = []*Chat{chat} response.Messages = buildSystemMessages([]v1protocol.MembershipUpdateEvent{event}, m.systemMessagesTranslations) return &response, m.saveChat(chat) } func (m *Messenger) AddAdminsToGroupChat(ctx context.Context, chatID string, members []string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse logger := m.logger.With(zap.String("site", "AddAdminsToGroupChat")) logger.Info("Add admins to group chat", zap.String("chatID", chatID), zap.Any("members", members)) chat, ok := m.allChats[chatID] if !ok { return nil, errors.New("can't find chat") } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } // Add members event := v1protocol.NewAdminsAddedEvent(members, group.NextClockValue()) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } recipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, recipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) response.Chats = []*Chat{chat} response.Messages = buildSystemMessages([]v1protocol.MembershipUpdateEvent{event}, m.systemMessagesTranslations) return &response, m.saveChat(chat) } func (m *Messenger) ConfirmJoiningGroup(ctx context.Context, chatID string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse chat, ok := m.allChats[chatID] if !ok { return nil, errors.New("can't find chat") } err := m.Join(*chat) if err != nil { return nil, err } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } event := v1protocol.NewMemberJoinedEvent( group.NextClockValue(), ) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } recipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, recipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) response.Chats = []*Chat{chat} response.Messages = buildSystemMessages([]v1protocol.MembershipUpdateEvent{event}, m.systemMessagesTranslations) return &response, m.saveChat(chat) } func (m *Messenger) LeaveGroupChat(ctx context.Context, chatID string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() var response MessengerResponse chat, ok := m.allChats[chatID] if !ok { return nil, errors.New("can't find chat") } err := m.Leave(*chat) if err != nil { return nil, err } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } event := v1protocol.NewMemberRemovedEvent( types.EncodeHex(crypto.FromECDSAPub(&m.identity.PublicKey)), group.NextClockValue(), ) event.ChatID = chat.ID err = event.Sign(m.identity) if err != nil { return nil, err } err = group.ProcessEvent(event) if err != nil { return nil, err } recipients, err := stringSliceToPublicKeys(group.Members(), true) if err != nil { return nil, err } if _, err := m.propagateMembershipUpdates(ctx, group, recipients, nil); err != nil { return nil, err } chat.updateChatFromProtocolGroup(group) chat.Active = false response.Chats = []*Chat{chat} response.Messages = buildSystemMessages([]v1protocol.MembershipUpdateEvent{event}, m.systemMessagesTranslations) return &response, m.saveChat(chat) } func (m *Messenger) propagateMembershipUpdates(ctx context.Context, group *v1protocol.Group, recipients []*ecdsa.PublicKey, chatMessage *protobuf.ChatMessage) ([]byte, error) { hasPairedDevices, err := m.hasPairedDevices() if err != nil { return nil, err } if !hasPairedDevices { // Filter out my key from the recipients n := 0 for _, recipient := range recipients { if !isPubKeyEqual(recipient, &m.identity.PublicKey) { recipients[n] = recipient n++ } } recipients = recipients[:n] } // Finally send membership updates to all recipients. return m.processor.SendMembershipUpdate( ctx, recipients, group, chatMessage, ) } func (m *Messenger) saveChat(chat *Chat) error { err := m.persistence.SaveChat(*chat) if err != nil { return err } m.allChats[chat.ID] = chat return nil } func (m *Messenger) saveChats(chats []*Chat) error { err := m.persistence.SaveChats(chats) if err != nil { return err } for _, chat := range chats { m.allChats[chat.ID] = chat } return nil } func (m *Messenger) SaveChat(chat *Chat) error { m.mutex.Lock() defer m.mutex.Unlock() err := m.saveChat(chat) if err != nil { return err } return nil } func (m *Messenger) Chats() []*Chat { m.mutex.Lock() defer m.mutex.Unlock() var chats []*Chat for _, c := range m.allChats { chats = append(chats, c) } return chats } func (m *Messenger) DeleteChat(chatID string) error { m.mutex.Lock() defer m.mutex.Unlock() err := m.persistence.DeleteChat(chatID) if err != nil { return err } delete(m.allChats, chatID) return nil } func (m *Messenger) SaveContact(contact *Contact) error { m.mutex.Lock() defer m.mutex.Unlock() identicon, err := identicon.GenerateBase64(contact.ID) if err != nil { return err } contact.Identicon = identicon name, err := alias.GenerateFromPublicKeyString(contact.ID) if err != nil { return err } contact.Alias = name err = m.persistence.SaveContact(contact, nil) if err != nil { return err } m.allContacts[contact.ID] = contact return nil } func (m *Messenger) BlockContact(contact *Contact) ([]*Chat, error) { m.mutex.Lock() defer m.mutex.Unlock() chats, err := m.persistence.BlockContact(contact) if err != nil { return nil, err } m.allContacts[contact.ID] = contact for _, chat := range chats { m.allChats[chat.ID] = chat } delete(m.allChats, contact.ID) return chats, nil } func (m *Messenger) Contacts() []*Contact { m.mutex.Lock() defer m.mutex.Unlock() var contacts []*Contact for _, contact := range m.allContacts { contacts = append(contacts, contact) } return contacts } func timestampInMs() uint64 { return uint64(time.Now().UnixNano() / int64(time.Millisecond)) } // ReSendChatMessage pulls a message from the database and sends it again func (m *Messenger) ReSendChatMessage(ctx context.Context, messageID string) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() logger := m.logger.With(zap.String("site", "ReSendChatMessage")) var response MessengerResponse message, err := m.persistence.MessageByID(messageID) if err != nil { return nil, err } if message == nil { return nil, errors.New("message not found") } if message.RawPayload == nil { return nil, errors.New("message payload not found, can't resend message") } chat, ok := m.allChats[message.LocalChatID] if !ok { return nil, errors.New("chat not found") } switch chat.ChatType { case ChatTypeOneToOne: publicKey, err := chat.PublicKey() if err != nil { return nil, err } logger.Debug("re-sending private message") id, err := m.processor.SendPrivateRaw(ctx, publicKey, message.RawPayload, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } message.ID = "0x" + hex.EncodeToString(id) err = m.sendToPairedDevices(ctx, message.RawPayload, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } case ChatTypePublic: logger.Debug("re-sending public message", zap.String("chatName", chat.Name)) id, err := m.processor.SendPublicRaw(ctx, chat.ID, message.RawPayload, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } message.ID = "0x" + hex.EncodeToString(id) case ChatTypePrivateGroupChat: logger.Debug("re-sending group message", zap.String("chatName", chat.Name)) recipients, err := chat.MembersAsPublicKeys() if err != nil { return nil, err } n := 0 for _, item := range recipients { if !isPubKeyEqual(item, &m.identity.PublicKey) { recipients[n] = item n++ } } id, err := m.processor.SendGroupRaw(ctx, recipients[:n], message.RawPayload, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } message.ID = "0x" + hex.EncodeToString(id) err = m.sendToPairedDevices(ctx, message.RawPayload, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } default: return nil, errors.New("chat type not supported") } response.Messages = []*Message{message} response.Chats = []*Chat{chat} return &response, nil } func (m *Messenger) hasPairedDevices() (bool, error) { activeInstallations, err := m.encryptor.GetOurActiveInstallations(&m.identity.PublicKey) if err != nil { return false, err } return len(activeInstallations) > 1, nil } // sendToPairedDevices will check if we have any paired devices and send to them if necessary func (m *Messenger) sendToPairedDevices(ctx context.Context, payload []byte, messageType protobuf.ApplicationMetadataMessage_Type) error { hasPairedDevices, err := m.hasPairedDevices() if err != nil { return err } // We send a message to any paired device if hasPairedDevices { _, err := m.processor.SendPrivateRaw(ctx, &m.identity.PublicKey, payload, messageType) if err != nil { return err } } return nil } // SendChatMessage takes a minimal message and sends it based on the corresponding chat func (m *Messenger) SendChatMessage(ctx context.Context, message *Message) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() logger := m.logger.With(zap.String("site", "Send"), zap.String("chatID", message.ChatId)) var response MessengerResponse // A valid added chat is required. chat, ok := m.allChats[message.ChatId] if !ok { return nil, errors.New("Chat not found") } clock := chat.LastClockValue timestamp := timestampInMs() if clock == 0 || clock < timestamp { clock = timestamp } else { clock = clock + 1 } message.LocalChatID = chat.ID message.Clock = clock message.Timestamp = timestamp message.From = "0x" + hex.EncodeToString(crypto.FromECDSAPub(&m.identity.PublicKey)) message.SigPubKey = &m.identity.PublicKey message.WhisperTimestamp = timestamp message.Seen = true message.OutgoingStatus = OutgoingStatusSending identicon, err := identicon.GenerateBase64(message.From) if err != nil { return nil, err } message.Identicon = identicon alias, err := alias.GenerateFromPublicKeyString(message.From) if err != nil { return nil, err } message.Alias = alias switch chat.ChatType { case ChatTypeOneToOne: publicKey, err := chat.PublicKey() if err != nil { return nil, err } logger.Debug("sending private message") message.MessageType = protobuf.ChatMessage_ONE_TO_ONE encodedMessage, err := proto.Marshal(message) if err != nil { return nil, err } message.RawPayload = encodedMessage id, err := m.processor.SendPrivateRaw(ctx, publicKey, encodedMessage, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } message.ID = "0x" + hex.EncodeToString(id) err = m.sendToPairedDevices(ctx, encodedMessage, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } case ChatTypePublic: logger.Debug("sending public message", zap.String("chatName", chat.Name)) message.MessageType = protobuf.ChatMessage_PUBLIC_GROUP encodedMessage, err := proto.Marshal(message) if err != nil { return nil, err } message.RawPayload = encodedMessage id, err := m.processor.SendPublicRaw(ctx, chat.ID, encodedMessage, protobuf.ApplicationMetadataMessage_CHAT_MESSAGE) if err != nil { return nil, err } message.ID = "0x" + hex.EncodeToString(id) case ChatTypePrivateGroupChat: logger.Debug("sending public message", zap.String("chatName", chat.Name)) message.MessageType = protobuf.ChatMessage_PRIVATE_GROUP logger.Debug("sending group message", zap.String("chatName", chat.Name)) recipients, err := chat.MembersAsPublicKeys() if err != nil { return nil, err } group, err := newProtocolGroupFromChat(chat) if err != nil { return nil, err } encodedMessage, err := m.propagateMembershipUpdates(ctx, group, recipients, &message.ChatMessage) if err != nil { return nil, err } id := v1protocol.MessageID(&m.identity.PublicKey, encodedMessage) message.ID = "0x" + hex.EncodeToString(id) default: return nil, errors.New("chat type not supported") } err = message.PrepareContent() if err != nil { return nil, err } jsonMessage, err := json.Marshal(message) if err != nil { return nil, err } chat.LastClockValue = clock chat.LastMessage = jsonMessage chat.Timestamp = int64(timestamp) err = m.persistence.SaveMessagesLegacy([]*Message{message}) if err != nil { return nil, err } response.Chats = []*Chat{chat} response.Messages = []*Message{message} return &response, m.saveChat(chat) } // SendRaw takes encoded data, encrypts it and sends through the wire. // DEPRECATED func (m *Messenger) SendRaw(ctx context.Context, chat Chat, data []byte) ([]byte, error) { publicKey, err := chat.PublicKey() if err != nil { return nil, err } if publicKey != nil { return m.processor.SendPrivateRaw(ctx, publicKey, data, protobuf.ApplicationMetadataMessage_UNKNOWN) } else if chat.Name != "" { return m.processor.SendPublicRaw(ctx, chat.Name, data, protobuf.ApplicationMetadataMessage_UNKNOWN) } return nil, errors.New("chat is neither public nor private") } // RetrieveAll retrieves messages from all filters, processes them and returns a // MessengerResponse to the client func (m *Messenger) RetrieveAll() (*MessengerResponse, error) { chatWithMessages, err := m.transport.RetrieveRawAll() if err != nil { return nil, err } return m.handleRetrievedMessages(chatWithMessages) } type ReceivedMessageState struct { // Message is the protobuf message received Message protobuf.ChatMessage // MessageID is the ID of the message MessageID string // WhisperTimestamp is the whisper timestamp of the message WhisperTimestamp uint64 // Contact is the contact associated with the author of the message Contact *Contact // PublicKey is the public key of the author of the message PublicKey *ecdsa.PublicKey // List of chats modified ModifiedChats map[string]bool PostProcessor *postProcessor } func (m *Messenger) handleChatMessage(state *ReceivedMessageState) (*Message, error) { logger := m.logger.With(zap.String("site", "handleChatMessage")) if err := ValidateReceivedChatMessage(&state.Message); err != nil { logger.Warn("failed to validate message", zap.Error(err)) return nil, err } receivedMessage := &Message{ ID: state.MessageID, ChatMessage: state.Message, From: state.Contact.ID, Alias: state.Contact.Alias, SigPubKey: state.PublicKey, Identicon: state.Contact.Identicon, WhisperTimestamp: state.WhisperTimestamp, } receivedMessage.PrepareContent() chat, err := state.PostProcessor.matchMessage(receivedMessage, m.allChats) if err != nil { return nil, err } // If deleted-at is greater, ignore message if chat.DeletedAtClockValue >= receivedMessage.Clock { return nil, nil } // Set the LocalChatID for the message receivedMessage.LocalChatID = chat.ID if c, ok := m.allChats[chat.ID]; ok { chat = c } // Set the LocalChatID for the message receivedMessage.LocalChatID = chat.ID // Increase unviewed count if !isPubKeyEqual(receivedMessage.SigPubKey, &m.identity.PublicKey) { chat.UnviewedMessagesCount++ } else { // Our own message, mark as sent receivedMessage.OutgoingStatus = OutgoingStatusSent } // Update chat timestamp chat.Timestamp = int64(timestampInMs()) // Update last clock value if chat.LastClockValue <= receivedMessage.Clock { chat.LastClockValue = receivedMessage.Clock encodedLastMessage, err := json.Marshal(receivedMessage) if err != nil { return nil, err } chat.LastMessage = encodedLastMessage } // Set chat active chat.Active = true // Set in the modified maps chat state.ModifiedChats[chat.ID] = true m.allChats[chat.ID] = chat return receivedMessage, nil } func (m *Messenger) messageExists(messageID string, existingMessagesMap map[string]bool) (bool, error) { if _, ok := existingMessagesMap[messageID]; ok { return true, nil } existingMessagesMap[messageID] = true // Check against the database, this is probably a bit slow for // each message, but for now might do, we'll make it faster later existingMessage, err := m.persistence.MessageByID(messageID) if err != nil && err != errRecordNotFound { return false, err } if existingMessage != nil { return true, nil } return false, nil } func (m *Messenger) handleRetrievedMessages(chatWithMessages map[transport.Filter][]*types.Message) (*MessengerResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() response := &MessengerResponse{ Chats: []*Chat{}, Messages: []*Message{}, } postProcessor := newPostProcessor(m, postProcessorConfig{MatchChat: true}) logger := m.logger.With(zap.String("site", "RetrieveAll")) rawMessages := make(map[transport.Filter][]*v1protocol.StatusMessage) modifiedChats := make(map[string]bool) modifiedContacts := make(map[string]bool) existingMessagesMap := make(map[string]bool) for chat, messages := range chatWithMessages { for _, shhMessage := range messages { // TODO: fix this to use an exported method. statusMessages, err := m.processor.handleMessages(shhMessage, true) if err != nil { logger.Info("failed to decode messages", zap.Error(err)) continue } for _, msg := range statusMessages { publicKey := msg.SigPubKey() // Check for messages from blocked users senderID := types.EncodeHex(crypto.FromECDSAPub(publicKey)) if _, ok := m.allContacts[senderID]; ok && m.allContacts[senderID].IsBlocked() { continue } // Don't process duplicates messageID := types.EncodeHex(msg.ID) exists, err := m.messageExists(messageID, existingMessagesMap) if err != nil { logger.Warn("failed to check message exists", zap.Error(err)) } if exists { continue } var contact *Contact if c, ok := m.allContacts[senderID]; ok { contact = c } else { c, err := buildContact(publicKey) if err != nil { logger.Info("failed to build contact", zap.Error(err)) continue } contact = c m.allContacts[senderID] = c modifiedContacts[contact.ID] = true } messageState := &ReceivedMessageState{ MessageID: messageID, WhisperTimestamp: uint64(msg.TransportMessage.Timestamp) * 1000, Contact: contact, PublicKey: publicKey, ModifiedChats: modifiedChats, PostProcessor: postProcessor, } if msg.ParsedMessage != nil { switch msg.ParsedMessage.(type) { case protobuf.MembershipUpdateMessage: rawMembershipUpdate := msg.ParsedMessage.(protobuf.MembershipUpdateMessage) membershipUpdate, err := v1protocol.MembershipUpdateMessageFromProtobuf(&rawMembershipUpdate) if err != nil { logger.Warn("failed to process membership update", zap.Error(err)) continue } chat, systemMessages, err := HandleMembershipUpdate(m.allChats[membershipUpdate.ChatID], membershipUpdate, types.EncodeHex(crypto.FromECDSAPub(&m.identity.PublicKey)), m.systemMessagesTranslations) if err != nil { logger.Warn("failed to process membership update", zap.Error(err)) continue } for _, message := range systemMessages { messageID := message.ID exists, err := m.messageExists(messageID, existingMessagesMap) if err != nil { logger.Warn("failed to check message exists", zap.Error(err)) } if exists { continue } response.Messages = append(response.Messages, message) } // Store in chats map as it might be a new one m.allChats[chat.ID] = chat // Set in the map modifiedChats[chat.ID] = true if rawMembershipUpdate.Message != nil { messageState.Message = *rawMembershipUpdate.Message receivedMessage, err := m.handleChatMessage(messageState) if err != nil { logger.Warn("failed to process message", zap.Error(err)) continue } // Add to response if receivedMessage != nil { response.Messages = append(response.Messages, receivedMessage) } } case protobuf.ChatMessage: messageState.Message = msg.ParsedMessage.(protobuf.ChatMessage) receivedMessage, err := m.handleChatMessage(messageState) if err != nil { logger.Warn("failed to process message", zap.Error(err)) continue } // Add to response if receivedMessage != nil { response.Messages = append(response.Messages, receivedMessage) } default: // RawMessage, not processed here, pass straight to the client rawMessages[chat] = append(rawMessages[chat], msg) } } else { rawMessages[chat] = append(rawMessages[chat], msg) } } } } for id := range modifiedChats { response.Chats = append(response.Chats, m.allChats[id]) } for id := range modifiedContacts { response.Contacts = append(response.Contacts, m.allContacts[id]) } var err error if len(response.Chats) > 0 { err = m.saveChats(response.Chats) if err != nil { return nil, err } } if len(response.Messages) > 0 { err = m.SaveMessages(response.Messages) if err != nil { return nil, err } } if len(response.Contacts) > 0 { err = m.persistence.SaveContacts(response.Contacts) if err != nil { return nil, err } } for filter, messages := range rawMessages { response.RawMessages = append(response.RawMessages, &RawResponse{Filter: &filter, Messages: messages}) } return response, nil } func (m *Messenger) RequestHistoricMessages( ctx context.Context, peer []byte, // should be removed after mailserver logic is ported from, to uint32, cursor []byte, ) ([]byte, error) { return m.transport.SendMessagesRequest(ctx, peer, from, to, cursor) } // DEPRECATED func (m *Messenger) LoadFilters(filters []*transport.Filter) ([]*transport.Filter, error) { return m.transport.LoadFilters(filters) } // DEPRECATED func (m *Messenger) RemoveFilters(filters []*transport.Filter) error { return m.transport.RemoveFilters(filters) } // DEPRECATED func (m *Messenger) ConfirmMessagesProcessed(messageIDs [][]byte) error { for _, id := range messageIDs { if err := m.encryptor.ConfirmMessageProcessed(id); err != nil { return err } } return nil } // DEPRECATED: required by status-react. func (m *Messenger) MessageByID(id string) (*Message, error) { return m.persistence.MessageByID(id) } // DEPRECATED: required by status-react. func (m *Messenger) MessagesExist(ids []string) (map[string]bool, error) { return m.persistence.MessagesExist(ids) } // DEPRECATED: required by status-react. func (m *Messenger) MessageByChatID(chatID, cursor string, limit int) ([]*Message, string, error) { return m.persistence.MessageByChatID(chatID, cursor, limit) } // DEPRECATED: required by status-react. func (m *Messenger) SaveMessages(messages []*Message) error { return m.persistence.SaveMessagesLegacy(messages) } // DEPRECATED: required by status-react. func (m *Messenger) DeleteMessage(id string) error { return m.persistence.DeleteMessage(id) } // DEPRECATED: required by status-react. func (m *Messenger) DeleteMessagesByChatID(id string) error { return m.persistence.DeleteMessagesByChatID(id) } // DEPRECATED: required by status-react. func (m *Messenger) MarkMessagesSeen(chatID string, ids []string) error { m.mutex.Lock() defer m.mutex.Unlock() err := m.persistence.MarkMessagesSeen(chatID, ids) if err != nil { return err } chat, err := m.persistence.Chat(chatID) if err != nil { return err } m.allChats[chatID] = chat return nil } // DEPRECATED: required by status-react. func (m *Messenger) UpdateMessageOutgoingStatus(id, newOutgoingStatus string) error { return m.persistence.UpdateMessageOutgoingStatus(id, newOutgoingStatus) } // postProcessor performs a set of actions on newly retrieved messages. // If persist is true, it saves the messages into the database. // If matchChat is true, it matches each messages against a Chat instance. type postProcessor struct { myPublicKey *ecdsa.PublicKey persistence *sqlitePersistence logger *zap.Logger config postProcessorConfig } type postProcessorConfig struct { MatchChat bool // match each messages to a chat; may result in a new chat creation Persist bool // if true, all sent and received user messages will be persisted Parse bool // if true, it will parse the content } func newPostProcessor(m *Messenger, config postProcessorConfig) *postProcessor { return &postProcessor{ myPublicKey: &m.identity.PublicKey, persistence: m.persistence, logger: m.logger, config: config, } } func (p *postProcessor) matchMessage(message *Message, chats map[string]*Chat) (*Chat, error) { if message.SigPubKey == nil { p.logger.Error("public key can't be empty") return nil, errors.New("received a message with empty public key") } switch { case message.MessageType == protobuf.ChatMessage_PUBLIC_GROUP: // For public messages, all outgoing and incoming messages have the same chatID // equal to a public chat name. chatID := message.ChatId chat := chats[chatID] if chat == nil { return nil, errors.New("received a public message from non-existing chat") } return chat, nil case message.MessageType == protobuf.ChatMessage_ONE_TO_ONE && isPubKeyEqual(message.SigPubKey, p.myPublicKey): // It's a private message coming from us so we rely on Message.ChatId // If chat does not exist, it should be created to support multidevice synchronization. chatID := message.ChatId chat := chats[chatID] if chat == nil { if len(chatID) != PubKeyStringLength { return nil, errors.New("invalid pubkey length") } bytePubKey, err := hex.DecodeString(chatID[2:]) if err != nil { return nil, errors.Wrap(err, "failed to decode hex chatID") } pubKey, err := crypto.UnmarshalPubkey(bytePubKey) if err != nil { return nil, errors.Wrap(err, "failed to decode pubkey") } newChat := CreateOneToOneChat(chatID[:8], pubKey) chat = &newChat } return chat, nil case message.MessageType == protobuf.ChatMessage_ONE_TO_ONE: // It's an incoming private message. ChatID is calculated from the signature. // If a chat does not exist, a new one is created and saved. chatID := types.EncodeHex(crypto.FromECDSAPub(message.SigPubKey)) chat := chats[chatID] if chat == nil { // TODO: this should be a three-word name used in the mobile client newChat := CreateOneToOneChat(chatID[:8], message.SigPubKey) chat = &newChat } return chat, nil case message.MessageType == protobuf.ChatMessage_PRIVATE_GROUP: // In the case of a group message, ChatID is the same for all messages belonging to a group. // It needs to be verified if the signature public key belongs to the chat. chatID := message.ChatId chat := chats[chatID] if chat == nil { return nil, errors.New("received group chat message for non-existing chat") } theirKeyHex := types.EncodeHex(crypto.FromECDSAPub(message.SigPubKey)) myKeyHex := types.EncodeHex(crypto.FromECDSAPub(p.myPublicKey)) var theyJoined bool var iJoined bool for _, member := range chat.Members { if member.ID == theirKeyHex && member.Joined { theyJoined = true } } for _, member := range chat.Members { if member.ID == myKeyHex && member.Joined { iJoined = true } } if theyJoined && iJoined { return chat, nil } return nil, errors.New("did not find a matching group chat") default: return nil, errors.New("can not match a chat because there is no valid case") } } // Identicon returns an identicon based on the input string func Identicon(id string) (string, error) { return identicon.GenerateBase64(id) } // VerifyENSNames verifies that a registered ENS name matches the expected public key func (m *Messenger) VerifyENSNames(rpcEndpoint, contractAddress string, ensDetails []enstypes.ENSDetails) (map[string]enstypes.ENSResponse, error) { m.mutex.Lock() defer m.mutex.Unlock() verifier := m.node.NewENSVerifier(m.logger) ensResponse, err := verifier.CheckBatch(ensDetails, rpcEndpoint, contractAddress) if err != nil { return nil, err } // Update contacts var contacts []*Contact for _, details := range ensResponse { if details.Error == nil { contact, ok := m.allContacts[details.PublicKeyString] if !ok { contact, err = buildContact(details.PublicKey) if err != nil { return nil, err } } contact.ENSVerified = details.Verified contact.ENSVerifiedAt = details.VerifiedAt contact.Name = details.Name contacts = append(contacts, contact) } else { m.logger.Warn("Failed to resolve ens name", zap.String("name", details.Name), zap.String("publicKey", details.PublicKeyString), zap.Error(details.Error), ) } } if len(contacts) != 0 { err = m.persistence.SaveContacts(contacts) if err != nil { return nil, err } } return ensResponse, nil } // GenerateAlias name returns the generated name given a public key hex encoded prefixed with 0x func GenerateAlias(id string) (string, error) { return alias.GenerateFromPublicKeyString(id) }