512 lines
14 KiB
Go
512 lines
14 KiB
Go
package ext
|
|
|
|
import (
|
|
"context"
|
|
"crypto/ecdsa"
|
|
"database/sql"
|
|
"math/big"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"github.com/syndtr/goleveldb/leveldb"
|
|
|
|
commongethtypes "github.com/ethereum/go-ethereum/common"
|
|
gethtypes "github.com/ethereum/go-ethereum/core/types"
|
|
"github.com/ethereum/go-ethereum/ethclient"
|
|
"github.com/ethereum/go-ethereum/log"
|
|
"github.com/ethereum/go-ethereum/node"
|
|
"github.com/ethereum/go-ethereum/p2p"
|
|
"github.com/ethereum/go-ethereum/p2p/enode"
|
|
"github.com/ethereum/go-ethereum/rpc"
|
|
|
|
"github.com/status-im/status-go/db"
|
|
coretypes "github.com/status-im/status-go/eth-node/core/types"
|
|
"github.com/status-im/status-go/eth-node/types"
|
|
"github.com/status-im/status-go/multiaccounts"
|
|
"github.com/status-im/status-go/multiaccounts/accounts"
|
|
"github.com/status-im/status-go/params"
|
|
"github.com/status-im/status-go/protocol"
|
|
"github.com/status-im/status-go/protocol/pushnotificationclient"
|
|
"github.com/status-im/status-go/protocol/pushnotificationserver"
|
|
"github.com/status-im/status-go/protocol/transport"
|
|
"github.com/status-im/status-go/services/ext/mailservers"
|
|
localnotifications "github.com/status-im/status-go/services/local-notifications"
|
|
"github.com/status-im/status-go/services/wallet"
|
|
"github.com/status-im/status-go/signal"
|
|
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
const (
|
|
// defaultConnectionsTarget used in Service.Start if configured connection target is 0.
|
|
defaultConnectionsTarget = 1
|
|
// defaultTimeoutWaitAdded is a timeout to use to establish initial connections.
|
|
defaultTimeoutWaitAdded = 5 * time.Second
|
|
)
|
|
|
|
// EnvelopeEventsHandler used for two different event types.
|
|
type EnvelopeEventsHandler interface {
|
|
EnvelopeSent([][]byte)
|
|
EnvelopeExpired([][]byte, error)
|
|
MailServerRequestCompleted(types.Hash, types.Hash, []byte, error)
|
|
MailServerRequestExpired(types.Hash)
|
|
}
|
|
|
|
// Service is a service that provides some additional API to whisper-based protocols like Whisper or Waku.
|
|
type Service struct {
|
|
messenger *protocol.Messenger
|
|
identity *ecdsa.PrivateKey
|
|
cancelMessenger chan struct{}
|
|
storage db.TransactionalStorage
|
|
n types.Node
|
|
config params.ShhextConfig
|
|
mailMonitor *MailRequestMonitor
|
|
requestsRegistry *RequestsRegistry
|
|
server *p2p.Server
|
|
eventSub mailservers.EnvelopeEventSubscriber
|
|
peerStore *mailservers.PeerStore
|
|
cache *mailservers.Cache
|
|
connManager *mailservers.ConnectionManager
|
|
lastUsedMonitor *mailservers.LastUsedConnectionMonitor
|
|
accountsDB *accounts.Database
|
|
multiAccountsDB *multiaccounts.Database
|
|
account *multiaccounts.Account
|
|
}
|
|
|
|
// Make sure that Service implements node.Service interface.
|
|
var _ node.Service = (*Service)(nil)
|
|
|
|
func New(
|
|
config params.ShhextConfig,
|
|
n types.Node,
|
|
ldb *leveldb.DB,
|
|
mailMonitor *MailRequestMonitor,
|
|
reqRegistry *RequestsRegistry,
|
|
eventSub mailservers.EnvelopeEventSubscriber,
|
|
) *Service {
|
|
cache := mailservers.NewCache(ldb)
|
|
peerStore := mailservers.NewPeerStore(cache)
|
|
return &Service{
|
|
storage: db.NewLevelDBStorage(ldb),
|
|
n: n,
|
|
config: config,
|
|
mailMonitor: mailMonitor,
|
|
requestsRegistry: reqRegistry,
|
|
peerStore: peerStore,
|
|
cache: mailservers.NewCache(ldb),
|
|
eventSub: eventSub,
|
|
}
|
|
}
|
|
|
|
func (s *Service) NodeID() *ecdsa.PrivateKey {
|
|
if s.server == nil {
|
|
return nil
|
|
}
|
|
return s.server.PrivateKey
|
|
}
|
|
|
|
func (s *Service) RequestsRegistry() *RequestsRegistry {
|
|
return s.requestsRegistry
|
|
}
|
|
|
|
func (s *Service) GetPeer(rawURL string) (*enode.Node, error) {
|
|
if len(rawURL) == 0 {
|
|
return mailservers.GetFirstConnected(s.server, s.peerStore)
|
|
}
|
|
return enode.ParseV4(rawURL)
|
|
}
|
|
|
|
func (s *Service) InitProtocol(identity *ecdsa.PrivateKey, db *sql.DB, multiAccountDb *multiaccounts.Database, acc *multiaccounts.Account, logger *zap.Logger) error {
|
|
if !s.config.PFSEnabled {
|
|
return nil
|
|
}
|
|
|
|
// If Messenger has been already set up, we need to shut it down
|
|
// before we init it again. Otherwise, it will lead to goroutines leakage
|
|
// due to not stopped filters.
|
|
if s.messenger != nil {
|
|
if err := s.messenger.Shutdown(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
s.identity = identity
|
|
|
|
dataDir := filepath.Clean(s.config.BackupDisabledDataDir)
|
|
|
|
if err := os.MkdirAll(dataDir, os.ModePerm); err != nil {
|
|
return err
|
|
}
|
|
|
|
envelopesMonitorConfig := &transport.EnvelopesMonitorConfig{
|
|
MaxAttempts: s.config.MaxMessageDeliveryAttempts,
|
|
MailserverConfirmationsEnabled: s.config.MailServerConfirmations,
|
|
IsMailserver: func(peer types.EnodeID) bool {
|
|
return s.peerStore.Exist(peer)
|
|
},
|
|
EnvelopeEventsHandler: EnvelopeSignalHandler{},
|
|
Logger: logger,
|
|
}
|
|
s.accountsDB = accounts.NewDB(db)
|
|
s.multiAccountsDB = multiAccountDb
|
|
s.account = acc
|
|
|
|
options, err := buildMessengerOptions(s.config, identity, db, s.multiAccountsDB, acc, envelopesMonitorConfig, s.accountsDB, logger)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
messenger, err := protocol.NewMessenger(
|
|
identity,
|
|
s.n,
|
|
s.config.InstallationID,
|
|
options...,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.messenger = messenger
|
|
return messenger.Init()
|
|
}
|
|
|
|
func (s *Service) StartMessenger() error {
|
|
// Start a loop that retrieves all messages and propagates them to status-react.
|
|
s.cancelMessenger = make(chan struct{})
|
|
err := s.messenger.Start()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go s.retrieveMessagesLoop(time.Second, s.cancelMessenger)
|
|
go s.verifyTransactionLoop(30*time.Second, s.cancelMessenger)
|
|
go s.verifyENSLoop(30*time.Second, s.cancelMessenger)
|
|
return nil
|
|
}
|
|
|
|
func (s *Service) retrieveMessagesLoop(tick time.Duration, cancel <-chan struct{}) {
|
|
ticker := time.NewTicker(tick)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
response, err := s.messenger.RetrieveAll()
|
|
if err != nil {
|
|
log.Error("failed to retrieve raw messages", "err", err)
|
|
continue
|
|
}
|
|
if !response.IsEmpty() {
|
|
PublisherSignalHandler{}.NewMessages(response)
|
|
localnotifications.SendMessageNotifications(response.Notifications)
|
|
}
|
|
case <-cancel:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
type verifyTransactionClient struct {
|
|
chainID *big.Int
|
|
url string
|
|
}
|
|
|
|
func (c *verifyTransactionClient) TransactionByHash(ctx context.Context, hash types.Hash) (coretypes.Message, coretypes.TransactionStatus, error) {
|
|
signer := gethtypes.NewEIP155Signer(c.chainID)
|
|
client, err := ethclient.Dial(c.url)
|
|
if err != nil {
|
|
return coretypes.Message{}, coretypes.TransactionStatusPending, err
|
|
}
|
|
|
|
transaction, pending, err := client.TransactionByHash(ctx, commongethtypes.BytesToHash(hash.Bytes()))
|
|
if err != nil {
|
|
return coretypes.Message{}, coretypes.TransactionStatusPending, err
|
|
}
|
|
|
|
message, err := transaction.AsMessage(signer)
|
|
if err != nil {
|
|
return coretypes.Message{}, coretypes.TransactionStatusPending, err
|
|
}
|
|
from := types.BytesToAddress(message.From().Bytes())
|
|
to := types.BytesToAddress(message.To().Bytes())
|
|
|
|
if pending {
|
|
return coretypes.NewMessage(
|
|
from,
|
|
&to,
|
|
message.Nonce(),
|
|
message.Value(),
|
|
message.Gas(),
|
|
message.GasPrice(),
|
|
message.Data(),
|
|
message.CheckNonce(),
|
|
), coretypes.TransactionStatusPending, nil
|
|
}
|
|
|
|
receipt, err := client.TransactionReceipt(ctx, commongethtypes.BytesToHash(hash.Bytes()))
|
|
if err != nil {
|
|
return coretypes.Message{}, coretypes.TransactionStatusPending, err
|
|
}
|
|
|
|
coremessage := coretypes.NewMessage(
|
|
from,
|
|
&to,
|
|
message.Nonce(),
|
|
message.Value(),
|
|
message.Gas(),
|
|
message.GasPrice(),
|
|
message.Data(),
|
|
message.CheckNonce(),
|
|
)
|
|
|
|
// Token transfer, check the logs
|
|
if len(coremessage.Data()) != 0 {
|
|
if wallet.IsTokenTransfer(receipt.Logs) {
|
|
return coremessage, coretypes.TransactionStatus(receipt.Status), nil
|
|
}
|
|
return coremessage, coretypes.TransactionStatusFailed, nil
|
|
}
|
|
|
|
return coremessage, coretypes.TransactionStatus(receipt.Status), nil
|
|
}
|
|
|
|
func (s *Service) verifyENSLoop(tick time.Duration, cancel <-chan struct{}) {
|
|
if s.config.VerifyENSURL == "" || s.config.VerifyENSContractAddress == "" {
|
|
log.Warn("not starting ENS loop")
|
|
return
|
|
}
|
|
|
|
ticker := time.NewTicker(tick)
|
|
defer ticker.Stop()
|
|
|
|
ctx, cancelVerifyENS := context.WithCancel(context.Background())
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
response, err := s.messenger.VerifyENSNames(ctx, s.config.VerifyENSURL, s.config.VerifyENSContractAddress)
|
|
if err != nil {
|
|
log.Error("failed to validate ens", "err", err)
|
|
continue
|
|
}
|
|
if !response.IsEmpty() {
|
|
PublisherSignalHandler{}.NewMessages(response)
|
|
}
|
|
case <-cancel:
|
|
cancelVerifyENS()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) verifyTransactionLoop(tick time.Duration, cancel <-chan struct{}) {
|
|
if s.config.VerifyTransactionURL == "" {
|
|
log.Warn("not starting transaction loop")
|
|
return
|
|
}
|
|
|
|
ticker := time.NewTicker(tick)
|
|
defer ticker.Stop()
|
|
|
|
ctx, cancelVerifyTransaction := context.WithCancel(context.Background())
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
accounts, err := s.accountsDB.GetAccounts()
|
|
if err != nil {
|
|
log.Error("failed to retrieve accounts", "err", err)
|
|
}
|
|
var wallets []types.Address
|
|
for _, account := range accounts {
|
|
if account.IsOwnAccount() {
|
|
wallets = append(wallets, types.BytesToAddress(account.Address.Bytes()))
|
|
}
|
|
}
|
|
|
|
response, err := s.messenger.ValidateTransactions(ctx, wallets)
|
|
if err != nil {
|
|
log.Error("failed to validate transactions", "err", err)
|
|
continue
|
|
}
|
|
if !response.IsEmpty() {
|
|
PublisherSignalHandler{}.NewMessages(response)
|
|
}
|
|
case <-cancel:
|
|
cancelVerifyTransaction()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) ConfirmMessagesProcessed(messageIDs [][]byte) error {
|
|
return s.messenger.ConfirmMessagesProcessed(messageIDs)
|
|
}
|
|
|
|
func (s *Service) EnableInstallation(installationID string) error {
|
|
return s.messenger.EnableInstallation(installationID)
|
|
}
|
|
|
|
// DisableInstallation disables an installation for multi-device sync.
|
|
func (s *Service) DisableInstallation(installationID string) error {
|
|
return s.messenger.DisableInstallation(installationID)
|
|
}
|
|
|
|
// UpdateMailservers updates information about selected mail servers.
|
|
func (s *Service) UpdateMailservers(nodes []*enode.Node) error {
|
|
if len(nodes) > 0 && s.messenger != nil {
|
|
s.messenger.SetMailserver(nodes[0].ID().Bytes())
|
|
}
|
|
if err := s.peerStore.Update(nodes); err != nil {
|
|
return err
|
|
}
|
|
if s.connManager != nil {
|
|
s.connManager.Notify(nodes)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Protocols returns a new protocols list. In this case, there are none.
|
|
func (s *Service) Protocols() []p2p.Protocol {
|
|
return []p2p.Protocol{}
|
|
}
|
|
|
|
// APIs returns a list of new APIs.
|
|
func (s *Service) APIs() []rpc.API {
|
|
panic("this is abstract service, use shhext or wakuext implementation")
|
|
}
|
|
|
|
// Start is run when a service is started.
|
|
// It does nothing in this case but is required by `node.Service` interface.
|
|
func (s *Service) Start(server *p2p.Server) error {
|
|
if s.config.EnableConnectionManager {
|
|
connectionsTarget := s.config.ConnectionTarget
|
|
if connectionsTarget == 0 {
|
|
connectionsTarget = defaultConnectionsTarget
|
|
}
|
|
maxFailures := s.config.MaxServerFailures
|
|
// if not defined change server on first expired event
|
|
if maxFailures == 0 {
|
|
maxFailures = 1
|
|
}
|
|
s.connManager = mailservers.NewConnectionManager(server, s.eventSub, connectionsTarget, maxFailures, defaultTimeoutWaitAdded)
|
|
s.connManager.Start()
|
|
if err := mailservers.EnsureUsedRecordsAddedFirst(s.peerStore, s.connManager); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if s.config.EnableLastUsedMonitor {
|
|
s.lastUsedMonitor = mailservers.NewLastUsedConnectionMonitor(s.peerStore, s.cache, s.eventSub)
|
|
s.lastUsedMonitor.Start()
|
|
}
|
|
s.mailMonitor.Start()
|
|
s.server = server
|
|
return nil
|
|
}
|
|
|
|
// Stop is run when a service is stopped.
|
|
func (s *Service) Stop() error {
|
|
log.Info("Stopping shhext service")
|
|
if s.config.EnableConnectionManager {
|
|
s.connManager.Stop()
|
|
}
|
|
if s.config.EnableLastUsedMonitor {
|
|
s.lastUsedMonitor.Stop()
|
|
}
|
|
s.requestsRegistry.Clear()
|
|
s.mailMonitor.Stop()
|
|
|
|
if s.cancelMessenger != nil {
|
|
select {
|
|
case <-s.cancelMessenger:
|
|
// channel already closed
|
|
default:
|
|
close(s.cancelMessenger)
|
|
s.cancelMessenger = nil
|
|
}
|
|
}
|
|
|
|
if s.messenger != nil {
|
|
if err := s.messenger.Shutdown(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func onNegotiatedFilters(filters []*transport.Filter) {
|
|
var signalFilters []*signal.Filter
|
|
for _, filter := range filters {
|
|
|
|
signalFilter := &signal.Filter{
|
|
ChatID: filter.ChatID,
|
|
SymKeyID: filter.SymKeyID,
|
|
Listen: filter.Listen,
|
|
FilterID: filter.FilterID,
|
|
Identity: filter.Identity,
|
|
Topic: filter.Topic,
|
|
}
|
|
|
|
signalFilters = append(signalFilters, signalFilter)
|
|
}
|
|
if len(filters) != 0 {
|
|
handler := PublisherSignalHandler{}
|
|
handler.FilterAdded(signalFilters)
|
|
}
|
|
}
|
|
|
|
func buildMessengerOptions(
|
|
config params.ShhextConfig,
|
|
identity *ecdsa.PrivateKey,
|
|
db *sql.DB,
|
|
multiAccounts *multiaccounts.Database,
|
|
account *multiaccounts.Account,
|
|
envelopesMonitorConfig *transport.EnvelopesMonitorConfig,
|
|
accountsDB *accounts.Database,
|
|
logger *zap.Logger,
|
|
) ([]protocol.Option, error) {
|
|
options := []protocol.Option{
|
|
protocol.WithCustomLogger(logger),
|
|
protocol.WithPushNotifications(),
|
|
protocol.WithDatabase(db),
|
|
protocol.WithMultiAccounts(multiAccounts),
|
|
protocol.WithAccount(account),
|
|
protocol.WithEnvelopesMonitorConfig(envelopesMonitorConfig),
|
|
protocol.WithOnNegotiatedFilters(onNegotiatedFilters),
|
|
}
|
|
|
|
if config.DataSyncEnabled {
|
|
options = append(options, protocol.WithDatasync())
|
|
}
|
|
settings, err := accountsDB.GetSettings()
|
|
if err != sql.ErrNoRows && err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if settings.PushNotificationsServerEnabled {
|
|
config := &pushnotificationserver.Config{
|
|
Enabled: true,
|
|
Logger: logger,
|
|
}
|
|
options = append(options, protocol.WithPushNotificationServerConfig(config))
|
|
}
|
|
|
|
options = append(options, protocol.WithPushNotificationClientConfig(&pushnotificationclient.Config{
|
|
DefaultServers: config.DefaultPushNotificationsServers,
|
|
BlockMentions: settings.PushNotificationsBlockMentions,
|
|
SendEnabled: settings.SendPushNotifications,
|
|
AllowFromContactsOnly: settings.PushNotificationsFromContactsOnly,
|
|
RemoteNotificationsEnabled: settings.RemotePushNotificationsEnabled,
|
|
}))
|
|
|
|
if config.VerifyTransactionURL != "" {
|
|
client := &verifyTransactionClient{
|
|
url: config.VerifyTransactionURL,
|
|
chainID: big.NewInt(config.VerifyTransactionChainID),
|
|
}
|
|
options = append(options, protocol.WithVerifyTransactionClient(client))
|
|
}
|
|
|
|
return options, nil
|
|
}
|