Relay transactions with exponential decaying probability

blockmanager.go

@@ -12,6 +12,7 @@ import (
 	"github.com/gcash/neutrino/chainsync"
 	"math"
 	"math/big"
+	"math/rand"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -47,6 +48,10 @@ const (
 	// maxCFCheckptsPerQuery is the maximum number of filter header
 	// checkpoints we can query for within a single message over the wire.
 	maxCFCheckptsPerQuery = wire.MaxCFHeadersPerMsg / wire.CFCheckptInterval
+
+	// defaultRelayMetric is the default probability of relaying a transaction
+	// per block.
+	defaultRelayMetric = 50
 )
 
 // filterStoreLookup
@@ -207,6 +212,8 @@ type blockManager struct {
 	blocksPerRetarget   int32 // target timespan / target time per block
 
 	requestedTxns map[chainhash.Hash]struct{}
+	relayMetric   int
+	relayRand     *rand.Rand
 }
 
 // newBlockManager returns a new bitcoin block manager.  Use Start to begin
@@ -240,6 +247,8 @@ func newBlockManager(s *ChainService,
 		minRetargetTimespan: targetTimespan / adjustmentFactor,
 		maxRetargetTimespan: targetTimespan * adjustmentFactor,
 		requestedTxns:       make(map[chainhash.Hash]struct{}),
+		relayMetric:         defaultRelayMetric,
+		relayRand:           rand.New(rand.NewSource(time.Now().UnixNano())),
 		firstPeerSignal:     firstPeerSignal,
 	}
 
@@ -2195,6 +2204,27 @@ func (b *blockManager) handleTxMsg(tmsg *txMsg) {
 	}
 	b.server.mempool.AddTransaction(tmsg.tx)
 	delete(b.requestedTxns, *txHash)
+
+	// We want to decide whether or not to relay this transaction. We want to relay
+	// for privacy reasons to make it so the remote peer cannot tell if the transaction
+	// is ours or a tx that we are relaying. The downside here is the transactions
+	// are not validated so we might be relaying an invalid transaction. Currently bchd
+	// nodes do not ban peers which relay invalid transactions, however we still do not
+	// want to cause an amplification attack. So our criteria for relaying is we relay
+	// with an exponential decaying probability.
+
+	if b.randomRelay() && b.server.shouldRelayTx(tmsg.tx.MsgTx()) {
+		if err := b.server.sendTransaction(tmsg.tx.MsgTx()); err != nil {
+			log.Errorf("Relay of mempool tx error: %v", err)
+		}
+
+		b.relayMetric *= 2
+	}
+}
+
+// randomRelay returns true with a 1 / relayMetric probability.
+func (b *blockManager) randomRelay() bool {
+	return b.relayRand.Intn(b.relayMetric) == 0
 }
 
 // QueueHeaders adds the passed headers message and peer to the block handling
@@ -2558,6 +2588,9 @@ func (b *blockManager) handleHeadersMsg(hmsg *headersMsg) {
 	// Clear the mempool to free up memory. This may mean we might receive
 	// transactions we've previously downloaded but this is rather unlikely.
 	b.server.mempool.Clear()
+
+	// Reset the relay metric.
+	b.relayMetric = defaultRelayMetric
 }
 
 // checkHeaderSanity checks the PoW, and timestamp of a block header.

neutrino.go

@@ -576,6 +576,12 @@ type Config struct {
 	// Proxy is an address to use to connect remote peers using the socks5 proxy.
 	Proxy string
 
+	// ShouldRelayTx is an optional function that can be provided to narrow
+	// the range of transactions that will be relayed to the remote peer. For
+	// example, if you only make p2pkh txs you might only want to relay p2pkh txs.
+	// The default, if this is nil, will only relay 2 output p2pkh txs.
+	ShouldRelayTx func(tx *wire.MsgTx) bool
+
 	// AssertFilterHeader is an optional field that allows the creator of
 	// the ChainService to ensure that if any chain data exists, it's
 	// compliant with the expected filter header state. If neutrino starts
@@ -639,6 +645,8 @@ type ChainService struct {
 
 	blocksOnly bool
 
+	shouldRelayTx func(tx *wire.MsgTx) bool
+
 	mempool *Mempool
 
 	proxy string
@@ -673,6 +681,25 @@ func NewChainService(cfg Config) (*ChainService, error) {
 		nameResolver = net.LookupIP
 	}
 
+	if cfg.ShouldRelayTx == nil {
+		cfg.ShouldRelayTx = func(tx *wire.MsgTx) bool {
+			if len(tx.TxOut) != 2 {
+				return false
+			}
+
+			for _, out := range tx.TxOut {
+				class, _, _, err := txscript.ExtractPkScriptAddrs(out.PkScript, &cfg.ChainParams)
+				if err != nil {
+					return false
+				}
+				if class != txscript.PubKeyHashTy {
+					return false
+				}
+			}
+			return true
+		}
+	}
+
 	// When creating the addr manager, we'll check to see if the user has
 	// provided their own resolution function. If so, then we'll use that
 	// instead as this may be proxying requests over an anonymizing
@@ -695,6 +722,7 @@ func NewChainService(cfg Config) (*ChainService, error) {
 		nameResolver:      nameResolver,
 		dialer:            dialer,
 		blocksOnly:        cfg.BlocksOnly,
+		shouldRelayTx:     cfg.ShouldRelayTx,
 		mempool:           NewMempool(),
 		proxy:             cfg.Proxy,
 	}