102 lines
3.0 KiB
Go
102 lines
3.0 KiB
Go
|
package pubsub
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"encoding/binary"
|
||
|
|
"sync"
|
||
|
|
|
||
|
|
"github.com/libp2p/go-libp2p/core/peer"
|
||
|
|
)
|
||
|
|
|
||
|
|
// PeerMetadataStore is an interface for storing and retrieving per peer metadata
|
||
|
|
type PeerMetadataStore interface {
|
||
|
|
// Get retrieves the metadata associated with a peer;
|
||
|
|
// It should return nil if there is no metadata associated with the peer and not an error.
|
||
|
|
Get(context.Context, peer.ID) ([]byte, error)
|
||
|
|
// Put sets the metadata associated with a peer.
|
||
|
|
Put(context.Context, peer.ID, []byte) error
|
||
|
|
}
|
||
|
|
|
||
|
|
// BasicSeqnoValidator is a basic validator, usable as a default validator, that ignores replayed
|
||
|
|
// messages outside the seen cache window. The validator uses the message seqno as a peer-specific
|
||
|
|
// nonce to decide whether the message should be propagated, comparing to the maximal nonce store
|
||
|
|
// in the peer metadata store. This is useful to ensure that there can be no infinitely propagating
|
||
|
|
// messages in the network regardless of the seen cache span and network diameter.
|
||
|
|
// It requires that pubsub is instantiated with a strict message signing policy and that seqnos
|
||
|
|
// are not disabled, ie it doesn't support anonymous mode.
|
||
|
|
//
|
||
|
|
// Warning: See https://github.com/libp2p/rust-libp2p/issues/3453
|
||
|
|
// TL;DR: rust is currently violating the spec by issuing a random seqno, which creates an
|
||
|
|
// interoperability hazard. We expect this issue to be addressed in the not so distant future,
|
||
|
|
// but keep this in mind if you are in a mixed environment with (older) rust nodes.
|
||
|
|
type BasicSeqnoValidator struct {
|
||
|
|
mx sync.RWMutex
|
||
|
|
meta PeerMetadataStore
|
||
|
|
}
|
||
|
|
|
||
|
|
// NewBasicSeqnoValidator constructs a BasicSeqnoValidator using the givven PeerMetadataStore.
|
||
|
|
func NewBasicSeqnoValidator(meta PeerMetadataStore) ValidatorEx {
|
||
|
|
val := &BasicSeqnoValidator{
|
||
|
|
meta: meta,
|
||
|
|
}
|
||
|
|
return val.validate
|
||
|
|
}
|
||
|
|
|
||
|
|
func (v *BasicSeqnoValidator) validate(ctx context.Context, _ peer.ID, m *Message) ValidationResult {
|
||
|
|
p := m.GetFrom()
|
||
|
|
|
||
|
|
v.mx.RLock()
|
||
|
|
nonceBytes, err := v.meta.Get(ctx, p)
|
||
|
|
v.mx.RUnlock()
|
||
|
|
|
||
|
|
if err != nil {
|
||
|
|
log.Warn("error retrieving peer nonce: %s", err)
|
||
|
|
return ValidationIgnore
|
||
|
|
}
|
||
|
|
|
||
|
|
var nonce uint64
|
||
|
|
if len(nonceBytes) > 0 {
|
||
|
|
nonce = binary.BigEndian.Uint64(nonceBytes)
|
||
|
|
}
|
||
|
|
|
||
|
|
var seqno uint64
|
||
|
|
seqnoBytes := m.GetSeqno()
|
||
|
|
if len(seqnoBytes) > 0 {
|
||
|
|
seqno = binary.BigEndian.Uint64(seqnoBytes)
|
||
|
|
}
|
||
|
|
|
||
|
|
// compare against the largest seen nonce
|
||
|
|
if seqno <= nonce {
|
||
|
|
return ValidationIgnore
|
||
|
|
}
|
||
|
|
|
||
|
|
// get the nonce and compare again with an exclusive lock before commiting (cf concurrent validation)
|
||
|
|
v.mx.Lock()
|
||
|
|
defer v.mx.Unlock()
|
||
|
|
|
||
|
|
nonceBytes, err = v.meta.Get(ctx, p)
|
||
|
|
if err != nil {
|
||
|
|
log.Warn("error retrieving peer nonce: %s", err)
|
||
|
|
return ValidationIgnore
|
||
|
|
}
|
||
|
|
|
||
|
|
if len(nonceBytes) > 0 {
|
||
|
|
nonce = binary.BigEndian.Uint64(nonceBytes)
|
||
|
|
}
|
||
|
|
|
||
|
|
if seqno <= nonce {
|
||
|
|
return ValidationIgnore
|
||
|
|
}
|
||
|
|
|
||
|
|
// update the nonce
|
||
|
|
nonceBytes = make([]byte, 8)
|
||
|
|
binary.BigEndian.PutUint64(nonceBytes, seqno)
|
||
|
|
|
||
|
|
err = v.meta.Put(ctx, p, nonceBytes)
|
||
|
|
if err != nil {
|
||
|
|
log.Warn("error storing peer nonce: %s", err)
|
||
|
|
}
|
||
|
|
|
||
|
|
return ValidationAccept
|
||
|
|
}
|