gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

eth: implement eth66 (#22241)

Browse Source 
* eth/protocols/eth: split up the eth protocol handlers

* eth/protocols/eth: define eth-66 protocol messages

* eth/protocols/eth: poc implement getblockheaders on eth/66

* eth/protocols/eth: implement remaining eth-66 handlers

* eth/protocols: define handler map for eth 66

* eth/downloader: use protocol constants from eth package

* eth/protocols/eth: add ETH66 capability

* eth/downloader: tests for eth66

* eth/downloader: fix error in tests

* eth/protocols/eth: use eth66 for outgoing requests

* eth/protocols/eth: remove unused error type

* eth/protocols/eth: define protocol length

* eth/protocols/eth: fix pooled tx over eth66

* protocols/eth/handlers: revert behavioural change which caused tests to fail

* eth/downloader: fix failing test

* eth/protocols/eth: add testcases + fix flaw with header requests

* eth/protocols: change comments

* eth/protocols/eth: review fixes + fixed flaw in RequestOneHeader

* eth/protocols: documentation

* eth/protocols/eth: review concerns about types
This commit is contained in:
Martin Holst Swende
2021-02-18 17:54:29 +01:00
committed by GitHub
parent b1835b3855
commit 6ec1561044
8 changed files with 1121 additions and 396 deletions
Download Patch File Download Diff File Expand all files Collapse all files

401
eth/protocols/eth/handler.go
View File

@@ -17,19 +17,17 @@
package eth
import (
"encoding/json"
"fmt"
"math/big"
"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/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/p2p/enode"
"github.com/ethereum/go-ethereum/p2p/enr"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/trie"
)
@@ -166,6 +164,64 @@ func Handle(backend Backend, peer *Peer) error {
}
}
type msgHandler func(backend Backend, msg Decoder, peer *Peer) error
type Decoder interface {
Decode(val interface{}) error
Time() time.Time
}
var eth64 = map[uint64]msgHandler{
GetBlockHeadersMsg: handleGetBlockHeaders,
BlockHeadersMsg: handleBlockHeaders,
GetBlockBodiesMsg: handleGetBlockBodies,
BlockBodiesMsg: handleBlockBodies,
GetNodeDataMsg: handleGetNodeData,
NodeDataMsg: handleNodeData,
GetReceiptsMsg: handleGetReceipts,
ReceiptsMsg: handleReceipts,
NewBlockHashesMsg: handleNewBlockhashes,
NewBlockMsg: handleNewBlock,
TransactionsMsg: handleTransactions,
}
var eth65 = map[uint64]msgHandler{
// old 64 messages
GetBlockHeadersMsg: handleGetBlockHeaders,
BlockHeadersMsg: handleBlockHeaders,
GetBlockBodiesMsg: handleGetBlockBodies,
BlockBodiesMsg: handleBlockBodies,
GetNodeDataMsg: handleGetNodeData,
NodeDataMsg: handleNodeData,
GetReceiptsMsg: handleGetReceipts,
ReceiptsMsg: handleReceipts,
NewBlockHashesMsg: handleNewBlockhashes,
NewBlockMsg: handleNewBlock,
TransactionsMsg: handleTransactions,
// New eth65 messages
NewPooledTransactionHashesMsg: handleNewPooledTransactionHashes,
GetPooledTransactionsMsg: handleGetPooledTransactions,
PooledTransactionsMsg: handlePooledTransactions,
}
var eth66 = map[uint64]msgHandler{
// eth64 announcement messages (no id)
NewBlockHashesMsg: handleNewBlockhashes,
NewBlockMsg: handleNewBlock,
TransactionsMsg: handleTransactions,
// eth65 announcement messages (no id)
NewPooledTransactionHashesMsg: handleNewPooledTransactionHashes,
// eth66 messages with request-id
GetBlockHeadersMsg: handleGetBlockHeaders66,
BlockHeadersMsg: handleBlockHeaders66,
GetBlockBodiesMsg: handleGetBlockBodies66,
BlockBodiesMsg: handleBlockBodies66,
GetNodeDataMsg: handleGetNodeData66,
NodeDataMsg: handleNodeData66,
GetReceiptsMsg: handleGetReceipts66,
ReceiptsMsg: handleReceipts66,
GetPooledTransactionsMsg: handleGetPooledTransactions66,
PooledTransactionsMsg: handlePooledTransactions66,
}
// handleMessage is invoked whenever an inbound message is received from a remote
// peer. The remote connection is torn down upon returning any error.
func handleMessage(backend Backend, peer *Peer) error {
@@ -179,334 +235,15 @@ func handleMessage(backend Backend, peer *Peer) error {
}
defer msg.Discard()
// Handle the message depending on its contents
switch {
case msg.Code == StatusMsg:
// Status messages should never arrive after the handshake
return fmt.Errorf("%w: uncontrolled status message", errExtraStatusMsg)
// Block header query, collect the requested headers and reply
case msg.Code == GetBlockHeadersMsg:
// Decode the complex header query
var query GetBlockHeadersPacket
if err := msg.Decode(&query); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
hashMode := query.Origin.Hash != (common.Hash{})
first := true
maxNonCanonical := uint64(100)
// Gather headers until the fetch or network limits is reached
var (
bytes common.StorageSize
headers []*types.Header
unknown bool
lookups int
)
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit &&
len(headers) < maxHeadersServe && lookups < 2*maxHeadersServe {
lookups++
// Retrieve the next header satisfying the query
var origin *types.Header
if hashMode {
if first {
first = false
origin = backend.Chain().GetHeaderByHash(query.Origin.Hash)
if origin != nil {
query.Origin.Number = origin.Number.Uint64()
}
} else {
origin = backend.Chain().GetHeader(query.Origin.Hash, query.Origin.Number)
}
} else {
origin = backend.Chain().GetHeaderByNumber(query.Origin.Number)
}
if origin == nil {
break
}
headers = append(headers, origin)
bytes += estHeaderSize
// Advance to the next header of the query
switch {
case hashMode && query.Reverse:
// Hash based traversal towards the genesis block
ancestor := query.Skip + 1
if ancestor == 0 {
unknown = true
} else {
query.Origin.Hash, query.Origin.Number = backend.Chain().GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical)
unknown = (query.Origin.Hash == common.Hash{})
}
case hashMode && !query.Reverse:
// Hash based traversal towards the leaf block
var (
current = origin.Number.Uint64()
next = current + query.Skip + 1
)
if next <= current {
infos, _ := json.MarshalIndent(peer.Peer.Info(), "", " ")
peer.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
unknown = true
} else {
if header := backend.Chain().GetHeaderByNumber(next); header != nil {
nextHash := header.Hash()
expOldHash, _ := backend.Chain().GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical)
if expOldHash == query.Origin.Hash {
query.Origin.Hash, query.Origin.Number = nextHash, next
} else {
unknown = true
}
} else {
unknown = true
}
}
case query.Reverse:
// Number based traversal towards the genesis block
if query.Origin.Number >= query.Skip+1 {
query.Origin.Number -= query.Skip + 1
} else {
unknown = true
}
case !query.Reverse:
// Number based traversal towards the leaf block
query.Origin.Number += query.Skip + 1
}
}
return peer.SendBlockHeaders(headers)
case msg.Code == BlockHeadersMsg:
// A batch of headers arrived to one of our previous requests
res := new(BlockHeadersPacket)
if err := msg.Decode(res); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
return backend.Handle(peer, res)
case msg.Code == GetBlockBodiesMsg:
// Decode the block body retrieval message
var query GetBlockBodiesPacket
if err := msg.Decode(&query); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Gather blocks until the fetch or network limits is reached
var (
bytes int
bodies []rlp.RawValue
)
for lookups, hash := range query {
if bytes >= softResponseLimit || len(bodies) >= maxBodiesServe ||
lookups >= 2*maxBodiesServe {
break
}
if data := backend.Chain().GetBodyRLP(hash); len(data) != 0 {
bodies = append(bodies, data)
bytes += len(data)
}
}
return peer.SendBlockBodiesRLP(bodies)
case msg.Code == BlockBodiesMsg:
// A batch of block bodies arrived to one of our previous requests
res := new(BlockBodiesPacket)
if err := msg.Decode(res); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
return backend.Handle(peer, res)
case msg.Code == GetNodeDataMsg:
// Decode the trie node data retrieval message
var query GetNodeDataPacket
if err := msg.Decode(&query); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Gather state data until the fetch or network limits is reached
var (
bytes int
nodes [][]byte
)
for lookups, hash := range query {
if bytes >= softResponseLimit || len(nodes) >= maxNodeDataServe ||
lookups >= 2*maxNodeDataServe {
break
}
// Retrieve the requested state entry
if bloom := backend.StateBloom(); bloom != nil && !bloom.Contains(hash[:]) {
// Only lookup the trie node if there's chance that we actually have it
continue
}
entry, err := backend.Chain().TrieNode(hash)
if len(entry) == 0 || err != nil {
// Read the contract code with prefix only to save unnecessary lookups.
entry, err = backend.Chain().ContractCodeWithPrefix(hash)
}
if err == nil && len(entry) > 0 {
nodes = append(nodes, entry)
bytes += len(entry)
}
}
return peer.SendNodeData(nodes)
case msg.Code == NodeDataMsg:
// A batch of node state data arrived to one of our previous requests
res := new(NodeDataPacket)
if err := msg.Decode(res); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
return backend.Handle(peer, res)
case msg.Code == GetReceiptsMsg:
// Decode the block receipts retrieval message
var query GetReceiptsPacket
if err := msg.Decode(&query); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Gather state data until the fetch or network limits is reached
var (
bytes int
receipts []rlp.RawValue
)
for lookups, hash := range query {
if bytes >= softResponseLimit || len(receipts) >= maxReceiptsServe ||
lookups >= 2*maxReceiptsServe {
break
}
// Retrieve the requested block's receipts
results := backend.Chain().GetReceiptsByHash(hash)
if results == nil {
if header := backend.Chain().GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
continue
}
}
// If known, encode and queue for response packet
if encoded, err := rlp.EncodeToBytes(results); err != nil {
log.Error("Failed to encode receipt", "err", err)
} else {
receipts = append(receipts, encoded)
bytes += len(encoded)
}
}
return peer.SendReceiptsRLP(receipts)
case msg.Code == ReceiptsMsg:
// A batch of receipts arrived to one of our previous requests
res := new(ReceiptsPacket)
if err := msg.Decode(res); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
return backend.Handle(peer, res)
case msg.Code == NewBlockHashesMsg:
// A batch of new block announcements just arrived
ann := new(NewBlockHashesPacket)
if err := msg.Decode(ann); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Mark the hashes as present at the remote node
for _, block := range *ann {
peer.markBlock(block.Hash)
}
// Deliver them all to the backend for queuing
return backend.Handle(peer, ann)
case msg.Code == NewBlockMsg:
// Retrieve and decode the propagated block
ann := new(NewBlockPacket)
if err := msg.Decode(ann); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
if hash := types.CalcUncleHash(ann.Block.Uncles()); hash != ann.Block.UncleHash() {
log.Warn("Propagated block has invalid uncles", "have", hash, "exp", ann.Block.UncleHash())
break // TODO(karalabe): return error eventually, but wait a few releases
}
if hash := types.DeriveSha(ann.Block.Transactions(), trie.NewStackTrie(nil)); hash != ann.Block.TxHash() {
log.Warn("Propagated block has invalid body", "have", hash, "exp", ann.Block.TxHash())
break // TODO(karalabe): return error eventually, but wait a few releases
}
if err := ann.sanityCheck(); err != nil {
return err
}
ann.Block.ReceivedAt = msg.ReceivedAt
ann.Block.ReceivedFrom = peer
// Mark the peer as owning the block
peer.markBlock(ann.Block.Hash())
return backend.Handle(peer, ann)
case msg.Code == NewPooledTransactionHashesMsg && peer.version >= ETH65:
// New transaction announcement arrived, make sure we have
// a valid and fresh chain to handle them
if !backend.AcceptTxs() {
break
}
ann := new(NewPooledTransactionHashesPacket)
if err := msg.Decode(ann); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Schedule all the unknown hashes for retrieval
for _, hash := range *ann {
peer.markTransaction(hash)
}
return backend.Handle(peer, ann)
case msg.Code == GetPooledTransactionsMsg && peer.version >= ETH65:
// Decode the pooled transactions retrieval message
var query GetPooledTransactionsPacket
if err := msg.Decode(&query); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
// Gather transactions until the fetch or network limits is reached
var (
bytes int
hashes []common.Hash
txs []rlp.RawValue
)
for _, hash := range query {
if bytes >= softResponseLimit {
break
}
// Retrieve the requested transaction, skipping if unknown to us
tx := backend.TxPool().Get(hash)
if tx == nil {
continue
}
// If known, encode and queue for response packet
if encoded, err := rlp.EncodeToBytes(tx); err != nil {
log.Error("Failed to encode transaction", "err", err)
} else {
hashes = append(hashes, hash)
txs = append(txs, encoded)
bytes += len(encoded)
}
}
return peer.SendPooledTransactionsRLP(hashes, txs)
case msg.Code == TransactionsMsg || (msg.Code == PooledTransactionsMsg && peer.version >= ETH65):
// Transactions arrived, make sure we have a valid and fresh chain to handle them
if !backend.AcceptTxs() {
break
}
// Transactions can be processed, parse all of them and deliver to the pool
var txs []*types.Transaction
if err := msg.Decode(&txs); err != nil {
return fmt.Errorf("%w: message %v: %v", errDecode, msg, err)
}
for i, tx := range txs {
// Validate and mark the remote transaction
if tx == nil {
return fmt.Errorf("%w: transaction %d is nil", errDecode, i)
}
peer.markTransaction(tx.Hash())
}
if msg.Code == PooledTransactionsMsg {
return backend.Handle(peer, (*PooledTransactionsPacket)(&txs))
}
return backend.Handle(peer, (*TransactionsPacket)(&txs))
default:
return fmt.Errorf("%w: %v", errInvalidMsgCode, msg.Code)
var handlers = eth64
if peer.Version() == ETH65 {
handlers = eth65
} else if peer.Version() >= ETH66 {
handlers = eth66
}
return nil
if handler := handlers[msg.Code]; handler != nil {
return handler(backend, msg, peer)
}
return fmt.Errorf("%w: %v", errInvalidMsgCode, msg.Code)
}