Enforce centralized sequencer in p2p block gossiping (#1418)

<!--
Please read and fill out this form before submitting your PR.

Please make sure you have reviewed our contributors guide before
submitting your
first PR.
-->

## Overview
Resolves #1417, resolves #1404

<!-- 
Please provide an explanation of the PR, including the appropriate
context,
background, goal, and rationale. If there is an issue with this
information,
please provide a tl;dr and link the issue. 
-->

## Checklist

<!-- 
Please complete the checklist to ensure that the PR is ready to be
reviewed.

IMPORTANT:
PRs should be left in Draft until the below checklist is completed.
-->

- [x] New and updated code has appropriate documentation
- [x] New and updated code has new and/or updated testing
- [x] Required CI checks are passing
- [ ] Visual proof for any user facing features like CLI or
documentation updates
- [x] Linked issues closed with keywords


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced block validation to reject certain blocks early in the
process.
- Implemented centralized sequencer validation to ensure block
sequencing integrity.

- **Bug Fixes**
- Adjusted the initialization of nodes in test suites to align with
updated genesis document and signing key configurations.
- Updated error handling in block management to provide more informative
feedback.

- **Refactor**
- Streamlined block manager initialization with improved error handling.
- Updated test cases to reflect new validator set and signing key
initialization methods.
- Refined node test initialization to include signing key handling and
error propagation.

- **Documentation**
	- Added new error messages to improve clarity on validation failures.

- **Tests**
- Enhanced integration and unit tests to cover new validation checks and
error conditions.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Connor O'Hara <connor@switchboard.xyz>

Author: Manav-Aggarwal

block/manager.go

@@ -133,11 +133,6 @@ func NewManager(
 		s.DAHeight = conf.DAStartHeight
 	}
 
-	proposerAddress, err := getAddress(proposerKey)
-	if err != nil {
-		return nil, err
-	}
-
 	if conf.DABlockTime == 0 {
 		logger.Info("Using default DA block time", "DABlockTime", defaultDABlockTime)
 		conf.DABlockTime = defaultDABlockTime
@@ -148,6 +143,11 @@ func NewManager(
 		conf.BlockTime = defaultBlockTime
 	}
 
+	proposerAddress, err := getAddress(proposerKey)
+	if err != nil {
+		return nil, err
+	}
+
 	exec := state.NewBlockExecutor(proposerAddress, genesis.ChainID, mempool, proxyApp, eventBus, logger)
 	if s.LastBlockHeight+1 == uint64(genesis.InitialHeight) {
 		res, err := exec.InitChain(genesis)
@@ -465,6 +465,10 @@ func (m *Manager) BlockStoreRetrieveLoop(ctx context.Context) {
 					return
 				default:
 				}
+				// early validation to reject junk blocks
+				if !m.isUsingExpectedCentralizedSequencer(block) {
+					continue
+				}
 				m.logger.Debug("block retrieved from p2p block sync", "blockHeight", block.Height(), "daHeight", daHeight)
 				m.blockInCh <- NewBlockEvent{block, daHeight}
 			}
@@ -547,12 +551,8 @@ func (m *Manager) processNextDABlock(ctx context.Context) error {
 			}
 			m.logger.Debug("retrieved potential blocks", "n", len(blockResp.Blocks), "daHeight", daHeight)
 			for _, block := range blockResp.Blocks {
-				// received block is not from the expected centralized sequencer
-				if !bytes.Equal(block.SignedHeader.ProposerAddress, m.genesis.Validators[0].Address.Bytes()) {
-					continue
-				}
 				// early validation to reject junk blocks
-				if block.ValidateBasic() != nil {
+				if !m.isUsingExpectedCentralizedSequencer(block) {
 					continue
 				}
 				blockHash := block.Hash().String()
@@ -588,6 +588,10 @@ func (m *Manager) processNextDABlock(ctx context.Context) error {
 	return err
 }
 
+func (m *Manager) isUsingExpectedCentralizedSequencer(block *types.Block) bool {
+	return bytes.Equal(block.SignedHeader.ProposerAddress, m.genesis.Validators[0].Address.Bytes()) && block.ValidateBasic() == nil
+}
+
 func (m *Manager) fetchBlock(ctx context.Context, daHeight uint64) (da.ResultRetrieveBlocks, error) {
 	var err error
 	blockRes := m.dalc.RetrieveBlocks(ctx, daHeight)

block/manager_test.go

@@ -20,11 +20,11 @@ import (
 )
 
 func TestInitialState(t *testing.T) {
-	genesisValidators, _ := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, _ := types.GetGenesisWithPrivkey()
 	genesis := &cmtypes.GenesisDoc{
 		ChainID:       "genesis id",
 		InitialHeight: 100,
-		Validators:    genesisValidators,
+		Validators:    genesisDoc.Validators,
 	}
 	sampleState := types.State{
 		ChainID:         "state id",

node/full_client_test.go

@@ -65,7 +65,9 @@ func getRPC(t *testing.T) (*mocks.Application, *FullClient) {
 	app.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil)
 	key, _, _ := crypto.GenerateEd25519Key(crand.Reader)
 	ctx := context.Background()
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 	node, err := newFullNode(
 		ctx,
 		config.NodeConfig{
@@ -74,10 +76,7 @@ func getRPC(t *testing.T) (*mocks.Application, *FullClient) {
 		key,
 		signingKey,
 		proxy.NewLocalClientCreator(app),
-		&cmtypes.GenesisDoc{
-			ChainID:    "test",
-			Validators: genesisValidators,
-		},
+		genesisDoc,
 		log.TestingLogger(),
 	)
 	require.NoError(err)
@@ -492,7 +491,9 @@ func TestTx(t *testing.T) {
 	mockApp.On("PrepareProposal", mock.Anything, mock.Anything).Return(prepareProposalResponse).Maybe()
 	mockApp.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil)
 	key, _, _ := crypto.GenerateEd25519Key(crand.Reader)
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	node, err := newFullNode(ctx, config.NodeConfig{
@@ -502,7 +503,7 @@ func TestTx(t *testing.T) {
 			BlockTime: 1 * time.Second, // blocks must be at least 1 sec apart for adjacent headers to get verified correctly
 		}},
 		key, signingKey, proxy.NewLocalClientCreator(mockApp),
-		&cmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators},
+		genesisDoc,
 		test.NewFileLogger(t))
 	require.NoError(err)
 	require.NotNil(node)
@@ -740,7 +741,9 @@ func TestMempool2Nodes(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
 
-	genesisValidators, signingKey1 := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey1, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 
 	app := &mocks.Application{}
 	app.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil)
@@ -769,7 +772,7 @@ func TestMempool2Nodes(t *testing.T) {
 			ListenAddress: "/ip4/127.0.0.1/tcp/9001",
 		},
 		BlockManagerConfig: getBMConfig(),
-	}, key1, signingKey1, proxy.NewLocalClientCreator(app), &cmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
+	}, key1, signingKey1, proxy.NewLocalClientCreator(app), genesisDoc, log.TestingLogger())
 	require.NoError(err)
 	require.NotNil(node1)
 
@@ -779,7 +782,7 @@ func TestMempool2Nodes(t *testing.T) {
 			ListenAddress: "/ip4/127.0.0.1/tcp/9002",
 			Seeds:         "/ip4/127.0.0.1/tcp/9001/p2p/" + id1.Pretty(),
 		},
-	}, key2, signingKey2, proxy.NewLocalClientCreator(app), &cmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
+	}, key2, signingKey2, proxy.NewLocalClientCreator(app), genesisDoc, log.TestingLogger())
 	require.NoError(err)
 	require.NotNil(node2)
 
@@ -834,8 +837,10 @@ func TestStatus(t *testing.T) {
 	app.On("PrepareProposal", mock.Anything, mock.Anything).Return(prepareProposalResponse).Maybe()
 	app.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil)
 	key, _, _ := crypto.GenerateEd25519Key(crand.Reader)
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
-	pubKey := genesisValidators[0].PubKey
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
+	pubKey := genesisDoc.Validators[0].PubKey
 
 	node, err := newFullNode(
 		context.Background(),
@@ -852,10 +857,7 @@ func TestStatus(t *testing.T) {
 		key,
 		signingKey,
 		proxy.NewLocalClientCreator(app),
-		&cmtypes.GenesisDoc{
-			ChainID:    "test",
-			Validators: genesisValidators,
-		},
+		genesisDoc,
 		test.NewFileLogger(t),
 	)
 	require.NoError(err)
@@ -890,8 +892,8 @@ func TestStatus(t *testing.T) {
 	assert.Equal(int64(2), resp.SyncInfo.LatestBlockHeight)
 
 	// Changed the RPC method to get this from the genesis.
-	assert.Equal(genesisValidators[0].Address, resp.ValidatorInfo.Address)
-	assert.Equal(genesisValidators[0].PubKey, resp.ValidatorInfo.PubKey)
+	assert.Equal(genesisDoc.Validators[0].Address, resp.ValidatorInfo.Address)
+	assert.Equal(genesisDoc.Validators[0].PubKey, resp.ValidatorInfo.PubKey)
 	// hardcode to 1, shouldn't matter because it's a centralized sequencer
 	assert.Equal(int64(1), resp.ValidatorInfo.VotingPower)
 
@@ -945,7 +947,9 @@ func TestFutureGenesisTime(t *testing.T) {
 	mockApp.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil)
 	mockApp.On("CheckTx", mock.Anything, mock.Anything).Return(&abci.ResponseCheckTx{}, nil)
 	key, _, _ := crypto.GenerateEd25519Key(crand.Reader)
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 	genesisTime := time.Now().Local().Add(time.Second * time.Duration(1))
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -961,7 +965,7 @@ func TestFutureGenesisTime(t *testing.T) {
 			ChainID:       "test",
 			InitialHeight: 1,
 			GenesisTime:   genesisTime,
-			Validators:    genesisValidators,
+			Validators:    genesisDoc.Validators,
 		},
 		test.NewFileLogger(t))
 	require.NoError(err)

node/full_node_integration_test.go

@@ -1,7 +1,6 @@
 package node
 
 import (
-	"bytes"
 	"context"
 	"crypto/rand"
 	"errors"
@@ -13,7 +12,6 @@ import (
 	"time"
 
 	"github.com/cometbft/cometbft/crypto/ed25519"
-	"github.com/cometbft/cometbft/p2p"
 	"github.com/stretchr/testify/assert"
 
 	abci "github.com/cometbft/cometbft/abci/types"
@@ -26,7 +24,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/rollkit/rollkit/config"
-	"github.com/rollkit/rollkit/da"
 	test "github.com/rollkit/rollkit/test/log"
 	"github.com/rollkit/rollkit/test/mocks"
 	"github.com/rollkit/rollkit/types"
@@ -40,78 +37,6 @@ func prepareProposalResponse(_ context.Context, req *abci.RequestPrepareProposal
 	}, nil
 }
 
-func TestCentralizedSequencer(t *testing.T) {
-	require := require.New(t)
-	assert := assert.New(t)
-	genDoc, privkey := types.GetGenesisWithPrivkey()
-	genDoc.AppHash = make([]byte, 32)
-	nodeKey := &p2p.NodeKey{
-		PrivKey: privkey,
-	}
-	signingKey, err := types.GetNodeKey(nodeKey)
-	require.NoError(err)
-
-	app := &mocks.Application{}
-	app.On("InitChain", mock.Anything, mock.Anything).Return(&abci.ResponseInitChain{}, nil)
-	app.On("CheckTx", mock.Anything, mock.Anything).Return(&abci.ResponseCheckTx{}, nil)
-	app.On("PrepareProposal", mock.Anything, mock.Anything).Return(prepareProposalResponse).Maybe()
-	app.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil)
-	app.On("FinalizeBlock", mock.Anything, mock.Anything).Return(finalizeBlockResponse)
-	app.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil)
-
-	vals := types.GetValidatorSetFromGenesis(genDoc)
-
-	dalc := getMockDA()
-
-	blockManagerConfig := config.BlockManagerConfig{
-		BlockTime:     1 * time.Second,
-		DAStartHeight: 1,
-		DABlockTime:   1 * time.Second,
-	}
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	node, err := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr, Aggregator: false, BlockManagerConfig: blockManagerConfig}, signingKey, signingKey, proxy.NewLocalClientCreator(app), genDoc, test.NewFileLogger(t))
-	require.NoError(err)
-	node.dalc = dalc
-	node.blockManager.SetDALC(dalc)
-
-	err = node.Start()
-	defer func() {
-		assert.NoError(node.Stop())
-	}()
-	require.NoError(err)
-
-	lastState, err := node.Store.GetState()
-	require.NoError(err)
-	lastResults := lastState.LastResultsHash
-
-	validBlock, err := types.GetFirstBlock(privkey, &vals, lastResults, false, false)
-	require.NoError(err)
-	err = validBlock.ValidateBasic()
-	require.NoError(err)
-	junkProposerBlock, err := types.GetFirstBlock(privkey, &vals, lastResults, true, false)
-	require.NoError(err)
-	err = junkProposerBlock.ValidateBasic()
-	require.Error(err)
-	sigInvalidBlock, err := types.GetFirstBlock(privkey, &vals, lastResults, false, true)
-	require.NoError(err)
-	err = sigInvalidBlock.ValidateBasic()
-	require.Error(err)
-	submitResp := dalc.SubmitBlocks(ctx, []*types.Block{validBlock, junkProposerBlock, sigInvalidBlock})
-	fmt.Println(submitResp)
-	require.Equal(submitResp.Code, da.StatusSuccess)
-	require.NoError(testutils.Retry(300, 100*time.Millisecond, func() error {
-		block, err := node.Store.GetBlock(1)
-		if err != nil {
-			return err
-		}
-		if !bytes.Equal(block.Hash(), validBlock.Hash()) {
-			return fmt.Errorf("unexpected block")
-		}
-		return nil
-	}))
-}
-
 func TestAggregatorMode(t *testing.T) {
 	assert := assert.New(t)
 	require := require.New(t)
@@ -125,13 +50,15 @@ func TestAggregatorMode(t *testing.T) {
 	app.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil)
 
 	key, _, _ := crypto.GenerateEd25519Key(rand.Reader)
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 	blockManagerConfig := config.BlockManagerConfig{
 		BlockTime: 1 * time.Second,
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	node, err := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr, Aggregator: true, BlockManagerConfig: blockManagerConfig}, key, signingKey, proxy.NewLocalClientCreator(app), &cmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
+	node, err := newFullNode(ctx, config.NodeConfig{DAAddress: MockServerAddr, Aggregator: true, BlockManagerConfig: blockManagerConfig}, key, signingKey, proxy.NewLocalClientCreator(app), genesisDoc, log.TestingLogger())
 	require.NoError(err)
 	require.NotNil(node)
 
@@ -238,7 +165,9 @@ func TestLazyAggregator(t *testing.T) {
 	app.On("Commit", mock.Anything, mock.Anything).Return(&abci.ResponseCommit{}, nil)
 
 	key, _, _ := crypto.GenerateEd25519Key(rand.Reader)
-	genesisValidators, signingKey := types.GetGenesisValidatorSetWithSigner()
+	genesisDoc, genesisValidatorKey := types.GetGenesisWithPrivkey()
+	signingKey, err := types.PrivKeyToSigningKey(genesisValidatorKey)
+	require.NoError(err)
 	blockManagerConfig := config.BlockManagerConfig{
 		// After the genesis header is published, the syncer is started
 		// which takes little longer (due to initialization) and the syncer
@@ -257,7 +186,7 @@ func TestLazyAggregator(t *testing.T) {
 		Aggregator:         true,
 		BlockManagerConfig: blockManagerConfig,
 		LazyAggregator:     true,
-	}, key, signingKey, proxy.NewLocalClientCreator(app), &cmtypes.GenesisDoc{ChainID: "test", Validators: genesisValidators}, log.TestingLogger())
+	}, key, signingKey, proxy.NewLocalClientCreator(app), genesisDoc, log.TestingLogger())
 	assert.False(node.IsRunning())
 	assert.NoError(err)