Merge branch 'master' into quaq/blob-verification

SECURITY.md

@@ -0,0 +1,20 @@
+# Security Policy
+
+## Version Information
+
+Please see [Releases](https://github.com/Layr-Labs/eigenda/releases) and we recommend using the [most recently released version](https://github.com/Layr-Labs/eigenda/releases/latest).
+
+## Audit reports
+
+Audit reports are published in the `docs` folder: https://github.com/Layr-Labs/eigenda/master/docs/audits 
+
+| Date | Report Link |
+| ------- | ----------- |
+| 202404 | [pdf](https://github.com/Layr-Labs/eigenda/blob/security-doc/docs/audits/Sigma_Prime_EigenDA_Offchain_Security_Assessment_Report.pdf) |
+
+## Reporting a Vulnerability
+
+**Please do not file a public ticket** mentioning the vulnerability.
+
+Please report security vulnerabilities to [email protected] with the all the relavent details included in the email.
+

api/clients/mock/relay_client.go

@@ -39,6 +39,15 @@ func (c *MockRelayClient) GetChunksByIndex(ctx context.Context, relayKey corev2.
 	return args.Get(0).([][]byte), args.Error(1)
 }
 
+func (c *MockRelayClient) GetSockets() map[corev2.RelayKey]string {
+	args := c.Called()
+	if args.Get(0) == nil {
+		return nil
+	}
+
+	return args.Get(0).(map[corev2.RelayKey]string)
+}
+
 func (c *MockRelayClient) Close() error {
 	args := c.Called()
 	return args.Error(0)

api/clients/relay_client.go

@@ -39,6 +39,8 @@ type RelayClient interface {
 	// The returned slice has the same length and ordering as the input slice, and the i-th element is the bundle for the i-th request.
 	// Each bundle is a sequence of frames in raw form (i.e., serialized core.Bundle bytearray).
 	GetChunksByIndex(ctx context.Context, relayKey corev2.RelayKey, requests []*ChunkRequestByIndex) ([][]byte, error)
+	// GetSockets returns the relay sockets
+	GetSockets() map[corev2.RelayKey]string
 	Close() error
 }
 
@@ -65,6 +67,8 @@ func NewRelayClient(config *RelayClientConfig, logger logging.Logger) (*relayCli
 		return nil, fmt.Errorf("invalid config: %v", config)
 	}
 
+	logger.Info("creating relay client", "config", config)
+
 	initOnce := sync.Map{}
 	for key := range config.Sockets {
 		initOnce.Store(key, &sync.Once{})
@@ -73,7 +77,7 @@ func NewRelayClient(config *RelayClientConfig, logger logging.Logger) (*relayCli
 		config: config,
 
 		initOnce: &initOnce,
-		logger:   logger,
+		logger:   logger.With("component", "RelayClient"),
 	}, nil
 }
 
@@ -196,6 +200,10 @@ func (c *relayClient) initOnceGrpcConnection(key corev2.RelayKey) error {
 	return initErr
 }
 
+func (c *relayClient) GetSockets() map[corev2.RelayKey]string {
+	return c.config.Sockets
+}
+
 func (c *relayClient) Close() error {
 	var errList *multierror.Error
 	c.conns.Range(func(k, v interface{}) bool {

common/aws/cli.go

@@ -37,12 +37,6 @@ type ClientConfig struct {
 	// FragmentParallelismConstant helps determine the size of the pool of workers to help upload/download files.
 	// A non-zero value for this parameter adds a constant number of workers. Default is 0.
 	FragmentParallelismConstant int
-	// FragmentReadTimeout is used to bound the maximum time to wait for a single fragmented read.
-	// Default is 30 seconds.
-	FragmentReadTimeout time.Duration
-	// FragmentWriteTimeout is used to bound the maximum time to wait for a single fragmented write.
-	// Default is 30 seconds.
-	FragmentWriteTimeout time.Duration
 }
 
 func ClientFlags(envPrefix string, flagPrefix string) []cli.Flag {
@@ -120,8 +114,6 @@ func ReadClientConfig(ctx *cli.Context, flagPrefix string) ClientConfig {
 		EndpointURL:                 ctx.GlobalString(common.PrefixFlag(flagPrefix, EndpointURLFlagName)),
 		FragmentParallelismFactor:   ctx.GlobalInt(common.PrefixFlag(flagPrefix, FragmentParallelismFactorFlagName)),
 		FragmentParallelismConstant: ctx.GlobalInt(common.PrefixFlag(flagPrefix, FragmentParallelismConstantFlagName)),
-		FragmentReadTimeout:         ctx.GlobalDuration(common.PrefixFlag(flagPrefix, FragmentReadTimeoutFlagName)),
-		FragmentWriteTimeout:        ctx.GlobalDuration(common.PrefixFlag(flagPrefix, FragmentWriteTimeoutFlagName)),
 	}
 }
 
@@ -131,7 +123,5 @@ func DefaultClientConfig() *ClientConfig {
 		Region:                      "us-east-2",
 		FragmentParallelismFactor:   8,
 		FragmentParallelismConstant: 0,
-		FragmentReadTimeout:         30 * time.Second,
-		FragmentWriteTimeout:        30 * time.Second,
 	}
 }

common/aws/s3/client.go

@@ -18,6 +18,10 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+const (
+	defaultBlobBufferSizeByte = 128 * 1024
+)
+
 var (
 	once              sync.Once
 	ref               *client
@@ -106,14 +110,20 @@ func NewClient(ctx context.Context, cfg commonaws.ClientConfig, logger logging.L
 }
 
 func (s *client) DownloadObject(ctx context.Context, bucket string, key string) ([]byte, error) {
+	objectSize := defaultBlobBufferSizeByte
+	size, err := s.HeadObject(ctx, bucket, key)
+	if err == nil {
+		objectSize = int(*size)
+	}
+	buffer := manager.NewWriteAtBuffer(make([]byte, 0, objectSize))
+
 	var partMiBs int64 = 10
 	downloader := manager.NewDownloader(s.s3Client, func(d *manager.Downloader) {
 		d.PartSize = partMiBs * 1024 * 1024 // 10MB per part
 		d.Concurrency = 3                   //The number of goroutines to spin up in parallel per call to Upload when sending parts
 	})
 
-	buffer := manager.NewWriteAtBuffer([]byte{})
-	_, err := downloader.Download(ctx, buffer, &s3.GetObjectInput{
+	_, err = downloader.Download(ctx, buffer, &s3.GetObjectInput{
 		Bucket: aws.String(bucket),
 		Key:    aws.String(key),
 	})
@@ -223,9 +233,6 @@ func (s *client) FragmentedUploadObject(
 	}
 	resultChannel := make(chan error, len(fragments))
 
-	ctx, cancel := context.WithTimeout(ctx, s.cfg.FragmentWriteTimeout)
-	defer cancel()
-
 	for _, fragment := range fragments {
 		fragmentCapture := fragment
 		s.concurrencyLimiter <- struct{}{}
@@ -283,9 +290,6 @@ func (s *client) FragmentedDownloadObject(
 	}
 	resultChannel := make(chan *readResult, len(fragmentKeys))
 
-	ctx, cancel := context.WithTimeout(ctx, s.cfg.FragmentWriteTimeout)
-	defer cancel()
-
 	for i, fragmentKey := range fragmentKeys {
 		boundFragmentKey := fragmentKey
 		boundI := i

common/metrics/config.go

@@ -0,0 +1,10 @@
+package metrics
+
+// Config provides configuration for a Metrics instance.
+type Config struct {
+	// Namespace is the namespace for the metrics.
+	Namespace string
+
+	// HTTPPort is the port to serve metrics on.
+	HTTPPort int
+}

common/metrics/count_metric.go

@@ -0,0 +1,101 @@
+package metrics
+
+import (
+	"fmt"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var _ CountMetric = &countMetric{}
+
+// countMetric a standard implementation of the CountMetric.
+type countMetric struct {
+	Metric
+
+	// logger is the logger used to log errors.
+	logger logging.Logger
+
+	// name is the name of the metric.
+	name string
+
+	// description is the description of the metric.
+	description string
+
+	// counter is the prometheus counter used to report this metric.
+	vec *prometheus.CounterVec
+
+	// labeler is the label maker used to create labels for this metric.
+	labeler *labelMaker
+}
+
+// newCountMetric creates a new CountMetric instance.
+func newCountMetric(
+	logger logging.Logger,
+	registry *prometheus.Registry,
+	namespace string,
+	name string,
+	description string,
+	labelTemplate any) (CountMetric, error) {
+
+	labeler, err := newLabelMaker(labelTemplate)
+	if err != nil {
+		return nil, err
+	}
+
+	vec := promauto.With(registry).NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Name:      fmt.Sprintf("%s_count", name),
+		},
+		labeler.getKeys(),
+	)
+
+	return &countMetric{
+		logger:      logger,
+		name:        name,
+		description: description,
+		vec:         vec,
+		labeler:     labeler,
+	}, nil
+}
+
+func (m *countMetric) Name() string {
+	return m.name
+}
+
+func (m *countMetric) Unit() string {
+	return "count"
+}
+
+func (m *countMetric) Description() string {
+	return m.description
+}
+
+func (m *countMetric) Type() string {
+	return "counter"
+}
+
+func (m *countMetric) LabelFields() []string {
+	return m.labeler.getKeys()
+}
+
+func (m *countMetric) Increment(label ...any) {
+	m.Add(1, label...)
+}
+
+func (m *countMetric) Add(value float64, label ...any) {
+	var l any
+	if len(label) > 0 {
+		l = label[0]
+	}
+
+	values, err := m.labeler.extractValues(l)
+	if err != nil {
+		m.logger.Errorf("error extracting values from label for metric %s: %v", m.name, err)
+		return
+	}
+
+	observer := m.vec.WithLabelValues(values...)
+	observer.Add(value)
+}

common/metrics/gauge_metric.go

@@ -0,0 +1,103 @@
+package metrics
+
+import (
+	"fmt"
+	"github.com/Layr-Labs/eigensdk-go/logging"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var _ GaugeMetric = &gaugeMetric{}
+
+// gaugeMetric is a standard implementation of the GaugeMetric interface via prometheus.
+type gaugeMetric struct {
+	Metric
+
+	// logger is the logger used to log errors.
+	logger logging.Logger
+
+	// name is the name of the metric.
+	name string
+
+	// unit is the unit of the metric.
+	unit string
+
+	// description is the description of the metric.
+	description string
+
+	// gauge is the prometheus gauge used to report this metric.
+	vec *prometheus.GaugeVec
+
+	// labeler is the label maker used to create labels for this metric.
+	labeler *labelMaker
+}
+
+// newGaugeMetric creates a new GaugeMetric instance.
+func newGaugeMetric(
+	logger logging.Logger,
+	registry *prometheus.Registry,
+	namespace string,
+	name string,
+	unit string,
+	description string,
+	labelTemplate any) (GaugeMetric, error) {
+
+	labeler, err := newLabelMaker(labelTemplate)
+	if err != nil {
+		return nil, err
+	}
+
+	vec := promauto.With(registry).NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Name:      fmt.Sprintf("%s_%s", name, unit),
+		},
+		labeler.getKeys(),
+	)
+
+	return &gaugeMetric{
+		logger:      logger,
+		name:        name,
+		unit:        unit,
+		description: description,
+		vec:         vec,
+		labeler:     labeler,
+	}, nil
+}
+
+func (m *gaugeMetric) Name() string {
+	return m.name
+}
+
+func (m *gaugeMetric) Unit() string {
+	return m.unit
+}
+
+func (m *gaugeMetric) Description() string {
+	return m.description
+}
+
+func (m *gaugeMetric) Type() string {
+	return "gauge"
+}
+
+func (m *gaugeMetric) LabelFields() []string {
+	return m.labeler.getKeys()
+}
+
+func (m *gaugeMetric) Set(value float64, label ...any) {
+	var l any
+	if len(label) > 0 {
+		l = label[0]
+	}
+
+	values, err := m.labeler.extractValues(l)
+	if err != nil {
+		m.logger.Errorf("failed to extract values from label: %v", err)
+		return
+	}
+
+	observer := m.vec.WithLabelValues(values...)
+
+	observer.Set(value)
+}