feat(coordinator): Add Auth (#382)

* feat: Add event ingester auth + sentry presets * Generate node name dynamically * Linting * Linting * linting * feat: Update golangci-lint version to v1.61 * build: Update golangci-lint-action to v6版本 * fix: Update authorization enabled status to false * feat: Add AuthConfig struct with validation method * feat: Add auth to coordinator * fix: Update auth enabled to false in xatu server config * feat: Add auth to coordinator
ethpandaops · Oct 2, 2024 · 4c47a93 · 4c47a93
1 parent b8e84ba
commit 4c47a93
Show file tree

Hide file tree

Showing 5 changed files with 151 additions and 2 deletions.
diff --git a/deploy/local/docker-compose/xatu-server.yaml b/deploy/local/docker-compose/xatu-server.yaml
@@ -36,6 +36,9 @@ tracing:
 services:
   coordinator:
     enabled: true # requires persistence to be enabled
+    auth:
+      enabled: false
+      secret: "super-secret-token"
     nodeRecord:
       maxQueueSize: 51200
       batchTimeout: 5s

diff --git a/pkg/server/service/coordinator/auth.go b/pkg/server/service/coordinator/auth.go
@@ -0,0 +1,41 @@
+package coordinator
+
+import (
+	"context"
+	"strings"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
+)
+
+func (c *Client) validateAuth(ctx context.Context, md metadata.MD) error {
+	if c.config.Auth.Enabled == nil || !*c.config.Auth.Enabled {
+		return nil
+	}
+
+	authHeader := md.Get("Authorization")
+
+	if len(authHeader) == 0 {
+		return status.Errorf(codes.Unauthenticated, "missing authorization header")
+	}
+
+	token := authHeader[0]
+	if token == "" {
+		return status.Errorf(codes.Unauthenticated, "empty authorization header")
+	}
+
+	// Extract the secret from the token
+	split := strings.Split(token, " ")
+	if len(split) != 2 {
+		return status.Errorf(codes.Unauthenticated, "invalid authorization header")
+	}
+
+	secret := split[1]
+
+	if secret != c.config.Auth.Secret {
+		return status.Errorf(codes.Unauthenticated, "invalid secret")
+	}
+
+	return nil
+}
diff --git a/pkg/server/service/coordinator/client.go b/pkg/server/service/coordinator/client.go
@@ -20,6 +20,7 @@ import (
 	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
@@ -89,6 +90,17 @@ func (c *Client) Stop(ctx context.Context) error {
 }
 
 func (c *Client) CreateNodeRecords(ctx context.Context, req *xatu.CreateNodeRecordsRequest) (*xatu.CreateNodeRecordsResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	for _, record := range req.NodeRecords {
 		// TODO(sam.calder-mason): Derive client id/name from the request jwt
 		c.metrics.AddNodeRecordReceived(1, "unknown")
@@ -133,6 +145,17 @@ func (c *Client) CreateNodeRecords(ctx context.Context, req *xatu.CreateNodeReco
 }
 
 func (c *Client) ListStalledExecutionNodeRecords(ctx context.Context, req *xatu.ListStalledExecutionNodeRecordsRequest) (*xatu.ListStalledExecutionNodeRecordsResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	pageSize := int(req.PageSize)
 	if pageSize == 0 {
 		pageSize = 100
@@ -159,6 +182,17 @@ func (c *Client) ListStalledExecutionNodeRecords(ctx context.Context, req *xatu.
 }
 
 func (c *Client) CreateExecutionNodeRecordStatus(ctx context.Context, req *xatu.CreateExecutionNodeRecordStatusRequest) (*xatu.CreateExecutionNodeRecordStatusResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	if req.Status == nil {
 		return nil, status.Errorf(codes.InvalidArgument, "status is required")
 	}
@@ -216,6 +250,17 @@ func (c *Client) CreateExecutionNodeRecordStatus(ctx context.Context, req *xatu.
 }
 
 func (c *Client) CoordinateExecutionNodeRecords(ctx context.Context, req *xatu.CoordinateExecutionNodeRecordsRequest) (*xatu.CoordinateExecutionNodeRecordsResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	targetedNodes := []string{}
 	ignoredNodeRecords := []string{}
 	activities := []*node.Activity{}
@@ -286,6 +331,17 @@ func (c *Client) CoordinateExecutionNodeRecords(ctx context.Context, req *xatu.C
 }
 
 func (c *Client) GetDiscoveryNodeRecord(ctx context.Context, req *xatu.GetDiscoveryNodeRecordRequest) (*xatu.GetDiscoveryNodeRecordResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	records, err := c.persistence.ListNodeRecordExecutions(ctx, req.NetworkIds, req.ForkIdHashes, 100)
 	if err != nil {
 		return nil, status.Error(codes.Internal, perrors.Wrap(err, "failed to get discovery node records from db").Error())
@@ -304,6 +360,17 @@ func (c *Client) GetDiscoveryNodeRecord(ctx context.Context, req *xatu.GetDiscov
 }
 
 func (c *Client) GetCannonLocation(ctx context.Context, req *xatu.GetCannonLocationRequest) (*xatu.GetCannonLocationResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	location, err := c.persistence.GetCannonLocationByNetworkIDAndType(ctx, req.NetworkId, req.Type.Enum().String())
 	if err != nil && err != persistence.ErrCannonLocationNotFound {
 		return nil, status.Error(codes.Internal, perrors.Wrap(err, "failed to get cannon location from db").Error())
@@ -326,6 +393,17 @@ func (c *Client) GetCannonLocation(ctx context.Context, req *xatu.GetCannonLocat
 }
 
 func (c *Client) UpsertCannonLocation(ctx context.Context, req *xatu.UpsertCannonLocationRequest) (*xatu.UpsertCannonLocationResponse, error) {
+	if c.config.Auth.Enabled != nil && *c.config.Auth.Enabled {
+		md, ok := metadata.FromIncomingContext(ctx)
+		if !ok {
+			return nil, status.Errorf(codes.Unauthenticated, "missing metadata")
+		}
+
+		if err := c.validateAuth(ctx, md); err != nil {
+			return nil, err
+		}
+	}
+
 	newLocation := &cannon.Location{}
 
 	err := newLocation.Marshal(req.Location)

diff --git a/pkg/server/service/coordinator/config.go b/pkg/server/service/coordinator/config.go
@@ -1,10 +1,33 @@
 package coordinator
 
-import "github.com/ethpandaops/xatu/pkg/server/service/coordinator/node"
+import (
+	"errors"
+	"fmt"
+
+	"github.com/ethpandaops/xatu/pkg/server/service/coordinator/node"
+)
 
 type Config struct {
 	Enabled    bool        `yaml:"enabled" default:"false"`
 	NodeRecord node.Config `yaml:"nodeRecord"`
+	Auth       AuthConfig  `yaml:"auth"`
+}
+
+type AuthConfig struct {
+	Enabled *bool  `yaml:"enabled" default:"false"`
+	Secret  string `yaml:"secret"`
+}
+
+func (c *AuthConfig) Validate() error {
+	if c.Enabled == nil || !*c.Enabled {
+		return nil
+	}
+
+	if c.Secret == "" {
+		return errors.New("secret is required")
+	}
+
+	return nil
 }
 
 func (c *Config) Validate() error {
@@ -16,5 +39,9 @@ func (c *Config) Validate() error {
 		return err
 	}
 
+	if err := c.Auth.Validate(); err != nil {
+		return fmt.Errorf("invalid auth config: %w", err)
+	}
+
 	return nil
 }
diff --git a/pkg/server/service/event-ingester/auth/authorization.go b/pkg/server/service/event-ingester/auth/authorization.go
@@ -61,7 +61,7 @@ func (a *Authorization) Start(ctx context.Context) error {
 			userNames[user] = true
 		}
 
-		a.log.WithField("group", group).WithField("users", len(group.Users().Usernames())).Info("Loaded group with users")
+		a.log.WithField("group", group.Name()).WithField("users", len(group.Users().Usernames())).Info("Loaded group with users")
 	}
 
 	return nil