From 1987264bc357c5266a13053c0f350491ea6acaac Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Fri, 3 Jun 2022 14:44:53 -0400 Subject: [PATCH] Migrate storage to database repo (#2132) --- api/util.go | 8 --- internal/daemon/controller/common/common.go | 25 ++++---- internal/daemon/controller/controller.go | 62 ++++++++++++------- internal/daemon/controller/listeners.go | 10 ++- internal/daemon/controller/listeners_test.go | 4 -- internal/daemon/worker/worker.go | 13 ++-- internal/servers/repository_worker.go | 8 +++ internal/servers/repository_workerauth.go | 24 ++++--- .../servers/repository_workerauth_test.go | 14 ++++- internal/servers/worker.go | 2 +- 10 files changed, 102 insertions(+), 68 deletions(-) diff --git a/api/util.go b/api/util.go index 4dd69368c..67669db03 100644 --- a/api/util.go +++ b/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 diff --git a/internal/daemon/controller/common/common.go b/internal/daemon/controller/common/common.go index 4945b44d9..f520d4f99 100644 --- a/internal/daemon/controller/common/common.go +++ b/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) ) diff --git a/internal/daemon/controller/controller.go b/internal/daemon/controller/controller.go index 79daf9700..212fdcc55 100644 --- a/internal/daemon/controller/controller.go +++ b/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 } diff --git a/internal/daemon/controller/listeners.go b/internal/daemon/controller/listeners.go index b4bd1046a..3475819b2 100644 --- a/internal/daemon/controller/listeners.go +++ b/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, ) diff --git a/internal/daemon/controller/listeners_test.go b/internal/daemon/controller/listeners_test.go index db1119291..30a3b769c 100644 --- a/internal/daemon/controller/listeners_test.go +++ b/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) diff --git a/internal/daemon/worker/worker.go b/internal/daemon/worker/worker.go index dce7259a1..5cca1f524 100644 --- a/internal/daemon/worker/worker.go +++ b/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) } diff --git a/internal/servers/repository_worker.go b/internal/servers/repository_worker.go index 3740af566..e6f20568d 100644 --- a/internal/servers/repository_worker.go +++ b/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")) } diff --git a/internal/servers/repository_workerauth.go b/internal/servers/repository_workerauth.go index 44bd70520..ef5db2bb9 100644 --- a/internal/servers/repository_workerauth.go +++ b/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 } diff --git a/internal/servers/repository_workerauth_test.go b/internal/servers/repository_workerauth_test.go index 11f7bea3b..f6d54a4be 100644 --- a/internal/servers/repository_workerauth_test.go +++ b/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, diff --git a/internal/servers/worker.go b/internal/servers/worker.go index 4789f810f..4d7beabfd 100644 --- a/internal/servers/worker.go +++ b/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