les: add checkpoint challenge for light client

les/client_handler.go

@@ -32,23 +32,30 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 )
 
+var checkpointChallengeTimeout = 15 * time.Second // Time allowance for a server to reply to the checkpoint progress challenge
+
 // clientHandler is responsible for receiving and processing all incoming server
 // responses.
 type clientHandler struct {
 	ulc        *ulc
+	clock      mclock.Clock
 	checkpoint *params.TrustedCheckpoint
 	fetcher    *lightFetcher
 	downloader *downloader.Downloader
 	backend    *LightEthereum
 
-	closeCh  chan struct{}
+	closeCh chan struct{}
-	wg       sync.WaitGroup // WaitGroup used to track all connected peers.
+	wg      sync.WaitGroup // WaitGroup used to track all connected peers.
-	syncDone func()         // Test hooks when syncing is done.
+
+	// Testing fields or hooks
+	ignoreCheckpoint bool   // Indicator whether ignore received checkpoint
+	syncDone         func() // Test hooks when syncing is done.
 }
 
 func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.TrustedCheckpoint, backend *LightEthereum) *clientHandler {
 	handler := &clientHandler{
 		checkpoint: checkpoint,
+		clock:      mclock.System{},
 		backend:    backend,
 		closeCh:    make(chan struct{}),
 	}
@@ -62,7 +69,7 @@ func newClientHandler(ulcServers []string, ulcFraction int, checkpoint *params.T
 	}
 	var height uint64
 	if checkpoint != nil {
-		height = (checkpoint.SectionIndex+1)*params.CHTFrequency - 1
+		height = (checkpoint.SectionIndex+1)*backend.iConfig.ChtSize - 1
 	}
 	handler.fetcher = newLightFetcher(handler)
 	handler.downloader = downloader.New(height, backend.chainDb, nil, backend.eventMux, nil, backend.blockchain, handler.removePeer)
@@ -133,6 +140,26 @@ func (h *clientHandler) handle(p *serverPeer) error {
 	if p.poolEntry != nil {
 		h.backend.serverPool.registered(p.poolEntry)
 	}
+	// If we have a trusted CHT, reject all peers below that (avoid light sync eclipse)
+	if h.checkpoint != nil {
+		// Request the peer's checkpoint header for chain height/weight validation
+		wrapPeer := &peerConnection{handler: h, peer: p}
+		if err := wrapPeer.RequestHeadersByNumber((h.checkpoint.SectionIndex+1)*h.backend.iConfig.ChtSize-1, 1, 0, false); err != nil {
+			return err
+		}
+		// Start a timer to disconnect if the peer doesn't reply in time
+		p.syncDrop = h.clock.AfterFunc(checkpointChallengeTimeout, func() {
+			p.Log().Warn("Checkpoint challenge timed out, dropping", "addr", p.RemoteAddr(), "type", p.Name())
+			h.removePeer(p.id)
+		})
+		// Make sure it's cleaned up if the peer dies off
+		defer func() {
+			if p.syncDrop != nil {
+				p.syncDrop.Stop()
+				p.syncDrop = nil
+			}
+		}()
+	}
 	// Mark the peer starts to be served.
 	atomic.StoreUint32(&p.serving, 1)
 	defer atomic.StoreUint32(&p.serving, 0)
@@ -205,6 +232,27 @@ func (h *clientHandler) handleMsg(p *serverPeer) error {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
 		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
+
+		// If no headers were received, but we're expecting a checkpoint header,
+		// drop the unsynced server.
+		if len(resp.Headers) == 0 && p.syncDrop != nil {
+			p.syncDrop.Stop()
+			p.syncDrop = nil
+			return errResp(ErrUselessPeer, "msg %v: %v", msg, err)
+		}
+		// If we are still waiting the checkpoint response.
+		if len(resp.Headers) == 1 && p.syncDrop != nil {
+			if !h.ignoreCheckpoint && resp.Headers[0].Number.Uint64() == (h.checkpoint.SectionIndex+1)*h.backend.iConfig.ChtSize-1 {
+				// First stop timer anyway.
+				p.syncDrop.Stop()
+				p.syncDrop = nil
+				if resp.Headers[0].Hash() != h.checkpoint.SectionHead {
+					return errResp(ErrUselessPeer, "msg %v: %v", msg, err)
+				}
+				return nil
+			}
+		}
+		// Deliver response header to concrete requester.
 		if h.fetcher.requestedID(resp.ReqID) {
 			h.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
 		} else {

les/handler_test.go

@@ -640,3 +640,51 @@ func TestStopResumeLes3(t *testing.T) {
 		}
 	}
 }
+
+func TestCheckpointChallengeLes2(t *testing.T) { testCheckpointChallenge(t, 2) }
+func TestCheckpointChallengeLes3(t *testing.T) { testCheckpointChallenge(t, 3) }
+
+func testCheckpointChallenge(t *testing.T, protocol int) {
+	config := light.TestServerIndexerConfig
+
+	waitIndexers := func(cIndexer, bIndexer, btIndexer *core.ChainIndexer) {
+		for {
+			cs, _, _ := cIndexer.Sections()
+			bts, _, _ := btIndexer.Sections()
+			if cs >= 1 && bts >= 1 {
+				break
+			}
+			time.Sleep(10 * time.Millisecond)
+		}
+	}
+	// Generate 512+4 blocks (totally 1 CHT sections)
+	server, client, tearDown := newClientServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, nil, 0, false, false)
+	defer tearDown()
+
+	s, _, head := server.chtIndexer.Sections()
+	cp := &params.TrustedCheckpoint{
+		SectionIndex: 0,
+		SectionHead:  head,
+		CHTRoot:      light.GetChtRoot(server.db, s-1, head),
+		BloomRoot:    light.GetBloomTrieRoot(server.db, s-1, head),
+	}
+	// Register the assembled checkpoint as hardcoded one.
+	client.handler.clock = &mclock.Simulated{}
+	client.handler.checkpoint = cp
+	client.handler.backend.blockchain.AddTrustedCheckpoint(cp)
+
+	// Create connected peer pair.
+	newTestPeerPair("peer", protocol, server.handler, client.handler)
+	client.handler.clock.(*mclock.Simulated).Run(checkpointChallengeTimeout)
+	if client.handler.backend.peers.len() != 1 {
+		t.Fatalf("Should pass checkpoint challenge")
+	}
+
+	client.handler.ignoreCheckpoint = true // Explicitly ignore all received headers, trigger timer
+	// Create connected peer pair.
+	newTestPeerPair("peer2", protocol, server.handler, client.handler)
+	client.handler.clock.(*mclock.Simulated).Run(checkpointChallengeTimeout)
+	if client.handler.backend.peers.len() != 1 {
+		t.Fatalf("Shouldn't pass checkpoint challenge")
+	}
+}

les/peer.go

@@ -357,6 +357,7 @@ type serverPeer struct {
 
 	poolEntry *poolEntry              // Statistic for server peer.
 	fcServer  *flowcontrol.ServerNode // Client side mirror token bucket.
+	syncDrop  mclock.Timer            // Timed connection dropper if sync progress isn't validated in time
 
 	// Statistics
 	errCount    int // Counter the invalid responses server has replied