initial implementation for common ancestor approach

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)

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

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

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

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

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

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)