feat: Ensure Header integrity on DA

block/internal/submitting/da_submitter.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/rs/zerolog"
-	"google.golang.org/protobuf/proto"
 
 	"github.com/evstack/ev-node/block/internal/cache"
 	"github.com/evstack/ev-node/block/internal/common"
@@ -161,7 +160,7 @@ func (s *DASubmitter) recordFailure(reason common.DASubmitterFailureReason) {
 }
 
 // SubmitHeaders submits pending headers to DA layer
-func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) error {
+func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager, signer signer.Signer) error {
 	headers, err := cache.GetPendingHeaders(ctx)
 	if err != nil {
 		return fmt.Errorf("failed to get pending headers: %w", err)
@@ -171,15 +170,29 @@ func (s *DASubmitter) SubmitHeaders(ctx context.Context, cache cache.Manager) er
 		return nil
 	}
 
+	if signer == nil {
+		return fmt.Errorf("signer is nil")
+	}
+
 	s.logger.Info().Int("count", len(headers)).Msg("submitting headers to DA")
 
 	return submitToDA(s, ctx, headers,
 		func(header *types.SignedHeader) ([]byte, error) {
-			headerPb, err := header.ToProto()
+			// A. Marshal the inner SignedHeader content to bytes (canonical representation for signing)
+			//    This effectively signs "Fields 1-3" of the intended DAHeaderEnvelope.
+			contentBytes, err := header.MarshalBinary()
 			if err != nil {
-				return nil, fmt.Errorf("failed to convert header to proto: %w", err)
+				return nil, fmt.Errorf("failed to marshal signed header for envelope signing: %w", err)
 			}
-			return proto.Marshal(headerPb)
+
+			// B. Sign the contentBytes with the envelope signer (aggregator)
+			envelopeSignature, err := signer.Sign(contentBytes)
+			if err != nil {
+				return nil, fmt.Errorf("failed to sign envelope: %w", err)
+			}
+
+			// C. Create the envelope and marshal it
+			return header.MarshalDAEnvelope(envelopeSignature)
 		},
 		func(submitted []*types.SignedHeader, res *datypes.ResultSubmit) {
 			for _, header := range submitted {

block/internal/submitting/da_submitter_integration_test.go

@@ -99,7 +99,7 @@ func TestDASubmitter_SubmitHeadersAndData_MarksInclusionAndUpdatesLastSubmitted(
 	daSubmitter := NewDASubmitter(client, cfg, gen, common.DefaultBlockOptions(), common.NopMetrics(), zerolog.Nop())
 
 	// Submit headers and data
-	require.NoError(t, daSubmitter.SubmitHeaders(context.Background(), cm))
+	require.NoError(t, daSubmitter.SubmitHeaders(context.Background(), cm, n))
 	require.NoError(t, daSubmitter.SubmitData(context.Background(), cm, n, gen))
 
 	// After submission, inclusion markers should be set

block/internal/submitting/da_submitter_test.go

@@ -213,7 +213,7 @@ func TestDASubmitter_SubmitHeaders_Success(t *testing.T) {
 	require.NoError(t, batch2.Commit())
 
 	// Submit headers
-	err = submitter.SubmitHeaders(ctx, cm)
+	err = submitter.SubmitHeaders(ctx, cm, signer)
 	require.NoError(t, err)
 
 	// Verify headers are marked as DA included
@@ -229,8 +229,11 @@ func TestDASubmitter_SubmitHeaders_NoPendingHeaders(t *testing.T) {
 	submitter, _, cm, mockDA, _ := setupDASubmitterTest(t)
 	ctx := context.Background()
 
+	// Create test signer
+	_, _, signer := createTestSigner(t)
+
 	// Submit headers when none are pending
-	err := submitter.SubmitHeaders(ctx, cm)
+	err := submitter.SubmitHeaders(ctx, cm, signer)
 	require.NoError(t, err) // Should succeed with no action
 	mockDA.AssertNotCalled(t, "Submit", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 }

block/internal/submitting/submitter.go

@@ -25,7 +25,7 @@ import (
 
 // daSubmitterAPI defines minimal methods needed by Submitter for DA submissions.
 type daSubmitterAPI interface {
-	SubmitHeaders(ctx context.Context, cache cache.Manager) error
+	SubmitHeaders(ctx context.Context, cache cache.Manager, signer signer.Signer) error
 	SubmitData(ctx context.Context, cache cache.Manager, signer signer.Signer, genesis genesis.Genesis) error
 }
 
@@ -158,7 +158,7 @@ func (s *Submitter) daSubmissionLoop() {
 							s.logger.Debug().Time("t", time.Now()).Uint64("headers", headersNb).Msg("Header submission completed")
 							s.headerSubmissionMtx.Unlock()
 						}()
-						if err := s.daSubmitter.SubmitHeaders(s.ctx, s.cache); err != nil {
+						if err := s.daSubmitter.SubmitHeaders(s.ctx, s.cache, s.signer); err != nil {
 							// Check for unrecoverable errors that indicate a critical issue
 							if errors.Is(err, common.ErrOversizedItem) {
 								s.logger.Error().Err(err).

block/internal/submitting/submitter_test.go

@@ -399,7 +399,7 @@ type fakeDASubmitter struct {
 	chData chan struct{}
 }
 
-func (f *fakeDASubmitter) SubmitHeaders(ctx context.Context, _ cache.Manager) error {
+func (f *fakeDASubmitter) SubmitHeaders(ctx context.Context, _ cache.Manager, _ signer.Signer) error {
 	select {
 	case f.chHdr <- struct{}{}:
 	default:

block/internal/syncing/da_retriever.go

@@ -34,6 +34,10 @@ type daRetriever struct {
 	// on restart, will be refetch as da height is updated by syncer
 	pendingHeaders map[uint64]*types.SignedHeader
 	pendingData    map[uint64]*types.Data
+
+	// strictMode indicates if the node has seen a valid DAHeaderEnvelope
+	// and should now reject all legacy/unsigned headers.
+	strictMode bool
 }
 
 // NewDARetriever creates a new DA retriever
@@ -50,6 +54,7 @@ func NewDARetriever(
 		logger:         logger.With().Str("component", "da_retriever").Logger(),
 		pendingHeaders: make(map[uint64]*types.SignedHeader),
 		pendingData:    make(map[uint64]*types.Data),
+		strictMode:     false,
 	}
 }
 
@@ -228,15 +233,58 @@ func (r *daRetriever) processBlobs(ctx context.Context, blobs [][]byte, daHeight
 // tryDecodeHeader attempts to decode a blob as a header
 func (r *daRetriever) tryDecodeHeader(bz []byte, daHeight uint64) *types.SignedHeader {
 	header := new(types.SignedHeader)
-	var headerPb pb.SignedHeader
 
-	if err := proto.Unmarshal(bz, &headerPb); err != nil {
-		return nil
-	}
+	isValidEnvelope := false
 
-	if err := header.FromProto(&headerPb); err != nil {
+	// Attempt to unmarshal as DAHeaderEnvelope and get the envelope signature
+	if envelopeSignature, err := header.UnmarshalDAEnvelope(bz); err != nil {
+		// If in strict mode, we REQUIRE an envelope.
+		if r.strictMode {
+			r.logger.Warn().Err(err).Msg("strict mode is enabled, rejecting non-envelope blob")
+			return nil
+		}
+
+		// Fallback for backward compatibility (only if NOT in strict mode)
+		r.logger.Debug().Msg("trying legacy decoding")
+		var headerPb pb.SignedHeader
+		if errLegacy := proto.Unmarshal(bz, &headerPb); errLegacy != nil {
+			return nil
+		}
+		if errLegacy := header.FromProto(&headerPb); errLegacy != nil {
+			return nil
+		}
+	} else {
+		// We have a structurally valid envelope (or at least it parsed)
+		if len(envelopeSignature) > 0 {
+			if header.Signer.PubKey == nil {
+				r.logger.Debug().Msg("header signer has no pubkey, cannot verify envelope")
+				return nil
+			}
+			payload, err := header.MarshalBinary()
+			if err != nil {
+				r.logger.Debug().Err(err).Msg("failed to marshal header for verification")
+				return nil
+			}
+			if valid, err := header.Signer.PubKey.Verify(payload, envelopeSignature); err != nil || !valid {
+				r.logger.Info().Err(err).Msg("DA envelope signature verification failed")
+				return nil
+			}
+			r.logger.Debug().Uint64("height", header.Height()).Msg("DA envelope signature verified")
+			isValidEnvelope = true
+		}
+	}
+	if r.strictMode && !isValidEnvelope {
+		r.logger.Warn().Msg("strict mode: rejecting block that is not a fully valid envelope")
 		return nil
 	}
+	// Mode Switch Logic
+	if isValidEnvelope && !r.strictMode {
+		r.logger.Info().Uint64("height", header.Height()).Msg("valid DA envelope detected, switching to STRICT MODE")
+		r.strictMode = true
+	}
+
+	// Legacy blob support implies: strictMode == false AND (!isValidEnvelope).
+	// We fall through here.
 
 	// Basic validation
 	if err := header.Header.ValidateBasic(); err != nil {

block/internal/syncing/da_retriever_strict_test.go

@@ -0,0 +1,97 @@
+package syncing
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/evstack/ev-node/pkg/config"
+	"github.com/evstack/ev-node/pkg/genesis"
+	"github.com/evstack/ev-node/types"
+)
+
+func TestDARetriever_StrictEnvelopeMode_Switch(t *testing.T) {
+	// Setup keys
+	addr, pub, signer := buildSyncTestSigner(t)
+	gen := genesis.Genesis{ChainID: "tchain", InitialHeight: 1, StartTime: time.Now().Add(-time.Second), ProposerAddress: addr}
+
+	r := newTestDARetriever(t, nil, config.DefaultConfig(), gen)
+
+	// 1. Create a Legacy Header (SignedHeader marshaled directly)
+	// This simulates old blobs on the network before the upgrade.
+	legacyHeader := &types.SignedHeader{
+		Header: types.Header{
+			BaseHeader:      types.BaseHeader{ChainID: gen.ChainID, Height: 1, Time: uint64(time.Now().UnixNano())},
+			ProposerAddress: addr,
+		},
+		Signer: types.Signer{PubKey: pub, Address: addr},
+	}
+	// Sign it
+	bz, err := types.DefaultAggregatorNodeSignatureBytesProvider(&legacyHeader.Header)
+	require.NoError(t, err)
+	sig, err := signer.Sign(bz)
+	require.NoError(t, err)
+	legacyHeader.Signature = sig
+
+	legacyBlob, err := legacyHeader.MarshalBinary()
+	require.NoError(t, err)
+
+	// 2. Create an Envelope Header (DAHeaderEnvelope)
+	// This simulates a new blob after upgrade.
+	envelopeHeader := &types.SignedHeader{
+		Header: types.Header{
+			BaseHeader:      types.BaseHeader{ChainID: gen.ChainID, Height: 2, Time: uint64(time.Now().UnixNano())},
+			ProposerAddress: addr,
+		},
+		Signer: types.Signer{PubKey: pub, Address: addr},
+	}
+	// Sign content
+	bz2, err := types.DefaultAggregatorNodeSignatureBytesProvider(&envelopeHeader.Header)
+	require.NoError(t, err)
+	sig2, err := signer.Sign(bz2)
+	require.NoError(t, err)
+	envelopeHeader.Signature = sig2
+
+	// Create Envelope
+	// We need to sign the envelope itself.
+	// The `SubmitHeaders` logic wraps it. We emulate it here using `MarshalDAEnvelope`.
+	// First get canonical content bytes (fields 1-3)
+	contentBytes, err := envelopeHeader.MarshalBinary()
+	require.NoError(t, err)
+	// Sign envelope
+	envSig, err := signer.Sign(contentBytes)
+	require.NoError(t, err)
+	// Marshal to envelope
+	envelopeBlob, err := envelopeHeader.MarshalDAEnvelope(envSig)
+	require.NoError(t, err)
+
+	// --- Test Scenario ---
+
+	// A. Initial State: StrictMode is false. Legacy blob should be accepted.
+	assert.False(t, r.strictMode)
+
+	decodedLegacy := r.tryDecodeHeader(legacyBlob, 100)
+	require.NotNil(t, decodedLegacy)
+	assert.Equal(t, uint64(1), decodedLegacy.Height())
+
+	// StrictMode should still be false because it was a legacy blob
+	assert.False(t, r.strictMode)
+
+	// B. Receiving Envelope: Should be accepted and Switch StrictMode to true.
+	decodedEnvelope := r.tryDecodeHeader(envelopeBlob, 101)
+	require.NotNil(t, decodedEnvelope)
+	assert.Equal(t, uint64(2), decodedEnvelope.Height())
+
+	assert.True(t, r.strictMode, "retriever should have switched to strict mode")
+
+	// C. Receiving Legacy again: Should be REJECTED now.
+	// We reuse the same legacyBlob (or a new one, doesn't matter, structure is legacy).
+	decodedLegacyAgain := r.tryDecodeHeader(legacyBlob, 102)
+	assert.Nil(t, decodedLegacyAgain, "legacy blob should be rejected in strict mode")
+
+	// D. Receiving Envelope again: Should still be accepted.
+	decodedEnvelopeAgain := r.tryDecodeHeader(envelopeBlob, 103)
+	require.NotNil(t, decodedEnvelopeAgain)
+}

proto/evnode/v1/evnode.proto

@@ -47,6 +47,17 @@ message SignedHeader {
   Header header = 1;
   bytes signature = 2;
   Signer signer = 3;
+  // Reserved for DAHeaderEnvelope envelope_signature
+  reserved 4;
+}
+
+// DAHeaderEnvelope is a wrapper around SignedHeader for DA submission.
+// It is binary compatible with SignedHeader (fields 1-3) but adds an envelope signature.
+message DAHeaderEnvelope {
+  Header header = 1;
+  bytes signature = 2;
+  Signer signer = 3;
+  bytes envelope_signature = 4;
 }
 
 // Signer is a signer of a block in the blockchain.