eth: enforce announcement metadatas and drop peers violating the protocol

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
+	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
+	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)
+	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
+	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
+	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 {
+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, mclock.System{}, nil)
+	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{}),
+		waitslots:   make(map[string]map[common.Hash]*txMetadata),
-		announces:   make(map[string]map[common.Hash]struct{}),
+		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
+			// traces of the hash from internal trackers. That said, compare any
-			for _, hash := range delivery.hashes {
+			// 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 {

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

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
 }

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 {

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)
 			}