package node import ( "context" "errors" "fmt" "math/rand" "net" "sync" "time" golog "github.com/ipfs/go-log/v2" "github.com/libp2p/go-libp2p" "go.uber.org/zap" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/libp2p/go-libp2p/core/event" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peerstore" "github.com/libp2p/go-libp2p/core/protocol" "github.com/libp2p/go-libp2p/p2p/discovery/backoff" ws "github.com/libp2p/go-libp2p/p2p/transport/websocket" ma "github.com/multiformats/go-multiaddr" "go.opencensus.io/stats" "github.com/waku-org/go-waku/logging" "github.com/waku-org/go-waku/waku/try" v2 "github.com/waku-org/go-waku/waku/v2" "github.com/waku-org/go-waku/waku/v2/discv5" "github.com/waku-org/go-waku/waku/v2/metrics" "github.com/waku-org/go-waku/waku/v2/protocol/enr" "github.com/waku-org/go-waku/waku/v2/protocol/filter" "github.com/waku-org/go-waku/waku/v2/protocol/legacy_filter" "github.com/waku-org/go-waku/waku/v2/protocol/lightpush" "github.com/waku-org/go-waku/waku/v2/protocol/pb" "github.com/waku-org/go-waku/waku/v2/protocol/peer_exchange" "github.com/waku-org/go-waku/waku/v2/protocol/relay" "github.com/waku-org/go-waku/waku/v2/protocol/store" "github.com/waku-org/go-waku/waku/v2/rendezvous" "github.com/waku-org/go-waku/waku/v2/timesource" "github.com/waku-org/go-waku/waku/v2/utils" ) type Peer struct { ID peer.ID `json:"peerID"` Protocols []protocol.ID `json:"protocols"` Addrs []ma.Multiaddr `json:"addrs"` Connected bool `json:"connected"` } type storeFactory func(w *WakuNode) store.Store type byte32 = [32]byte type IdentityCredential = struct { IDTrapdoor byte32 `json:"idTrapdoor"` IDNullifier byte32 `json:"idNullifier"` IDSecretHash byte32 `json:"idSecretHash"` IDCommitment byte32 `json:"idCommitment"` } type RLNRelay interface { IdentityCredential() (IdentityCredential, error) MembershipIndex() (uint, error) AppendRLNProof(msg *pb.WakuMessage, senderEpochTime time.Time) error Stop() } type WakuNode struct { host host.Host opts *WakuNodeParameters log *zap.Logger timesource timesource.Timesource relay Service lightPush Service peerConnector PeerConnectorService discoveryV5 Service peerExchange Service rendezvous Service legacyFilter ReceptorService filterFullnode ReceptorService filterLightnode Service store ReceptorService rlnRelay RLNRelay wakuFlag enr.WakuEnrBitfield localNode *enode.LocalNode bcaster relay.Broadcaster connectionNotif ConnectionNotifier protocolEventSub event.Subscription identificationEventSub event.Subscription addressChangesSub event.Subscription enrChangeCh chan struct{} keepAliveMutex sync.Mutex keepAliveFails map[peer.ID]int cancel context.CancelFunc wg *sync.WaitGroup // Channel passed to WakuNode constructor // receiving connection status notifications connStatusChan chan ConnStatus storeFactory storeFactory } func defaultStoreFactory(w *WakuNode) store.Store { return store.NewWakuStore(w.opts.messageProvider, w.timesource, w.log) } // New is used to instantiate a WakuNode using a set of WakuNodeOptions func New(opts ...WakuNodeOption) (*WakuNode, error) { params := new(WakuNodeParameters) params.libP2POpts = DefaultLibP2POptions opts = append(DefaultWakuNodeOptions, opts...) for _, opt := range opts { err := opt(params) if err != nil { return nil, err } } if params.logger == nil { params.logger = utils.Logger() golog.SetAllLoggers(params.logLevel) } if params.privKey == nil { prvKey, err := crypto.GenerateKey() if err != nil { return nil, err } params.privKey = prvKey } if params.enableWSS { params.libP2POpts = append(params.libP2POpts, libp2p.Transport(ws.New, ws.WithTLSConfig(params.tlsConfig))) } else { // Enable WS transport by default params.libP2POpts = append(params.libP2POpts, libp2p.Transport(ws.New)) } // Setting default host address if none was provided if params.hostAddr == nil { err := WithHostAddress(&net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: 0})(params) if err != nil { return nil, err } } if len(params.multiAddr) > 0 { params.libP2POpts = append(params.libP2POpts, libp2p.ListenAddrs(params.multiAddr...)) } params.libP2POpts = append(params.libP2POpts, params.Identity()) if params.addressFactory != nil { params.libP2POpts = append(params.libP2POpts, libp2p.AddrsFactory(params.addressFactory)) } var err error w := new(WakuNode) w.bcaster = relay.NewBroadcaster(1024) w.opts = params w.log = params.logger.Named("node2") w.wg = &sync.WaitGroup{} w.keepAliveFails = make(map[peer.ID]int) w.wakuFlag = enr.NewWakuEnrBitfield(w.opts.enableLightPush, w.opts.enableLegacyFilter, w.opts.enableStore, w.opts.enableRelay) if params.enableNTP { w.timesource = timesource.NewNTPTimesource(w.opts.ntpURLs, w.log) } else { w.timesource = timesource.NewDefaultClock() } w.localNode, err = w.newLocalnode(w.opts.privKey) if err != nil { w.log.Error("creating localnode", zap.Error(err)) } // Setup peer connection strategy cacheSize := 600 rngSrc := rand.NewSource(rand.Int63()) minBackoff, maxBackoff := time.Second*30, time.Hour bkf := backoff.NewExponentialBackoff(minBackoff, maxBackoff, backoff.FullJitter, time.Second, 5.0, 0, rand.New(rngSrc)) w.peerConnector, err = v2.NewPeerConnectionStrategy(cacheSize, w.opts.discoveryMinPeers, network.DialPeerTimeout, bkf, w.log) if err != nil { w.log.Error("creating peer connection strategy", zap.Error(err)) } if w.opts.enableDiscV5 { err := w.mountDiscV5() if err != nil { return nil, err } } w.peerExchange, err = peer_exchange.NewWakuPeerExchange(w.DiscV5(), w.peerConnector, w.log) if err != nil { return nil, err } var rendezvousPoints []peer.ID for _, p := range w.opts.rendezvousNodes { peerID, err := utils.GetPeerID(p) if err != nil { return nil, err } rendezvousPoints = append(rendezvousPoints, peerID) } w.rendezvous = rendezvous.NewRendezvous(w.opts.enableRendezvousServer, w.opts.rendezvousDB, w.opts.enableRendezvous, rendezvousPoints, w.peerConnector, w.log) w.relay = relay.NewWakuRelay(w.bcaster, w.opts.minRelayPeersToPublish, w.timesource, w.log, w.opts.wOpts...) w.legacyFilter = legacy_filter.NewWakuFilter(w.bcaster, w.opts.isLegacyFilterFullnode, w.timesource, w.log, w.opts.legacyFilterOpts...) w.filterFullnode = filter.NewWakuFilterFullnode(w.timesource, w.log, w.opts.filterOpts...) w.filterLightnode = filter.NewWakuFilterLightnode(w.bcaster, w.timesource, w.log) w.lightPush = lightpush.NewWakuLightPush(w.Relay(), w.log) if params.storeFactory != nil { w.storeFactory = params.storeFactory } else { w.storeFactory = defaultStoreFactory } if params.connStatusC != nil { w.connStatusChan = params.connStatusC } return w, nil } func (w *WakuNode) watchMultiaddressChanges(ctx context.Context) { defer w.wg.Done() addrs := w.ListenAddresses() first := make(chan struct{}, 1) first <- struct{}{} for { select { case <-ctx.Done(): return case <-first: w.log.Info("listening", logging.MultiAddrs("multiaddr", addrs...)) w.enrChangeCh <- struct{}{} case <-w.addressChangesSub.Out(): newAddrs := w.ListenAddresses() diff := false if len(addrs) != len(newAddrs) { diff = true } else { for i := range newAddrs { if addrs[i].String() != newAddrs[i].String() { diff = true break } } } if diff { addrs = newAddrs w.log.Info("listening addresses update received", logging.MultiAddrs("multiaddr", addrs...)) _ = w.setupENR(ctx, addrs) w.enrChangeCh <- struct{}{} } } } } // Start initializes all the protocols that were setup in the WakuNode func (w *WakuNode) Start(ctx context.Context) error { ctx, cancel := context.WithCancel(ctx) w.cancel = cancel host, err := libp2p.New(w.opts.libP2POpts...) if err != nil { return err } w.host = host if w.protocolEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerProtocolsUpdated)); err != nil { return err } if w.identificationEventSub, err = host.EventBus().Subscribe(new(event.EvtPeerIdentificationCompleted)); err != nil { return err } if w.addressChangesSub, err = host.EventBus().Subscribe(new(event.EvtLocalAddressesUpdated)); err != nil { return err } w.connectionNotif = NewConnectionNotifier(ctx, w.host, w.log) w.host.Network().Notify(w.connectionNotif) w.enrChangeCh = make(chan struct{}, 10) w.wg.Add(3) go w.connectednessListener(ctx) go w.watchMultiaddressChanges(ctx) go w.watchENRChanges(ctx) err = w.bcaster.Start(ctx) if err != nil { return err } if w.opts.keepAliveInterval > time.Duration(0) { w.wg.Add(1) go w.startKeepAlive(ctx, w.opts.keepAliveInterval) } w.peerConnector.SetHost(host) err = w.peerConnector.Start(ctx) if err != nil { return err } if w.opts.enableNTP { err := w.timesource.Start(ctx) if err != nil { return err } } w.relay.SetHost(host) if w.opts.enableRelay { err := w.relay.Start(ctx) if err != nil { return err } if !w.opts.noDefaultWakuTopic { sub, err := w.Relay().Subscribe(ctx) if err != nil { return err } sub.Unsubscribe() } } w.store = w.storeFactory(w) w.store.SetHost(host) if w.opts.enableStore { sub := w.bcaster.RegisterForAll() err := w.startStore(ctx, sub) if err != nil { return err } w.log.Info("Subscribing store to broadcaster") } w.lightPush.SetHost(host) if w.opts.enableLightPush { if err := w.lightPush.Start(ctx); err != nil { return err } } w.legacyFilter.SetHost(host) if w.opts.enableLegacyFilter { sub := w.bcaster.RegisterForAll() err := w.legacyFilter.Start(ctx, sub) if err != nil { return err } w.log.Info("Subscribing filter to broadcaster") } w.filterFullnode.SetHost(host) if w.opts.enableFilterFullNode { sub := w.bcaster.RegisterForAll() err := w.filterFullnode.Start(ctx, sub) if err != nil { return err } w.log.Info("Subscribing filterV2 to broadcaster") } w.filterLightnode.SetHost(host) if w.opts.enableFilterLightNode { err := w.filterLightnode.Start(ctx) if err != nil { return err } } err = w.setupENR(ctx, w.ListenAddresses()) if err != nil { return err } w.peerExchange.SetHost(host) if w.opts.enablePeerExchange { err := w.peerExchange.Start(ctx) if err != nil { return err } } w.rendezvous.SetHost(host) if w.opts.enableRendezvousServer || w.opts.enableRendezvous { err := w.rendezvous.Start(ctx) if err != nil { return err } } if w.opts.enableRLN { err = w.mountRlnRelay(ctx) if err != nil { return err } } return nil } // Stop stops the WakuNode and closess all connections to the host func (w *WakuNode) Stop() { if w.cancel == nil { return } w.cancel() w.bcaster.Stop() defer w.connectionNotif.Close() defer w.protocolEventSub.Close() defer w.identificationEventSub.Close() defer w.addressChangesSub.Close() if w.opts.enableRendezvousServer || w.opts.enableRendezvous { w.rendezvous.Stop() } w.relay.Stop() w.lightPush.Stop() w.store.Stop() w.legacyFilter.Stop() w.filterFullnode.Stop() w.peerExchange.Stop() if w.opts.enableDiscV5 { w.discoveryV5.Stop() } w.peerConnector.Stop() _ = w.stopRlnRelay() w.timesource.Stop() w.host.Close() w.wg.Wait() close(w.enrChangeCh) w.cancel = nil } // Host returns the libp2p Host used by the WakuNode func (w *WakuNode) Host() host.Host { return w.host } // ID returns the base58 encoded ID from the host func (w *WakuNode) ID() string { return w.host.ID().Pretty() } func (w *WakuNode) watchENRChanges(ctx context.Context) { defer w.wg.Done() var prevNodeVal string for { select { case <-ctx.Done(): return case <-w.enrChangeCh: if w.localNode != nil { currNodeVal := w.localNode.Node().String() if prevNodeVal != currNodeVal { if prevNodeVal == "" { w.log.Info("enr record", logging.ENode("enr", w.localNode.Node())) } else { w.log.Info("new enr record", logging.ENode("enr", w.localNode.Node())) } prevNodeVal = currNodeVal } } } } } // ListenAddresses returns all the multiaddresses used by the host func (w *WakuNode) ListenAddresses() []ma.Multiaddr { hostInfo, _ := ma.NewMultiaddr(fmt.Sprintf("/p2p/%s", w.host.ID().Pretty())) var result []ma.Multiaddr for _, addr := range w.host.Addrs() { result = append(result, addr.Encapsulate(hostInfo)) } return result } // ENR returns the ENR address of the node func (w *WakuNode) ENR() *enode.Node { return w.localNode.Node() } // Timesource returns the timesource used by this node to obtain the current wall time // Depending on the configuration it will be the local time or a ntp syncd time func (w *WakuNode) Timesource() timesource.Timesource { return w.timesource } // Relay is used to access any operation related to Waku Relay protocol func (w *WakuNode) Relay() *relay.WakuRelay { if result, ok := w.relay.(*relay.WakuRelay); ok { return result } return nil } // Store is used to access any operation related to Waku Store protocol func (w *WakuNode) Store() store.Store { return w.store.(store.Store) } // LegacyFilter is used to access any operation related to Waku LegacyFilter protocol func (w *WakuNode) LegacyFilter() *legacy_filter.WakuFilter { if result, ok := w.legacyFilter.(*legacy_filter.WakuFilter); ok { return result } return nil } // FilterLightnode is used to access any operation related to Waku Filter protocol Full node feature func (w *WakuNode) FilterFullnode() *filter.WakuFilterFullNode { if result, ok := w.filterFullnode.(*filter.WakuFilterFullNode); ok { return result } return nil } // FilterFullnode is used to access any operation related to Waku Filter protocol Light node feature func (w *WakuNode) FilterLightnode() *filter.WakuFilterLightnode { if result, ok := w.filterLightnode.(*filter.WakuFilterLightnode); ok { return result } return nil } // Lightpush is used to access any operation related to Waku Lightpush protocol func (w *WakuNode) Lightpush() *lightpush.WakuLightPush { if result, ok := w.lightPush.(*lightpush.WakuLightPush); ok { return result } return nil } // DiscV5 is used to access any operation related to DiscoveryV5 func (w *WakuNode) DiscV5() *discv5.DiscoveryV5 { if result, ok := w.discoveryV5.(*discv5.DiscoveryV5); ok { return result } return nil } // PeerExchange is used to access any operation related to Peer Exchange func (w *WakuNode) PeerExchange() *peer_exchange.WakuPeerExchange { if result, ok := w.peerExchange.(*peer_exchange.WakuPeerExchange); ok { return result } return nil } // Broadcaster is used to access the message broadcaster that is used to push // messages to different protocols func (w *WakuNode) Broadcaster() relay.Broadcaster { return w.bcaster } // Publish will attempt to publish a message via WakuRelay if there are enough // peers available, otherwise it will attempt to publish via Lightpush protocol func (w *WakuNode) Publish(ctx context.Context, msg *pb.WakuMessage) error { if !w.opts.enableLightPush && !w.opts.enableRelay { return errors.New("cannot publish message, relay and lightpush are disabled") } hash := msg.Hash(relay.DefaultWakuTopic) err := try.Do(func(attempt int) (bool, error) { var err error relay := w.Relay() lightpush := w.Lightpush() if relay == nil || !relay.EnoughPeersToPublish() { w.log.Debug("publishing message via lightpush", logging.HexBytes("hash", hash)) _, err = lightpush.Publish(ctx, msg) } else { w.log.Debug("publishing message via relay", logging.HexBytes("hash", hash)) _, err = relay.Publish(ctx, msg) } return attempt < maxPublishAttempt, err }) return err } func (w *WakuNode) mountDiscV5() error { discV5Options := []discv5.DiscoveryV5Option{ discv5.WithBootnodes(w.opts.discV5bootnodes), discv5.WithUDPPort(w.opts.udpPort), discv5.WithAutoUpdate(w.opts.discV5autoUpdate), } if w.opts.advertiseAddrs != nil { discV5Options = append(discV5Options, discv5.WithAdvertiseAddr(w.opts.advertiseAddrs)) } var err error w.discoveryV5, err = discv5.NewDiscoveryV5(w.opts.privKey, w.localNode, w.peerConnector, w.log, discV5Options...) return err } func (w *WakuNode) startStore(ctx context.Context, sub relay.Subscription) error { err := w.store.Start(ctx, sub) if err != nil { w.log.Error("starting store", zap.Error(err)) return err } if len(w.opts.resumeNodes) != 0 { // TODO: extract this to a function and run it when you go offline // TODO: determine if a store is listening to a topic var peerIDs []peer.ID for _, n := range w.opts.resumeNodes { pID, err := w.AddPeer(n, store.StoreID_v20beta4) if err != nil { w.log.Warn("adding peer to peerstore", logging.MultiAddrs("peer", n), zap.Error(err)) } peerIDs = append(peerIDs, pID) } if !w.opts.noDefaultWakuTopic { w.wg.Add(1) go func() { defer w.wg.Done() ctxWithTimeout, ctxCancel := context.WithTimeout(ctx, 20*time.Second) defer ctxCancel() if _, err := w.store.(store.Store).Resume(ctxWithTimeout, string(relay.DefaultWakuTopic), peerIDs); err != nil { w.log.Error("Could not resume history", zap.Error(err)) time.Sleep(10 * time.Second) } }() } } return nil } func (w *WakuNode) addPeer(info *peer.AddrInfo, protocols ...protocol.ID) error { w.log.Info("adding peer to peerstore", logging.HostID("peer", info.ID)) w.host.Peerstore().AddAddrs(info.ID, info.Addrs, peerstore.PermanentAddrTTL) err := w.host.Peerstore().AddProtocols(info.ID, protocols...) if err != nil { return err } return nil } // AddPeer is used to add a peer and the protocols it support to the node peerstore func (w *WakuNode) AddPeer(address ma.Multiaddr, protocols ...protocol.ID) (peer.ID, error) { info, err := peer.AddrInfoFromP2pAddr(address) if err != nil { return "", err } return info.ID, w.addPeer(info, protocols...) } // DialPeerWithMultiAddress is used to connect to a peer using a multiaddress func (w *WakuNode) DialPeerWithMultiAddress(ctx context.Context, address ma.Multiaddr) error { info, err := peer.AddrInfoFromP2pAddr(address) if err != nil { return err } return w.connect(ctx, *info) } // DialPeer is used to connect to a peer using a string containing a multiaddress func (w *WakuNode) DialPeer(ctx context.Context, address string) error { p, err := ma.NewMultiaddr(address) if err != nil { return err } info, err := peer.AddrInfoFromP2pAddr(p) if err != nil { return err } return w.connect(ctx, *info) } func (w *WakuNode) connect(ctx context.Context, info peer.AddrInfo) error { err := w.host.Connect(ctx, info) if err != nil { return err } stats.Record(ctx, metrics.Dials.M(1)) return nil } // DialPeerByID is used to connect to an already known peer func (w *WakuNode) DialPeerByID(ctx context.Context, peerID peer.ID) error { info := w.host.Peerstore().PeerInfo(peerID) return w.connect(ctx, info) } // ClosePeerByAddress is used to disconnect from a peer using its multiaddress func (w *WakuNode) ClosePeerByAddress(address string) error { p, err := ma.NewMultiaddr(address) if err != nil { return err } // Extract the peer ID from the multiaddr. info, err := peer.AddrInfoFromP2pAddr(p) if err != nil { return err } return w.ClosePeerById(info.ID) } // ClosePeerById is used to close a connection to a peer func (w *WakuNode) ClosePeerById(id peer.ID) error { err := w.host.Network().ClosePeer(id) if err != nil { return err } return nil } // PeerCount return the number of connected peers func (w *WakuNode) PeerCount() int { return len(w.host.Network().Peers()) } // PeerStats returns a list of peers and the protocols supported by them func (w *WakuNode) PeerStats() PeerStats { p := make(PeerStats) for _, peerID := range w.host.Network().Peers() { protocols, err := w.host.Peerstore().GetProtocols(peerID) if err != nil { continue } p[peerID] = protocols } return p } // Set the bootnodes on discv5 func (w *WakuNode) SetDiscV5Bootnodes(nodes []*enode.Node) error { w.opts.discV5bootnodes = nodes return w.DiscV5().SetBootnodes(nodes) } // Peers return the list of peers, addresses, protocols supported and connection status func (w *WakuNode) Peers() ([]*Peer, error) { var peers []*Peer for _, peerId := range w.host.Peerstore().Peers() { connected := w.host.Network().Connectedness(peerId) == network.Connected protocols, err := w.host.Peerstore().GetProtocols(peerId) if err != nil { return nil, err } addrs := w.host.Peerstore().Addrs(peerId) peers = append(peers, &Peer{ ID: peerId, Protocols: protocols, Connected: connected, Addrs: addrs, }) } return peers, nil }