go-waku/waku/node.go

519 lines
14 KiB
Go
Raw Normal View History

package waku
import (
"context"
"crypto/ecdsa"
2021-04-19 00:03:16 +00:00
"database/sql"
2022-07-25 12:24:42 +00:00
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"net"
"os"
"os/signal"
"syscall"
"time"
2022-07-25 12:24:42 +00:00
"github.com/ethereum/go-ethereum/accounts/keystore"
2022-07-28 14:04:33 +00:00
"github.com/ethereum/go-ethereum/common"
2021-03-23 14:46:16 +00:00
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/p2p/enode"
dssql "github.com/ipfs/go-ds-sql"
"github.com/urfave/cli/v2"
"go.uber.org/zap"
"github.com/libp2p/go-libp2p"
2021-11-16 14:22:01 +00:00
"github.com/libp2p/go-libp2p-core/discovery"
2022-05-27 13:25:06 +00:00
"github.com/libp2p/go-libp2p-core/peer"
2021-10-15 02:15:02 +00:00
"github.com/libp2p/go-libp2p/config"
2022-08-25 20:36:04 +00:00
"github.com/libp2p/go-libp2p/p2p/transport/tcp"
"github.com/libp2p/go-libp2p-peerstore/pstoreds"
pubsub "github.com/libp2p/go-libp2p-pubsub"
2022-08-25 20:36:04 +00:00
ws "github.com/libp2p/go-libp2p/p2p/transport/websocket"
2021-04-22 13:07:22 +00:00
"github.com/multiformats/go-multiaddr"
2021-10-05 02:13:54 +00:00
rendezvous "github.com/status-im/go-waku-rendezvous"
2022-05-27 13:25:06 +00:00
"github.com/status-im/go-waku/logging"
"github.com/status-im/go-waku/waku/metrics"
"github.com/status-im/go-waku/waku/persistence"
"github.com/status-im/go-waku/waku/persistence/sqlite"
2021-11-16 14:22:01 +00:00
"github.com/status-im/go-waku/waku/v2/dnsdisc"
"github.com/status-im/go-waku/waku/v2/node"
"github.com/status-im/go-waku/waku/v2/protocol/filter"
"github.com/status-im/go-waku/waku/v2/protocol/lightpush"
"github.com/status-im/go-waku/waku/v2/protocol/relay"
"github.com/status-im/go-waku/waku/v2/protocol/store"
"github.com/status-im/go-waku/waku/v2/rest"
"github.com/status-im/go-waku/waku/v2/rpc"
"github.com/status-im/go-waku/waku/v2/utils"
)
func failOnErr(err error, msg string) {
if err != nil {
utils.Logger().Fatal(msg, zap.Error(err))
}
}
2021-11-17 16:19:42 +00:00
func freePort() (int, error) {
addr, err := net.ResolveTCPAddr("tcp", "localhost:0")
if err != nil {
return 0, err
}
l, err := net.ListenTCP("tcp", addr)
if err != nil {
return 0, err
}
port := l.Addr().(*net.TCPAddr).Port
err = l.Close()
if err != nil {
return 0, err
}
return port, nil
}
const dialTimeout = 7 * time.Second
2021-10-09 18:18:53 +00:00
// Execute starts a go-waku node with settings determined by the Options parameter
func Execute(options Options) {
if options.GenerateKey {
2022-07-25 12:24:42 +00:00
if err := writePrivateKeyToFile(options.KeyFile, []byte(options.KeyPasswd), options.Overwrite); err != nil {
failOnErr(err, "nodekey error")
}
return
}
2021-10-15 02:15:02 +00:00
hostAddr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", options.Address, options.Port))
failOnErr(err, "invalid host address")
prvKey, err := getPrivKey(options)
failOnErr(err, "nodekey error")
2022-05-27 19:55:35 +00:00
p2pPrvKey := utils.EcdsaPrivKeyToSecp256k1PrivKey(prvKey)
id, err := peer.IDFromPublicKey(p2pPrvKey.GetPublic())
2022-05-27 13:25:06 +00:00
failOnErr(err, "deriving peer ID from private key")
logger := utils.Logger().With(logging.HostID("node", id))
if options.DBPath == "" && options.UseDB {
failOnErr(errors.New("dbpath can't be null"), "")
}
var db *sql.DB
if options.UseDB {
db, err = sqlite.NewDB(options.DBPath)
failOnErr(err, "Could not connect to DB")
logger.Debug("using database: ", zap.String("path", options.DBPath))
} else {
db, err = sqlite.NewDB(":memory:")
failOnErr(err, "Could not create in-memory DB")
logger.Debug("using in-memory database")
}
2021-04-22 13:07:22 +00:00
ctx := context.Background()
2021-04-19 00:03:16 +00:00
var metricsServer *metrics.Server
if options.Metrics.Enable {
metricsServer = metrics.NewMetricsServer(options.Metrics.Address, options.Metrics.Port, logger)
go metricsServer.Start()
}
2021-04-19 00:03:16 +00:00
nodeOpts := []node.WakuNodeOption{
2022-05-27 13:25:06 +00:00
node.WithLogger(logger),
node.WithPrivateKey(prvKey),
2021-11-17 16:19:42 +00:00
node.WithHostAddress(hostAddr),
node.WithKeepAlive(options.KeepAlive),
}
2021-04-19 00:03:16 +00:00
2021-10-15 02:15:02 +00:00
if options.AdvertiseAddress != "" {
advertiseAddr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%d", options.AdvertiseAddress, options.Port))
2021-11-17 16:19:42 +00:00
failOnErr(err, "Invalid advertise address")
if advertiseAddr.Port == 0 {
for {
p, err := freePort()
if err == nil {
advertiseAddr.Port = p
break
}
}
}
2022-03-22 00:48:46 +00:00
nodeOpts = append(nodeOpts, node.WithAdvertiseAddress(advertiseAddr))
}
if options.Dns4DomainName != "" {
nodeOpts = append(nodeOpts, node.WithDns4Domain(options.Dns4DomainName))
2021-10-15 02:15:02 +00:00
}
2022-03-22 13:12:58 +00:00
libp2pOpts := node.DefaultLibP2POptions
if options.AdvertiseAddress == "" {
libp2pOpts = append(libp2pOpts, libp2p.NATPortMap()) // Attempt to open ports using uPNP for NATed hosts.)
}
if options.Websocket.Enable {
2022-08-25 20:36:04 +00:00
nodeOpts = append(nodeOpts, node.WithWebsockets(options.Websocket.Address, options.Websocket.WSPort))
2022-03-22 13:12:58 +00:00
}
if options.Websocket.Secure {
2022-08-25 20:36:04 +00:00
nodeOpts = append(nodeOpts, node.WithSecureWebsockets(options.Websocket.Address, options.Websocket.WSSPort, options.Websocket.CertPath, options.Websocket.KeyPath))
}
2021-04-19 00:03:16 +00:00
if options.ShowAddresses {
2021-10-15 02:15:02 +00:00
printListeningAddresses(ctx, nodeOpts, options)
return
}
2022-06-19 21:47:39 +00:00
if options.Version {
2022-08-09 14:15:16 +00:00
fmt.Printf("version / git commit hash: %s-%s\n", node.Version, node.GitCommit)
2022-06-19 21:47:39 +00:00
return
}
if options.UseDB {
2022-06-19 21:47:39 +00:00
if options.PersistPeers {
// Create persistent peerstore
queries, err := sqlite.NewQueries("peerstore", db)
failOnErr(err, "Peerstore")
2022-06-19 21:47:39 +00:00
datastore := dssql.NewDatastore(db, queries)
opts := pstoreds.DefaultOpts()
peerStore, err := pstoreds.NewPeerstore(ctx, datastore, opts)
failOnErr(err, "Peerstore")
2022-06-19 21:47:39 +00:00
libp2pOpts = append(libp2pOpts, libp2p.Peerstore(peerStore))
}
}
nodeOpts = append(nodeOpts, node.WithLibP2POptions(libp2pOpts...))
2021-10-01 17:49:50 +00:00
if options.Relay.Enable {
var wakurelayopts []pubsub.Option
wakurelayopts = append(wakurelayopts, pubsub.WithPeerExchange(options.Relay.PeerExchange))
nodeOpts = append(nodeOpts, node.WithWakuRelayAndMinPeers(options.Relay.MinRelayPeersToPublish, wakurelayopts...))
}
2021-06-10 12:59:51 +00:00
if options.RendezvousServer.Enable {
db, err := persistence.NewRendezVousLevelDB(options.RendezvousServer.DBPath)
failOnErr(err, "RendezvousDB")
storage := rendezvous.NewStorage(db)
nodeOpts = append(nodeOpts, node.WithRendezvousServer(storage))
}
if options.Filter.Enable {
nodeOpts = append(nodeOpts, node.WithWakuFilter(!options.Filter.DisableFullNode, filter.WithTimeout(options.Filter.Timeout)))
}
2021-06-28 14:14:28 +00:00
if options.Store.Enable {
2022-06-13 18:30:35 +00:00
if options.Store.PersistMessages {
nodeOpts = append(nodeOpts, node.WithWakuStore(true, options.Store.ShouldResume))
dbStore, err := persistence.NewDBStore(logger, persistence.WithDB(db), persistence.WithRetentionPolicy(options.Store.RetentionMaxMessages, options.Store.RetentionTime))
2022-06-13 18:30:35 +00:00
failOnErr(err, "DBStore")
nodeOpts = append(nodeOpts, node.WithMessageProvider(dbStore))
} else {
nodeOpts = append(nodeOpts, node.WithWakuStore(false, false))
}
}
if options.LightPush.Enable {
nodeOpts = append(nodeOpts, node.WithLightPush())
}
2021-04-12 17:59:41 +00:00
if options.Rendezvous.Enable {
2021-11-16 14:22:01 +00:00
nodeOpts = append(nodeOpts, node.WithRendezvous(pubsub.WithDiscoveryOpts(discovery.Limit(45), discovery.TTL(time.Duration(20)*time.Second))))
}
2022-05-05 19:17:35 +00:00
var discoveredNodes []dnsdisc.DiscoveredNode
if options.DNSDiscovery.Enable {
if len(options.DNSDiscovery.URLs.Value()) != 0 {
for _, url := range options.DNSDiscovery.URLs.Value() {
logger.Info("attempting DNS discovery with ", zap.String("URL", url))
nodes, err := dnsdisc.RetrieveNodes(ctx, url, dnsdisc.WithNameserver(options.DNSDiscovery.Nameserver))
if err != nil {
logger.Warn("dns discovery error ", zap.Error(err))
} else {
var discAddresses []multiaddr.Multiaddr
for _, n := range nodes {
discAddresses = append(discAddresses, n.Addresses...)
}
logger.Info("found dns entries ", logging.MultiAddrs("nodes", discAddresses...))
discoveredNodes = nodes
}
2022-05-05 19:17:35 +00:00
}
} else {
2022-05-27 13:25:06 +00:00
logger.Fatal("DNS discovery URL is required")
2022-05-05 19:17:35 +00:00
}
}
2021-11-17 16:19:42 +00:00
if options.DiscV5.Enable {
var bootnodes []*enode.Node
for _, addr := range options.DiscV5.Nodes.Value() {
2021-11-17 16:19:42 +00:00
bootnode, err := enode.Parse(enode.ValidSchemes, addr)
if err != nil {
2022-05-27 13:25:06 +00:00
logger.Fatal("parsing ENR", zap.Error(err))
2021-11-17 16:19:42 +00:00
}
bootnodes = append(bootnodes, bootnode)
}
2022-05-05 19:17:35 +00:00
for _, n := range discoveredNodes {
if n.ENR != nil {
bootnodes = append(bootnodes, n.ENR)
}
}
2021-11-17 16:19:42 +00:00
nodeOpts = append(nodeOpts, node.WithDiscoveryV5(options.DiscV5.Port, bootnodes, options.DiscV5.AutoUpdate, pubsub.WithDiscoveryOpts(discovery.Limit(45), discovery.TTL(time.Duration(20)*time.Second))))
}
2022-09-11 21:08:58 +00:00
checkForRLN(logger, options, &nodeOpts)
2022-07-05 21:28:34 +00:00
wakuNode, err := node.New(ctx, nodeOpts...)
2021-04-15 21:23:07 +00:00
failOnErr(err, "Wakunode")
addPeers(wakuNode, options.Rendezvous.Nodes, string(rendezvous.RendezvousID_v001))
addPeers(wakuNode, options.Store.Nodes, string(store.StoreID_v20beta4))
addPeers(wakuNode, options.LightPush.Nodes, string(lightpush.LightPushID_v20beta1))
addPeers(wakuNode, options.Filter.Nodes, string(filter.FilterID_v20beta1))
2021-10-05 02:13:54 +00:00
2021-10-06 15:42:57 +00:00
if err = wakuNode.Start(); err != nil {
2022-05-27 13:25:06 +00:00
logger.Fatal("starting waku node", zap.Error(err))
2021-10-06 15:42:57 +00:00
}
2021-10-05 02:13:54 +00:00
2021-11-17 16:19:42 +00:00
if options.DiscV5.Enable {
if err = wakuNode.DiscV5().Start(); err != nil {
2022-05-27 13:25:06 +00:00
logger.Fatal("starting discovery v5", zap.Error(err))
2021-11-17 16:19:42 +00:00
}
}
if len(options.Relay.Topics.Value()) == 0 {
options.Relay.Topics = *cli.NewStringSlice(relay.DefaultWakuTopic)
}
if options.Relay.Enable {
for _, nodeTopic := range options.Relay.Topics.Value() {
nodeTopic := nodeTopic
sub, err := wakuNode.Relay().SubscribeToTopic(ctx, nodeTopic)
failOnErr(err, "Error subscring to topic")
wakuNode.Broadcaster().Unregister(&nodeTopic, sub.C)
}
}
for _, n := range options.StaticNodes {
go func(ctx context.Context, node multiaddr.Multiaddr) {
ctx, cancel := context.WithTimeout(ctx, dialTimeout)
defer cancel()
err = wakuNode.DialPeerWithMultiAddress(ctx, node)
if err != nil {
2022-05-27 13:25:06 +00:00
logger.Error("dialing peer", zap.Error(err))
}
}(ctx, n)
}
2022-05-05 19:17:35 +00:00
if len(discoveredNodes) != 0 {
for _, n := range discoveredNodes {
for _, m := range n.Addresses {
go func(ctx context.Context, m multiaddr.Multiaddr) {
ctx, cancel := context.WithTimeout(ctx, dialTimeout)
2022-05-05 19:17:35 +00:00
defer cancel()
err = wakuNode.DialPeerWithMultiAddress(ctx, m)
if err != nil {
logger.Error("dialing peer", logging.MultiAddrs("peer", m), zap.Error(err))
2022-05-05 19:17:35 +00:00
}
}(ctx, m)
}
}
}
2022-08-12 12:44:13 +00:00
onStartRLN(wakuNode, options)
2022-08-09 00:02:08 +00:00
var rpcServer *rpc.WakuRpc
if options.RPCServer.Enable {
rpcServer = rpc.NewWakuRpc(wakuNode, options.RPCServer.Address, options.RPCServer.Port, options.RPCServer.Admin, options.RPCServer.Private, logger)
rpcServer.Start()
}
var restServer *rest.WakuRest
if options.RESTServer.Enable {
restServer = rest.NewWakuRest(wakuNode, options.RESTServer.Address, options.RESTServer.Port, options.RESTServer.Admin, options.RESTServer.Private, options.RESTServer.RelayCacheCapacity, logger)
restServer.Start()
}
logger.Info("Node setup complete")
// Wait for a SIGINT or SIGTERM signal
ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
<-ch
2022-05-27 13:25:06 +00:00
logger.Info("Received signal, shutting down...")
2021-04-19 00:03:16 +00:00
// shut the node down
wakuNode.Stop()
if options.RPCServer.Enable {
err := rpcServer.Stop(ctx)
failOnErr(err, "RPCClose")
}
if options.RESTServer.Enable {
err := restServer.Stop(ctx)
failOnErr(err, "RESTClose")
}
if options.Metrics.Enable {
2021-10-18 12:43:17 +00:00
err = metricsServer.Stop(ctx)
failOnErr(err, "MetricsClose")
}
if options.UseDB {
err = db.Close()
failOnErr(err, "DBClose")
}
}
func addPeers(wakuNode *node.WakuNode, addresses []multiaddr.Multiaddr, protocols ...string) {
for _, addr := range addresses {
_, err := wakuNode.AddPeer(addr, protocols...)
failOnErr(err, "error adding peer")
}
}
2022-07-25 12:24:42 +00:00
func loadPrivateKeyFromFile(path string, passwd string) (*ecdsa.PrivateKey, error) {
src, err := ioutil.ReadFile(path)
if err != nil {
return nil, err
}
2022-07-25 12:24:42 +00:00
var encryptedK keystore.CryptoJSON
err = json.Unmarshal(src, &encryptedK)
if err != nil {
return nil, err
}
2022-07-25 12:24:42 +00:00
pKey, err := keystore.DecryptDataV3(encryptedK, passwd)
2022-05-27 19:55:35 +00:00
if err != nil {
return nil, err
}
2022-07-25 12:24:42 +00:00
return crypto.ToECDSA(pKey)
}
2022-08-09 00:02:08 +00:00
func checkForFileExistence(path string, overwrite bool) error {
_, err := os.Stat(path)
2021-12-08 14:21:30 +00:00
if err == nil && !overwrite {
return fmt.Errorf("%s already exists. Use --overwrite to overwrite the file", path)
}
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
return err
}
2021-12-08 14:21:30 +00:00
return nil
}
func generatePrivateKey() ([]byte, error) {
key, err := crypto.GenerateKey()
if err != nil {
2021-12-08 14:21:30 +00:00
return nil, err
}
2022-07-25 12:24:42 +00:00
return key.D.Bytes(), nil
2021-12-08 14:21:30 +00:00
}
2022-07-25 12:24:42 +00:00
func writePrivateKeyToFile(path string, passwd []byte, overwrite bool) error {
2022-08-09 00:02:08 +00:00
if err := checkForFileExistence(path, overwrite); err != nil {
2021-12-08 14:21:30 +00:00
return err
}
2022-07-25 12:24:42 +00:00
key, err := generatePrivateKey()
if err != nil {
return err
}
encryptedK, err := keystore.EncryptDataV3(key, passwd, keystore.StandardScryptN, keystore.StandardScryptP)
if err != nil {
return err
}
output, err := json.Marshal(encryptedK)
2021-12-08 14:21:30 +00:00
if err != nil {
return err
}
return ioutil.WriteFile(path, output, 0600)
}
func getPrivKey(options Options) (*ecdsa.PrivateKey, error) {
var prvKey *ecdsa.PrivateKey
var err error
if options.NodeKey != nil {
prvKey = options.NodeKey
} else {
// TODO: once https://github.com/urfave/cli/issues/1272 is fixed, remove env variable logic
keyString := os.Getenv("GOWAKU-NODEKEY")
if keyString != "" {
2022-08-04 21:39:12 +00:00
if prvKey, err = crypto.ToECDSA(common.FromHex(keyString)); err != nil {
return nil, fmt.Errorf("error converting key into valid ecdsa key: %w", err)
}
} else {
if _, err := os.Stat(options.KeyFile); err == nil {
2022-07-25 12:24:42 +00:00
if prvKey, err = loadPrivateKeyFromFile(options.KeyFile, options.KeyPasswd); err != nil {
return nil, fmt.Errorf("could not read keyfile: %w", err)
}
} else {
if os.IsNotExist(err) {
2021-10-05 02:13:54 +00:00
if prvKey, err = crypto.GenerateKey(); err != nil {
return nil, fmt.Errorf("error generating key: %w", err)
}
} else {
return nil, fmt.Errorf("could not read keyfile: %w", err)
}
}
}
}
return prvKey, nil
}
2021-10-15 02:15:02 +00:00
func printListeningAddresses(ctx context.Context, nodeOpts []node.WakuNodeOption, options Options) {
params := new(node.WakuNodeParameters)
for _, opt := range nodeOpts {
err := opt(params)
if err != nil {
panic(err)
}
}
2021-10-15 02:15:02 +00:00
var libp2pOpts []config.Option
2021-11-17 16:19:42 +00:00
libp2pOpts = append(libp2pOpts,
params.Identity(),
libp2p.ListenAddrs(params.MultiAddresses()...),
)
2021-10-15 02:15:02 +00:00
2022-08-25 20:36:04 +00:00
if options.Websocket.Secure {
transports := libp2p.ChainOptions(
libp2p.Transport(tcp.NewTCPTransport),
libp2p.Transport(ws.New, ws.WithTLSConfig(params.TLSConfig())),
)
libp2pOpts = append(libp2pOpts, transports)
}
2021-11-17 16:19:42 +00:00
addrFactory := params.AddressFactory()
if addrFactory != nil {
libp2pOpts = append(libp2pOpts, libp2p.AddrsFactory(addrFactory))
2021-10-15 02:15:02 +00:00
}
2022-03-22 13:12:58 +00:00
h, err := libp2p.New(libp2pOpts...)
if err != nil {
panic(err)
}
hostInfo, _ := multiaddr.NewMultiaddr(fmt.Sprintf("/p2p/%s", h.ID().Pretty()))
2021-10-15 02:15:02 +00:00
for _, addr := range h.Addrs() {
fmt.Println(addr.Encapsulate(hostInfo))
}
}