123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218 |
- // Copyright 2015 The go-ethereum Authors
- // This file is part of the go-ethereum library.
- //
- // The go-ethereum library is free software: you can redistribute it and/or modify
- // it under the terms of the GNU Lesser General Public License as published by
- // the Free Software Foundation, either version 3 of the License, or
- // (at your option) any later version.
- //
- // The go-ethereum library is distributed in the hope that it will be useful,
- // but WITHOUT ANY WARRANTY; without even the implied warranty of
- // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- // GNU Lesser General Public License for more details.
- //
- // You should have received a copy of the GNU Lesser General Public License
- // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
- package eth
- import (
- "errors"
- "fmt"
- "math/big"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common"
- "github.com/ethereum/go-ethereum/core"
- "github.com/ethereum/go-ethereum/core/types"
- "github.com/ethereum/go-ethereum/eth/protocols/eth"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/p2p/enode"
- "github.com/ethereum/go-ethereum/trie"
- )
- // ethHandler implements the eth.Backend interface to handle the various network
- // packets that are sent as replies or broadcasts.
- type ethHandler handler
- func (h *ethHandler) Chain() *core.BlockChain { return h.chain }
- func (h *ethHandler) StateBloom() *trie.SyncBloom { return h.stateBloom }
- func (h *ethHandler) TxPool() eth.TxPool { return h.txpool }
- // RunPeer is invoked when a peer joins on the `eth` protocol.
- func (h *ethHandler) RunPeer(peer *eth.Peer, hand eth.Handler) error {
- return (*handler)(h).runEthPeer(peer, hand)
- }
- // PeerInfo retrieves all known `eth` information about a peer.
- func (h *ethHandler) PeerInfo(id enode.ID) interface{} {
- if p := h.peers.peer(id.String()); p != nil {
- return p.info()
- }
- return nil
- }
- // AcceptTxs retrieves whether transaction processing is enabled on the node
- // or if inbound transactions should simply be dropped.
- func (h *ethHandler) AcceptTxs() bool {
- return atomic.LoadUint32(&h.acceptTxs) == 1
- }
- // Handle is invoked from a peer's message handler when it receives a new remote
- // message that the handler couldn't consume and serve itself.
- func (h *ethHandler) Handle(peer *eth.Peer, packet eth.Packet) error {
- // Consume any broadcasts and announces, forwarding the rest to the downloader
- switch packet := packet.(type) {
- case *eth.BlockHeadersPacket:
- return h.handleHeaders(peer, *packet)
- case *eth.BlockBodiesPacket:
- txset, uncleset := packet.Unpack()
- return h.handleBodies(peer, txset, uncleset)
- case *eth.NodeDataPacket:
- if err := h.downloader.DeliverNodeData(peer.ID(), *packet); err != nil {
- log.Debug("Failed to deliver node state data", "err", err)
- }
- return nil
- case *eth.ReceiptsPacket:
- if err := h.downloader.DeliverReceipts(peer.ID(), *packet); err != nil {
- log.Debug("Failed to deliver receipts", "err", err)
- }
- return nil
- case *eth.NewBlockHashesPacket:
- hashes, numbers := packet.Unpack()
- return h.handleBlockAnnounces(peer, hashes, numbers)
- case *eth.NewBlockPacket:
- return h.handleBlockBroadcast(peer, packet.Block, packet.TD)
- case *eth.NewPooledTransactionHashesPacket:
- return h.txFetcher.Notify(peer.ID(), *packet)
- case *eth.TransactionsPacket:
- return h.txFetcher.Enqueue(peer.ID(), *packet, false)
- case *eth.PooledTransactionsPacket:
- return h.txFetcher.Enqueue(peer.ID(), *packet, true)
- default:
- return fmt.Errorf("unexpected eth packet type: %T", packet)
- }
- }
- // handleHeaders is invoked from a peer's message handler when it transmits a batch
- // of headers for the local node to process.
- func (h *ethHandler) handleHeaders(peer *eth.Peer, headers []*types.Header) error {
- p := h.peers.peer(peer.ID())
- if p == nil {
- return errors.New("unregistered during callback")
- }
- // If no headers were received, but we're expencting a checkpoint header, consider it that
- if len(headers) == 0 && p.syncDrop != nil {
- // Stop the timer either way, decide later to drop or not
- p.syncDrop.Stop()
- p.syncDrop = nil
- // If we're doing a fast (or snap) sync, we must enforce the checkpoint block to avoid
- // eclipse attacks. Unsynced nodes are welcome to connect after we're done
- // joining the network
- if atomic.LoadUint32(&h.fastSync) == 1 {
- peer.Log().Warn("Dropping unsynced node during sync", "addr", peer.RemoteAddr(), "type", peer.Name())
- return errors.New("unsynced node cannot serve sync")
- }
- }
- // Filter out any explicitly requested headers, deliver the rest to the downloader
- filter := len(headers) == 1
- if filter {
- // If it's a potential sync progress check, validate the content and advertised chain weight
- if p.syncDrop != nil && headers[0].Number.Uint64() == h.checkpointNumber {
- // Disable the sync drop timer
- p.syncDrop.Stop()
- p.syncDrop = nil
- // Validate the header and either drop the peer or continue
- if headers[0].Hash() != h.checkpointHash {
- return errors.New("checkpoint hash mismatch")
- }
- return nil
- }
- // Otherwise if it's a whitelisted block, validate against the set
- if want, ok := h.authorizationList[headers[0].Number.Uint64()]; ok {
- if hash := headers[0].Hash(); want != hash {
- peer.Log().Info("Whitelist mismatch, dropping peer", "number", headers[0].Number.Uint64(), "hash", hash, "want", want)
- return errors.New("whitelist block mismatch")
- }
- peer.Log().Debug("Whitelist block verified", "number", headers[0].Number.Uint64(), "hash", want)
- }
- // Irrelevant of the fork checks, send the header to the fetcher just in case
- headers = h.blockFetcher.FilterHeaders(peer.ID(), headers, time.Now())
- }
- if len(headers) > 0 || !filter {
- err := h.downloader.DeliverHeaders(peer.ID(), headers)
- if err != nil {
- log.Debug("Failed to deliver headers", "err", err)
- }
- }
- return nil
- }
- // handleBodies is invoked from a peer's message handler when it transmits a batch
- // of block bodies for the local node to process.
- func (h *ethHandler) handleBodies(peer *eth.Peer, txs [][]*types.Transaction, uncles [][]*types.Header) error {
- // Filter out any explicitly requested bodies, deliver the rest to the downloader
- filter := len(txs) > 0 || len(uncles) > 0
- if filter {
- txs, uncles = h.blockFetcher.FilterBodies(peer.ID(), txs, uncles, time.Now())
- }
- if len(txs) > 0 || len(uncles) > 0 || !filter {
- err := h.downloader.DeliverBodies(peer.ID(), txs, uncles)
- if err != nil {
- log.Debug("Failed to deliver bodies", "err", err)
- }
- }
- return nil
- }
- // handleBlockAnnounces is invoked from a peer's message handler when it transmits a
- // batch of block announcements for the local node to process.
- func (h *ethHandler) handleBlockAnnounces(peer *eth.Peer, hashes []common.Hash, numbers []uint64) error {
- // Schedule all the unknown hashes for retrieval
- var (
- unknownHashes = make([]common.Hash, 0, len(hashes))
- unknownNumbers = make([]uint64, 0, len(numbers))
- )
- for i := 0; i < len(hashes); i++ {
- if !h.chain.HasBlock(hashes[i], numbers[i]) {
- unknownHashes = append(unknownHashes, hashes[i])
- unknownNumbers = append(unknownNumbers, numbers[i])
- }
- }
- for i := 0; i < len(unknownHashes); i++ {
- h.blockFetcher.Notify(peer.ID(), unknownHashes[i], unknownNumbers[i], time.Now(), peer.RequestOneHeader, peer.RequestBodies)
- }
- return nil
- }
- // handleBlockBroadcast is invoked from a peer's message handler when it transmits a
- // block broadcast for the local node to process.
- func (h *ethHandler) handleBlockBroadcast(peer *eth.Peer, block *types.Block, td *big.Int) error {
- // Schedule the block for import
- h.blockFetcher.Enqueue(peer.ID(), block)
- // Assuming the block is importable by the peer, but possibly not yet done so,
- // calculate the head hash and TD that the peer truly must have.
- var (
- trueHead = block.ParentHash()
- trueTD = new(big.Int).Sub(td, block.Difficulty())
- )
- // Update the peer's total difficulty if better than the previous
- if _, td := peer.Head(); trueTD.Cmp(td) > 0 {
- peer.SetHead(trueHead, trueTD)
- h.chainSync.handlePeerEvent(peer)
- }
- return nil
- }
|