fix: reject unexpected DA proposers early

Author: randygrok

block/internal/syncing/da_retriever.go

@@ -32,13 +32,17 @@ type pendingDataCleaner interface {
 	removePendingData(height uint64)
 }
 
+type expectedProposerProvider func(height uint64) ([]byte, bool)
+
 // daRetriever handles DA retrieval operations for syncing
 type daRetriever struct {
 	client  da.Client
 	cache   cache.CacheManager
 	genesis genesis.Genesis
 	logger  zerolog.Logger
 
+	expectedProposer expectedProposerProvider
+
 	mu sync.Mutex
 	// transient cache, only full event need to be passed to the syncer
 	// on restart, will be refetch as da height is updated by syncer
@@ -68,6 +72,10 @@ func NewDARetriever(
 	}
 }
 
+func (r *daRetriever) setExpectedProposerProvider(provider expectedProposerProvider) {
+	r.expectedProposer = provider
+}
+
 // RetrieveFromDA retrieves blocks from the specified DA height and returns height events
 func (r *daRetriever) RetrieveFromDA(ctx context.Context, daHeight uint64) ([]common.DAHeightEvent, error) {
 	r.logger.Debug().Uint64("da_height", daHeight).Msg("retrieving from DA")
@@ -309,6 +317,11 @@ func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 		return nil
 	}
 
+	if err := r.assertExpectedProposer(header); err != nil {
+		r.logger.Debug().Err(err).Msg("unexpected proposer")
+		return nil
+	}
+
 	if isValidEnvelope && !r.strictMode {
 		r.logger.Info().Uint64("height", header.Height()).Msg("valid DA envelope detected, switching to STRICT MODE")
 		r.strictMode = true
@@ -329,6 +342,21 @@ func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedH
 	return header
 }
 
+func (r *daRetriever) assertExpectedProposer(header *types.SignedHeader) error {
+	if r.expectedProposer == nil {
+		return nil
+	}
+
+	expected, ok := r.expectedProposer(header.Height())
+	if !ok || len(expected) == 0 {
+		return nil
+	}
+	if !bytes.Equal(header.ProposerAddress, expected) {
+		return fmt.Errorf("%w - got: %x, want: %x", types.ErrUnexpectedProposer, header.ProposerAddress, expected)
+	}
+	return nil
+}
+
 // tryDecodeData attempts to decode a blob as signed data
 func (r *daRetriever) tryDecodeData(bz []byte, daHeight uint64) *types.Data {
 	var signedData types.SignedData

block/internal/syncing/da_retriever_test.go

@@ -338,6 +338,33 @@ func TestDARetriever_ProcessBlobs_KeepsDataForLaterHeaderAfterCandidateEvent(t *
 	require.Equal(t, data.Data.DACommitment().String(), events[0].Data.DACommitment().String())
 }
 
+func TestDARetriever_ProcessBlobs_RejectsUnexpectedProposerBeforePendingHeader(t *testing.T) {
+	expectedAddr, expectedPub, expectedSigner := buildSyncTestSigner(t)
+	wrongAddr, wrongPub, wrongSigner := buildSyncTestSigner(t)
+	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: expectedAddr}
+
+	r := newTestDARetriever(t, nil, config.DefaultConfig(), gen)
+	r.setExpectedProposerProvider(func(height uint64) ([]byte, bool) {
+		if height != 5 {
+			return nil, false
+		}
+		return expectedAddr, true
+	})
+
+	_, data := makeSignedDataBytes(t, gen.ChainID, 5, expectedAddr, expectedPub, expectedSigner, 1)
+	wrongHeaderBin, _ := makeSignedHeaderBytes(t, gen.ChainID, 5, wrongAddr, wrongPub, wrongSigner, nil, &data.Data, nil)
+	correctHeaderBin, correctHeader := makeSignedHeaderBytes(t, gen.ChainID, 5, expectedAddr, expectedPub, expectedSigner, nil, &data.Data, nil)
+
+	events := r.processBlobs(context.Background(), [][]byte{wrongHeaderBin}, 100)
+	require.Empty(t, events)
+	require.NotContains(t, r.pendingHeaders, uint64(5), "unexpected proposer must not occupy the pending header slot")
+
+	events = r.processBlobs(context.Background(), [][]byte{correctHeaderBin}, 101)
+	require.Empty(t, events)
+	require.Contains(t, r.pendingHeaders, uint64(5), "expected proposer should be accepted as pending while data is missing")
+	require.Equal(t, correctHeader.Hash().String(), r.pendingHeaders[5].Hash().String())
+}
+
 func TestDARetriever_ProcessBlobs_MultipleHeadersCrossDAHeightMatching(t *testing.T) {
 
 	addr, pub, signer := buildSyncTestSigner(t)

block/internal/syncing/syncer.go

@@ -181,7 +181,9 @@ func (s *Syncer) Start(ctx context.Context) (err error) {
 	}
 
 	// Initialize handlers
-	s.daRetriever = NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
+	daRetriever := NewDARetriever(s.daClient, s.cache, s.genesis, s.logger)
+	daRetriever.setExpectedProposerProvider(s.expectedProposerForHeight)
+	s.daRetriever = daRetriever
 	if s.config.Instrumentation.IsTracingEnabled() {
 		s.daRetriever = WithTracingDARetriever(s.daRetriever)
 	}
@@ -302,6 +304,14 @@ func (s *Syncer) SetLastState(state types.State) {
 	s.lastState.Store(&state)
 }
 
+func (s *Syncer) expectedProposerForHeight(height uint64) ([]byte, bool) {
+	state := s.getLastState()
+	if height != state.LastBlockHeight+1 || len(state.NextProposerAddress) == 0 {
+		return nil, false
+	}
+	return state.NextProposerAddress, true
+}
+
 // initializeState loads the current sync state
 func (s *Syncer) initializeState() error {
 	// Load state from store

block/internal/syncing/syncer_test.go

@@ -214,6 +214,22 @@ func TestSyncer_ValidateBlock_UsesStateNextProposer(t *testing.T) {
 	require.Contains(t, err.Error(), "unexpected proposer")
 }
 
+func TestSyncer_ExpectedProposerForHeight_OnlyNextHeight(t *testing.T) {
+	addr, _, _ := buildSyncTestSigner(t)
+	s := &Syncer{lastState: &atomic.Pointer[types.State]{}}
+	s.SetLastState(types.State{
+		LastBlockHeight:     4,
+		NextProposerAddress: addr,
+	})
+
+	got, ok := s.expectedProposerForHeight(5)
+	require.True(t, ok)
+	require.Equal(t, addr, got)
+
+	_, ok = s.expectedProposerForHeight(6)
+	require.False(t, ok)
+}
+
 func TestSyncer_TrySyncNextBlock_ClassifiesExternalValidationFailures(t *testing.T) {
 	expectedAddr, expectedPub, expectedSigner := buildSyncTestSigner(t)
 	wrongAddr, wrongPub, wrongSigner := buildSyncTestSigner(t)