Skip to content

Commit

Permalink
Remove IPNI advertisement relay over pubsub via Lotus node
Browse files Browse the repository at this point in the history 
The initial implementation of IPNI used GossipSub as a way to propagate
IPNI advertisement chain. To do this the propagation had to be relayed
through the Lotus node due to strict Filecoin GossipSub validation
rules.

Since then IPNI has moved on to roll out its own sync protocol that
works over HTTP, and HTTP-over-libp2p. This move has been the official
way of advertising content to IPNI federation over a year now.

Therefore, remove the ad relay over pubsub via Lotus node as it is now
considered to have reached its EOL as a mechanism for advertising to
IPNI.
  • Loading branch information
@masih
masih committed Dec 10, 2024
1 parent 8b3de48 commit 4d997f0
Show file tree
Hide file tree
Showing 9 changed files with 8 additions and 428 deletions.
11 changes: 0 additions & 11 deletions build/params_shared_funcs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,17 +14,6 @@ import (

func BlocksTopic(netName dtypes.NetworkName) string { return "/fil/blocks/" + string(netName) }
func MessagesTopic(netName dtypes.NetworkName) string { return "/fil/msgs/" + string(netName) }
func IndexerIngestTopic(netName dtypes.NetworkName) string {

nn := string(netName)
// The network name testnetnet is here for historical reasons.
// Going forward we aim to use the name `mainnet` where possible.
if nn == "testnetnet" {
nn = "mainnet"
}

return "/indexer/ingest/" + nn
}
func DhtProtocolName(netName dtypes.NetworkName) protocol.ID {
return protocol.ID("/fil/kad/" + string(netName))
}
Expand Down
170 changes: 0 additions & 170 deletions chain/sub/incoming.go
View file
Original file line number Diff line number Diff line change
@@ -1,19 +1,15 @@
package sub

import (
"bytes"
"context"
"encoding/binary"
"errors"
"sync"
"time"

lru "github.com/hashicorp/golang-lru/v2"
bserv "github.com/ipfs/boxo/blockservice"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
logging "github.com/ipfs/go-log/v2"
"github.com/ipni/go-libipni/announce/message"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"github.com/libp2p/go-libp2p/core/connmgr"
"github.com/libp2p/go-libp2p/core/peer"
Expand All @@ -30,10 +26,8 @@ import (
"github.com/filecoin-project/lotus/chain/consensus"
"github.com/filecoin-project/lotus/chain/messagepool"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/sub/ratelimit"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/metrics"
"github.com/filecoin-project/lotus/node/impl/full"
)

var log = logging.Logger("sub")
Expand Down Expand Up @@ -470,167 +464,3 @@ func recordFailure(ctx context.Context, metric *stats.Int64Measure, failureType
)
stats.Record(ctx, metric.M(1))
}

type peerMsgInfo struct {
peerID peer.ID
lastCid cid.Cid
lastSeqno uint64
rateLimit *ratelimit.Window
mutex sync.Mutex
}

type IndexerMessageValidator struct {
self peer.ID

peerCache *lru.TwoQueueCache[address.Address, *peerMsgInfo]
chainApi full.ChainModuleAPI
stateApi full.StateModuleAPI
}

func NewIndexerMessageValidator(self peer.ID, chainApi full.ChainModuleAPI, stateApi full.StateModuleAPI) *IndexerMessageValidator {
peerCache, _ := lru.New2Q[address.Address, *peerMsgInfo](8192)

return &IndexerMessageValidator{
self: self,
peerCache: peerCache,
chainApi: chainApi,
stateApi: stateApi,
}
}

func (v *IndexerMessageValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) pubsub.ValidationResult {
// This chain-node should not be publishing its own messages. These are
// relayed from market-nodes. If a node appears to be local, reject it.
if pid == v.self {
log.Debug("ignoring indexer message from self")
stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
return pubsub.ValidationIgnore
}
originPeer := msg.GetFrom()
if originPeer == v.self {
log.Debug("ignoring indexer message originating from self")
stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
return pubsub.ValidationIgnore
}

idxrMsg := message.Message{}
err := idxrMsg.UnmarshalCBOR(bytes.NewBuffer(msg.Data))
if err != nil {
log.Errorw("Could not decode indexer pubsub message", "err", err)
return pubsub.ValidationReject
}
if len(idxrMsg.ExtraData) == 0 {
log.Debugw("ignoring message missing miner id", "peer", originPeer)
return pubsub.ValidationIgnore
}

// Get miner info from lotus
minerAddr, err := address.NewFromBytes(idxrMsg.ExtraData)
if err != nil {
log.Warnw("cannot parse extra data as miner address", "err", err, "extraData", idxrMsg.ExtraData)
return pubsub.ValidationReject
}

msgCid := idxrMsg.Cid

msgInfo, cached := v.peerCache.Get(minerAddr)
if !cached {
msgInfo = &peerMsgInfo{}
}

// Lock this peer's message info.
msgInfo.mutex.Lock()
defer msgInfo.mutex.Unlock()

var seqno uint64
if cached {
// Reject replayed messages.
seqno = binary.BigEndian.Uint64(msg.Message.GetSeqno())
if seqno <= msgInfo.lastSeqno {
log.Debugf("ignoring replayed indexer message")
return pubsub.ValidationIgnore
}
}

if !cached || originPeer != msgInfo.peerID {
// Check that the miner ID maps to the peer that sent the message.
err = v.authenticateMessage(ctx, minerAddr, originPeer)
if err != nil {
log.Warnw("cannot authenticate message", "err", err, "peer", originPeer, "minerID", minerAddr)
stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
return pubsub.ValidationReject
}
msgInfo.peerID = originPeer
if !cached {
// Add msgInfo to cache only after being authenticated. If two
// messages from the same peer are handled concurrently, there is a
// small chance that one msgInfo could replace the other here when
// the info is first cached. This is OK, so no need to prevent it.
v.peerCache.Add(minerAddr, msgInfo)
}
}

// Update message info cache with the latest message's sequence number.
msgInfo.lastSeqno = seqno

// See if message needs to be ignored due to rate limiting.
if v.rateLimitPeer(msgInfo, msgCid) {
return pubsub.ValidationIgnore
}

stats.Record(ctx, metrics.IndexerMessageValidationSuccess.M(1))
return pubsub.ValidationAccept
}

func (v *IndexerMessageValidator) rateLimitPeer(msgInfo *peerMsgInfo, msgCid cid.Cid) bool {
const (
msgLimit = 5
msgTimeLimit = 10 * time.Second
repeatTimeLimit = 2 * time.Hour
)

timeWindow := msgInfo.rateLimit

// Check overall message rate.
if timeWindow == nil {
timeWindow = ratelimit.NewWindow(msgLimit, msgTimeLimit)
msgInfo.rateLimit = timeWindow
} else if msgInfo.lastCid == msgCid {
// Check if this is a repeat of the previous message data.
if time.Since(timeWindow.Newest()) < repeatTimeLimit {
log.Warnw("ignoring repeated indexer message", "sender", msgInfo.peerID)
return true
}
}

err := timeWindow.Add()
if err != nil {
log.Warnw("ignoring indexer message", "sender", msgInfo.peerID, "err", err)
return true
}

msgInfo.lastCid = msgCid

return false
}

func (v *IndexerMessageValidator) authenticateMessage(ctx context.Context, minerAddress address.Address, peerID peer.ID) error {
ts, err := v.chainApi.ChainHead(ctx)
if err != nil {
return err
}

minerInfo, err := v.stateApi.StateMinerInfo(ctx, minerAddress, ts.Key())
if err != nil {
return err
}

if minerInfo.PeerId == nil {
return xerrors.New("no peer id for miner")
}
if *minerInfo.PeerId != peerID {
return xerrors.New("miner id does not map to peer that sent message")
}

return nil
}
Loading

0 comments on commit 4d997f0

Please sign in to comment.