mirror of
https://source.quilibrium.com/quilibrium/ceremonyclient.git
synced 2024-11-15 12:45:19 +00:00
563 lines
15 KiB
Go
563 lines
15 KiB
Go
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
|
|
}
|