Add Session tracker to DB, App, and Windows Desktop Sessions (#12304)

api/client/client.go

@@ -2431,7 +2431,7 @@ func (c *Client) CreateRegisterChallenge(ctx context.Context, in *proto.CreateRe
 
 // GenerateCertAuthorityCRL generates an empty CRL for a CA.
 func (c *Client) GenerateCertAuthorityCRL(ctx context.Context, req *proto.CertAuthorityRequest) (*proto.CRL, error) {
-	resp, err := c.grpc.GenerateCertAuthorityCRL(ctx, req)
+	resp, err := c.grpc.GenerateCertAuthorityCRL(ctx, req, c.callOpts...)
 	return resp, trail.FromGRPC(err)
 }
 
@@ -2532,21 +2532,48 @@ func GetResourcesWithFilters(ctx context.Context, clt ListResourcesClient, req p
 }
 
 // CreateSessionTracker creates a tracker resource for an active session.
-func (c *Client) CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error) {
-	resp, err := c.grpc.CreateSessionTracker(ctx, req)
-	return resp, trail.FromGRPC(err)
+func (c *Client) CreateSessionTracker(ctx context.Context, st types.SessionTracker) (types.SessionTracker, error) {
+	v1, ok := st.(*types.SessionTrackerV1)
+	if !ok {
+		return nil, trace.BadParameter("invalid type %T, expected *types.SessionTrackerV1", st)
+	}
+
+	req := &proto.CreateSessionTrackerRequest{SessionTracker: v1}
+
+	// DELETE IN 11.0.0
+	// Early v9 versions use a flattened out types.SessionTrackerV1
+	req.ID = v1.Spec.SessionID
+	req.Type = v1.Spec.Kind
+	req.Reason = v1.Spec.Reason
+	req.Invited = v1.Spec.Invited
+	req.Hostname = v1.Spec.Hostname
+	req.Address = v1.Spec.Address
+	req.ClusterName = v1.Spec.ClusterName
+	req.Login = v1.Spec.Login
+	req.Expires = v1.Spec.Expires
+	req.KubernetesCluster = v1.Spec.KubernetesCluster
+	req.HostUser = v1.Spec.HostUser
+	if len(v1.Spec.Participants) > 0 {
+		req.Initiator = &v1.Spec.Participants[0]
+	}
+
+	tracker, err := c.grpc.CreateSessionTracker(ctx, req, c.callOpts...)
+	if err != nil {
+		return nil, trail.FromGRPC(err)
+	}
+	return tracker, nil
 }
 
 // GetSessionTracker returns the current state of a session tracker for an active session.
 func (c *Client) GetSessionTracker(ctx context.Context, sessionID string) (types.SessionTracker, error) {
 	req := &proto.GetSessionTrackerRequest{SessionID: sessionID}
-	resp, err := c.grpc.GetSessionTracker(ctx, req)
+	resp, err := c.grpc.GetSessionTracker(ctx, req, c.callOpts...)
 	return resp, trail.FromGRPC(err)
 }
 
 // GetActiveSessionTrackers returns a list of active session trackers.
 func (c *Client) GetActiveSessionTrackers(ctx context.Context) ([]types.SessionTracker, error) {
-	stream, err := c.grpc.GetActiveSessionTrackers(ctx, &empty.Empty{})
+	stream, err := c.grpc.GetActiveSessionTrackers(ctx, &empty.Empty{}, c.callOpts...)
 	if err != nil {
 		return nil, trail.FromGRPC(err)
 	}
@@ -2570,18 +2597,18 @@ func (c *Client) GetActiveSessionTrackers(ctx context.Context) ([]types.SessionT
 
 // RemoveSessionTracker removes a tracker resource for an active session.
 func (c *Client) RemoveSessionTracker(ctx context.Context, sessionID string) error {
-	_, err := c.grpc.RemoveSessionTracker(ctx, &proto.RemoveSessionTrackerRequest{SessionID: sessionID})
+	_, err := c.grpc.RemoveSessionTracker(ctx, &proto.RemoveSessionTrackerRequest{SessionID: sessionID}, c.callOpts...)
 	return trail.FromGRPC(err)
 }
 
 // UpdateSessionTracker updates a tracker resource for an active session.
 func (c *Client) UpdateSessionTracker(ctx context.Context, req *proto.UpdateSessionTrackerRequest) error {
-	_, err := c.grpc.UpdateSessionTracker(ctx, req)
+	_, err := c.grpc.UpdateSessionTracker(ctx, req, c.callOpts...)
 	return trail.FromGRPC(err)
 }
 
 // MaintainSessionPresence establishes a channel used to continuously verify the presence for a session.
 func (c *Client) MaintainSessionPresence(ctx context.Context) (proto.AuthService_MaintainSessionPresenceClient, error) {
-	stream, err := c.grpc.MaintainSessionPresence(ctx)
+	stream, err := c.grpc.MaintainSessionPresence(ctx, c.callOpts...)
 	return stream, trail.FromGRPC(err)
 }

api/client/proto/authservice.proto

@@ -1538,35 +1538,45 @@ message ListResourcesResponse {
 // This is not specific to any session type. Relevant fields should be set for a given session type.
 message CreateSessionTrackerRequest {
     // Namespace is a session namespace, separating sessions from each other.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Namespace = 1 [ (gogoproto.jsontag) = "namespace,omitempty" ];
 
     // Type describes what type of session this is.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Type = 2 [ (gogoproto.jsontag) = "type,omitempty" ];
 
     // Reason is an arbitrary string that may be used to describe the session and/or it's
     // purpose.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Reason = 3 [ (gogoproto.jsontag) = "reason,omitempty" ];
 
     // Invited is a list of invited users, this field is interpreted by different
     // clients on a best-effort basis and used for delivering notifications to invited users.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     repeated string Invited = 4 [ (gogoproto.jsontag) = "invited,omitempty" ];
 
     // Hostname is the address of the target this session is connected to.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Hostname = 5 [ (gogoproto.jsontag) = "target_hostname,omitempty" ];
 
     // Address is the address of the target this session is connected to.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Address = 6 [ (gogoproto.jsontag) = "target_address,omitempty" ];
 
     // ClusterName is the name of cluster that this session belongs to.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string ClusterName = 7 [ (gogoproto.jsontag) = "cluster_name,omitempty" ];
 
     // Login is the local login/user on the target used by the session.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string Login = 8 [ (gogoproto.jsontag) = "login,omitempty" ];
 
     // Initiator is the participant that initiated the session.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     types.Participant Initiator = 9 [ (gogoproto.jsontag) = "initiator,omitempty" ];
 
     // Expires encodes the time at which this session expires and becomes invalid.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     google.protobuf.Timestamp Expires = 10 [
         (gogoproto.stdtime) = true,
         (gogoproto.nullable) = false,
@@ -1574,19 +1584,27 @@ message CreateSessionTrackerRequest {
     ];
 
     // The Kubernetes cluster this session belongs to.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string KubernetesCluster = 11 [ (gogoproto.jsontag) = "kubernetes_cluster,omitempty" ];
 
     // HostUser is the user regarded as the owner of this session, RBAC checks are performed
     // against the require policies of this user.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string HostUser = 12 [ (gogoproto.jsontag) = "host_user,omitempty" ];
 
     // ID is the ID of the session.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     string ID = 13 [ (gogoproto.jsontag) = "id,omitempty" ];
 
     // HostPolicies is a list of RBAC policy sets held by the host user at the time of session
     // creation.
+    // DELETE IN V11 - deprecated/reserve in favor of SessionTracker field.
     repeated types.SessionTrackerPolicySet HostPolicies = 14
         [ (gogoproto.jsontag) = "host_policies,omitempty" ];
+
+    // SessionTracker is the session tracker to be created.
+    types.SessionTrackerV1 SessionTracker = 15
+        [ (gogoproto.jsontag) = "session_tracker,omitempty" ];
 }
 
 // GetSessionTrackerRequest is a request to fetch a session resource.

api/defaults/defaults.go

@@ -48,6 +48,9 @@ const (
 	// deviation added to this time to avoid lots of simultaneous
 	// heartbeats coming to auth server
 	ServerAnnounceTTL = 600 * time.Second
+
+	// SessionTrackerTTL defines the default base ttl of a session tracker.
+	SessionTrackerTTL = time.Hour
 )
 
 var (

api/types/events/events.pb.go

@@ -1655,7 +1655,9 @@ func (m *AccessRequestCreate) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_AccessRequestCreate proto.InternalMessageInfo
 
-// ResourceID is a unique identifier for a teleport resource.
+// ResourceID is a unique identifier for a teleport resource. This is duplicated
+// from api/types/types.proto to decouple the api and events types and because
+// neither file currently imports the other.
 type ResourceID struct {
 	// ClusterName is the name of the cluster the resource is in.
 	ClusterName string `protobuf:"bytes,1,opt,name=ClusterName,proto3" json:"cluster"`

api/types/session_tracker.go

@@ -19,15 +19,20 @@ package types
 import (
 	"time"
 
+	"github.com/gravitational/teleport/api/defaults"
+
 	"github.com/gravitational/trace"
 )
 
 const (
-	SSHSessionKind        SessionKind            = "ssh"
-	KubernetesSessionKind SessionKind            = "k8s"
-	SessionObserverMode   SessionParticipantMode = "observer"
-	SessionModeratorMode  SessionParticipantMode = "moderator"
-	SessionPeerMode       SessionParticipantMode = "peer"
+	SSHSessionKind            SessionKind            = "ssh"
+	KubernetesSessionKind     SessionKind            = "k8s"
+	DatabaseSessionKind       SessionKind            = "db"
+	AppSessionKind            SessionKind            = "app"
+	WindowsDesktopSessionKind SessionKind            = "desktop"
+	SessionObserverMode       SessionParticipantMode = "observer"
+	SessionModeratorMode      SessionParticipantMode = "moderator"
+	SessionPeerMode           SessionParticipantMode = "peer"
 )
 
 // SessionKind is a type of session.
@@ -52,6 +57,9 @@ type SessionTracker interface {
 	// SetState sets the state of the session.
 	SetState(SessionState) error
 
+	// SetCreated sets the time at which the session was created.
+	SetCreated(time.Time)
+
 	// GetCreated returns the time at which the session was created.
 	GetCreated() time.Time
 
@@ -186,6 +194,19 @@ func (s *SessionTrackerV1) CheckAndSetDefaults() error {
 		return trace.Wrap(err)
 	}
 
+	if s.GetCreated().IsZero() {
+		s.SetCreated(time.Now())
+	}
+
+	if s.Expiry().IsZero() {
+		// By default, resource expiration should match session expiration.
+		expiry := s.GetExpires()
+		if expiry.IsZero() {
+			expiry = s.GetCreated().Add(defaults.SessionTrackerTTL)
+		}
+		s.SetExpiry(expiry)
+	}
+
 	return nil
 }
 
@@ -220,6 +241,11 @@ func (s *SessionTrackerV1) GetCreated() time.Time {
 	return s.Spec.Created
 }
 
+// SetCreated returns the time at which the session was created.
+func (s *SessionTrackerV1) SetCreated(created time.Time) {
+	s.Spec.Created = created
+}
+
 // GetExpires return the time at which the session expires.
 func (s *SessionTrackerV1) GetExpires() time.Time {
 	return s.Spec.Expires

api/types/types.proto

@@ -3098,6 +3098,18 @@ message SessionTrackerSpecV1 {
     // creation.
     repeated SessionTrackerPolicySet HostPolicies = 16
         [ (gogoproto.jsontag) = "host_roles,omitempty" ];
+
+    // DatabaseName is the database server this session belongs to.
+    string DatabaseName = 17 [ (gogoproto.jsontag) = "database_name,omitempty" ];
+
+    // AppName is the app server this session belongs to.
+    string AppName = 18 [ (gogoproto.jsontag) = "app_name,omitempty" ];
+
+    // AppSessionID is the unique ID of the app access certificate used to start this app session.
+    string AppSessionID = 19 [ (gogoproto.jsontag) = "app_session_id,omitempty" ];
+
+    // DesktopName is the windows desktop server this session belongs to.
+    string DesktopName = 20 [ (gogoproto.jsontag) = "desktop_name,omitempty" ];
 }
 
 // SessionTrackerPolicySet is a set of RBAC policies held by the session tracker

lib/auth/auth.go

@@ -67,6 +67,7 @@ import (
 	"github.com/gravitational/teleport/lib/events"
 	kubeutils "github.com/gravitational/teleport/lib/kube/utils"
 	"github.com/gravitational/teleport/lib/limiter"
+	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/services/local"
 	"github.com/gravitational/teleport/lib/session"
@@ -2930,8 +2931,17 @@ func (a *Server) GetApp(ctx context.Context, name string) (types.Application, er
 }
 
 // CreateSessionTracker creates a tracker resource for an active session.
-func (a *Server) CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error) {
-	return a.SessionTrackerService.CreateSessionTracker(ctx, req)
+func (a *Server) CreateSessionTracker(ctx context.Context, tracker types.SessionTracker) (types.SessionTracker, error) {
+	// Don't allow sessions that require moderation without the enterprise feature enabled.
+	for _, policySet := range tracker.GetHostPolicySets() {
+		if len(policySet.RequireSessionJoin) != 0 {
+			if !modules.GetModules().Features().ModeratedSessions {
+				return nil, trace.AccessDenied("this Teleport cluster is not licensed for moderated sessions, please contact the cluster administrator")
+			}
+		}
+	}
+
+	return a.SessionTrackerService.CreateSessionTracker(ctx, tracker)
 }
 
 // GetActiveSessionTrackers returns a list of active session trackers.

lib/auth/auth_with_roles.go

@@ -195,10 +195,24 @@ func (a *ServerWithRoles) actionForKindSSHSession(namespace, verb string, sid se
 	return trace.Wrap(a.actionWithExtendedContext(namespace, types.KindSSHSession, verb, extendContext))
 }
 
-// hasBuiltinRole checks the type of the role set returned and the name.
-// Returns true if role set is builtin and the name matches.
-func (a *ServerWithRoles) hasBuiltinRole(name string) bool {
-	return HasBuiltinRole(a.context.Checker, name)
+// serverAction returns an access denied error if the role is not one of the builtin server roles.
+func (a *ServerWithRoles) serverAction() error {
+	role, ok := a.context.Identity.(BuiltinRole)
+	if !ok || !role.IsServer() {
+		return trace.AccessDenied("this request can be only executed by a teleport built-in server")
+	}
+	return nil
+}
+
+// hasBuiltinRole checks that the attached checker is a BuiltinRoleSet
+// and whether any of the given roles match the role set.
+func (a *ServerWithRoles) hasBuiltinRole(roles ...types.SystemRole) bool {
+	for _, role := range roles {
+		if HasBuiltinRole(a.context.Checker, string(role)) {
+			return true
+		}
+	}
+	return false
 }
 
 // HasBuiltinRole checks the type of the role set returned and the name.
@@ -240,18 +254,23 @@ func hasLocalUserRole(checker services.AccessChecker) bool {
 }
 
 // CreateSessionTracker creates a tracker resource for an active session.
-func (a *ServerWithRoles) CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error) {
-	if !a.hasBuiltinRole(string(types.RoleKube)) && !a.hasBuiltinRole(string(types.RoleNode)) && !a.hasBuiltinRole(string(types.RoleProxy)) {
-		return nil, trace.AccessDenied("this request can be only executed by a node, proxy or kube service")
+func (a *ServerWithRoles) CreateSessionTracker(ctx context.Context, tracker types.SessionTracker) (types.SessionTracker, error) {
+	if err := a.serverAction(); err != nil {
+		return nil, trace.Wrap(err)
 	}
 
-	return a.authServer.CreateSessionTracker(ctx, req)
+	tracker, err := a.authServer.CreateSessionTracker(ctx, tracker)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return tracker, nil
+
 }
 
 // GetSessionTracker returns the current state of a session tracker for an active session.
 func (a *ServerWithRoles) GetSessionTracker(ctx context.Context, sessionID string) (types.SessionTracker, error) {
-	if !a.hasBuiltinRole(string(types.RoleKube)) && !a.hasBuiltinRole(string(types.RoleNode)) && !a.hasBuiltinRole(string(types.RoleProxy)) {
-		return nil, trace.AccessDenied("this request can be only executed by a node, proxy or kube service")
+	if err := a.serverAction(); err != nil {
+		return nil, trace.Wrap(err)
 	}
 
 	return a.authServer.GetSessionTracker(ctx, sessionID)
@@ -316,8 +335,8 @@ func (a *ServerWithRoles) GetActiveSessionTrackers(ctx context.Context) ([]types
 
 // RemoveSessionTracker removes a tracker resource for an active session.
 func (a *ServerWithRoles) RemoveSessionTracker(ctx context.Context, sessionID string) error {
-	if !a.hasBuiltinRole(string(types.RoleKube)) && !a.hasBuiltinRole(string(types.RoleNode)) && !a.hasBuiltinRole(string(types.RoleProxy)) {
-		return trace.AccessDenied("this request can be only executed by a node, proxy or kube service")
+	if err := a.serverAction(); err != nil {
+		return trace.Wrap(err)
 	}
 
 	return a.authServer.RemoveSessionTracker(ctx, sessionID)
@@ -325,8 +344,8 @@ func (a *ServerWithRoles) RemoveSessionTracker(ctx context.Context, sessionID st
 
 // UpdateSessionTracker updates a tracker resource for an active session.
 func (a *ServerWithRoles) UpdateSessionTracker(ctx context.Context, req *proto.UpdateSessionTrackerRequest) error {
-	if !a.hasBuiltinRole(string(types.RoleKube)) && !a.hasBuiltinRole(string(types.RoleNode)) && !a.hasBuiltinRole(string(types.RoleProxy)) {
-		return trace.AccessDenied("this request can be only executed by a node, proxy or kube service")
+	if err := a.serverAction(); err != nil {
+		return trace.Wrap(err)
 	}
 
 	return a.authServer.UpdateSessionTracker(ctx, req)
@@ -337,7 +356,7 @@ func (a *ServerWithRoles) UpdateSessionTracker(ctx context.Context, req *proto.U
 func (a *ServerWithRoles) AuthenticateWebUser(req AuthenticateUserRequest) (types.WebSession, error) {
 	// authentication request has it's own authentication, however this limits the requests
 	// types to proxies to make it harder to break
-	if !a.hasBuiltinRole(string(types.RoleProxy)) {
+	if !a.hasBuiltinRole(types.RoleProxy) {
 		return nil, trace.AccessDenied("this request can be only executed by a proxy")
 	}
 	return a.authServer.AuthenticateWebUser(req)
@@ -348,7 +367,7 @@ func (a *ServerWithRoles) AuthenticateWebUser(req AuthenticateUserRequest) (type
 func (a *ServerWithRoles) AuthenticateSSHUser(req AuthenticateSSHRequest) (*SSHLoginResponse, error) {
 	// authentication request has it's own authentication, however this limits the requests
 	// types to proxies to make it harder to break
-	if !a.hasBuiltinRole(string(types.RoleProxy)) {
+	if !a.hasBuiltinRole(types.RoleProxy) {
 		return nil, trace.AccessDenied("this request can be only executed by a proxy")
 	}
 	return a.authServer.AuthenticateSSHUser(req)
@@ -599,7 +618,7 @@ func (a *ServerWithRoles) UpsertNode(ctx context.Context, s types.Server) (*type
 //
 // This logic has moved to KeepAliveServer.
 func (a *ServerWithRoles) KeepAliveNode(ctx context.Context, handle types.KeepAlive) error {
-	if !a.hasBuiltinRole(string(types.RoleNode)) {
+	if !a.hasBuiltinRole(types.RoleNode) {
 		return trace.AccessDenied("[10] access denied")
 	}
 	clusterName, err := a.GetDomainName()
@@ -635,7 +654,7 @@ func (a *ServerWithRoles) KeepAliveServer(ctx context.Context, handle types.Keep
 		if serverName != handle.Name {
 			return trace.AccessDenied("access denied")
 		}
-		if !a.hasBuiltinRole(string(types.RoleNode)) {
+		if !a.hasBuiltinRole(types.RoleNode) {
 			return trace.AccessDenied("access denied")
 		}
 		if err := a.action(apidefaults.Namespace, types.KindNode, types.VerbUpdate); err != nil {
@@ -651,7 +670,7 @@ func (a *ServerWithRoles) KeepAliveServer(ctx context.Context, handle types.Keep
 				return trace.AccessDenied("access denied")
 			}
 		}
-		if !a.hasBuiltinRole(string(types.RoleApp)) {
+		if !a.hasBuiltinRole(types.RoleApp) {
 			return trace.AccessDenied("access denied")
 		}
 		if err := a.action(apidefaults.Namespace, types.KindAppServer, types.VerbUpdate); err != nil {
@@ -664,7 +683,7 @@ func (a *ServerWithRoles) KeepAliveServer(ctx context.Context, handle types.Keep
 		if serverName != handle.HostID {
 			return trace.AccessDenied("access denied")
 		}
-		if !a.hasBuiltinRole(string(types.RoleDatabase)) {
+		if !a.hasBuiltinRole(types.RoleDatabase) {
 			return trace.AccessDenied("access denied")
 		}
 		if err := a.action(apidefaults.Namespace, types.KindDatabaseServer, types.VerbUpdate); err != nil {
@@ -674,14 +693,14 @@ func (a *ServerWithRoles) KeepAliveServer(ctx context.Context, handle types.Keep
 		if serverName != handle.Name {
 			return trace.AccessDenied("access denied")
 		}
-		if !a.hasBuiltinRole(string(types.RoleWindowsDesktop)) {
+		if !a.hasBuiltinRole(types.RoleWindowsDesktop) {
 			return trace.AccessDenied("access denied")
 		}
 		if err := a.action(apidefaults.Namespace, types.KindWindowsDesktopService, types.VerbUpdate); err != nil {
 			return trace.Wrap(err)
 		}
 	case constants.KeepAliveKube:
-		if serverName != handle.Name || !a.hasBuiltinRole(string(types.RoleKube)) {
+		if serverName != handle.Name || !a.hasBuiltinRole(types.RoleKube) {
 			return trace.AccessDenied("access denied")
 		}
 		if err := a.action(apidefaults.Namespace, types.KindKubeService, types.VerbUpdate); err != nil {
@@ -763,9 +782,9 @@ func (a *ServerWithRoles) NewWatcher(ctx context.Context, watch types.Watch) (ty
 		}
 	}
 	switch {
-	case a.hasBuiltinRole(string(types.RoleProxy)):
+	case a.hasBuiltinRole(types.RoleProxy):
 		watch.QueueSize = defaults.ProxyQueueSize
-	case a.hasBuiltinRole(string(types.RoleNode)):
+	case a.hasBuiltinRole(types.RoleNode):
 		watch.QueueSize = defaults.NodeQueueSize
 	}
 	return a.authServer.NewWatcher(ctx, watch)
@@ -798,8 +817,7 @@ func (a *ServerWithRoles) checkAccessToNode(server types.Server) error {
 	// In addition, allow proxy (and remote proxy) to access all nodes for its
 	// smart resolution address resolution. Once the smart resolution logic is
 	// moved to the auth server, this logic can be removed.
-	if a.hasBuiltinRole(string(types.RoleAdmin)) ||
-		a.hasBuiltinRole(string(types.RoleProxy)) ||
+	if a.hasBuiltinRole(types.RoleAdmin, types.RoleProxy) ||
 		a.hasRemoteBuiltinRole(string(types.RoleRemoteProxy)) {
 		return nil
 	}
@@ -1585,7 +1603,7 @@ func (a *ServerWithRoles) SubmitAccessReview(ctx context.Context, params types.A
 
 	// review author must match calling user, except in the case of the builtin admin role.  we make this
 	// exception in order to allow for convenient testing with local tctl connections.
-	if !a.hasBuiltinRole(string(types.RoleAdmin)) {
+	if !a.hasBuiltinRole(types.RoleAdmin) {
 		if params.Review.Author != a.context.User.GetName() {
 			return nil, trace.AccessDenied("user %q cannot submit reviews on behalf of %q", a.context.User.GetName(), params.Review.Author)
 		}
@@ -1711,7 +1729,7 @@ func (a *ServerWithRoles) GetUsers(withSecrets bool) ([]types.User, error) {
 	if withSecrets {
 		// TODO(fspmarshall): replace admin requirement with VerbReadWithSecrets once we've
 		// migrated to that model.
-		if !a.hasBuiltinRole(string(types.RoleAdmin)) {
+		if !a.hasBuiltinRole(types.RoleAdmin) {
 			err := trace.AccessDenied("user %q requested access to all users with secrets", a.context.User.GetName())
 			log.Warning(err)
 			if err := a.authServer.emitter.EmitAuditEvent(a.authServer.closeCtx, &apievents.UserLogin{
@@ -1742,7 +1760,7 @@ func (a *ServerWithRoles) GetUser(name string, withSecrets bool) (types.User, er
 	if withSecrets {
 		// TODO(fspmarshall): replace admin requirement with VerbReadWithSecrets once we've
 		// migrated to that model.
-		if !a.hasBuiltinRole(string(types.RoleAdmin)) {
+		if !a.hasBuiltinRole(types.RoleAdmin) {
 			err := trace.AccessDenied("user %q requested access to user %q with secrets", a.context.User.GetName(), name)
 			log.Warning(err)
 			if err := a.authServer.emitter.EmitAuditEvent(a.authServer.closeCtx, &apievents.UserLogin{
@@ -1852,7 +1870,7 @@ func (a *ServerWithRoles) generateUserCerts(ctx context.Context, req proto.UserC
 
 	// this prevents clients who have no chance at getting a cert and impersonating anyone
 	// from enumerating local users and hitting database
-	if !a.hasBuiltinRole(string(types.RoleAdmin)) && !a.context.Checker.CanImpersonateSomeone() && req.Username != a.context.User.GetName() {
+	if !a.hasBuiltinRole(types.RoleAdmin) && !a.context.Checker.CanImpersonateSomeone() && req.Username != a.context.User.GetName() {
 		return nil, trace.AccessDenied("access denied: impersonation is not allowed")
 	}
 
@@ -1973,7 +1991,7 @@ func (a *ServerWithRoles) generateUserCerts(ctx context.Context, req proto.UserC
 	checker := services.NewRoleSet(parsedRoles...)
 
 	switch {
-	case a.hasBuiltinRole(string(types.RoleAdmin)):
+	case a.hasBuiltinRole(types.RoleAdmin):
 		// builtin admins can impersonate anyone
 		// this is required for local tctl commands to work
 	case req.Username == a.context.User.GetName():
@@ -2043,7 +2061,7 @@ func (a *ServerWithRoles) generateUserCerts(ctx context.Context, req proto.UserC
 		ttl:               req.Expires.Sub(a.authServer.GetClock().Now()),
 		compatibility:     req.Format,
 		publicKey:         req.PublicKey,
-		overrideRoleTTL:   a.hasBuiltinRole(string(types.RoleAdmin)),
+		overrideRoleTTL:   a.hasBuiltinRole(types.RoleAdmin),
 		routeToCluster:    req.RouteToCluster,
 		kubernetesCluster: req.KubernetesCluster,
 		dbService:         req.RouteToDatabase.ServiceName,
@@ -3215,7 +3233,7 @@ func (a *ServerWithRoles) DeleteSemaphore(ctx context.Context, filter types.Sema
 // signed certificate if successful.
 func (a *ServerWithRoles) ProcessKubeCSR(req KubeCSR) (*KubeCSRResponse, error) {
 	// limits the requests types to proxies to make it harder to break
-	if !a.hasBuiltinRole(string(types.RoleProxy)) {
+	if !a.hasBuiltinRole(types.RoleProxy) {
 		return nil, trace.AccessDenied("this request can be only executed by a proxy")
 	}
 	return a.authServer.ProcessKubeCSR(req)
@@ -3272,7 +3290,7 @@ func (a *ServerWithRoles) DeleteAllDatabaseServers(ctx context.Context, namespac
 func (a *ServerWithRoles) SignDatabaseCSR(ctx context.Context, req *proto.DatabaseCSRRequest) (*proto.DatabaseCSRResponse, error) {
 	// Only proxy is allowed to request this certificate when proxying
 	// database client connection to a remote database service.
-	if !a.hasBuiltinRole(string(types.RoleProxy)) {
+	if !a.hasBuiltinRole(types.RoleProxy) {
 		return nil, trace.AccessDenied("this request can only be executed by a proxy service")
 	}
 	return a.authServer.SignDatabaseCSR(ctx, req)
@@ -3294,7 +3312,7 @@ func (a *ServerWithRoles) SignDatabaseCSR(ctx context.Context, req *proto.Databa
 func (a *ServerWithRoles) GenerateDatabaseCert(ctx context.Context, req *proto.DatabaseCertRequest) (*proto.DatabaseCertResponse, error) {
 	// Check if this is a local cluster admin, or a datababase service, or a
 	// user that is allowed to impersonate database service.
-	if !a.hasBuiltinRole(string(types.RoleDatabase)) && !a.hasBuiltinRole(string(types.RoleAdmin)) {
+	if !a.hasBuiltinRole(types.RoleDatabase, types.RoleAdmin) {
 		if err := a.canImpersonateBuiltinRole(types.RoleDatabase); err != nil {
 			log.WithError(err).Warnf("User %v tried to generate database certificate but is not allowed to impersonate %q system role.",
 				a.context.User.GetName(), types.RoleDatabase)
@@ -4259,9 +4277,7 @@ func (a *ServerWithRoles) DeleteAllWindowsDesktops(ctx context.Context) error {
 
 func (a *ServerWithRoles) filterWindowsDesktops(desktops []types.WindowsDesktop) ([]types.WindowsDesktop, error) {
 	// For certain built-in roles allow full access
-	if a.hasBuiltinRole(string(types.RoleAdmin)) ||
-		a.hasBuiltinRole(string(types.RoleProxy)) ||
-		a.hasBuiltinRole(string(types.RoleWindowsDesktop)) {
+	if a.hasBuiltinRole(types.RoleAdmin, types.RoleProxy, types.RoleWindowsDesktop) {
 		return desktops, nil
 	}
 
@@ -4288,7 +4304,7 @@ func (a *ServerWithRoles) checkAccessToWindowsDesktop(w types.WindowsDesktop) er
 // authentication.
 func (a *ServerWithRoles) GenerateWindowsDesktopCert(ctx context.Context, req *proto.WindowsDesktopCertRequest) (*proto.WindowsDesktopCertResponse, error) {
 	// Only windows_desktop_service should be requesting Windows certificates.
-	if !a.hasBuiltinRole(string(types.RoleWindowsDesktop)) {
+	if !a.hasBuiltinRole(types.RoleWindowsDesktop) {
 		return nil, trace.AccessDenied("access denied")
 	}
 	return a.authServer.GenerateWindowsDesktopCert(ctx, req)
@@ -4350,7 +4366,7 @@ func (a *ServerWithRoles) GetAccountRecoveryCodes(ctx context.Context, req *prot
 // GenerateCertAuthorityCRL generates an empty CRL for a CA.
 func (a *ServerWithRoles) GenerateCertAuthorityCRL(ctx context.Context, caType types.CertAuthType) ([]byte, error) {
 	// Only windows_desktop_service should be requesting CRLs
-	if !a.hasBuiltinRole(string(types.RoleWindowsDesktop)) {
+	if !a.hasBuiltinRole(types.RoleWindowsDesktop) {
 		return nil, trace.AccessDenied("access denied")
 	}
 	crl, err := a.authServer.GenerateCertAuthorityCRL(ctx, caType)

lib/auth/grpcserver.go

@@ -3622,17 +3622,43 @@ func (g *GRPCServer) CreateSessionTracker(ctx context.Context, req *proto.Create
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	session, err := auth.ServerWithRoles.CreateSessionTracker(ctx, req)
+
+	var createTracker types.SessionTracker = req.SessionTracker
+	// DELETE IN 11.0.0
+	// Early v9 versions use a flattened out types.SessionTrackerV1
+	if req.SessionTracker == nil {
+		spec := types.SessionTrackerSpecV1{
+			SessionID:         req.ID,
+			Kind:              req.Type,
+			State:             types.SessionState_SessionStatePending,
+			Reason:            req.Reason,
+			Invited:           req.Invited,
+			Hostname:          req.Hostname,
+			Address:           req.Address,
+			ClusterName:       req.ClusterName,
+			Login:             req.Login,
+			Participants:      []types.Participant{*req.Initiator},
+			Expires:           req.Expires,
+			KubernetesCluster: req.KubernetesCluster,
+			HostUser:          req.HostUser,
+		}
+		createTracker, err = types.NewSessionTracker(spec)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+	}
+
+	tracker, err := auth.ServerWithRoles.CreateSessionTracker(ctx, createTracker)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
-	defined, ok := session.(*types.SessionTrackerV1)
+	v1, ok := tracker.(*types.SessionTrackerV1)
 	if !ok {
-		return nil, trace.BadParameter("unexpected session type %T", session)
+		return nil, trace.BadParameter("unexpected session type %T", tracker)
 	}
 
-	return defined, nil
+	return v1, nil
 }
 
 // GetSessionTracker returns the current state of a session tracker for an active session.

lib/auth/moderated_sessions_test.go

@@ -20,11 +20,12 @@ import (
 	"context"
 	"testing"
 
-	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/backend/memory"
 	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/services/local"
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 	"github.com/stretchr/testify/require"
 )
 
@@ -35,17 +36,23 @@ func TestUnmoderatedSessionsAllowed(t *testing.T) {
 		ModeratedSessions: false, // Explicily turn off moderated sessions.
 	}})
 
+	srv := &Server{
+		clock: clockwork.NewRealClock(),
+	}
+
 	bk, err := memory.New(memory.Config{})
 	require.NoError(t, err)
 
-	srv, err := local.NewSessionTrackerService(bk)
+	srv.Services.SessionTrackerService, err = local.NewSessionTrackerService(bk)
 	require.NoError(t, err)
 
-	tracker, err := srv.CreateSessionTracker(context.Background(), &proto.CreateSessionTrackerRequest{
-		ID:        "foo",
-		Initiator: &types.Participant{},
+	tracker, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID: "foo",
 	})
+	require.NoError(t, err)
+	tracker.AddParticipant(types.Participant{})
 
+	_, err = srv.CreateSessionTracker(context.Background(), tracker)
 	require.NoError(t, err)
 	require.NotNil(t, tracker)
 }
@@ -57,15 +64,18 @@ func TestModeratedSessionsDisabled(t *testing.T) {
 		ModeratedSessions: false, // Explicily turn off moderated sessions.
 	}})
 
+	srv := &Server{
+		clock: clockwork.NewRealClock(),
+	}
+
 	bk, err := memory.New(memory.Config{})
 	require.NoError(t, err)
 
-	srv, err := local.NewSessionTrackerService(bk)
+	srv.Services.SessionTrackerService, err = local.NewSessionTrackerService(bk)
 	require.NoError(t, err)
 
-	tracker, err := srv.CreateSessionTracker(context.Background(), &proto.CreateSessionTrackerRequest{
-		ID:        "foo",
-		Initiator: &types.Participant{},
+	tracker, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID: "foo",
 		HostPolicies: []*types.SessionTrackerPolicySet{
 			{
 				Name:    "foo",
@@ -78,8 +88,12 @@ func TestModeratedSessionsDisabled(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
+	tracker.AddParticipant(types.Participant{})
 
+	tracker, err = srv.CreateSessionTracker(context.Background(), tracker)
 	require.Error(t, err)
+	require.True(t, trace.IsAccessDenied(err))
 	require.Nil(t, tracker)
 	require.Contains(t, err.Error(), "this Teleport cluster is not licensed for moderated sessions, please contact the cluster administrator")
 }
@@ -91,15 +105,18 @@ func TestModeratedSesssionsEnabled(t *testing.T) {
 		ModeratedSessions: true,
 	}})
 
+	srv := &Server{
+		clock: clockwork.NewRealClock(),
+	}
+
 	bk, err := memory.New(memory.Config{})
 	require.NoError(t, err)
 
-	srv, err := local.NewSessionTrackerService(bk)
+	srv.Services.SessionTrackerService, err = local.NewSessionTrackerService(bk)
 	require.NoError(t, err)
 
-	tracker, err := srv.CreateSessionTracker(context.Background(), &proto.CreateSessionTrackerRequest{
-		ID:        "foo",
-		Initiator: &types.Participant{},
+	tracker, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID: "foo",
 		HostPolicies: []*types.SessionTrackerPolicySet{
 			{
 				Name:    "foo",
@@ -112,7 +129,10 @@ func TestModeratedSesssionsEnabled(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
+	tracker.AddParticipant(types.Participant{})
 
+	_, err = srv.CreateSessionTracker(context.Background(), tracker)
 	require.NoError(t, err)
 	require.NotNil(t, tracker)
 }

lib/defaults/defaults.go

@@ -282,16 +282,9 @@ const (
 	// no name is provided at connection time.
 	DefaultRedisUsername = "default"
 
-	// SessionTrackerTTL defines the default base ttl of a session tracker.
-	SessionTrackerTTL = time.Hour
-
-	// SessionTrackerExpirationUpdateInterval is the default interval on which an active
-	// session's expiration will be extended.
-	SessionTrackerExpirationUpdateInterval = SessionTrackerTTL / 6
-
 	// AbandonedUploadPollingRate defines how often to check for
 	// abandoned uploads which need to be completed.
-	AbandonedUploadPollingRate = SessionTrackerTTL / 6
+	AbandonedUploadPollingRate = defaults.SessionTrackerTTL / 6
 
 	// UploadGracePeriod is a period after which non-completed
 	// upload is considered abandoned and will be completed by the reconciler

lib/events/complete.go

@@ -134,23 +134,13 @@ func (u *UploadCompleter) Serve(ctx context.Context) error {
 		case <-u.closeC:
 			return nil
 		case <-ctx.Done():
-			return trace.Wrap(ctx.Err(), "Context canceled")
+			return nil
 		}
 	}
 }
 
 // checkUploads fetches uploads and completes any abandoned uploads
 func (u *UploadCompleter) checkUploads(ctx context.Context) error {
-	trackers, err := u.cfg.SessionTracker.GetActiveSessionTrackers(ctx)
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	var activeSessionIDs []string
-	for _, st := range trackers {
-		activeSessionIDs = append(activeSessionIDs, st.GetSessionID())
-	}
-
 	uploads, err := u.cfg.Uploader.ListUploads(ctx)
 	if err != nil {
 		return trace.Wrap(err)
@@ -177,8 +167,10 @@ func (u *UploadCompleter) checkUploads(ctx context.Context) error {
 			}
 		}
 
-		if apiutils.SliceContainsStr(activeSessionIDs, upload.SessionID.String()) {
+		if _, err := u.cfg.SessionTracker.GetSessionTracker(ctx, upload.SessionID.String()); err == nil {
 			continue
+		} else if !trace.IsNotFound(err) {
+			return trace.Wrap(err)
 		}
 
 		parts, err := u.cfg.Uploader.ListParts(ctx, upload)

lib/events/complete_test.go

@@ -233,7 +233,7 @@ func (m *mockSessionTrackerService) GetSessionTracker(ctx context.Context, sessi
 	return nil, trace.NotFound("tracker not found")
 }
 
-func (m *mockSessionTrackerService) CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error) {
+func (m *mockSessionTrackerService) CreateSessionTracker(ctx context.Context, st types.SessionTracker) (types.SessionTracker, error) {
 	return nil, trace.NotImplemented("CreateSessionTracker is not implemented")
 }
 

lib/kube/proxy/sess.go

@@ -26,10 +26,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/gravitational/teleport"
-	"github.com/gravitational/teleport/api/client/proto"
-	"github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
 	apievents "github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/auth"
@@ -38,6 +35,8 @@ import (
 	tsession "github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/lib/srv"
 	"github.com/gravitational/teleport/lib/utils"
+
+	"github.com/google/uuid"
 	"github.com/gravitational/trace"
 	"github.com/julienschmidt/httprouter"
 	log "github.com/sirupsen/logrus"
@@ -272,10 +271,7 @@ type session struct {
 
 	terminalSizeQueue *multiResizeQueue
 
-	state types.SessionState
-
-	// stateUpdate is used to notify listeners about state updates
-	stateUpdate *sync.Cond
+	tracker *srv.SessionTracker
 
 	accessEvaluator auth.SessionAccessEvaluator
 
@@ -336,7 +332,6 @@ func newSession(ctx authContext, forwarder *Forwarder, req *http.Request, params
 		partiesHistorical:              make(map[uuid.UUID]*party),
 		log:                            log,
 		io:                             io,
-		state:                          types.SessionState_SessionStatePending,
 		accessEvaluator:                accessEvaluator,
 		emitter:                        events.NewDiscardEmitter(),
 		terminalSizeQueue:              newMultiResizeQueue(),
@@ -346,7 +341,6 @@ func newSession(ctx authContext, forwarder *Forwarder, req *http.Request, params
 		initiator:                      initiator.ID,
 		expires:                        time.Now().UTC().Add(sessionMaxLifetime),
 		PresenceEnabled:                ctx.Identity.GetIdentity().MFAVerified != "",
-		stateUpdate:                    sync.NewCond(&sync.Mutex{}),
 		displayParticipantRequirements: utils.AsBool(q.Get("displayParticipantRequirements")),
 	}
 
@@ -362,53 +356,20 @@ func newSession(ctx authContext, forwarder *Forwarder, req *http.Request, params
 		}
 	}()
 
-	err = s.trackerCreate(initiator, policySets)
-	if err != nil {
+	if err := s.trackSession(initiator, policySets); err != nil {
 		return nil, trace.Wrap(err)
 	}
 
 	return s, nil
 }
 
-// waitOnAccess puts the session in pending mode and waits for the session
-// to fulfill the access requirements again.
-func (s *session) waitOnAccess() {
-	s.io.Off()
-	s.BroadcastMessage("Session paused, Waiting for required participants...")
-
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-
-outer:
-	for {
-		switch s.state {
-		case types.SessionState_SessionStatePending:
-			continue
-		case types.SessionState_SessionStateTerminated:
-			return
-		case types.SessionState_SessionStateRunning:
-			break outer
-		}
-
-		s.stateUpdate.Wait()
-	}
-
-	s.BroadcastMessage("Resuming session...")
-	s.io.On()
-}
-
 // checkPresence checks the presence timestamp of involved moderators
 // and kicks them if they are not active.
 func (s *session) checkPresence() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	sess, err := s.trackerGet()
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	for _, participant := range sess.GetParticipants() {
+	for _, participant := range s.tracker.GetParticipants() {
 		if participant.ID == s.initiator.String() {
 			continue
 		}
@@ -538,8 +499,7 @@ func (s *session) launch() error {
 		}
 	}()
 
-	err = s.trackerUpdateState(types.SessionState_SessionStateRunning)
-	if err != nil {
+	if err := s.tracker.UpdateState(s.forwarder.ctx, types.SessionState_SessionStateRunning); err != nil {
 		s.log.Warn("Failed to set tracker state to running")
 	}
 
@@ -788,16 +748,18 @@ func (s *session) join(p *party) error {
 		}
 	}
 
-	s.stateUpdate.L.Lock()
-	state := s.state
-	s.stateUpdate.L.Unlock()
-
-	if state == types.SessionState_SessionStateTerminated {
+	if s.tracker.GetState() == types.SessionState_SessionStateTerminated {
 		return trace.AccessDenied("The requested session is not active")
 	}
 
-	err := s.trackerAddParticipant(p)
-	if err != nil {
+	s.log.Debugf("Tracking participant: %s", p.ID)
+	participant := &types.Participant{
+		ID:         p.ID.String(),
+		User:       p.Ctx.User.GetName(),
+		Mode:       string(p.Mode),
+		LastActive: time.Now().UTC(),
+	}
+	if err := s.tracker.AddParticipant(s.forwarder.ctx, participant); err != nil {
 		return trace.Wrap(err)
 	}
 
@@ -830,8 +792,7 @@ func (s *session) join(p *party) error {
 	}
 
 	recentWrites := s.io.GetRecentHistory()
-	_, err = p.Client.stdoutStream().Write(recentWrites)
-	if err != nil {
+	if _, err := p.Client.stdoutStream().Write(recentWrites); err != nil {
 		s.log.Warnf("Failed to write history to client: %v.", err)
 	}
 
@@ -901,10 +862,7 @@ func (s *session) BroadcastMessage(format string, args ...interface{}) {
 
 // leave removes a party from the session.
 func (s *session) leave(id uuid.UUID) error {
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-
-	if s.state == types.SessionState_SessionStateTerminated {
+	if s.tracker.GetState() == types.SessionState_SessionStateTerminated {
 		return nil
 	}
 
@@ -945,7 +903,8 @@ func (s *session) leave(id uuid.UUID) error {
 		s.forwarder.log.WithError(err).Warn("Failed to emit event.")
 	}
 
-	err := s.trackerRemoveParticipant(party.ID.String())
+	s.log.Debugf("No longer tracking participant: %v", party.ID)
+	err := s.tracker.RemoveParticipant(s.forwarder.ctx, party.ID.String())
 	if err != nil {
 		return trace.Wrap(err)
 	}
@@ -975,21 +934,26 @@ func (s *session) leave(id uuid.UUID) error {
 	if !canStart {
 		if options.TerminateOnLeave {
 			go func() {
-				err := s.Close()
-				if err != nil {
+				if err := s.Close(); err != nil {
 					s.log.WithError(err).Errorf("Failed to close session")
 				}
 			}()
-		} else {
-			s.state = types.SessionState_SessionStatePending
-			s.stateUpdate.Broadcast()
-			err := s.trackerUpdateState(types.SessionState_SessionStateRunning)
-			if err != nil {
-				s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStateRunning)
-			}
-
-			go s.waitOnAccess()
+			return nil
 		}
+
+		// pause session and wait for another party to resume
+		s.io.Off()
+		s.BroadcastMessage("Session paused, Waiting for required participants...")
+		if err := s.tracker.UpdateState(s.forwarder.ctx, types.SessionState_SessionStatePending); err != nil {
+			s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStatePending)
+		}
+
+		go func() {
+			if state := s.tracker.WaitForStateUpdate(types.SessionState_SessionStatePending); state == types.SessionState_SessionStateRunning {
+				s.BroadcastMessage("Resuming session...")
+				s.io.On()
+			}
+		}()
 	}
 
 	return nil
@@ -1037,21 +1001,17 @@ func (s *session) Close() error {
 
 	s.closeOnce.Do(func() {
 		s.BroadcastMessage("Closing session...")
-		s.stateUpdate.L.Lock()
-		defer s.stateUpdate.L.Unlock()
-		s.state = types.SessionState_SessionStateTerminated
+
 		s.io.Close()
-		s.stateUpdate.Broadcast()
-		err := s.trackerUpdateState(types.SessionState_SessionStateTerminated)
-		if err != nil {
-			s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStateTerminated)
+
+		if err := s.tracker.Close(s.forwarder.ctx); err != nil {
+			s.log.WithError(err).Debug("Failed to close session tracker")
 		}
 
 		s.log.Debugf("Closing session %v.", s.id.String())
 		close(s.closeC)
 		for id, party := range s.parties {
-			err = party.Close()
-			if err != nil {
+			if err := party.Close(); err != nil {
 				s.log.WithError(err).Errorf("Failed to disconnect party %v", id.String())
 			}
 		}
@@ -1079,85 +1039,40 @@ func getRolesByName(forwarder *Forwarder, roleNames []string) ([]types.Role, err
 	return roles, nil
 }
 
-func (s *session) trackerGet() (types.SessionTracker, error) {
-	sess, err := s.forwarder.cfg.AuthClient.GetSessionTracker(s.forwarder.ctx, s.id.String())
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	return sess, nil
-}
-
-func (s *session) trackerCreate(p *party, policySets []*types.SessionTrackerPolicySet) error {
-	initiator := &types.Participant{
-		ID:         p.ID.String(),
-		User:       p.Ctx.User.GetName(),
-		LastActive: time.Now().UTC(),
-	}
-
-	req := &proto.CreateSessionTrackerRequest{
-		ID:                s.id.String(),
-		Namespace:         defaults.Namespace,
-		Type:              string(types.KubernetesSessionKind),
-		Hostname:          s.podName,
-		ClusterName:       s.ctx.teleportCluster.name,
-		Initiator:         initiator,
-		Expires:           s.expires,
+// trackSession creates a new session tracker for the kube session.
+// While ctx is open, the session tracker's expiration will be extended
+// on an interval until the session tracker is closed.
+func (s *session) trackSession(p *party, policySet []*types.SessionTrackerPolicySet) error {
+	trackerSpec := types.SessionTrackerSpecV1{
+		SessionID:   s.id.String(),
+		Kind:        string(types.KubernetesSessionKind),
+		State:       types.SessionState_SessionStatePending,
+		Hostname:    s.podName,
+		ClusterName: s.ctx.teleportCluster.name,
+		Participants: []types.Participant{{
+			ID:         p.ID.String(),
+			User:       p.Ctx.User.GetName(),
+			LastActive: time.Now().UTC(),
+		}},
 		KubernetesCluster: s.ctx.kubeCluster,
-		HostUser:          initiator.User,
-		HostPolicies:      policySets,
+		HostUser:          p.Ctx.User.GetName(),
+		HostPolicies:      policySet,
 		Login:             "root",
+		Created:           time.Now(),
 	}
 
-	_, err := s.forwarder.cfg.AuthClient.CreateSessionTracker(s.forwarder.ctx, req)
-	return trace.Wrap(err)
-}
-
-func (s *session) trackerAddParticipant(participant *party) error {
-	s.log.Debugf("Tracking participant: %v", participant.ID.String())
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_AddParticipant{
-			AddParticipant: &proto.SessionTrackerAddParticipant{
-				Participant: &types.Participant{
-					ID:         participant.ID.String(),
-					User:       participant.Ctx.User.GetName(),
-					Mode:       string(participant.Mode),
-					LastActive: time.Now().UTC(),
-				},
-			},
-		},
+	s.log.Debug("Creating session tracker")
+	var err error
+	s.tracker, err = srv.NewSessionTracker(s.forwarder.ctx, trackerSpec, s.forwarder.cfg.AuthClient)
+	if err != nil {
+		return trace.Wrap(err)
 	}
 
-	err := s.forwarder.cfg.AuthClient.UpdateSessionTracker(s.forwarder.ctx, req)
-	return trace.Wrap(err)
-}
+	go func() {
+		if err := s.tracker.UpdateExpirationLoop(s.forwarder.ctx, s.forwarder.cfg.Clock); err != nil {
+			s.log.WithError(err).Debug("Failed to update session tracker expiration")
+		}
+	}()
 
-func (s *session) trackerRemoveParticipant(participantID string) error {
-	s.log.Debugf("Not tracking participant: %v", participantID)
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_RemoveParticipant{
-			RemoveParticipant: &proto.SessionTrackerRemoveParticipant{
-				ParticipantID: participantID,
-			},
-		},
-	}
-
-	err := s.forwarder.cfg.AuthClient.UpdateSessionTracker(s.forwarder.ctx, req)
-	return trace.Wrap(err)
-}
-
-func (s *session) trackerUpdateState(state types.SessionState) error {
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_UpdateState{
-			UpdateState: &proto.SessionTrackerUpdateState{
-				State: state,
-			},
-		},
-	}
-
-	err := s.forwarder.cfg.AuthClient.UpdateSessionTracker(s.forwarder.ctx, req)
-	return trace.Wrap(err)
+	return nil
 }

lib/services/local/sessiontracker.go

@@ -23,10 +23,7 @@ import (
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/backend"
-	"github.com/gravitational/teleport/lib/defaults"
-	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/services"
-	"github.com/gravitational/teleport/lib/utils"
 
 	"github.com/gravitational/trace"
 	"github.com/sirupsen/logrus"
@@ -53,7 +50,7 @@ func (s *sessionTracker) loadSession(ctx context.Context, sessionID string) (typ
 		return nil, trace.Wrap(err)
 	}
 
-	session, err := unmarshalSession(sessionJSON.Value)
+	session, err := services.UnmarshalSessionTracker(sessionJSON.Value)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -69,7 +66,7 @@ func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user str
 			return trace.Wrap(err)
 		}
 
-		session, err := unmarshalSession(sessionItem.Value)
+		session, err := services.UnmarshalSessionTracker(sessionItem.Value)
 		if err != nil {
 			return trace.Wrap(err)
 		}
@@ -79,7 +76,7 @@ func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user str
 			return trace.Wrap(err)
 		}
 
-		sessionJSON, err := marshalSession(session)
+		sessionJSON, err := services.MarshalSessionTracker(session)
 		if err != nil {
 			return trace.Wrap(err)
 		}
@@ -129,7 +126,7 @@ func (s *sessionTracker) GetActiveSessionTrackers(ctx context.Context) ([]types.
 	var noExpiry []backend.Item
 	now := s.bk.Clock().Now()
 	for _, item := range result.Items {
-		session, err := unmarshalSession(item.Value)
+		session, err := services.UnmarshalSessionTracker(item.Value)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
@@ -171,62 +168,23 @@ func (s *sessionTracker) GetActiveSessionTrackers(ctx context.Context) ([]types.
 }
 
 // CreateSessionTracker creates a tracker resource for an active session.
-func (s *sessionTracker) CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error) {
-	// Don't allow sessions that require moderation without the enterprise feature enabled.
-	for _, policySet := range req.HostPolicies {
-		if len(policySet.RequireSessionJoin) != 0 {
-			if !modules.GetModules().Features().ModeratedSessions {
-				return nil, trace.AccessDenied(
-					"this Teleport cluster is not licensed for moderated sessions, please contact the cluster administrator")
-			}
-		}
-	}
-
-	now := s.bk.Clock().Now()
-	spec := types.SessionTrackerSpecV1{
-		SessionID:         req.ID,
-		Kind:              req.Type,
-		State:             types.SessionState_SessionStatePending,
-		Created:           now,
-		Reason:            req.Reason,
-		Invited:           req.Invited,
-		Hostname:          req.Hostname,
-		Address:           req.Address,
-		ClusterName:       req.ClusterName,
-		Login:             req.Login,
-		Participants:      []types.Participant{*req.Initiator},
-		Expires:           req.Expires,
-		KubernetesCluster: req.KubernetesCluster,
-		HostUser:          req.HostUser,
-	}
-
-	session, err := types.NewSessionTracker(spec)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	// By default, resource expiration should match session expiration.
-	session.SetExpiry(session.GetExpires())
-	if session.Expiry().IsZero() {
-		session.SetExpiry(now.Add(defaults.SessionTrackerTTL))
-	}
-
-	json, err := marshalSession(session)
+func (s *sessionTracker) CreateSessionTracker(ctx context.Context, tracker types.SessionTracker) (types.SessionTracker, error) {
+	json, err := services.MarshalSessionTracker(tracker)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
 	item := backend.Item{
-		Key:     backend.Key(sessionPrefix, session.GetSessionID()),
+		Key:     backend.Key(sessionPrefix, tracker.GetSessionID()),
 		Value:   json,
-		Expires: session.Expiry(),
+		Expires: tracker.Expiry(),
 	}
 	_, err = s.bk.Put(ctx, item)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
 
-	return session, nil
+	return tracker, nil
 }
 
 // UpdateSessionTracker updates a tracker resource for an active session.
@@ -237,7 +195,7 @@ func (s *sessionTracker) UpdateSessionTracker(ctx context.Context, req *proto.Up
 			return trace.Wrap(err)
 		}
 
-		session, err := unmarshalSession(sessionItem.Value)
+		session, err := services.UnmarshalSessionTracker(sessionItem.Value)
 		if err != nil {
 			return trace.Wrap(err)
 		}
@@ -258,7 +216,7 @@ func (s *sessionTracker) UpdateSessionTracker(ctx context.Context, req *proto.Up
 			return trace.BadParameter("unrecognized session version %T", session)
 		}
 
-		sessionJSON, err := marshalSession(session)
+		sessionJSON, err := services.MarshalSessionTracker(session)
 		if err != nil {
 			return trace.Wrap(err)
 		}
@@ -288,59 +246,3 @@ func (s *sessionTracker) UpdateSessionTracker(ctx context.Context, req *proto.Up
 func (s *sessionTracker) RemoveSessionTracker(ctx context.Context, sessionID string) error {
 	return trace.Wrap(s.bk.Delete(ctx, backend.Key(sessionPrefix, sessionID)))
 }
-
-// unmarshalSession unmarshals the Session resource from JSON.
-func unmarshalSession(bytes []byte, opts ...services.MarshalOption) (types.SessionTracker, error) {
-	var session types.SessionTrackerV1
-
-	if len(bytes) == 0 {
-		return nil, trace.BadParameter("missing resource data")
-	}
-
-	cfg, err := services.CollectOptions(opts)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	if err := utils.FastUnmarshal(bytes, &session); err != nil {
-		return nil, trace.BadParameter(err.Error())
-	}
-
-	if err := session.CheckAndSetDefaults(); err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	if cfg.ID != 0 {
-		session.SetResourceID(cfg.ID)
-	}
-
-	if !cfg.Expires.IsZero() {
-		session.SetExpiry(cfg.Expires)
-	}
-
-	return &session, nil
-}
-
-// marshalSession marshals the Session resource to JSON.
-func marshalSession(session types.SessionTracker, opts ...services.MarshalOption) ([]byte, error) {
-	if err := session.CheckAndSetDefaults(); err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	cfg, err := services.CollectOptions(opts)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	switch session := session.(type) {
-	case *types.SessionTrackerV1:
-		if !cfg.PreserveResourceID {
-			copy := *session
-			copy.SetResourceID(0)
-			session = &copy
-		}
-		return utils.FastMarshal(session)
-	default:
-		return nil, trace.BadParameter("unrecognized session version %T", session)
-	}
-}

lib/services/local/sessiontracker_test.go

@@ -18,11 +18,12 @@ import (
 	"testing"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/gravitational/teleport/api/client/proto"
-	"github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/backend/memory"
+
+	"github.com/google/uuid"
+	"github.com/gravitational/trace"
 	"github.com/stretchr/testify/require"
 )
 
@@ -32,29 +33,34 @@ func TestSessionTrackerStorage(t *testing.T) {
 	bk, err := memory.New(memory.Config{})
 	require.NoError(t, err)
 
-	id := uuid.New().String()
+	sid := uuid.New().String()
 	srv, err := NewSessionTrackerService(bk)
 	require.NoError(t, err)
 
-	session, err := srv.CreateSessionTracker(ctx, &proto.CreateSessionTrackerRequest{
-		Namespace:   defaults.Namespace,
-		ID:          id,
-		Type:        types.KindSSHSession,
+	tracker, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID:   sid,
+		Kind:        types.KindSSHSession,
 		Hostname:    "hostname",
 		ClusterName: "cluster",
 		Login:       "root",
-		Initiator: &types.Participant{
-			ID:   uuid.New().String(),
-			User: "eve",
-			Mode: string(types.SessionPeerMode),
+		Participants: []types.Participant{
+			{
+				ID:   uuid.New().String(),
+				User: "eve",
+				Mode: string(types.SessionPeerMode),
+			},
 		},
 		Expires: time.Now().UTC().Add(24 * time.Hour),
 	})
 	require.NoError(t, err)
 
+	_, err = srv.CreateSessionTracker(ctx, tracker)
+	require.NoError(t, err)
+
 	bobID := uuid.New().String()
-	err = srv.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
-		SessionID: id,
+
+	req := &proto.UpdateSessionTrackerRequest{
+		SessionID: sid,
 		Update: &proto.UpdateSessionTrackerRequest_AddParticipant{
 			AddParticipant: &proto.SessionTrackerAddParticipant{
 				Participant: &types.Participant{
@@ -64,30 +70,36 @@ func TestSessionTrackerStorage(t *testing.T) {
 				},
 			},
 		},
-	})
+	}
+
+	err = srv.UpdateSessionTracker(ctx, req)
 	require.NoError(t, err)
 
-	err = srv.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
-		SessionID: id,
+	req = &proto.UpdateSessionTrackerRequest{
+		SessionID: sid,
 		Update: &proto.UpdateSessionTrackerRequest_RemoveParticipant{
 			RemoveParticipant: &proto.SessionTrackerRemoveParticipant{
 				ParticipantID: bobID,
 			},
 		},
-	})
+	}
+
+	err = srv.UpdateSessionTracker(ctx, req)
 	require.NoError(t, err)
 
 	sessions, err := srv.GetActiveSessionTrackers(ctx)
 	require.NoError(t, err)
 	require.Len(t, sessions, 1)
-	require.Len(t, session.GetParticipants(), 1)
+	tracker = sessions[0]
+	require.Len(t, tracker.GetParticipants(), 1)
 
-	err = srv.RemoveSessionTracker(ctx, session.GetSessionID())
+	err = srv.RemoveSessionTracker(ctx, sid)
 	require.NoError(t, err)
 
-	session, err = srv.GetSessionTracker(ctx, session.GetSessionID())
+	tracker, err = srv.GetSessionTracker(ctx, sid)
 	require.Error(t, err)
-	require.Nil(t, session)
+	require.True(t, trace.IsNotFound(err))
+	require.Nil(t, tracker)
 }
 
 func TestSessionTrackerImplicitExpiry(t *testing.T) {
@@ -100,29 +112,44 @@ func TestSessionTrackerImplicitExpiry(t *testing.T) {
 	srv, err := NewSessionTrackerService(bk)
 	require.NoError(t, err)
 
-	req1 := proto.CreateSessionTrackerRequest{
-		Namespace:   defaults.Namespace,
-		ID:          id,
-		Type:        types.KindSSHSession,
+	tracker1, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID:   id,
+		Kind:        types.KindSSHSession,
 		Hostname:    "hostname",
 		ClusterName: "cluster",
 		Login:       "foo",
-		Initiator: &types.Participant{
-			ID:   uuid.New().String(),
-			User: "eve",
-			Mode: string(types.SessionPeerMode),
+		Participants: []types.Participant{
+			{
+				ID:   uuid.New().String(),
+				User: "eve",
+				Mode: string(types.SessionPeerMode),
+			},
 		},
 		Expires: time.Now().UTC().Add(time.Second),
-	}
-
-	req2 := req1
-	req2.ID = id2
-	req2.Expires = time.Now().UTC().Add(24 * time.Hour)
-
-	_, err = srv.CreateSessionTracker(ctx, &req1)
+	})
 	require.NoError(t, err)
 
-	_, err = srv.CreateSessionTracker(ctx, &req2)
+	_, err = srv.CreateSessionTracker(ctx, tracker1)
+	require.NoError(t, err)
+
+	tracker2, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID:   id2,
+		Kind:        types.KindSSHSession,
+		Hostname:    "hostname",
+		ClusterName: "cluster",
+		Login:       "foo",
+		Participants: []types.Participant{
+			{
+				ID:   uuid.New().String(),
+				User: "eve",
+				Mode: string(types.SessionPeerMode),
+			},
+		},
+		Expires: time.Now().UTC().Add(24 * time.Hour),
+	})
+	require.NoError(t, err)
+
+	_, err = srv.CreateSessionTracker(ctx, tracker2)
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {

lib/services/sessiontracker.go

@@ -21,6 +21,9 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/utils"
+
+	"github.com/gravitational/trace"
 )
 
 // SessionTrackerService is a realtime session service that has information about
@@ -33,7 +36,7 @@ type SessionTrackerService interface {
 	GetSessionTracker(ctx context.Context, sessionID string) (types.SessionTracker, error)
 
 	// CreateSessionTracker creates a tracker resource for an active session.
-	CreateSessionTracker(ctx context.Context, req *proto.CreateSessionTrackerRequest) (types.SessionTracker, error)
+	CreateSessionTracker(ctx context.Context, st types.SessionTracker) (types.SessionTracker, error)
 
 	// UpdateSessionTracker updates a tracker resource for an active session.
 	UpdateSessionTracker(ctx context.Context, req *proto.UpdateSessionTrackerRequest) error
@@ -44,3 +47,35 @@ type SessionTrackerService interface {
 	// UpdatePresence updates the presence status of a user in a session.
 	UpdatePresence(ctx context.Context, sessionID, user string) error
 }
+
+// UnmarshalSessionTracker unmarshals the Session resource from JSON.
+func UnmarshalSessionTracker(bytes []byte) (types.SessionTracker, error) {
+	if len(bytes) == 0 {
+		return nil, trace.BadParameter("missing resource data")
+	}
+
+	var session types.SessionTrackerV1
+	if err := utils.FastUnmarshal(bytes, &session); err != nil {
+		return nil, trace.BadParameter(err.Error())
+	}
+
+	if err := session.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &session, nil
+}
+
+// MarshalSessionTracker marshals the Session resource to JSON.
+func MarshalSessionTracker(session types.SessionTracker) ([]byte, error) {
+	if err := session.CheckAndSetDefaults(); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	switch session := session.(type) {
+	case *types.SessionTrackerV1:
+		return utils.FastMarshal(session)
+	default:
+		return nil, trace.BadParameter("unrecognized session version %T", session)
+	}
+}

lib/srv/app/server.go

@@ -175,7 +175,7 @@ type Server struct {
 
 	proxyPort string
 
-	cache *sessionCache
+	cache *sessionChunkCache
 
 	awsSigner *appaws.SigningService
 
@@ -253,7 +253,7 @@ func New(ctx context.Context, c *Config) (*Server, error) {
 
 	// Create a new session cache, this holds sessions that can be used to
 	// forward requests.
-	s.cache, err = newSessionCache(s.closeContext, s.log)
+	s.cache, err = s.newSessionChunkCache()
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -513,6 +513,11 @@ func (s *Server) Close() error {
 		errs = append(errs, err)
 	}
 
+	// Close the session cache and its remaining sessions. Sessions
+	// use server.closeContext to complete cleanup, so we must wait
+	// for sessions to finish closing before closing the context.
+	s.cache.closeAllSessions()
+
 	// Signal to any blocking go routine that it should exit.
 	s.closeFunc()
 
@@ -682,7 +687,7 @@ func (s *Server) authorize(ctx context.Context, r *http.Request) (*tlsca.Identit
 // getSession returns a request session used to proxy the request to the
 // target application. Always checks if the session is valid first and if so,
 // will return a cached session, otherwise will create one.
-func (s *Server) getSession(ctx context.Context, identity *tlsca.Identity, app types.Application) (*session, error) {
+func (s *Server) getSession(ctx context.Context, identity *tlsca.Identity, app types.Application) (*sessionChunk, error) {
 	// If a cached forwarder exists, return it right away.
 	session, err := s.cache.get(identity.RouteToApp.SessionID)
 	if err == nil {
@@ -690,15 +695,7 @@ func (s *Server) getSession(ctx context.Context, identity *tlsca.Identity, app t
 	}
 
 	// Create a new session with a recorder and forwarder in it.
-	session, err = s.newSession(ctx, identity, app)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	// Put the session in the cache so the next request can use it for 5 minutes
-	// or the time until the certificate expires, whichever comes first.
-	ttl := utils.MinTTL(identity.Expires.Sub(s.c.Clock.Now()), 5*time.Minute)
-	err = s.cache.set(identity.RouteToApp.SessionID, session, ttl)
+	session, err = s.newSessionChunk(ctx, identity, app)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/srv/app/session.go

@@ -31,8 +31,10 @@ import (
 	"github.com/gravitational/teleport/lib/events"
 	"github.com/gravitational/teleport/lib/events/filesessions"
 	"github.com/gravitational/teleport/lib/services"
-	session_pkg "github.com/gravitational/teleport/lib/session"
+	rsession "github.com/gravitational/teleport/lib/session"
+	"github.com/gravitational/teleport/lib/srv"
 	"github.com/gravitational/teleport/lib/tlsca"
+	"github.com/gravitational/teleport/lib/utils"
 
 	"github.com/gravitational/oxy/forward"
 	"github.com/gravitational/trace"
@@ -42,18 +44,42 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-// session holds a request forwarder and audit log for this chunk.
-type session struct {
+// sessionChunk holds an open request forwarder and audit log for an app session.
+//
+// An app session is only bounded by the lifetime of the certificate in
+// the caller's identity, so we create sessionChunks to track and record
+// chunks of live app session activity.
+//
+// Each chunk will emit an "app.session.chunk" event with the chunk ID
+// corresponding to the session chunk's uploaded recording. These emitted
+// chunk IDs can be used to aggregate all session uploads tied to the
+// overarching identity SessionID.
+type sessionChunk struct {
+	closeC chan struct{}
+	// id is the session chunk's uuid, which is used as the id of its session upload.
+	id string
 	// fwd can rewrite and forward requests to the target application.
 	fwd *forward.Forwarder
 	// streamWriter can emit events to the audit log.
 	streamWriter events.StreamWriter
 }
 
-// newSession creates a new session.
-func (s *Server) newSession(ctx context.Context, identity *tlsca.Identity, app types.Application) (*session, error) {
+// newSessionChunk creates a new chunk session.
+func (s *Server) newSessionChunk(ctx context.Context, identity *tlsca.Identity, app types.Application) (*sessionChunk, error) {
+	sess := &sessionChunk{
+		id:     uuid.New().String(),
+		closeC: make(chan struct{}),
+	}
+
+	// Create a session tracker so that other services, such as the
+	// session upload completer, can track the session chunk's lifetime.
+	if err := s.createTracker(sess, identity); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
 	// Create the stream writer that will write this chunk to the audit log.
-	streamWriter, err := s.newStreamWriter(identity, app)
+	var err error
+	sess.streamWriter, err = s.newStreamWriter(identity, app, sess.id)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -79,7 +105,7 @@ func (s *Server) newSession(ctx context.Context, identity *tlsca.Identity, app t
 	// Create a rewriting transport that will be used to forward requests.
 	transport, err := newTransport(s.closeContext,
 		&transportConfig{
-			w:            streamWriter,
+			w:            sess.streamWriter,
 			app:          app,
 			publicPort:   s.proxyPort,
 			cipherSuites: s.c.CipherSuites,
@@ -91,7 +117,8 @@ func (s *Server) newSession(ctx context.Context, identity *tlsca.Identity, app t
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	fwd, err := forward.New(
+
+	sess.fwd, err = forward.New(
 		forward.FlushInterval(100*time.Millisecond),
 		forward.RoundTripper(transport),
 		forward.Logger(logrus.StandardLogger()),
@@ -102,15 +129,33 @@ func (s *Server) newSession(ctx context.Context, identity *tlsca.Identity, app t
 		return nil, trace.Wrap(err)
 	}
 
-	return &session{
-		fwd:          fwd,
-		streamWriter: streamWriter,
-	}, nil
+	// Put the session chunk in the cache so that upcoming requests can use it for
+	// 5 minutes or the time until the certificate expires, whichever comes first.
+	ttl := utils.MinTTL(identity.Expires.Sub(s.c.Clock.Now()), 5*time.Minute)
+	err = s.cache.set(identity.RouteToApp.SessionID, sess, ttl)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return sess, nil
 }
 
-// newStreamWriter creates a streamer that will be used to stream the
-// requests that occur within this session to the audit log.
-func (s *Server) newStreamWriter(identity *tlsca.Identity, app types.Application) (events.StreamWriter, error) {
+func (s *sessionChunk) close(ctx context.Context) error {
+	close(s.closeC)
+	err := s.streamWriter.Close(ctx)
+	return trace.Wrap(err)
+}
+
+func (s *Server) closeSession(sess *sessionChunk) {
+	if err := sess.close(s.closeContext); err != nil {
+		s.log.WithError(err).Debugf("Error closing session %v", sess.id)
+	}
+}
+
+// newStreamWriter creates a session stream that will be used to record
+// requests that occur within this session chunk and upload the recording
+// to the Auth server.
+func (s *Server) newStreamWriter(identity *tlsca.Identity, app types.Application, chunkID string) (events.StreamWriter, error) {
 	recConfig, err := s.c.AccessPoint.GetSessionRecordingConfig(s.closeContext)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -121,11 +166,6 @@ func (s *Server) newStreamWriter(identity *tlsca.Identity, app types.Application
 		return nil, trace.Wrap(err)
 	}
 
-	// Each chunk has its own ID. Create a new UUID for this chunk which will be
-	// emitted in a new event to the audit log that can be use to aggregate all
-	// chunks for a particular session.
-	chunkID := uuid.New().String()
-
 	// Create a sync or async streamer depending on configuration of cluster.
 	streamer, err := s.newStreamer(s.closeContext, chunkID, recConfig)
 	if err != nil {
@@ -137,7 +177,7 @@ func (s *Server) newStreamWriter(identity *tlsca.Identity, app types.Application
 		Context:      s.closeContext,
 		Streamer:     streamer,
 		Clock:        s.c.Clock,
-		SessionID:    session_pkg.ID(chunkID),
+		SessionID:    rsession.ID(chunkID),
 		Namespace:    apidefaults.Namespace,
 		ServerID:     s.c.HostID,
 		RecordOutput: recConfig.GetMode() != types.RecordOff,
@@ -182,13 +222,13 @@ func (s *Server) newStreamWriter(identity *tlsca.Identity, app types.Application
 // of the server and the session, sync streamer sends the events
 // directly to the auth server and blocks if the events can not be received,
 // async streamer buffers the events to disk and uploads the events later
-func (s *Server) newStreamer(ctx context.Context, sessionID string, recConfig types.SessionRecordingConfig) (events.Streamer, error) {
+func (s *Server) newStreamer(ctx context.Context, chunkID string, recConfig types.SessionRecordingConfig) (events.Streamer, error) {
 	if services.IsRecordSync(recConfig.GetMode()) {
-		s.log.Debugf("Using sync streamer for session %v.", sessionID)
+		s.log.Debugf("Using sync streamer for session chunk %v.", chunkID)
 		return s.c.AuthClient, nil
 	}
 
-	s.log.Debugf("Using async streamer for session %v.", sessionID)
+	s.log.Debugf("Using async streamer for session chunk %v.", chunkID)
 	uploadDir := filepath.Join(
 		s.c.DataDir, teleport.LogsDir, teleport.ComponentUpload,
 		events.StreamingLogsDir, apidefaults.Namespace,
@@ -200,43 +240,72 @@ func (s *Server) newStreamer(ctx context.Context, sessionID string, recConfig ty
 	return fileStreamer, nil
 }
 
-// sessionCache holds a cache of sessions that are used to forward requests.
-type sessionCache struct {
-	mu    sync.Mutex
-	cache *ttlmap.TTLMap
-
-	closeContext context.Context
-
-	log *logrus.Entry
-}
-
-// newSessionCache creates a new session cache.
-func newSessionCache(ctx context.Context, log *logrus.Entry) (*sessionCache, error) {
-	var err error
-
-	s := &sessionCache{
-		closeContext: ctx,
-		log:          log,
+// createTracker creates a new session tracker for the session chunk.
+func (s *Server) createTracker(sess *sessionChunk, identity *tlsca.Identity) error {
+	trackerSpec := types.SessionTrackerSpecV1{
+		SessionID:   sess.id,
+		Kind:        string(types.AppSessionKind),
+		State:       types.SessionState_SessionStateRunning,
+		Hostname:    s.c.HostID,
+		AppName:     identity.RouteToApp.Name,
+		ClusterName: identity.RouteToApp.ClusterName,
+		Login:       identity.GetUserMetadata().Login,
+		Participants: []types.Participant{{
+			User: identity.Username,
+		}},
+		HostUser:     identity.Username,
+		Created:      s.c.Clock.Now(),
+		AppSessionID: identity.RouteToApp.SessionID,
 	}
 
-	// Cache of request forwarders. Set an expire function that can be used to
-	// close and upload the stream of events to the Audit Log.
-	s.cache, err = ttlmap.New(defaults.ClientCacheSize, ttlmap.CallOnExpire(s.expire))
+	s.log.Debugf("Creating tracker for session chunk %v", sess.id)
+	tracker, err := srv.NewSessionTracker(s.closeContext, trackerSpec, s.c.AuthClient)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	go func() {
+		<-sess.closeC
+		if err := tracker.Close(s.closeContext); err != nil {
+			s.log.WithError(err).Debugf("Failed to close session tracker for session chunk %v", sess.id)
+		}
+	}()
+
+	return nil
+}
+
+// sessionChunkCache holds a cache of session chunks.
+type sessionChunkCache struct {
+	srv *Server
+
+	mu    sync.Mutex
+	cache *ttlmap.TTLMap
+}
+
+// newSessionChunkCache creates a new session chunk cache.
+func (s *Server) newSessionChunkCache() (*sessionChunkCache, error) {
+	sessionCache := &sessionChunkCache{srv: s}
+
+	// Cache of session chunks. Set an expire function that can be used
+	// to close and upload the stream of events to the Audit Log.
+	var err error
+	sessionCache.cache, err = ttlmap.New(defaults.ClientCacheSize, ttlmap.CallOnExpire(sessionCache.expire), ttlmap.Clock(s.c.Clock))
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
-	go s.expireSessions()
 
-	return s, nil
+	go sessionCache.expireSessions()
+
+	return sessionCache, nil
 }
 
-// get will fetch the session from the cache.
-func (s *sessionCache) get(key string) (*session, error) {
+// get will fetch the session chunk from the cache.
+func (s *sessionChunkCache) get(key string) (*sessionChunk, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if f, ok := s.cache.Get(key); ok {
-		if fwd, fok := f.(*session); fok {
+		if fwd, fok := f.(*sessionChunk); fok {
 			return fwd, nil
 		}
 		return nil, trace.BadParameter("invalid type stored in cache: %T", f)
@@ -244,43 +313,37 @@ func (s *sessionCache) get(key string) (*session, error) {
 	return nil, trace.NotFound("session not found")
 }
 
-// set will add the session to the cache.
-func (s *sessionCache) set(key string, value *session, ttl time.Duration) error {
+// set will add the session chunk to the cache.
+func (s *sessionChunkCache) set(sessionID string, sess *sessionChunk, ttl time.Duration) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	if err := s.cache.Set(key, value, ttl); err != nil {
+	if err := s.cache.Set(sessionID, sess, ttl); err != nil {
 		return trace.Wrap(err)
 	}
 	return nil
 }
 
 // expire will close the stream writer.
-func (s *sessionCache) expire(key string, el interface{}) {
-	session, ok := el.(*session)
-	if !ok {
-		s.log.Debugf("Invalid type stored in cache: %T.", el)
-		return
-	}
-
-	// Closing the stream writer may trigger a flush operation which could be
-	// time-consuming. Launch in another goroutine since this occurs under a
+func (s *sessionChunkCache) expire(key string, el interface{}) {
+	// Closing the session stream writer may trigger a flush operation which could
+	// be time-consuming. Launch in another goroutine since this occurs under a
 	// lock and expire can get called during a "get" operation on the ttlmap.
-	go s.closeStreamWriter(s.closeContext, session)
-
-	s.log.Debugf("Closing expired stream %v.", key)
+	go s.closeSession(el)
+	s.srv.log.Debugf("Closing expired stream %v.", key)
 }
 
-// closeStreamWriter will close the stream writer. This could be a
-// time-consuming operation.
-func (s *sessionCache) closeStreamWriter(ctx context.Context, session *session) {
-	if err := session.streamWriter.Close(ctx); err != nil {
-		s.log.Debugf("Failed to close stream writer: %v.", err)
+func (s *sessionChunkCache) closeSession(el interface{}) {
+	switch sess := el.(type) {
+	case *sessionChunk:
+		s.srv.closeSession(sess)
+	default:
+		s.srv.log.Debugf("Invalid type stored in cache: %T.", el)
 	}
 }
 
 // expireSessions ticks every second trying to close expired sessions.
-func (s *sessionCache) expireSessions() {
+func (s *sessionChunkCache) expireSessions() {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 
@@ -288,16 +351,26 @@ func (s *sessionCache) expireSessions() {
 		select {
 		case <-ticker.C:
 			s.expiredSessions()
-		case <-s.closeContext.Done():
+		case <-s.srv.closeContext.Done():
 			return
 		}
 	}
 }
 
 // expiredSession tries to expire sessions in the cache.
-func (s *sessionCache) expiredSessions() {
+func (s *sessionChunkCache) expiredSessions() {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	s.cache.RemoveExpired(10)
 }
+
+// closeAllSessions will remove and close all sessions in the cache.
+func (s *sessionChunkCache) closeAllSessions() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, session, ok := s.cache.Pop(); ok; _, session, ok = s.cache.Pop() {
+		s.closeSession(session)
+	}
+}

lib/srv/db/access_test.go

@@ -1694,7 +1694,7 @@ func setupTestContext(ctx context.Context, t *testing.T, withDatabases ...withDa
 
 	// Create and start test auth server.
 	authServer, err := auth.NewTestAuthServer(auth.TestAuthServerConfig{
-		Clock:       clockwork.NewFakeClockAt(time.Now()),
+		Clock:       testCtx.clock,
 		ClusterName: testCtx.clusterName,
 		Dir:         t.TempDir(),
 	})
@@ -1881,7 +1881,7 @@ func (c *testContext) setupDatabaseServer(ctx context.Context, t *testing.T, p a
 
 	// Create database server agent itself.
 	server, err := New(ctx, Config{
-		Clock:            clockwork.NewFakeClockAt(time.Now()),
+		Clock:            c.clock,
 		DataDir:          t.TempDir(),
 		AuthClient:       c.authClient,
 		AccessPoint:      c.authClient,

lib/srv/db/server.go

@@ -680,15 +680,24 @@ func (s *Server) handleConnection(ctx context.Context, clientConn net.Conn) erro
 		return trace.Wrap(err)
 	}
 
+	// Create a session tracker so that other services, such as
+	// the session upload completer, can track the session's lifetime.
+	cancelCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// Create a session tracker so that other services, such as
+	// the session upload completer, can track the session's lifetime.
+	if err := s.trackSession(cancelCtx, sessionCtx); err != nil {
+		return trace.Wrap(err)
+	}
+
 	streamWriter, err := s.newStreamWriter(sessionCtx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
 	defer func() {
-		// Closing the stream writer is needed to flush all recorded data
-		// and trigger upload. Do it in a goroutine since depending on
-		// session size it can take a while, and we don't want to block
-		// the client.
+		// Close session stream in a goroutine since depending on session size
+		// it can take a while, and we don't want to block the client.
 		go func() {
 			// Use the server closing context to make sure that upload
 			// continues beyond the session lifetime.
@@ -866,3 +875,45 @@ func fetchMySQLVersion(ctx context.Context, database types.Database) error {
 
 	return nil
 }
+
+// trackSession creates a new session tracker for the database session.
+// While ctx is open, the session tracker's expiration will be extended
+// on an interval. Once the ctx is closed, the session tracker's state
+// will be updated to terminated.
+func (s *Server) trackSession(ctx context.Context, sessionCtx *common.Session) error {
+	trackerSpec := types.SessionTrackerSpecV1{
+		SessionID:    sessionCtx.ID,
+		Kind:         string(types.DatabaseSessionKind),
+		State:        types.SessionState_SessionStateRunning,
+		Hostname:     sessionCtx.HostID,
+		DatabaseName: sessionCtx.DatabaseName,
+		ClusterName:  sessionCtx.ClusterName,
+		Login:        sessionCtx.Identity.GetUserMetadata().Login,
+		Participants: []types.Participant{{
+			User: sessionCtx.Identity.Username,
+		}},
+		HostUser: sessionCtx.Identity.Username,
+		Created:  s.cfg.Clock.Now(),
+	}
+
+	s.log.Debugf("Creating tracker for session %v", sessionCtx.ID)
+	tracker, err := srv.NewSessionTracker(s.closeContext, trackerSpec, s.cfg.AuthClient)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	go func() {
+		if err := tracker.UpdateExpirationLoop(ctx, s.cfg.Clock); err != nil {
+			s.log.WithError(err).Debugf("Failed to update session tracker expiration for session %v", sessionCtx.ID)
+		}
+	}()
+
+	go func() {
+		<-ctx.Done()
+		if err := tracker.Close(s.closeContext); err != nil {
+			s.log.WithError(err).Debugf("Failed to close session tracker for session %v", sessionCtx.ID)
+		}
+	}()
+
+	return nil
+}

lib/srv/desktop/windows_server.go

@@ -777,23 +777,6 @@ func (s *WindowsService) connectRDP(ctx context.Context, log logrus.FieldLogger,
 		log.Infof("desktop session %v will not be recorded, user %v's roles disable recording", string(sessionID), authCtx.User.GetName())
 	}
 
-	sw, err := s.newStreamWriter(recordSession, string(sessionID))
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	// Closing the stream writer is needed to flush all recorded data
-	// and trigger the upload. Do it in a goroutine since depending on
-	// the session size it can take a while, and we don't want to block
-	// the client.
-	defer func() {
-		go func() {
-			if err := sw.Close(context.Background()); err != nil {
-				log.WithError(err).Errorf("closing stream writer for desktop session %v", sessionID.String())
-			}
-		}()
-	}()
-
 	var windowsUser string
 	authorize := func(login string) error {
 		windowsUser = login // capture attempted login user
@@ -814,6 +797,29 @@ func (s *WindowsService) connectRDP(ctx context.Context, log logrus.FieldLogger,
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	// Create a session tracker so that other services, such as
+	// the session upload completer, can track the session's lifetime.
+	if err := s.trackSession(ctx, &identity, windowsUser, string(sessionID), desktop); err != nil {
+		return trace.Wrap(err)
+	}
+
+	sw, err := s.newStreamWriter(recordSession, string(sessionID))
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	// Closing the stream writer is needed to flush all recorded data
+	// and trigger the upload. Do it in a goroutine since depending on
+	// the session size it can take a while, and we don't want to block
+	// the client.
+	defer func() {
+		go func() {
+			if err := sw.Close(context.Background()); err != nil {
+				log.WithError(err).Errorf("closing stream writer for desktop session %v", sessionID.String())
+			}
+		}()
+	}()
+
 	delay := timer()
 	tdpConn.OnSend = s.makeTDPSendHandler(ctx, sw, delay, &identity, string(sessionID), desktop.GetAddr())
 	tdpConn.OnRecv = s.makeTDPReceiveHandler(ctx, sw, delay, &identity, string(sessionID), desktop.GetAddr())
@@ -1252,6 +1258,49 @@ func (s *WindowsService) generateCredentials(ctx context.Context, username, doma
 	return certDER, keyDER, nil
 }
 
+// trackSession creates a session tracker for the given sessionID and
+// attributes, and starts a goroutine to continually extend the tracker
+// expiration while the session is active. Once the given ctx is closed,
+// the tracker will be marked as terminated.
+func (s *WindowsService) trackSession(ctx context.Context, id *tlsca.Identity, windowsUser string, sessionID string, desktop types.WindowsDesktop) error {
+	trackerSpec := types.SessionTrackerSpecV1{
+		SessionID:   sessionID,
+		Kind:        string(types.WindowsDesktopSessionKind),
+		State:       types.SessionState_SessionStateRunning,
+		Hostname:    s.cfg.Hostname,
+		Address:     desktop.GetAddr(),
+		DesktopName: desktop.GetName(),
+		ClusterName: s.clusterName,
+		Login:       windowsUser,
+		Participants: []types.Participant{{
+			User: id.Username,
+		}},
+		HostUser: id.Username,
+		Created:  s.cfg.Clock.Now(),
+	}
+
+	s.cfg.Log.Debugf("Creating tracker for session %v", sessionID)
+	tracker, err := srv.NewSessionTracker(ctx, trackerSpec, s.cfg.AuthClient)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	go func() {
+		if err := tracker.UpdateExpirationLoop(ctx, s.cfg.Clock); err != nil {
+			s.cfg.Log.WithError(err).Debugf("Failed to update session tracker expiration for session %v", sessionID)
+		}
+	}()
+
+	go func() {
+		<-ctx.Done()
+		if err := tracker.Close(s.closeCtx); err != nil {
+			s.cfg.Log.WithError(err).Debugf("Failed to close session tracker for session %v", sessionID)
+		}
+	}()
+
+	return nil
+}
+
 // The following vars contain the various object identifiers required for smartcard
 // login certificates.
 //

lib/srv/sess.go

@@ -25,12 +25,7 @@ import (
 	"sync"
 	"time"
 
-	"golang.org/x/crypto/ssh"
-
-	"github.com/google/uuid"
 	"github.com/gravitational/teleport"
-	"github.com/gravitational/teleport/api/client/proto"
-	apidefaults "github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
 	apievents "github.com/gravitational/teleport/api/types/events"
 	"github.com/gravitational/teleport/lib/auth"
@@ -42,13 +37,14 @@ import (
 	rsession "github.com/gravitational/teleport/lib/session"
 	"github.com/gravitational/teleport/lib/sshutils"
 	"github.com/gravitational/teleport/lib/utils"
-	"github.com/gravitational/trace/trail"
+
+	"github.com/google/uuid"
+	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
 	"github.com/moby/term"
-
-	"github.com/gravitational/trace"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/crypto/ssh"
 )
 
 const sessionRecorderID = "session-recorder"
@@ -419,11 +415,9 @@ type session struct {
 	// serverCtx is used to control clean up of internal resources
 	serverCtx context.Context
 
-	state types.SessionState
-
 	access auth.SessionAccessEvaluator
 
-	stateUpdate *sync.Cond
+	tracker *SessionTracker
 
 	initiator string
 
@@ -431,8 +425,6 @@ type session struct {
 
 	presenceEnabled bool
 
-	started bool
-
 	doneCh chan struct{}
 
 	bpfContext *bpf.SessionContext
@@ -524,12 +516,10 @@ func newSession(id rsession.ID, r *SessionRegistry, ctx *ServerContext) (*sessio
 		stopC:                          make(chan struct{}),
 		startTime:                      startTime,
 		serverCtx:                      ctx.srv.Context(),
-		state:                          types.SessionState_SessionStatePending,
 		access:                         auth.NewSessionAccessEvaluator(policySets, types.SSHSessionKind),
 		scx:                            ctx,
 		presenceEnabled:                ctx.Identity.Certificate.Extensions[teleport.CertExtensionMFAVerified] != "",
 		io:                             NewTermManager(),
-		stateUpdate:                    sync.NewCond(&sync.Mutex{}),
 		doneCh:                         make(chan struct{}),
 		initiator:                      ctx.Identity.TeleportUser,
 		displayParticipantRequirements: utils.AsBool(ctx.env[teleport.EnvSSHSessionDisplayParticipantRequirements]),
@@ -552,12 +542,10 @@ func newSession(id rsession.ID, r *SessionRegistry, ctx *ServerContext) (*sessio
 		}
 	}()
 
-	err = sess.trackerCreate(ctx.Identity.TeleportUser, policySets)
-	if err != nil {
+	if err = sess.trackSession(ctx.Identity.TeleportUser, policySets); err != nil {
 		if trace.IsNotImplemented(err) {
 			return nil, trace.NotImplemented("Attempted to use Moderated Sessions with an Auth Server below the minimum version of 9.0.0.")
 		}
-
 		return nil, trace.Wrap(err)
 	}
 
@@ -608,24 +596,16 @@ func (s *session) Stop() {
 	// Close and kill terminal
 	if s.term != nil {
 		if err := s.term.Close(); err != nil {
-			s.log.Debugf("Failed to close the shell: %v", err)
+			s.log.WithError(err).Debug("Failed to close the shell")
 		}
 		if err := s.term.Kill(); err != nil {
-			s.log.Debugf("Failed to kill the shell: %v", err)
+			s.log.WithError(err).Debug("Failed to kill the shell")
 		}
 	}
 
-	// Remove session parties and close client connections.
-	for _, p := range s.parties {
-		p.closeUnderSessionLock()
-	}
-
-	// Set session state to terminated
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-	err := s.trackerUpdateState(types.SessionState_SessionStateTerminated)
-	if err != nil {
-		s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStateTerminated)
+	// Close session tracker and mark it as terminated
+	if err := s.tracker.Close(s.serverCtx); err != nil {
+		s.log.WithError(err).Debug("Failed to close session tracker")
 	}
 }
 
@@ -641,6 +621,11 @@ func (s *session) Close() error {
 
 	serverSessions.Dec()
 
+	// Remove session parties and close client connections.
+	for _, p := range s.getParties() {
+		p.Close()
+	}
+
 	// Remove session from registry
 	s.registry.removeSession(s)
 
@@ -664,31 +649,6 @@ func (s *session) Close() error {
 	return nil
 }
 
-func (s *session) waitOnAccess() error {
-	s.io.Off()
-	s.BroadcastMessage("Session paused, Waiting for required participants...")
-
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-outer:
-	for {
-		switch s.state {
-		case types.SessionState_SessionStatePending:
-			continue
-		case types.SessionState_SessionStateTerminated:
-			return nil
-		case types.SessionState_SessionStateRunning:
-			break outer
-		}
-
-		s.stateUpdate.Wait()
-	}
-
-	s.BroadcastMessage("Resuming session...")
-	s.io.On()
-	return nil
-}
-
 func (s *session) BroadcastMessage(format string, args ...interface{}) {
 	if s.access.IsModerated() && !services.IsRecordAtProxy(s.scx.SessionRecordingConfig.GetMode()) {
 		s.io.BroadcastMessage(fmt.Sprintf(format, args...))
@@ -826,7 +786,10 @@ func (s *session) emitSessionLeaveEvent(ctx *ServerContext) {
 		}
 		_, _, err = p.sconn.SendRequest(teleport.SessionEvent, false, eventPayload)
 		if err != nil {
-			s.log.Warnf("Unable to send %v to %v: %v.", events.SessionLeaveEvent, p.sconn.RemoteAddr(), err)
+			// The party's connection may already be closed, in which case we expect an EOF
+			if !trace.IsEOF(err) {
+				s.log.Warnf("Unable to send %v to %v: %v.", events.SessionLeaveEvent, p.sconn.RemoteAddr(), err)
+			}
 			continue
 		}
 		s.log.Debugf("Sent %v to %v.", events.SessionLeaveEvent, p.sconn.RemoteAddr())
@@ -889,20 +852,15 @@ func (s *session) setEndingContext(ctx *ServerContext) {
 	s.endingContext = ctx
 }
 
+// launch launches the session.
+// Must be called under session Lock.
 func (s *session) launch(ctx *ServerContext) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-
 	s.log.Debugf("Launching session %v.", s.id)
 	s.BroadcastMessage("Connecting to %v over SSH", ctx.srv.GetInfo().GetHostname())
 
 	s.io.On()
 
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-
-	err := s.trackerUpdateState(types.SessionState_SessionStateRunning)
-	if err != nil {
+	if err := s.tracker.UpdateState(s.serverCtx, types.SessionState_SessionStateRunning); err != nil {
 		s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStateRunning)
 	}
 
@@ -1246,43 +1204,48 @@ func (s *session) removePartyUnderLock(p *party) error {
 	// Remove participant from in-memory map of party members.
 	delete(s.parties, p.id)
 
+	s.BroadcastMessage("User %v left the session.", p.user)
+
 	// Update session tracker
-	if err := s.trackerRemoveParticipant(p.user); err != nil {
+	s.log.Debugf("No longer tracking participant: %v", p.id)
+	if err := s.tracker.RemoveParticipant(s.serverCtx, p.id.String()); err != nil {
 		return trace.Wrap(err)
 	}
 
 	// Remove party for the term writer
 	s.io.DeleteWriter(string(p.id))
 
+	// Emit session leave event to both the Audit Log as well as over the
+	// "x-teleport-event" channel in the SSH connection.
+	s.emitSessionLeaveEvent(p.ctx)
+
 	canRun, policyOptions, err := s.checkIfStart()
 	if err != nil {
 		return trace.Wrap(err)
 	}
 
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-
-	if !canRun && s.state == types.SessionState_SessionStateRunning {
+	if !canRun {
 		if policyOptions.TerminateOnLeave {
 			// Force termination in goroutine to avoid deadlock
 			go s.registry.ForceTerminate(s.scx)
 			return nil
 		}
 
-		err := s.trackerUpdateState(types.SessionState_SessionStatePending)
-		if err != nil {
+		// pause session and wait for another party to resume
+		s.io.Off()
+		s.BroadcastMessage("Session paused, Waiting for required participants...")
+		if err := s.tracker.UpdateState(s.serverCtx, types.SessionState_SessionStatePending); err != nil {
 			s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStatePending)
 		}
 
-		go s.waitOnAccess()
+		go func() {
+			if state := s.tracker.WaitForStateUpdate(types.SessionState_SessionStatePending); state == types.SessionState_SessionStateRunning {
+				s.BroadcastMessage("Resuming session...")
+				s.io.On()
+			}
+		}()
 	}
 
-	s.BroadcastMessage("User %v left the session.", p.user)
-
-	// Emit session leave event to both the Audit Log as well as over the
-	// "x-teleport-event" channel in the SSH connection.
-	s.emitSessionLeaveEvent(p.ctx)
-
 	// If the leaving party was the last one in the session, start the lingerAndDie
 	// goroutine. Parties that join during the linger duration will cancel the
 	// goroutine to prevent the session from ending with active parties.
@@ -1404,12 +1367,7 @@ func (s *session) checkPresence() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	sess, err := s.trackerGet()
-	if err != nil {
-		return trace.Wrap(err)
-	}
-
-	for _, participant := range sess.GetParticipants() {
+	for _, participant := range s.tracker.GetParticipants() {
 		if participant.User == s.initiator {
 			continue
 		}
@@ -1454,6 +1412,16 @@ func (s *session) addParty(p *party, mode types.SessionParticipantMode) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	if s.login != p.login {
+		return trace.AccessDenied(
+			"can't switch users from %v to %v for session %v",
+			s.login, p.login, s.id)
+	}
+
+	if s.tracker.GetState() == types.SessionState_SessionStateTerminated {
+		return trace.AccessDenied("The requested session is not active")
+	}
+
 	if len(s.parties) == 0 {
 		canStart, _, err := s.checkIfStart()
 		if err != nil {
@@ -1477,8 +1445,14 @@ func (s *session) addParty(p *party, mode types.SessionParticipantMode) error {
 	s.participants[p.id] = p
 	p.ctx.AddCloser(p)
 
-	err := s.trackerAddParticipant(p)
-	if err != nil {
+	s.log.Debugf("Tracking participant: %s", p.id)
+	participant := &types.Participant{
+		ID:         p.id.String(),
+		User:       p.user,
+		Mode:       string(p.mode),
+		LastActive: time.Now().UTC(),
+	}
+	if err := s.tracker.AddParticipant(s.serverCtx, participant); err != nil {
 		return trace.Wrap(err)
 	}
 
@@ -1504,38 +1478,24 @@ func (s *session) addParty(p *party, mode types.SessionParticipantMode) error {
 		}()
 	}
 
-	s.stateUpdate.L.Lock()
-	defer s.stateUpdate.L.Unlock()
-	if s.state == types.SessionState_SessionStatePending {
+	if s.tracker.GetState() == types.SessionState_SessionStatePending {
 		canStart, _, err := s.checkIfStart()
 		if err != nil {
 			return trace.Wrap(err)
 		}
 
 		if canStart {
-			if !s.started {
-				s.started = true
-
-				go func() {
-					err := s.launch(s.scx)
-					if err != nil {
-						s.log.Errorf("Failed to launch session %v: %v", s.id, err)
-					}
-				}()
-			} else {
-				err := s.trackerUpdateState(types.SessionState_SessionStateRunning)
-				if err != nil {
-					s.log.Warnf("Failed to set tracker state to %v", types.SessionState_SessionStateRunning)
-				}
+			if err := s.launch(s.scx); err != nil {
+				s.log.Errorf("Failed to launch session %v: %v", s.id, err)
 			}
-		} else if !s.started {
-			base := "Waiting for required participants..."
+			return nil
+		}
 
-			if s.displayParticipantRequirements {
-				s.BroadcastMessage(base+"\r\n%v", s.access.PrettyRequirementsList())
-			} else {
-				s.BroadcastMessage(base)
-			}
+		base := "Waiting for required participants..."
+		if s.displayParticipantRequirements {
+			s.BroadcastMessage(base+"\r\n%v", s.access.PrettyRequirementsList())
+		} else {
+			s.BroadcastMessage(base)
 		}
 	}
 
@@ -1669,135 +1629,47 @@ func (p *party) closeUnderSessionLock() {
 	})
 }
 
-func (s *session) trackerGet() (types.SessionTracker, error) {
-	// get the session from the registry
-	sess, err := s.registry.SessionTrackerService.GetSessionTracker(s.serverCtx, s.id.String())
-	if err != nil {
-		return nil, trace.Wrap(err)
+// trackSession creates a new session tracker for the ssh session.
+// While ctx is open, the session tracker's expiration will be extended
+// on an interval until the session tracker is closed.
+func (s *session) trackSession(teleportUser string, policySet []*types.SessionTrackerPolicySet) error {
+	trackerSpec := types.SessionTrackerSpecV1{
+		SessionID:   s.id.String(),
+		Kind:        string(types.SSHSessionKind),
+		State:       types.SessionState_SessionStatePending,
+		Hostname:    s.registry.Srv.GetInfo().GetHostname(),
+		Address:     s.scx.srv.ID(),
+		ClusterName: s.scx.ClusterName,
+		Login:       s.login,
+		Participants: []types.Participant{{
+			ID:         teleportUser,
+			User:       teleportUser,
+			LastActive: s.registry.clock.Now(),
+		}},
+		HostUser:     teleportUser,
+		Reason:       s.scx.env[teleport.EnvSSHSessionReason],
+		HostPolicies: policySet,
+		Created:      s.registry.clock.Now(),
 	}
 
-	return sess, nil
-}
-
-func (s *session) trackerCreate(teleportUser string, policySet []*types.SessionTrackerPolicySet) error {
-	s.log.Debug("Creating tracker")
-	initator := &types.Participant{
-		ID:         teleportUser,
-		User:       teleportUser,
-		LastActive: time.Now().UTC(),
-	}
-
-	reason := s.scx.env[teleport.EnvSSHSessionReason]
-	var invited []string
-
 	if s.scx.env[teleport.EnvSSHSessionInvited] != "" {
-		err := json.Unmarshal([]byte(s.scx.env[teleport.EnvSSHSessionInvited]), &invited)
-		if err != nil {
+		if err := json.Unmarshal([]byte(s.scx.env[teleport.EnvSSHSessionInvited]), &trackerSpec.Invited); err != nil {
 			return trace.Wrap(err)
 		}
 	}
 
-	req := &proto.CreateSessionTrackerRequest{
-		ID:           s.id.String(),
-		Namespace:    apidefaults.Namespace,
-		Type:         string(types.KubernetesSessionKind),
-		Hostname:     s.registry.Srv.GetInfo().GetHostname(),
-		Address:      s.scx.srv.ID(),
-		ClusterName:  s.scx.ClusterName,
-		Login:        "root",
-		Initiator:    initator,
-		HostUser:     initator.User,
-		Reason:       reason,
-		Invited:      invited,
-		HostPolicies: policySet,
-	}
-
-	_, err := s.registry.SessionTrackerService.CreateSessionTracker(s.serverCtx, req)
+	s.log.Debug("Creating session tracker")
+	var err error
+	s.tracker, err = NewSessionTracker(s.serverCtx, trackerSpec, s.registry.SessionTrackerService)
 	if err != nil {
-		return trail.FromGRPC(err)
+		return trace.Wrap(err)
 	}
 
-	// Start go routine to push back session expiration while session is still active.
 	go func() {
-		ticker := s.registry.clock.NewTicker(defaults.SessionTrackerExpirationUpdateInterval)
-		defer ticker.Stop()
-		for {
-			select {
-			case time := <-ticker.Chan():
-				if err := s.trackerUpdateExpiry(time.Add(defaults.SessionTrackerTTL)); err != nil {
-					s.log.WithError(err).Warningf("Failed to update session tracker expiration.")
-				}
-			case <-s.stopC:
-				return
-			}
+		if err := s.tracker.UpdateExpirationLoop(s.serverCtx, s.registry.clock); err != nil {
+			s.log.WithError(err).Debug("Failed to update session tracker expiration")
 		}
 	}()
 
 	return nil
 }
-
-func (s *session) trackerAddParticipant(participant *party) error {
-	s.log.Debugf("Tracking participant: %v", participant.user)
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_AddParticipant{
-			AddParticipant: &proto.SessionTrackerAddParticipant{
-				Participant: &types.Participant{
-					ID:         participant.user,
-					User:       participant.user,
-					Mode:       string(participant.mode),
-					LastActive: time.Now().UTC(),
-				},
-			},
-		},
-	}
-
-	err := s.registry.SessionTrackerService.UpdateSessionTracker(s.serverCtx, req)
-	return trace.Wrap(err)
-}
-
-func (s *session) trackerRemoveParticipant(participantID string) error {
-	s.log.Debugf("Not tracking participant: %v", participantID)
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_RemoveParticipant{
-			RemoveParticipant: &proto.SessionTrackerRemoveParticipant{
-				ParticipantID: participantID,
-			},
-		},
-	}
-
-	err := s.registry.SessionTrackerService.UpdateSessionTracker(s.serverCtx, req)
-	return trace.Wrap(err)
-}
-
-func (s *session) trackerUpdateState(state types.SessionState) error {
-	s.state = state
-	s.stateUpdate.Broadcast()
-
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_UpdateState{
-			UpdateState: &proto.SessionTrackerUpdateState{
-				State: state,
-			},
-		},
-	}
-
-	err := s.registry.SessionTrackerService.UpdateSessionTracker(s.serverCtx, req)
-	return trace.Wrap(err)
-}
-
-func (s *session) trackerUpdateExpiry(expires time.Time) error {
-	req := &proto.UpdateSessionTrackerRequest{
-		SessionID: s.id.String(),
-		Update: &proto.UpdateSessionTrackerRequest_UpdateExpiry{
-			UpdateExpiry: &proto.SessionTrackerUpdateExpiry{
-				Expires: &expires,
-			},
-		},
-	}
-
-	err := s.registry.SessionTrackerService.UpdateSessionTracker(s.serverCtx, req)
-	return trace.Wrap(err)
-}

lib/srv/sess_test.go

@@ -368,58 +368,6 @@ func testJoinSession(t *testing.T, reg *SessionRegistry, sess *session) {
 	require.NoError(t, err)
 }
 
-// TestSessionTracker tests session tracker lifecycle
-func TestSessionTracker(t *testing.T) {
-	ctx := context.Background()
-
-	srv := newMockServer(t)
-
-	// Use a separate clock from srv so we can use BlockUntil.
-	regClock := clockwork.NewFakeClock()
-	reg, err := NewSessionRegistry(SessionRegistryConfig{
-		Srv:                   srv,
-		SessionTrackerService: srv.auth,
-		clock:                 regClock,
-	})
-
-	require.NoError(t, err)
-	t.Cleanup(func() { reg.Close() })
-
-	// Session tracker should be created for a new session
-	sess := testOpenSession(t, reg)
-	tracker, err := srv.auth.GetSessionTracker(ctx, sess.ID())
-	require.NoError(t, err)
-
-	// Session tracker's expiration should be updated on an interval
-	// while the session is active.
-	regClock.BlockUntil(1)
-	regClock.Advance(defaults.SessionTrackerExpirationUpdateInterval)
-	srv.clock.Advance(defaults.SessionTrackerExpirationUpdateInterval)
-
-	trackerUpdated := func() bool {
-		updatedTracker, err := srv.auth.GetSessionTracker(ctx, sess.ID())
-		require.NoError(t, err)
-		return updatedTracker.Expiry().Equal(tracker.Expiry().Add(defaults.SessionTrackerExpirationUpdateInterval))
-	}
-	require.Eventually(t, trackerUpdated, time.Second*5, time.Millisecond*500)
-
-	// Once the sesssion is closed and the last set
-	// expiration is up, the tracker should be deleted.
-	sess.Close()
-	regClock.Advance(defaults.SessionTrackerTTL)
-	srv.clock.Advance(defaults.SessionTrackerTTL)
-
-	trackerDeleted := func() bool {
-		_, err := srv.auth.GetSessionTracker(ctx, sess.ID())
-		if err == nil {
-			return false
-		}
-		require.True(t, trace.IsNotFound(err))
-		return true
-	}
-	require.Eventually(t, trackerDeleted, time.Second*5, time.Millisecond*500)
-}
-
 func testOpenSession(t *testing.T, reg *SessionRegistry) *session {
 	scx := newTestServerContext(t, reg.Srv)
 

lib/srv/sessiontracker.go

@@ -0,0 +1,194 @@
+/*
+Copyright 2022 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package srv
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/gravitational/teleport/api/client/proto"
+	apidefaults "github.com/gravitational/teleport/api/defaults"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/services"
+	"github.com/jonboulle/clockwork"
+
+	"github.com/gravitational/trace"
+)
+
+// SessionTracker is a session tracker for a specific session. It tracks
+// the session in memory and broadcasts updates to the given service (backend).
+type SessionTracker struct {
+	closeC chan struct{}
+	// tracker is the in memory session tracker
+	tracker types.SessionTracker
+	// trackerCond is used to provide synchronized access to tracker
+	// and to broadcast state changes.
+	trackerCond *sync.Cond
+	// service is used to share session tracker updates with the service
+	service services.SessionTrackerService
+}
+
+// NewSessionTracker returns a new SessionTracker for the given types.SessionTracker
+func NewSessionTracker(ctx context.Context, trackerSpec types.SessionTrackerSpecV1, service services.SessionTrackerService) (*SessionTracker, error) {
+	if service == nil {
+		return nil, trace.BadParameter("missing parameter service")
+	}
+
+	t, err := types.NewSessionTracker(trackerSpec)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	if t, err = service.CreateSessionTracker(ctx, t); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &SessionTracker{
+		service:     service,
+		tracker:     t,
+		trackerCond: sync.NewCond(&sync.Mutex{}),
+		closeC:      make(chan struct{}),
+	}, nil
+}
+
+// Close closes the session tracker and sets the tracker state to terminated
+func (s *SessionTracker) Close(ctx context.Context) error {
+	close(s.closeC)
+	err := s.UpdateState(ctx, types.SessionState_SessionStateTerminated)
+	return trace.Wrap(err)
+}
+
+const sessionTrackerExpirationUpdateInterval = apidefaults.SessionTrackerTTL / 6
+
+// UpdateExpirationLoop extends the session tracker expiration by 1 hour every 10 minutes
+// until the SessionTracker or ctx is closed.
+func (s *SessionTracker) UpdateExpirationLoop(ctx context.Context, clock clockwork.Clock) error {
+	ticker := clock.NewTicker(sessionTrackerExpirationUpdateInterval)
+	defer ticker.Stop()
+	return s.updateExpirationLoop(ctx, ticker)
+}
+
+// updateExpirationLoop is used in tests
+func (s *SessionTracker) updateExpirationLoop(ctx context.Context, ticker clockwork.Ticker) error {
+	for {
+		select {
+		case time := <-ticker.Chan():
+			expiry := time.Add(apidefaults.SessionTrackerTTL)
+			if err := s.UpdateExpiration(ctx, expiry); err != nil {
+				return trace.Wrap(err)
+			}
+		case <-ctx.Done():
+			return trace.Wrap(ctx.Err())
+		case <-s.closeC:
+			return nil
+		}
+	}
+}
+
+func (s *SessionTracker) UpdateExpiration(ctx context.Context, expiry time.Time) error {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	s.tracker.SetExpiry(expiry)
+	s.trackerCond.Broadcast()
+
+	err := s.service.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
+		SessionID: s.tracker.GetSessionID(),
+		Update: &proto.UpdateSessionTrackerRequest_UpdateExpiry{
+			UpdateExpiry: &proto.SessionTrackerUpdateExpiry{
+				Expires: &expiry,
+			},
+		},
+	})
+	return trace.Wrap(err)
+}
+
+func (s *SessionTracker) AddParticipant(ctx context.Context, p *types.Participant) error {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	s.tracker.AddParticipant(*p)
+	s.trackerCond.Broadcast()
+
+	err := s.service.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
+		SessionID: s.tracker.GetSessionID(),
+		Update: &proto.UpdateSessionTrackerRequest_AddParticipant{
+			AddParticipant: &proto.SessionTrackerAddParticipant{
+				Participant: p,
+			},
+		},
+	})
+	return trace.Wrap(err)
+}
+
+func (s *SessionTracker) RemoveParticipant(ctx context.Context, participantID string) error {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	s.tracker.RemoveParticipant(participantID)
+	s.trackerCond.Broadcast()
+
+	err := s.service.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
+		SessionID: s.tracker.GetSessionID(),
+		Update: &proto.UpdateSessionTrackerRequest_RemoveParticipant{
+			RemoveParticipant: &proto.SessionTrackerRemoveParticipant{
+				ParticipantID: participantID,
+			},
+		},
+	})
+	return trace.Wrap(err)
+}
+
+func (s *SessionTracker) UpdateState(ctx context.Context, state types.SessionState) error {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	s.tracker.SetState(state)
+	s.trackerCond.Broadcast()
+
+	err := s.service.UpdateSessionTracker(ctx, &proto.UpdateSessionTrackerRequest{
+		SessionID: s.tracker.GetSessionID(),
+		Update: &proto.UpdateSessionTrackerRequest_UpdateState{
+			UpdateState: &proto.SessionTrackerUpdateState{
+				State: state,
+			},
+		},
+	})
+	return trace.Wrap(err)
+}
+
+// WaitForStateUpdate waits for the tracker's state to be updated and returns the new state.
+func (s *SessionTracker) WaitForStateUpdate(initialState types.SessionState) types.SessionState {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+
+	for {
+		if state := s.tracker.GetState(); state != initialState {
+			return state
+		}
+		s.trackerCond.Wait()
+	}
+}
+
+func (s *SessionTracker) GetState() types.SessionState {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	return s.tracker.GetState()
+}
+
+func (s *SessionTracker) GetParticipants() []types.Participant {
+	s.trackerCond.L.Lock()
+	defer s.trackerCond.L.Unlock()
+	return s.tracker.GetParticipants()
+}

lib/srv/sessiontracker_test.go

@@ -0,0 +1,155 @@
+/*
+Copyright 2022 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package srv
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/gravitational/teleport/api/client/proto"
+	"github.com/gravitational/teleport/api/types"
+
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSessionTracker(t *testing.T) {
+	ctx := context.Background()
+	clock := clockwork.NewFakeClock()
+
+	mockService := &mockSessiontrackerService{
+		trackers: make(map[string]types.SessionTracker),
+	}
+
+	sessID := "sessionID"
+	trackerSpec := types.SessionTrackerSpecV1{
+		Created:   clock.Now(),
+		SessionID: sessID,
+	}
+
+	// Create a new session tracker
+	tracker, err := NewSessionTracker(ctx, trackerSpec, mockService)
+	require.NoError(t, err)
+	require.NotNil(t, tracker)
+	require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+
+	t.Run("UpdateExpirationLoop", func(t *testing.T) {
+		cancelCtx, cancel := context.WithCancel(ctx)
+		done := make(chan struct{})
+
+		duration := time.Minute
+		ticker := clock.NewTicker(duration)
+		defer ticker.Stop()
+
+		// Start update expiration goroutine
+		go func() {
+			tracker.updateExpirationLoop(cancelCtx, ticker)
+			close(done)
+		}()
+
+		// lock expiry and advance clock
+		tracker.trackerCond.L.Lock()
+		clock.Advance(duration)
+		expectedExpiry := tracker.tracker.Expiry().Add(duration)
+
+		// wait for expiration to get updated
+		tracker.trackerCond.Wait()
+		tracker.trackerCond.L.Unlock()
+		require.Equal(t, expectedExpiry, tracker.tracker.Expiry())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+
+		// cancelling the goroutine's ctx should halt the update loop
+		cancel()
+		_, ok := <-done
+		require.False(t, ok)
+	})
+
+	t.Run("State", func(t *testing.T) {
+		stateUpdate := make(chan types.SessionState)
+		go func() {
+			stateUpdate <- tracker.WaitForStateUpdate(types.SessionState_SessionStatePending)
+		}()
+
+		err = tracker.UpdateState(ctx, types.SessionState_SessionStatePending)
+		require.NoError(t, err)
+		require.Equal(t, types.SessionState_SessionStatePending, tracker.GetState())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+
+		err = tracker.UpdateState(ctx, types.SessionState_SessionStateRunning)
+		require.NoError(t, err)
+		require.Equal(t, types.SessionState_SessionStateRunning, tracker.GetState())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+
+		// WaitForStateUpdate should ignore the pending update and then catch the running update
+		require.Equal(t, types.SessionState_SessionStateRunning, <-stateUpdate)
+	})
+
+	t.Run("Participants", func(t *testing.T) {
+		participantID := "userID"
+
+		p := &types.Participant{ID: participantID}
+		err = tracker.AddParticipant(ctx, p)
+		require.NoError(t, err)
+		require.Equal(t, []types.Participant{*p}, tracker.GetParticipants())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+
+		err = tracker.RemoveParticipant(ctx, participantID)
+		require.NoError(t, err)
+		require.Empty(t, tracker.GetParticipants())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+	})
+
+	t.Run("Close", func(t *testing.T) {
+		// Closing the tracker should update the state to terminated
+		err = tracker.Close(ctx)
+		require.NoError(t, err)
+		require.Equal(t, types.SessionState_SessionStateTerminated, tracker.GetState())
+		require.Equal(t, tracker.tracker, mockService.trackers[sessID])
+	})
+}
+
+type mockSessiontrackerService struct {
+	trackers map[string]types.SessionTracker
+}
+
+func (m *mockSessiontrackerService) GetActiveSessionTrackers(ctx context.Context) ([]types.SessionTracker, error) {
+	return nil, trace.NotImplemented("")
+}
+
+func (m *mockSessiontrackerService) GetSessionTracker(ctx context.Context, sessionID string) (types.SessionTracker, error) {
+	return nil, trace.NotImplemented("")
+}
+
+func (m *mockSessiontrackerService) UpdateSessionTracker(ctx context.Context, req *proto.UpdateSessionTrackerRequest) error {
+	// m.trackers[req.SessionID] will be updated as a pointer reference
+	return nil
+}
+
+func (m *mockSessiontrackerService) RemoveSessionTracker(ctx context.Context, sessionID string) error {
+	return trace.NotImplemented("")
+}
+
+func (m *mockSessiontrackerService) UpdatePresence(ctx context.Context, sessionID, user string) error {
+	return trace.NotImplemented("")
+}
+
+func (m *mockSessiontrackerService) CreateSessionTracker(ctx context.Context, tracker types.SessionTracker) (types.SessionTracker, error) {
+	m.trackers[tracker.GetSessionID()] = tracker
+	return tracker, nil
+}

tool/tsh/tsh_test.go

@@ -1945,7 +1945,8 @@ func TestSerializeKubeSessions(t *testing.T) {
     "kind": "session_tracker",
     "version": "v1",
     "metadata": {
-      "name": "id"
+      "name": "id",
+      "expires": "1970-01-01T00:00:00Z"
     },
     "spec": {
       "session_id": "id",