2020-07-06 08:54:22 +00:00
|
|
|
package common
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"crypto/ecdsa"
|
|
|
|
"database/sql"
|
2020-07-21 15:41:10 +00:00
|
|
|
"sync"
|
2019-09-02 09:29:06 +00:00
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/golang/protobuf/proto"
|
|
|
|
"github.com/pkg/errors"
|
2020-01-02 09:10:19 +00:00
|
|
|
datasyncnode "github.com/vacp2p/mvds/node"
|
|
|
|
datasyncproto "github.com/vacp2p/mvds/protobuf"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
|
2019-11-23 17:57:05 +00:00
|
|
|
"github.com/status-im/status-go/eth-node/crypto"
|
|
|
|
"github.com/status-im/status-go/eth-node/types"
|
2019-11-21 16:19:22 +00:00
|
|
|
"github.com/status-im/status-go/protocol/datasync"
|
|
|
|
datasyncpeer "github.com/status-im/status-go/protocol/datasync/peer"
|
|
|
|
"github.com/status-im/status-go/protocol/encryption"
|
2020-07-31 12:22:05 +00:00
|
|
|
"github.com/status-im/status-go/protocol/encryption/sharedsecret"
|
2020-07-25 21:25:40 +00:00
|
|
|
"github.com/status-im/status-go/protocol/protobuf"
|
2020-01-13 19:17:30 +00:00
|
|
|
"github.com/status-im/status-go/protocol/transport"
|
2019-11-21 16:19:22 +00:00
|
|
|
v1protocol "github.com/status-im/status-go/protocol/v1"
|
2019-09-02 09:29:06 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// Whisper message properties.
|
|
|
|
const (
|
2020-05-13 13:19:29 +00:00
|
|
|
whisperTTL = 15
|
|
|
|
whisperDefaultPoW = 0.002
|
|
|
|
// whisperLargeSizePoW is the PoWTarget for larger payload sizes
|
|
|
|
whisperLargeSizePoW = 0.000002
|
|
|
|
// largeSizeInBytes is when should we be using a lower POW.
|
|
|
|
// Roughly this is 50KB
|
|
|
|
largeSizeInBytes = 50000
|
|
|
|
whisperPoWTime = 5
|
2019-09-02 09:29:06 +00:00
|
|
|
)
|
|
|
|
|
2023-10-12 15:45:23 +00:00
|
|
|
// RekeyCompatibility indicates whether we should be sending
|
|
|
|
// keys in 1-to-1 messages as well as in the newer format
|
|
|
|
var RekeyCompatibility = true
|
|
|
|
|
2020-07-06 08:54:22 +00:00
|
|
|
// SentMessage reprent a message that has been passed to the transport layer
|
|
|
|
type SentMessage struct {
|
|
|
|
PublicKey *ecdsa.PublicKey
|
|
|
|
Spec *encryption.ProtocolMessageSpec
|
|
|
|
MessageIDs [][]byte
|
|
|
|
}
|
|
|
|
|
2021-10-07 11:10:23 +00:00
|
|
|
type MessageEventType uint32
|
|
|
|
|
|
|
|
const (
|
|
|
|
MessageScheduled = iota + 1
|
|
|
|
MessageSent
|
|
|
|
)
|
|
|
|
|
|
|
|
type MessageEvent struct {
|
2022-07-15 13:19:37 +00:00
|
|
|
Recipient *ecdsa.PublicKey
|
2021-10-07 11:10:23 +00:00
|
|
|
Type MessageEventType
|
|
|
|
SentMessage *SentMessage
|
|
|
|
RawMessage *RawMessage
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
type MessageSender struct {
|
2021-03-09 12:48:15 +00:00
|
|
|
identity *ecdsa.PrivateKey
|
|
|
|
datasync *datasync.DataSync
|
2021-10-21 12:39:19 +00:00
|
|
|
database *sql.DB
|
2021-03-09 12:48:15 +00:00
|
|
|
protocol *encryption.Protocol
|
2021-04-22 13:36:18 +00:00
|
|
|
transport *transport.Transport
|
2021-03-09 12:48:15 +00:00
|
|
|
logger *zap.Logger
|
|
|
|
persistence *RawMessagesPersistence
|
2020-07-22 07:41:40 +00:00
|
|
|
|
2021-10-21 12:39:19 +00:00
|
|
|
datasyncEnabled bool
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// ephemeralKeys is a map that contains the ephemeral keys of the client, used
|
|
|
|
// to decrypt messages
|
|
|
|
ephemeralKeys map[string]*ecdsa.PrivateKey
|
|
|
|
ephemeralKeysMutex sync.Mutex
|
|
|
|
|
2021-10-07 11:10:23 +00:00
|
|
|
// messageEventsSubscriptions contains all the subscriptions for message events
|
|
|
|
messageEventsSubscriptions []chan<- *MessageEvent
|
2020-07-06 08:54:22 +00:00
|
|
|
|
|
|
|
featureFlags FeatureFlags
|
2020-07-31 12:22:05 +00:00
|
|
|
|
|
|
|
// handleSharedSecrets is a callback that is called every time a new shared secret is negotiated
|
|
|
|
handleSharedSecrets func([]*sharedsecret.Secret) error
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func NewMessageSender(
|
2019-09-02 09:29:06 +00:00
|
|
|
identity *ecdsa.PrivateKey,
|
|
|
|
database *sql.DB,
|
|
|
|
enc *encryption.Protocol,
|
2021-04-22 13:36:18 +00:00
|
|
|
transport *transport.Transport,
|
2019-09-02 09:29:06 +00:00
|
|
|
logger *zap.Logger,
|
2020-07-06 08:54:22 +00:00
|
|
|
features FeatureFlags,
|
2021-06-23 14:13:48 +00:00
|
|
|
) (*MessageSender, error) {
|
2020-02-10 11:22:37 +00:00
|
|
|
dataSyncTransport := datasync.NewNodeTransport()
|
2019-09-02 09:29:06 +00:00
|
|
|
dataSyncNode, err := datasyncnode.NewPersistentNode(
|
|
|
|
database,
|
|
|
|
dataSyncTransport,
|
|
|
|
datasyncpeer.PublicKeyToPeerID(identity.PublicKey),
|
|
|
|
datasyncnode.BATCH,
|
|
|
|
datasync.CalculateSendTime,
|
|
|
|
logger,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2020-07-06 08:54:22 +00:00
|
|
|
ds := datasync.New(dataSyncNode, dataSyncTransport, features.Datasync, logger)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
p := &MessageSender{
|
2021-10-21 12:39:19 +00:00
|
|
|
identity: identity,
|
|
|
|
datasyncEnabled: features.Datasync,
|
|
|
|
datasync: ds,
|
|
|
|
protocol: enc,
|
|
|
|
database: database,
|
|
|
|
persistence: NewRawMessagesPersistence(database),
|
|
|
|
transport: transport,
|
|
|
|
logger: logger,
|
|
|
|
ephemeralKeys: make(map[string]*ecdsa.PrivateKey),
|
|
|
|
featureFlags: features,
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Initializing DataSync is required to encrypt and send messages.
|
|
|
|
// With DataSync enabled, messages are added to the DataSync
|
|
|
|
// but actual encrypt and send calls are postponed.
|
|
|
|
// sendDataSync is responsible for encrypting and sending postponed messages.
|
2020-07-06 08:54:22 +00:00
|
|
|
if features.Datasync {
|
2020-11-03 12:42:42 +00:00
|
|
|
// We set the max message size to 3/4 of the allowed message size, to leave
|
|
|
|
// room for encryption.
|
|
|
|
// Messages will be tried to send in any case, even if they exceed this
|
|
|
|
// value
|
|
|
|
ds.Init(p.sendDataSync, transport.MaxMessageSize()/4*3, logger)
|
2020-11-24 06:25:54 +00:00
|
|
|
ds.Start(datasync.DatasyncTicker)
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return p, nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) Stop() {
|
2021-10-07 11:10:23 +00:00
|
|
|
for _, c := range s.messageEventsSubscriptions {
|
2020-07-06 08:54:22 +00:00
|
|
|
close(c)
|
|
|
|
}
|
2021-10-07 11:10:23 +00:00
|
|
|
s.messageEventsSubscriptions = nil
|
2021-06-23 14:13:48 +00:00
|
|
|
s.datasync.Stop() // idempotent op
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SetHandleSharedSecrets(handler func([]*sharedsecret.Secret) error) {
|
|
|
|
s.handleSharedSecrets = handler
|
2020-07-31 12:22:05 +00:00
|
|
|
}
|
|
|
|
|
2020-05-13 13:24:52 +00:00
|
|
|
// SendPrivate takes encoded data, encrypts it and sends through the wire.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SendPrivate(
|
2019-09-02 09:29:06 +00:00
|
|
|
ctx context.Context,
|
2019-10-14 14:10:48 +00:00
|
|
|
recipient *ecdsa.PublicKey,
|
2021-02-23 15:47:45 +00:00
|
|
|
rawMessage *RawMessage,
|
2019-09-02 09:29:06 +00:00
|
|
|
) ([]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Debug(
|
2019-09-02 09:29:06 +00:00
|
|
|
"sending a private message",
|
2020-07-21 15:41:10 +00:00
|
|
|
zap.String("public-key", types.EncodeHex(crypto.FromECDSAPub(recipient))),
|
2020-05-13 13:24:52 +00:00
|
|
|
zap.String("site", "SendPrivate"),
|
2019-09-02 09:29:06 +00:00
|
|
|
)
|
2020-07-22 07:41:40 +00:00
|
|
|
// Currently we don't support sending through datasync and setting custom waku fields,
|
|
|
|
// as the datasync interface is not rich enough to propagate that information, so we
|
|
|
|
// would have to add some complexity to handle this.
|
2023-11-08 18:05:33 +00:00
|
|
|
if rawMessage.ResendAutomatically && (rawMessage.Sender != nil || rawMessage.SkipEncryptionLayer || rawMessage.SendOnPersonalTopic) {
|
2021-01-18 09:12:03 +00:00
|
|
|
return nil, errors.New("setting identity, skip-encryption or personal topic and datasync not supported")
|
2020-07-22 07:41:40 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Set sender identity if not specified
|
|
|
|
if rawMessage.Sender == nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
rawMessage.Sender = s.identity
|
2020-07-22 07:41:40 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
return s.sendPrivate(ctx, recipient, rawMessage)
|
2019-09-26 07:01:17 +00:00
|
|
|
}
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2021-01-11 10:32:51 +00:00
|
|
|
// SendCommunityMessage takes encoded data, encrypts it and sends through the wire
|
|
|
|
// using the community topic and their key
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SendCommunityMessage(
|
2021-01-11 10:32:51 +00:00
|
|
|
ctx context.Context,
|
|
|
|
rawMessage RawMessage,
|
|
|
|
) ([]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Debug(
|
2021-01-11 10:32:51 +00:00
|
|
|
"sending a community message",
|
2022-05-27 09:14:40 +00:00
|
|
|
zap.String("communityId", types.EncodeHex(rawMessage.CommunityID)),
|
|
|
|
zap.String("site", "SendCommunityMessage"),
|
2021-01-11 10:32:51 +00:00
|
|
|
)
|
2021-06-23 14:13:48 +00:00
|
|
|
rawMessage.Sender = s.identity
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2022-05-27 09:14:40 +00:00
|
|
|
return s.sendCommunity(ctx, &rawMessage)
|
2021-01-11 10:32:51 +00:00
|
|
|
}
|
|
|
|
|
2023-10-12 19:21:49 +00:00
|
|
|
// SendPubsubTopicKey sends the protected topic key for a community to a list of recipients
|
|
|
|
func (s *MessageSender) SendPubsubTopicKey(
|
|
|
|
ctx context.Context,
|
|
|
|
rawMessage *RawMessage,
|
|
|
|
) ([]byte, error) {
|
|
|
|
s.logger.Debug(
|
|
|
|
"sending the protected topic key for a community",
|
|
|
|
zap.String("communityId", types.EncodeHex(rawMessage.CommunityID)),
|
|
|
|
zap.String("site", "SendPubsubTopicKey"),
|
|
|
|
)
|
|
|
|
rawMessage.Sender = s.identity
|
|
|
|
messageID, err := s.getMessageID(rawMessage)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
|
|
|
|
|
|
|
// Notify before dispatching, otherwise the dispatch subscription might happen
|
|
|
|
// earlier than the scheduled
|
|
|
|
s.notifyOnScheduledMessage(nil, rawMessage)
|
|
|
|
|
|
|
|
// Send to each recipients
|
|
|
|
for _, recipient := range rawMessage.Recipients {
|
|
|
|
_, err = s.sendPrivate(ctx, recipient, rawMessage)
|
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to send message")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return messageID, nil
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// SendGroup takes encoded data, encrypts it and sends through the wire,
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
// always return the messageID
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SendGroup(
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
ctx context.Context,
|
|
|
|
recipients []*ecdsa.PublicKey,
|
2020-07-28 13:22:22 +00:00
|
|
|
rawMessage RawMessage,
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
) ([]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Debug(
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
"sending a private group message",
|
2020-05-13 13:24:52 +00:00
|
|
|
zap.String("site", "SendGroup"),
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
)
|
2020-07-22 07:41:40 +00:00
|
|
|
// Set sender if not specified
|
2020-07-21 15:41:10 +00:00
|
|
|
if rawMessage.Sender == nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
rawMessage.Sender = s.identity
|
2020-07-21 15:41:10 +00:00
|
|
|
}
|
2020-07-22 07:41:40 +00:00
|
|
|
|
|
|
|
// Calculate messageID first and set on raw message
|
2021-06-23 14:13:48 +00:00
|
|
|
wrappedMessage, err := s.wrapMessageV1(&rawMessage)
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
|
|
|
}
|
2020-07-21 15:41:10 +00:00
|
|
|
messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, wrappedMessage)
|
2020-07-22 07:41:40 +00:00
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
|
2023-06-29 09:58:20 +00:00
|
|
|
// We call it only once, and we nil the function after so it doesn't get called again
|
|
|
|
if rawMessage.BeforeDispatch != nil {
|
|
|
|
if err := rawMessage.BeforeDispatch(&rawMessage); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
rawMessage.BeforeDispatch = nil
|
|
|
|
|
2020-05-13 13:24:52 +00:00
|
|
|
// Send to each recipients
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
for _, recipient := range recipients {
|
2021-06-23 14:13:48 +00:00
|
|
|
_, err = s.sendPrivate(ctx, recipient, &rawMessage)
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to send message")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2022-05-27 09:14:40 +00:00
|
|
|
func (s *MessageSender) getMessageID(rawMessage *RawMessage) (types.HexBytes, error) {
|
|
|
|
wrappedMessage, err := s.wrapMessageV1(rawMessage)
|
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
|
|
|
}
|
|
|
|
|
|
|
|
messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, wrappedMessage)
|
|
|
|
|
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2022-12-14 22:11:25 +00:00
|
|
|
func ShouldCommunityMessageBeEncrypted(msgType protobuf.ApplicationMetadataMessage_Type) bool {
|
|
|
|
return msgType == protobuf.ApplicationMetadataMessage_CHAT_MESSAGE ||
|
|
|
|
msgType == protobuf.ApplicationMetadataMessage_EDIT_MESSAGE ||
|
|
|
|
msgType == protobuf.ApplicationMetadataMessage_DELETE_MESSAGE ||
|
|
|
|
msgType == protobuf.ApplicationMetadataMessage_PIN_MESSAGE ||
|
|
|
|
msgType == protobuf.ApplicationMetadataMessage_EMOJI_REACTION
|
|
|
|
}
|
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
// sendCommunity sends a message that's to be sent in a community
|
|
|
|
// If it's a chat message, it will go to the respective topic derived by the
|
|
|
|
// chat id, if it's not a chat message, it will go to the community topic.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) sendCommunity(
|
2021-01-11 10:32:51 +00:00
|
|
|
ctx context.Context,
|
|
|
|
rawMessage *RawMessage,
|
|
|
|
) ([]byte, error) {
|
2022-05-27 09:14:40 +00:00
|
|
|
s.logger.Debug("sending community message", zap.String("recipient", types.EncodeHex(crypto.FromECDSAPub(&rawMessage.Sender.PublicKey))))
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
// Set sender
|
|
|
|
if rawMessage.Sender == nil {
|
|
|
|
rawMessage.Sender = s.identity
|
|
|
|
}
|
|
|
|
|
2022-05-27 09:14:40 +00:00
|
|
|
messageID, err := s.getMessageID(rawMessage)
|
2021-01-11 10:32:51 +00:00
|
|
|
if err != nil {
|
2022-05-27 09:14:40 +00:00
|
|
|
return nil, err
|
2021-01-11 10:32:51 +00:00
|
|
|
}
|
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
2022-05-27 09:14:40 +00:00
|
|
|
messageIDs := [][]byte{messageID}
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2023-06-20 16:12:59 +00:00
|
|
|
if rawMessage.BeforeDispatch != nil {
|
|
|
|
if err := rawMessage.BeforeDispatch(rawMessage); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
2021-01-11 10:32:51 +00:00
|
|
|
// Notify before dispatching, otherwise the dispatch subscription might happen
|
|
|
|
// earlier than the scheduled
|
2022-07-15 13:19:37 +00:00
|
|
|
s.notifyOnScheduledMessage(nil, rawMessage)
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
var hash []byte
|
|
|
|
var newMessage *types.NewMessage
|
|
|
|
|
|
|
|
// Check if it's a key exchange message. In this case we send it
|
|
|
|
// to all the recipients
|
2022-05-27 09:14:40 +00:00
|
|
|
if rawMessage.CommunityKeyExMsgType != KeyExMsgNone {
|
2023-10-12 15:45:23 +00:00
|
|
|
forceRekey := rawMessage.CommunityKeyExMsgType == KeyExMsgRekey
|
|
|
|
// If rekeycompatibility is on, we always
|
|
|
|
// want to execute below, otherwise we execute
|
|
|
|
// only when we want to fill up old keys to a given user
|
|
|
|
if RekeyCompatibility || !forceRekey {
|
|
|
|
keyExMessageSpecs, err := s.protocol.GetKeyExMessageSpecs(rawMessage.HashRatchetGroupID, s.identity, rawMessage.Recipients, forceRekey)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
for i, spec := range keyExMessageSpecs {
|
|
|
|
recipient := rawMessage.Recipients[i]
|
|
|
|
_, _, err = s.sendMessageSpec(ctx, recipient, spec, messageIDs)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
2022-05-27 09:14:40 +00:00
|
|
|
}
|
|
|
|
|
2023-10-12 15:45:23 +00:00
|
|
|
if forceRekey {
|
|
|
|
|
|
|
|
var ratchet *encryption.HashRatchetKeyCompatibility
|
|
|
|
// We have just rekeyed, pull the latest
|
|
|
|
if RekeyCompatibility {
|
|
|
|
ratchet, err = s.protocol.GetCurrentKeyForGroup(rawMessage.HashRatchetGroupID)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
// We send the message over the community topic
|
|
|
|
messages, err := s.protocol.BuildHashRatchetReKeyGroupMessage(s.identity, rawMessage.Recipients, rawMessage.HashRatchetGroupID, ratchet)
|
2022-05-27 09:14:40 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2023-10-12 15:45:23 +00:00
|
|
|
for _, m := range messages {
|
|
|
|
payload, err := proto.Marshal(m.Message)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
rawMessage.Payload = payload
|
|
|
|
newMessage = &types.NewMessage{
|
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: payload,
|
|
|
|
PowTarget: calculatePoW(payload),
|
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
}
|
|
|
|
|
|
|
|
newMessage.Ephemeral = rawMessage.Ephemeral
|
|
|
|
newMessage.PubsubTopic = rawMessage.PubsubTopic
|
|
|
|
|
|
|
|
messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, payload)
|
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
|
|
|
|
|
|
|
// notify before dispatching
|
|
|
|
s.notifyOnScheduledMessage(nil, rawMessage)
|
|
|
|
|
|
|
|
_, err = s.transport.SendPublic(ctx, newMessage, types.EncodeHex(rawMessage.CommunityID))
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2022-05-27 09:14:40 +00:00
|
|
|
}
|
2022-09-15 14:13:35 +00:00
|
|
|
return nil, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
wrappedMessage, err := s.wrapMessageV1(rawMessage)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// If it's a chat message, we send it on the community chat topic
|
2022-12-14 22:11:25 +00:00
|
|
|
if ShouldCommunityMessageBeEncrypted(rawMessage.MessageType) {
|
2023-06-23 10:49:26 +00:00
|
|
|
messageSpec, err := s.protocol.BuildHashRatchetMessage(rawMessage.HashRatchetGroupID, wrappedMessage)
|
2022-05-27 09:14:40 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
payload, err := proto.Marshal(messageSpec.Message)
|
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to marshal")
|
|
|
|
}
|
2022-09-21 16:05:29 +00:00
|
|
|
hash, newMessage, err = s.dispatchCommunityChatMessage(ctx, rawMessage, payload)
|
2022-09-15 14:13:35 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2022-09-05 12:08:44 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
sentMessage := &SentMessage{
|
|
|
|
Spec: messageSpec,
|
|
|
|
MessageIDs: messageIDs,
|
|
|
|
}
|
2022-09-05 12:08:44 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
s.notifyOnSentMessage(sentMessage)
|
2021-10-29 14:29:28 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
} else {
|
2021-01-11 10:32:51 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
payload := wrappedMessage
|
2022-09-05 12:08:44 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
pubkey, err := crypto.DecompressPubkey(rawMessage.CommunityID)
|
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to decompress pubkey")
|
|
|
|
}
|
2023-05-22 21:38:02 +00:00
|
|
|
hash, newMessage, err = s.dispatchCommunityMessage(ctx, pubkey, payload, messageIDs, rawMessage.PubsubTopic)
|
2022-09-15 14:13:35 +00:00
|
|
|
if err != nil {
|
|
|
|
s.logger.Error("failed to send a community message", zap.Error(err))
|
|
|
|
return nil, errors.Wrap(err, "failed to send a message spec")
|
2022-09-05 12:08:44 +00:00
|
|
|
}
|
2022-05-27 09:14:40 +00:00
|
|
|
|
2022-09-15 14:13:35 +00:00
|
|
|
s.logger.Debug("sent community message ", zap.String("messageID", messageID.String()), zap.String("hash", types.EncodeHex(hash)))
|
2022-05-27 09:14:40 +00:00
|
|
|
}
|
2022-09-15 14:13:35 +00:00
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
s.transport.Track(messageIDs, hash, newMessage)
|
2021-01-11 10:32:51 +00:00
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2019-09-26 07:01:17 +00:00
|
|
|
// sendPrivate sends data to the recipient identifying with a given public key.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) sendPrivate(
|
2019-09-26 07:01:17 +00:00
|
|
|
ctx context.Context,
|
2019-10-14 14:10:48 +00:00
|
|
|
recipient *ecdsa.PublicKey,
|
2020-05-13 13:24:52 +00:00
|
|
|
rawMessage *RawMessage,
|
2019-09-26 07:01:17 +00:00
|
|
|
) ([]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Debug("sending private message", zap.String("recipient", types.EncodeHex(crypto.FromECDSAPub(recipient))))
|
2020-07-21 15:41:10 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
wrappedMessage, err := s.wrapMessageV1(rawMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
|
|
|
}
|
|
|
|
|
2020-07-21 15:41:10 +00:00
|
|
|
messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, wrappedMessage)
|
2020-07-22 07:41:40 +00:00
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
2023-11-10 00:29:15 +00:00
|
|
|
rawMessage.PubsubTopic = transport.DefaultShardPubsubTopic() // TODO: determine which pubsub topic should be used for 1:1 messages
|
2020-07-22 07:41:40 +00:00
|
|
|
|
2023-06-20 16:12:59 +00:00
|
|
|
if rawMessage.BeforeDispatch != nil {
|
|
|
|
if err := rawMessage.BeforeDispatch(rawMessage); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// Notify before dispatching, otherwise the dispatch subscription might happen
|
|
|
|
// earlier than the scheduled
|
2022-07-15 13:19:37 +00:00
|
|
|
s.notifyOnScheduledMessage(recipient, rawMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
if s.featureFlags.Datasync && rawMessage.ResendAutomatically {
|
2020-07-21 15:41:10 +00:00
|
|
|
// No need to call transport tracking.
|
|
|
|
// It is done in a data sync dispatch step.
|
2021-06-23 14:13:48 +00:00
|
|
|
datasyncID, err := s.addToDataSync(recipient, wrappedMessage)
|
2021-02-23 15:47:45 +00:00
|
|
|
if err != nil {
|
2019-09-02 09:29:06 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to send message with datasync")
|
|
|
|
}
|
2021-03-09 12:48:15 +00:00
|
|
|
// We don't need to receive confirmations from our own devices
|
2021-06-23 14:13:48 +00:00
|
|
|
if !IsPubKeyEqual(recipient, &s.identity.PublicKey) {
|
2021-03-09 12:48:15 +00:00
|
|
|
confirmation := &RawMessageConfirmation{
|
|
|
|
DataSyncID: datasyncID,
|
|
|
|
MessageID: messageID,
|
|
|
|
PublicKey: crypto.CompressPubkey(recipient),
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
err = s.persistence.InsertPendingConfirmation(confirmation)
|
2021-03-09 12:48:15 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
2023-11-08 18:05:33 +00:00
|
|
|
} else if rawMessage.SkipEncryptionLayer {
|
2023-07-04 20:10:51 +00:00
|
|
|
// When SkipProtocolLayer is set we don't pass the message to the encryption layer
|
2020-07-21 15:41:10 +00:00
|
|
|
messageIDs := [][]byte{messageID}
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, newMessage, err := s.sendPrivateRawMessage(ctx, rawMessage, recipient, wrappedMessage, messageIDs)
|
2020-07-21 15:41:10 +00:00
|
|
|
if err != nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Error("failed to send a private message", zap.Error(err))
|
2020-07-21 15:41:10 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to send a message spec")
|
|
|
|
}
|
|
|
|
|
2023-07-04 20:10:51 +00:00
|
|
|
s.logger.Debug("sent private message skipProtocolLayer", zap.String("messageID", messageID.String()), zap.String("hash", types.EncodeHex(hash)))
|
2021-10-29 14:29:28 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.transport.Track(messageIDs, hash, newMessage)
|
2020-07-21 15:41:10 +00:00
|
|
|
|
2019-09-02 09:29:06 +00:00
|
|
|
} else {
|
2021-09-21 15:47:04 +00:00
|
|
|
messageSpec, err := s.protocol.BuildEncryptedMessage(rawMessage.Sender, recipient, wrappedMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to encrypt message")
|
|
|
|
}
|
|
|
|
|
2020-07-31 12:22:05 +00:00
|
|
|
// The shared secret needs to be handle before we send a message
|
|
|
|
// otherwise the topic might not be set up before we receive a message
|
2021-06-23 14:13:48 +00:00
|
|
|
if s.handleSharedSecrets != nil {
|
|
|
|
err := s.handleSharedSecrets([]*sharedsecret.Secret{messageSpec.SharedSecret})
|
2020-07-31 12:22:05 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2020-06-30 07:50:59 +00:00
|
|
|
messageIDs := [][]byte{messageID}
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, newMessage, err := s.sendMessageSpec(ctx, recipient, messageSpec, messageIDs)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Error("failed to send a private message", zap.Error(err))
|
2019-09-02 09:29:06 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to send a message spec")
|
|
|
|
}
|
|
|
|
|
2021-10-29 14:29:28 +00:00
|
|
|
s.logger.Debug("sent private message without datasync", zap.String("messageID", messageID.String()), zap.String("hash", types.EncodeHex(hash)))
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.transport.Track(messageIDs, hash, newMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2020-01-10 18:59:01 +00:00
|
|
|
// sendPairInstallation sends data to the recipients, using DH
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SendPairInstallation(
|
2019-10-14 14:10:48 +00:00
|
|
|
ctx context.Context,
|
2020-01-10 18:59:01 +00:00
|
|
|
recipient *ecdsa.PublicKey,
|
2020-07-28 13:22:22 +00:00
|
|
|
rawMessage RawMessage,
|
2020-01-10 18:59:01 +00:00
|
|
|
) ([]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Debug("sending private message", zap.String("recipient", types.EncodeHex(crypto.FromECDSAPub(recipient))))
|
2020-01-10 18:59:01 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
wrappedMessage, err := s.wrapMessageV1(&rawMessage)
|
2020-01-10 18:59:01 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
messageSpec, err := s.protocol.BuildDHMessage(s.identity, recipient, wrappedMessage)
|
2020-01-10 18:59:01 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to encrypt message")
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
messageID := v1protocol.MessageID(&s.identity.PublicKey, wrappedMessage)
|
2020-06-30 07:50:59 +00:00
|
|
|
messageIDs := [][]byte{messageID}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, newMessage, err := s.sendMessageSpec(ctx, recipient, messageSpec, messageIDs)
|
2020-01-10 18:59:01 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to send a message spec")
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.transport.Track(messageIDs, hash, newMessage)
|
2020-01-10 18:59:01 +00:00
|
|
|
|
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) encodeMembershipUpdate(
|
2021-01-08 11:26:15 +00:00
|
|
|
message v1protocol.MembershipUpdateMessage,
|
2020-07-25 21:25:40 +00:00
|
|
|
chatEntity ChatEntity,
|
2019-12-02 15:34:05 +00:00
|
|
|
) ([]byte, error) {
|
2019-10-14 14:10:48 +00:00
|
|
|
|
2020-07-27 10:13:22 +00:00
|
|
|
if chatEntity != nil {
|
|
|
|
chatEntityProtobuf := chatEntity.GetProtobuf()
|
2020-07-28 08:02:51 +00:00
|
|
|
switch chatEntityProtobuf := chatEntityProtobuf.(type) {
|
2020-07-27 10:13:22 +00:00
|
|
|
case *protobuf.ChatMessage:
|
2020-07-28 08:02:51 +00:00
|
|
|
message.Message = chatEntityProtobuf
|
2020-07-27 10:13:22 +00:00
|
|
|
case *protobuf.EmojiReaction:
|
2020-07-28 08:02:51 +00:00
|
|
|
message.EmojiReaction = chatEntityProtobuf
|
2020-07-27 10:13:22 +00:00
|
|
|
|
|
|
|
}
|
2019-10-14 14:10:48 +00:00
|
|
|
}
|
2020-07-27 10:13:22 +00:00
|
|
|
|
2019-11-21 16:19:22 +00:00
|
|
|
encodedMessage, err := v1protocol.EncodeMembershipUpdateMessage(message)
|
2019-10-14 14:10:48 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to encode membership update message")
|
|
|
|
}
|
|
|
|
|
2020-01-10 18:59:01 +00:00
|
|
|
return encodedMessage, nil
|
2019-10-14 14:10:48 +00:00
|
|
|
}
|
|
|
|
|
2021-01-08 11:26:15 +00:00
|
|
|
// EncodeMembershipUpdate takes a group and an optional chat message and returns the protobuf representation to be sent on the wire.
|
|
|
|
// All the events in a group are encoded and added to the payload
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) EncodeMembershipUpdate(
|
2021-01-08 11:26:15 +00:00
|
|
|
group *v1protocol.Group,
|
|
|
|
chatEntity ChatEntity,
|
|
|
|
) ([]byte, error) {
|
|
|
|
message := v1protocol.MembershipUpdateMessage{
|
|
|
|
ChatID: group.ChatID(),
|
|
|
|
Events: group.Events(),
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
return s.encodeMembershipUpdate(message, chatEntity)
|
2021-01-08 11:26:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// EncodeAbridgedMembershipUpdate takes a group and an optional chat message and returns the protobuf representation to be sent on the wire.
|
2022-09-13 16:07:34 +00:00
|
|
|
// Only the events relevant to the current group are encoded
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) EncodeAbridgedMembershipUpdate(
|
2021-01-08 11:26:15 +00:00
|
|
|
group *v1protocol.Group,
|
|
|
|
chatEntity ChatEntity,
|
|
|
|
) ([]byte, error) {
|
|
|
|
message := v1protocol.MembershipUpdateMessage{
|
|
|
|
ChatID: group.ChatID(),
|
2022-09-13 16:07:34 +00:00
|
|
|
Events: group.AbridgedEvents(),
|
2021-01-08 11:26:15 +00:00
|
|
|
}
|
2021-06-23 14:13:48 +00:00
|
|
|
return s.encodeMembershipUpdate(message, chatEntity)
|
2021-01-08 11:26:15 +00:00
|
|
|
}
|
|
|
|
|
2022-09-21 16:05:29 +00:00
|
|
|
func (s *MessageSender) dispatchCommunityChatMessage(ctx context.Context, rawMessage *RawMessage, wrappedMessage []byte) ([]byte, *types.NewMessage, error) {
|
2022-09-05 12:08:44 +00:00
|
|
|
|
|
|
|
newMessage := &types.NewMessage{
|
2023-05-22 21:38:02 +00:00
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: wrappedMessage,
|
|
|
|
PowTarget: calculatePoW(wrappedMessage),
|
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
PubsubTopic: rawMessage.PubsubTopic,
|
2022-09-05 12:08:44 +00:00
|
|
|
}
|
|
|
|
|
2023-06-20 16:12:59 +00:00
|
|
|
if rawMessage.BeforeDispatch != nil {
|
|
|
|
if err := rawMessage.BeforeDispatch(rawMessage); err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
// notify before dispatching
|
|
|
|
s.notifyOnScheduledMessage(nil, rawMessage)
|
|
|
|
|
|
|
|
hash, err := s.transport.SendPublic(ctx, newMessage, rawMessage.LocalChatID)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return hash, newMessage, nil
|
|
|
|
}
|
|
|
|
|
2020-05-13 13:24:52 +00:00
|
|
|
// SendPublic takes encoded data, encrypts it and sends through the wire.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) SendPublic(
|
2019-12-02 15:34:05 +00:00
|
|
|
ctx context.Context,
|
|
|
|
chatName string,
|
2020-07-28 13:22:22 +00:00
|
|
|
rawMessage RawMessage,
|
2019-12-02 15:34:05 +00:00
|
|
|
) ([]byte, error) {
|
2020-07-22 07:41:40 +00:00
|
|
|
// Set sender
|
2020-07-21 15:41:10 +00:00
|
|
|
if rawMessage.Sender == nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
rawMessage.Sender = s.identity
|
2020-07-21 15:41:10 +00:00
|
|
|
}
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
wrappedMessage, err := s.wrapMessageV1(&rawMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
|
|
|
}
|
|
|
|
|
2020-07-31 12:22:05 +00:00
|
|
|
var newMessage *types.NewMessage
|
2020-09-02 14:11:16 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
messageSpec, err := s.protocol.BuildPublicMessage(s.identity, wrappedMessage)
|
2020-09-02 14:11:16 +00:00
|
|
|
if err != nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Error("failed to send a public message", zap.Error(err))
|
2020-09-02 14:11:16 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to wrap a public message in the encryption layer")
|
|
|
|
}
|
|
|
|
|
2023-11-08 18:05:33 +00:00
|
|
|
if !rawMessage.SkipEncryptionLayer {
|
2020-07-31 12:22:05 +00:00
|
|
|
newMessage, err = MessageSpecToWhisper(messageSpec)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
newMessage = &types.NewMessage{
|
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: wrappedMessage,
|
|
|
|
PowTarget: calculatePoW(wrappedMessage),
|
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
}
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2022-11-07 20:01:06 +00:00
|
|
|
newMessage.Ephemeral = rawMessage.Ephemeral
|
2023-05-22 21:38:02 +00:00
|
|
|
newMessage.PubsubTopic = rawMessage.PubsubTopic
|
2022-11-07 20:01:06 +00:00
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
messageID := v1protocol.MessageID(&rawMessage.Sender.PublicKey, wrappedMessage)
|
|
|
|
rawMessage.ID = types.EncodeHex(messageID)
|
|
|
|
|
2023-06-20 16:12:59 +00:00
|
|
|
if rawMessage.BeforeDispatch != nil {
|
|
|
|
if err := rawMessage.BeforeDispatch(&rawMessage); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// notify before dispatching
|
2022-07-15 13:19:37 +00:00
|
|
|
s.notifyOnScheduledMessage(nil, &rawMessage)
|
2020-07-22 07:41:40 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err := s.transport.SendPublic(ctx, newMessage, chatName)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
2021-10-29 14:29:28 +00:00
|
|
|
s.logger.Debug("sent public message", zap.String("messageID", messageID.String()), zap.String("hash", types.EncodeHex(hash)))
|
|
|
|
|
2020-09-02 14:11:16 +00:00
|
|
|
sentMessage := &SentMessage{
|
|
|
|
Spec: messageSpec,
|
|
|
|
MessageIDs: [][]byte{messageID},
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.notifyOnSentMessage(sentMessage)
|
2020-09-02 14:11:16 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.transport.Track([][]byte{messageID}, hash, newMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
return messageID, nil
|
|
|
|
}
|
|
|
|
|
2021-02-23 15:47:45 +00:00
|
|
|
// unwrapDatasyncMessage tries to unwrap message as datasync one and in case of success
|
|
|
|
// returns cloned messages with replaced payloads
|
|
|
|
func unwrapDatasyncMessage(m *v1protocol.StatusMessage, datasync *datasync.DataSync) ([]*v1protocol.StatusMessage, [][]byte, error) {
|
|
|
|
var statusMessages []*v1protocol.StatusMessage
|
|
|
|
|
|
|
|
payloads, acks, err := datasync.UnwrapPayloadsAndAcks(
|
|
|
|
m.SigPubKey(),
|
2023-11-08 18:05:33 +00:00
|
|
|
m.EncryptionLayer.Payload,
|
2021-02-23 15:47:45 +00:00
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, payload := range payloads {
|
|
|
|
message, err := m.Clone()
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
2023-11-08 18:05:33 +00:00
|
|
|
message.EncryptionLayer.Payload = payload
|
2021-02-23 15:47:45 +00:00
|
|
|
statusMessages = append(statusMessages, message)
|
|
|
|
}
|
|
|
|
return statusMessages, acks, nil
|
|
|
|
}
|
|
|
|
|
2020-07-06 08:54:22 +00:00
|
|
|
// HandleMessages expects a whisper message as input, and it will go through
|
2019-09-02 09:29:06 +00:00
|
|
|
// a series of transformations until the message is parsed into an application
|
|
|
|
// layer message, or in case of Raw methods, the processing stops at the layer
|
2019-09-26 07:01:17 +00:00
|
|
|
// before.
|
|
|
|
// It returns an error only if the processing of required steps failed.
|
2023-07-04 20:53:53 +00:00
|
|
|
func (s *MessageSender) HandleMessages(shhMessage *types.Message) ([]*v1protocol.StatusMessage, [][]byte, error) {
|
2021-06-23 14:13:48 +00:00
|
|
|
logger := s.logger.With(zap.String("site", "handleMessages"))
|
2019-12-17 20:51:01 +00:00
|
|
|
hlogger := logger.With(zap.ByteString("hash", shhMessage.Hash))
|
2019-11-21 16:19:22 +00:00
|
|
|
var statusMessage v1protocol.StatusMessage
|
2021-02-23 15:47:45 +00:00
|
|
|
var statusMessages []*v1protocol.StatusMessage
|
2022-09-21 16:05:29 +00:00
|
|
|
var acks [][]byte
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2023-11-08 18:05:33 +00:00
|
|
|
err := statusMessage.HandleTransportLayer(shhMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
hlogger.Error("failed to handle transport layer message", zap.Error(err))
|
2021-02-23 15:47:45 +00:00
|
|
|
return nil, nil, err
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
err = s.handleEncryptionLayer(context.Background(), &statusMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
hlogger.Debug("failed to handle an encryption message", zap.Error(err))
|
|
|
|
}
|
|
|
|
|
2022-09-21 16:05:29 +00:00
|
|
|
// Hash ratchet with a group id not found yet
|
2023-11-08 18:05:33 +00:00
|
|
|
if err == encryption.ErrHashRatchetGroupIDNotFound && len(statusMessage.EncryptionLayer.HashRatchetInfo) == 1 {
|
|
|
|
info := statusMessage.EncryptionLayer.HashRatchetInfo[0]
|
2022-09-21 16:05:29 +00:00
|
|
|
err := s.persistence.SaveHashRatchetMessage(info.GroupID, info.KeyID, shhMessage)
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Check if there are undecrypted message
|
2023-11-08 18:05:33 +00:00
|
|
|
for _, hashRatchetInfo := range statusMessage.EncryptionLayer.HashRatchetInfo {
|
2023-10-12 15:45:23 +00:00
|
|
|
messages, err := s.persistence.GetHashRatchetMessages(hashRatchetInfo.KeyID)
|
2022-09-21 16:05:29 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
2023-10-30 11:53:37 +00:00
|
|
|
var processedIds [][]byte
|
2022-09-21 16:05:29 +00:00
|
|
|
for _, message := range messages {
|
|
|
|
var statusMessage v1protocol.StatusMessage
|
2023-11-08 18:05:33 +00:00
|
|
|
err := statusMessage.HandleTransportLayer(message)
|
2022-09-21 16:05:29 +00:00
|
|
|
if err != nil {
|
|
|
|
hlogger.Error("failed to handle transport layer message", zap.Error(err))
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
2023-10-29 08:04:01 +00:00
|
|
|
err = s.handleEncryptionLayer(context.Background(), &statusMessage)
|
|
|
|
if err != nil {
|
|
|
|
hlogger.Debug("failed to handle an encryption message", zap.Error(err))
|
2023-10-30 11:53:37 +00:00
|
|
|
continue
|
2023-10-29 08:04:01 +00:00
|
|
|
}
|
2023-10-30 11:53:37 +00:00
|
|
|
processedIds = append(processedIds, message.Hash)
|
2023-10-29 08:04:01 +00:00
|
|
|
|
2022-09-21 16:05:29 +00:00
|
|
|
stms, as, err := unwrapDatasyncMessage(&statusMessage, s.datasync)
|
|
|
|
if err != nil {
|
|
|
|
hlogger.Debug("failed to handle datasync message", zap.Error(err))
|
|
|
|
//that wasn't a datasync message, so use the original payload
|
2023-10-29 08:04:01 +00:00
|
|
|
statusMessages = append(statusMessages, &statusMessage)
|
2022-09-21 16:05:29 +00:00
|
|
|
|
|
|
|
} else {
|
|
|
|
statusMessages = append(statusMessages, stms...)
|
|
|
|
acks = append(acks, as...)
|
|
|
|
}
|
|
|
|
}
|
2023-10-30 11:53:37 +00:00
|
|
|
err = s.persistence.DeleteHashRatchetMessages(processedIds)
|
|
|
|
if err != nil {
|
|
|
|
s.logger.Warn("failed to delete hash ratchet messages", zap.Error(err))
|
|
|
|
return nil, nil, err
|
|
|
|
|
|
|
|
}
|
2022-09-21 16:05:29 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
stms, as, err := unwrapDatasyncMessage(&statusMessage, s.datasync)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
hlogger.Debug("failed to handle datasync message", zap.Error(err))
|
2021-02-23 15:47:45 +00:00
|
|
|
//that wasn't a datasync message, so use the original payload
|
2023-10-29 08:04:01 +00:00
|
|
|
statusMessages = append(statusMessages, &statusMessage)
|
2022-09-21 16:05:29 +00:00
|
|
|
} else {
|
|
|
|
statusMessages = append(statusMessages, stms...)
|
|
|
|
acks = append(acks, as...)
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
for _, statusMessage := range statusMessages {
|
2023-11-08 18:05:33 +00:00
|
|
|
err := statusMessage.HandleApplicationLayer()
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
hlogger.Error("failed to handle application metadata layer message", zap.Error(err))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-02-23 15:47:45 +00:00
|
|
|
return statusMessages, acks, nil
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// fetchDecryptionKey returns the private key associated with this public key, and returns true if it's an ephemeral key
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) fetchDecryptionKey(destination *ecdsa.PublicKey) (*ecdsa.PrivateKey, bool) {
|
2020-07-21 15:41:10 +00:00
|
|
|
destinationID := types.EncodeHex(crypto.FromECDSAPub(destination))
|
2020-07-22 07:41:40 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.ephemeralKeysMutex.Lock()
|
|
|
|
decryptionKey, ok := s.ephemeralKeys[destinationID]
|
|
|
|
s.ephemeralKeysMutex.Unlock()
|
2020-07-22 07:41:40 +00:00
|
|
|
|
|
|
|
// the key is not there, fallback on identity
|
2020-07-21 15:41:10 +00:00
|
|
|
if !ok {
|
2021-06-23 14:13:48 +00:00
|
|
|
return s.identity, false
|
2020-07-21 15:41:10 +00:00
|
|
|
}
|
2020-07-22 07:41:40 +00:00
|
|
|
return decryptionKey, true
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) handleEncryptionLayer(ctx context.Context, message *v1protocol.StatusMessage) error {
|
|
|
|
logger := s.logger.With(zap.String("site", "handleEncryptionLayer"))
|
2020-07-22 07:41:40 +00:00
|
|
|
publicKey := message.SigPubKey()
|
|
|
|
|
|
|
|
// if it's an ephemeral key, we don't negotiate a topic
|
2023-11-08 18:05:33 +00:00
|
|
|
decryptionKey, skipNegotiation := s.fetchDecryptionKey(message.TransportLayer.Dst)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
2023-11-08 18:05:33 +00:00
|
|
|
err := message.HandleEncryptionLayer(decryptionKey, publicKey, s.protocol, skipNegotiation)
|
2020-07-22 07:41:40 +00:00
|
|
|
|
|
|
|
// if it's an ephemeral key, we don't have to handle a device not found error
|
|
|
|
if err == encryption.ErrDeviceNotFound && !skipNegotiation {
|
2021-06-23 14:13:48 +00:00
|
|
|
if err := s.handleErrDeviceNotFound(ctx, publicKey); err != nil {
|
2019-11-04 10:08:22 +00:00
|
|
|
logger.Error("failed to handle ErrDeviceNotFound", zap.Error(err))
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
if err != nil {
|
2023-10-29 08:04:01 +00:00
|
|
|
logger.Error("failed to handle an encrypted message", zap.Error(err))
|
|
|
|
return err
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) handleErrDeviceNotFound(ctx context.Context, publicKey *ecdsa.PublicKey) error {
|
2019-09-02 09:29:06 +00:00
|
|
|
now := time.Now().Unix()
|
2021-06-23 14:13:48 +00:00
|
|
|
advertise, err := s.protocol.ShouldAdvertiseBundle(publicKey, now)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
if !advertise {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
messageSpec, err := s.protocol.BuildBundleAdvertiseMessage(s.identity, publicKey)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
ctx, cancel := context.WithTimeout(ctx, time.Second)
|
|
|
|
defer cancel()
|
2020-06-30 07:50:59 +00:00
|
|
|
// We don't pass an array of messageIDs as no action needs to be taken
|
|
|
|
// when sending a bundle
|
2021-06-23 14:13:48 +00:00
|
|
|
_, _, err = s.sendMessageSpec(ctx, publicKey, messageSpec, nil)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.protocol.ConfirmBundleAdvertisement(publicKey, now)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) wrapMessageV1(rawMessage *RawMessage) ([]byte, error) {
|
2020-07-21 15:41:10 +00:00
|
|
|
wrappedMessage, err := v1protocol.WrapMessageV1(rawMessage.Payload, rawMessage.MessageType, rawMessage.Sender)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to wrap message")
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
Move to protobuf for Message type (#1706)
* Use a single Message type `v1/message.go` and `message.go` are the same now, and they embed `protobuf.ChatMessage`
* Use `SendChatMessage` for sending chat messages, this is basically the old `Send` but a bit more flexible so we can send different message types (stickers,commands), and not just text.
* Remove dedup from services/shhext. Because now we process in status-protocol, dedup makes less sense, as those messages are going to be processed anyway, so removing for now, we can re-evaluate if bringing it to status-go or not.
* Change the various retrieveX method to a single one:
`RetrieveAll` will be processing those messages that it can process (Currently only `Message`), and return the rest in `RawMessages` (still transit). The format for the response is:
`Chats`: -> The chats updated by receiving the message
`Messages`: -> The messages retrieved (already matched to a chat)
`Contacts`: -> The contacts updated by the messages
`RawMessages` -> Anything else that can't be parsed, eventually as we move everything to status-protocol-go this will go away.
2019-12-05 16:25:34 +00:00
|
|
|
return wrappedMessage, nil
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) addToDataSync(publicKey *ecdsa.PublicKey, message []byte) ([]byte, error) {
|
|
|
|
groupID := datasync.ToOneToOneGroupID(&s.identity.PublicKey, publicKey)
|
2019-09-02 09:29:06 +00:00
|
|
|
peerID := datasyncpeer.PublicKeyToPeerID(*publicKey)
|
2021-06-23 14:13:48 +00:00
|
|
|
exist, err := s.datasync.IsPeerInGroup(groupID, peerID)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
2021-02-23 15:47:45 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to check if peer is in group")
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
if !exist {
|
2021-06-23 14:13:48 +00:00
|
|
|
if err := s.datasync.AddPeer(groupID, peerID); err != nil {
|
2021-02-23 15:47:45 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to add peer")
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
}
|
2021-06-23 14:13:48 +00:00
|
|
|
id, err := s.datasync.AppendMessage(groupID, message)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
2021-02-23 15:47:45 +00:00
|
|
|
return nil, errors.Wrap(err, "failed to append message to datasync")
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-02-23 15:47:45 +00:00
|
|
|
return id[:], nil
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// sendDataSync sends a message scheduled by the data sync layer.
|
2019-09-26 07:01:17 +00:00
|
|
|
// Data Sync layer calls this method "dispatch" function.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) sendDataSync(ctx context.Context, publicKey *ecdsa.PublicKey, marshalledDatasyncPayload []byte, payload *datasyncproto.Payload) error {
|
2020-07-10 13:26:06 +00:00
|
|
|
// Calculate the messageIDs
|
2019-09-02 09:29:06 +00:00
|
|
|
messageIDs := make([][]byte, 0, len(payload.Messages))
|
2021-10-29 14:29:28 +00:00
|
|
|
hexMessageIDs := make([]string, 0, len(payload.Messages))
|
2019-09-02 09:29:06 +00:00
|
|
|
for _, payload := range payload.Messages {
|
2021-10-29 14:29:28 +00:00
|
|
|
mid := v1protocol.MessageID(&s.identity.PublicKey, payload.Body)
|
|
|
|
messageIDs = append(messageIDs, mid)
|
|
|
|
hexMessageIDs = append(hexMessageIDs, mid.String())
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2021-09-21 15:47:04 +00:00
|
|
|
messageSpec, err := s.protocol.BuildEncryptedMessage(s.identity, publicKey, marshalledDatasyncPayload)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return errors.Wrap(err, "failed to encrypt message")
|
|
|
|
}
|
|
|
|
|
2020-07-31 12:22:05 +00:00
|
|
|
// The shared secret needs to be handle before we send a message
|
|
|
|
// otherwise the topic might not be set up before we receive a message
|
2021-06-23 14:13:48 +00:00
|
|
|
if s.handleSharedSecrets != nil {
|
|
|
|
err := s.handleSharedSecrets([]*sharedsecret.Secret{messageSpec.SharedSecret})
|
2020-07-31 12:22:05 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, newMessage, err := s.sendMessageSpec(ctx, publicKey, messageSpec, messageIDs)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
2021-06-23 14:13:48 +00:00
|
|
|
s.logger.Error("failed to send a datasync message", zap.Error(err))
|
2019-09-02 09:29:06 +00:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2021-10-29 14:29:28 +00:00
|
|
|
s.logger.Debug("sent private messages", zap.Any("messageIDs", hexMessageIDs), zap.String("hash", types.EncodeHex(hash)))
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.transport.Track(messageIDs, hash, newMessage)
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2020-07-31 12:22:05 +00:00
|
|
|
// sendPrivateRawMessage sends a message not wrapped in an encryption layer
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) sendPrivateRawMessage(ctx context.Context, rawMessage *RawMessage, publicKey *ecdsa.PublicKey, payload []byte, messageIDs [][]byte) ([]byte, *types.NewMessage, error) {
|
2020-07-21 15:41:10 +00:00
|
|
|
newMessage := &types.NewMessage{
|
2023-05-22 21:38:02 +00:00
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: payload,
|
|
|
|
PowTarget: calculatePoW(payload),
|
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
PubsubTopic: rawMessage.PubsubTopic,
|
2020-07-21 15:41:10 +00:00
|
|
|
}
|
2021-01-18 09:12:03 +00:00
|
|
|
var hash []byte
|
|
|
|
var err error
|
2020-07-21 15:41:10 +00:00
|
|
|
|
2021-01-18 09:12:03 +00:00
|
|
|
if rawMessage.SendOnPersonalTopic {
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err = s.transport.SendPrivateOnPersonalTopic(ctx, newMessage, publicKey)
|
2021-01-18 09:12:03 +00:00
|
|
|
} else {
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err = s.transport.SendPrivateWithPartitioned(ctx, newMessage, publicKey)
|
2021-01-18 09:12:03 +00:00
|
|
|
}
|
2020-07-21 15:41:10 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return hash, newMessage, nil
|
|
|
|
}
|
|
|
|
|
2022-09-05 12:08:44 +00:00
|
|
|
// sendCommunityMessage sends a message not wrapped in an encryption layer
|
2021-01-11 10:32:51 +00:00
|
|
|
// to a community
|
2023-05-22 21:38:02 +00:00
|
|
|
func (s *MessageSender) dispatchCommunityMessage(ctx context.Context, publicKey *ecdsa.PublicKey, payload []byte, messageIDs [][]byte, pubsubTopic string) ([]byte, *types.NewMessage, error) {
|
2021-01-11 10:32:51 +00:00
|
|
|
newMessage := &types.NewMessage{
|
2023-05-22 21:38:02 +00:00
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: payload,
|
|
|
|
PowTarget: calculatePoW(payload),
|
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
PubsubTopic: pubsubTopic,
|
2021-01-11 10:32:51 +00:00
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err := s.transport.SendCommunityMessage(ctx, newMessage, publicKey)
|
2021-01-11 10:32:51 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
return hash, newMessage, nil
|
|
|
|
}
|
|
|
|
|
2019-09-26 07:01:17 +00:00
|
|
|
// sendMessageSpec analyses the spec properties and selects a proper transport method.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) sendMessageSpec(ctx context.Context, publicKey *ecdsa.PublicKey, messageSpec *encryption.ProtocolMessageSpec, messageIDs [][]byte) ([]byte, *types.NewMessage, error) {
|
2020-07-06 08:54:22 +00:00
|
|
|
newMessage, err := MessageSpecToWhisper(messageSpec)
|
2019-09-02 09:29:06 +00:00
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
logger := s.logger.With(zap.String("site", "sendMessageSpec"))
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
var hash []byte
|
|
|
|
|
2020-07-31 12:22:05 +00:00
|
|
|
// process shared secret
|
|
|
|
if messageSpec.AgreedSecret {
|
2019-09-02 09:29:06 +00:00
|
|
|
logger.Debug("sending using shared secret")
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err = s.transport.SendPrivateWithSharedSecret(ctx, newMessage, publicKey, messageSpec.SharedSecret.Key)
|
2020-07-31 12:22:05 +00:00
|
|
|
} else {
|
2019-09-02 09:29:06 +00:00
|
|
|
logger.Debug("sending partitioned topic")
|
2021-06-23 14:13:48 +00:00
|
|
|
hash, err = s.transport.SendPrivateWithPartitioned(ctx, newMessage, publicKey)
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
if err != nil {
|
|
|
|
return nil, nil, err
|
|
|
|
}
|
|
|
|
|
2020-07-06 08:54:22 +00:00
|
|
|
sentMessage := &SentMessage{
|
|
|
|
PublicKey: publicKey,
|
|
|
|
Spec: messageSpec,
|
|
|
|
MessageIDs: messageIDs,
|
|
|
|
}
|
2020-06-30 07:50:59 +00:00
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
s.notifyOnSentMessage(sentMessage)
|
2020-07-22 07:41:40 +00:00
|
|
|
|
|
|
|
return hash, newMessage, nil
|
|
|
|
}
|
|
|
|
|
2021-10-07 11:10:23 +00:00
|
|
|
func (s *MessageSender) SubscribeToMessageEvents() <-chan *MessageEvent {
|
|
|
|
c := make(chan *MessageEvent, 100)
|
|
|
|
s.messageEventsSubscriptions = append(s.messageEventsSubscriptions, c)
|
2020-07-22 07:41:40 +00:00
|
|
|
return c
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) notifyOnSentMessage(sentMessage *SentMessage) {
|
2021-10-07 11:10:23 +00:00
|
|
|
event := &MessageEvent{
|
|
|
|
Type: MessageSent,
|
|
|
|
SentMessage: sentMessage,
|
|
|
|
}
|
2020-07-06 08:54:22 +00:00
|
|
|
// Publish on channels, drop if buffer is full
|
2021-10-07 11:10:23 +00:00
|
|
|
for _, c := range s.messageEventsSubscriptions {
|
2020-07-06 08:54:22 +00:00
|
|
|
select {
|
2021-10-07 11:10:23 +00:00
|
|
|
case c <- event:
|
2020-07-06 08:54:22 +00:00
|
|
|
default:
|
2021-10-07 11:10:23 +00:00
|
|
|
s.logger.Warn("message events subscription channel full when publishing sent event, dropping message")
|
2020-06-30 07:50:59 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-09-02 09:29:06 +00:00
|
|
|
}
|
|
|
|
|
2022-07-15 13:19:37 +00:00
|
|
|
func (s *MessageSender) notifyOnScheduledMessage(recipient *ecdsa.PublicKey, message *RawMessage) {
|
2021-10-07 11:10:23 +00:00
|
|
|
event := &MessageEvent{
|
2022-07-15 13:19:37 +00:00
|
|
|
Recipient: recipient,
|
2021-10-07 11:10:23 +00:00
|
|
|
Type: MessageScheduled,
|
|
|
|
RawMessage: message,
|
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// Publish on channels, drop if buffer is full
|
2021-10-07 11:10:23 +00:00
|
|
|
for _, c := range s.messageEventsSubscriptions {
|
2020-07-22 07:41:40 +00:00
|
|
|
select {
|
2021-10-07 11:10:23 +00:00
|
|
|
case c <- event:
|
2020-07-22 07:41:40 +00:00
|
|
|
default:
|
2021-10-07 11:10:23 +00:00
|
|
|
s.logger.Warn("message events subscription channel full when publishing scheduled event, dropping message")
|
2020-07-22 07:41:40 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) JoinPublic(id string) (*transport.Filter, error) {
|
|
|
|
return s.transport.JoinPublic(id)
|
2020-07-09 16:52:26 +00:00
|
|
|
}
|
|
|
|
|
2020-07-22 07:41:40 +00:00
|
|
|
// AddEphemeralKey adds an ephemeral key that we will be listening to
|
|
|
|
// note that we never removed them from now, as waku/whisper does not
|
|
|
|
// recalculate topics on removal, so effectively there's no benefit.
|
|
|
|
// On restart they will be gone.
|
2021-06-23 14:13:48 +00:00
|
|
|
func (s *MessageSender) AddEphemeralKey(privateKey *ecdsa.PrivateKey) (*transport.Filter, error) {
|
|
|
|
s.ephemeralKeysMutex.Lock()
|
|
|
|
s.ephemeralKeys[types.EncodeHex(crypto.FromECDSAPub(&privateKey.PublicKey))] = privateKey
|
|
|
|
s.ephemeralKeysMutex.Unlock()
|
|
|
|
return s.transport.LoadKeyFilters(privateKey)
|
2020-07-21 15:41:10 +00:00
|
|
|
}
|
|
|
|
|
2020-07-06 08:54:22 +00:00
|
|
|
func MessageSpecToWhisper(spec *encryption.ProtocolMessageSpec) (*types.NewMessage, error) {
|
2019-11-23 17:57:05 +00:00
|
|
|
var newMessage *types.NewMessage
|
2019-09-02 09:29:06 +00:00
|
|
|
|
|
|
|
payload, err := proto.Marshal(spec.Message)
|
|
|
|
if err != nil {
|
|
|
|
return newMessage, err
|
|
|
|
}
|
|
|
|
|
2019-11-23 17:57:05 +00:00
|
|
|
newMessage = &types.NewMessage{
|
2019-09-02 09:29:06 +00:00
|
|
|
TTL: whisperTTL,
|
|
|
|
Payload: payload,
|
2020-05-13 13:19:29 +00:00
|
|
|
PowTarget: calculatePoW(payload),
|
2019-09-02 09:29:06 +00:00
|
|
|
PowTime: whisperPoWTime,
|
|
|
|
}
|
|
|
|
return newMessage, nil
|
|
|
|
}
|
|
|
|
|
2020-05-13 13:19:29 +00:00
|
|
|
// calculatePoW returns the PoWTarget to be used.
|
2020-05-14 05:40:40 +00:00
|
|
|
// We check the size and arbitrarily set it to a lower PoW if the packet is
|
|
|
|
// greater than 50KB. We do this as the defaultPoW is too high for clients to send
|
2020-05-13 13:19:29 +00:00
|
|
|
// large messages.
|
|
|
|
func calculatePoW(payload []byte) float64 {
|
|
|
|
if len(payload) > largeSizeInBytes {
|
|
|
|
return whisperLargeSizePoW
|
|
|
|
}
|
|
|
|
return whisperDefaultPoW
|
|
|
|
}
|
2021-10-21 12:39:19 +00:00
|
|
|
|
|
|
|
func (s *MessageSender) StopDatasync() {
|
|
|
|
s.datasync.Stop()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (s *MessageSender) StartDatasync() {
|
|
|
|
dataSyncTransport := datasync.NewNodeTransport()
|
|
|
|
dataSyncNode, err := datasyncnode.NewPersistentNode(
|
|
|
|
s.database,
|
|
|
|
dataSyncTransport,
|
|
|
|
datasyncpeer.PublicKeyToPeerID(s.identity.PublicKey),
|
|
|
|
datasyncnode.BATCH,
|
|
|
|
datasync.CalculateSendTime,
|
|
|
|
s.logger,
|
|
|
|
)
|
|
|
|
if err != nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
ds := datasync.New(dataSyncNode, dataSyncTransport, true, s.logger)
|
|
|
|
|
|
|
|
if s.datasyncEnabled {
|
|
|
|
ds.Init(s.sendDataSync, s.transport.MaxMessageSize()/4*3, s.logger)
|
|
|
|
ds.Start(datasync.DatasyncTicker)
|
|
|
|
}
|
|
|
|
|
|
|
|
s.datasync = ds
|
|
|
|
}
|
2023-05-24 14:25:10 +00:00
|
|
|
|
2023-06-28 10:53:46 +00:00
|
|
|
// GetCurrentKeyForGroup returns the latest key timestampID belonging to a key group
|
2023-10-12 15:45:23 +00:00
|
|
|
func (s *MessageSender) GetCurrentKeyForGroup(groupID []byte) (*encryption.HashRatchetKeyCompatibility, error) {
|
2023-06-28 10:53:46 +00:00
|
|
|
return s.protocol.GetCurrentKeyForGroup(groupID)
|
|
|
|
}
|
|
|
|
|
2023-05-24 14:25:10 +00:00
|
|
|
// GetKeyIDsForGroup returns a slice of key IDs belonging to a given group ID
|
2023-10-12 15:45:23 +00:00
|
|
|
func (s *MessageSender) GetKeysForGroup(groupID []byte) ([]*encryption.HashRatchetKeyCompatibility, error) {
|
|
|
|
return s.protocol.GetKeysForGroup(groupID)
|
2023-05-24 14:25:10 +00:00
|
|
|
}
|