eth: implemented peer consensus based witness verification

eth/backend.go

@@ -411,13 +411,15 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		EventMux:                eth.eventMux,
 		RequiredBlocks:          config.RequiredBlocks,
 		EthAPI:                  blockChainAPI,
+		gasCeil:                 config.Miner.GasCeil,
 		checker:                 checker,
 		enableBlockTracking:     eth.config.EnableBlockTracking,
 		txAnnouncementOnly:      eth.p2pServer.TxAnnouncementOnly,
 		witnessProtocol:         eth.config.WitnessProtocol,
 		syncWithWitnesses:       eth.config.SyncWithWitnesses,
 		syncAndProduceWitnesses: eth.config.SyncAndProduceWitnesses,
 		fastForwardThreshold:    config.FastForwardThreshold,
+		p2pServer:               eth.p2pServer,
 	}); err != nil {
 		return nil, err
 	}

eth/fetcher/block_fetcher.go

@@ -109,6 +109,7 @@
 
 // peerDropFn is a callback type for dropping a peer detected as malicious.
 type peerDropFn func(id string)
+type peerJailFn func(id string) // Function to jail a peer by peer ID (string)
 
 // blockAnnounce is the hash notification of the availability of a new block in the
 // network.
@@ -235,6 +236,7 @@
 	insertHeaders  headersInsertFn    // Injects a batch of headers into the chain
 	insertChain    chainInsertFn      // Injects a batch of blocks into the chain
 	dropPeer       peerDropFn         // Drops a peer for misbehaving
+	jailPeer       peerJailFn         // Jails a peer to prevent reconnection (optional, can be nil)
 
 	// Testing hooks
 	announceChangeHook func(common.Hash, bool)           // Method to call upon adding or deleting a hash from the blockAnnounce list
@@ -249,7 +251,7 @@
 }
 
 // NewBlockFetcher creates a block fetcher to retrieve blocks based on hash announcements.
-func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn, enableBlockTracking bool, requireWitness bool) *BlockFetcher {
+func NewBlockFetcher(light bool, getHeader HeaderRetrievalFn, getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertHeaders headersInsertFn, insertChain chainInsertFn, dropPeer peerDropFn, jailPeer peerJailFn, enableBlockTracking bool, requireWitness bool, gasCeil uint64) *BlockFetcher {
 	f := &BlockFetcher{
 		light:               light,
 		notify:              make(chan *blockAnnounce),
@@ -275,6 +277,7 @@
 		insertHeaders:       insertHeaders,
 		insertChain:         insertChain,
 		dropPeer:            dropPeer,
+		jailPeer:            jailPeer,
 		enableBlockTracking: enableBlockTracking,
 		requireWitness:      requireWitness,
 	}
@@ -283,10 +286,12 @@
 	f.wm = newWitnessManager(
 		f.quit,
 		f.dropPeer,
+		f.jailPeer,
 		f.enqueueCh,
 		f.getBlock,
 		f.getHeader,
 		f.chainHeight,
+		gasCeil,
 	)
 
 	return f
@@ -1297,3 +1302,8 @@
 		f.wm.forget(hash)
 	}
 }
+
+// GetWitnessManager returns the witness manager for external access
+func (f *BlockFetcher) GetWitnessManager() *witnessManager {
+	return f.wm
+}

eth/fetcher/block_fetcher_race_test.go

@@ -50,8 +50,10 @@ func TestBlockFetcherConcurrentMapAccess(t *testing.T) {
 		insertHeaders,
 		insertChain,
 		dropPeer,
+		nil,   // no peer jailing
 		false, // no block tracking
 		false, // no witness requirement
+		0,     // no gas ceiling
 	)
 
 	// Start the fetcher
@@ -245,10 +247,12 @@ func TestWitnessManagerConcurrentAccess(t *testing.T) {
 	manager := newWitnessManager(
 		quit,
 		dropPeer,
+		nil,
 		enqueueCh,
 		getBlock,
 		getHeader,
 		chainHeight,
+		0,
 	)
 
 	// Start the witness manager
@@ -478,8 +482,10 @@ func TestBlockFetcherMapStateConsistency(t *testing.T) {
 		insertHeaders,
 		insertChain,
 		dropPeer,
+		nil,
 		false,
 		false,
+		0,
 	)
 
 	fetcher.Start()
@@ -534,10 +540,12 @@ func TestWitnessManagerStateConsistency(t *testing.T) {
 	manager := newWitnessManager(
 		quit,
 		dropPeer,
+		nil,
 		enqueueCh,
 		getBlock,
 		getHeader,
 		chainHeight,
+		0,
 	)
 
 	block := createTestBlock(101)
@@ -594,8 +602,10 @@ func TestBlockFetcherMemoryLeaks(t *testing.T) {
 		insertHeaders,
 		insertChain,
 		dropPeer,
+		nil,
 		false,
 		false,
+		0,
 	)
 
 	fetcher.Start()
@@ -648,10 +658,12 @@ func TestWitnessManagerMemoryLeaks(t *testing.T) {
 	manager := newWitnessManager(
 		quit,
 		dropPeer,
+		nil,
 		enqueueCh,
 		getBlock,
 		getHeader,
 		chainHeight,
+		0,
 	)
 
 	// Add and remove many entries to test cleanup

eth/fetcher/block_fetcher_test.go

@@ -109,7 +109,7 @@ func newTester(light bool) *fetcherTester {
 		blocks:  map[common.Hash]*types.Block{genesis.Hash(): genesis},
 		drops:   make(map[string]bool),
 	}
-	tester.fetcher = NewBlockFetcher(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer, false, false)
+	tester.fetcher = NewBlockFetcher(light, tester.getHeader, tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertHeaders, tester.insertChain, tester.dropPeer, nil, false, false, 0)
 	tester.fetcher.Start()
 
 	return tester

eth/fetcher/witness_manager.go

@@ -30,6 +30,16 @@
 
 	witnessCacheSize = 10
 	witnessCacheTTL  = 2 * time.Minute
+
+	// Witness verification constants
+	witnessVerificationPeers   = 2               // Number of random peers to query for verification
+	witnessVerificationTimeout = 5 * time.Second // Timeout for verification queries
+
+	// Witness size estimation constants
+	// Assuming 1M gas results in 1MB witness, and max page size is 15MB
+	gasPerMB             = 1_000_000 // 1M gas per MB of witness
+	maxPageSizeMB        = 15        // Maximum page size in MB
+	witnessPageThreshold = 10        // Default threshold if gas ceil not available
 )
 
 // witnessRequestState tracks the state of a pending witness request.
@@ -52,6 +62,7 @@
 	// Parent fetcher fields/methods required
 	parentQuit        <-chan struct{}        // Parent fetcher's quit channel
 	parentDropPeer    peerDropFn             // Function to drop a misbehaving peer
+	parentJailPeer    peerJailFn             // Function to jail a peer to prevent reconnection (optional)
 	parentEnqueueCh   chan<- *enqueueRequest // Channel to send completed blocks+witnesses back
 	parentGetBlock    blockRetrievalFn       // Function to check if block is known locally
 	parentGetHeader   HeaderRetrievalFn      // Function to check if header is known locally (needed for checks)
@@ -62,6 +73,9 @@
 	witnessUnavailable map[common.Hash]time.Time                    // Tracks hashes whose witnesses are known to be unavailable, with expiry times.
 	witnessCache       *ttlcache.Cache[common.Hash, *cachedWitness] // TTL cache of witnesses that arrived before their blocks
 
+	// Witness verification state
+	gasCeil uint64 // Gas ceiling for calculating dynamic page threshold
+
 	// Communication channels (owned by witnessManager)
 	injectNeedWitnessCh chan *injectBlockNeedWitnessMsg // Injected blocks needing witness fetch
 	injectWitnessCh     chan *injectedWitnessMsg        // Injected witnesses from broadcast
@@ -83,13 +97,15 @@
 }
 
 // newWitnessManager creates and initializes a new witnessManager.
 func newWitnessManager(
 	parentQuit <-chan struct{},
 	parentDropPeer peerDropFn,
+	parentJailPeer peerJailFn,
 	parentEnqueueCh chan<- *enqueueRequest,
 	parentGetBlock blockRetrievalFn,
 	parentGetHeader HeaderRetrievalFn,
 	parentChainHeight chainHeightFn,
+	gasCeil uint64,
 ) *witnessManager {
 	// Create TTL cache with 1 minute expiration for witnesses
 	witnessCache := ttlcache.New[common.Hash, *cachedWitness](
@@ -100,13 +116,15 @@
 	m := &witnessManager{
 		parentQuit:          parentQuit,
 		parentDropPeer:      parentDropPeer,
+		parentJailPeer:      parentJailPeer,
 		parentEnqueueCh:     parentEnqueueCh,
 		parentGetBlock:      parentGetBlock,
 		parentGetHeader:     parentGetHeader,
 		parentChainHeight:   parentChainHeight,
 		pending:             make(map[common.Hash]*witnessRequestState),
 		witnessUnavailable:  make(map[common.Hash]time.Time),
 		witnessCache:        witnessCache,
+		gasCeil:             gasCeil,
 		injectNeedWitnessCh: make(chan *injectBlockNeedWitnessMsg, 10),
 		injectWitnessCh:     make(chan *injectedWitnessMsg, 10),
 		witnessTimer:        time.NewTimer(0),
@@ -887,3 +905,114 @@
 }
 
 var ErrNoWitnessPeerAvailable = errors.New("no peer with witness available") // Define a potential specific error
+
+// calculatePageThreshold calculates the dynamic page threshold based on gas ceiling
+// Formula: ceil(gasCeil (in millions) / maxPageSizeMB)
+// Example: 50M gas / 15MB per page = ceil(3.33) = 4 pages
+func (m *witnessManager) calculatePageThreshold() uint64 {
+	if m.gasCeil == 0 {
+		return witnessPageThreshold // Return default if gas ceil not set
+	}
+
+	// Convert gas ceil to millions and divide by max page size in MB using ceiling division
+	gasCeilMB := m.gasCeil / gasPerMB
+
+	// Ceiling division: (a + b - 1) / b
+	threshold := (gasCeilMB + maxPageSizeMB - 1) / maxPageSizeMB
+
+	// Ensure minimum threshold of 1 page
+	if threshold < 1 {
+		threshold = 1
+	}
+
+	log.Debug("[wm] Calculated dynamic page threshold", "gasCeil", m.gasCeil, "gasCeilMB", gasCeilMB, "threshold", threshold)
+	return threshold
+}
+
+// getConsensusPageCountWithOriginal gets consensus page count including the original peer
+func (m *witnessManager) getConsensusPageCountWithOriginal(peers []string, hash common.Hash, originalPageCount uint64, getWitnessPageCount func(peer string, hash common.Hash) (uint64, error)) uint64 {
+	// Start with the original peer's count
+	countMap := make(map[uint64]int)
+	countMap[originalPageCount] = 1
+
+	// Query random peers and add their counts
+	for _, peer := range peers {
+		pageCount, err := getWitnessPageCount(peer, hash)
+		if err == nil {
+			countMap[pageCount]++
+		}
+	}
+
+	// Find the most common page count (majority vote)
+	var maxCount int
+	var consensusCount uint64
+	for count, freq := range countMap {
+		if freq > maxCount {
+			maxCount = freq
+			consensusCount = count
+		}
+	}
+
+	// Log the consensus result
+	log.Debug("[wm] Consensus calculation", "counts", countMap, "consensus", consensusCount, "maxVotes", maxCount)
+
+	// Only return consensus if we have majority (at least 2 out of 3)
+	if maxCount >= 2 {
+		return consensusCount
+	}
+
+	// No clear majority, return 0 (will be treated as no consensus)
+	return 0
+}
+
+// CheckWitnessPageCount checks if a witness page count should trigger verification
+// Returns true if peer is honest (or under threshold), false if peer should be dropped
+func (m *witnessManager) CheckWitnessPageCount(hash common.Hash, pageCount uint64, peer string, getRandomPeers func() []string, getWitnessPageCount func(peer string, hash common.Hash) (uint64, error)) bool {
+	// Calculate dynamic threshold based on gas ceiling
+	threshold := m.calculatePageThreshold()
+
+	// If page count is within threshold, no verification needed
+	// No peer queries are made in this case
+	if pageCount <= threshold {
+		log.Debug("[wm] Witness page count within threshold, no verification needed", "peer", peer, "pageCount", pageCount, "threshold", threshold)
+		return true
+	}
+
+	// Page count exceeds threshold - verify synchronously
+	log.Debug("[wm] Witness page count exceeds threshold, running synchronous verification", "peer", peer, "hash", hash, "pageCount", pageCount, "threshold", threshold)
+	return m.verifyWitnessPageCountSync(hash, pageCount, peer, getRandomPeers, getWitnessPageCount)
+}
+
+// verifyWitnessPageCountSync verifies a witness page count synchronously and returns result
+func (m *witnessManager) verifyWitnessPageCountSync(hash common.Hash, reportedPageCount uint64, reportingPeer string, getRandomPeers func() []string, getWitnessPageCount func(peer string, hash common.Hash) (uint64, error)) bool {
+	// Get random peers for verification
+	randomPeers := getRandomPeers()
+	if len(randomPeers) < witnessVerificationPeers {
+		// Not enough peers for verification, assume honest (conservative approach)
+		log.Debug("[wm] Not enough peers for verification, assuming honest", "peer", reportingPeer, "availablePeers", len(randomPeers))
+		return true
+	}
+
+	// Select random peers for verification
+	selectedPeers := randomPeers[:witnessVerificationPeers]
+
+	// Query selected peers for page count and include original peer's count in consensus
+	consensusPageCount := m.getConsensusPageCountWithOriginal(selectedPeers, hash, reportedPageCount, getWitnessPageCount)
+
+	// Determine if original peer is honest based on majority consensus
+	if consensusPageCount != reportedPageCount && consensusPageCount != 0 {
+		// Peer is dishonest - drop and jail immediately
+		log.Warn("Dropping dishonest peer - consensus verification failed", "peer", reportingPeer, "reported", reportedPageCount, "consensus", consensusPageCount)
+		m.parentDropPeer(reportingPeer)
+		// Also jail the peer to prevent reconnection
+		if m.parentJailPeer != nil {
+			log.Warn("Jailing dishonest peer", "peer", reportingPeer)
+			m.parentJailPeer(reportingPeer)
+		}
+		return false
+	}
+
+	// Peer is honest or no consensus (assume honest to avoid false positives)
+	log.Debug("[wm] Peer verification successful", "peer", reportingPeer, "pageCount", reportedPageCount, "hash", hash)
+	return true
+}