Migrate storage to database repo (#2132)

api/util.go

@@ -1,13 +1,5 @@
 package api
 
-import (
-	"github.com/fatih/structs"
-)
-
-func init() {
-	structs.DefaultTagName = "json"
-}
-
 type GenericResult interface {
 	GetItem() interface{}
 	GetResponse() *Response

internal/daemon/controller/common/common.go

@@ -14,16 +14,17 @@ import (
 )
 
 type (
-	AuthTokenRepoFactory       = oidc.AuthTokenRepoFactory
-	VaultCredentialRepoFactory = func() (*vault.Repository, error)
-	IamRepoFactory             func() (*iam.Repository, error)
-	OidcAuthRepoFactory        = oidc.OidcRepoFactory
-	PasswordAuthRepoFactory    func() (*password.Repository, error)
-	ServersRepoFactory         func() (*servers.Repository, error)
-	StaticRepoFactory          func() (*static.Repository, error)
-	PluginHostRepoFactory      func() (*pluginhost.Repository, error)
-	HostPluginRepoFactory      func() (*hostplugin.Repository, error)
-	SessionRepoFactory         func() (*session.Repository, error)
-	ConnectionRepoFactory      func() (*session.ConnectionRepository, error)
-	TargetRepoFactory          func() (*target.Repository, error)
+	AuthTokenRepoFactory         = oidc.AuthTokenRepoFactory
+	VaultCredentialRepoFactory   = func() (*vault.Repository, error)
+	IamRepoFactory               func() (*iam.Repository, error)
+	OidcAuthRepoFactory          = oidc.OidcRepoFactory
+	PasswordAuthRepoFactory      func() (*password.Repository, error)
+	ServersRepoFactory           func() (*servers.Repository, error)
+	StaticRepoFactory            func() (*static.Repository, error)
+	PluginHostRepoFactory        func() (*pluginhost.Repository, error)
+	HostPluginRepoFactory        func() (*hostplugin.Repository, error)
+	SessionRepoFactory           func() (*session.Repository, error)
+	ConnectionRepoFactory        func() (*session.ConnectionRepository, error)
+	TargetRepoFactory            func() (*target.Repository, error)
+	WorkerAuthRepoStorageFactory func() (*servers.WorkerAuthRepositoryStorage, error)
 )

internal/daemon/controller/controller.go

@@ -27,6 +27,7 @@ import (
 	"github.com/hashicorp/boundary/internal/scheduler"
 	"github.com/hashicorp/boundary/internal/scheduler/job"
 	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/servers/store"
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/types/scope"
@@ -36,9 +37,7 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
 	"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
-	"github.com/hashicorp/nodeenrollment/registration"
 	"github.com/hashicorp/nodeenrollment/rotation"
-	nodeefile "github.com/hashicorp/nodeenrollment/storage/file"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mr-tron/base58"
 	ua "go.uber.org/atomic"
@@ -70,18 +69,19 @@ type Controller struct {
 	apiGrpcGatewayTicket  string
 
 	// Repo factory methods
-	AuthTokenRepoFn       common.AuthTokenRepoFactory
-	VaultCredentialRepoFn common.VaultCredentialRepoFactory
-	IamRepoFn             common.IamRepoFactory
-	OidcRepoFn            common.OidcAuthRepoFactory
-	PasswordAuthRepoFn    common.PasswordAuthRepoFactory
-	ServersRepoFn         common.ServersRepoFactory
-	SessionRepoFn         common.SessionRepoFactory
-	ConnectionRepoFn      common.ConnectionRepoFactory
-	StaticHostRepoFn      common.StaticRepoFactory
-	PluginHostRepoFn      common.PluginHostRepoFactory
-	HostPluginRepoFn      common.HostPluginRepoFactory
-	TargetRepoFn          common.TargetRepoFactory
+	AuthTokenRepoFn         common.AuthTokenRepoFactory
+	VaultCredentialRepoFn   common.VaultCredentialRepoFactory
+	IamRepoFn               common.IamRepoFactory
+	OidcRepoFn              common.OidcAuthRepoFactory
+	PasswordAuthRepoFn      common.PasswordAuthRepoFactory
+	ServersRepoFn           common.ServersRepoFactory
+	SessionRepoFn           common.SessionRepoFactory
+	ConnectionRepoFn        common.ConnectionRepoFactory
+	StaticHostRepoFn        common.StaticRepoFactory
+	PluginHostRepoFn        common.PluginHostRepoFactory
+	HostPluginRepoFn        common.HostPluginRepoFactory
+	TargetRepoFn            common.TargetRepoFactory
+	WorkerAuthRepoStorageFn common.WorkerAuthRepoStorageFactory
 
 	scheduler *scheduler.Scheduler
 
@@ -92,9 +92,6 @@ type Controller struct {
 	// Used to signal the Health Service to start
 	// replying to queries with "503 Service Unavailable".
 	HealthService *health.Service
-
-	// PoC: Testing bits for BYOW
-	NodeeFileStorage *nodeefile.FileStorage
 }
 
 func New(ctx context.Context, conf *Config) (*Controller, error) {
@@ -304,6 +301,9 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 	c.ConnectionRepoFn = func() (*session.ConnectionRepository, error) {
 		return session.NewConnectionRepository(ctx, dbase, dbase, c.kms)
 	}
+	c.WorkerAuthRepoStorageFn = func() (*servers.WorkerAuthRepositoryStorage, error) {
+		return servers.NewRepositoryStorage(ctx, dbase, dbase, c.kms)
+	}
 
 	return c, nil
 }
@@ -317,12 +317,11 @@ func (c *Controller) Start() error {
 
 	c.baseContext, c.baseCancel = context.WithCancel(context.Background())
 
-	var err error
-	c.NodeeFileStorage, err = nodeefile.NewFileStorage(c.baseContext)
+	workerAuthStorage, err := c.WorkerAuthRepoStorageFn()
 	if err != nil {
-		return err
+		return fmt.Errorf("error fetching worker auth storage: %w", err)
 	}
-	_, err = rotation.RotateRootCertificates(c.baseContext, c.NodeeFileStorage)
+	_, err = rotation.RotateRootCertificates(c.baseContext, workerAuthStorage)
 	if err != nil {
 		return err
 	}
@@ -410,17 +409,32 @@ func (c *Controller) WorkerStatusUpdateTimes() *sync.Map {
 	return c.workerStatusUpdateTimes
 }
 
+// This is a temporary function until the API is up
 func (c *Controller) AuthorizeNodeeWorker(request string) error {
-	const op = "controller.(Controller).AuthorizeWorker"
+	const op = "controller.(Controller).AuthorizeNodeeWorker"
 	reqBytes, err := base58.FastBase58Decoding(request)
 	if err != nil {
-		return fmt.Errorf("(%s) error base64-decoding fetch node creds next proto value", op)
+		return fmt.Errorf("(%s) error base58-decoding fetch node creds next proto value: %w", op, err)
 	}
 	// Decode the proto into the request
 	req := new(types.FetchNodeCredentialsRequest)
 	if err := proto.Unmarshal(reqBytes, req); err != nil {
 		return fmt.Errorf("(%s) error unmarshaling common name value: %w", op, err)
 	}
-	_, err = registration.AuthorizeNode(c.baseContext, c.NodeeFileStorage, req)
+
+	serversRepo, err := c.ServersRepoFn()
+	if err != nil {
+		return fmt.Errorf("(%s) error fetching servers repo: %w", op, err)
+	}
+
+	_, err = serversRepo.CreateWorker(c.baseContext, &servers.Worker{
+		Worker: &store.Worker{
+			ScopeId: scope.Global.String(),
+		},
+	}, servers.WithFetchNodeCredentialsRequest(req))
+	if err != nil {
+		return fmt.Errorf("(%s) error creating worker: %w", op, err)
+	}
+
 	return err
 }

internal/daemon/controller/listeners.go

@@ -106,10 +106,16 @@ func (c *Controller) configureForApi(ln *base.ServerListener) ([]func(), error)
 
 func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error) {
 	const op = "controller.configureForCluster"
+
+	workerAuthStorage, err := c.WorkerAuthRepoStorageFn()
+	if err != nil {
+		return nil, fmt.Errorf("error fetching worker auth storage: %w", err)
+	}
+
 	l, err := protocol.NewInterceptingListener(
 		&protocol.InterceptingListenerConfiguration{
 			Context:      c.baseContext,
-			Storage:      c.NodeeFileStorage,
+			Storage:      workerAuthStorage,
 			BaseListener: ln.ClusterListener,
 			BaseTlsConfiguration: &tls.Config{
 				GetConfigForClient: c.validateWorkerTls,
@@ -143,7 +149,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 	pbs.RegisterSessionServiceServer(workerServer, workerService)
 
 	multihopService, err := handlers.NewMultihopServiceServer(
-		c.NodeeFileStorage,
+		workerAuthStorage,
 		true,
 		nil,
 	)

internal/daemon/controller/listeners_test.go

@@ -23,7 +23,6 @@ import (
 	"github.com/hashicorp/go-secure-stdlib/base62"
 	"github.com/hashicorp/go-secure-stdlib/configutil/v2"
 	"github.com/hashicorp/go-secure-stdlib/listenerutil"
-	nodeefile "github.com/hashicorp/nodeenrollment/storage/file"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/proto"
@@ -251,9 +250,6 @@ func TestStartListeners(t *testing.T) {
 			c.baseContext = ctx
 			c.baseCancel = cancel
 
-			c.NodeeFileStorage, err = nodeefile.NewFileStorage(c.baseContext)
-			require.NoError(t, err)
-
 			err = c.startListeners()
 			require.NoError(t, err)
 

internal/daemon/worker/worker.go

@@ -161,6 +161,12 @@ func New(conf *Config) (*Worker, error) {
 		return nil, fmt.Errorf("exactly one proxy listener is required")
 	}
 
+	w.NodeeFileStorage, err = nodeefile.NewFileStorage(w.baseContext,
+		nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.StoragePath))
+	if err != nil {
+		return nil, err
+	}
+
 	return w, nil
 }
 
@@ -178,13 +184,6 @@ func (w *Worker) Start() error {
 	controllerResolver := manual.NewBuilderWithScheme(scheme)
 	w.controllerResolver.Store(controllerResolver)
 
-	var err error
-	w.NodeeFileStorage, err = nodeefile.NewFileStorage(w.baseContext,
-		nodeefile.WithBaseDirectory(w.conf.RawConfig.Worker.StoragePath))
-	if err != nil {
-		return err
-	}
-
 	if err := w.startListeners(); err != nil {
 		return fmt.Errorf("error starting worker listeners: %w", err)
 	}

internal/servers/repository_worker.go

@@ -14,6 +14,7 @@ import (
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
 	"github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/registration"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 // DeleteWorker will delete a worker from the repository.
@@ -413,6 +414,12 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op
 	if worker.PublicId, err = opts.withNewIdFunc(ctx); err != nil {
 		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to generate worker id"))
 	}
+	state, err := structpb.NewStruct(map[string]any{
+		"worker_id": worker.PublicId,
+	})
+	if err != nil {
+		return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error generating state struct", err))
+	}
 
 	var databaseWrapper wrapping.Wrapper
 	var workerAuthRepo *WorkerAuthRepositoryStorage
@@ -447,6 +454,7 @@ func (r *Repository) CreateWorker(ctx context.Context, worker *Worker, opt ...Op
 				if err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to authorize node"))
 				}
+				nodeInfo.State = state
 				if err := StoreNodeInformationTx(ctx, w, databaseWrapper, nodeInfo); err != nil {
 					return errors.Wrap(ctx, err, op, errors.WithMsg("unable to store node information"))
 				}

internal/servers/repository_workerauth.go

@@ -18,6 +18,7 @@ import (
 	"github.com/hashicorp/boundary/internal/types/scope"
 	"github.com/hashicorp/go-dbw"
 	wrapping "github.com/hashicorp/go-kms-wrapping/v2"
+	"github.com/hashicorp/nodeenrollment"
 	nodee "github.com/hashicorp/nodeenrollment"
 	"github.com/hashicorp/nodeenrollment/types"
 	"github.com/mitchellh/mapstructure"
@@ -30,7 +31,7 @@ var (
 )
 
 type rootCertificatesVersion struct {
-	Version uint32
+	Version uint32 `mapstructure:"version" structs:"version"`
 }
 
 // WorkerAuthRepositoryStorage is the Worker Auth database repository
@@ -340,21 +341,26 @@ func (r *WorkerAuthRepositoryStorage) Load(ctx context.Context, msg nodee.Messag
 		return errors.New(ctx, errors.InvalidParameter, op, "given message cannot be loaded as it has no ID")
 	}
 
+	var err error
 	switch t := msg.(type) {
 	case *types.NodeInformation:
-		err := r.loadNodeInformation(ctx, t)
-		if err != nil {
-			return errors.Wrap(ctx, err, op)
-		}
+		err = r.loadNodeInformation(ctx, t)
+
 	case *types.RootCertificates:
-		err := r.loadRootCertificates(ctx, t)
-		if err != nil {
-			return errors.Wrap(ctx, err, op)
-		}
+		err = r.loadRootCertificates(ctx, t)
+
 	default:
 		return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("message type %T not supported for Load", t))
 	}
 
+	if err != nil {
+		if err == nodeenrollment.ErrNotFound {
+			// Don't wrap as this will confuse things
+			return err
+		}
+		return errors.Wrap(ctx, err, op)
+	}
+
 	return nil
 }
 

internal/servers/repository_workerauth_test.go

@@ -346,13 +346,25 @@ func TestStoreNodeInformationTx(t *testing.T) {
 			wantErrContains: "encrypt-error",
 		},
 		{
-			name:            "create-error",
+			name:            "create-error-no-db",
 			writer:          &db.Db{},
 			databaseWrapper: databaseWrapper,
 			node:            testNodeInfoFn(),
 			wantErr:         true,
 			wantErrContains: "db.Create",
 		},
+		{
+			name:            "create-error-validation",
+			writer:          rw,
+			databaseWrapper: databaseWrapper,
+			node: func() *types.NodeInformation {
+				ni := testNodeInfoFn()
+				ni.State = nil
+				return ni
+			}(),
+			wantErr:         true,
+			wantErrContains: "missing WorkerId",
+		},
 		{
 			name:            "success",
 			writer:          rw,

internal/servers/worker.go

@@ -13,7 +13,7 @@ import (
 )
 
 type workerAuthWorkerId struct {
-	WorkerId string
+	WorkerId string `mapstructure:"worker_id" structs:"worker_id"`
 }
 
 // AttachWorkerIdToState accepts a workerId and creates a struct for use with the Nodeenrollment lib