feat(queue)!: add priv/pub key signing

cmd/vela-server/queue.go

@@ -18,11 +18,12 @@ func setupQueue(c *cli.Context) (queue.Service, error) {
 
 	// queue configuration
 	_setup := &queue.Setup{
-		Driver:  c.String("queue.driver"),
-		Address: c.String("queue.addr"),
-		Cluster: c.Bool("queue.cluster"),
-		Routes:  c.StringSlice("queue.routes"),
-		Timeout: c.Duration("queue.pop.timeout"),
+		Driver:     c.String("queue.driver"),
+		Address:    c.String("queue.addr"),
+		Cluster:    c.Bool("queue.cluster"),
+		Routes:     c.StringSlice("queue.routes"),
+		Timeout:    c.Duration("queue.pop.timeout"),
+		PrivateKey: c.String("queue.private-key"),
 	}
 
 	// setup the queue

cmd/vela-server/server.go

@@ -98,6 +98,7 @@ func server(c *cli.Context) error {
 		middleware.Secret(c.String("vela-secret")),
 		middleware.Secrets(secrets),
 		middleware.Scm(scm),
+		middleware.QueueSigningPrivateKey(c.String("queue.private-key")),
 		middleware.Allowlist(c.StringSlice("vela-repo-allowlist")),
 		middleware.DefaultBuildLimit(c.Int64("default-build-limit")),
 		middleware.DefaultTimeout(c.Int64("default-build-timeout")),

docker-compose.yml

@@ -42,6 +42,7 @@ services:
       VELA_USER_ACCESS_TOKEN_DURATION: 60m
       VELA_DISABLE_WEBHOOK_VALIDATION: 'true'
       VELA_ENABLE_SECURE_COOKIE: 'false'
+      VELA_QUEUE_SIGNING_PRIVATE_KEY: 'tCIevHOBq6DdN5SSBtteXUusjjd0fOqzk2eyi0DMq04NewmShNKQeUbbp3vkvIckb4pCxc+vxUo+mYf/vzOaSg=='
       VELA_REPO_ALLOWLIST: '*'
       VELA_SCHEDULE_ALLOWLIST: '*'
     env_file:
@@ -78,6 +79,7 @@ services:
       VELA_SERVER_SECRET: 'zB7mrKDTZqNeNTD8z47yG4DHywspAh'
       WORKER_ADDR: 'http://worker:8080'
       WORKER_CHECK_IN: 5m
+      VELA_QUEUE_SIGNING_PUBLIC_KEY: 'DXsJkoTSkHlG26d75LyHJG+KQsXPr8VKPpmH/78zmko='
     restart: always
     ports:
       - '8081:8080'

go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/spf13/afero v1.9.5
 	github.com/urfave/cli/v2 v2.25.7
 	go.starlark.net v0.0.0-20230725161458-0d7263928a74
+	golang.org/x/crypto v0.11.0
 	golang.org/x/oauth2 v0.9.0
 	golang.org/x/sync v0.3.0
 	gopkg.in/square/go-jose.v2 v2.6.0
@@ -116,7 +117,6 @@ require (
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
 	github.com/yuin/gopher-lua v1.1.0 // indirect
 	golang.org/x/arch v0.3.0 // indirect
-	golang.org/x/crypto v0.11.0 // indirect
 	golang.org/x/net v0.12.0 // indirect
 	golang.org/x/sys v0.10.0 // indirect
 	golang.org/x/text v0.11.0 // indirect

queue/flags.go

@@ -50,4 +50,16 @@ var Flags = []cli.Flag{
 		Usage:    "timeout for requests that pop items off the queue",
 		Value:    60 * time.Second,
 	},
+	&cli.StringFlag{
+		EnvVars:  []string{"VELA_QUEUE_SIGNING_PRIVATE_KEY"},
+		FilePath: "/vela/signing.key",
+		Name:     "queue.private-key",
+		Usage:    "set value of base64 encoded queue signing private key",
+	},
+	&cli.StringFlag{
+		EnvVars:  []string{"VELA_QUEUE_SIGNING_PUBLIC_KEY"},
+		FilePath: "/vela/signing.pub",
+		Name:     "queue.public-key",
+		Usage:    "set value of base64 encoded queue signing public key",
+	},
 }

queue/redis/length_test.go

@@ -29,7 +29,7 @@ func TestRedis_Length(t *testing.T) {
 	}
 
 	// setup redis mock
-	_redis, err := NewTest("vela", "vela:second", "vela:third")
+	_redis, err := NewTest(_signingPrivateKey, _signingPublicKey, "vela", "vela:second", "vela:third")
 	if err != nil {
 		t.Errorf("unable to create queue service: %v", err)
 	}

queue/redis/opts.go

@@ -5,6 +5,8 @@
 package redis
 
 import (
+	"encoding/base64"
+	"errors"
 	"fmt"
 	"time"
 )
@@ -69,3 +71,75 @@ func WithTimeout(timeout time.Duration) ClientOpt {
 		return nil
 	}
 }
+
+// WithPrivateKey sets the private key in the queue client for Redis.
+//
+//nolint:dupl // ignore similar code
+func WithPrivateKey(key string) ClientOpt {
+	return func(c *client) error {
+		c.Logger.Trace("configuring private key in redis queue client")
+
+		if len(key) == 0 {
+			c.Logger.Warn("unable to base64 decode private key, provided key is empty. queue service will be unable to sign items")
+			return nil
+		}
+
+		decoded, err := base64.StdEncoding.DecodeString(key)
+		if err != nil {
+			return err
+		}
+
+		if len(decoded) == 0 {
+			return errors.New("unable to base64 decode private key, decoded key is empty")
+		}
+
+		c.config.PrivateKey = new([64]byte)
+		copy(c.config.PrivateKey[:], decoded)
+
+		if c.config.PrivateKey == nil {
+			return errors.New("unable to copy decoded queue signing private key, copied key is nil")
+		}
+
+		if len(c.config.PrivateKey) == 0 {
+			return errors.New("unable to copy decoded queue signing private key, copied key is empty")
+		}
+
+		return nil
+	}
+}
+
+// WithPublicKey sets the public key in the queue client for Redis.
+//
+//nolint:dupl // ignore similar code
+func WithPublicKey(key string) ClientOpt {
+	return func(c *client) error {
+		c.Logger.Tracef("configuring public key in redis queue client")
+
+		if len(key) == 0 {
+			c.Logger.Warn("unable to base64 decode public key, provided key is empty. queue service will be unable to open items")
+			return nil
+		}
+
+		decoded, err := base64.StdEncoding.DecodeString(key)
+		if err != nil {
+			return err
+		}
+
+		if len(decoded) == 0 {
+			return errors.New("unable to base64 decode public key, decoded key is empty")
+		}
+
+		c.config.PublicKey = new([32]byte)
+		copy(c.config.PublicKey[:], decoded)
+
+		if c.config.PublicKey == nil {
+			return errors.New("unable to copy decoded queue signing public key, copied key is nil")
+		}
+
+		if len(c.config.PublicKey) == 0 {
+			return errors.New("unable to copy decoded queue signing public key, copied key is empty")
+		}
+
+		return nil
+	}
+}

queue/redis/opts_test.go

@@ -5,6 +5,7 @@
 package redis
 
 import (
+	"encoding/base64"
 	"fmt"
 	"reflect"
 	"testing"
@@ -180,3 +181,139 @@ func TestRedis_ClientOpt_WithCluster(t *testing.T) {
 		}
 	}
 }
+
+func TestRedis_ClientOpt_WithSigningPrivateKey(t *testing.T) {
+	// setup tests
+	// create a local fake redis instance
+	//
+	// https://pkg.go.dev/github.com/alicebob/miniredis/v2#Run
+	_redis, err := miniredis.Run()
+	if err != nil {
+		t.Errorf("unable to create miniredis instance: %v", err)
+	}
+	defer _redis.Close()
+
+	tests := []struct {
+		failure bool
+		privKey string
+		want    string
+	}{
+		{ //valid key input
+			failure: false,
+			privKey: "tCIevHOBq6DdN5SSBtteXUusjjd0fOqzk2eyi0DMq04NewmShNKQeUbbp3vkvIckb4pCxc+vxUo+mYf/vzOaSg==",
+			want:    "tCIevHOBq6DdN5SSBtteXUusjjd0fOqzk2eyi0DMq04NewmShNKQeUbbp3vkvIckb4pCxc+vxUo+mYf/vzOaSg==",
+		},
+		{ //empty key input
+			failure: false,
+			privKey: "",
+			want:    "",
+		},
+		{ //invalid base64 encoded input
+			failure: true,
+			privKey: "abc123",
+			want:    "",
+		},
+	}
+
+	// run tests
+	for _, test := range tests {
+		_service, err := New(
+			WithAddress(fmt.Sprintf("redis://%s", _redis.Addr())),
+			WithPrivateKey(test.privKey),
+		)
+
+		if test.failure {
+			if err == nil {
+				t.Errorf("WithPrivateKey should have returned err")
+			}
+
+			continue
+		}
+
+		if err != nil {
+			t.Errorf("WithPrivateKey returned err: %v", err)
+		}
+
+		got := ""
+		if _service.config.PrivateKey != nil {
+			got = fmt.Sprintf("%s", *_service.config.PrivateKey)
+		} else {
+			got = ""
+		}
+
+		w, _ := base64.StdEncoding.DecodeString(test.want)
+
+		want := string(w)
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("WithPrivateKey is %v, want %v", got, want)
+		}
+	}
+}
+
+func TestRedis_ClientOpt_WithSigningPublicKey(t *testing.T) {
+	// setup tests
+	// create a local fake redis instance
+	//
+	// https://pkg.go.dev/github.com/alicebob/miniredis/v2#Run
+	_redis, err := miniredis.Run()
+	if err != nil {
+		t.Errorf("unable to create miniredis instance: %v", err)
+	}
+	defer _redis.Close()
+
+	tests := []struct {
+		failure bool
+		pubKey  string
+		want    string
+	}{
+		{ //valid key input
+			failure: false,
+			pubKey:  "DXsJkoTSkHlG26d75LyHJG+KQsXPr8VKPpmH/78zmko=",
+			want:    "DXsJkoTSkHlG26d75LyHJG+KQsXPr8VKPpmH/78zmko=",
+		},
+		{ //empty key input
+			failure: false,
+			pubKey:  "",
+			want:    "",
+		},
+		{ //invalid base64 encoded input
+			failure: true,
+			pubKey:  "abc123",
+			want:    "",
+		},
+	}
+
+	// run tests
+	for _, test := range tests {
+		_service, err := New(
+			WithAddress(fmt.Sprintf("redis://%s", _redis.Addr())),
+			WithPublicKey(test.pubKey),
+		)
+
+		if test.failure {
+			if err == nil {
+				t.Errorf("WithPublicKey should have returned err")
+			}
+
+			continue
+		}
+
+		if err != nil {
+			t.Errorf("WithPublicKey returned err: %v", err)
+		}
+
+		got := ""
+		if _service.config.PublicKey != nil {
+			got = fmt.Sprintf("%s", *_service.config.PublicKey)
+		} else {
+			got = ""
+		}
+
+		w, _ := base64.StdEncoding.DecodeString(test.want)
+
+		want := string(w)
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("SigningPublicKey is %v, want %v", got, want)
+		}
+	}
+}

queue/redis/pop.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/go-vela/types"
 	"github.com/redis/go-redis/v9"
+	"golang.org/x/crypto/nacl/sign"
 )
 
 // Pop grabs an item from the specified channel off the queue.
@@ -35,10 +36,28 @@ func (c *client) Pop(ctx context.Context) (*types.Item, error) {
 		return nil, err
 	}
 
-	item := new(types.Item)
+	// this should already be validated on startup
+	if c.config.PublicKey == nil || len(*c.config.PublicKey) != 32 {
+		return nil, errors.New("no valid signing public key provided")
+	}
+
+	// extract signed item from pop results
+	signed := []byte(result[1])
+
+	var opened, out []byte
+
+	// open the item using the public key generated using sign
+	//
+	// https://pkg.go.dev/golang.org/x/[email protected]/nacl/sign
+	opened, ok := sign.Open(out, signed, c.config.PublicKey)
+	if !ok {
+		return nil, errors.New("unable to open signed item")
+	}
 
 	// unmarshal result into queue item
-	err = json.Unmarshal([]byte(result[1]), item)
+	item := new(types.Item)
+
+	err = json.Unmarshal(opened, item)
 	if err != nil {
 		return nil, err
 	}