From 3c6a45e5370577ac838367b12475038e14734bfe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 5 Oct 2023 18:52:32 +0300 Subject: [PATCH] eth: enforce announcement metadatas and drop peers violating the protocol --- eth/fetcher/tx_fetcher.go | 106 ++++++++++++++------ eth/fetcher/tx_fetcher_test.go | 31 +++++- eth/handler.go | 2 +- eth/handler_eth.go | 4 +- tests/fuzzers/txfetcher/txfetcher_fuzzer.go | 3 +- 5 files changed, 110 insertions(+), 36 deletions(-) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index a11b5e2164..07417e2656 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/eth/fetcher/tx_fetcher.go @@ -105,6 +105,14 @@ var ( type txAnnounce struct { origin string // Identifier of the peer originating the notification hashes []common.Hash // Batch of transaction hashes being announced + metas []*txMetadata // Batch of metadatas associated with the hashes (nil before eth/68) +} + +// txMetadata is a set of extra data transmitted along the announcement for better +// fetch scheduling. +type txMetadata struct { + kind byte // Transaction consensus type + size uint32 // Transaction size in bytes } // txRequest represents an in-flight transaction retrieval request destined to @@ -120,6 +128,7 @@ type txRequest struct { type txDelivery struct { origin string // Identifier of the peer originating the notification hashes []common.Hash // Batch of transaction hashes having been delivered + metas []txMetadata // Batch of metadatas associated with the delivered hashes direct bool // Whether this is a direct reply or a broadcast } @@ -155,14 +164,14 @@ type TxFetcher struct { // Stage 1: Waiting lists for newly discovered transactions that might be // broadcast without needing explicit request/reply round trips. - waitlist map[common.Hash]map[string]struct{} // Transactions waiting for an potential broadcast - waittime map[common.Hash]mclock.AbsTime // Timestamps when transactions were added to the waitlist - waitslots map[string]map[common.Hash]struct{} // Waiting announcements grouped by peer (DoS protection) + waitlist map[common.Hash]map[string]struct{} // Transactions waiting for an potential broadcast + waittime map[common.Hash]mclock.AbsTime // Timestamps when transactions were added to the waitlist + waitslots map[string]map[common.Hash]*txMetadata // Waiting announcements grouped by peer (DoS protection) // Stage 2: Queue of transactions that waiting to be allocated to some peer // to be retrieved directly. - announces map[string]map[common.Hash]struct{} // Set of announced transactions, grouped by origin peer - announced map[common.Hash]map[string]struct{} // Set of download locations, grouped by transaction hash + announces map[string]map[common.Hash]*txMetadata // Set of announced transactions, grouped by origin peer + announced map[common.Hash]map[string]struct{} // Set of download locations, grouped by transaction hash // Stage 3: Set of transactions currently being retrieved, some which may be // fulfilled and some rescheduled. Note, this step shares 'announces' from the @@ -175,6 +184,7 @@ type TxFetcher struct { hasTx func(common.Hash) bool // Retrieves a tx from the local txpool addTxs func([]*types.Transaction) []error // Insert a batch of transactions into local txpool fetchTxs func(string, []common.Hash) error // Retrieves a set of txs from a remote peer + dropPeer func(string) // Drops a peer in case of announcement violation step chan struct{} // Notification channel when the fetcher loop iterates clock mclock.Clock // Time wrapper to simulate in tests @@ -183,14 +193,14 @@ type TxFetcher struct { // NewTxFetcher creates a transaction fetcher to retrieve transaction // based on hash announcements. -func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error) *TxFetcher { - return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, mclock.System{}, nil) +func NewTxFetcher(hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string)) *TxFetcher { + return NewTxFetcherForTests(hasTx, addTxs, fetchTxs, dropPeer, mclock.System{}, nil) } // NewTxFetcherForTests is a testing method to mock out the realtime clock with // a simulated version and the internal randomness with a deterministic one. func NewTxFetcherForTests( - hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, + hasTx func(common.Hash) bool, addTxs func([]*types.Transaction) []error, fetchTxs func(string, []common.Hash) error, dropPeer func(string), clock mclock.Clock, rand *mrand.Rand) *TxFetcher { return &TxFetcher{ notify: make(chan *txAnnounce), @@ -199,8 +209,8 @@ func NewTxFetcherForTests( quit: make(chan struct{}), waitlist: make(map[common.Hash]map[string]struct{}), waittime: make(map[common.Hash]mclock.AbsTime), - waitslots: make(map[string]map[common.Hash]struct{}), - announces: make(map[string]map[common.Hash]struct{}), + waitslots: make(map[string]map[common.Hash]*txMetadata), + announces: make(map[string]map[common.Hash]*txMetadata), announced: make(map[common.Hash]map[string]struct{}), fetching: make(map[common.Hash]string), requests: make(map[string]*txRequest), @@ -209,6 +219,7 @@ func NewTxFetcherForTests( hasTx: hasTx, addTxs: addTxs, fetchTxs: fetchTxs, + dropPeer: dropPeer, clock: clock, rand: rand, } @@ -216,7 +227,7 @@ func NewTxFetcherForTests( // Notify announces the fetcher of the potential availability of a new batch of // transactions in the network. -func (f *TxFetcher) Notify(peer string, hashes []common.Hash) error { +func (f *TxFetcher) Notify(peer string, types []byte, sizes []uint32, hashes []common.Hash) error { // Keep track of all the announced transactions txAnnounceInMeter.Mark(int64(len(hashes))) @@ -226,28 +237,35 @@ func (f *TxFetcher) Notify(peer string, hashes []common.Hash) error { // still valuable to check here because it runs concurrent to the internal // loop, so anything caught here is time saved internally. var ( - unknowns = make([]common.Hash, 0, len(hashes)) + unknownHashes = make([]common.Hash, 0, len(hashes)) + unknownMetas = make([]*txMetadata, 0, len(hashes)) + duplicate int64 underpriced int64 ) - for _, hash := range hashes { + for i, hash := range hashes { switch { case f.hasTx(hash): duplicate++ case f.isKnownUnderpriced(hash): underpriced++ default: - unknowns = append(unknowns, hash) + unknownHashes = append(unknownHashes, hash) + if types == nil { + unknownMetas = append(unknownMetas, nil) + } else { + unknownMetas = append(unknownMetas, &txMetadata{kind: types[i], size: sizes[i]}) + } } } txAnnounceKnownMeter.Mark(duplicate) txAnnounceUnderpricedMeter.Mark(underpriced) // If anything's left to announce, push it into the internal loop - if len(unknowns) == 0 { + if len(unknownHashes) == 0 { return nil } - announce := &txAnnounce{origin: peer, hashes: unknowns} + announce := &txAnnounce{origin: peer, hashes: unknownHashes, metas: unknownMetas} select { case f.notify <- announce: return nil @@ -290,6 +308,7 @@ func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) // re-requesting them and dropping the peer in case of malicious transfers. var ( added = make([]common.Hash, 0, len(txs)) + metas = make([]txMetadata, 0, len(txs)) ) // proceed in batches for i := 0; i < len(txs); i += 128 { @@ -325,6 +344,10 @@ func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) otherreject++ } added = append(added, batch[j].Hash()) + metas = append(metas, txMetadata{ + kind: batch[j].Type(), + size: uint32(batch[j].Size()), + }) } knownMeter.Mark(duplicate) underpricedMeter.Mark(underpriced) @@ -337,7 +360,7 @@ func (f *TxFetcher) Enqueue(peer string, txs []*types.Transaction, direct bool) } } select { - case f.cleanup <- &txDelivery{origin: peer, hashes: added, direct: direct}: + case f.cleanup <- &txDelivery{origin: peer, hashes: added, metas: metas, direct: direct}: return nil case <-f.quit: return errTerminated @@ -394,13 +417,15 @@ func (f *TxFetcher) loop() { want := used + len(ann.hashes) if want > maxTxAnnounces { txAnnounceDOSMeter.Mark(int64(want - maxTxAnnounces)) + ann.hashes = ann.hashes[:want-maxTxAnnounces] + ann.metas = ann.metas[:want-maxTxAnnounces] } // All is well, schedule the remainder of the transactions idleWait := len(f.waittime) == 0 _, oldPeer := f.announces[ann.origin] - for _, hash := range ann.hashes { + for i, hash := range ann.hashes { // If the transaction is already downloading, add it to the list // of possible alternates (in case the current retrieval fails) and // also account it for the peer. @@ -409,9 +434,9 @@ func (f *TxFetcher) loop() { // Stage 2 and 3 share the set of origins per tx if announces := f.announces[ann.origin]; announces != nil { - announces[hash] = struct{}{} + announces[hash] = ann.metas[i] } else { - f.announces[ann.origin] = map[common.Hash]struct{}{hash: {}} + f.announces[ann.origin] = map[common.Hash]*txMetadata{hash: ann.metas[i]} } continue } @@ -422,9 +447,9 @@ func (f *TxFetcher) loop() { // Stage 2 and 3 share the set of origins per tx if announces := f.announces[ann.origin]; announces != nil { - announces[hash] = struct{}{} + announces[hash] = ann.metas[i] } else { - f.announces[ann.origin] = map[common.Hash]struct{}{hash: {}} + f.announces[ann.origin] = map[common.Hash]*txMetadata{hash: ann.metas[i]} } continue } @@ -435,9 +460,9 @@ func (f *TxFetcher) loop() { f.waitlist[hash][ann.origin] = struct{}{} if waitslots := f.waitslots[ann.origin]; waitslots != nil { - waitslots[hash] = struct{}{} + waitslots[hash] = ann.metas[i] } else { - f.waitslots[ann.origin] = map[common.Hash]struct{}{hash: {}} + f.waitslots[ann.origin] = map[common.Hash]*txMetadata{hash: ann.metas[i]} } continue } @@ -446,9 +471,9 @@ func (f *TxFetcher) loop() { f.waittime[hash] = f.clock.Now() if waitslots := f.waitslots[ann.origin]; waitslots != nil { - waitslots[hash] = struct{}{} + waitslots[hash] = ann.metas[i] } else { - f.waitslots[ann.origin] = map[common.Hash]struct{}{hash: {}} + f.waitslots[ann.origin] = map[common.Hash]*txMetadata{hash: ann.metas[i]} } } // If a new item was added to the waitlist, schedule it into the fetcher @@ -474,9 +499,9 @@ func (f *TxFetcher) loop() { f.announced[hash] = f.waitlist[hash] for peer := range f.waitlist[hash] { if announces := f.announces[peer]; announces != nil { - announces[hash] = struct{}{} + announces[hash] = f.waitslots[peer][hash] } else { - f.announces[peer] = map[common.Hash]struct{}{hash: {}} + f.announces[peer] = map[common.Hash]*txMetadata{hash: f.waitslots[peer][hash]} } delete(f.waitslots[peer], hash) if len(f.waitslots[peer]) == 0 { @@ -545,10 +570,20 @@ func (f *TxFetcher) loop() { case delivery := <-f.cleanup: // Independent if the delivery was direct or broadcast, remove all - // traces of the hash from internal trackers - for _, hash := range delivery.hashes { + // traces of the hash from internal trackers. That said, compare any + // advertised metadata with the real ones and drop bad peers. + for i, hash := range delivery.hashes { if _, ok := f.waitlist[hash]; ok { for peer, txset := range f.waitslots { + if meta := txset[hash]; meta != nil { + if delivery.metas[i].kind != meta.kind { + log.Warn("Announced transaction type mismatch", "peer", peer, "tx", hash, "type", delivery.metas[i].kind, "ann", meta.kind) + f.dropPeer(peer) + } else if delivery.metas[i].size != meta.size { + log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", delivery.metas[i].size, "ann", meta.size) + f.dropPeer(peer) + } + } delete(txset, hash) if len(txset) == 0 { delete(f.waitslots, peer) @@ -558,6 +593,15 @@ func (f *TxFetcher) loop() { delete(f.waittime, hash) } else { for peer, txset := range f.announces { + if meta := txset[hash]; meta != nil { + if delivery.metas[i].kind != meta.kind { + log.Warn("Announced transaction type mismatch", "peer", peer, "tx", hash, "type", delivery.metas[i].kind, "ann", meta.kind) + f.dropPeer(peer) + } else if delivery.metas[i].size != meta.size { + log.Warn("Announced transaction size mismatch", "peer", peer, "tx", hash, "size", delivery.metas[i].size, "ann", meta.size) + f.dropPeer(peer) + } + } delete(txset, hash) if len(txset) == 0 { delete(f.announces, peer) @@ -859,7 +903,7 @@ func (f *TxFetcher) forEachPeer(peers map[string]struct{}, do func(peer string)) // forEachHash does a range loop over a map of hashes in production, but during // testing it does a deterministic sorted random to allow reproducing issues. -func (f *TxFetcher) forEachHash(hashes map[common.Hash]struct{}, do func(hash common.Hash) bool) { +func (f *TxFetcher) forEachHash(hashes map[common.Hash]*txMetadata, do func(hash common.Hash) bool) { // If we're running production, use whatever Go's map gives us if f.rand == nil { for hash := range hashes { diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 980c1a6c26..1a80264ef6 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/eth/fetcher/tx_fetcher_test.go @@ -81,6 +81,7 @@ func TestTransactionFetcherWaiting(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -171,6 +172,7 @@ func TestTransactionFetcherSkipWaiting(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -234,6 +236,7 @@ func TestTransactionFetcherSingletonRequesting(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -313,6 +316,7 @@ func TestTransactionFetcherFailedRescheduling(t *testing.T) { <-proceed return errors.New("peer disconnected") }, + nil, ) }, steps: []interface{}{ @@ -382,6 +386,7 @@ func TestTransactionFetcherCleanup(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -421,6 +426,7 @@ func TestTransactionFetcherCleanupEmpty(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -459,6 +465,7 @@ func TestTransactionFetcherMissingRescheduling(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -505,6 +512,7 @@ func TestTransactionFetcherMissingCleanup(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -543,6 +551,7 @@ func TestTransactionFetcherBroadcasts(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -591,6 +600,7 @@ func TestTransactionFetcherWaitTimerResets(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -648,6 +658,7 @@ func TestTransactionFetcherTimeoutRescheduling(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -713,6 +724,7 @@ func TestTransactionFetcherTimeoutTimerResets(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -772,6 +784,7 @@ func TestTransactionFetcherRateLimiting(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -810,6 +823,7 @@ func TestTransactionFetcherDoSProtection(t *testing.T) { func(common.Hash) bool { return false }, nil, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -877,6 +891,7 @@ func TestTransactionFetcherUnderpricedDedup(t *testing.T) { return errs }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -946,6 +961,7 @@ func TestTransactionFetcherUnderpricedDoSProtection(t *testing.T) { return errs }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: append(steps, []interface{}{ @@ -968,6 +984,7 @@ func TestTransactionFetcherOutOfBoundDeliveries(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1021,6 +1038,7 @@ func TestTransactionFetcherDrop(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1087,6 +1105,7 @@ func TestTransactionFetcherDropRescheduling(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1120,6 +1139,12 @@ func TestTransactionFetcherDropRescheduling(t *testing.T) { }) } +// Tests that announced transactions with the wrong transaction type or size will +// result in a dropped peer. +func TestInvalidAnnounceMetadata(t *testing.T) { + +} + // This test reproduces a crash caught by the fuzzer. The root cause was a // dangling transaction timing out and clashing on re-add with a concurrently // announced one. @@ -1132,6 +1157,7 @@ func TestTransactionFetcherFuzzCrash01(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1159,6 +1185,7 @@ func TestTransactionFetcherFuzzCrash02(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1188,6 +1215,7 @@ func TestTransactionFetcherFuzzCrash03(t *testing.T) { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, ) }, steps: []interface{}{ @@ -1224,6 +1252,7 @@ func TestTransactionFetcherFuzzCrash04(t *testing.T) { <-proceed return errors.New("peer disconnected") }, + nil, ) }, steps: []interface{}{ @@ -1276,7 +1305,7 @@ func testTransactionFetcher(t *testing.T, tt txFetcherTest) { for i, step := range tt.steps { switch step := step.(type) { case doTxNotify: - if err := fetcher.Notify(step.peer, step.hashes); err != nil { + if err := fetcher.Notify(step.peer, nil, nil, step.hashes); err != nil { t.Errorf("step %d: %v", i, err) } <-wait // Fetcher needs to process this, wait until it's done diff --git a/eth/handler.go b/eth/handler.go index 0c0c17fee1..f0021e5644 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -277,7 +277,7 @@ func newHandler(config *handlerConfig) (*handler, error) { addTxs := func(txs []*types.Transaction) []error { return h.txpool.Add(txs, false, false) } - h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, addTxs, fetchTx) + h.txFetcher = fetcher.NewTxFetcher(h.txpool.Has, addTxs, fetchTx, h.removePeer) h.chainSync = newChainSyncer(h) return h, nil } diff --git a/eth/handler_eth.go b/eth/handler_eth.go index e844b36cca..2a839f615f 100644 --- a/eth/handler_eth.go +++ b/eth/handler_eth.go @@ -68,10 +68,10 @@ func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error { return h.handleBlockBroadcast(peer, packet.Block, packet.TD) case *eth.NewPooledTransactionHashesPacket67: - return h.txFetcher.Notify(peer.ID(), *packet) + return h.txFetcher.Notify(peer.ID(), nil, nil, *packet) case *eth.NewPooledTransactionHashesPacket68: - return h.txFetcher.Notify(peer.ID(), packet.Hashes) + return h.txFetcher.Notify(peer.ID(), packet.Types, packet.Sizes, packet.Hashes) case *eth.TransactionsPacket: for _, tx := range *packet { diff --git a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go index d1d6fdc665..6c4526bca8 100644 --- a/tests/fuzzers/txfetcher/txfetcher_fuzzer.go +++ b/tests/fuzzers/txfetcher/txfetcher_fuzzer.go @@ -83,6 +83,7 @@ func Fuzz(input []byte) int { return make([]error, len(txs)) }, func(string, []common.Hash) error { return nil }, + nil, clock, rand, ) f.Start() @@ -128,7 +129,7 @@ func Fuzz(input []byte) int { if verbose { fmt.Println("Notify", peer, announceIdxs) } - if err := f.Notify(peer, announces); err != nil { + if err := f.Notify(peer, nil, nil, announces); err != nil { panic(err) }