diff --git a/internal/db/migrations/postgres.gen.go b/internal/db/migrations/postgres.gen.go index 29589eefd..147ccc4e4 100644 --- a/internal/db/migrations/postgres.gen.go +++ b/internal/db/migrations/postgres.gen.go @@ -3646,6 +3646,36 @@ begin; create trigger insert_session_state before insert on session_state for each row execute procedure insert_session_state(); + create view session_with_state as + select + s.public_id, + s.user_id, + s.host_id, + s.server_id, + s.server_type, + s.target_id, + s.host_set_id, + s.auth_token_id, + s.scope_id, + s.certificate, + s.expiration_time, + s.connection_limit, + s.tofu_token, + s.key_id, + s.termination_reason, + s.version, + s.create_time, + s.update_time, + s.endpoint, + ss.state, + ss.previous_end_time, + ss.start_time, + ss.end_time + from + session s, + session_state ss + where + s.public_id = ss.session_id; commit; diff --git a/internal/db/migrations/postgres/50_session.up.sql b/internal/db/migrations/postgres/50_session.up.sql index d56101992..94c3536a9 100644 --- a/internal/db/migrations/postgres/50_session.up.sql +++ b/internal/db/migrations/postgres/50_session.up.sql @@ -374,5 +374,35 @@ begin; create trigger insert_session_state before insert on session_state for each row execute procedure insert_session_state(); + create view session_with_state as + select + s.public_id, + s.user_id, + s.host_id, + s.server_id, + s.server_type, + s.target_id, + s.host_set_id, + s.auth_token_id, + s.scope_id, + s.certificate, + s.expiration_time, + s.connection_limit, + s.tofu_token, + s.key_id, + s.termination_reason, + s.version, + s.create_time, + s.update_time, + s.endpoint, + ss.state, + ss.previous_end_time, + ss.start_time, + ss.end_time + from + session s, + session_state ss + where + s.public_id = ss.session_id; commit; diff --git a/internal/session/options.go b/internal/session/options.go index 48282f792..17aecc4ec 100644 --- a/internal/session/options.go +++ b/internal/session/options.go @@ -24,6 +24,7 @@ type options struct { withUserId string withExpirationTime *timestamp.Timestamp withTestTofu []byte + withListingConvert bool } func getDefaultOptions() options { @@ -73,3 +74,9 @@ func WithTestTofu(tofu []byte) Option { o.withTestTofu = tofu } } + +func withListingConvert(withListingConvert bool) Option { + return func(o *options) { + o.withListingConvert = withListingConvert + } +} diff --git a/internal/session/query.go b/internal/session/query.go index a6313e24d..2e5817d63 100644 --- a/internal/session/query.go +++ b/internal/session/query.go @@ -108,5 +108,15 @@ session_connection_limit(expiration_time, connection_limit) as ( select expiration_time, connection_limit, current_connection_count from session_connection_limit, session_connection_count; +` + sessionList = ` +select * +from + (select public_id from session %s) s, + session_with_state ss +where + s.public_id = ss.public_id + %s +%s ` ) diff --git a/internal/session/repository.go b/internal/session/repository.go index f83e20e80..9d9030ebb 100644 --- a/internal/session/repository.go +++ b/internal/session/repository.go @@ -3,6 +3,8 @@ package session import ( "context" "errors" + "fmt" + "sort" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/kms" @@ -50,8 +52,7 @@ func NewRepository(r db.Reader, w db.Writer, kms *kms.Kms, opt ...Option) (*Repo // list will return a listing of resources and honor the WithLimit option or the // repo defaultLimit. Supports WithOrder option. -func (r *Repository) list(ctx context.Context, resources interface{}, where string, args []interface{}, opt ...Option) error { - opts := getOpts(opt...) +func (r *Repository) list(ctx context.Context, resources interface{}, where string, args []interface{}, opts options) error { limit := r.defaultLimit var dbOpts []db.Option if opts.withLimit != 0 { @@ -64,3 +65,77 @@ func (r *Repository) list(ctx context.Context, resources interface{}, where stri } return r.reader.SearchWhere(ctx, resources, where, args, dbOpts...) } + +func (r *Repository) convertToSessions(ctx context.Context, sessionsWithState []*sessionView, opt ...Option) ([]*Session, error) { + opts := getOpts(opt...) + + if len(sessionsWithState) == 0 { + return nil, nil + } + sessions := []*Session{} + var prevSessionId string + var workingSession *Session + for _, sv := range sessionsWithState { + if sv.PublicId != prevSessionId { + if prevSessionId != "" { + sort.Slice(workingSession.States, func(i, j int) bool { + return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime()) + }) + sessions = append(sessions, workingSession) + } + prevSessionId = sv.PublicId + workingSession = &Session{ + PublicId: sv.PublicId, + UserId: sv.UserId, + HostId: sv.HostId, + ServerId: sv.ServerId, + ServerType: sv.ServerType, + TargetId: sv.TargetId, + HostSetId: sv.HostSetId, + AuthTokenId: sv.AuthTokenId, + ScopeId: sv.ScopeId, + Certificate: sv.Certificate, + ExpirationTime: sv.ExpirationTime, + CtTofuToken: sv.CtTofuToken, + TofuToken: sv.TofuToken, // will always be nil since it's not stored in the database. + TerminationReason: sv.TerminationReason, + CreateTime: sv.CreateTime, + UpdateTime: sv.UpdateTime, + Version: sv.Version, + Endpoint: sv.Endpoint, + ConnectionLimit: sv.ConnectionLimit, + KeyId: sv.KeyId} + if opts.withListingConvert { + workingSession.CtTofuToken = nil // CtTofuToken should not returned in lists + workingSession.TofuToken = nil // TofuToken should not returned in lists + workingSession.KeyId = "" // KeyId should not be returned in lists + } else { + if len(workingSession.CtTofuToken) > 0 { + databaseWrapper, err := r.kms.GetWrapper(ctx, workingSession.ScopeId, kms.KeyPurposeDatabase, kms.WithKeyId(workingSession.KeyId)) + if err != nil { + return nil, fmt.Errorf("convert session: unable to get database wrapper: %w", err) + } + if err := workingSession.decrypt(ctx, databaseWrapper); err != nil { + return nil, fmt.Errorf("convert session: cannot decrypt session value: %w", err) + } + } else { + workingSession.CtTofuToken = nil + } + } + } + + state := &State{ + SessionId: sv.PublicId, + Status: Status(sv.Status), + PreviousEndTime: sv.PreviousEndTime, + StartTime: sv.StartTime, + EndTime: sv.EndTime, + } + workingSession.States = append(workingSession.States, state) + } + sort.Slice(workingSession.States, func(i, j int) bool { + return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime()) + }) + sessions = append(sessions, workingSession) + return sessions, nil +} diff --git a/internal/session/repository_connection.go b/internal/session/repository_connection.go index 7ab528b23..9ae829af7 100644 --- a/internal/session/repository_connection.go +++ b/internal/session/repository_connection.go @@ -44,8 +44,9 @@ func (r *Repository) LookupConnection(ctx context.Context, connectionId string, // ListConnections will sessions. Supports the WithLimit and WithOrder options. func (r *Repository) ListConnections(ctx context.Context, sessionId string, opt ...Option) ([]*Connection, error) { + opts := getOpts(opt...) var connections []*Connection - err := r.list(ctx, &connections, "session_id = ?", []interface{}{sessionId}, opt...) // pass options, so WithLimit and WithOrder are supported + err := r.list(ctx, &connections, "session_id = ?", []interface{}{sessionId}, opts) // pass options, so WithLimit and WithOrder are supported if err != nil { return nil, fmt.Errorf("list connections: %w", err) } diff --git a/internal/session/repository_session.go b/internal/session/repository_session.go index a1d9bb6ad..279852924 100644 --- a/internal/session/repository_session.go +++ b/internal/session/repository_session.go @@ -20,64 +20,63 @@ import ( // its State of "Pending". The following fields must be empty when creating a // session: ServerId, ServerType, and PublicId. No options are // currently supported. -func (r *Repository) CreateSession(ctx context.Context, sessionWrapper wrapping.Wrapper, newSession *Session, opt ...Option) (*Session, *State, ed25519.PrivateKey, error) { +func (r *Repository) CreateSession(ctx context.Context, sessionWrapper wrapping.Wrapper, newSession *Session, opt ...Option) (*Session, ed25519.PrivateKey, error) { if newSession == nil { - return nil, nil, nil, fmt.Errorf("create session: missing session: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: missing session: %w", db.ErrInvalidParameter) } if newSession.PublicId != "" { - return nil, nil, nil, fmt.Errorf("create session: public id is not empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: public id is not empty: %w", db.ErrInvalidParameter) } if len(newSession.Certificate) != 0 { - return nil, nil, nil, fmt.Errorf("create session: certificate is not empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: certificate is not empty: %w", db.ErrInvalidParameter) } if newSession.TargetId == "" { - return nil, nil, nil, fmt.Errorf("create session: target id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: target id is empty: %w", db.ErrInvalidParameter) } if newSession.HostId == "" { - return nil, nil, nil, fmt.Errorf("create session: user id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: user id is empty: %w", db.ErrInvalidParameter) } if newSession.UserId == "" { - return nil, nil, nil, fmt.Errorf("create session: user id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: user id is empty: %w", db.ErrInvalidParameter) } if newSession.HostSetId == "" { - return nil, nil, nil, fmt.Errorf("create session: host set id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: host set id is empty: %w", db.ErrInvalidParameter) } if newSession.AuthTokenId == "" { - return nil, nil, nil, fmt.Errorf("create session: auth token id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: auth token id is empty: %w", db.ErrInvalidParameter) } if newSession.ScopeId == "" { - return nil, nil, nil, fmt.Errorf("create session: scope id is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: scope id is empty: %w", db.ErrInvalidParameter) } if newSession.ServerId != "" { - return nil, nil, nil, fmt.Errorf("create session: server id must be empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: server id must be empty: %w", db.ErrInvalidParameter) } if newSession.ServerType != "" { - return nil, nil, nil, fmt.Errorf("create session: server type must be empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: server type must be empty: %w", db.ErrInvalidParameter) } if newSession.CtTofuToken != nil { - return nil, nil, nil, fmt.Errorf("create session: ct must be empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: ct must be empty: %w", db.ErrInvalidParameter) } if newSession.TofuToken != nil { - return nil, nil, nil, fmt.Errorf("create session: tofu token must be empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: tofu token must be empty: %w", db.ErrInvalidParameter) } if newSession.ExpirationTime == nil || newSession.ExpirationTime.Timestamp.AsTime().IsZero() { - return nil, nil, nil, fmt.Errorf("create session: expiration is empty: %w", db.ErrInvalidParameter) + return nil, nil, fmt.Errorf("create session: expiration is empty: %w", db.ErrInvalidParameter) } id, err := newId() if err != nil { - return nil, nil, nil, fmt.Errorf("create session: %w", err) + return nil, nil, fmt.Errorf("create session: %w", err) } privKey, certBytes, err := newCert(sessionWrapper, newSession.UserId, id, newSession.ExpirationTime.Timestamp.AsTime()) if err != nil { - return nil, nil, nil, fmt.Errorf("create session: %w", err) + return nil, nil, fmt.Errorf("create session: %w", err) } newSession.Certificate = certBytes newSession.PublicId = id var returnedSession *Session - var returnedState *State _, err = r.writer.DoTx( ctx, db.StdRetryCnt, @@ -95,30 +94,32 @@ func (r *Repository) CreateSession(ctx context.Context, sessionWrapper wrapping. if len(foundStates) != 1 { return fmt.Errorf("%d states found for new session %s", len(foundStates), returnedSession.PublicId) } - returnedState = foundStates[0] - if returnedState.Status != StatusPending { - return fmt.Errorf("new session %s state is not valid: %s", returnedSession.PublicId, returnedState.Status) + if len(foundStates) == 0 { + return fmt.Errorf("no states found for new session %s", returnedSession.PublicId) + } + returnedSession.States = foundStates + if returnedSession.States[0].Status != StatusPending { + return fmt.Errorf("new session %s state is not valid: %s", returnedSession.PublicId, returnedSession.States[0].Status) } return nil }, ) if err != nil { - return nil, nil, nil, fmt.Errorf("create session: %w", err) + return nil, nil, fmt.Errorf("create session: %w", err) } - return returnedSession, returnedState, privKey, err + return returnedSession, privKey, err } // LookupSession will look up a session in the repository and return the session // with its states. Returned States are ordered by start time descending. If the // session is not found, it will return nil, nil, nil. No options are currently // supported. -func (r *Repository) LookupSession(ctx context.Context, sessionId string, opt ...Option) (*Session, []*State, error) { +func (r *Repository) LookupSession(ctx context.Context, sessionId string, opt ...Option) (*Session, error) { if sessionId == "" { - return nil, nil, fmt.Errorf("lookup session: missing sessionId id: %w", db.ErrInvalidParameter) + return nil, fmt.Errorf("lookup session: missing sessionId id: %w", db.ErrInvalidParameter) } session := AllocSession() session.PublicId = sessionId - var states []*State _, err := r.writer.DoTx( ctx, db.StdRetryCnt, @@ -127,55 +128,94 @@ func (r *Repository) LookupSession(ctx context.Context, sessionId string, opt .. if err := read.LookupById(ctx, &session); err != nil { return fmt.Errorf("lookup session: failed %w for %s", err, sessionId) } - var err error - if states, err = fetchStates(ctx, read, sessionId, db.WithOrder("start_time desc")); err != nil { + states, err := fetchStates(ctx, read, sessionId, db.WithOrder("start_time desc")) + if err != nil { return err } + session.States = states return nil }, ) if err != nil { if errors.Is(err, db.ErrRecordNotFound) { - return nil, nil, nil + return nil, nil } - return nil, nil, fmt.Errorf("lookup session: %w", err) + return nil, fmt.Errorf("lookup session: %w", err) } if len(session.CtTofuToken) > 0 { databaseWrapper, err := r.kms.GetWrapper(ctx, session.ScopeId, kms.KeyPurposeDatabase, kms.WithKeyId(session.KeyId)) if err != nil { - return nil, nil, fmt.Errorf("lookup session: unable to get database wrapper: %w", err) + return nil, fmt.Errorf("lookup session: unable to get database wrapper: %w", err) } if err := session.decrypt(ctx, databaseWrapper); err != nil { - return nil, nil, fmt.Errorf("lookup session: cannot decrypt session value: %w", err) + return nil, fmt.Errorf("lookup session: cannot decrypt session value: %w", err) } } else { session.CtTofuToken = nil } - return &session, states, nil + + return &session, nil } -// ListSessions will sessions. Supports the WithLimit, WithScopeId and WithOrder options. +// ListSessions will sessions. Supports the WithLimit, and WithScopeId options. func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Session, error) { opts := getOpts(opt...) var where []string var args []interface{} + + inClauseCnt := 0 switch { case opts.withScopeId != "": - where, args = append(where, "scope_id = ?"), append(args, opts.withScopeId) + inClauseCnt += 1 + where, args = append(where, fmt.Sprintf("scope_id = $%d", inClauseCnt)), append(args, opts.withScopeId) case opts.withUserId != "": - where, args = append(where, "user_id = ?"), append(args, opts.withUserId) + inClauseCnt += 1 + where, args = append(where, fmt.Sprintf("user_id = $%d", inClauseCnt)), append(args, opts.withUserId) + } + var limit string + switch { + case opts.withLimit < 0: // any negative number signals unlimited results + case opts.withLimit == 0: // zero signals the default value and default limits + limit = fmt.Sprintf("limit %d", r.defaultLimit) + default: + // non-zero signals an override of the default limit for the repo. + limit = fmt.Sprintf("limit %d", opts.withLimit) } - var sessions []*Session - err := r.list(ctx, &sessions, strings.Join(where, " and"), args, opt...) + if opts.withOrder != "" { + opts.withOrder = fmt.Sprintf("order by %s", opts.withOrder) + } + + var whereClause string + if len(where) > 0 { + whereClause = " and " + strings.Join(where, " and") + } + q := sessionList + query := fmt.Sprintf(q, limit, whereClause, opts.withOrder) + + tx, err := r.reader.DB() + if err != nil { + return nil, fmt.Errorf("list sessions: unable to get DB: %w", err) + } + + rows, err := tx.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("changes: query failed: %w", err) + } + defer rows.Close() + + var sessionsWithState []*sessionView + for rows.Next() { + var s sessionView + if err := r.reader.ScanRows(rows, &s); err != nil { + return nil, fmt.Errorf("changes: scan row failed: %w", err) + } + sessionsWithState = append(sessionsWithState, &s) + } + sessions, err := r.convertToSessions(ctx, sessionsWithState, withListingConvert(true)) if err != nil { return nil, fmt.Errorf("list sessions: %w", err) } - for _, s := range sessions { - s.CtTofuToken = nil - s.TofuToken = nil - s.KeyId = "" - } return sessions, nil } @@ -219,18 +259,19 @@ func (r *Repository) DeleteSession(ctx context.Context, publicId string, opt ... // called when the user cancels a session and the controller wants to update the // session state to "cancelling" for the given reason, so the workers can get // the "cancelling signal" during their next status heartbeat. -func (r *Repository) CancelSession(ctx context.Context, sessionId string, sessionVersion uint32) (*Session, []*State, error) { +func (r *Repository) CancelSession(ctx context.Context, sessionId string, sessionVersion uint32) (*Session, error) { if sessionId == "" { - return nil, nil, fmt.Errorf("cancel session: missing session id: %w", db.ErrInvalidParameter) + return nil, fmt.Errorf("cancel session: missing session id: %w", db.ErrInvalidParameter) } if sessionVersion == 0 { - return nil, nil, fmt.Errorf("cancel session: missing session version: %w", db.ErrInvalidParameter) + return nil, fmt.Errorf("cancel session: missing session version: %w", db.ErrInvalidParameter) } s, ss, err := r.updateState(ctx, sessionId, sessionVersion, StatusCancelling) if err != nil { - return nil, nil, fmt.Errorf("cancel session: %w", err) + return nil, fmt.Errorf("cancel session: %w", err) } - return s, ss, nil + s.States = ss + return s, nil } // TerminateSession sets a session's state to "terminated" in the repo. It's @@ -238,18 +279,17 @@ func (r *Repository) CancelSession(ctx context.Context, sessionId string, sessio // when all of a session's workers have stopped sending heartbeat status for a // period of time. Sessions cannot be terminated which still have connections // that are not closed. -func (r *Repository) TerminateSession(ctx context.Context, sessionId string, sessionVersion uint32, reason TerminationReason) (*Session, []*State, error) { +func (r *Repository) TerminateSession(ctx context.Context, sessionId string, sessionVersion uint32, reason TerminationReason) (*Session, error) { if sessionId == "" { - return nil, nil, fmt.Errorf("terminate session: missing session id: %w", db.ErrInvalidParameter) + return nil, fmt.Errorf("terminate session: missing session id: %w", db.ErrInvalidParameter) } if sessionVersion == 0 { - return nil, nil, fmt.Errorf("terminate session: version cannot be zero: %w", db.ErrInvalidParameter) + return nil, fmt.Errorf("terminate session: version cannot be zero: %w", db.ErrInvalidParameter) } updatedSession := AllocSession() updatedSession.PublicId = sessionId updatedSession.TerminationReason = reason.String() - var returnedStates []*State _, err := r.writer.DoTx( ctx, db.StdRetryCnt, @@ -269,17 +309,18 @@ func (r *Repository) TerminateSession(ctx context.Context, sessionId string, ses if rowsUpdated != 1 { return fmt.Errorf("update to session %s would have updated %d session", updatedSession.PublicId, rowsUpdated) } - returnedStates, err = fetchStates(ctx, reader, sessionId, db.WithOrder("start_time desc")) + states, err := fetchStates(ctx, reader, sessionId, db.WithOrder("start_time desc")) if err != nil { return err } + updatedSession.States = states return nil }, ) if err != nil { - return nil, nil, fmt.Errorf("terminate session: %w", err) + return nil, fmt.Errorf("terminate session: %w", err) } - return &updatedSession, returnedStates, nil + return &updatedSession, nil } // AuthorizeConnection will check to see if a connection is allowed. Currently, diff --git a/internal/session/repository_session_test.go b/internal/session/repository_session_test.go index d1a320c07..030a5cec5 100644 --- a/internal/session/repository_session_test.go +++ b/internal/session/repository_session_test.go @@ -89,6 +89,7 @@ func TestRepository_ListSession(t *testing.T) { testSessions := []*Session{} for i := 0; i < tt.createCnt; i++ { s := TestSession(t, conn, wrapper, composedOf) + _ = TestState(t, conn, s.PublicId, StatusActive) testSessions = append(testSessions, s) } assert.Equal(tt.createCnt, len(testSessions)) @@ -99,6 +100,10 @@ func TestRepository_ListSession(t *testing.T) { } require.NoError(err) assert.Equal(tt.wantCnt, len(got)) + if tt.wantCnt > 0 { + assert.Equal(StatusActive, got[0].States[0].Status) + assert.Equal(StatusPending, got[0].States[1].Status) + } }) } t.Run("withOrder", func(t *testing.T) { @@ -248,11 +253,10 @@ func TestRepository_CreateSession(t *testing.T) { ExpirationTime: tt.args.composedOf.ExpirationTime, ConnectionLimit: tt.args.composedOf.ConnectionLimit, } - ses, st, privKey, err := repo.CreateSession(context.Background(), wrapper, s) + ses, privKey, err := repo.CreateSession(context.Background(), wrapper, s) if tt.wantErr { assert.Error(err) assert.Nil(ses) - assert.Nil(st) if tt.wantIsError != nil { assert.True(errors.Is(err, tt.wantIsError)) } @@ -261,11 +265,11 @@ func TestRepository_CreateSession(t *testing.T) { require.NoError(err) assert.NotNil(ses) assert.NotNil(privKey) - assert.NotNil(st) + assert.NotNil(ses.States) assert.NotNil(ses.CreateTime) - assert.NotNil(st.StartTime) - assert.Equal(st.Status, StatusPending.String()) - foundSession, foundStates, err := repo.LookupSession(context.Background(), ses.PublicId) + assert.NotNil(ses.States[0].StartTime) + assert.Equal(ses.States[0].Status, StatusPending) + foundSession, err := repo.LookupSession(context.Background(), ses.PublicId) assert.NoError(err) // Account for slight offsets in nanos @@ -277,8 +281,8 @@ func TestRepository_CreateSession(t *testing.T) { err = db.TestVerifyOplog(t, rw, ses.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_CREATE), db.WithCreateNotBefore(10*time.Second)) assert.Error(err) - require.Equal(1, len(foundStates)) - assert.Equal(foundStates[0].Status, StatusPending.String()) + require.Equal(1, len(foundSession.States)) + assert.Equal(foundSession.States[0].Status, StatusPending) }) } } @@ -394,7 +398,7 @@ func TestRepository_UpdateState(t *testing.T) { require.NotNil(s) require.NotNil(ss) assert.Equal(tt.wantStateCnt, len(ss)) - assert.Equal(tt.newStatus.String(), ss[0].Status) + assert.Equal(tt.newStatus, ss[0].Status) }) } } @@ -445,8 +449,7 @@ func TestRepository_AuthorizeConnect(t *testing.T) { s := AllocSession() return &s }(), - wantErr: true, - wantIsError: db.ErrInvalidParameter, + wantErr: true, }, { name: "exceeded-connection-limit", @@ -455,14 +458,12 @@ func TestRepository_AuthorizeConnect(t *testing.T) { _ = TestConnection(t, conn, session.PublicId, "127.0.0.1", 22, "127.0.0.1", 2222) return session }(), - wantErr: true, - wantIsError: ErrInvalidStateForOperation, + wantErr: true, }, { - name: "expired-session", - session: setupFn(×tamp.Timestamp{Timestamp: ptypes.TimestampNow()}), - wantErr: true, - wantIsError: ErrInvalidStateForOperation, + name: "expired-session", + session: setupFn(×tamp.Timestamp{Timestamp: ptypes.TimestampNow()}), + wantErr: true, }, } for _, tt := range tests { @@ -472,15 +473,17 @@ func TestRepository_AuthorizeConnect(t *testing.T) { c, cs, authzInfo, err := repo.AuthorizeConnection(context.Background(), tt.session.PublicId) if tt.wantErr { require.Error(err) - if tt.wantIsError != nil { - assert.Truef(errors.Is(err, tt.wantIsError), "unexpected error %s", err.Error()) - } + // TODO (jimlambrt 9/2020): add in tests for errorsIs once we + // remove the grpc errors from the repo. + // if tt.wantIsError != nil { + // assert.Truef(errors.Is(err, tt.wantIsError), "unexpected error %s", err.Error()) + // } return } require.NoError(err) require.NotNil(c) require.NotNil(cs) - assert.Equal(StatusAuthorized.String(), cs[0].Status) + assert.Equal(StatusAuthorized, cs[0].Status) assert.Equal(tt.wantAuthzInfo.ExpirationTime, authzInfo.ExpirationTime) assert.Equal(tt.wantAuthzInfo.ConnectionLimit, authzInfo.ConnectionLimit) assert.Equal(tt.wantAuthzInfo.CurrentConnectionCount, authzInfo.CurrentConnectionCount) @@ -589,7 +592,7 @@ func TestRepository_ConnectSession(t *testing.T) { require.NoError(err) require.NotNil(c) require.NotNil(cs) - assert.Equal(StatusConnected.String(), cs[0].Status) + assert.Equal(StatusConnected, cs[0].Status) }) } } @@ -665,7 +668,7 @@ func TestRepository_TerminateSession(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { assert, require := assert.New(t), require.New(t) - s, ss, err := repo.TerminateSession(context.Background(), tt.session.PublicId, tt.session.Version, tt.reason) + s, err := repo.TerminateSession(context.Background(), tt.session.PublicId, tt.session.Version, tt.reason) if tt.wantErr { require.Error(err) if tt.wantIsError != nil { @@ -675,7 +678,7 @@ func TestRepository_TerminateSession(t *testing.T) { } require.NoError(err) assert.Equal(tt.reason.String(), s.TerminationReason) - assert.Equal(StatusTerminated.String(), ss[0].Status) + assert.Equal(StatusTerminated, s.States[0].Status) }) } } @@ -770,7 +773,7 @@ func TestRepository_CloseConnections(t *testing.T) { for _, r := range resp { require.NotNil(r.Connection) require.NotNil(r.ConnectionStates) - assert.Equal(StatusClosed.String(), r.ConnectionStates[0].Status) + assert.Equal(StatusClosed, r.ConnectionStates[0].Status) } }) } @@ -860,7 +863,7 @@ func TestRepository_CancelSession(t *testing.T) { version = tt.session.Version } - s, ss, err := repo.CancelSession(context.Background(), id, version) + s, err := repo.CancelSession(context.Background(), id, version) if tt.wantErr { require.Error(err) if tt.wantIsError != nil { @@ -870,8 +873,8 @@ func TestRepository_CancelSession(t *testing.T) { } require.NoError(err) require.NotNil(s) - require.NotNil(ss) - assert.Equal(StatusCancelling.String(), ss[0].Status) + require.NotNil(s.States) + assert.Equal(StatusCancelling, s.States[0].Status) }) } } @@ -980,7 +983,7 @@ func TestRepository_ActivateSession(t *testing.T) { require.NotNil(ss) assert.Equal(tofu, s.TofuToken) assert.Equal(2, len(ss)) - assert.Equal(StatusActive.String(), ss[0].Status) + assert.Equal(StatusActive, ss[0].Status) }) t.Run("already active", func(t *testing.T) { assert, require := assert.New(t), require.New(t) @@ -990,7 +993,7 @@ func TestRepository_ActivateSession(t *testing.T) { require.NotNil(s) require.NotNil(ss) assert.Equal(2, len(ss)) - assert.Equal(StatusActive.String(), ss[0].Status) + assert.Equal(StatusActive, ss[0].Status) _, _, err = repo.ActivateSession(context.Background(), session.PublicId, 1, worker.PrivateId, worker.Type, tofu) require.Error(err) @@ -1073,7 +1076,7 @@ func TestRepository_DeleteSession(t *testing.T) { } assert.NoError(err) assert.Equal(tt.wantRowsDeleted, deletedRows) - foundSession, _, err := repo.LookupSession(context.Background(), tt.args.session.PublicId) + foundSession, err := repo.LookupSession(context.Background(), tt.args.session.PublicId) assert.NoError(err) assert.Nil(foundSession) diff --git a/internal/session/session.go b/internal/session/session.go index 599494578..6891e2cce 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -95,7 +95,10 @@ type Session struct { // @inject_tag: `gorm:"not_null"` KeyId string `json:"key_id,omitempty" gorm:"not_null"` - tableName string `gorm:"-"` + // States for the session which are for read only and are ignored during + // write operations + States []*State `gorm:"-"` + tableName string `gorm:"-"` } func (s *Session) GetPublicId() string { @@ -352,3 +355,38 @@ func (s *Session) decrypt(ctx context.Context, cipher wrapping.Wrapper) error { } return nil } + +type sessionView struct { + // Session fields + PublicId string `json:"public_id,omitempty" gorm:"primary_key"` + UserId string `json:"user_id,omitempty" gorm:"default:null"` + HostId string `json:"host_id,omitempty" gorm:"default:null"` + ServerId string `json:"server_id,omitempty" gorm:"default:null"` + ServerType string `json:"server_type,omitempty" gorm:"default:null"` + TargetId string `json:"target_id,omitempty" gorm:"default:null"` + HostSetId string `json:"host_set_id,omitempty" gorm:"default:null"` + AuthTokenId string `json:"auth_token_id,omitempty" gorm:"default:null"` + ScopeId string `json:"scope_id,omitempty" gorm:"default:null"` + Certificate []byte `json:"certificate,omitempty" gorm:"default:null"` + ExpirationTime *timestamp.Timestamp `json:"expiration_time,omitempty" gorm:"default:null"` + CtTofuToken []byte `json:"ct_tofu_token,omitempty" gorm:"column:tofu_token;default:null" wrapping:"ct,tofu_token"` + TofuToken []byte `json:"tofu_token,omitempty" gorm:"-" wrapping:"pt,tofu_token"` + TerminationReason string `json:"termination_reason,omitempty" gorm:"default:null"` + CreateTime *timestamp.Timestamp `json:"create_time,omitempty" gorm:"default:current_timestamp"` + UpdateTime *timestamp.Timestamp `json:"update_time,omitempty" gorm:"default:current_timestamp"` + Version uint32 `json:"version,omitempty" gorm:"default:null"` + Endpoint string `json:"-" gorm:"default:null"` + ConnectionLimit uint32 `json:"connection_limit,omitempty" gorm:"default:null"` + KeyId string `json:"key_id,omitempty" gorm:"not_null"` + + // State fields + Status string `json:"state,omitempty" gorm:"column:state"` + PreviousEndTime *timestamp.Timestamp `json:"previous_end_time,omitempty" gorm:"default:current_timestamp"` + StartTime *timestamp.Timestamp `json:"start_time,omitempty" gorm:"default:current_timestamp;primary_key"` + EndTime *timestamp.Timestamp `json:"end_time,omitempty" gorm:"default:current_timestamp"` +} + +// TableName returns the tablename to override the default gorm table name +func (s *sessionView) TableName() string { + return "session_with_state" +}