Add scaffolding for external host plugins and convert aws/azure over (#1581)

sdk/plugins/host/load.go

@@ -0,0 +1,213 @@
+package external_host_plugins
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"fmt"
+	"io/fs"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"runtime"
+	"strings"
+
+	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+)
+
+// NOTE: This package could probably use some reflect based bits to allow
+// loading other types of plugins later. That is an exercise left for future
+// refactoring, but worth calling out now.
+
+type pluginInfo struct {
+	containerFs  fs.FS
+	filename     string
+	creationFunc func() (pb.HostPluginServiceClient, error)
+}
+
+// CreateHostPlugin takes in a type, parses the various options to look for a
+// plugin matching that name, and returns a host plugin client, a cleanup
+// function to execute on shutdown of the enclosing program, and an error.
+func CreateHostPlugin(
+	ctx context.Context,
+	pluginType string,
+	opt ...Option,
+) (
+	hp pb.HostPluginServiceClient,
+	cleanup func() error,
+	retErr error,
+) {
+	defer func() {
+		if retErr != nil && cleanup != nil {
+			_ = cleanup()
+		}
+	}()
+
+	pluginType = strings.ToLower(pluginType)
+
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error parsing host plugin options: %w", err)
+	}
+	if len(opts.withHostPluginsSources) == 0 {
+		return nil, nil, fmt.Errorf("no host plugins available")
+	}
+
+	// First, scan available plugins, then find the right one to use
+	pluginMap := map[string]pluginInfo{}
+	var plugin pluginInfo
+	{
+		for _, sourceInfo := range opts.withHostPluginsSources {
+			switch {
+			case sourceInfo.pluginFs != nil:
+				dirs, err := fs.ReadDir(sourceInfo.pluginFs, ".")
+				if err != nil {
+					return nil, nil, fmt.Errorf("error scanning host plugins: %w", err)
+				}
+				// Store a match between the config type string and the expected plugin name
+				for _, entry := range dirs {
+					pluginType := strings.TrimSuffix(strings.TrimPrefix(entry.Name(), sourceInfo.pluginFsPrefix), ".gz")
+					if runtime.GOOS == "windows" {
+						pluginType = strings.TrimSuffix(pluginType, ".exe")
+					}
+					pluginMap[pluginType] = pluginInfo{
+						containerFs: sourceInfo.pluginFs,
+						filename:    entry.Name(),
+					}
+				}
+			case sourceInfo.pluginMap != nil:
+				for k, creationFunc := range sourceInfo.pluginMap {
+					pluginMap[k] = pluginInfo{creationFunc: creationFunc, filename: k}
+				}
+			}
+		}
+
+		plugin = pluginMap[pluginType]
+		if plugin.filename == "" && plugin.creationFunc == nil {
+			return nil, nil, fmt.Errorf("unknown host plugin type %q", pluginType)
+		}
+	}
+
+	// If the source is just a func, execute it and skip ahead; otherwise it's a plugin, so instantiate it
+	{
+		switch {
+		case plugin.creationFunc != nil:
+			hp, err = plugin.creationFunc()
+			if err != nil {
+				return nil, nil, fmt.Errorf("error performing direct instantiation of host plugin with type %q: %w", plugin.filename, err)
+			}
+
+		case plugin.containerFs != nil:
+			hp, cleanup, err = executeHostPlugin(plugin, opt...)
+			if err != nil {
+				return nil, cleanup, err
+			}
+		}
+	}
+
+	return hp, cleanup, nil
+}
+
+// executeHostPlugin takes in the discovered plugin information and spins out
+// the actual binary, returning a client that talks to it.
+func executeHostPlugin(plugin pluginInfo, opt ...Option) (pb.HostPluginServiceClient, func() error, error) {
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, nil, fmt.Errorf("error parsing host plugin options: %w", err)
+	}
+
+	// Open and basic validation
+	file, err := plugin.containerFs.Open(plugin.filename)
+	if err != nil {
+		return nil, nil, err
+	}
+	stat, err := file.Stat()
+	if err != nil {
+		return nil, nil, fmt.Errorf("error discovering host plugin information: %w", err)
+	}
+	if stat.IsDir() {
+		return nil, nil, fmt.Errorf("host plugin is a directory, not a file")
+	}
+
+	// Read in plugin bytes
+	expLen := stat.Size()
+	buf := make([]byte, expLen)
+	readLen, err := file.Read(buf)
+	if err != nil {
+		file.Close()
+		return nil, nil, fmt.Errorf("error reading host plugin bytes: %w", err)
+	}
+	if err := file.Close(); err != nil {
+		return nil, nil, fmt.Errorf("error closing host plugin file: %w", err)
+	}
+	if int64(readLen) != expLen {
+		return nil, nil, fmt.Errorf("reading host plugin expected %d bytes, read %d", expLen, readLen)
+	}
+
+	executedFileName := plugin.filename
+
+	// If it's compressed, uncompress it
+	if strings.HasSuffix(plugin.filename, ".gz") {
+		executedFileName = strings.TrimSuffix(plugin.filename, ".gz")
+		gzipReader, err := gzip.NewReader(bytes.NewReader(buf))
+		if err != nil {
+			return nil, nil, fmt.Errorf("error creating gzip decompression reader: %w", err)
+		}
+		uncompBuf := new(bytes.Buffer)
+		_, err = uncompBuf.ReadFrom(gzipReader)
+		gzipReader.Close()
+		if err != nil {
+			return nil, nil, fmt.Errorf("error reading gzip compressed data from reader: %w", err)
+		}
+		buf = uncompBuf.Bytes()
+	}
+
+	cleanup := func() error {
+		return nil
+	}
+
+	// Now, create a temp dir and write out the plugin bytes
+	dir := opts.withHostPluginExecutionDir
+	if dir == "" {
+		tmpDir, err := ioutil.TempDir("", "*")
+		if err != nil {
+			return nil, nil, fmt.Errorf("error creating tmp dir for kms execution: %w", err)
+		}
+		cleanup = func() error {
+			return os.RemoveAll(tmpDir)
+		}
+		dir = tmpDir
+	}
+	pluginPath := filepath.Join(dir, executedFileName)
+	if err := ioutil.WriteFile(pluginPath, buf, fs.FileMode(0700)); err != nil {
+		return nil, cleanup, fmt.Errorf("error writing out host plugin for execution: %w", err)
+	}
+
+	// Execute the plugin
+	client, err := NewHostPluginClient(pluginPath, WithLogger(opts.withLogger))
+	if err != nil {
+		return nil, cleanup, fmt.Errorf("error fetching host plugin client: %w", err)
+	}
+	origCleanup := cleanup
+	cleanup = func() error {
+		client.Kill()
+		return origCleanup()
+	}
+	rpcClient, err := client.Client()
+	if err != nil {
+		return nil, cleanup, fmt.Errorf("error fetching host plugin rpc client: %w", err)
+	}
+
+	raw, err := rpcClient.Dispense(hostServicePluginSetName)
+	if err != nil {
+		return nil, cleanup, fmt.Errorf("error dispensing host plugin: %w", err)
+	}
+
+	var ok bool
+	hp, ok := raw.(pb.HostPluginServiceClient)
+	if !ok {
+		return nil, cleanup, fmt.Errorf("error converting rpc host plugin to host plugin interface: %w", err)
+	}
+
+	return hp, cleanup, nil
+}

sdk/plugins/host/options.go

@@ -0,0 +1,119 @@
+package external_host_plugins
+
+import (
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+
+	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/hashicorp/go-hclog"
+)
+
+// getOpts iterates the inbound Options and returns a struct
+func getOpts(opt ...Option) (*options, error) {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		if o == nil {
+			continue
+		}
+		if err := o(opts); err != nil {
+			return nil, fmt.Errorf("error running option function: %w", err)
+		}
+	}
+	return opts, nil
+}
+
+// Option - a type that wraps an interface for compile-time safety but can
+// contain an option for this package or for wrappers implementing this
+// interface.
+type Option func(*options) error
+
+type options struct {
+	withHostPluginsSources     []pluginSourceInfo
+	withHostPluginExecutionDir string
+	withLogger                 hclog.Logger
+}
+
+func getDefaultOptions() *options {
+	return &options{}
+}
+
+// pluginSourceInfo contains information about plugin sources. Each entry in the
+// final slice of withHostPluginsSources will either have the Fs/FsPrefix values
+// populated or have the map populated. The former is for loading and executing
+// a plugin via go-plugin from the filesystem; the latter is for executing it
+// directly in-memory.
+type pluginSourceInfo struct {
+	pluginMap      map[string]func() (pb.HostPluginServiceClient, error)
+	pluginFs       fs.FS
+	pluginFsPrefix string
+}
+
+// WithLogger allows passing a logger to the plugin library for debugging
+func WithLogger(logger hclog.Logger) Option {
+	return func(o *options) error {
+		o.withLogger = logger
+		return nil
+	}
+}
+
+// WithHostPluginsFilesystem provides an fs.FS containing plugins that can be
+// executed to provide Host functionality. This can be specified multiple times;
+// all FSes will be scanned. If there are conflicts, the last one wins (this
+// property is shared with WithHostPluginsMap). The prefix will be stripped from
+// each entry when determining the plugin type.
+func WithHostPluginsFilesystem(prefix string, plugins fs.FS) Option {
+	return func(o *options) error {
+		if plugins == nil {
+			return errors.New("nil host plugin filesystem passed into option")
+		}
+		o.withHostPluginsSources = append(o.withHostPluginsSources,
+			pluginSourceInfo{
+				pluginFs:       plugins,
+				pluginFsPrefix: prefix,
+			},
+		)
+		return nil
+	}
+}
+
+// WithHostPluginsMap provides a map containing functions that can be called to
+// provide implementations of the server. This can be specified multiple times;
+// all FSes will be scanned. If there are conflicts, the last one wins (this
+// property is shared with WithHostPluginsFilesystem).
+func WithHostPluginsMap(plugins map[string]func() (pb.HostPluginServiceClient, error)) Option {
+	return func(o *options) error {
+		if len(plugins) == 0 {
+			return errors.New("no entries in host plugins map passed into option")
+		}
+		o.withHostPluginsSources = append(o.withHostPluginsSources,
+			pluginSourceInfo{
+				pluginMap: plugins,
+			},
+		)
+		return nil
+	}
+}
+
+// WithHostPluginExecutionDir allows setting a specific directory for
+// writing out and executing plugins; if not set, os.TempDir will be used
+// to create a suitable directory.
+func WithHostPluginExecutionDir(dir string) Option {
+	return func(o *options) error {
+		if dir == "" {
+			// We always call this with the option, so if it's not actually set,
+			// don't error
+			return nil
+		}
+		fi, err := os.Stat(dir)
+		if err != nil {
+			return fmt.Errorf("error while performing stat to validate path %q is a directory: %w", dir, err)
+		}
+		if !fi.IsDir() {
+			return fmt.Errorf("given plugin execution path %q is not a directory", dir)
+		}
+		o.withHostPluginExecutionDir = dir
+		return nil
+	}
+}

sdk/plugins/host/plugin.go

@@ -0,0 +1,89 @@
+package external_host_plugins
+
+import (
+	"context"
+	"fmt"
+	"os/exec"
+
+	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/hashicorp/go-plugin"
+	"google.golang.org/grpc"
+)
+
+const (
+	hostServicePluginSetName = "host-plugin"
+)
+
+// HandshakeConfig is a shared config that can be used regardless of plugin, to
+// avoid having to know type-specific things about each plugin
+var HandshakeConfig = plugin.HandshakeConfig{
+	MagicCookieKey:   "HASHICORP_BOUNDARY_HOST_PLUGIN",
+	MagicCookieValue: hostServicePluginSetName,
+}
+
+// ServeHostPlugin is a generic function to start serving a host plugin service as a
+// plugin
+func ServeHostPlugin(svc pb.HostPluginServiceServer, opt ...Option) error {
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return err
+	}
+	hostServiceServer, err := NewHostPluginServiceServer(svc)
+	if err != nil {
+		return err
+	}
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: HandshakeConfig,
+		VersionedPlugins: map[int]plugin.PluginSet{
+			1: {hostServicePluginSetName: hostServiceServer},
+		},
+		Logger:     opts.withLogger,
+		GRPCServer: plugin.DefaultGRPCServer,
+	})
+	return nil
+}
+
+type hostPlugin struct {
+	plugin.Plugin
+
+	impl pb.HostPluginServiceServer
+}
+
+func NewHostPluginServiceServer(impl pb.HostPluginServiceServer) (*hostPlugin, error) {
+	if impl == nil {
+		return nil, fmt.Errorf("empty underlying host plugin passed in")
+	}
+	return &hostPlugin{
+		impl: impl,
+	}, nil
+}
+
+func NewHostPluginClient(pluginPath string, opt ...Option) (*plugin.Client, error) {
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, err
+	}
+	hostServiceClient := &hostPlugin{}
+
+	return plugin.NewClient(&plugin.ClientConfig{
+		HandshakeConfig: HandshakeConfig,
+		VersionedPlugins: map[int]plugin.PluginSet{
+			1: {hostServicePluginSetName: hostServiceClient},
+		},
+		Cmd: exec.Command(pluginPath),
+		AllowedProtocols: []plugin.Protocol{
+			plugin.ProtocolGRPC,
+		},
+		Logger:   opts.withLogger,
+		AutoMTLS: true,
+	}), nil
+}
+
+func (h *hostPlugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error {
+	pb.RegisterHostPluginServiceServer(s, h.impl)
+	return nil
+}
+
+func (h *hostPlugin) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error) {
+	return pb.NewHostPluginServiceClient(c), nil
+}