Relay authentication

api/proto/relay/relay.proto

@@ -49,12 +49,14 @@ message GetChunksRequest {
   // 1. the operator id
   // 2. for each chunk request:
   //    a. if the chunk request is a request by index:
-  //       i. the blob key
-  //       ii. the start index
-  //       iii. the end index
+  //       i.   a one byte ASCII representation of the character "i" (aka Ox69)
+  //       ii.  the blob key
+  //       iii. the start index
+  //       iv.  the end index
   //    b. if the chunk request is a request by range:
-  //       i. the blob key
-  //       ii. each requested chunk index, in order
+  //       i.   a one byte ASCII representation of the character "r" (aka Ox72)
+  //       ii.  the blob key
+  //       iii. each requested chunk index, in order
   bytes operator_signature = 3;
 }
 

relay/auth/authenticator.go

@@ -6,24 +6,28 @@ import (
 	"fmt"
 	pb "github.com/Layr-Labs/eigenda/api/grpc/relay"
 	"github.com/Layr-Labs/eigenda/core"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"sync"
 	"time"
 )
 
 // RequestAuthenticator authenticates requests to the relay service. This object is thread safe.
 type RequestAuthenticator interface {
 	// AuthenticateGetChunksRequest authenticates a GetChunksRequest, returning an error if the request is invalid.
-	// The address is the address of the peer that sent the request. This may be used to cache auth results
+	// The origin is the address of the peer that sent the request. This may be used to cache auth results
 	// in order to save server resources.
 	AuthenticateGetChunksRequest(
-		address string,
+		origin string,
 		request *pb.GetChunksRequest,
 		now time.Time) error
+
+	// PreloadCache preloads the key cache with the public keys of all operators that are currently indexed.
+	PreloadCache() error
 }
 
 // authenticationTimeout is used to track the expiration of an auth.
 type authenticationTimeout struct {
-	clientID   string
+	origin     string
 	expiration time.Time
 }
 
@@ -47,29 +51,52 @@ type requestAuthenticator struct {
 	savedAuthLock sync.Mutex
 
 	// keyCache is used to cache the public keys of operators. Operator keys are assumed to never change.
-	keyCache sync.Map
+	keyCache *lru.Cache[core.OperatorID, *core.G2Point]
 }
 
 // NewRequestAuthenticator creates a new RequestAuthenticator.
 func NewRequestAuthenticator(
 	ics core.IndexedChainState,
-	authenticationTimeoutDuration time.Duration) RequestAuthenticator {
+	keyCacheSize int,
+	authenticationTimeoutDuration time.Duration) (RequestAuthenticator, error) {
+
+	keyCache, err := lru.New[core.OperatorID, *core.G2Point](keyCacheSize)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create key cache: %w", err)
+	}
 
 	return &requestAuthenticator{
 		ics:                           ics,
 		authenticatedClients:          make(map[string]struct{}),
 		authenticationTimeouts:        make([]*authenticationTimeout, 0),
 		authenticationTimeoutDuration: authenticationTimeoutDuration,
-		keyCache:                      sync.Map{},
+		keyCache:                      keyCache,
+	}, nil
+}
+
+func (a *requestAuthenticator) PreloadCache() error {
+	blockNumber, err := a.ics.GetCurrentBlockNumber()
+	if err != nil {
+		return fmt.Errorf("failed to get current block number: %w", err)
 	}
+	operators, err := a.ics.GetIndexedOperators(context.Background(), blockNumber)
+	if err != nil {
+		return fmt.Errorf("failed to get operators: %w", err)
+	}
+
+	for operatorID, operator := range operators {
+		a.keyCache.Add(operatorID, operator.PubkeyG2)
+	}
+
+	return nil
 }
 
 func (a *requestAuthenticator) AuthenticateGetChunksRequest(
-	address string,
+	origin string,
 	request *pb.GetChunksRequest,
 	now time.Time) error {
 
-	if a.isAuthenticationStillValid(now, address) {
+	if a.isAuthenticationStillValid(now, origin) {
 		// We've recently authenticated this client. Do not authenticate again for a while.
 		return nil
 	}
@@ -95,15 +122,14 @@ func (a *requestAuthenticator) AuthenticateGetChunksRequest(
 		return errors.New("signature verification failed")
 	}
 
-	a.saveAuthenticationResult(now, address)
+	a.saveAuthenticationResult(now, origin)
 	return nil
 }
 
 // getOperatorKey returns the public key of the operator with the given ID, caching the result.
 func (a *requestAuthenticator) getOperatorKey(operatorID core.OperatorID) (*core.G2Point, error) {
-	untypedKey, ok := a.keyCache.Load(operatorID)
+	key, ok := a.keyCache.Get(operatorID)
 	if ok {
-		key := untypedKey.(*core.G2Point)
 		return key, nil
 	}
 
@@ -120,14 +146,14 @@ func (a *requestAuthenticator) getOperatorKey(operatorID core.OperatorID) (*core
 	if !ok {
 		return nil, errors.New("operator not found")
 	}
-	key := operator.PubkeyG2
+	key = operator.PubkeyG2
 
-	a.keyCache.Store(operatorID, key)
+	a.keyCache.Add(operatorID, key)
 	return key, nil
 }
 
 // saveAuthenticationResult saves the result of an auth.
-func (a *requestAuthenticator) saveAuthenticationResult(now time.Time, address string) {
+func (a *requestAuthenticator) saveAuthenticationResult(now time.Time, origin string) {
 	if a.authenticationTimeoutDuration == 0 {
 		// Authentication saving is disabled.
 		return
@@ -136,10 +162,10 @@ func (a *requestAuthenticator) saveAuthenticationResult(now time.Time, address s
 	a.savedAuthLock.Lock()
 	defer a.savedAuthLock.Unlock()
 
-	a.authenticatedClients[address] = struct{}{}
+	a.authenticatedClients[origin] = struct{}{}
 	a.authenticationTimeouts = append(a.authenticationTimeouts,
 		&authenticationTimeout{
-			clientID:   address,
+			origin:     origin,
 			expiration: now.Add(a.authenticationTimeoutDuration),
 		})
 }
@@ -160,13 +186,14 @@ func (a *requestAuthenticator) isAuthenticationStillValid(now time.Time, address
 }
 
 // removeOldAuthentications removes any authentications that have expired.
+// This method is not thread safe and should be called with the savedAuthLock held.
 func (a *requestAuthenticator) removeOldAuthentications(now time.Time) {
 	index := 0
 	for ; index < len(a.authenticationTimeouts); index++ {
 		if a.authenticationTimeouts[index].expiration.After(now) {
 			break
 		}
-		delete(a.authenticatedClients, a.authenticationTimeouts[index].clientID)
+		delete(a.authenticatedClients, a.authenticationTimeouts[index].origin)
 	}
 	if index > 0 {
 		a.authenticationTimeouts = a.authenticationTimeouts[index:]

relay/auth/authenticator_test.go

@@ -57,12 +57,13 @@ func TestValidRequest(t *testing.T) {
 
 	timeout := 10 * time.Second
 
-	authenticator := NewRequestAuthenticator(ics, timeout)
+	authenticator, err := NewRequestAuthenticator(ics, 1024, timeout)
+	require.NoError(t, err)
 
 	request := randomGetChunksRequest()
 	request.OperatorId = operatorID[:]
-	SignGetChunksRequest(ics.KeyPairs[operatorID], request)
-	signature := request.OperatorSignature
+	signature := SignGetChunksRequest(ics.KeyPairs[operatorID], request)
+	request.OperatorSignature = signature
 
 	now := time.Now()
 
@@ -119,12 +120,13 @@ func TestAuthenticationSavingDisabled(t *testing.T) {
 	// This disables saving of authentication results.
 	timeout := time.Duration(0)
 
-	authenticator := NewRequestAuthenticator(ics, timeout)
+	authenticator, err := NewRequestAuthenticator(ics, 1024, timeout)
+	require.NoError(t, err)
 
 	request := randomGetChunksRequest()
 	request.OperatorId = operatorID[:]
-	SignGetChunksRequest(ics.KeyPairs[operatorID], request)
-	signature := request.OperatorSignature
+	signature := SignGetChunksRequest(ics.KeyPairs[operatorID], request)
+	request.OperatorSignature = signature
 
 	now := time.Now()
 
@@ -162,7 +164,8 @@ func TestNonExistingClient(t *testing.T) {
 
 	timeout := 10 * time.Second
 
-	authenticator := NewRequestAuthenticator(ics, timeout)
+	authenticator, err := NewRequestAuthenticator(ics, 1024, timeout)
+	require.NoError(t, err)
 
 	invalidOperatorID := tu.RandomBytes(32)
 
@@ -191,11 +194,12 @@ func TestBadSignature(t *testing.T) {
 
 	timeout := 10 * time.Second
 
-	authenticator := NewRequestAuthenticator(ics, timeout)
+	authenticator, err := NewRequestAuthenticator(ics, 1024, timeout)
+	require.NoError(t, err)
 
 	request := randomGetChunksRequest()
 	request.OperatorId = operatorID[:]
-	SignGetChunksRequest(ics.KeyPairs[operatorID], request)
+	request.OperatorSignature = SignGetChunksRequest(ics.KeyPairs[operatorID], request)
 
 	now := time.Now()
 

relay/auth/request_signing.go

@@ -7,6 +7,11 @@ import (
 	"golang.org/x/crypto/sha3"
 )
 
+var (
+	iByte = []byte{0x69}
+	rByte = []byte{0x72}
+)
+
 // HashGetChunksRequest hashes the given GetChunksRequest.
 func HashGetChunksRequest(request *pb.GetChunksRequest) []byte {
 
@@ -19,6 +24,7 @@ func HashGetChunksRequest(request *pb.GetChunksRequest) []byte {
 	for _, chunkRequest := range request.GetChunkRequests() {
 		if chunkRequest.GetByIndex() != nil {
 			getByIndex := chunkRequest.GetByIndex()
+			hasher.Write(iByte)
 			hasher.Write(getByIndex.BlobKey)
 			for _, index := range getByIndex.ChunkIndices {
 				indexBytes := make([]byte, 4)
@@ -27,6 +33,7 @@ func HashGetChunksRequest(request *pb.GetChunksRequest) []byte {
 			}
 		} else {
 			getByRange := chunkRequest.GetByRange()
+			hasher.Write(rByte)
 			hasher.Write(getByRange.BlobKey)
 
 			startBytes := make([]byte, 4)
@@ -42,9 +49,10 @@ func HashGetChunksRequest(request *pb.GetChunksRequest) []byte {
 	return hasher.Sum(nil)
 }
 
-// SignGetChunksRequest signs the given GetChunksRequest with the given private key.
-func SignGetChunksRequest(keys *core.KeyPair, request *pb.GetChunksRequest) {
+// SignGetChunksRequest signs the given GetChunksRequest with the given private key. Does not
+// write the signature into the request.
+func SignGetChunksRequest(keys *core.KeyPair, request *pb.GetChunksRequest) []byte {
 	hash := HashGetChunksRequest(request)
 	signature := keys.SignMessage(([32]byte)(hash))
-	request.OperatorSignature = signature.G1Point.Serialize()
+	return signature.G1Point.Serialize()
 }

relay/cmd/config.go

@@ -79,8 +79,9 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 				GetChunkBytesBurstinessClient:   ctx.Int(flags.GetChunkBytesBurstinessClientFlag.Name),
 				MaxConcurrentGetChunkOpsClient:  ctx.Int(flags.MaxConcurrentGetChunkOpsClientFlag.Name),
 			},
-			AuthenticationTimeout:  ctx.Duration(flags.AuthenticationTimeoutFlag.Name),
-			AuthenticationDisabled: ctx.Bool(flags.AuthenticationDisabledFlag.Name),
+			AuthenticationKeyCacheSize: ctx.Int(flags.AuthenticationKeyCacheSizeFlag.Name),
+			AuthenticationTimeout:      ctx.Duration(flags.AuthenticationTimeoutFlag.Name),
+			AuthenticationDisabled:     ctx.Bool(flags.AuthenticationDisabledFlag.Name),
 		},
 		EthClientConfig:               geth.ReadEthClientConfig(ctx),
 		IndexerPullInterval:           ctx.Duration(flags.IndexerPullIntervalFlag.Name),

relay/cmd/flags/flags.go

@@ -210,6 +210,13 @@ var (
 		EnvVar:   common.PrefixEnvVar(envVarPrefix, "INDEXER_PULL_INTERVAL"),
 		Value:    5 * time.Minute,
 	}
+	AuthenticationKeyCacheSizeFlag = cli.IntFlag{
+		Name:     common.PrefixFlag(FlagPrefix, "authentication-key-cache-size"),
+		Usage:    "Max number of items in the authentication key cache",
+		Required: false,
+		EnvVar:   common.PrefixEnvVar(envVarPrefix, "AUTHENTICATION_KEY_CACHE_SIZE"),
+		Value:    1024 * 1024,
+	}
 	AuthenticationTimeoutFlag = cli.DurationFlag{
 		Name:     common.PrefixFlag(FlagPrefix, "authentication-timeout"),
 		Usage:    "Duration to keep authentication results",
@@ -232,8 +239,6 @@ var requiredFlags = []cli.Flag{
 	RelayIDsFlag,
 	BlsOperatorStateRetrieverAddrFlag,
 	EigenDAServiceManagerAddrFlag,
-	AuthenticationTimeoutFlag,
-	AuthenticationDisabledFlag,
 }
 
 var optionalFlags = []cli.Flag{
@@ -260,6 +265,9 @@ var optionalFlags = []cli.Flag{
 	GetChunkBytesBurstinessClientFlag,
 	MaxConcurrentGetChunkOpsClientFlag,
 	IndexerPullIntervalFlag,
+	AuthenticationKeyCacheSizeFlag,
+	AuthenticationTimeoutFlag,
+	AuthenticationDisabledFlag,
 }
 
 var Flags []cli.Flag

relay/server.go

@@ -93,6 +93,9 @@ type Config struct {
 	// RateLimits contains configuration for rate limiting.
 	RateLimits limiter.Config
 
+	// AuthenticationKeyCacheSize is the maximum number of operator public keys that can be cached.
+	AuthenticationKeyCacheSize int
+
 	// AuthenticationTimeout is the duration for which an authentication is "cached". A request from the same client
 	// within this duration will not trigger a new authentication in order to save resources. If zero, then each request
 	// will be authenticated independently, regardless of timing.
@@ -145,7 +148,18 @@ func NewServer(
 
 	var authenticator auth.RequestAuthenticator
 	if !config.AuthenticationDisabled {
-		authenticator = auth.NewRequestAuthenticator(ics, config.AuthenticationTimeout)
+		authenticator, err = auth.NewRequestAuthenticator(
+			ics,
+			config.AuthenticationKeyCacheSize,
+			config.AuthenticationTimeout)
+		if err != nil {
+			return nil, fmt.Errorf("error creating authenticator: %w", err)
+		}
+
+		err = authenticator.PreloadCache()
+		if err != nil {
+			return nil, fmt.Errorf("error preloading operator key cache: %w", err)
+		}
 	}
 
 	return &Server{
@@ -219,13 +233,13 @@ func (s *Server) GetChunks(ctx context.Context, request *pb.GetChunksRequest) (*
 			"too many chunk requests provided, max is %d", s.config.MaxKeysPerGetChunksRequest)
 	}
 
-	client, ok := peer.FromContext(ctx)
-	if !ok {
-		return nil, errors.New("could not get peer information")
-	}
-	clientAddress := client.Addr.String()
-
 	if s.authenticator != nil {
+		client, ok := peer.FromContext(ctx)
+		if !ok {
+			return nil, errors.New("could not get peer information")
+		}
+		clientAddress := client.Addr.String()
+
 		err := s.authenticator.AuthenticateGetChunksRequest(clientAddress, request, time.Now())
 		if err != nil {
 			return nil, fmt.Errorf("auth failed: %w", err)