status-im
/
status-go
mirror of https://github.com/status-im/status-go.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

remove nimbus

This commit is contained in:
Andrea Maria Piana 2021-05-14 11:26:16 +02:00
parent 57b1bc193f
commit 566e9a3ade
31 changed files with 1 additions and 3739 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -73,7 +73,3 @@ _assets/compose/bootnode/keys
# do not vendor nested vendor/ dirs
vendor/**/vendor
# Nimbus
/vendor/github.com/status-im/nimbus/
**/eth-node/bridge/nimbus/libnimbus.*

11
Makefile
View File

@ -75,15 +75,6 @@ HELP_FUN = \
print "\n"; \
}
nimbus: ##@build Build Nimbus
./eth-node/bridge/nimbus/build-nimbus.sh
nimbus-statusgo: nimbus ##@build Build status-go (based on Nimbus node) as statusd server
C_INCLUDE_PATH="./eth-node/bridge/nimbus" go build -mod=vendor -i -o $(GOBIN)/statusd -v -tags '$(BUILD_TAGS) nimbus' $(BUILD_FLAGS) ./cmd/statusd && \
cp vendor/github.com/status-im/status-go/eth-node/bridge/nimbus/libnimbus.so $(GOBIN)
@echo "Compilation done."
@echo "Run \"build/bin/statusd -h\" to view available commands."
statusgo: ##@build Build status-go as statusd server
go build -mod=vendor -i -o $(GOBIN)/statusd -v -tags '$(BUILD_TAGS)' $(BUILD_FLAGS) ./cmd/statusd
@echo "Compilation done."
@ -323,7 +314,7 @@ ci: lint canary-test test-unit test-e2e ##@tests Run all linters and tests at on
ci-race: lint canary-test test-unit test-e2e-race ##@tests Run all linters and tests at once + race
clean: ##@other Cleanup
rm -fr build/bin/* mailserver-config.json vendor/github.com/status-im/nimbus
rm -fr build/bin/* mailserver-config.json
git clean -xf
deep-clean: clean

29
account/accounts_nimbus.go
View File

@ -1,29 +0,0 @@
// +build nimbus
package account
import (
"github.com/status-im/status-go/account/generator"
)
// NimbusManager represents account manager interface.
type NimbusManager struct {
*Manager
}
// NewNimbusManager returns new node account manager.
func NewNimbusManager() *NimbusManager {
m := &NimbusManager{}
m.Manager = &Manager{accountsGenerator: generator.New(m)}
return m
}
// // InitKeystore sets key manager and key store.
// func (m *Manager) InitKeystore(keydir string) error {
// m.mu.Lock()
// defer m.mu.Unlock()
// var err error
// m.keystore, err = makeKeyStore(keydir)
// return err
// }

4
account/keystore_geth.go
View File

@ -1,7 +1,3 @@
// +build !nimbus
// TODO: Make independent version for Nimbus
package account
import (

49
account/keystore_nimbus.go
View File

@ -1,49 +0,0 @@
// +build nimbus
package account
import (
"io/ioutil"
"os"
"github.com/ethereum/go-ethereum/accounts"
"github.com/ethereum/go-ethereum/accounts/keystore"
gethbridge "github.com/status-im/status-go/eth-node/bridge/geth"
// nimbusbridge "github.com/status-im/status-go/eth-node/bridge/nimbus"
"github.com/status-im/status-go/eth-node/types"
)
// makeAccountManager creates ethereum accounts.Manager with single disk backend and lightweight kdf.
// If keydir is empty new temporary directory with go-ethereum-keystore will be intialized.
func makeAccountManager(keydir string) (manager *accounts.Manager, err error) {
if keydir == "" {
// There is no datadir.
keydir, err = ioutil.TempDir("", "nimbus-keystore")
}
if err != nil {
return nil, err
}
if err := os.MkdirAll(keydir, 0700); err != nil {
return nil, err
}
config := accounts.Config{InsecureUnlockAllowed: false}
return accounts.NewManager(&config, keystore.NewKeyStore(keydir, keystore.LightScryptN, keystore.LightScryptP)), nil
}
func makeKeyStore(manager *accounts.Manager) (types.KeyStore, error) {
backends := manager.Backends(keystore.KeyStoreType)
if len(backends) == 0 {
return nil, ErrAccountKeyStoreMissing
}
keyStore, ok := backends[0].(*keystore.KeyStore)
if !ok {
return nil, ErrAccountKeyStoreMissing
}
return gethbridge.WrapKeyStore(keyStore), nil
}
// func makeKeyStore(_ string) (types.KeyStore, error) {
// return nimbusbridge.WrapKeyStore(), nil
// }

2
api/geth_backend.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package api
import (

1011
api/nimbus_backend.go
View File

File diff suppressed because it is too large Load Diff

40
eth-node/bridge/nimbus/build-nimbus.sh
View File

@ -1,40 +0,0 @@
#!/usr/bin/env bash
# Pre-requisites: Git, Nix
set -e
GIT_ROOT=$(cd "${BASH_SOURCE%/*}" && git rev-parse --show-toplevel)
# NOTE: To use a local Nimbus repository, uncomment and edit the following line
#nimbus_dir=~/src/github.com/status-im/nimbus
target_dir="${GIT_ROOT}/vendor/github.com/status-im/status-go/eth-node/bridge/nimbus"
if [ -z "$nimbus_dir" ]; then
# The git ref of Nimbus to fetch and build. This should represent a commit SHA or a tag, for reproducible builds
nimbus_ref='feature/android-api' # TODO: Use a tag once
nimbus_src='https://github.com/status-im/nimbus/'
nimbus_dir="${GIT_ROOT}/vendor/github.com/status-im/nimbus"
trap "rm -rf $nimbus_dir" ERR INT QUIT
# Clone nimbus repo into vendor directory, if necessary
if [ -d "$nimbus_dir" ]; then
cd $nimbus_dir && git reset --hard $nimbus_ref; cd -
else
# List fetched from vendorDeps array in https://github.com/status-im/nimbus/blob/master/nix/nimbus-wrappers.nix#L9-L12
vendor_paths=( nim-chronicles nim-faststreams nim-json-serialization nim-chronos nim-eth nim-json nim-metrics nim-secp256k1 nim-serialization nim-stew nim-stint nimcrypto )
vendor_path_opts="${vendor_paths[@]/#/--recurse-submodules=vendor/}"
git clone $nimbus_src --progress ${vendor_path_opts} --depth 1 -j8 -b $nimbus_ref $nimbus_dir
fi
fi
# Build Nimbus wrappers and copy them into the Nimbus bridge in status-eth-node
build_dir=$(nix-build --pure --no-out-link -A wrappers-native $nimbus_dir/nix/default.nix)
rm -f ${target_dir}/libnimbus.*
mkdir -p ${target_dir}
cp -f ${build_dir}/include/* ${build_dir}/lib/libnimbus.so \
${target_dir}/
chmod +w ${target_dir}/libnimbus.{so,h}

16
eth-node/bridge/nimbus/cfuncs.go
View File

@ -1,16 +0,0 @@
// +build nimbus
package nimbusbridge
/*
#include <libnimbus.h>
// onMessageHandler gateway function
void onMessageHandler_cgo(received_message * msg, void* udata)
{
void onMessageHandler(received_message* msg, void* udata);
onMessageHandler(msg, udata);
}
*/
import "C"

61
eth-node/bridge/nimbus/filter.go
View File

@ -1,61 +0,0 @@
// +build nimbus
package nimbusbridge
// https://golang.org/cmd/cgo/
/*
#include <stddef.h>
#include <stdbool.h>
#include <stdlib.h>
#include <libnimbus.h>
*/
import "C"
import (
"unsafe"
"github.com/status-im/status-go/eth-node/types"
)
type nimbusFilterWrapper struct {
filter *C.filter_options
id string
own bool
}
// NewNimbusFilterWrapper returns an object that wraps Nimbus's Filter in a types interface
func NewNimbusFilterWrapper(f *C.filter_options, id string, own bool) types.Filter {
wrapper := &nimbusFilterWrapper{
filter: f,
id: id,
own: own,
}
return wrapper
}
// GetNimbusFilterFrom retrieves the underlying whisper Filter struct from a wrapped Filter interface
func GetNimbusFilterFrom(f types.Filter) *C.filter_options {
return f.(*nimbusFilterWrapper).filter
}
// ID returns the filter ID
func (w *nimbusFilterWrapper) ID() string {
return w.id
}
// Free frees the C memory associated with the filter
func (w *nimbusFilterWrapper) Free() {
if !w.own {
panic("native filter is not owned by Go")
}
if w.filter.privateKeyID != nil {
C.free(unsafe.Pointer(w.filter.privateKeyID))
w.filter.privateKeyID = nil
}
if w.filter.symKeyID != nil {
C.free(unsafe.Pointer(w.filter.symKeyID))
w.filter.symKeyID = nil
}
}

263
eth-node/bridge/nimbus/keystore.go
View File

@ -1,263 +0,0 @@
// +build nimbus
package nimbusbridge
// https://golang.org/cmd/cgo/
/*
#include <stddef.h>
#include <stdbool.h>
#include <stdlib.h>
#include <libnimbus.h>
*/
import "C"
import (
"crypto/ecdsa"
"encoding/binary"
"errors"
"fmt"
"math/big"
"unsafe"
"github.com/btcsuite/btcd/btcec"
"github.com/pborman/uuid"
"github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/extkeys"
)
var (
ErrInvalidSeed = errors.New("seed is invalid")
ErrInvalidKeyLen = errors.New("Nimbus serialized extended key length is invalid")
)
type nimbusKeyStoreAdapter struct {
}
// WrapKeyStore creates a types.KeyStore wrapper over the singleton Nimbus node
func WrapKeyStore() types.KeyStore {
return &nimbusKeyStoreAdapter{}
}
func (k *nimbusKeyStoreAdapter) ImportECDSA(priv *ecdsa.PrivateKey, passphrase string) (types.Account, error) {
fmt.Println("ImportECDSA")
panic("ImportECDSA")
var privateKeyC unsafe.Pointer
if priv != nil {
privateKeyC = C.CBytes(crypto.FromECDSA(priv))
defer C.free(privateKeyC)
}
passphraseC := C.CString(passphrase)
defer C.free(unsafe.Pointer(passphraseC))
var nimbusAccount C.account
if !C.nimbus_keystore_import_ecdsa((*C.uchar)(privateKeyC), passphraseC, &nimbusAccount) {
return types.Account{}, errors.New("failed to import ECDSA private key")
}
return accountFrom(&nimbusAccount), nil
}
func (k *nimbusKeyStoreAdapter) ImportSingleExtendedKey(extKey *extkeys.ExtendedKey, passphrase string) (types.Account, error) {
fmt.Println("ImportSingleExtendedKey")
panic("ImportSingleExtendedKey")
extKeyJSONC := C.CString(extKey.String())
defer C.free(unsafe.Pointer(extKeyJSONC))
passphraseC := C.CString(passphrase)
defer C.free(unsafe.Pointer(passphraseC))
var nimbusAccount C.account
if !C.nimbus_keystore_import_single_extendedkey(extKeyJSONC, passphraseC, &nimbusAccount) {
return types.Account{}, errors.New("failed to import extended key")
}
return accountFrom(&nimbusAccount), nil
}
func (k *nimbusKeyStoreAdapter) ImportExtendedKeyForPurpose(keyPurpose extkeys.KeyPurpose, extKey *extkeys.ExtendedKey, passphrase string) (types.Account, error) {
fmt.Println("ImportExtendedKeyForPurpose")
passphraseC := C.CString(passphrase)
defer C.free(unsafe.Pointer(passphraseC))
extKeyJSONC := C.CString(extKey.String())
defer C.free(unsafe.Pointer(extKeyJSONC))
var nimbusAccount C.account
if !C.nimbus_keystore_import_extendedkeyforpurpose(C.int(keyPurpose), extKeyJSONC, passphraseC, &nimbusAccount) {
return types.Account{}, errors.New("failed to import extended key")
}
return accountFrom(&nimbusAccount), nil
}
func (k *nimbusKeyStoreAdapter) AccountDecryptedKey(a types.Account, auth string) (types.Account, *types.Key, error) {
fmt.Println("AccountDecryptedKey")
panic("AccountDecryptedKey")
authC := C.CString(auth)
defer C.free(unsafe.Pointer(authC))
var nimbusAccount C.account
err := nimbusAccountFrom(a, &nimbusAccount)
if err != nil {
return types.Account{}, nil, err
}
var nimbusKey C.key
if !C.nimbus_keystore_account_decrypted_key(authC, &nimbusAccount, &nimbusKey) {
return types.Account{}, nil, errors.New("failed to decrypt account key")
}
key, err := keyFrom(&nimbusKey)
if err != nil {
return types.Account{}, nil, err
}
return accountFrom(&nimbusAccount), key, nil
}
func (k *nimbusKeyStoreAdapter) Delete(a types.Account, auth string) error {
fmt.Println("Delete")
var nimbusAccount C.account
err := nimbusAccountFrom(a, &nimbusAccount)
if err != nil {
return err
}
authC := C.CString(auth)
defer C.free(unsafe.Pointer(authC))
if !C.nimbus_keystore_delete(&nimbusAccount, authC) {
return errors.New("failed to delete account")
}
return nil
}
func nimbusAccountFrom(account types.Account, nimbusAccount *C.account) error {
fmt.Println("nimbusAccountFrom")
err := copyAddressToCBuffer(&nimbusAccount.address[0], account.Address.Bytes())
if err != nil {
return err
}
if account.URL == "" {
nimbusAccount.url[0] = C.char(0)
} else if len(account.URL) >= C.URL_LEN {
return errors.New("URL is too long to fit in Nimbus struct")
} else {
copyURLToCBuffer(&nimbusAccount.url[0], account.URL)
}
return err
}
func accountFrom(nimbusAccount *C.account) types.Account {
return types.Account{
Address: types.BytesToAddress(C.GoBytes(unsafe.Pointer(&nimbusAccount.address[0]), C.ADDRESS_LEN)),
URL: C.GoString(&nimbusAccount.url[0]),
}
}
// copyAddressToCBuffer copies a Go buffer to a C buffer without allocating new memory
func copyAddressToCBuffer(dst *C.uchar, src []byte) error {
if len(src) != C.ADDRESS_LEN {
return errors.New("invalid buffer size")
}
p := (*[C.ADDRESS_LEN]C.uchar)(unsafe.Pointer(dst))
for index, b := range src {
p[index] = C.uchar(b)
}
return nil
}
// copyURLToCBuffer copies a Go buffer to a C buffer without allocating new memory
func copyURLToCBuffer(dst *C.char, src string) error {
if len(src)+1 > C.URL_LEN {
return errors.New("URL is too long to fit in Nimbus struct")
}
p := (*[C.URL_LEN]C.uchar)(unsafe.Pointer(dst))
for index := 0; index <= len(src); index++ {
p[index] = C.uchar(src[index])
}
return nil
}
func keyFrom(k *C.key) (*types.Key, error) {
fmt.Println("keyFrom")
if k == nil {
return nil, nil
}
var err error
key := types.Key{
ID: uuid.Parse(C.GoString(&k.id[0])),
}
key.Address = types.BytesToAddress(C.GoBytes(unsafe.Pointer(&k.address[0]), C.ADDRESS_LEN))
key.PrivateKey, err = crypto.ToECDSA(C.GoBytes(unsafe.Pointer(&k.privateKeyID[0]), C.PRIVKEY_LEN))
if err != nil {
return nil, err
}
key.ExtendedKey, err = newExtKeyFromBuffer(C.GoBytes(unsafe.Pointer(&k.extKey[0]), C.EXTKEY_LEN))
if err != nil {
return nil, err
}
return &key, err
}
// newExtKeyFromBuffer returns a new extended key instance from a serialized
// extended key.
func newExtKeyFromBuffer(key []byte) (*extkeys.ExtendedKey, error) {
if len(key) == 0 {
return &extkeys.ExtendedKey{}, nil
}
if len(key) != C.EXTKEY_LEN {
return nil, ErrInvalidKeyLen
}
// The serialized format is:
// version (4) || depth (1) || parent fingerprint (4)) ||
// child num (4) || chain code (32) || key data (33)
payload := key
// Deserialize each of the payload fields.
version := payload[:4]
depth := payload[4:5][0]
fingerPrint := payload[5:9]
childNumber := binary.BigEndian.Uint32(payload[9:13])
chainCode := payload[13:45]
keyData := payload[45:78]
// The key data is a private key if it starts with 0x00. Serialized
// compressed pubkeys either start with 0x02 or 0x03.
isPrivate := keyData[0] == 0x00
if isPrivate {
// Ensure the private key is valid. It must be within the range
// of the order of the secp256k1 curve and not be 0.
keyData = keyData[1:]
keyNum := new(big.Int).SetBytes(keyData)
if keyNum.Cmp(btcec.S256().N) >= 0 || keyNum.Sign() == 0 {
return nil, ErrInvalidSeed
}
} else {
// Ensure the public key parses correctly and is actually on the
// secp256k1 curve.
_, err := btcec.ParsePubKey(keyData, btcec.S256())
if err != nil {
return nil, err
}
}
return &extkeys.ExtendedKey{
Version: version,
KeyData: keyData,
ChainCode: chainCode,
FingerPrint: fingerPrint,
Depth: depth,
ChildNumber: childNumber,
IsPrivate: isPrivate,
}, nil
}

159
eth-node/bridge/nimbus/node.go
View File

@ -1,159 +0,0 @@
// +build nimbus
package nimbusbridge
// https://golang.org/cmd/cgo/
/*
#cgo LDFLAGS: -Wl,-rpath,'$ORIGIN' -L${SRCDIR} -lnimbus -lm
#include <stddef.h>
#include <stdbool.h>
#include <stdlib.h>
#include <libnimbus.h>
*/
import "C"
import (
"crypto/ecdsa"
"errors"
"fmt"
"runtime"
"strconv"
"strings"
"sync"
"syscall"
"time"
"unsafe"
"go.uber.org/zap"
"github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types"
enstypes "github.com/status-im/status-go/eth-node/types/ens"
)
type nimbusNodeWrapper struct {
mu sync.Mutex
routineQueue *RoutineQueue
tid int
nodeStarted bool
cancelPollingChan chan struct{}
w types.Whisper
}
type Node interface {
types.Node
StartNimbus(privateKey *ecdsa.PrivateKey, listenAddr string, staging bool) error
Stop()
}
func NewNodeBridge() Node {
c := make(chan Node, 1)
go func(c chan<- Node, delay time.Duration) {
runtime.LockOSThread()
defer runtime.UnlockOSThread()
n := &nimbusNodeWrapper{
routineQueue: NewRoutineQueue(),
tid: syscall.Gettid(),
cancelPollingChan: make(chan struct{}, 1),
}
c <- n
for {
select {
case <-time.After(delay):
n.poll()
case <-n.cancelPollingChan:
return
}
}
}(c, 50*time.Millisecond)
return <-c
}
func (n *nimbusNodeWrapper) StartNimbus(privateKey *ecdsa.PrivateKey, listenAddr string, staging bool) error {
return n.routineQueue.Send(func(c chan<- callReturn) {
c <- callReturn{err: startNimbus(privateKey, listenAddr, staging)}
n.nodeStarted = true
}).err
}
func (n *nimbusNodeWrapper) Stop() {
if n.cancelPollingChan != nil {
close(n.cancelPollingChan)
n.nodeStarted = false
n.cancelPollingChan = nil
}
}
func (n *nimbusNodeWrapper) NewENSVerifier(_ *zap.Logger) enstypes.ENSVerifier {
panic("not implemented")
}
func (n *nimbusNodeWrapper) GetWhisper(ctx interface{}) (types.Whisper, error) {
n.mu.Lock()
defer n.mu.Unlock()
if n.w == nil {
n.w = NewNimbusWhisperWrapper(n.routineQueue)
}
return n.w, nil
}
func (w *nimbusNodeWrapper) GetWaku(ctx interface{}) (types.Waku, error) {
panic("not implemented")
}
func (n *nimbusNodeWrapper) AddPeer(url string) error {
urlC := C.CString(url)
defer C.free(unsafe.Pointer(urlC))
if !C.nimbus_add_peer(urlC) {
return fmt.Errorf("failed to add peer: %s", url)
}
return nil
}
func (n *nimbusNodeWrapper) RemovePeer(url string) error {
panic("TODO: RemovePeer")
}
func (n *nimbusNodeWrapper) poll() {
if syscall.Gettid() != n.tid {
panic("poll called from wrong thread")
}
if n.nodeStarted {
C.nimbus_poll()
}
n.routineQueue.HandleEvent()
}
func startNimbus(privateKey *ecdsa.PrivateKey, listenAddr string, staging bool) error {
C.NimMain()
if listenAddr == "" {
listenAddr = ":30304"
}
addrParts := strings.Split(listenAddr, ":")
port, err := strconv.Atoi(addrParts[len(addrParts)-1])
if err != nil {
return fmt.Errorf("failed to parse port number from %s", listenAddr)
}
var privateKeyC unsafe.Pointer
if privateKey != nil {
privateKeyC = C.CBytes(crypto.FromECDSA(privateKey))
defer C.free(privateKeyC)
}
if !C.nimbus_start(C.ushort(port), true, false, 0.002, (*C.uchar)(privateKeyC), C.bool(staging)) {
return errors.New("failed to start Nimbus node")
}
return nil
}

212
eth-node/bridge/nimbus/public_whisper_api.go
View File

@ -1,212 +0,0 @@
// +build nimbus
package nimbusbridge
// https://golang.org/cmd/cgo/
/*
#include <stddef.h>
#include <stdbool.h>
#include <stdlib.h>
#include <libnimbus.h>
*/
import "C"
import (
"container/list"
"context"
"errors"
"fmt"
"sync"
"unsafe"
"github.com/status-im/status-go/eth-node/types"
)
type nimbusPublicWhisperAPIWrapper struct {
filterMessagesMu *sync.Mutex
filterMessages *map[string]*list.List
routineQueue *RoutineQueue
}
// NewNimbusPublicWhisperAPIWrapper returns an object that wraps Nimbus's PublicWhisperAPI in a types interface
func NewNimbusPublicWhisperAPIWrapper(filterMessagesMu *sync.Mutex, filterMessages *map[string]*list.List, routineQueue *RoutineQueue) types.PublicWhisperAPI {
return &nimbusPublicWhisperAPIWrapper{
filterMessagesMu: filterMessagesMu,
filterMessages: filterMessages,
routineQueue: routineQueue,
}
}
// AddPrivateKey imports the given private key.
func (w *nimbusPublicWhisperAPIWrapper) AddPrivateKey(ctx context.Context, privateKey types.HexBytes) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
privKeyC := C.CBytes(privateKey)
defer C.free(unsafe.Pointer(privKeyC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if C.nimbus_add_keypair((*C.uchar)(privKeyC), (*C.uchar)(idC)) {
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
} else {
c <- callReturn{err: errors.New("failed to add private key to Nimbus")}
}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
// GenerateSymKeyFromPassword derives a key from the given password, stores it, and returns its ID.
func (w *nimbusPublicWhisperAPIWrapper) GenerateSymKeyFromPassword(ctx context.Context, passwd string) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
passwordC := C.CString(passwd)
defer C.free(unsafe.Pointer(passwordC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if C.nimbus_add_symkey_from_password(passwordC, (*C.uchar)(idC)) {
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
} else {
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
// DeleteKeyPair removes the key with the given key if it exists.
func (w *nimbusPublicWhisperAPIWrapper) DeleteKeyPair(ctx context.Context, key string) (bool, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC, err := decodeHexID(key)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(keyC))
c <- callReturn{value: C.nimbus_delete_keypair(keyC)}
})
if retVal.err != nil {
return false, retVal.err
}
return retVal.value.(bool), nil
}
// NewMessageFilter creates a new filter that can be used to poll for
// (new) messages that satisfy the given criteria.
func (w *nimbusPublicWhisperAPIWrapper) NewMessageFilter(req types.Criteria) (string, error) {
// topics := make([]whisper.TopicType, len(req.Topics))
// for index, tt := range req.Topics {
// topics[index] = whisper.TopicType(tt)
// }
// criteria := whisper.Criteria{
// SymKeyID: req.SymKeyID,
// PrivateKeyID: req.PrivateKeyID,
// Sig: req.Sig,
// MinPow: req.MinPow,
// Topics: topics,
// AllowP2P: req.AllowP2P,
// }
// return w.publicWhisperAPI.NewMessageFilter(criteria)
// TODO
return "", errors.New("not implemented")
}
// GetFilterMessages returns the messages that match the filter criteria and
// are received between the last poll and now.
func (w *nimbusPublicWhisperAPIWrapper) GetFilterMessages(id string) ([]*types.Message, error) {
idC := C.CString(id)
defer C.free(unsafe.Pointer(idC))
var (
messageList *list.List
ok bool
)
w.filterMessagesMu.Lock()
defer w.filterMessagesMu.Unlock()
if messageList, ok = (*w.filterMessages)[id]; !ok {
return nil, fmt.Errorf("no filter with ID %s", id)
}
retVal := make([]*types.Message, messageList.Len())
if messageList.Len() == 0 {
return retVal, nil
}
elem := messageList.Front()
index := 0
for elem != nil {
retVal[index] = (elem.Value).(*types.Message)
index++
next := elem.Next()
messageList.Remove(elem)
elem = next
}
return retVal, nil
}
// Post posts a message on the Whisper network.
// returns the hash of the message in case of success.
func (w *nimbusPublicWhisperAPIWrapper) Post(ctx context.Context, req types.NewMessage) ([]byte, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
msg := C.post_message{
ttl: C.uint32_t(req.TTL),
powTime: C.double(req.PowTime),
powTarget: C.double(req.PowTarget),
}
if req.SigID != "" {
sourceID, err := decodeHexID(req.SigID)
if err != nil {
c <- callReturn{err: err}
return
}
msg.sourceID = sourceID
defer C.free(unsafe.Pointer(sourceID))
}
if req.SymKeyID != "" {
symKeyID, err := decodeHexID(req.SymKeyID)
if err != nil {
c <- callReturn{err: err}
return
}
msg.symKeyID = symKeyID
defer C.free(unsafe.Pointer(symKeyID))
}
if req.PublicKey != nil && len(req.PublicKey) > 0 {
msg.pubKey = (*C.uchar)(C.CBytes(req.PublicKey[1:]))
defer C.free(unsafe.Pointer(msg.pubKey))
}
msg.payloadLen = C.size_t(len(req.Payload))
msg.payload = (*C.uchar)(C.CBytes(req.Payload))
defer C.free(unsafe.Pointer(msg.payload))
msg.paddingLen = C.size_t(len(req.Padding))
msg.padding = (*C.uchar)(C.CBytes(req.Padding))
defer C.free(unsafe.Pointer(msg.padding))
copyTopicToCBuffer(&msg.topic[0], req.Topic[:])
// TODO: return envelope hash once nimbus_post is improved to return it
if C.nimbus_post(&msg) {
c <- callReturn{value: make([]byte, 0)}
return
}
c <- callReturn{err: fmt.Errorf("failed to post message symkeyid=%s pubkey=%#x topic=%#x", req.SymKeyID, req.PublicKey, req.Topic[:])}
// hashC := C.nimbus_post(&msg)
// if hashC == nil {
// return nil, errors.New("Nimbus failed to post message")
// }
// return hex.DecodeString(C.GoString(hashC))
})
if retVal.err != nil {
return nil, retVal.err
}
return retVal.value.([]byte), nil
}

64
eth-node/bridge/nimbus/routine_queue.go
View File

@ -1,64 +0,0 @@
// +build nimbus
package nimbusbridge
import (
"syscall"
)
// RoutineQueue provides a mechanism for marshalling function calls
// so that they are run in a specific thread (the thread where
// RoutineQueue is initialized).
type RoutineQueue struct {
tid int
events chan event
}
type callReturn struct {
value interface{}
err error
}
// NewRoutineQueue returns a new RoutineQueue object.
func NewRoutineQueue() *RoutineQueue {
q := &RoutineQueue{
tid: syscall.Gettid(),
events: make(chan event, 20),
}
return q
}
// event represents an event triggered by the user.
type event struct {
f func(chan<- callReturn)
done chan callReturn
}
func (q *RoutineQueue) HandleEvent() {
if syscall.Gettid() != q.tid {
panic("HandleEvent called from wrong thread")
}
select {
case ev := <-q.events:
ev.f(ev.done)
default:
return
}
}
// Send executes the passed function. This method can be called safely from a
// goroutine in order to execute a Nimbus function. It is important to note that the
// passed function won't be executed immediately, instead it will be added to
// the user events queue.
func (q *RoutineQueue) Send(f func(chan<- callReturn)) callReturn {
ev := event{f: f, done: make(chan callReturn, 1)}
defer close(ev.done)
if syscall.Gettid() == q.tid {
f(ev.done)
return <-ev.done
}
q.events <- ev
return <-ev.done
}

431
eth-node/bridge/nimbus/whisper.go
View File

@ -1,431 +0,0 @@
// +build nimbus
package nimbusbridge
// https://golang.org/cmd/cgo/
/*
#include <stddef.h>
#include <stdbool.h>
#include <stdlib.h>
#include <libnimbus.h>
void onMessageHandler_cgo(received_message* msg, void* udata); // Forward declaration.
*/
import "C"
import (
"container/list"
"crypto/ecdsa"
"errors"
"fmt"
"sync"
"time"
"unsafe"
gopointer "github.com/mattn/go-pointer"
"github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types"
)
type nimbusWhisperWrapper struct {
timesource func() time.Time
filters map[string]types.Filter
filterMessagesMu sync.Mutex
filterMessages map[string]*list.List
routineQueue *RoutineQueue
}
// NewNimbusWhisperWrapper returns an object that wraps Nimbus' Whisper in a types interface
func NewNimbusWhisperWrapper(routineQueue *RoutineQueue) types.Whisper {
return &nimbusWhisperWrapper{
timesource: func() time.Time { return time.Now() },
filters: map[string]types.Filter{},
filterMessages: map[string]*list.List{},
routineQueue: routineQueue,
}
}
func (w *nimbusWhisperWrapper) PublicWhisperAPI() types.PublicWhisperAPI {
return NewNimbusPublicWhisperAPIWrapper(&w.filterMessagesMu, &w.filterMessages, w.routineQueue)
}
// MinPow returns the PoW value required by this node.
func (w *nimbusWhisperWrapper) MinPow() float64 {
return w.routineQueue.Send(func(c chan<- callReturn) {
c <- callReturn{value: float64(C.nimbus_get_min_pow())}
}).value.(float64)
}
// BloomFilter returns the aggregated bloom filter for all the topics of interest.
// The nodes are required to send only messages that match the advertised bloom filter.
// If a message does not match the bloom, it will tantamount to spam, and the peer will
// be disconnected.
func (w *nimbusWhisperWrapper) BloomFilter() []byte {
return w.routineQueue.Send(func(c chan<- callReturn) {
// Allocate a buffer for Nimbus to return the bloom filter on
dataC := C.malloc(C.size_t(C.BLOOM_LEN))
defer C.free(unsafe.Pointer(dataC))
C.nimbus_get_bloom_filter((*C.uchar)(dataC))
// Move the returned data into a Go array
data := make([]byte, C.BLOOM_LEN)
copy(data, C.GoBytes(dataC, C.BLOOM_LEN))
c <- callReturn{value: data}
}).value.([]byte)
}
// GetCurrentTime returns current time.
func (w *nimbusWhisperWrapper) GetCurrentTime() time.Time {
return w.timesource()
}
// SetTimeSource assigns a particular source of time to a whisper object.
func (w *nimbusWhisperWrapper) SetTimeSource(timesource func() time.Time) {
w.timesource = timesource
}
func (w *nimbusWhisperWrapper) SubscribeEnvelopeEvents(eventsProxy chan<- types.EnvelopeEvent) types.Subscription {
// TODO: when mailserver support is implemented
panic("not implemented")
}
func (w *nimbusWhisperWrapper) GetPrivateKey(id string) (*ecdsa.PrivateKey, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
privKeyC := C.malloc(types.AesKeyLength)
defer C.free(unsafe.Pointer(privKeyC))
if !C.nimbus_get_private_key(idC, (*C.uchar)(privKeyC)) {
c <- callReturn{err: errors.New("failed to get private key from Nimbus")}
return
}
pk, err := crypto.ToECDSA(C.GoBytes(privKeyC, C.PRIVKEY_LEN))
if err != nil {
c <- callReturn{err: err}
return
}
c <- callReturn{value: pk}
})
if retVal.err != nil {
return nil, retVal.err
}
return retVal.value.(*ecdsa.PrivateKey), nil
}
// AddKeyPair imports a asymmetric private key and returns a deterministic identifier.
func (w *nimbusWhisperWrapper) AddKeyPair(key *ecdsa.PrivateKey) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
privKey := crypto.FromECDSA(key)
privKeyC := C.CBytes(privKey)
defer C.free(unsafe.Pointer(privKeyC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if !C.nimbus_add_keypair((*C.uchar)(privKeyC), (*C.uchar)(idC)) {
c <- callReturn{err: errors.New("failed to add keypair to Nimbus")}
return
}
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
// DeleteKeyPair deletes the key with the specified ID if it exists.
func (w *nimbusWhisperWrapper) DeleteKeyPair(keyID string) bool {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC, err := decodeHexID(keyID)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(keyC))
c <- callReturn{value: C.nimbus_delete_keypair(keyC)}
})
if retVal.err != nil {
return false
}
return retVal.value.(bool)
}
// DeleteKeyPairs removes all cryptographic identities known to the node
func (w *nimbusWhisperWrapper) DeleteKeyPairs() error {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
C.nimbus_delete_keypairs()
c <- callReturn{}
})
return retVal.err
}
func (w *nimbusWhisperWrapper) AddSymKeyDirect(key []byte) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
keyC := C.CBytes(key)
defer C.free(unsafe.Pointer(keyC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if !C.nimbus_add_symkey((*C.uchar)(keyC), (*C.uchar)(idC)) {
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
return
}
c <- callReturn{value: types.EncodeHex(C.GoBytes(idC, C.ID_LEN))}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) AddSymKeyFromPassword(password string) (string, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
passwordC := C.CString(password)
defer C.free(unsafe.Pointer(passwordC))
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if C.nimbus_add_symkey_from_password(passwordC, (*C.uchar)(idC)) {
id := C.GoBytes(idC, C.ID_LEN)
c <- callReturn{value: types.EncodeHex(id)}
} else {
c <- callReturn{err: errors.New("failed to add symkey to Nimbus")}
}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) DeleteSymKey(id string) bool {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
c <- callReturn{value: C.nimbus_delete_symkey(idC)}
})
if retVal.err != nil {
return false
}
return retVal.value.(bool)
}
func (w *nimbusWhisperWrapper) GetSymKey(id string) ([]byte, error) {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
// Allocate a buffer for Nimbus to return the symkey on
dataC := C.malloc(C.size_t(C.SYMKEY_LEN))
defer C.free(unsafe.Pointer(dataC))
if !C.nimbus_get_symkey(idC, (*C.uchar)(dataC)) {
c <- callReturn{err: errors.New("symkey not found")}
return
}
c <- callReturn{value: C.GoBytes(dataC, C.SYMKEY_LEN)}
})
if retVal.err != nil {
return nil, retVal.err
}
return retVal.value.([]byte), nil
}
//export onMessageHandler
func onMessageHandler(msg *C.received_message, udata unsafe.Pointer) {
messageList := (gopointer.Restore(udata)).(*list.List)
topic := types.TopicType{}
copy(topic[:], C.GoBytes(unsafe.Pointer(&msg.topic[0]), types.TopicLength)[:types.TopicLength])
wrappedMsg := &types.Message{
TTL: uint32(msg.ttl),
Timestamp: uint32(msg.timestamp),
Topic: topic,
Payload: C.GoBytes(unsafe.Pointer(msg.decoded), C.int(msg.decodedLen)),
PoW: float64(msg.pow),
Hash: C.GoBytes(unsafe.Pointer(&msg.hash[0]), types.HashLength),
P2P: true,
}
if msg.source != nil {
wrappedMsg.Sig = append([]byte{0x04}, C.GoBytes(unsafe.Pointer(msg.source), types.PubKeyLength)...)
}
if msg.recipientPublicKey != nil {
wrappedMsg.Dst = append([]byte{0x04}, C.GoBytes(unsafe.Pointer(msg.recipientPublicKey), types.PubKeyLength)...)
}
messageList.PushBack(wrappedMsg)
}
func (w *nimbusWhisperWrapper) Subscribe(opts *types.SubscriptionOptions) (string, error) {
f, err := w.createFilterWrapper("", opts)
if err != nil {
return "", err
}
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
// Create a message store for this filter, so we can add new messages to it from the nimbus_subscribe_filter callback
messageList := list.New()
idC := C.malloc(C.size_t(C.ID_LEN))
defer C.free(idC)
if !C.nimbus_subscribe_filter(
GetNimbusFilterFrom(f),
(C.received_msg_handler)(unsafe.Pointer(C.onMessageHandler_cgo)), gopointer.Save(messageList),
(*C.uchar)(idC)) {
c <- callReturn{err: errors.New("failed to subscribe to filter in Nimbus")}
return
}
filterID := C.GoString((*C.char)(idC))
w.filterMessagesMu.Lock()
w.filterMessages[filterID] = messageList // TODO: Check if this is done too late (race condition with onMessageHandler)
w.filterMessagesMu.Unlock()
f.(*nimbusFilterWrapper).id = filterID
c <- callReturn{value: filterID}
})
if retVal.err != nil {
return "", retVal.err
}
return retVal.value.(string), nil
}
func (w *nimbusWhisperWrapper) GetFilter(id string) types.Filter {
idC := C.CString(id)
defer C.free(unsafe.Pointer(idC))
panic("GetFilter not implemented")
// pFilter := C.nimbus_get_filter(idC)
// return NewNimbusFilterWrapper(pFilter, id, false)
}
func (w *nimbusWhisperWrapper) Unsubscribe(id string) error {
retVal := w.routineQueue.Send(func(c chan<- callReturn) {
idC, err := decodeHexID(id)
if err != nil {
c <- callReturn{err: err}
return
}
defer C.free(unsafe.Pointer(idC))
if ok := C.nimbus_unsubscribe_filter(idC); !ok {
c <- callReturn{err: errors.New("filter not found")}
return
}
w.filterMessagesMu.Lock()
if messageList, ok := w.filterMessages[id]; ok {
gopointer.Unref(gopointer.Save(messageList))
delete(w.filterMessages, id)
}
w.filterMessagesMu.Unlock()
if f, ok := w.filters[id]; ok {
f.(*nimbusFilterWrapper).Free()
delete(w.filters, id)
}
c <- callReturn{err: nil}
})
return retVal.err
}
func decodeHexID(id string) (*C.uint8_t, error) {
idBytes, err := types.DecodeHex(id)
if err == nil && len(idBytes) != C.ID_LEN {
err = fmt.Errorf("ID length must be %v bytes, actual length is %v", C.ID_LEN, len(idBytes))
}
if err != nil {
return nil, err
}
return (*C.uint8_t)(C.CBytes(idBytes)), nil
}
// copyTopicToCBuffer copies a Go topic buffer to a C topic buffer without allocating new memory
func copyTopicToCBuffer(dst *C.uchar, topic []byte) {
if len(topic) != types.TopicLength {
panic("invalid Whisper topic buffer size")
}
p := (*[types.TopicLength]C.uchar)(unsafe.Pointer(dst))
for index, b := range topic {
p[index] = C.uchar(b)
}
}
func (w *nimbusWhisperWrapper) createFilterWrapper(id string, opts *types.SubscriptionOptions) (types.Filter, error) {
if len(opts.Topics) != 1 {
return nil, errors.New("currently only 1 topic is supported by the Nimbus bridge")
}
filter := C.filter_options{
minPow: C.double(opts.PoW),
allowP2P: C.int(1),
}
copyTopicToCBuffer(&filter.topic[0], opts.Topics[0])
if opts.PrivateKeyID != "" {
if idC, err := decodeHexID(opts.PrivateKeyID); err == nil {
filter.privateKeyID = idC
} else {
return nil, err
}
}
if opts.SymKeyID != "" {
if idC, err := decodeHexID(opts.SymKeyID); err == nil {
filter.symKeyID = idC
} else {
return nil, err
}
}
return NewNimbusFilterWrapper(&filter, id, true), nil
}
func (w *nimbusWhisperWrapper) SendMessagesRequest(peerID []byte, r types.MessagesRequest) error {
return errors.New("not implemented")
}
// RequestHistoricMessages sends a message with p2pRequestCode to a specific peer,
// which is known to implement MailServer interface, and is supposed to process this
// request and respond with a number of peer-to-peer messages (possibly expired),
// which are not supposed to be forwarded any further.
// The whisper protocol is agnostic of the format and contents of envelope.
func (w *nimbusWhisperWrapper) RequestHistoricMessagesWithTimeout(peerID []byte, envelope types.Envelope, timeout time.Duration) error {
return errors.New("not implemented")
}
// SyncMessages can be sent between two Mail Servers and syncs envelopes between them.
func (w *nimbusWhisperWrapper) SyncMessages(peerID []byte, req types.SyncMailRequest) error {
return errors.New("not implemented")
}

2
mobile/status_geth.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package statusgo
import (

9
mobile/status_nimbus.go
View File

@ -1,9 +0,0 @@
// +build nimbus
package statusgo
import (
"github.com/status-im/status-go/api"
)
var statusBackend = api.NewNimbusStatusBackend()

2
node/get_status_node.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package node
import (

2
node/geth_node.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package node
import (

2
node/geth_node_test.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package node
import (

392
node/nimbus_node.go
View File

@ -1,392 +0,0 @@
// +build nimbus
package node
import (
"errors"
"fmt"
"reflect"
"github.com/syndtr/goleveldb/leveldb"
gethrpc "github.com/ethereum/go-ethereum/rpc"
"github.com/status-im/status-go/params"
nimbussvc "github.com/status-im/status-go/services/nimbus"
"github.com/status-im/status-go/services/nodebridge"
"github.com/status-im/status-go/services/shhext"
"github.com/status-im/status-go/services/status"
"github.com/status-im/status-go/timesource"
)
// Errors related to node and services creation.
var (
// ErrNodeMakeFailureFormat = "error creating p2p node: %s"
ErrWhisperServiceRegistrationFailure = errors.New("failed to register the Whisper service")
// ErrLightEthRegistrationFailure = errors.New("failed to register the LES service")
ErrLightEthRegistrationFailureUpstreamEnabled = errors.New("failed to register the LES service, upstream is also configured")
// ErrPersonalServiceRegistrationFailure = errors.New("failed to register the personal api service")
ErrStatusServiceRegistrationFailure = errors.New("failed to register the Status service")
// ErrPeerServiceRegistrationFailure = errors.New("failed to register the Peer service")
// ErrIncentivisationServiceRegistrationFailure = errors.New("failed to register the Incentivisation service")
)
func (n *NimbusStatusNode) activateServices(config *params.NodeConfig, db *leveldb.DB) error {
// start Ethereum service if we are not expected to use an upstream server
if !config.UpstreamConfig.Enabled {
} else {
if config.LightEthConfig.Enabled {
return ErrLightEthRegistrationFailureUpstreamEnabled
}
n.log.Info("LES protocol is disabled")
// `personal_sign` and `personal_ecRecover` methods are important to
// keep DApps working.
// Usually, they are provided by an ETH or a LES service, but when using
// upstream, we don't start any of these, so we need to start our own
// implementation.
// if err := n.activatePersonalService(accs, config); err != nil {
// return fmt.Errorf("%v: %v", ErrPersonalServiceRegistrationFailure, err)
// }
}
if err := n.activateNodeServices(config, db); err != nil {
return err
}
return nil
}
func (n *NimbusStatusNode) activateNodeServices(config *params.NodeConfig, db *leveldb.DB) error {
// start Whisper service.
if err := n.activateShhService(config, db); err != nil {
return fmt.Errorf("%v: %v", ErrWhisperServiceRegistrationFailure, err)
}
// // start Waku service
// if err := activateWakuService(stack, config, db); err != nil {
// return fmt.Errorf("%v: %v", ErrWakuServiceRegistrationFailure, err)
// }
// start incentivisation service
// if err := n.activateIncentivisationService(config); err != nil {
// return fmt.Errorf("%v: %v", ErrIncentivisationServiceRegistrationFailure, err)
// }
// start status service.
if err := n.activateStatusService(config); err != nil {
return fmt.Errorf("%v: %v", ErrStatusServiceRegistrationFailure, err)
}
// start peer service
// if err := activatePeerService(n); err != nil {
// return fmt.Errorf("%v: %v", ErrPeerServiceRegistrationFailure, err)
// }
return nil
}
// // activateLightEthService configures and registers the eth.Ethereum service with a given node.
// func activateLightEthService(stack *node.Node, accs *accounts.Manager, config *params.NodeConfig) error {
// if !config.LightEthConfig.Enabled {
// logger.Info("LES protocol is disabled")
// return nil
// }
// genesis, err := calculateGenesis(config.NetworkID)
// if err != nil {
// return err
// }
// ethConf := eth.DefaultConfig
// ethConf.Genesis = genesis
// ethConf.SyncMode = downloader.LightSync
// ethConf.NetworkId = config.NetworkID
// ethConf.DatabaseCache = config.LightEthConfig.DatabaseCache
// return stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
// // NOTE(dshulyak) here we set our instance of the accounts manager.
// // without sharing same instance selected account won't be visible for personal_* methods.
// nctx := &node.ServiceContext{}
// *nctx = *ctx
// nctx.AccountManager = accs
// return les.New(nctx, &ethConf)
// })
// }
// func activatePersonalService(stack *node.Node, accs *accounts.Manager, config *params.NodeConfig) error {
// return stack.Register(func(*node.ServiceContext) (node.Service, error) {
// svc := personal.New(accs)
// return svc, nil
// })
// }
// func (n *NimbusStatusNode) activatePersonalService(accs *accounts.Manager, config *params.NodeConfig) error {
// return n.Register(func(*nimbussvc.ServiceContext) (nimbussvc.Service, error) {
// svc := personal.New(accs)
// return svc, nil
// })
// }
func (n *NimbusStatusNode) activateStatusService(config *params.NodeConfig) error {
if !config.EnableStatusService {
n.log.Info("Status service api is disabled")
return nil
}
return n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
var service *nodebridge.WhisperService
if err := ctx.Service(&service); err != nil {
return nil, err
}
svc := status.New(service.Whisper)
return svc, nil
})
}
// func (n *NimbusStatusNode) activatePeerService() error {
// return n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
// svc := peer.New()
// return svc, nil
// })
// }
// func registerWhisperMailServer(whisperService *whisper.Whisper, config *params.WhisperConfig) (err error) {
// var mailServer mailserver.WhisperMailServer
// whisperService.RegisterMailServer(&mailServer)
// return mailServer.Init(whisperService, config)
// }
// func registerWakuMailServer(wakuService *waku.Waku, config *params.WakuConfig) (err error) {
// var mailServer mailserver.WakuMailServer
// wakuService.RegisterMailServer(&mailServer)
// return mailServer.Init(wakuService, config)
// }
// activateShhService configures Whisper and adds it to the given node.
func (n *NimbusStatusNode) activateShhService(config *params.NodeConfig, db *leveldb.DB) (err error) {
if !config.WhisperConfig.Enabled {
n.log.Info("SHH protocol is disabled")
return nil
}
if config.EnableNTPSync {
if err = n.Register(func(*nimbussvc.ServiceContext) (nimbussvc.Service, error) {
return timesource.Default(), nil
}); err != nil {
return
}
}
// err = n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
// return n.createShhService(ctx, &config.WhisperConfig, &config.ClusterConfig)
// })
// if err != nil {
// return
// }
// Register eth-node node bridge
err = n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
return &nodebridge.NodeService{Node: n.node}, nil
})
if err != nil {
return
}
// Register Whisper eth-node bridge
err = n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
n.log.Info("Creating WhisperService")
var ethnode *nodebridge.NodeService
if err := ctx.Service(&ethnode); err != nil {
return nil, err
}
w, err := ethnode.Node.GetWhisper(ctx)
if err != nil {
n.log.Error("GetWhisper returned error", "err", err)
return nil, err
}
return &nodebridge.WhisperService{Whisper: w}, nil
})
if err != nil {
return
}
// TODO(dshulyak) add a config option to enable it by default, but disable if app is started from statusd
return n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
var ethnode *nodebridge.NodeService
if err := ctx.Service(&ethnode); err != nil {
return nil, err
}
return shhext.NewNimbus(ethnode.Node, ctx, "shhext", db, config.ShhextConfig), nil
})
}
// activateWakuService configures Waku and adds it to the given node.
func (n *NimbusStatusNode) activateWakuService(config *params.NodeConfig, db *leveldb.DB) (err error) {
if !config.WakuConfig.Enabled {
n.log.Info("Waku protocol is disabled")
return nil
}
panic("not implemented")
// err = n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
// return createWakuService(ctx, &config.WakuConfig, &config.ClusterConfig)
// })
// if err != nil {
// return
// }
// // TODO(dshulyak) add a config option to enable it by default, but disable if app is started from statusd
// return n.Register(func(ctx *nimbussvc.ServiceContext) (nimbussvc.Service, error) {
// var ethnode *nodebridge.NodeService
// if err := ctx.Service(&ethnode); err != nil {
// return nil, err
// }
// return shhext.New(ethnode.Node, ctx, "wakuext", shhext.EnvelopeSignalHandler{}, db, config.ShhextConfig), nil
// })
}
// Register injects a new service into the node's stack. The service created by
// the passed constructor must be unique in its type with regard to sibling ones.
func (n *NimbusStatusNode) Register(constructor nimbussvc.ServiceConstructor) error {
n.lock.Lock()
defer n.lock.Unlock()
if n.isRunning() {
return ErrNodeRunning
}
n.serviceFuncs = append(n.serviceFuncs, constructor)
return nil
}
func (n *NimbusStatusNode) startServices() error {
services := make(map[reflect.Type]nimbussvc.Service)
for _, constructor := range n.serviceFuncs {
// Create a new context for the particular service
ctxServices := make(map[reflect.Type]nimbussvc.Service)
for kind, s := range services { // copy needed for threaded access
ctxServices[kind] = s
}
ctx := nimbussvc.NewServiceContext(n.config, ctxServices)
//EventMux: n.eventmux,
//AccountManager: n.accman,
// Construct and save the service
service, err := constructor(ctx)
if err != nil {
n.log.Info("Service constructor returned error", "err", err)
return err
}
kind := reflect.TypeOf(service)
if _, exists := services[kind]; exists {
return &nimbussvc.DuplicateServiceError{Kind: kind}
}
services[kind] = service
}
// Start each of the services
var started []reflect.Type
for kind, service := range services {
// Start the next service, stopping all previous upon failure
if err := service.StartService(); err != nil {
for _, kind := range started {
services[kind].Stop()
}
return err
}
// Mark the service started for potential cleanup
started = append(started, kind)
}
// Lastly start the configured RPC interfaces
if err := n.startRPC(services); err != nil {
for _, service := range services {
service.Stop()
}
return err
}
// Finish initializing the startup
n.services = services
return nil
}
// startRPC is a helper method to start all the various RPC endpoint during node
// startup. It's not meant to be called at any time afterwards as it makes certain
// assumptions about the state of the node.
func (n *NimbusStatusNode) startRPC(services map[reflect.Type]nimbussvc.Service) error {
// Gather all the possible APIs to surface
apis := []gethrpc.API{}
for _, service := range services {
apis = append(apis, service.APIs()...)
}
// Start the various API endpoints, terminating all in case of errors
if err := n.startInProc(apis); err != nil {
return err
}
if err := n.startPublicInProc(apis, n.config.FormatAPIModules()); err != nil {
n.stopInProc()
return err
}
// All API endpoints started successfully
n.rpcAPIs = apis
return nil
}
// startInProc initializes an in-process RPC endpoint.
func (n *NimbusStatusNode) startInProc(apis []gethrpc.API) error {
n.log.Debug("startInProc", "apis", apis)
// Register all the APIs exposed by the services
handler := gethrpc.NewServer()
for _, api := range apis {
n.log.Debug("Registering InProc", "namespace", api.Namespace)
if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
return err
}
n.log.Debug("InProc registered", "namespace", api.Namespace)
}
n.inprocHandler = handler
return nil
}
// stopInProc terminates the in-process RPC endpoint.
func (n *NimbusStatusNode) stopInProc() {
if n.inprocHandler != nil {
n.inprocHandler.Stop()
n.inprocHandler = nil
}
}
// startPublicInProc initializes an in-process RPC endpoint for public APIs.
func (n *NimbusStatusNode) startPublicInProc(apis []gethrpc.API, modules []string) error {
n.log.Debug("startPublicInProc", "apis", apis, "modules", modules)
// Generate the whitelist based on the allowed modules
whitelist := make(map[string]bool)
for _, module := range modules {
whitelist[module] = true
}
// Register all the public APIs exposed by the services
handler := gethrpc.NewServer()
for _, api := range apis {
if whitelist[api.Namespace] || (len(whitelist) == 0 && api.Public) {
n.log.Debug("Registering InProc public", "service", api.Service, "namespace", api.Namespace)
if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
return err
}
n.log.Debug("InProc public registered", "service", api.Service, "namespace", api.Namespace)
}
}
n.inprocPublicHandler = handler
return nil
}
// stopPublicInProc terminates the in-process RPC endpoint for public APIs.
func (n *NimbusStatusNode) stopPublicInProc() {
if n.inprocPublicHandler != nil {
n.inprocPublicHandler.Stop()
n.inprocPublicHandler = nil
}
}

798
node/nimbus_status_node.go
View File

@ -1,798 +0,0 @@
// +build nimbus
package node
import (
"context"
"crypto/ecdsa"
"errors"
"fmt"
"os"
"path/filepath"
"reflect"
"strings"
"sync"
"github.com/syndtr/goleveldb/leveldb"
"github.com/ethereum/go-ethereum/log"
gethrpc "github.com/ethereum/go-ethereum/rpc"
"github.com/status-im/status-go/db"
nimbusbridge "github.com/status-im/status-go/eth-node/bridge/nimbus"
"github.com/status-im/status-go/eth-node/crypto"
"github.com/status-im/status-go/eth-node/types"
"github.com/status-im/status-go/params"
"github.com/status-im/status-go/rpc"
nimbussvc "github.com/status-im/status-go/services/nimbus"
"github.com/status-im/status-go/services/nodebridge"
"github.com/status-im/status-go/services/shhext"
"github.com/status-im/status-go/services/status"
)
// // tickerResolution is the delta to check blockchain sync progress.
// const tickerResolution = time.Second
// errors
var (
ErrNodeRunning = errors.New("node is already running")
ErrNodeStopped = errors.New("node not started")
ErrNoRunningNode = errors.New("there is no running node")
ErrServiceUnknown = errors.New("service unknown")
)
// NimbusStatusNode abstracts contained geth node and provides helper methods to
// interact with it.
type NimbusStatusNode struct {
mu sync.RWMutex
//eventmux *event.TypeMux // Event multiplexer used between the services of a stack
config *params.NodeConfig // Status node configuration
privateKey *ecdsa.PrivateKey
node nimbusbridge.Node
nodeRunning bool
rpcClient *rpc.Client // reference to public RPC client
rpcPrivateClient *rpc.Client // reference to private RPC client (can call private APIs)
rpcAPIs []gethrpc.API // List of APIs currently provided by the node
inprocHandler *gethrpc.Server // In-process RPC request handler to process the API requests
inprocPublicHandler *gethrpc.Server // In-process RPC request handler to process the public API requests
serviceFuncs []nimbussvc.ServiceConstructor // Service constructors (in dependency order)
services map[reflect.Type]nimbussvc.Service // Currently running services
// discovery discovery.Discovery
// register *peers.Register
// peerPool *peers.PeerPool
db *leveldb.DB // used as a cache for PeerPool
//stop chan struct{} // Channel to wait for termination notifications
lock sync.RWMutex
log log.Logger
}
// NewNimbus makes new instance of NimbusStatusNode.
func NewNimbus() *NimbusStatusNode {
return &NimbusStatusNode{
//eventmux: new(event.TypeMux),
log: log.New("package", "status-go/node.NimbusStatusNode"),
}
}
// Config exposes reference to running node's configuration
func (n *NimbusStatusNode) Config() *params.NodeConfig {
n.mu.RLock()
defer n.mu.RUnlock()
return n.config
}
// GethNode returns underlying geth node.
// func (n *NimbusStatusNode) GethNode() *node.Node {
// n.mu.RLock()
// defer n.mu.RUnlock()
// return n.gethNode
// }
// Server retrieves the currently running P2P network layer.
// func (n *NimbusStatusNode) Server() *p2p.Server {
// n.mu.RLock()
// defer n.mu.RUnlock()
// if n.gethNode == nil {
// return nil
// }
// return n.gethNode.Server()
// }
// Start starts current NimbusStatusNode, failing if it's already started.
// It accepts a list of services that should be added to the node.
func (n *NimbusStatusNode) Start(config *params.NodeConfig, services ...nimbussvc.ServiceConstructor) error {
panic("Start")
return n.StartWithOptions(config, NimbusStartOptions{
Services: services,
StartDiscovery: true,
// AccountsManager: accs,
})
}
// NimbusStartOptions allows to control some parameters of Start() method.
type NimbusStartOptions struct {
Node types.Node
Services []nimbussvc.ServiceConstructor
StartDiscovery bool
// AccountsManager *accounts.Manager
}
// StartWithOptions starts current NimbusStatusNode, failing if it's already started.
// It takes some options that allows to further configure starting process.
func (n *NimbusStatusNode) StartWithOptions(config *params.NodeConfig, options NimbusStartOptions) error {
n.mu.Lock()
defer n.mu.Unlock()
if n.isRunning() {
n.log.Debug("node is already running")
return ErrNodeRunning
}
n.log.Debug("starting with NodeConfig", "ClusterConfig", config.ClusterConfig)
db, err := db.Create(config.DataDir, params.StatusDatabase)
if err != nil {
return fmt.Errorf("failed to create database at %s: %v", config.DataDir, err)
}
n.db = db
err = n.startWithDB(config, db, options.Services)
// continue only if there was no error when starting node with a db
if err == nil && options.StartDiscovery && n.discoveryEnabled() {
// err = n.startDiscovery()
}
if err != nil {
if dberr := db.Close(); dberr != nil {
n.log.Error("error while closing leveldb after node crash", "error", dberr)
}
n.db = nil
return err
}
return nil
}
func (n *NimbusStatusNode) startWithDB(config *params.NodeConfig, db *leveldb.DB, services []nimbussvc.ServiceConstructor) error {
if err := n.createNode(config, services, db); err != nil {
return err
}
if err := n.setupRPCClient(); err != nil {
return err
}
return nil
}
func (n *NimbusStatusNode) createNode(config *params.NodeConfig, services []nimbussvc.ServiceConstructor, db *leveldb.DB) error {
var privateKey *ecdsa.PrivateKey
if config.NodeKey != "" {
var err error
privateKey, err = crypto.HexToECDSA(config.NodeKey)
if err != nil {
return err
}
}
n.privateKey = privateKey
n.node = nimbusbridge.NewNodeBridge()
err := n.activateServices(config, db)
if err != nil {
return err
}
if err = n.start(config, services); err != nil {
return err
}
return nil
}
// start starts current NimbusStatusNode, will fail if it's already started.
func (n *NimbusStatusNode) start(config *params.NodeConfig, services []nimbussvc.ServiceConstructor) error {
for _, service := range services {
if err := n.Register(service); err != nil {
return err
}
}
n.config = config
n.startServices()
err := n.node.StartNimbus(n.privateKey, config.ListenAddr, true)
n.nodeRunning = err == nil
return err
}
func (n *NimbusStatusNode) setupRPCClient() (err error) {
// setup public RPC client
gethNodeClient := gethrpc.DialInProc(n.inprocPublicHandler)
n.rpcClient, err = rpc.NewClient(gethNodeClient, n.config.UpstreamConfig)
if err != nil {
return
}
// setup private RPC client
gethNodePrivateClient := gethrpc.DialInProc(n.inprocHandler)
n.rpcPrivateClient, err = rpc.NewClient(gethNodePrivateClient, n.config.UpstreamConfig)
return
}
func (n *NimbusStatusNode) discoveryEnabled() bool {
return n.config != nil && (!n.config.NoDiscovery || n.config.Rendezvous) && n.config.ClusterConfig.Enabled
}
// func (n *NimbusStatusNode) discoverNode() (*enode.Node, error) {
// if !n.isRunning() {
// return nil, nil
// }
// server := n.gethNode.Server()
// discNode := server.Self()
// if n.config.AdvertiseAddr == "" {
// return discNode, nil
// }
// n.log.Info("Using AdvertiseAddr for rendezvous", "addr", n.config.AdvertiseAddr)
// r := discNode.Record()
// r.Set(enr.IP(net.ParseIP(n.config.AdvertiseAddr)))
// if err := enode.SignV4(r, server.PrivateKey); err != nil {
// return nil, err
// }
// return enode.New(enode.ValidSchemes[r.IdentityScheme()], r)
// }
// func (n *NimbusStatusNode) startRendezvous() (discovery.Discovery, error) {
// if !n.config.Rendezvous {
// return nil, errors.New("rendezvous is not enabled")
// }
// if len(n.config.ClusterConfig.RendezvousNodes) == 0 {
// return nil, errors.New("rendezvous node must be provided if rendezvous discovery is enabled")
// }
// maddrs := make([]ma.Multiaddr, len(n.config.ClusterConfig.RendezvousNodes))
// for i, addr := range n.config.ClusterConfig.RendezvousNodes {
// var err error
// maddrs[i], err = ma.NewMultiaddr(addr)
// if err != nil {
// return nil, fmt.Errorf("failed to parse rendezvous node %s: %v", n.config.ClusterConfig.RendezvousNodes[0], err)
// }
// }
// node, err := n.discoverNode()
// if err != nil {
// return nil, fmt.Errorf("failed to get a discover node: %v", err)
// }
// return discovery.NewRendezvous(maddrs, n.gethNode.Server().PrivateKey, node)
// }
// StartDiscovery starts the peers discovery protocols depending on the node config.
func (n *NimbusStatusNode) StartDiscovery() error {
n.mu.Lock()
defer n.mu.Unlock()
if n.discoveryEnabled() {
// return n.startDiscovery()
}
return nil
}
// func (n *NimbusStatusNode) startDiscovery() error {
// if n.isDiscoveryRunning() {
// return ErrDiscoveryRunning
// }
// discoveries := []discovery.Discovery{}
// if !n.config.NoDiscovery {
// discoveries = append(discoveries, discovery.NewDiscV5(
// n.gethNode.Server().PrivateKey,
// n.config.ListenAddr,
// parseNodesV5(n.config.ClusterConfig.BootNodes)))
// }
// if n.config.Rendezvous {
// d, err := n.startRendezvous()
// if err != nil {
// return err
// }
// discoveries = append(discoveries, d)
// }
// if len(discoveries) == 0 {
// return errors.New("wasn't able to register any discovery")
// } else if len(discoveries) > 1 {
// n.discovery = discovery.NewMultiplexer(discoveries)
// } else {
// n.discovery = discoveries[0]
// }
// log.Debug(
// "using discovery",
// "instance", reflect.TypeOf(n.discovery),
// "registerTopics", n.config.RegisterTopics,
// "requireTopics", n.config.RequireTopics,
// )
// n.register = peers.NewRegister(n.discovery, n.config.RegisterTopics...)
// options := peers.NewDefaultOptions()
// // TODO(dshulyak) consider adding a flag to define this behaviour
// options.AllowStop = len(n.config.RegisterTopics) == 0
// options.TrustedMailServers = parseNodesToNodeID(n.config.ClusterConfig.TrustedMailServers)
// options.MailServerRegistryAddress = n.config.MailServerRegistryAddress
// n.peerPool = peers.NewPeerPool(
// n.discovery,
// n.config.RequireTopics,
// peers.NewCache(n.db),
// options,
// )
// if err := n.discovery.Start(); err != nil {
// return err
// }
// if err := n.register.Start(); err != nil {
// return err
// }
// return n.peerPool.Start(n.gethNode.Server(), n.rpcClient)
// }
// Stop will stop current NimbusStatusNode. A stopped node cannot be resumed.
func (n *NimbusStatusNode) Stop() error {
n.mu.Lock()
defer n.mu.Unlock()
if !n.isRunning() {
return ErrNoRunningNode
}
var errs []error
// Terminate all subsystems and collect any errors
if err := n.stop(); err != nil && err != ErrNodeStopped {
errs = append(errs, err)
}
// Report any errors that might have occurred
switch len(errs) {
case 0:
return nil
case 1:
return errs[0]
default:
return fmt.Errorf("%v", errs)
}
}
// StopError is returned if a Node fails to stop either any of its registered
// services or itself.
type StopError struct {
Server error
Services map[reflect.Type]error
}
// Error generates a textual representation of the stop error.
func (e *StopError) Error() string {
return fmt.Sprintf("server: %v, services: %v", e.Server, e.Services)
}
// stop will stop current NimbusStatusNode. A stopped node cannot be resumed.
func (n *NimbusStatusNode) stop() error {
// if n.isDiscoveryRunning() {
// if err := n.stopDiscovery(); err != nil {
// n.log.Error("Error stopping the discovery components", "error", err)
// }
// n.register = nil
// n.peerPool = nil
// n.discovery = nil
// }
// if err := n.gethNode.Stop(); err != nil {
// return err
// }
// Terminate the API, services and the p2p server.
n.stopPublicInProc()
n.stopInProc()
n.rpcClient = nil
n.rpcPrivateClient = nil
n.rpcAPIs = nil
failure := &StopError{
Services: make(map[reflect.Type]error),
}
for kind, service := range n.services {
if err := service.Stop(); err != nil {
failure.Services[kind] = err
}
}
n.services = nil
// We need to clear `node` because config is passed to `Start()`
// and may be completely different. Similarly with `config`.
if n.node != nil {
n.node.Stop()
n.node = nil
}
n.nodeRunning = false
n.config = nil
if n.db != nil {
err := n.db.Close()
n.db = nil
return err
}
if len(failure.Services) > 0 {
return failure
}
return nil
}
func (n *NimbusStatusNode) isDiscoveryRunning() bool {
return false //n.register != nil || n.peerPool != nil || n.discovery != nil
}
// func (n *NimbusStatusNode) stopDiscovery() error {
// n.register.Stop()
// n.peerPool.Stop()
// return n.discovery.Stop()
// }
// ResetChainData removes chain data if node is not running.
func (n *NimbusStatusNode) ResetChainData(config *params.NodeConfig) error {
n.mu.Lock()
defer n.mu.Unlock()
if n.isRunning() {
return ErrNodeRunning
}
chainDataDir := filepath.Join(config.DataDir, config.Name, "lightchaindata")
if _, err := os.Stat(chainDataDir); os.IsNotExist(err) {
return err
}
err := os.RemoveAll(chainDataDir)
if err == nil {
n.log.Info("Chain data has been removed", "dir", chainDataDir)
}
return err
}
// IsRunning confirm that node is running.
func (n *NimbusStatusNode) IsRunning() bool {
n.mu.RLock()
defer n.mu.RUnlock()
return n.isRunning()
}
func (n *NimbusStatusNode) isRunning() bool {
return n.node != nil && n.nodeRunning // && n.gethNode.Server() != nil
}
// populateStaticPeers connects current node with our publicly available LES/SHH/Swarm cluster
func (n *NimbusStatusNode) populateStaticPeers() error {
if !n.config.ClusterConfig.Enabled {
n.log.Info("Static peers are disabled")
return nil
}
for _, enode := range n.config.ClusterConfig.StaticNodes {
if err := n.addPeer(enode); err != nil {
n.log.Error("Static peer addition failed", "error", err)
return err
}
n.log.Info("Static peer added", "enode", enode)
}
return nil
}
func (n *NimbusStatusNode) removeStaticPeers() error {
if !n.config.ClusterConfig.Enabled {
n.log.Info("Static peers are disabled")
return nil
}
for _, enode := range n.config.ClusterConfig.StaticNodes {
if err := n.removePeer(enode); err != nil {
n.log.Error("Static peer deletion failed", "error", err)
return err
}
n.log.Info("Static peer deleted", "enode", enode)
}
return nil
}
// ReconnectStaticPeers removes and adds static peers to a server.
func (n *NimbusStatusNode) ReconnectStaticPeers() error {
n.mu.Lock()
defer n.mu.Unlock()
if !n.isRunning() {
return ErrNoRunningNode
}
if err := n.removeStaticPeers(); err != nil {
return err
}
return n.populateStaticPeers()
}
// AddPeer adds new static peer node
func (n *NimbusStatusNode) AddPeer(url string) error {
n.mu.RLock()
defer n.mu.RUnlock()
return n.addPeer(url)
}
// addPeer adds new static peer node
func (n *NimbusStatusNode) addPeer(url string) error {
if !n.isRunning() {
return ErrNoRunningNode
}
n.node.AddPeer(url)
return nil
}
func (n *NimbusStatusNode) removePeer(url string) error {
if !n.isRunning() {
return ErrNoRunningNode
}
n.node.RemovePeer(url)
return nil
}
// PeerCount returns the number of connected peers.
func (n *NimbusStatusNode) PeerCount() int {
n.mu.RLock()
defer n.mu.RUnlock()
if !n.isRunning() {
return 0
}
return 1
//return n.gethNode.Server().PeerCount()
}
// Service retrieves a currently running service registered of a specific type.
func (n *NimbusStatusNode) Service(service interface{}) error {
n.lock.RLock()
defer n.lock.RUnlock()
// Short circuit if the node's not running
if !n.isRunning() {
return ErrNodeStopped
}
// Otherwise try to find the service to return
element := reflect.ValueOf(service).Elem()
if running, ok := n.services[element.Type()]; ok {
element.Set(reflect.ValueOf(running))
return nil
}
return ErrServiceUnknown
}
// // LightEthereumService exposes reference to LES service running on top of the node
// func (n *NimbusStatusNode) LightEthereumService() (l *les.LightEthereum, err error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// err = n.Service(&l)
// return
// }
// StatusService exposes reference to status service running on top of the node
func (n *NimbusStatusNode) StatusService() (st *status.Service, err error) {
n.mu.RLock()
defer n.mu.RUnlock()
err = n.Service(&st)
return
}
// // PeerService exposes reference to peer service running on top of the node.
// func (n *NimbusStatusNode) PeerService() (st *peer.Service, err error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// err = n.Service(&st)
// return
// }
// WhisperService exposes reference to Whisper service running on top of the node
func (n *NimbusStatusNode) WhisperService() (w *nodebridge.WhisperService, err error) {
n.mu.RLock()
defer n.mu.RUnlock()
err = n.Service(&w)
return
}
// ShhExtService exposes reference to shh extension service running on top of the node
func (n *NimbusStatusNode) ShhExtService() (s *shhext.NimbusService, err error) {
n.mu.RLock()
defer n.mu.RUnlock()
err = n.Service(&s)
return
}
// // WalletService returns wallet.Service instance if it was started.
// func (n *NimbusStatusNode) WalletService() (s *wallet.Service, err error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// err = n.Service(&s)
// return
// }
// // BrowsersService returns browsers.Service instance if it was started.
// func (n *NimbusStatusNode) BrowsersService() (s *browsers.Service, err error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// err = n.Service(&s)
// return
// }
// // PermissionsService returns browsers.Service instance if it was started.
// func (n *NimbusStatusNode) PermissionsService() (s *permissions.Service, err error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// err = n.Service(&s)
// return
// }
// // AccountManager exposes reference to node's accounts manager
// func (n *NimbusStatusNode) AccountManager() (*accounts.Manager, error) {
// n.mu.RLock()
// defer n.mu.RUnlock()
// if n.gethNode == nil {
// return nil, ErrNoGethNode
// }
// return n.gethNode.AccountManager(), nil
// }
// RPCClient exposes reference to RPC client connected to the running node.
func (n *NimbusStatusNode) RPCClient() *rpc.Client {
n.mu.RLock()
defer n.mu.RUnlock()
return n.rpcClient
}
// RPCPrivateClient exposes reference to RPC client connected to the running node
// that can call both public and private APIs.
func (n *NimbusStatusNode) RPCPrivateClient() *rpc.Client {
n.mu.Lock()
defer n.mu.Unlock()
return n.rpcPrivateClient
}
// ChaosModeCheckRPCClientsUpstreamURL updates RPCClient and RPCPrivateClient upstream URLs,
// if defined, without restarting the node. This is required for the Chaos Unicorn Day.
// Additionally, if the passed URL is Infura, it changes it to httpbin.org/status/500.
func (n *NimbusStatusNode) ChaosModeCheckRPCClientsUpstreamURL(on bool) error {
url := n.config.UpstreamConfig.URL
if on {
if strings.Contains(url, "infura.io") {
url = "https://httpbin.org/status/500"
}
}
publicClient := n.RPCClient()
if publicClient != nil {
if err := publicClient.UpdateUpstreamURL(url); err != nil {
return err
}
}
privateClient := n.RPCPrivateClient()
if privateClient != nil {
if err := privateClient.UpdateUpstreamURL(url); err != nil {
return err
}
}
return nil
}
// EnsureSync waits until blockchain synchronization
// is complete and returns.
func (n *NimbusStatusNode) EnsureSync(ctx context.Context) error {
// Don't wait for any blockchain sync for the
// local private chain as blocks are never mined.
if n.config.NetworkID == 0 || n.config.NetworkID == params.StatusChainNetworkID {
return nil
}
return n.ensureSync(ctx)
}
func (n *NimbusStatusNode) ensureSync(ctx context.Context) error {
return errors.New("Sync not implemented")
// les, err := n.LightEthereumService()
// if err != nil {
// return fmt.Errorf("failed to get LES service: %v", err)
// }
// downloader := les.Downloader()
// if downloader == nil {
// return errors.New("LightEthereumService downloader is nil")
// }
// progress := downloader.Progress()
// if n.PeerCount() > 0 && progress.CurrentBlock >= progress.HighestBlock {
// n.log.Debug("Synchronization completed", "current block", progress.CurrentBlock, "highest block", progress.HighestBlock)
// return nil
// }
// ticker := time.NewTicker(tickerResolution)
// defer ticker.Stop()
// progressTicker := time.NewTicker(time.Minute)
// defer progressTicker.Stop()
// for {
// select {
// case <-ctx.Done():
// return errors.New("timeout during node synchronization")
// case <-ticker.C:
// if n.PeerCount() == 0 {
// n.log.Debug("No established connections with any peers, continue waiting for a sync")
// continue
// }
// if downloader.Synchronising() {
// n.log.Debug("Synchronization is in progress")
// continue
// }
// progress = downloader.Progress()
// if progress.CurrentBlock >= progress.HighestBlock {
// n.log.Info("Synchronization completed", "current block", progress.CurrentBlock, "highest block", progress.HighestBlock)
// return nil
// }
// n.log.Debug("Synchronization is not finished", "current", progress.CurrentBlock, "highest", progress.HighestBlock)
// case <-progressTicker.C:
// progress = downloader.Progress()
// n.log.Warn("Synchronization is not finished", "current", progress.CurrentBlock, "highest", progress.HighestBlock)
// }
// }
}
// // Discover sets up the discovery for a specific topic.
// func (n *NimbusStatusNode) Discover(topic string, max, min int) (err error) {
// if n.peerPool == nil {
// return errors.New("peerPool not running")
// }
// return n.peerPool.UpdateTopic(topic, params.Limits{
// Max: max,
// Min: min,
// })
// }

15
services/accounts/service_nimbus.go
View File

@ -1,15 +0,0 @@
// +build nimbus
package accounts
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that Service implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*Service)(nil)
// Start a service.
func (s *Service) StartService() error {
return nil
}

2
services/ext/mailrequests.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package ext
import (

2
services/ext/mailrequests_test.go
View File

@ -1,5 +1,3 @@
// +build !nimbus
package ext
import (

84
services/nimbus/service.go
View File

@ -1,84 +0,0 @@
// +build nimbus
package nimbus
import (
"errors"
"fmt"
"reflect"
gethrpc "github.com/ethereum/go-ethereum/rpc"
"github.com/status-im/status-go/params"
)
// errors
var (
ErrNodeStopped = errors.New("node not started")
ErrServiceUnknown = errors.New("service unknown")
)
// DuplicateServiceError is returned during Node startup if a registered service
// constructor returns a service of the same type that was already started.
type DuplicateServiceError struct {
Kind reflect.Type
}
// Error generates a textual representation of the duplicate service error.
func (e *DuplicateServiceError) Error() string {
return fmt.Sprintf("duplicate service: %v", e.Kind)
}
// ServiceContext is a collection of service independent options inherited from
// the protocol stack, that is passed to all constructors to be optionally used;
// as well as utility methods to operate on the service environment.
type ServiceContext struct {
config *params.NodeConfig
services map[reflect.Type]Service // Index of the already constructed services
// EventMux *event.TypeMux // Event multiplexer used for decoupled notifications
// AccountManager *accounts.Manager // Account manager created by the node.
}
func NewServiceContext(config *params.NodeConfig, services map[reflect.Type]Service) *ServiceContext {
return &ServiceContext{
config: config,
services: services,
}
}
// Service retrieves a currently running service registered of a specific type.
func (ctx *ServiceContext) Service(service interface{}) error {
element := reflect.ValueOf(service).Elem()
if running, ok := ctx.services[element.Type()]; ok {
element.Set(reflect.ValueOf(running))
return nil
}
return ErrServiceUnknown
}
// ServiceConstructor is the function signature of the constructors needed to be
// registered for service instantiation.
type ServiceConstructor func(ctx *ServiceContext) (Service, error)
// Service is an individual protocol that can be registered into a node.
//
// Notes:
//
// • Service life-cycle management is delegated to the node. The service is allowed to
// initialize itself upon creation, but no goroutines should be spun up outside of the
// Start method.
//
// • Restart logic is not required as the node will create a fresh instance
// every time a service is started.
type Service interface {
// APIs retrieves the list of RPC descriptors the service provides
APIs() []gethrpc.API
// StartService is called after all services have been constructed and the networking
// layer was also initialized to spawn any goroutines required by the service.
StartService() error
// Stop terminates all goroutines belonging to the service, blocking until they
// are all terminated.
Stop() error
}

14
services/nodebridge/node_service_nimbus.go
View File

@ -1,14 +0,0 @@
// +build nimbus
package nodebridge
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that NodeService implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*NodeService)(nil)
func (w *NodeService) StartService() error {
return nil
}

14
services/nodebridge/whisper_service_nimbus.go
View File

@ -1,14 +0,0 @@
// +build nimbus
package nodebridge
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that WhisperService implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*WhisperService)(nil)
func (w *WhisperService) StartService() error {
return nil
}

15
services/rpcfilters/service_nimbus.go
View File

@ -1,15 +0,0 @@
// +build nimbus
package rpcfilters
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that Service implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*Service)(nil)
// StartService is run when a service is started.
func (s *Service) StartService() error {
return s.Start(nil)
}

16
services/status/service_nimbus.go
View File

@ -1,16 +0,0 @@
// +build nimbus
package status
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that Service implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*Service)(nil)
// StartService is run when a service is started.
// It does nothing in this case but is required by `nimbussvc.Service` interface.
func (s *Service) StartService() error {
return nil
}

15
services/subscriptions/service_nimbus.go
View File

@ -1,15 +0,0 @@
// +build nimbus
package subscriptions
import (
nimbussvc "github.com/status-im/status-go/services/nimbus"
)
// Make sure that Service implements nimbussvc.Service interface.
var _ nimbussvc.Service = (*Service)(nil)
// StartService is run when a service is started.
func (s *Service) StartService() error {
return nil
}