//go:build nimbus // +build nimbus package shhext import ( "context" "github.com/status-im/status-go/protocol/common" "github.com/ethereum/go-ethereum/log" "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" "github.com/status-im/status-go/protocol/encryption/multidevice" "github.com/status-im/status-go/protocol/transport" ) // ----- // PUBLIC API // ----- // NimbusPublicAPI extends whisper public API. type NimbusPublicAPI struct { service *NimbusService publicAPI types.PublicWhisperAPI log log.Logger } // NewPublicAPI returns instance of the public API. func NewNimbusPublicAPI(s *NimbusService) *NimbusPublicAPI { return &NimbusPublicAPI{ service: s, publicAPI: s.w.PublicWhisperAPI(), log: log.New("package", "status-go/services/sshext.NimbusPublicAPI"), } } // func (api *NimbusPublicAPI) getPeer(rawurl string) (*enode.Node, error) { // if len(rawurl) == 0 { // return mailservers.GetFirstConnected(api.service.server, api.service.peerStore) // } // return enode.ParseV4(rawurl) // } // // RetryConfig specifies configuration for retries with timeout and max amount of retries. // type RetryConfig struct { // BaseTimeout time.Duration // // StepTimeout defines duration increase per each retry. // StepTimeout time.Duration // MaxRetries int // } // RequestMessagesSync repeats MessagesRequest using configuration in retry conf. // func (api *NimbusPublicAPI) RequestMessagesSync(conf RetryConfig, r MessagesRequest) (MessagesResponse, error) { // var resp MessagesResponse // shh := api.service.w // events := make(chan types.EnvelopeEvent, 10) // var ( // requestID types.HexBytes // err error // retries int // ) // for retries <= conf.MaxRetries { // sub := shh.SubscribeEnvelopeEvents(events) // r.Timeout = conf.BaseTimeout + conf.StepTimeout*time.Duration(retries) // timeout := r.Timeout // // FIXME this weird conversion is required because MessagesRequest expects seconds but defines time.Duration // r.Timeout = time.Duration(int(r.Timeout.Seconds())) // requestID, err = api.RequestMessages(context.Background(), r) // if err != nil { // sub.Unsubscribe() // return resp, err // } // mailServerResp, err := waitForExpiredOrCompleted(types.BytesToHash(requestID), events, timeout) // sub.Unsubscribe() // if err == nil { // resp.Cursor = hex.EncodeToString(mailServerResp.Cursor) // resp.Error = mailServerResp.Error // return resp, nil // } // retries++ // api.log.Error("[RequestMessagesSync] failed", "err", err, "retries", retries) // } // return resp, fmt.Errorf("failed to request messages after %d retries", retries) // } // func waitForExpiredOrCompleted(requestID types.Hash, events chan types.EnvelopeEvent, timeout time.Duration) (*types.MailServerResponse, error) { // expired := fmt.Errorf("request %x expired", requestID) // after := time.NewTimer(timeout) // defer after.Stop() // for { // var ev types.EnvelopeEvent // select { // case ev = <-events: // case <-after.C: // return nil, expired // } // if ev.Hash != requestID { // continue // } // switch ev.Event { // case types.EventMailServerRequestCompleted: // data, ok := ev.Data.(*types.MailServerResponse) // if ok { // return data, nil // } // return nil, errors.New("invalid event data type") // case types.EventMailServerRequestExpired: // return nil, expired // } // } // } // // RequestMessages sends a request for historic messages to a MailServer. // func (api *NimbusPublicAPI) RequestMessages(_ context.Context, r MessagesRequest) (types.HexBytes, error) { // api.log.Info("RequestMessages", "request", r) // shh := api.service.w // now := api.service.w.GetCurrentTime() // r.setDefaults(now) // if r.From > r.To { // return nil, fmt.Errorf("Query range is invalid: from > to (%d > %d)", r.From, r.To) // } // mailServerNode, err := api.getPeer(r.MailServerPeer) // if err != nil { // return nil, fmt.Errorf("%v: %v", ErrInvalidMailServerPeer, err) // } // var ( // symKey []byte // publicKey *ecdsa.PublicKey // ) // if r.SymKeyID != "" { // symKey, err = shh.GetSymKey(r.SymKeyID) // if err != nil { // return nil, fmt.Errorf("%v: %v", ErrInvalidSymKeyID, err) // } // } else { // publicKey = mailServerNode.Pubkey() // } // payload, err := makeMessagesRequestPayload(r) // if err != nil { // return nil, err // } // envelope, err := makeEnvelop( // payload, // symKey, // publicKey, // api.service.nodeID, // shh.MinPow(), // now, // ) // if err != nil { // return nil, err // } // hash := envelope.Hash() // if !r.Force { // err = api.service.requestsRegistry.Register(hash, r.Topics) // if err != nil { // return nil, err // } // } // if err := shh.RequestHistoricMessagesWithTimeout(mailServerNode.ID().Bytes(), envelope, r.Timeout*time.Second); err != nil { // if !r.Force { // api.service.requestsRegistry.Unregister(hash) // } // return nil, err // } // return hash[:], nil // } // // createSyncMailRequest creates SyncMailRequest. It uses a full bloom filter // // if no topics are given. // func createSyncMailRequest(r SyncMessagesRequest) (types.SyncMailRequest, error) { // var bloom []byte // if len(r.Topics) > 0 { // bloom = topicsToBloom(r.Topics...) // } else { // bloom = types.MakeFullNodeBloom() // } // cursor, err := hex.DecodeString(r.Cursor) // if err != nil { // return types.SyncMailRequest{}, err // } // return types.SyncMailRequest{ // Lower: r.From, // Upper: r.To, // Bloom: bloom, // Limit: r.Limit, // Cursor: cursor, // }, nil // } // func createSyncMessagesResponse(r types.SyncEventResponse) SyncMessagesResponse { // return SyncMessagesResponse{ // Cursor: hex.EncodeToString(r.Cursor), // Error: r.Error, // } // } // // SyncMessages sends a request to a given MailServerPeer to sync historic messages. // // MailServerPeers needs to be added as a trusted peer first. // func (api *NimbusPublicAPI) SyncMessages(ctx context.Context, r SyncMessagesRequest) (SyncMessagesResponse, error) { // log.Info("SyncMessages start", "request", r) // var response SyncMessagesResponse // mailServerEnode, err := enode.ParseV4(r.MailServerPeer) // if err != nil { // return response, fmt.Errorf("invalid MailServerPeer: %v", err) // } // mailServerID := mailServerEnode.ID().Bytes() // request, err := createSyncMailRequest(r) // if err != nil { // return response, fmt.Errorf("failed to create a sync mail request: %v", err) // } // for { // log.Info("Sending a request to sync messages", "request", request) // resp, err := api.service.syncMessages(ctx, mailServerID, request) // if err != nil { // return response, err // } // log.Info("Syncing messages response", "error", resp.Error, "cursor", fmt.Sprintf("%#x", resp.Cursor)) // if resp.Error != "" || len(resp.Cursor) == 0 || !r.FollowCursor { // return createSyncMessagesResponse(resp), nil // } // request.Cursor = resp.Cursor // } // } // ConfirmMessagesProcessedByID is a method to confirm that messages was consumed by // the client side. // TODO: this is broken now as it requires dedup ID while a message hash should be used. func (api *NimbusPublicAPI) ConfirmMessagesProcessedByID(messageConfirmations []*Metadata) error { confirmationCount := len(messageConfirmations) dedupIDs := make([][]byte, confirmationCount) encryptionIDs := make([][]byte, confirmationCount) for i, confirmation := range messageConfirmations { dedupIDs[i] = confirmation.DedupID encryptionIDs[i] = confirmation.EncryptionID } return api.service.ConfirmMessagesProcessed(encryptionIDs) } // Post is used to send one-to-one for those who did not enabled device-to-device sync, // in other words don't use PFS-enabled messages. Otherwise, SendDirectMessage is used. // It's important to call NimbusPublicAPI.afterSend() so that the client receives a signal // with confirmation that the message left the device. func (api *NimbusPublicAPI) Post(ctx context.Context, newMessage types.NewMessage) (types.HexBytes, error) { return api.publicAPI.Post(ctx, newMessage) } func (api *NimbusPublicAPI) Join(chat protocol.Chat) error { return api.service.messenger.Join(chat) } func (api *NimbusPublicAPI) Leave(chat protocol.Chat) error { return api.service.messenger.Leave(chat) } func (api *NimbusPublicAPI) LeaveGroupChat(ctx Context, chatID string) (*protocol.MessengerResponse, error) { return api.service.messenger.LeaveGroupChat(ctx, chatID) } func (api *NimbusPublicAPI) CreateGroupChatWithMembers(ctx Context, name string, members []string) (*protocol.MessengerResponse, error) { return api.service.messenger.CreateGroupChatWithMembers(ctx, name, members) } func (api *NimbusPublicAPI) AddMembersToGroupChat(ctx Context, chatID string, members []string) (*protocol.MessengerResponse, error) { return api.service.messenger.AddMembersToGroupChat(ctx, chatID, members) } func (api *NimbusPublicAPI) RemoveMemberFromGroupChat(ctx Context, chatID string, member string) (*protocol.MessengerResponse, error) { return api.service.messenger.RemoveMembersFromGroupChat(ctx, chatID, []string{member}) } func (api *NimbusPublicAPI) AddAdminsToGroupChat(ctx Context, chatID string, members []string) (*protocol.MessengerResponse, error) { return api.service.messenger.AddAdminsToGroupChat(ctx, chatID, members) } func (api *NimbusPublicAPI) ConfirmJoiningGroup(ctx context.Context, chatID string) (*protocol.MessengerResponse, error) { return api.service.messenger.ConfirmJoiningGroup(ctx, chatID) } // func (api *NimbusPublicAPI) requestMessagesUsingPayload(request db.HistoryRequest, peer, symkeyID string, payload []byte, force bool, timeout time.Duration, topics []types.TopicType) (hash types.Hash, err error) { // shh := api.service.w // now := api.service.w.GetCurrentTime() // mailServerNode, err := api.getPeer(peer) // if err != nil { // return hash, fmt.Errorf("%v: %v", ErrInvalidMailServerPeer, err) // } // var ( // symKey []byte // publicKey *ecdsa.PublicKey // ) // if symkeyID != "" { // symKey, err = shh.GetSymKey(symkeyID) // if err != nil { // return hash, fmt.Errorf("%v: %v", ErrInvalidSymKeyID, err) // } // } else { // publicKey = mailServerNode.Pubkey() // } // envelope, err := makeEnvelop( // payload, // symKey, // publicKey, // api.service.nodeID, // shh.MinPow(), // now, // ) // if err != nil { // return hash, err // } // hash = envelope.Hash() // err = request.Replace(hash) // if err != nil { // return hash, err // } // if !force { // err = api.service.requestsRegistry.Register(hash, topics) // if err != nil { // return hash, err // } // } // if err := shh.RequestHistoricMessagesWithTimeout(mailServerNode.ID().Bytes(), envelope, timeout); err != nil { // if !force { // api.service.requestsRegistry.Unregister(hash) // } // return hash, err // } // return hash, nil // } // // InitiateHistoryRequests is a stateful API for initiating history request for each topic. // // Caller of this method needs to define only two parameters per each TopicRequest: // // - Topic // // - Duration in nanoseconds. Will be used to determine starting time for history request. // // After that status-go will guarantee that request for this topic and date will be performed. // func (api *NimbusPublicAPI) InitiateHistoryRequests(parent context.Context, request InitiateHistoryRequestParams) (rst []types.HexBytes, err error) { // tx := api.service.storage.NewTx() // defer func() { // if err == nil { // err = tx.Commit() // } // }() // ctx := NewContextFromService(parent, api.service, tx) // requests, err := api.service.historyUpdates.CreateRequests(ctx, request.Requests) // if err != nil { // return nil, err // } // var ( // payload []byte // hash types.Hash // ) // for i := range requests { // req := requests[i] // options := CreateTopicOptionsFromRequest(req) // bloom := options.ToBloomFilterOption() // payload, err = bloom.ToMessagesRequestPayload() // if err != nil { // return rst, err // } // hash, err = api.requestMessagesUsingPayload(req, request.Peer, request.SymKeyID, payload, request.Force, request.Timeout, options.Topics()) // if err != nil { // return rst, err // } // rst = append(rst, hash.Bytes()) // } // return rst, err // } // // CompleteRequest client must mark request completed when all envelopes were processed. // func (api *NimbusPublicAPI) CompleteRequest(parent context.Context, hex string) (err error) { // tx := api.service.storage.NewTx() // ctx := NewContextFromService(parent, api.service, tx) // err = api.service.historyUpdates.UpdateFinishedRequest(ctx, types.HexToHash(hex)) // if err == nil { // return tx.Commit() // } // return err // } func (api *NimbusPublicAPI) LoadFilters(parent context.Context, chats []*transport.Filter) ([]*transport.Filter, error) { return api.service.messenger.LoadFilters(chats) } func (api *NimbusPublicAPI) SaveChat(parent context.Context, chat *protocol.Chat) error { api.log.Info("saving chat", "chat", chat) return api.service.messenger.SaveChat(chat) } func (api *NimbusPublicAPI) Chats(parent context.Context) []*protocol.Chat { return api.service.messenger.Chats() } func (api *NimbusPublicAPI) DeleteChat(parent context.Context, chatID string) error { return api.service.messenger.DeleteChat(chatID) } func (api *NimbusPublicAPI) SaveContact(parent context.Context, contact *protocol.Contact) error { return api.service.messenger.SaveContact(contact) } func (api *NimbusPublicAPI) BlockContact(parent context.Context, contact *protocol.Contact) (*protocol.MessengerResponse, error) { api.log.Info("blocking contact", "contact", contact.ID) return api.service.messenger.BlockContact(contact) } func (api *NimbusPublicAPI) Contacts(parent context.Context) []*protocol.Contact { return api.service.messenger.Contacts() } func (api *NimbusPublicAPI) RemoveFilters(parent context.Context, chats []*transport.Filter) error { return api.service.messenger.RemoveFilters(chats) } // EnableInstallation enables an installation for multi-device sync. func (api *NimbusPublicAPI) EnableInstallation(installationID string) error { return api.service.messenger.EnableInstallation(installationID) } // DisableInstallation disables an installation for multi-device sync. func (api *NimbusPublicAPI) DisableInstallation(installationID string) error { return api.service.messenger.DisableInstallation(installationID) } // GetOurInstallations returns all the installations available given an identity func (api *NimbusPublicAPI) GetOurInstallations() []*multidevice.Installation { return api.service.messenger.Installations() } // SetInstallationMetadata sets the metadata for our own installation func (api *NimbusPublicAPI) SetInstallationMetadata(installationID string, data *multidevice.InstallationMetadata) error { return api.service.messenger.SetInstallationMetadata(installationID, data) } // VerifyENSNames takes a list of ensdetails and returns whether they match the public key specified func (api *NimbusPublicAPI) VerifyENSNames(details []enstypes.ENSDetails) (map[string]enstypes.ENSResponse, error) { return api.service.messenger.VerifyENSNames(api.service.config.VerifyENSURL, ensContractAddress, details) } type ApplicationMessagesResponse struct { Messages []*common.Message `json:"messages"` Cursor string `json:"cursor"` } func (api *NimbusPublicAPI) ChatMessages(chatID, cursor string, limit int) (*ApplicationMessagesResponse, error) { messages, cursor, err := api.service.messenger.MessageByChatID(chatID, cursor, limit) if err != nil { return nil, err } return &ApplicationMessagesResponse{ Messages: messages, Cursor: cursor, }, nil } func (api *NimbusPublicAPI) DeleteMessage(id string) error { return api.service.messenger.DeleteMessage(id) } func (api *NimbusPublicAPI) DeleteMessagesByChatID(id string) error { return api.service.messenger.DeleteMessagesByChatID(id) } func (api *NimbusPublicAPI) MarkMessagesSeen(chatID string, ids []string) error { return api.service.messenger.MarkMessagesSeen(chatID, ids) } func (api *NimbusPublicAPI) UpdateMessageOutgoingStatus(id, newOutgoingStatus string) error { return api.service.messenger.UpdateMessageOutgoingStatus(id, newOutgoingStatus) } func (api *PublicAPI) StartMessenger() error { return api.service.StartMessenger() } func (api *NimbusPublicAPI) SendChatMessage(ctx context.Context, message *common.Message) (*protocol.MessengerResponse, error) { return api.service.messenger.SendChatMessage(ctx, message) } func (api *NimbusPublicAPI) SendPinMessage(ctx context.Context, message *common.PinMessage) (*protocol.MessengerResponse, error) { return api.service.messenger.SendPinMessage(ctx, message) } func (api *NimbusPublicAPI) ReSendChatMessage(ctx context.Context, messageID string) error { return api.service.messenger.ReSendChatMessage(ctx, messageID) } func (api *NimbusPublicAPI) RequestTransaction(ctx context.Context, chatID, value, contract, address string) (*protocol.MessengerResponse, error) { return api.service.messenger.RequestTransaction(ctx, chatID, value, contract, address) } func (api *NimbusPublicAPI) RequestAddressForTransaction(ctx context.Context, chatID, from, value, contract string) (*protocol.MessengerResponse, error) { return api.service.messenger.RequestAddressForTransaction(ctx, chatID, from, value, contract) } func (api *NimbusPublicAPI) DeclineRequestAddressForTransaction(ctx context.Context, messageID string) (*protocol.MessengerResponse, error) { return api.service.messenger.DeclineRequestAddressForTransaction(ctx, messageID) } func (api *NimbusPublicAPI) DeclineRequestTransaction(ctx context.Context, messageID string) (*protocol.MessengerResponse, error) { return api.service.messenger.DeclineRequestTransaction(ctx, messageID) } func (api *NimbusPublicAPI) AcceptRequestAddressForTransaction(ctx context.Context, messageID, address string) (*protocol.MessengerResponse, error) { return api.service.messenger.AcceptRequestAddressForTransaction(ctx, messageID, address) } func (api *NimbusPublicAPI) SendTransaction(ctx context.Context, chatID, value, contract, transactionHash string, signature types.HexBytes) (*protocol.MessengerResponse, error) { return api.service.messenger.SendTransaction(ctx, chatID, value, contract, transactionHash, signature) } func (api *NimbusPublicAPI) AcceptRequestTransaction(ctx context.Context, transactionHash, messageID string, signature types.HexBytes) (*protocol.MessengerResponse, error) { return api.service.messenger.AcceptRequestTransaction(ctx, transactionHash, messageID, signature) } func (api *NimbusPublicAPI) SendContactUpdates(ctx context.Context, name, picture string) error { return api.service.messenger.SendContactUpdates(ctx, name, picture) } func (api *NimbusPublicAPI) SendContactUpdate(ctx context.Context, contactID, name, picture string) (*protocol.MessengerResponse, error) { return api.service.messenger.SendContactUpdate(ctx, contactID, name, picture) } func (api *NimbusPublicAPI) SendPairInstallation(ctx context.Context) (*protocol.MessengerResponse, error) { return api.service.messenger.SendPairInstallation(ctx) } func (api *NimbusPublicAPI) SyncDevices(ctx context.Context, name, picture string) error { return api.service.messenger.SyncDevices(ctx, name, picture) } // ----- // HELPER // ----- // makeEnvelop makes an envelop for a historic messages request. // Symmetric key is used to authenticate to MailServer. // PK is the current node ID. // func makeEnvelop( // payload []byte, // symKey []byte, // publicKey *ecdsa.PublicKey, // nodeID *ecdsa.PrivateKey, // pow float64, // now time.Time, // ) (types.Envelope, error) { // params := whisper.MessageParams{ // PoW: pow, // Payload: payload, // WorkTime: defaultWorkTime, // Src: nodeID, // } // // Either symKey or public key is required. // // This condition is verified in `message.Wrap()` method. // if len(symKey) > 0 { // params.KeySym = symKey // } else if publicKey != nil { // params.Dst = publicKey // } // message, err := whisper.NewSentMessage(¶ms) // if err != nil { // return nil, err // } // envelope, err := message.Wrap(¶ms, now) // if err != nil { // return nil, err // } // return nimbusbridge.NewNimbusEnvelopeWrapper(envelope), nil // } // // makeMessagesRequestPayload makes a specific payload for MailServer // // to request historic messages. // func makeMessagesRequestPayload(r MessagesRequest) ([]byte, error) { // cursor, err := hex.DecodeString(r.Cursor) // if err != nil { // return nil, fmt.Errorf("invalid cursor: %v", err) // } // if len(cursor) > 0 && len(cursor) != mailserver.CursorLength { // return nil, fmt.Errorf("invalid cursor size: expected %d but got %d", mailserver.CursorLength, len(cursor)) // } // payload := mailserver.MessagesRequestPayload{ // Lower: r.From, // Upper: r.To, // Bloom: createBloomFilter(r), // Limit: r.Limit, // Cursor: cursor, // // Client must tell the MailServer if it supports batch responses. // // This can be removed in the future. // Batch: true, // } // return rlp.EncodeToBytes(payload) // } // func createBloomFilter(r MessagesRequest) []byte { // if len(r.Topics) > 0 { // return topicsToBloom(r.Topics...) // } // return types.TopicToBloom(r.Topic) // } // func topicsToBloom(topics ...types.TopicType) []byte { // i := new(big.Int) // for _, topic := range topics { // bloom := types.TopicToBloom(topic) // i.Or(i, new(big.Int).SetBytes(bloom[:])) // } // combined := make([]byte, types.BloomFilterSize) // data := i.Bytes() // copy(combined[types.BloomFilterSize-len(data):], data[:]) // return combined // }