From bdf7bf7702d8159c80beb09ae6ad069bbb2b3fe9 Mon Sep 17 00:00:00 2001 From: Jeff Mitchell Date: Mon, 3 Apr 2023 17:44:41 -0400 Subject: [PATCH] Do not allow PKI-KMS workers to have names updated via API (#3143) Do not allow PKI-KMS workers to have names updated via API This would cause their public IDs to change, which we do not want. Descriptions are also disallowed, since those come from the config file. This also adds logic to allow upgrading from old KMS method to new, with a test. Co-authored-by: Irena Rindos --- CHANGELOG.md | 6 +- Makefile | 2 +- .../handlers/workers/worker_service.go | 20 +- .../handlers/workers/worker_service_test.go | 318 +++++++++++------- internal/daemon/worker/testing.go | 4 +- internal/server/public_ids.go | 6 +- internal/server/repository_worker.go | 40 ++- internal/server/repository_worker_test.go | 38 ++- internal/server/repository_workerauth.go | 27 +- .../cluster/multi_controller_worker_test.go | 3 +- .../tests/cluster/switch_kms_method_test.go | 93 +++++ internal/tests/cluster/unix_listener_test.go | 3 +- internal/tests/cluster/worker_replay_test.go | 2 +- 13 files changed, 421 insertions(+), 141 deletions(-) create mode 100644 internal/tests/cluster/switch_kms_method_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index d6de2f11d..88a10c93b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,7 +11,11 @@ Canonical reference for changes, improvements, and bugfixes for Boundary. only recommended if compatibility with pre-0.13 workers using the KMS auth method is required. Requiring opting in removes some potentially confusing behavior for deciding when to use the old versus new mechanism. To opt in, add - `use_deprecated_kms_auth_method = true` to the `worker` config block. + `use_deprecated_kms_auth_method = true` to the `worker` config block. Note + that if a 0.13+ worker using KMS connects to a 0.13+ controller using KMS, the + transition to the new method will happen automatically. To go back to the old + method after that will require the worker to be deleted and re-added with the + `use_deprecated_kms_auth_method` config field specified. * When grants are added to roles additional validity checking is now performed. This extra validity checking is designed to reject grants that are not [documented grant diff --git a/Makefile b/Makefile index 608cf7193..6be0efb94 100644 --- a/Makefile +++ b/Makefile @@ -36,7 +36,7 @@ golangci-lint: $(eval GOLINT_INSTALLED := $(shell which golangci-lint)) if [ "$(GOLINT_INSTALLED)" = "" ]; then \ - sh scripts/install-golangci-lint.sh -b $(GO_PATH)/bin v1.51.2; \ + sh scripts/install-golangci-lint.sh -b $(GO_PATH)/bin v1.52.2; \ fi; .PHONY: cleangen diff --git a/internal/daemon/controller/handlers/workers/worker_service.go b/internal/daemon/controller/handlers/workers/worker_service.go index a28a09587..d372a02ba 100644 --- a/internal/daemon/controller/handlers/workers/worker_service.go +++ b/internal/daemon/controller/handlers/workers/worker_service.go @@ -7,6 +7,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + stderrors "errors" "fmt" "strings" @@ -334,8 +335,25 @@ func (s Service) UpdateWorker(ctx context.Context, req *pbs.UpdateWorkerRequest) if authResults.Error != nil { return nil, authResults.Error } + w, err := s.updateInRepo(ctx, authResults.Scope.GetId(), req.GetId(), req.GetUpdateMask().GetPaths(), req.GetItem()) - if err != nil { + switch { + case err == nil: + case stderrors.Is(err, server.ErrCannotUpdateKmsWorkerViaApi): + // Treat this like a "bad field" error on name even though we couldn't + // return it in validation without having to make an additional call and + // a lot of additional logic + return nil, handlers.ValidateUpdateRequest(req, req.GetItem(), func() map[string]string { + badFields := make(map[string]string) + if req.GetItem().GetName().GetValue() != "" { + badFields[globals.NameField] = "KMS-registered workers cannot have their name updated via the API." + } + if req.GetItem().GetDescription().GetValue() != "" { + badFields[globals.DescriptionField] = "KMS-registered workers cannot have their description updated via the API." + } + return badFields + }, globals.WorkerPrefix) + default: return nil, err } diff --git a/internal/daemon/controller/handlers/workers/worker_service_test.go b/internal/daemon/controller/handlers/workers/worker_service_test.go index e2ca3a9e7..28bcd03ee 100644 --- a/internal/daemon/controller/handlers/workers/worker_service_test.go +++ b/internal/daemon/controller/handlers/workers/worker_service_test.go @@ -487,32 +487,45 @@ func TestUpdate(t *testing.T) { return connectedDownstreams } - wkr := server.TestPkiWorker(t, conn, wrapper, + pkiWkr := server.TestPkiWorker(t, conn, wrapper, server.WithName("default"), server.WithDescription("default")) - version := wkr.GetVersion() + pkiKmsWkr := server.TestPkiWorker(t, conn, wrapper, + server.WithName("default-kms"), + server.WithDescription("default-kms"), + server.WithNewIdFunc(func(ctx context.Context) (string, error) { + return server.NewWorkerIdFromScopeAndName(ctx, scope.Global.String(), "default-kms") + }), + ) + + pkiVersion := pkiWkr.GetVersion() + pkiKmsVersion := pkiKmsWkr.GetVersion() resetWorker := func() { - version++ - _, _, err = repo.UpdateWorker(context.Background(), wkr, version, []string{"Name", "Description"}) - require.NoError(t, err, "Failed to reset worker.") - version++ + pkiVersion = pkiWkr.Version + 1 + pkiKmsVersion = pkiKmsWkr.Version + 1 + wkr, _, err := repo.UpdateWorker(context.Background(), pkiWkr, pkiVersion, []string{"Name", "Description"}) + require.NoError(t, err, "Failed to reset pki worker.") + pkiVersion = wkr.Version } - wCreated := wkr.GetCreateTime().GetTimestamp().AsTime() - toMerge := &pbs.UpdateWorkerRequest{ - Id: wkr.GetPublicId(), + wCreated := pkiWkr.GetCreateTime().GetTimestamp().AsTime() + toMerge := func(wkr *server.Worker) *pbs.UpdateWorkerRequest { + return &pbs.UpdateWorkerRequest{ + Id: wkr.GetPublicId(), + } } workerService, err := NewService(ctx, repoFn, iamRepoFn, workerAuthRepoFn, nil) require.NoError(t, err) expectedScope := &scopes.ScopeInfo{Id: scope.Global.String(), Type: scope.Global.String(), Name: scope.Global.String(), Description: "Global Scope"} cases := []struct { - name string - req *pbs.UpdateWorkerRequest - res *pbs.UpdateWorkerResponse - err error + name string + reqIdFunc func(*server.Worker) string + req *pbs.UpdateWorkerRequest + res func(*server.Worker) *pbs.UpdateWorkerResponse + err error }{ { name: "Update an Existing Worker", @@ -525,20 +538,22 @@ func TestUpdate(t *testing.T) { Description: wrapperspb.String("desc"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Name: wrapperspb.String("name"), - Description: wrapperspb.String("desc"), - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + return &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + Name: wrapperspb.String("name"), + Description: wrapperspb.String("desc"), + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } }, }, { @@ -549,23 +564,25 @@ func TestUpdate(t *testing.T) { }, Item: &pb.Worker{ Name: wrapperspb.String("name"), - Description: wrapperspb.String("desc"), + Description: wrapperspb.String("default"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Name: wrapperspb.String("name"), - Description: wrapperspb.String("desc"), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + return &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + Name: wrapperspb.String("name"), + Description: wrapperspb.String("default"), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } }, }, { @@ -622,19 +639,21 @@ func TestUpdate(t *testing.T) { Description: wrapperspb.String("ignored"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Description: wrapperspb.String("default"), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + return &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + Description: wrapperspb.String(wkr.Description), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } }, }, { @@ -647,19 +666,26 @@ func TestUpdate(t *testing.T) { Name: wrapperspb.String("ignored"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Name: wrapperspb.String("default"), - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + ret := &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } + // In the previous test, the name will now be blank if it's the + // non-kms worker so must add it in to expected + if wkr.PublicId == pkiKmsWkr.PublicId { + ret.Item.Name = wrapperspb.String(wkr.Name) + } + return ret }, }, { @@ -673,20 +699,26 @@ func TestUpdate(t *testing.T) { Description: wrapperspb.String("ignored"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Name: wrapperspb.String("updated"), - Description: wrapperspb.String("default"), - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + ret := &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + Name: wrapperspb.String("updated"), + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } + // The name will not be updated if it's the pki-kms worker + if wkr.PublicId == pkiKmsWkr.PublicId { + ret.Item.Name = wrapperspb.String(wkr.Name) + } + return ret }, }, { @@ -700,20 +732,27 @@ func TestUpdate(t *testing.T) { Description: wrapperspb.String("notignored"), }, }, - res: &pbs.UpdateWorkerResponse{ - Item: &pb.Worker{ - Id: wkr.GetPublicId(), - ScopeId: wkr.GetScopeId(), - Scope: expectedScope, - Name: wrapperspb.String("default"), - ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, - Description: wrapperspb.String("notignored"), - CreatedTime: wkr.GetCreateTime().GetTimestamp(), - LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), - AuthorizedActions: testAuthorizedActions, - Type: PkiWorkerType, - DirectlyConnectedDownstreamWorkers: connectedDownstreams, - }, + res: func(wkr *server.Worker) *pbs.UpdateWorkerResponse { + ret := &pbs.UpdateWorkerResponse{ + Item: &pb.Worker{ + Id: wkr.GetPublicId(), + ScopeId: wkr.GetScopeId(), + Scope: expectedScope, + Name: wrapperspb.String("updated"), + ActiveConnectionCount: &wrapperspb.UInt32Value{Value: 0}, + Description: wrapperspb.String("notignored"), + CreatedTime: wkr.GetCreateTime().GetTimestamp(), + LastStatusTime: wkr.GetLastStatusTime().GetTimestamp(), + AuthorizedActions: testAuthorizedActions, + Type: PkiWorkerType, + DirectlyConnectedDownstreamWorkers: connectedDownstreams, + }, + } + // The name will not have been updated previously if it's the pki-kms worker + if wkr.PublicId == pkiKmsWkr.PublicId { + ret.Item.Name = wrapperspb.String(wkr.Name) + } + return ret }, }, { @@ -730,9 +769,11 @@ func TestUpdate(t *testing.T) { err: handlers.ApiErrorWithCode(codes.NotFound), }, { - name: "Cant change Id", + name: "Cant change Id pki worker", + reqIdFunc: func(worker *server.Worker) string { + return worker.GetPublicId() + }, req: &pbs.UpdateWorkerRequest{ - Id: wkr.GetPublicId(), UpdateMask: &field_mask.FieldMask{ Paths: []string{"id"}, }, @@ -853,36 +894,77 @@ func TestUpdate(t *testing.T) { }, } for _, tc := range cases { - t.Run(tc.name, func(t *testing.T) { - tc.req.Item.Version = version + for _, wkr := range []*server.Worker{pkiWkr, pkiKmsWkr} { + t.Run(fmt.Sprintf("%s-%s", tc.name, wkr.Name), func(t *testing.T) { + if wkr.PublicId == pkiWkr.PublicId { + tc.req.Item.Version = pkiVersion + } else { + tc.req.Item.Version = pkiKmsVersion + } - req := proto.Clone(toMerge).(*pbs.UpdateWorkerRequest) - proto.Merge(req, tc.req) + req := proto.Clone(toMerge(wkr)).(*pbs.UpdateWorkerRequest) + proto.Merge(req, tc.req) + if tc.reqIdFunc != nil { + req.Id = tc.reqIdFunc(wkr) + } - got, gErr := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, scope.Global.String()), req) - if tc.err != nil { - require.Error(t, gErr) - assert.True(t, errors.Is(gErr, tc.err), "UpdateWorker(%+v) got error %v, wanted %v", req, gErr, tc.err) - } + // Search through update masks to see if we're updating name + var basicInfoChange bool + if tc.req != nil && tc.req.UpdateMask != nil { + for _, field := range tc.req.UpdateMask.Paths { + if strings.ToLower(field) == "name" || strings.ToLower(field) == "description" { + basicInfoChange = true + } + if strings.Contains(field, ",") { + splitStr := strings.Split(field, ",") + for _, field := range splitStr { + if strings.ToLower(field) == "name" || strings.ToLower(field) == "description" { + basicInfoChange = true + } + } + } + } + } - if tc.err == nil { - defer resetWorker() - } + got, gErr := workerService.UpdateWorker(auth.DisabledAuthTestContext(iamRepoFn, scope.Global.String()), req) + // If it's a PKI-KMS worker and we wouldn't otherwise expect an + // error but we tried a name change, ensure we see the expected + // error here. + if wkr.PublicId == pkiKmsWkr.PublicId && basicInfoChange && tc.err == nil { + require.Error(t, gErr) + assert.Contains(t, gErr.Error(), "KMS-registered workers cannot have their") + assert.True(t, errors.Is(gErr, handlers.ApiErrorWithCode(codes.InvalidArgument))) + return + } + if tc.err != nil { + require.Error(t, gErr) + assert.True(t, errors.Is(gErr, tc.err), "UpdateWorker(%+v) got error %v, wanted %v", req, gErr, tc.err) + assert.Nil(t, got) + return + } + + // Should not have nil got at this point + require.NotNil(t, got) + + if tc.err == nil { + defer resetWorker() + } + + expRes := tc.res(wkr) - if got != nil { assert.NotNilf(t, tc.res, "Expected UpdateWorker response to be nil, but was %v", got) gotUpdateTime := got.GetItem().GetUpdatedTime().AsTime() // Verify updated set after it was created assert.True(t, gotUpdateTime.After(wCreated), "Updated resource should have been updated after its creation. Was updated %v, which is after %v", gotUpdateTime, wCreated) // Clear all values which are hard to compare against. - got.Item.UpdatedTime, tc.res.Item.UpdatedTime = nil, nil - } - if tc.res != nil { - tc.res.Item.Version = version + 1 - } - assert.Empty(t, cmp.Diff(got, tc.res, protocmp.Transform()), "UpdateWorker(%q) got response %q, wanted %q", req, got, tc.res) - }) + got.Item.UpdatedTime, expRes.Item.UpdatedTime = nil, nil + + expRes.Item.Version = tc.req.Item.Version + 1 + + assert.Empty(t, cmp.Diff(got, expRes, protocmp.Transform()), "UpdateWorker(%q) got response %q, wanted %q", req, got, expRes) + }) + } } } diff --git a/internal/daemon/worker/testing.go b/internal/daemon/worker/testing.go index dea6ad170..f8d8debc0 100644 --- a/internal/daemon/worker/testing.go +++ b/internal/daemon/worker/testing.go @@ -264,9 +264,7 @@ func NewTestWorker(t testing.TB, opts *TestWorkerOpts) *TestWorker { } } - if opts.UseDeprecatedKmsAuthMethod { - opts.Config.Worker.UseDeprecatedKmsAuthMethod = true - } + opts.Config.Worker.UseDeprecatedKmsAuthMethod = opts.UseDeprecatedKmsAuthMethod if len(opts.InitialUpstreams) > 0 { opts.Config.Worker.InitialUpstreams = opts.InitialUpstreams } diff --git a/internal/server/public_ids.go b/internal/server/public_ids.go index 8bad08cc7..ff80168e6 100644 --- a/internal/server/public_ids.go +++ b/internal/server/public_ids.go @@ -19,11 +19,11 @@ func newWorkerId(ctx context.Context) (string, error) { return id, nil } -// newWorkerIdFromScopeAndName generates a predictable public id based on the +// NewWorkerIdFromScopeAndName generates a predictable public id based on the // scope and the worker name. This should only be used on kms workers at // upsert time. -func newWorkerIdFromScopeAndName(ctx context.Context, scope, name string) (string, error) { - const op = "server.newWorkerIdFromScopeAndName" +func NewWorkerIdFromScopeAndName(ctx context.Context, scope, name string) (string, error) { + const op = "server.NewWorkerIdFromScopeAndName" id, err := db.NewPublicId(globals.WorkerPrefix, db.WithPrngValues([]string{scope, name})) if err != nil { return "", errors.Wrap(ctx, err, "server.newWorkerId") diff --git a/internal/server/repository_worker.go b/internal/server/repository_worker.go index b2a69a8cb..250cf71ad 100644 --- a/internal/server/repository_worker.go +++ b/internal/server/repository_worker.go @@ -5,6 +5,7 @@ package server import ( "context" + stderrors "errors" "fmt" "strings" @@ -22,6 +23,8 @@ import ( "google.golang.org/protobuf/types/known/timestamppb" ) +var ErrCannotUpdateKmsWorkerViaApi = stderrors.New("cannot update a kms worker's basic information via api") + // DeleteWorker will delete a worker from the repository. func (r *Repository) DeleteWorker(ctx context.Context, publicId string, _ ...Option) (int, error) { const op = "server.(Repository).DeleteWorker" @@ -282,7 +285,7 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt // are treated as the same worker. Allowing this only for kms workers // also ensures that we maintain the unique name constraint between pki // workers and kms workers. - workerId, err = newWorkerIdFromScopeAndName(ctx, worker.GetScopeId(), worker.GetName()) + workerId, err = NewWorkerIdFromScopeAndName(ctx, worker.GetScopeId(), worker.GetName()) if err != nil || workerId == "" { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id")) } @@ -299,9 +302,9 @@ func (r *Repository) UpsertWorkerStatus(ctx context.Context, worker *Worker, opt switch { case opts.withKeyId != "": // This case goes first because a key ID being supplied should - // be a clear indicator that we are working with a PKI - // workerClone, and a lack of one a clear indication we are - // working with a KMS workerClone. + // be a clear indicator that we are working with a PKI (or + // KMS-PKI) workerClone, and a lack of one a clear indication we + // are working with a KMS workerClone. // // Note: unlike in the below case, this purposefully leaves out // "description" since we want description changes for (non @@ -420,11 +423,12 @@ func setWorkerTags(ctx context.Context, w db.Writer, id string, ts TagSource, ta // UpdateWorker will update a worker in the repository and return the resulting // worker. fieldMaskPaths provides field_mask.proto paths for fields that should -// be updated. Fields will be set to NULL if the field is a zero value and +// be updated. Fields will be set to NULL if the field is a zero value and // included in fieldMask. Name, Description, and Address are the only updatable // fields, if no updatable fields are included in the fieldMaskPaths, then an -// error is returned. If any paths besides those listed above are included in -// the path then an error is returned. +// error is returned. If any paths besides those listed above are included in +// the path then an error is returned. If the worker is a KMS worker (whether +// via the old registration method or pki-kms) name updates will be disallowed. func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version uint32, fieldMaskPaths []string, opt ...Option) (*Worker, int, error) { const ( nameField = "name" @@ -470,6 +474,26 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { + // First we need to do a lookup so we can validate that this + // function is not being used for workers registered via KMS-PKI + // means + wAgg := &workerAggregate{PublicId: worker.GetPublicId()} + if err := reader.LookupById(ctx, wAgg); err != nil { + return errors.Wrap(ctx, err, op) + } + // If it's a KMS-PKI worker we do not want to allow + // name/description/other updates via the API, it should only come + // in via the upsert mechanism via status updates. If the public ID + // is predictably generated in the KMS fashion, it's a KMS-PKI + // worker. + workerId, err := NewWorkerIdFromScopeAndName(ctx, wAgg.ScopeId, wAgg.Name) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("error generating worker id in kms-pki name check case")) + } + if workerId == worker.PublicId { + return errors.Wrap(ctx, ErrCannotUpdateKmsWorkerViaApi, op, errors.WithCode(errors.InvalidParameter)) + } + worker := worker.clone() rowsUpdated, err = w.Update(ctx, worker, dbMask, nullFields, db.WithVersion(&version), db.WithWhere("server_worker.type = 'pki'")) @@ -484,7 +508,7 @@ func (r *Repository) UpdateWorker(ctx context.Context, worker *Worker, version u return errors.New(ctx, errors.MultipleRecords, op, "more than 1 resource would have been updated") } - wAgg := &workerAggregate{PublicId: worker.GetPublicId()} + wAgg = &workerAggregate{PublicId: worker.GetPublicId()} if err := reader.LookupById(ctx, wAgg); err != nil { return errors.Wrap(ctx, err, op) } diff --git a/internal/server/repository_worker_test.go b/internal/server/repository_worker_test.go index 6a0c342b2..89152b30b 100644 --- a/internal/server/repository_worker_test.go +++ b/internal/server/repository_worker_test.go @@ -297,6 +297,7 @@ func TestUpsertWorkerStatus(t *testing.T) { workers, err := repo.ListWorkers(ctx, []string{scope.Global.String()}, server.WithActiveWorkers(true)) require.NoError(t, err) assert.Len(t, workers, 1) + assert.Equal(t, server.KmsWorkerType.String(), workers[0].Type) // update again with a shutdown state wStatus3 := server.NewWorker(scope.Global.String(), @@ -335,6 +336,7 @@ func TestUpsertWorkerStatus(t *testing.T) { assert.Equal(t, uint32(1), worker.Version) assert.Equal(t, "pki_address", worker.GetAddress()) assert.Equal(t, "test-version", worker.ReleaseVersion) + assert.Equal(t, server.PkiWorkerType.String(), worker.Type) }) failureCases := []struct { @@ -1226,6 +1228,7 @@ func TestRepository_UpdateWorker(t *testing.T) { pkiCases := []struct { name string modifyWorker func(*testing.T, *server.Worker) + newIdFunc func(context.Context, string, string) func(context.Context) (string, error) path []string assertGot func(*testing.T, *server.Worker) wantErr bool @@ -1245,6 +1248,20 @@ func TestRepository_UpdateWorker(t *testing.T) { assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, + { + name: "update name against kms-pki", + modifyWorker: func(t *testing.T, w *server.Worker) { + t.Helper() + w.Name = "foo" + }, + newIdFunc: func(ctx context.Context, scopeId, name string) func(ctx context.Context) (string, error) { + return func(ctx context.Context) (string, error) { + return server.NewWorkerIdFromScopeAndName(ctx, scopeId, name) + } + }, + path: []string{"Name"}, + wantErr: true, + }, { name: "update description", modifyWorker: func(t *testing.T, w *server.Worker) { @@ -1260,10 +1277,29 @@ func TestRepository_UpdateWorker(t *testing.T) { assert.Greater(t, w.GetUpdateTime().AsTime(), w.GetCreateTime().AsTime()) }, }, + { + name: "update description against kms-pki", + modifyWorker: func(t *testing.T, w *server.Worker) { + t.Helper() + w.Description = "foo" + }, + newIdFunc: func(ctx context.Context, scopeId, name string) func(ctx context.Context) (string, error) { + return func(ctx context.Context) (string, error) { + return server.NewWorkerIdFromScopeAndName(ctx, scopeId, name) + } + }, + path: []string{"Description"}, + wantErr: true, + }, } for _, tt := range pkiCases { t.Run(tt.name, func(t *testing.T) { - wkr := server.TestPkiWorker(t, conn, wrapper) + name := strings.ReplaceAll(tt.name, " ", "-") + opts := []server.Option{server.WithName(name)} + if tt.newIdFunc != nil { + opts = append(opts, server.WithNewIdFunc(tt.newIdFunc(ctx, scope.Global.String(), name))) + } + wkr := server.TestPkiWorker(t, conn, wrapper, opts...) defer func() { _, err := repo.DeleteWorker(ctx, wkr.GetPublicId()) require.NoError(t, err) diff --git a/internal/server/repository_workerauth.go b/internal/server/repository_workerauth.go index 41a5fdddd..35a8a223e 100644 --- a/internal/server/repository_workerauth.go +++ b/internal/server/repository_workerauth.go @@ -159,7 +159,7 @@ func StoreNodeInformationTx(ctx context.Context, reader db.Reader, writer db.Wri WithReleaseVersion(fmt.Sprintf("Boundary v%s", wci.BoundaryVersion)), ) wConf.Type = PkiWorkerType.String() - wConf.PublicId, err = newWorkerIdFromScopeAndName(ctx, scopeId, wci.Name) + wConf.PublicId, err = NewWorkerIdFromScopeAndName(ctx, scopeId, wci.Name) if err != nil || wConf.PublicId == "" { return errors.Wrap(ctx, err, op, errors.WithMsg("error creating a worker id")) } @@ -167,6 +167,31 @@ func StoreNodeInformationTx(ctx context.Context, reader db.Reader, writer db.Wri Target: db.Columns{"public_id"}, Action: db.SetColumns([]string{"release_version", "name", "description"}), } + + // Before going ahead with this, we need to see if this is an upgrade + // case from the old KMS auth method. If so we need to delete first as + // type is immutable. + lookupWorker := allocWorker() + lookupWorker.PublicId = wConf.PublicId + err := reader.LookupById(ctx, &lookupWorker) + switch { + case err == nil: + if lookupWorker.Type == KmsWorkerType.String() { + // Delete it + rowsDeleted, err := writer.Delete(ctx, lookupWorker) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("error deleting kms typed worker with matching id")) + } + if rowsDeleted != 1 { + return errors.New(ctx, errors.NotSpecificIntegrity, op, fmt.Sprintf("deleted %d rows, expected to delete 1", rowsDeleted)) + } + } + case errors.Convert(err) != nil && errors.Convert(err).Code == errors.RecordNotFound: + // all good + default: + return errors.Wrap(ctx, err, op) + } + var withRowsAffected int64 err = writer.Create( ctx, diff --git a/internal/tests/cluster/multi_controller_worker_test.go b/internal/tests/cluster/multi_controller_worker_test.go index deea699bd..150fff598 100644 --- a/internal/tests/cluster/multi_controller_worker_test.go +++ b/internal/tests/cluster/multi_controller_worker_test.go @@ -118,7 +118,8 @@ func TestWorkerAppendInitialUpstreams(t *testing.T) { defer w1.Shutdown() // Wait for worker to send status - cancelCtx, _ := context.WithTimeout(ctx, 10*time.Second) + cancelCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + t.Cleanup(cancel) for { select { case <-time.After(500 * time.Millisecond): diff --git a/internal/tests/cluster/switch_kms_method_test.go b/internal/tests/cluster/switch_kms_method_test.go new file mode 100644 index 000000000..037490e25 --- /dev/null +++ b/internal/tests/cluster/switch_kms_method_test.go @@ -0,0 +1,93 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: MPL-2.0 + +package cluster + +import ( + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/cmd/config" + "github.com/hashicorp/boundary/internal/daemon/controller" + "github.com/hashicorp/boundary/internal/daemon/worker" + "github.com/hashicorp/boundary/internal/observability/event" + "github.com/hashicorp/boundary/internal/types/scope" + "github.com/hashicorp/go-hclog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWorkerUpgradeKmsAuthMethod(t *testing.T) { + ec := event.TestEventerConfig(t, "TestWorkerUpgradeKmsAuthMethod", event.TestWithObservationSink(t), event.TestWithSysSink(t)) + testLock := &sync.Mutex{} + testLogger := hclog.New(&hclog.LoggerOptions{ + Mutex: testLock, + Name: "test", + }) + require.NoError(t, event.InitSysEventer(testLogger, testLock, "use-TestWorkerUpgradeKmsAuthMethod", event.WithEventerConfig(&ec.EventerConfig))) + + conf, err := config.DevController() + conf.Eventing = &ec.EventerConfig + require.NoError(t, err) + c1 := controller.NewTestController(t, &controller.TestControllerOpts{ + Config: conf, + }) + defer c1.Shutdown() + serversRepo := c1.ServersRepo() + + conf, err = config.DevWorker() + conf.Eventing = &ec.EventerConfig + require.NoError(t, err) + w1 := worker.NewTestWorker(t, &worker.TestWorkerOpts{ + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + UseDeprecatedKmsAuthMethod: true, + }) + defer w1.Shutdown() + + // Give time for it to connect + time.Sleep(10 * time.Second) + + // Verify it's KMS type + workers, err := serversRepo.ListWorkers(c1.Context(), []string{scope.Global.String()}) + require.NoError(t, err) + require.Len(t, workers, 1) + assert.Equal(t, "kms", workers[0].Type) + oldId := workers[0].PublicId + + // Assert that the worker has connected + logBuf, err := os.ReadFile(ec.AllEvents.Name()) + require.NoError(t, err) + require.Equal(t, 1, strings.Count(string(logBuf), "worker successfully authed")) + + require.NoError(t, w1.Worker().Shutdown()) + + // Now, start up again, using the new method + w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ + Config: conf, + WorkerAuthKms: c1.Config().WorkerAuthKms, + InitialUpstreams: c1.ClusterAddrs(), + }) + defer w1.Shutdown() + + // Give time for it to connect + time.Sleep(10 * time.Second) + + // We should find only one nonce, and one successful worker authentication, + // both in the output and in the database + ec.AllEvents.Close() + logBuf, err = os.ReadFile(ec.AllEvents.Name()) + require.NoError(t, err) + require.Equal(t, 2, strings.Count(string(logBuf), "worker has successfully authenticated")) + + // Verify it's PKI type + workers, err = serversRepo.ListWorkers(c1.Context(), []string{scope.Global.String()}) + require.NoError(t, err) + require.Len(t, workers, 1) + assert.Equal(t, "pki", workers[0].Type) + assert.Equal(t, oldId, workers[0].PublicId) +} diff --git a/internal/tests/cluster/unix_listener_test.go b/internal/tests/cluster/unix_listener_test.go index babf0c244..3938b6abc 100644 --- a/internal/tests/cluster/unix_listener_test.go +++ b/internal/tests/cluster/unix_listener_test.go @@ -5,7 +5,6 @@ package cluster import ( "context" - "io/ioutil" "os" "path" "testing" @@ -29,7 +28,7 @@ func TestUnixListener(t *testing.T) { conf, err := config.DevController() require.NoError(err) - tempDir, err := ioutil.TempDir("", "boundary-unix-listener-test") + tempDir, err := os.MkdirTemp("", "boundary-unix-listener-test") require.NoError(err) defer func() { diff --git a/internal/tests/cluster/worker_replay_test.go b/internal/tests/cluster/worker_replay_test.go index 7095bfbed..a27d467f4 100644 --- a/internal/tests/cluster/worker_replay_test.go +++ b/internal/tests/cluster/worker_replay_test.go @@ -66,7 +66,7 @@ func TestWorkerReplay(t *testing.T) { require.NoError(t, w1.Worker().Shutdown()) // Now, start up again with the same nonce - w1 = worker.NewTestWorker(t, &worker.TestWorkerOpts{ + worker.NewTestWorker(t, &worker.TestWorkerOpts{ Config: conf, WorkerAuthKms: c1.Config().WorkerAuthKms, InitialUpstreams: c1.ClusterAddrs(),