diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f289220a..b9f3b42ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,8 @@ Canonical reference for changes, improvements, and bugfixes for Boundary. ([PR](https://github.com/hashicorp/boundary/pull/2544)) * workers: Fixed a panic that can happen in certain situations ([PR](https://github.com/hashicorp/boundary/pull/2553)) +* sessions: Fixed a panic in a controller when a worker is deleted while + sessions are ongoing ([PR](https://github.com/hashicorp/boundary/pull/2612)) ### Deprecations/Changes diff --git a/internal/auth/oidc/service_callback.go b/internal/auth/oidc/service_callback.go index c75d63bbd..ec90a0da5 100644 --- a/internal/auth/oidc/service_callback.go +++ b/internal/auth/oidc/service_callback.go @@ -230,9 +230,9 @@ func Callback( return "", errors.Wrap(ctx, err, op) } - scope, err := iamRepo.LookupScope(ctx, am.ScopeId) + _, err = iamRepo.LookupScope(ctx, am.ScopeId) if err != nil { - return "", errors.Wrap(ctx, err, op, errors.WithMsg("unable to lookup account scope: "+scope.PublicId)) + return "", errors.Wrap(ctx, err, op, errors.WithMsg("unable to lookup account scope: "+am.ScopeId)) } user, err := iamRepo.LookupUserWithLogin(ctx, acct.PublicId) diff --git a/internal/daemon/cluster/handlers/worker_service.go b/internal/daemon/cluster/handlers/worker_service.go index 5ff6d9f15..66197b89a 100644 --- a/internal/daemon/cluster/handlers/worker_service.go +++ b/internal/daemon/cluster/handlers/worker_service.go @@ -295,18 +295,21 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku if sessionInfo.WorkerFilter != "" { if req.WorkerId == "" { event.WriteError(ctx, op, errors.New("worker filter enabled for session but got no id information from worker")) - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Did not receive worker id when looking up session but filtering is enabled") + return nil, status.Errorf(codes.Internal, "Did not receive worker id when looking up session but filtering is enabled") } serversRepo, err := ws.serversRepoFn() if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error getting server repo")) - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err) + return nil, status.Errorf(codes.Internal, "Error acquiring server repo when looking up session: %v", err) } w, err := serversRepo.LookupWorker(ctx, req.WorkerId) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up worker", "worker_id", req.WorkerId)) - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error looking up worker: %v", err) - + return nil, status.Errorf(codes.Internal, "Error looking up worker: %v", err) + } + if w == nil { + event.WriteError(ctx, op, err, event.WithInfoMsg("error looking up worker", "worker_id", req.WorkerId)) + return nil, status.Errorf(codes.Internal, "Worker not found") } // Build the map for filtering. tagMap := w.CanonicalTags() @@ -315,7 +318,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku eval, err := bexpr.CreateEvaluator(sessionInfo.WorkerFilter) if err != nil { event.WriteError(ctx, op, err, event.WithInfoMsg("error creating worker filter evaluator", "worker_id", req.WorkerId)) - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, "Error creating worker filter evaluator: %v", err) + return nil, status.Errorf(codes.Internal, "Error creating worker filter evaluator: %v", err) } filterInput := map[string]any{ "name": w.GetName(), @@ -323,7 +326,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku } ok, err := eval.Evaluate(filterInput) if err != nil { - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, + return nil, status.Errorf(codes.Internal, fmt.Sprintf("Worker filter expression evaluation resulted in error: %s", err)) } if !ok { @@ -335,7 +338,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku creds, err := sessRepo.ListSessionCredentials(ctx, sessionInfo.ProjectId, sessionInfo.PublicId) if err != nil { - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, + return nil, status.Errorf(codes.Internal, fmt.Sprintf("Error retrieving session credentials: %s", err)) } var workerCreds []*pbs.Credential @@ -343,7 +346,7 @@ func (ws *workerServiceServer) LookupSession(ctx context.Context, req *pbs.Looku m := &pbs.Credential{} err = proto.Unmarshal(c, m) if err != nil { - return &pbs.LookupSessionResponse{}, status.Errorf(codes.Internal, + return nil, status.Errorf(codes.Internal, fmt.Sprintf("Error unmarshaling credentials: %s", err)) } workerCreds = append(workerCreds, m) @@ -398,6 +401,9 @@ func (ws *workerServiceServer) CancelSession(ctx context.Context, req *pbs.Cance if err != nil { return nil, err } + if ses == nil { + return nil, status.Error(codes.PermissionDenied, "Unknown session ID.") + } ses, err = sessRepo.CancelSession(ctx, req.GetSessionId(), ses.Version) if err != nil { diff --git a/internal/daemon/cluster/handlers/worker_service_test.go b/internal/daemon/cluster/handlers/worker_service_test.go index 61a3eac2d..83cce0452 100644 --- a/internal/daemon/cluster/handlers/worker_service_test.go +++ b/internal/daemon/cluster/handlers/worker_service_test.go @@ -27,6 +27,7 @@ import ( "github.com/stretchr/testify/require" "golang.org/x/crypto/ssh/testdata" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/testing/protocmp" ) func TestLookupSession(t *testing.T) { @@ -68,6 +69,18 @@ func TestLookupSession(t *testing.T) { Endpoint: "tcp://127.0.0.1:22", }) + worker1 := server.TestKmsWorker(t, conn, wrapper, server.WithName("testworker")) + sessWithWorkerFilter := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ProjectId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + WorkerFilter: fmt.Sprintf("%q matches %q", "/name", worker1.GetName()), + }) + sessWithCreds := session.TestSession(t, conn, wrapper, session.ComposedOf{ UserId: uId, HostId: h.GetPublicId(), @@ -134,17 +147,38 @@ func TestLookupSession(t *testing.T) { wantErr bool wantErrMsg string want *pbs.LookupSessionResponse - sessionId string + req *pbs.LookupSessionRequest }{ { - name: "Invalid session id", - sessionId: "s_fakesession", + name: "Invalid session id", + req: &pbs.LookupSessionRequest{ + SessionId: "s_fakesession", + }, wantErr: true, wantErrMsg: "rpc error: code = PermissionDenied desc = Unknown session ID.", }, { - name: "Valid", - sessionId: sess.PublicId, + name: "no worker id", + req: &pbs.LookupSessionRequest{ + SessionId: sessWithWorkerFilter.PublicId, + }, + wantErr: true, + wantErrMsg: "rpc error: code = Internal desc = Did not receive worker id when looking up session but filtering is enabled", + }, + { + name: "nonexistant worker id", + req: &pbs.LookupSessionRequest{ + SessionId: sessWithWorkerFilter.PublicId, + WorkerId: "w_nonexistingworker", + }, + wantErr: true, + wantErrMsg: "rpc error: code = Internal desc = Worker not found", + }, + { + name: "Valid", + req: &pbs.LookupSessionRequest{ + SessionId: sess.PublicId, + }, want: &pbs.LookupSessionResponse{ ConnectionLimit: 1, ConnectionsLeft: 1, @@ -158,8 +192,28 @@ func TestLookupSession(t *testing.T) { }, }, { - name: "Valid-with-worker-creds", - sessionId: sessWithCreds.PublicId, + name: "Valid with worker filter", + req: &pbs.LookupSessionRequest{ + SessionId: sess.PublicId, + WorkerId: worker1.GetPublicId(), + }, + want: &pbs.LookupSessionResponse{ + ConnectionLimit: 1, + ConnectionsLeft: 1, + Version: 1, + Endpoint: sess.Endpoint, + HostId: sess.HostId, + HostSetId: sess.HostSetId, + TargetId: sess.TargetId, + UserId: sess.UserId, + Status: pbs.SESSIONSTATUS_SESSIONSTATUS_PENDING, + }, + }, + { + name: "Valid-with-worker-creds", + req: &pbs.LookupSessionRequest{ + SessionId: sessWithCreds.PublicId, + }, want: &pbs.LookupSessionResponse{ ConnectionLimit: 1, ConnectionsLeft: 1, @@ -178,11 +232,8 @@ func TestLookupSession(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - req := &pbs.LookupSessionRequest{ - SessionId: tc.sessionId, - } - got, err := s.LookupSession(ctx, req) + got, err := s.LookupSession(ctx, tc.req) if tc.wantErr { require.Error(err) assert.Nil(got) @@ -202,6 +253,97 @@ func TestLookupSession(t *testing.T) { } } +func TestCancelSession(t *testing.T) { + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + org, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper)) + + serversRepoFn := func() (*server.Repository, error) { + return server.NewRepository(rw, rw, kms) + } + workerAuthRepoFn := func() (*server.WorkerAuthRepositoryStorage, error) { + return server.NewRepositoryStorage(ctx, rw, rw, kms) + } + sessionRepoFn := func(opts ...session.Option) (*session.Repository, error) { + return session.NewRepository(ctx, rw, rw, kms, opts...) + } + connectionRepoFn := func() (*session.ConnectionRepository, error) { + return session.NewConnectionRepository(ctx, rw, rw, kms) + } + + at := authtoken.TestAuthToken(t, conn, kms, org.GetPublicId()) + uId := at.GetIamUserId() + hc := static.TestCatalogs(t, conn, prj.GetPublicId(), 1)[0] + hs := static.TestSets(t, conn, hc.GetPublicId(), 1)[0] + h := static.TestHosts(t, conn, hc.GetPublicId(), 1)[0] + static.TestSetMembers(t, conn, hs.GetPublicId(), []*static.Host{h}) + tar := tcp.TestTarget(ctx, t, conn, prj.GetPublicId(), "test", target.WithHostSources([]string{hs.GetPublicId()})) + + sess := session.TestSession(t, conn, wrapper, session.ComposedOf{ + UserId: uId, + HostId: h.GetPublicId(), + TargetId: tar.GetPublicId(), + HostSetId: hs.GetPublicId(), + AuthTokenId: at.GetPublicId(), + ProjectId: prj.GetPublicId(), + Endpoint: "tcp://127.0.0.1:22", + }) + s := handlers.NewWorkerServiceServer(serversRepoFn, workerAuthRepoFn, sessionRepoFn, connectionRepoFn, new(sync.Map), kms, new(atomic.Int64)) + require.NotNil(t, s) + _ = sess + cases := []struct { + name string + wantErr bool + wantErrMsg string + want *pbs.CancelSessionResponse + req *pbs.CancelSessionRequest + }{ + { + name: "Invalid session id", + req: &pbs.CancelSessionRequest{ + SessionId: "s_fakesession", + }, + wantErr: true, + wantErrMsg: "rpc error: code = PermissionDenied desc = Unknown session ID.", + }, + { + name: "Valid", + req: &pbs.CancelSessionRequest{ + SessionId: sess.PublicId, + }, + want: &pbs.CancelSessionResponse{ + Status: pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING, + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + + got, err := s.CancelSession(ctx, tc.req) + if tc.wantErr { + require.Error(err) + assert.Nil(got) + assert.Equal(tc.wantErrMsg, err.Error()) + return + } + require.NoError(err) + require.NotNil(got) + assert.Empty( + cmp.Diff( + tc.want, + got, + protocmp.Transform(), + ), + ) + }) + } +} + // This test creates workers of both kms and pki type and verifies that the only // returned workers are those of kms type with the expected tag key/value func TestHcpbWorkers(t *testing.T) {