Pluggable secret backend

This commit extends SwarmKit secret management with pluggable secret
backends support. The solution uses the existing docker plugin
framework for loading plugins and the existing SwarmKit data backend for
storing them.

The approach is to add a new `driver` parameter to existing secrets,
which defines whether the values are taken as is or fetched from one of
the secret plugins. The loading of secrets is done using the standard
docker plugin infrastructure, which is already accessible in SwarmKit
and used in other flows (e.g., networking).
The fetched values are evaluated before assigning them to worker nodes,
so the payload is not stored in the raft store.

Remarks:
* I've added support for mocking the plugin subsystem when settings up
the controlapi server.
I preferred this approach over loading the full plugin subsystem in UT.

Work still needed in this CR:
- [ ] More unit tests (pending initial iteration)
- [ ] Customized error handling (e.g., customize error string for Not
Found)

Work still needed to complete this feature:
- [ ] Inject secrets as part of plugin initialization
- [ ] CLI support in docker
- [ ] Docs
- [ ] Support scheduling plugins in swarm
moby/moby#33575

Signed-off-by: liron <liron@twistlock.com>

api/specs.proto

@@ -393,6 +393,9 @@ message SecretSpec {
 	// The currently recognized values are:
 	// - golang: Go templating
 	Driver templating = 3;
+
+	// Driver is the the secret driver that is used to store the specified secret
+	Driver driver = 4;
 }
 
 // ConfigSpec specifies user-provided configuration files.

api/validation/secrets.go

@@ -0,0 +1,14 @@
+package validation
+
+import "fmt"
+
+// MaxSecretSize is the maximum byte length of the `Secret.Spec.Data` field.
+const MaxSecretSize = 500 * 1024 // 500KB
+
+// ValidateSecretPayload validates the secret payload size
+func ValidateSecretPayload(data []byte) error {
+	if len(data) >= MaxSecretSize || len(data) < 1 {
+		return fmt.Errorf("secret data must be larger than 0 and less than %d bytes", MaxSecretSize)
+	}
+	return nil
+}

cmd/swarmctl/secret/create.go

@@ -24,8 +24,13 @@ var createCmd = &cobra.Command{
 		var (
 			secretData []byte
 			err        error
+			driver     string
 		)
 
+		driver, err = flags.GetString("driver")
+		if err != nil {
+			return fmt.Errorf("Error reading secret driver %s", err.Error())
+		}
 		if flags.Changed("file") {
 			filename, err := flags.GetString("file")
 			if err != nil {
@@ -35,7 +40,7 @@ var createCmd = &cobra.Command{
 			if err != nil {
 				return fmt.Errorf("Error reading from file '%s': %s", filename, err.Error())
 			}
-		} else {
+		} else if driver == "" {
 			secretData, err = ioutil.ReadAll(os.Stdin)
 			if err != nil {
 				return fmt.Errorf("Error reading content from STDIN: %s", err.Error())
@@ -51,6 +56,9 @@ var createCmd = &cobra.Command{
 			Annotations: api.Annotations{Name: args[0]},
 			Data:        secretData,
 		}
+		if driver != "" {
+			spec.Driver = &api.Driver{Name: driver}
+		}
 
 		resp, err := client.CreateSecret(common.Context(cmd), &api.CreateSecretRequest{Spec: spec})
 		if err != nil {
@@ -63,4 +71,5 @@ var createCmd = &cobra.Command{
 
 func init() {
 	createCmd.Flags().StringP("file", "f", "", "Rather than read the secret from STDIN, read from the given file")
+	createCmd.Flags().StringP("driver", "d", "", "Rather than read the secret from STDIN, read the value from an external secret driver")
 }

cmd/swarmctl/secret/list.go

@@ -63,15 +63,21 @@ var (
 					// Ignore flushing errors - there's nothing we can do.
 					_ = w.Flush()
 				}()
-				common.PrintHeader(w, "ID", "Name", "Created")
+				common.PrintHeader(w, "ID", "Name", "Driver", "Created")
 				output = func(s *api.Secret) {
 					created, err := gogotypes.TimestampFromProto(s.Meta.CreatedAt)
 					if err != nil {
 						panic(err)
 					}
-					fmt.Fprintf(w, "%s\t%s\t%s\n",
+					var driver string
+					if s.Spec.Driver != nil {
+						driver = s.Spec.Driver.Name
+					}
+
+					fmt.Fprintf(w, "%s\t%s\t%s\t%s\n",
 						s.ID,
 						s.Spec.Annotations.Name,
+						driver,
 						humanize.Time(created),
 					)
 				}

manager/controlapi/secret.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/swarmkit/api"
+	"github.com/docker/swarmkit/api/validation"
 	"github.com/docker/swarmkit/identity"
 	"github.com/docker/swarmkit/log"
 	"github.com/docker/swarmkit/manager/state/store"
@@ -14,9 +15,6 @@ import (
 	"google.golang.org/grpc/codes"
 )
 
-// MaxSecretSize is the maximum byte length of the `Secret.Spec.Data` field.
-const MaxSecretSize = 500 * 1024 // 500KB
-
 // assumes spec is not nil
 func secretFromSecretSpec(spec *api.SecretSpec) *api.Secret {
 	return &api.Secret{
@@ -56,7 +54,6 @@ func (s *Server) UpdateSecret(ctx context.Context, request *api.UpdateSecretRequ
 	if request.SecretID == "" || request.SecretVersion == nil {
 		return nil, grpc.Errorf(codes.InvalidArgument, errInvalidArgument.Error())
 	}
-
 	var secret *api.Secret
 	err := s.store.Update(func(tx store.Tx) error {
 		secret = store.GetSecret(tx, request.SecretID)
@@ -245,9 +242,16 @@ func validateSecretSpec(spec *api.SecretSpec) error {
 	if err := validateConfigOrSecretAnnotations(spec.Annotations); err != nil {
 		return err
 	}
-
-	if len(spec.Data) >= MaxSecretSize || len(spec.Data) < 1 {
-		return grpc.Errorf(codes.InvalidArgument, "secret data must be larger than 0 and less than %d bytes", MaxSecretSize)
+	// Check if secret driver is defined
+	if spec.Driver != nil {
+		// Ensure secret driver has a name
+		if spec.Driver.Name == "" {
+			return grpc.Errorf(codes.InvalidArgument, "secret driver must have a name")
+		}
+		return nil
+	}
+	if err := validation.ValidateSecretPayload(spec.Data); err != nil {
+		return grpc.Errorf(codes.InvalidArgument, "%s", err.Error())
 	}
 	return nil
 }

manager/controlapi/secret_test.go

@@ -89,6 +89,16 @@ func TestValidateSecretSpec(t *testing.T) {
 		err := validateSecretSpec(good)
 		assert.NoError(t, err)
 	}
+
+	// Ensure secret driver has a name
+	spec := createSecretSpec("secret-driver", make([]byte, 1), nil)
+	spec.Driver = &api.Driver{}
+	err := validateSecretSpec(spec)
+	assert.Error(t, err)
+	assert.Equal(t, codes.InvalidArgument, grpc.Code(err), grpc.ErrorDesc(err))
+	spec.Driver.Name = "secret-driver"
+	err = validateSecretSpec(spec)
+	assert.NoError(t, err)
 }
 
 func TestCreateSecret(t *testing.T) {

manager/controlapi/server.go

@@ -10,7 +10,6 @@ import (
 )
 
 var (
-	errNotImplemented  = errors.New("not implemented")
 	errInvalidArgument = errors.New("invalid argument")
 )
 

manager/dispatcher/assignments.go

@@ -1,9 +1,12 @@
 package dispatcher
 
 import (
+	"fmt"
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/api/equality"
+	"github.com/docker/swarmkit/api/validation"
+	"github.com/docker/swarmkit/manager/drivers"
 	"github.com/docker/swarmkit/manager/state/store"
 )
 
@@ -24,15 +27,16 @@ type typeAndID struct {
 }
 
 type assignmentSet struct {
+	dp                   *drivers.DriverProvider
 	tasksMap             map[string]*api.Task
 	tasksUsingDependency map[typeAndID]map[string]struct{}
 	changes              map[typeAndID]*api.AssignmentChange
-
-	log *logrus.Entry
+	log                  *logrus.Entry
 }
 
-func newAssignmentSet(log *logrus.Entry) *assignmentSet {
+func newAssignmentSet(log *logrus.Entry, dp *drivers.DriverProvider) *assignmentSet {
 	return &assignmentSet{
+		dp:                   dp,
 		changes:              make(map[typeAndID]*api.AssignmentChange),
 		tasksMap:             make(map[string]*api.Task),
 		tasksUsingDependency: make(map[typeAndID]map[string]struct{}),
@@ -53,12 +57,13 @@ func (a *assignmentSet) addTaskDependencies(readTx store.ReadTx, t *api.Task) {
 		if len(a.tasksUsingDependency[mapKey]) == 0 {
 			a.tasksUsingDependency[mapKey] = make(map[string]struct{})
 
-			secret := store.GetSecret(readTx, secretID)
-			if secret == nil {
+			secret, err := a.secret(readTx, secretID)
+			if err != nil {
 				a.log.WithFields(logrus.Fields{
 					"secret.id":   secretID,
 					"secret.name": secretRef.SecretName,
-				}).Debug("secret not found")
+					"error":       err,
+				}).Error("failed to fetch secret")
 				continue
 			}
 
@@ -245,3 +250,29 @@ func (a *assignmentSet) message() api.AssignmentsMessage {
 
 	return message
 }
+
+// secret populates the secret value from raft store. For external secrets, the value is populated
+// from the secret driver.
+func (a *assignmentSet) secret(readTx store.ReadTx, secretID string) (*api.Secret, error) {
+	secret := store.GetSecret(readTx, secretID)
+	if secret == nil {
+		return nil, fmt.Errorf("secret not found")
+	}
+	if secret.Spec.Driver == nil {
+		return secret, nil
+	}
+	d, err := a.dp.NewSecretDriver(secret.Spec.Driver)
+	if err != nil {
+		return nil, err
+	}
+	value, err := d.Get(&secret.Spec)
+	if err != nil {
+		return nil, err
+	}
+	if err := validation.ValidateSecretPayload(value); err != nil {
+		return nil, err
+	}
+	// Assign the secret
+	secret.Spec.Data = value
+	return secret, nil
+}

manager/dispatcher/dispatcher.go

@@ -17,6 +17,7 @@ import (
 	"github.com/docker/swarmkit/api/equality"
 	"github.com/docker/swarmkit/ca"
 	"github.com/docker/swarmkit/log"
+	"github.com/docker/swarmkit/manager/drivers"
 	"github.com/docker/swarmkit/manager/state/store"
 	"github.com/docker/swarmkit/remotes"
 	"github.com/docker/swarmkit/watch"
@@ -125,6 +126,7 @@ type Dispatcher struct {
 	ctx                  context.Context
 	cancel               context.CancelFunc
 	clusterUpdateQueue   *watch.Queue
+	dp                   *drivers.DriverProvider
 
 	taskUpdates     map[string]*api.TaskStatus // indexed by task ID
 	taskUpdatesLock sync.Mutex
@@ -142,8 +144,9 @@ type Dispatcher struct {
 }
 
 // New returns Dispatcher with cluster interface(usually raft.Node).
-func New(cluster Cluster, c *Config) *Dispatcher {
+func New(cluster Cluster, c *Config, dp *drivers.DriverProvider) *Dispatcher {
 	d := &Dispatcher{
+		dp:                    dp,
 		nodes:                 newNodeStore(c.HeartbeatPeriod, c.HeartbeatEpsilon, c.GracePeriodMultiplier, c.RateLimitPeriod),
 		downNodes:             newNodeStore(defaultNodeDownPeriod, 0, 1, 0),
 		store:                 cluster.MemoryStore(),
@@ -836,7 +839,7 @@ func (d *Dispatcher) Assignments(r *api.AssignmentsRequest, stream api.Dispatche
 	var (
 		sequence    int64
 		appliesTo   string
-		assignments = newAssignmentSet(log)
+		assignments = newAssignmentSet(log, d.dp)
 	)
 
 	sendMessage := func(msg api.AssignmentsMessage, assignmentType api.AssignmentsMessage_Type) error {