feat: on-demand check

Author: hopeyen

disperser/apiserver/server_v2.go

@@ -157,6 +157,7 @@ func (s *DispersalServerV2) Start(ctx context.Context) error {
 	if err := s.RefreshOnchainState(ctx); err != nil {
 		return fmt.Errorf("failed to refresh onchain quorum state: %w", err)
 	}
+	s.logger.Debug("Refreshed onchain quorum state", "onchainState", s.onchainState.Load())
 
 	go func() {
 		ticker := time.NewTicker(s.onchainStateRefreshInterval)
@@ -165,6 +166,7 @@ func (s *DispersalServerV2) Start(ctx context.Context) error {
 		for {
 			select {
 			case <-ticker.C:
+				s.logger.Debug("Refreshing onchain quorum state", "onchainState", s.onchainState.Load())
 				if err := s.RefreshOnchainState(ctx); err != nil {
 					s.logger.Error("failed to refresh onchain quorum state", "err", err)
 				}
@@ -258,6 +260,8 @@ func (s *DispersalServerV2) RefreshOnchainState(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("failed to get blob version parameters: %w", err)
 	}
+
+	s.logger.Debug("Refreshed onchain quorum state", "quorumCount", quorumCount, "requiredQuorums", requiredQuorums, "blobParams", blobParams, "storeDurationBlocks", storeDurationBlocks, "blockStaleMeasure", blockStaleMeasure)
 	onchainState := &OnchainState{
 		QuorumCount:           quorumCount,
 		RequiredQuorums:       requiredQuorums,
@@ -267,6 +271,8 @@ func (s *DispersalServerV2) RefreshOnchainState(ctx context.Context) error {
 
 	s.onchainState.Store(onchainState)
 
+	s.logger.Debug("stored onchain quorum state", "onchainState", onchainState)
+
 	return nil
 }
 

node/auth/authenticator.go

@@ -47,8 +47,7 @@ type requestAuthenticator struct {
 	// reloaded from the chain state in case the key has been changed.
 	keyTimeoutDuration time.Duration
 
-	// disperserIDFilter is a function that returns true if the given disperser ID is valid.
-	disperserIDFilter func(uint32) bool
+	//TODO: add blacklistedDispersers is a set of disperser IDs that are blacklisted.
 }
 
 // NewRequestAuthenticator creates a new RequestAuthenticator.
@@ -57,7 +56,6 @@ func NewRequestAuthenticator(
 	chainReader core.Reader,
 	keyCacheSize int,
 	keyTimeoutDuration time.Duration,
-	disperserIDFilter func(uint32) bool,
 	now time.Time) (RequestAuthenticator, error) {
 
 	keyCache, err := lru.New[uint32, *keyWithTimeout](keyCacheSize)
@@ -69,7 +67,6 @@ func NewRequestAuthenticator(
 		chainReader:        chainReader,
 		keyCache:           keyCache,
 		keyTimeoutDuration: keyTimeoutDuration,
-		disperserIDFilter:  disperserIDFilter,
 	}
 
 	err = authenticator.preloadCache(ctx, now)
@@ -81,7 +78,7 @@ func NewRequestAuthenticator(
 }
 
 func (a *requestAuthenticator) preloadCache(ctx context.Context, now time.Time) error {
-	// this will need to be updated for decentralized dispersers
+	//TODO: preload disperser blacklist from storage
 	_, err := a.getDisperserKey(ctx, now, api.EigenLabsDisperserID)
 	if err != nil {
 		return fmt.Errorf("failed to get operator key: %w", err)
@@ -114,9 +111,7 @@ func (a *requestAuthenticator) getDisperserKey(
 	now time.Time,
 	disperserID uint32) (*gethcommon.Address, error) {
 
-	if !a.disperserIDFilter(disperserID) {
-		return nil, fmt.Errorf("invalid disperser ID: %d", disperserID)
-	}
+	//TODO: Reject blacklisted dispersers
 
 	key, ok := a.keyCache.Get(disperserID)
 	if ok {

node/auth/authenticator_test.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"errors"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -32,7 +31,6 @@ func TestValidRequest(t *testing.T) {
 		&chainReader,
 		10,
 		time.Minute,
-		func(uint32) bool { return true },
 		start)
 	require.NoError(t, err)
 
@@ -66,7 +64,6 @@ func TestInvalidRequestWrongHash(t *testing.T) {
 		&chainReader,
 		10,
 		time.Minute,
-		func(uint32) bool { return true },
 		start)
 	require.NoError(t, err)
 
@@ -100,7 +97,6 @@ func TestInvalidRequestWrongKey(t *testing.T) {
 		&chainReader,
 		10,
 		time.Minute,
-		func(uint32) bool { return true },
 		start)
 	require.NoError(t, err)
 
@@ -126,7 +122,7 @@ func TestInvalidRequestInvalidDisperserID(t *testing.T) {
 	require.NoError(t, err)
 	disperserAddress0 := crypto.PubkeyToAddress(*publicKey0)
 
-	// This disperser will be loaded on chain (simulated), but will fail the valid disperser ID filter.
+	// Non EigenLabs disperser is registered on-chain
 	publicKey1, privateKey1, err := rand.ECDSA()
 	require.NoError(t, err)
 	disperserAddress1 := crypto.PubkeyToAddress(*publicKey1)
@@ -137,20 +133,13 @@ func TestInvalidRequestInvalidDisperserID(t *testing.T) {
 	chainReader.Mock.On("GetDisperserAddress", uint32(1234)).Return(
 		nil, errors.New("disperser not found"))
 
-	filterCallCount := atomic.Uint32{}
-
 	authenticator, err := NewRequestAuthenticator(
 		context.Background(),
 		&chainReader,
 		10,
 		time.Minute,
-		func(id uint32) bool {
-			filterCallCount.Add(1)
-			return id != uint32(1)
-		},
 		start)
 	require.NoError(t, err)
-	require.Equal(t, uint32(1), filterCallCount.Load())
 
 	request := RandomStoreChunksRequest(rand)
 	request.DisperserID = 0
@@ -162,14 +151,16 @@ func TestInvalidRequestInvalidDisperserID(t *testing.T) {
 	expectedHash, err := hashing.HashStoreChunksRequest(request)
 	require.NoError(t, err)
 	require.Equal(t, expectedHash, hash)
-	require.Equal(t, uint32(2), filterCallCount.Load())
 
 	request.DisperserID = 1
 	signature, err = SignStoreChunksRequest(privateKey1, request)
 	require.NoError(t, err)
 	request.Signature = signature
-	_, err = authenticator.AuthenticateStoreChunksRequest(context.Background(), request, start)
-	require.Error(t, err)
+	hash, err = authenticator.AuthenticateStoreChunksRequest(context.Background(), request, start)
+	require.NoError(t, err)
+	expectedHash, err = hashing.HashStoreChunksRequest(request)
+	require.NoError(t, err)
+	require.Equal(t, expectedHash, hash)
 
 	request.DisperserID = 1234
 	signature, err = SignStoreChunksRequest(privateKey1, request)
@@ -196,7 +187,6 @@ func TestKeyExpiry(t *testing.T) {
 		&chainReader,
 		10,
 		time.Minute,
-		func(uint32) bool { return true },
 		start)
 	require.NoError(t, err)
 
@@ -260,7 +250,6 @@ func TestKeyCacheSize(t *testing.T) {
 		&chainReader,
 		cacheSize,
 		time.Minute,
-		func(uint32) bool { return true },
 		start)
 	require.NoError(t, err)
 

node/grpc/server_v2.go

@@ -14,6 +14,7 @@ import (
 	"github.com/Layr-Labs/eigenda/common/kvstore"
 	"github.com/Layr-Labs/eigenda/common/replay"
 	"github.com/Layr-Labs/eigenda/core"
+	"github.com/Layr-Labs/eigenda/core/meterer"
 	corev2 "github.com/Layr-Labs/eigenda/core/v2"
 	"github.com/Layr-Labs/eigenda/node"
 	"github.com/Layr-Labs/eigenda/node/auth"
@@ -58,9 +59,6 @@ func NewServerV2(
 			reader,
 			config.DispersalAuthenticationKeyCacheSize,
 			config.DisperserKeyTimeout,
-			func(id uint32) bool {
-				return id == api.EigenLabsDisperserID
-			},
 			time.Now())
 		if err != nil {
 			return nil, fmt.Errorf("failed to create authenticator: %w", err)
@@ -140,6 +138,15 @@ func (s *ServerV2) StoreChunks(ctx context.Context, in *pb.StoreChunksRequest) (
 		if err != nil {
 			return nil, api.NewErrorInvalidArg(fmt.Sprintf("failed to verify request: %v", err))
 		}
+		// TODO: move to blob authenticator later
+		for _, blob := range batch.BlobCertificates {
+			if meterer.IsOnDemandPayment(&blob.BlobHeader.PaymentMetadata) {
+				// Batch contains on-demand payments, so the chunk must be from EigenLabsDisperser
+				if in.DisperserID != api.EigenLabsDisperserID {
+					return nil, api.NewErrorInvalidArg("on-demand payments are only allowed for EigenLabsDisperser")
+				}
+			}
+		}
 	}
 
 	probe.SetStage("get_operator_state")