From cc7c5b3fd997ad658c901f05ba7ddc639ea56423 Mon Sep 17 00:00:00 2001 From: Manav Darji Date: Mon, 2 May 2022 16:57:02 +0530 Subject: [PATCH] initial implementation for common ancestor approach --- consensus/bor/rest.go | 28 ++++++++ eth/backend.go | 55 +++++++++++++++ eth/downloader/downloader.go | 107 +++++++++++++++++++++++++---- eth/handler.go | 23 ++++--- eth/handler_bor.go | 50 ++++++++++++++ eth/handler_eth_test.go | 22 ++++++ tests/bor/mocks/IHeimdallClient.go | 23 +++++++ 7 files changed, 287 insertions(+), 21 deletions(-) create mode 100644 eth/handler_bor.go diff --git a/consensus/bor/rest.go b/consensus/bor/rest.go index 6460b75a3c..627262ffe6 100644 --- a/consensus/bor/rest.go +++ b/consensus/bor/rest.go @@ -4,11 +4,13 @@ import ( "encoding/json" "fmt" "io/ioutil" + "math/big" "net/http" "net/url" "sort" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" ) @@ -23,10 +25,21 @@ type ResponseWithHeight struct { Result json.RawMessage `json:"result"` } +// Checkpoint defines a response object type of bor checkpoint +type Checkpoint struct { + Proposer common.Address `json:"proposer"` + StartBlock *big.Int `json:"start_block"` + EndBlock *big.Int `json:"end_block"` + RootHash common.Hash `json:"root_hash"` + BorChainID string `json:"bor_chain_id"` + Timestamp uint64 `json:"timestamp"` +} + type IHeimdallClient interface { Fetch(path string, query string) (*ResponseWithHeight, error) FetchWithRetry(path string, query string) (*ResponseWithHeight, error) FetchStateSyncEvents(fromID uint64, to int64) ([]*EventRecordWithTime, error) + FetchLatestCheckpoint() (*Checkpoint, error) Close() } @@ -76,6 +89,21 @@ func (h *HeimdallClient) FetchStateSyncEvents(fromID uint64, to int64) ([]*Event return eventRecords, nil } +// FetchLatestCheckpoint fetches the latest bor submitted checkpoint from heimdall +func (h *HeimdallClient) FetchLatestCheckpoint() (*Checkpoint, error) { + var checkpoint Checkpoint + + response, err := h.Fetch("/checkpoints/latest", "") + if err != nil { + return nil, err + } + if err := json.Unmarshal(response.Result, &checkpoint); err != nil { + return nil, err + } + + return &checkpoint, nil +} + // Fetch fetches response from heimdall func (h *HeimdallClient) Fetch(rawPath string, rawQuery string) (*ResponseWithHeight, error) { u, err := url.Parse(h.urlString) diff --git a/eth/backend.go b/eth/backend.go index 05d6c5c927..994a7d473f 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -100,6 +100,8 @@ type Ethereum struct { lock sync.RWMutex // Protects the variadic fields (e.g. gas price and etherbase) + closeCh chan struct{} // Channel to signal the background processes to exit + shutdownTracker *shutdowncheck.ShutdownTracker // Tracks if and when the node has shutdown ungracefully } @@ -161,6 +163,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { bloomRequests: make(chan chan *bloombits.Retrieval), bloomIndexer: core.NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms), p2pServer: stack.Server(), + closeCh: make(chan struct{}), shutdownTracker: shutdowncheck.NewShutdownTracker(chainDb), } @@ -252,6 +255,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) { BloomCache: uint64(cacheLimit), EventMux: eth.eventMux, Checkpoint: checkpoint, + EthAPI: ethAPI, PeerRequiredBlocks: config.PeerRequiredBlocks, }); err != nil { return nil, err @@ -584,6 +588,54 @@ func (s *Ethereum) Start() error { } // Start the networking layer and the light server if requested s.handler.Start(maxPeers) + + go s.startCheckpointWhitelistService() + return nil +} + +// StartCheckpointWhitelistService starts the goroutine to fetch checkpoints and update the +// checkpoint whitelist map. +func (s *Ethereum) startCheckpointWhitelistService() { + every := time.Duration(100) * time.Second + ticker := time.NewTicker(every) + defer ticker.Stop() + +Loop: + for { + select { + case <-ticker.C: + err := s.handleWhitelistCheckpoint() + if err != nil { + log.Warn(err.Error()) + } + case <-s.closeCh: + break Loop + } + } +} + +// handleWhitelistCheckpoint handles the checkpoint whitelist mechanism. +func (s *Ethereum) handleWhitelistCheckpoint() error { + var m sync.Mutex + + ethHandler := (*ethHandler)(s.handler) + + if !ethHandler.chain.Engine().(*bor.Bor).WithoutHeimdall { + endBlockNum, endBlockHash, err := ethHandler.fetchWhitelistCheckpoint() + if err != nil { + return err + } + + m.Lock() + // Update the checkpoint whitelist map. + ethHandler.downloader.EnqueueCheckpointWhitelist(endBlockNum, endBlockHash) + // If size of checkpoint whitelist map is greater than 10, remove the oldest entry. + if len(ethHandler.downloader.GetCheckpointWhitelist()) > 10 { + ethHandler.downloader.DequeueCheckpointWhitelist() + } + m.Unlock() + } + return nil } @@ -599,6 +651,9 @@ func (s *Ethereum) Stop() error { s.bloomIndexer.Close() close(s.closeBloomHandler) + // Close all bg processes + close(s.closeCh) + // closing consensus engine first, as miner has deps on it s.engine.Close() s.txPool.Stop() diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 1e68746f97..b2d70906aa 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -78,6 +78,7 @@ var ( errCanceled = errors.New("syncing canceled (requested)") errTooOld = errors.New("peer's protocol version too old") errNoAncestorFound = errors.New("no common ancestor found") + errCheckpointMismatch = errors.New("checkpoint mismatch") ErrMergeTransition = errors.New("legacy sync reached the merge") ) @@ -143,6 +144,10 @@ type Downloader struct { quitCh chan struct{} // Quit channel to signal termination quitLock sync.Mutex // Lock to prevent double closes + // Checkpoint whitelist + checkpointWhitelist map[uint64]common.Hash // Checkpoint whitelist, populated by reaching out to heimdall + checkpointOrder []uint64 // Checkpoint order, populated by reaching out to heimdall + // Testing hooks syncInitHook func(uint64, uint64) // Method to call upon initiating a new sync run bodyFetchHook func([]*types.Header) // Method to call upon starting a block body fetch @@ -209,18 +214,20 @@ func New(checkpoint uint64, stateDb ethdb.Database, mux *event.TypeMux, chain Bl lightchain = chain } dl := &Downloader{ - stateDB: stateDb, - mux: mux, - checkpoint: checkpoint, - queue: newQueue(blockCacheMaxItems, blockCacheInitialItems), - peers: newPeerSet(), - blockchain: chain, - lightchain: lightchain, - dropPeer: dropPeer, - headerProcCh: make(chan *headerTask, 1), - quitCh: make(chan struct{}), - SnapSyncer: snap.NewSyncer(stateDb), - stateSyncStart: make(chan *stateSync), + stateDB: stateDb, + mux: mux, + checkpoint: checkpoint, + queue: newQueue(blockCacheMaxItems, blockCacheInitialItems), + peers: newPeerSet(), + blockchain: chain, + lightchain: lightchain, + dropPeer: dropPeer, + headerProcCh: make(chan *headerTask, 1), + quitCh: make(chan struct{}), + SnapSyncer: snap.NewSyncer(stateDb), + stateSyncStart: make(chan *stateSync), + checkpointWhitelist: make(map[uint64]common.Hash), + checkpointOrder: make([]uint64, 0), } dl.skeleton = newSkeleton(stateDb, dl.peers, dropPeer, newBeaconBackfiller(dl, success)) @@ -332,6 +339,10 @@ func (d *Downloader) LegacySync(id string, head common.Hash, td, ttd *big.Int, m case nil, errBusy, errCanceled: return err } + if errors.Is(err, errCheckpointMismatch) { + // TODO: what better can be done here? + log.Warn("Mismatch in last checkpointed block", "peer", id, "err", err) + } if errors.Is(err, errInvalidChain) || errors.Is(err, errBadPeer) || errors.Is(err, errTimeout) || errors.Is(err, errStallingPeer) || errors.Is(err, errUnsyncedPeer) || errors.Is(err, errEmptyHeaderSet) || errors.Is(err, errPeersUnavailable) || errors.Is(err, errTooOld) || errors.Is(err, errInvalidAncestor) { @@ -764,12 +775,51 @@ func calculateRequestSpan(remoteHeight, localHeight uint64) (int64, int, int, ui return int64(from), count, span - 1, uint64(max) } +// isValidChain checks if the chain we're about to receive from this peer is valid or not +// in terms of reorgs. We won't reorg beyond the last bor checkpoint submitted to mainchain. +func (d *Downloader) isValidChain(p *peerConnection, remoteHeader *types.Header) (bool, error) { + // We want to validate the chain by comparing the last checkpointed block + // we're storing in `checkpointWhitelist` with the peer's block. + + // Check for availaibility of the last checkpointed block. + // This can be also be empty if our heimdall is not responsing + // or we're running without it. + if len(d.checkpointWhitelist) <= 0 { + // worst case, we don't have the checkpoints in memory + return true, nil + } + + // Fetch the last checkpoint entry + lastCheckpointBlockNum := d.checkpointOrder[len(d.checkpointOrder)-1] + lastCheckpointBlockHash := d.checkpointWhitelist[lastCheckpointBlockNum] + + headers, hashes, err := d.fetchHeadersByNumber(p, lastCheckpointBlockNum, 1, 0, false) + if err != nil || len(headers) == 0 { + // TODO: what better can be done here? + return true, nil + } + reqBlockNum := headers[0].Number.Uint64() + reqBlockHash := hashes[0] + + // Check against the checkpointed blocks + if reqBlockNum == lastCheckpointBlockNum && reqBlockHash == lastCheckpointBlockHash { + return true, nil + } + + return false, errCheckpointMismatch +} + // findAncestor tries to locate the common ancestor link of the local chain and // a remote peers blockchain. In the general case when our node was in sync and // on the correct chain, checking the top N links should already get us a match. // In the rare scenario when we ended up on a long reorganisation (i.e. none of // the head links match), we do a binary search to find the common ancestor. func (d *Downloader) findAncestor(p *peerConnection, remoteHeader *types.Header) (uint64, error) { + // Check the validity of chain to be downloaded + if _, err := d.isValidChain(p, remoteHeader); errors.Is(err, errCheckpointMismatch) { + return 0, err + } + // Figure out the valid ancestor range to prevent rewrite attacks var ( floor = int64(-1) @@ -1755,3 +1805,36 @@ func (d *Downloader) DeliverSnapPacket(peer *snap.Peer, packet snap.Packet) erro return fmt.Errorf("unexpected snap packet type: %T", packet) } } + +// Helper functions for checkpoint whitelist service + +// PurgeWhitelistMap purges data from checkpoint whitelist map +func (d *Downloader) PurgeWhitelistMap() error { + for k := range d.checkpointWhitelist { + delete(d.checkpointWhitelist, k) + } + return nil +} + +// EnqueueWhitelistBlock enqueues blockNumber, blockHash to the checkpoint whitelist map +func (d *Downloader) EnqueueCheckpointWhitelist(key uint64, val common.Hash) { + if _, ok := d.checkpointWhitelist[key]; !ok { + log.Debug("Enqueing new checkpoint whitelist", "block number", key, "block hash", val) + d.checkpointWhitelist[key] = val + d.checkpointOrder = append(d.checkpointOrder, key) + } +} + +// DequeueWhitelistBlock dequeues block, blockhash from the checkpoint whitelist map +func (d *Downloader) DequeueCheckpointWhitelist() { + if len(d.checkpointOrder) > 0 { + log.Debug("Dequeing checkpoint whitelist", "block number", d.checkpointOrder[0], "block hash", d.checkpointWhitelist[d.checkpointOrder[0]]) + delete(d.checkpointWhitelist, d.checkpointOrder[0]) + d.checkpointOrder = d.checkpointOrder[1:] + } +} + +// GetCheckpointWhitelist returns the checkpoints whitelisted. +func (d *Downloader) GetCheckpointWhitelist() map[uint64]common.Hash { + return d.checkpointWhitelist +} diff --git a/eth/handler.go b/eth/handler.go index 40edfa2d17..fa0852d69e 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -36,6 +36,7 @@ import ( "github.com/ethereum/go-ethereum/eth/protocols/snap" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/internal/ethapi" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/params" @@ -77,15 +78,16 @@ type txPool interface { // handlerConfig is the collection of initialization parameters to create a full // node network handler. type handlerConfig struct { - Database ethdb.Database // Database for direct sync insertions - Chain *core.BlockChain // Blockchain to serve data from - TxPool txPool // Transaction pool to propagate from - Merger *consensus.Merger // The manager for eth1/2 transition - Network uint64 // Network identifier to adfvertise - Sync downloader.SyncMode // Whether to snap or full sync - BloomCache uint64 // Megabytes to alloc for snap sync bloom - EventMux *event.TypeMux // Legacy event mux, deprecate for `feed` - Checkpoint *params.TrustedCheckpoint // Hard coded checkpoint for sync challenges + Database ethdb.Database // Database for direct sync insertions + Chain *core.BlockChain // Blockchain to serve data from + TxPool txPool // Transaction pool to propagate from + Merger *consensus.Merger // The manager for eth1/2 transition + Network uint64 // Network identifier to adfvertise + Sync downloader.SyncMode // Whether to snap or full sync + BloomCache uint64 // Megabytes to alloc for snap sync bloom + EventMux *event.TypeMux // Legacy event mux, deprecate for `feed` + Checkpoint *params.TrustedCheckpoint // Hard coded checkpoint for sync challenges + EthAPI *ethapi.PublicBlockChainAPI // EthAPI to interact PeerRequiredBlocks map[uint64]common.Hash // Hard coded map of required block hashes for sync challenges } @@ -111,6 +113,8 @@ type handler struct { peers *peerSet merger *consensus.Merger + ethAPI *ethapi.PublicBlockChainAPI // EthAPI to interact + eventMux *event.TypeMux txsCh chan core.NewTxsEvent txsSub event.Subscription @@ -141,6 +145,7 @@ func newHandler(config *handlerConfig) (*handler, error) { chain: config.Chain, peers: newPeerSet(), merger: config.Merger, + ethAPI: config.EthAPI, peerRequiredBlocks: config.PeerRequiredBlocks, quitSync: make(chan struct{}), } diff --git a/eth/handler_bor.go b/eth/handler_bor.go new file mode 100644 index 0000000000..b87c07f551 --- /dev/null +++ b/eth/handler_bor.go @@ -0,0 +1,50 @@ +package eth + +import ( + "context" + "fmt" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/consensus/bor" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rpc" +) + +// fetchWhitelistCheckpoint fetched the latest checkpoint from it's local heimdall +// and verifies the data against bor data. +func (h *ethHandler) fetchWhitelistCheckpoint() (uint64, common.Hash, error) { + // check for checkpoint whitelisting: bor + checkpoint, err := h.chain.Engine().(*bor.Bor).HeimdallClient.FetchLatestCheckpoint() + if err != nil { + log.Debug("Failed to fetch latest checkpoint for whitelisting") + return 0, common.Hash{}, fmt.Errorf("failed to fetch latest checkpoint") + } + + // check if we have the checkpoint blocks + head := h.ethAPI.BlockNumber() + if head < hexutil.Uint64(checkpoint.EndBlock.Uint64()) { + log.Debug("Head block behing checkpoint block", "head", head, "checkpoint end block", checkpoint.EndBlock) + return 0, common.Hash{}, fmt.Errorf("missing checkpoint blocks") + } + + // verify the root hash of checkpoint + roothash, err := h.ethAPI.GetRootHash(context.Background(), checkpoint.StartBlock.Uint64(), checkpoint.EndBlock.Uint64()) + if err != nil { + log.Debug("Failed to get root hash of checkpoint while whitelisting") + return 0, common.Hash{}, fmt.Errorf("failed to get local root hash") + } + if roothash != checkpoint.RootHash.String()[2:] { + log.Warn("Checkpoint root hash mismatch while whitelisting", "expected", checkpoint.RootHash.String()[2:], "got", roothash) + return 0, common.Hash{}, fmt.Errorf("checkpoint roothash mismatch") + } + + // fetch the end checkpoint block hash + block, err := h.ethAPI.GetBlockByNumber(context.Background(), rpc.BlockNumber(checkpoint.EndBlock.Uint64()), false) + if err != nil { + log.Debug("Failed to get end block hash of checkpoint while whitelisting") + return 0, common.Hash{}, fmt.Errorf("failed to get end block") + } + hash := fmt.Sprintf("%v", block["hash"]) + return checkpoint.EndBlock.Uint64(), common.HexToHash(hash), nil +} diff --git a/eth/handler_eth_test.go b/eth/handler_eth_test.go index 7d5027ae77..ec1329b059 100644 --- a/eth/handler_eth_test.go +++ b/eth/handler_eth_test.go @@ -39,6 +39,7 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/params" "github.com/ethereum/go-ethereum/rlp" + "gotest.tools/assert" ) // testEthHandler is a mock event handler to listen for inbound network requests @@ -746,3 +747,24 @@ func testBroadcastMalformedBlock(t *testing.T, protocol uint) { } } } + +// TestWhitelistCheckpoint checks the checkpoint whitelist map queue mechanism +func TestWhitelistCheckpoint(t *testing.T) { + t.Parallel() + + testHandler := newTestHandler() + defer testHandler.close() + + ethHandler := (*ethHandler)(testHandler.handler) + for i := 0; i < 10; i++ { + ethHandler.downloader.EnqueueCheckpointWhitelist(uint64(i), common.Hash{}) + } + + assert.Equal(t, len(ethHandler.downloader.GetCheckpointWhitelist()), 10, "expected 10 items in whitelist") + + ethHandler.downloader.EnqueueCheckpointWhitelist(11, common.Hash{}) + ethHandler.downloader.DequeueCheckpointWhitelist() + + assert.Equal(t, len(ethHandler.downloader.GetCheckpointWhitelist()), 10, "expected 10 items in whitelist") + +} diff --git a/tests/bor/mocks/IHeimdallClient.go b/tests/bor/mocks/IHeimdallClient.go index ac89487d3b..3ee599bb2e 100644 --- a/tests/bor/mocks/IHeimdallClient.go +++ b/tests/bor/mocks/IHeimdallClient.go @@ -40,6 +40,29 @@ func (_m *IHeimdallClient) Fetch(path string, query string) (*bor.ResponseWithHe return r0, r1 } +// FetchLatestCheckpoint provides a mock function with given fields: +func (_m *IHeimdallClient) FetchLatestCheckpoint() (*bor.Checkpoint, error) { + ret := _m.Called() + + var r0 *bor.Checkpoint + if rf, ok := ret.Get(0).(func() *bor.Checkpoint); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*bor.Checkpoint) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func() error); ok { + r1 = rf() + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // FetchStateSyncEvents provides a mock function with given fields: fromID, to func (_m *IHeimdallClient) FetchStateSyncEvents(fromID uint64, to int64) ([]*bor.EventRecordWithTime, error) { ret := _m.Called(fromID, to)