package p2p import ( "bytes" "context" "encoding/hex" "fmt" "sync" "time" "github.com/libp2p/go-libp2p" dht "github.com/libp2p/go-libp2p-kad-dht" libp2pconfig "github.com/libp2p/go-libp2p/config" "github.com/libp2p/go-libp2p/core/crypto" "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/p2p/discovery/routing" "github.com/libp2p/go-libp2p/p2p/discovery/util" "github.com/libp2p/go-libp2p/p2p/net/connmgr" "github.com/pkg/errors" "go.uber.org/zap" blossomsub "source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub" "source.quilibrium.com/quilibrium/monorepo/go-libp2p-blossomsub/pb" "source.quilibrium.com/quilibrium/monorepo/node/config" "source.quilibrium.com/quilibrium/monorepo/node/protobufs" ) type BlossomSub struct { ps *blossomsub.PubSub ctx context.Context logger *zap.Logger peerID peer.ID bitmaskMap map[string]*blossomsub.Bitmask h host.Host signKey crypto.PrivKey peerScore map[string]int64 peerScoreMx sync.Mutex isBootstrapPeer bool network uint8 } var _ PubSub = (*BlossomSub)(nil) var ErrNoPeersAvailable = errors.New("no peers available") var BITMASK_ALL = []byte{ 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, } var ANNOUNCE_PREFIX = "quilibrium-2.0.0-dusk-" func getPeerID(p2pConfig *config.P2PConfig) peer.ID { peerPrivKey, err := hex.DecodeString(p2pConfig.PeerPrivKey) if err != nil { panic(errors.Wrap(err, "error unmarshaling peerkey")) } privKey, err := crypto.UnmarshalEd448PrivateKey(peerPrivKey) if err != nil { panic(errors.Wrap(err, "error unmarshaling peerkey")) } pub := privKey.GetPublic() id, err := peer.IDFromPublicKey(pub) if err != nil { panic(errors.Wrap(err, "error getting peer id")) } return id } func NewBlossomSub( p2pConfig *config.P2PConfig, logger *zap.Logger, ) *BlossomSub { ctx := context.Background() opts := []libp2pconfig.Option{ libp2p.ListenAddrStrings(p2pConfig.ListenMultiaddr), } isBootstrapPeer := false peerId := getPeerID(p2pConfig) if p2pConfig.Network == 0 { for _, peerAddr := range config.BootstrapPeers { peerinfo, err := peer.AddrInfoFromString(peerAddr) if err != nil { panic(err) } if bytes.Equal([]byte(peerinfo.ID), []byte(peerId)) { isBootstrapPeer = true break } } } else { for _, peerAddr := range p2pConfig.BootstrapPeers { peerinfo, err := peer.AddrInfoFromString(peerAddr) if err != nil { panic(err) } if bytes.Equal([]byte(peerinfo.ID), []byte(peerId)) { isBootstrapPeer = true break } } } var privKey crypto.PrivKey if p2pConfig.PeerPrivKey != "" { peerPrivKey, err := hex.DecodeString(p2pConfig.PeerPrivKey) if err != nil { panic(errors.Wrap(err, "error unmarshaling peerkey")) } privKey, err = crypto.UnmarshalEd448PrivateKey(peerPrivKey) if err != nil { panic(errors.Wrap(err, "error unmarshaling peerkey")) } opts = append(opts, libp2p.Identity(privKey)) } if p2pConfig.LowWatermarkConnections != 0 && p2pConfig.HighWatermarkConnections != 0 { cm, err := connmgr.NewConnManager( int(p2pConfig.LowWatermarkConnections), int(p2pConfig.HighWatermarkConnections), connmgr.WithEmergencyTrim(true), ) if err != nil { panic(err) } opts = append(opts, libp2p.ConnectionManager(cm)) } bs := &BlossomSub{ ctx: ctx, logger: logger, bitmaskMap: make(map[string]*blossomsub.Bitmask), signKey: privKey, peerScore: make(map[string]int64), isBootstrapPeer: isBootstrapPeer, network: p2pConfig.Network, } h, err := libp2p.New(opts...) if err != nil { panic(errors.Wrap(err, "error constructing p2p")) } logger.Info("established peer id", zap.String("peer_id", h.ID().String())) kademliaDHT := initDHT(ctx, p2pConfig, logger, h, isBootstrapPeer) routingDiscovery := routing.NewRoutingDiscovery(kademliaDHT) util.Advertise(ctx, routingDiscovery, getNetworkNamespace(p2pConfig.Network)) var tracer *blossomsub.JSONTracer if p2pConfig.TraceLogFile == "" { tracer, err = blossomsub.NewStdoutJSONTracer() if err != nil { panic(errors.Wrap(err, "error building stdout tracer")) } } else { tracer, err = blossomsub.NewJSONTracer(p2pConfig.TraceLogFile) if err != nil { panic(errors.Wrap(err, "error building file tracer")) } } blossomOpts := []blossomsub.Option{} if tracer != nil { blossomOpts = append(blossomOpts, blossomsub.WithEventTracer(tracer)) } blossomOpts = append(blossomOpts, blossomsub.WithPeerScore( &blossomsub.PeerScoreParams{ SkipAtomicValidation: false, BitmaskScoreCap: 0, IPColocationFactorWeight: 0, IPColocationFactorThreshold: 6, BehaviourPenaltyWeight: 0, BehaviourPenaltyThreshold: 100, BehaviourPenaltyDecay: .5, DecayInterval: 10 * time.Second, DecayToZero: .1, RetainScore: 60 * time.Minute, AppSpecificScore: func(p peer.ID) float64 { return float64(bs.GetPeerScore([]byte(p))) }, AppSpecificWeight: 10.0, }, &blossomsub.PeerScoreThresholds{ SkipAtomicValidation: false, GossipThreshold: -2000, PublishThreshold: -5000, GraylistThreshold: -10000, AcceptPXThreshold: 1, OpportunisticGraftThreshold: 2, })) params := mergeDefaults(p2pConfig) rt := blossomsub.NewBlossomSubRouter(h, params) pubsub, err := blossomsub.NewBlossomSubWithRouter(ctx, h, rt, blossomOpts...) if err != nil { panic(err) } peerID := h.ID() bs.ps = pubsub bs.peerID = peerID bs.h = h bs.signKey = privKey return bs } func (b *BlossomSub) PublishToBitmask(bitmask []byte, data []byte) error { return b.ps.Publish(b.ctx, bitmask, data) } func (b *BlossomSub) Publish(address []byte, data []byte) error { bitmask := GetBloomFilter(address, 256, 3) return b.PublishToBitmask(bitmask, data) } func (b *BlossomSub) Subscribe( bitmask []byte, handler func(message *pb.Message) error, ) error { b.logger.Info("joining broadcast") bm, err := b.ps.Join(bitmask) if err != nil { b.logger.Error("join failed", zap.Error(err)) return errors.Wrap(err, "subscribe") } b.logger.Info("subscribe to bitmask", zap.Binary("bitmask", bitmask)) subs := []*blossomsub.Subscription{} for _, bit := range bm { sub, err := bit.Subscribe() if err != nil { b.logger.Error("subscription failed", zap.Error(err)) return errors.Wrap(err, "subscribe") } subs = append(subs, sub) } b.logger.Info( "begin streaming from bitmask", zap.Binary("bitmask", bitmask), ) for _, sub := range subs { copiedBitmask := make([]byte, len(bitmask)) copy(copiedBitmask[:], bitmask[:]) sub := sub go func() { for { m, err := sub.Next(b.ctx) if err != nil { b.logger.Error( "got error when fetching the next message", zap.Error(err), ) } if err = handler(m.Message); err != nil { b.logger.Debug("message handler returned error", zap.Error(err)) } } }() } return nil } func (b *BlossomSub) Unsubscribe(bitmask []byte, raw bool) { networkBitmask := append([]byte{b.network}, bitmask...) bm, ok := b.bitmaskMap[string(networkBitmask)] if !ok { return } bm.Close() } func (b *BlossomSub) GetPeerID() []byte { return []byte(b.peerID) } func initDHT( ctx context.Context, p2pConfig *config.P2PConfig, logger *zap.Logger, h host.Host, isBootstrapPeer bool, ) *dht.IpfsDHT { logger.Info("establishing dht") var kademliaDHT *dht.IpfsDHT var err error defaultBootstrapPeers := append([]string{}, p2pConfig.BootstrapPeers...) if p2pConfig.Network == 0 { defaultBootstrapPeers = config.BootstrapPeers } bootstrappers := []peer.AddrInfo{} for _, peerAddr := range defaultBootstrapPeers { peerinfo, err := peer.AddrInfoFromString(peerAddr) if err != nil { panic(err) } bootstrappers = append(bootstrappers, *peerinfo) } if isBootstrapPeer { kademliaDHT, err = dht.New( ctx, h, dht.Mode(dht.ModeServer), dht.BootstrapPeers(bootstrappers...), ) } else { panic( "this release is for bootstrap peers only, if you would like to run a " + "bootstrap node, please submit a pull request", ) } if err != nil { panic(err) } if err = kademliaDHT.Bootstrap(ctx); err != nil { panic(err) } reconnect := func() { for _, peerinfo := range bootstrappers { peerinfo := peerinfo go func() { if peerinfo.ID == h.ID() || h.Network().Connectedness(peerinfo.ID) == network.Connected || h.Network().Connectedness(peerinfo.ID) == network.Limited { return } if err := h.Connect(ctx, peerinfo); err != nil { logger.Debug("error while connecting to dht peer", zap.Error(err)) } else { h.ConnManager().Protect(peerinfo.ID, "bootstrap") logger.Debug( "connected to peer", zap.String("peer_id", peerinfo.ID.String()), ) } }() } } reconnect() go func() { for { time.Sleep(30 * time.Second) reconnect() } }() return kademliaDHT } func (b *BlossomSub) GetPeerScore(peerId []byte) int64 { b.peerScoreMx.Lock() score := b.peerScore[string(peerId)] b.peerScoreMx.Unlock() return score } func (b *BlossomSub) SetPeerScore(peerId []byte, score int64) { b.peerScoreMx.Lock() b.peerScore[string(peerId)] = score b.peerScoreMx.Unlock() } func (b *BlossomSub) GetBitmaskPeers() map[string][]string { peers := map[string][]string{} for _, k := range b.bitmaskMap { peers[fmt.Sprintf("%+x", k.Bitmask()[1:])] = []string{} for _, p := range k.ListPeers() { peers[fmt.Sprintf("%+x", k.Bitmask()[1:])] = append( peers[fmt.Sprintf("%+x", k.Bitmask()[1:])], p.String(), ) } } return peers } func (b *BlossomSub) GetPeerstoreCount() int { return len(b.h.Peerstore().Peers()) } func (b *BlossomSub) GetNetworkInfo() *protobufs.NetworkInfoResponse { resp := &protobufs.NetworkInfoResponse{} for _, p := range b.h.Network().Peers() { addrs := b.h.Peerstore().Addrs(p) multiaddrs := []string{} for _, a := range addrs { multiaddrs = append(multiaddrs, a.String()) } resp.NetworkInfo = append(resp.NetworkInfo, &protobufs.NetworkInfo{ PeerId: []byte(p), Multiaddrs: multiaddrs, PeerScore: b.ps.PeerScore(p), }) } return resp } func (b *BlossomSub) GetNetworkPeersCount() int { return len(b.h.Network().Peers()) } func (b *BlossomSub) GetPublicKey() []byte { pub, _ := b.signKey.GetPublic().Raw() return pub } func (b *BlossomSub) SignMessage(msg []byte) ([]byte, error) { sig, err := b.signKey.Sign(msg) return sig, errors.Wrap(err, "sign message") } func mergeDefaults(p2pConfig *config.P2PConfig) blossomsub.BlossomSubParams { p2pConfig.D = 0 p2pConfig.DLo = 0 p2pConfig.DHi = 0 p2pConfig.DScore = 0 if p2pConfig.DOut == 0 { p2pConfig.DOut = blossomsub.BlossomSubDout } if p2pConfig.HistoryLength == 0 { p2pConfig.HistoryLength = blossomsub.BlossomSubHistoryLength } if p2pConfig.HistoryGossip == 0 { p2pConfig.HistoryGossip = blossomsub.BlossomSubHistoryGossip } if p2pConfig.DLazy == 0 { p2pConfig.DLazy = blossomsub.BlossomSubDlazy } if p2pConfig.GossipRetransmission == 0 { p2pConfig.GossipRetransmission = blossomsub.BlossomSubGossipRetransmission } if p2pConfig.HeartbeatInitialDelay == 0 { p2pConfig.HeartbeatInitialDelay = blossomsub.BlossomSubHeartbeatInitialDelay } if p2pConfig.HeartbeatInterval == 0 { p2pConfig.HeartbeatInterval = blossomsub.BlossomSubHeartbeatInterval } if p2pConfig.FanoutTTL == 0 { p2pConfig.FanoutTTL = blossomsub.BlossomSubFanoutTTL } if p2pConfig.PrunePeers == 0 { p2pConfig.PrunePeers = blossomsub.BlossomSubPrunePeers } if p2pConfig.PruneBackoff == 0 { p2pConfig.PruneBackoff = blossomsub.BlossomSubPruneBackoff } if p2pConfig.UnsubscribeBackoff == 0 { p2pConfig.UnsubscribeBackoff = blossomsub.BlossomSubUnsubscribeBackoff } if p2pConfig.Connectors == 0 { p2pConfig.Connectors = blossomsub.BlossomSubConnectors } if p2pConfig.MaxPendingConnections == 0 { p2pConfig.MaxPendingConnections = blossomsub.BlossomSubMaxPendingConnections } if p2pConfig.ConnectionTimeout == 0 { p2pConfig.ConnectionTimeout = blossomsub.BlossomSubConnectionTimeout } if p2pConfig.DirectConnectTicks == 0 { p2pConfig.DirectConnectTicks = blossomsub.BlossomSubDirectConnectTicks } if p2pConfig.DirectConnectInitialDelay == 0 { p2pConfig.DirectConnectInitialDelay = blossomsub.BlossomSubDirectConnectInitialDelay } if p2pConfig.OpportunisticGraftTicks == 0 { p2pConfig.OpportunisticGraftTicks = blossomsub.BlossomSubOpportunisticGraftTicks } if p2pConfig.OpportunisticGraftPeers == 0 { p2pConfig.OpportunisticGraftPeers = blossomsub.BlossomSubOpportunisticGraftPeers } if p2pConfig.GraftFloodThreshold == 0 { p2pConfig.GraftFloodThreshold = blossomsub.BlossomSubGraftFloodThreshold } if p2pConfig.MaxIHaveLength == 0 { p2pConfig.MaxIHaveLength = blossomsub.BlossomSubMaxIHaveLength } if p2pConfig.MaxIHaveMessages == 0 { p2pConfig.MaxIHaveMessages = blossomsub.BlossomSubMaxIHaveMessages } if p2pConfig.IWantFollowupTime == 0 { p2pConfig.IWantFollowupTime = blossomsub.BlossomSubIWantFollowupTime } return blossomsub.BlossomSubParams{ D: p2pConfig.D, Dlo: p2pConfig.DLo, Dhi: p2pConfig.DHi, Dscore: p2pConfig.DScore, Dout: p2pConfig.DOut, HistoryLength: p2pConfig.HistoryLength, HistoryGossip: p2pConfig.HistoryGossip, Dlazy: p2pConfig.DLazy, GossipRetransmission: p2pConfig.GossipRetransmission, HeartbeatInitialDelay: p2pConfig.HeartbeatInitialDelay, HeartbeatInterval: p2pConfig.HeartbeatInterval, FanoutTTL: p2pConfig.FanoutTTL, PrunePeers: p2pConfig.PrunePeers, PruneBackoff: p2pConfig.PruneBackoff, UnsubscribeBackoff: p2pConfig.UnsubscribeBackoff, Connectors: p2pConfig.Connectors, MaxPendingConnections: p2pConfig.MaxPendingConnections, ConnectionTimeout: p2pConfig.ConnectionTimeout, DirectConnectTicks: p2pConfig.DirectConnectTicks, DirectConnectInitialDelay: p2pConfig.DirectConnectInitialDelay, OpportunisticGraftTicks: p2pConfig.OpportunisticGraftTicks, OpportunisticGraftPeers: p2pConfig.OpportunisticGraftPeers, GraftFloodThreshold: p2pConfig.GraftFloodThreshold, MaxIHaveLength: p2pConfig.MaxIHaveLength, MaxIHaveMessages: p2pConfig.MaxIHaveMessages, IWantFollowupTime: p2pConfig.IWantFollowupTime, SlowHeartbeatWarning: 0.1, } } func getNetworkNamespace(network uint8) string { var network_name string switch network { case 0: network_name = "mainnet" case 1: network_name = "testnet-primary" default: network_name = fmt.Sprintf("network-%d", network) } return ANNOUNCE_PREFIX + network_name }