fix: session trackers participants inconsistencies

api/proto/teleport/legacy/types/types.proto

@@ -6636,6 +6636,9 @@ message Participant {
     (gogoproto.nullable) = false,
     (gogoproto.jsontag) = "last_active,omitempty"
   ];
+
+  // Cluster is the cluster name the user is authenticated against.
+  string Cluster = 5 [(gogoproto.jsontag) = "cluster,omitempty"];
 }
 
 // UIConfigV1 represents the configuration for the web UI served by the proxy service

api/types/session_tracker.go

@@ -105,7 +105,7 @@ type SessionTracker interface {
 	RemoveParticipant(string) error
 
 	// UpdatePresence updates presence timestamp of a participant.
-	UpdatePresence(string, time.Time) error
+	UpdatePresence(username string, cluster string, t time.Time) error
 
 	// GetKubeCluster returns the name of the kubernetes cluster the session is running in.
 	GetKubeCluster() string
@@ -320,9 +320,14 @@ func (s *SessionTrackerV1) GetHostUser() string {
 }
 
 // UpdatePresence updates presence timestamp of a participant.
-func (s *SessionTrackerV1) UpdatePresence(user string, t time.Time) error {
+func (s *SessionTrackerV1) UpdatePresence(user, userCluster string, t time.Time) error {
 	idx := slices.IndexFunc(s.Spec.Participants, func(participant Participant) bool {
-		return participant.User == user
+		// participant.Cluster == "" is a legacy participant that was created
+		// before cluster field was added, so we allow updating presence for
+		// such participants as well.
+		// TODO(tigrato): Remove this in version 20.0.0
+		// TODO(tigrato): DELETE IN 20.0.0
+		return participant.User == user && (participant.Cluster == userCluster || participant.Cluster == "")
 	})
 
 	if idx < 0 {

api/types/session_tracker_test.go

@@ -34,12 +34,20 @@ func TestSessionTrackerV1_UpdatePresence(t *testing.T) {
 			{
 				ID:         "1",
 				User:       "llama",
+				Cluster:    "teleport-local",
 				Mode:       string(SessionPeerMode),
 				LastActive: now,
 			},
 			{
 				ID:         "2",
 				User:       "fish",
+				Cluster:    "teleport-remote",
+				Mode:       string(SessionModeratorMode),
+				LastActive: now,
+			},
+			{
+				ID:         "3",
+				User:       "cat",
 				Mode:       string(SessionModeratorMode),
 				LastActive: now,
 			},
@@ -48,11 +56,13 @@ func TestSessionTrackerV1_UpdatePresence(t *testing.T) {
 	require.NoError(t, err)
 
 	// Presence cannot be updated for a non-existent user
-	err = s.UpdatePresence("alpaca", now.Add(time.Hour))
+	err = s.UpdatePresence("alpaca", "", now.Add(time.Hour))
 	require.ErrorIs(t, err, trace.NotFound("participant alpaca not found"))
 
 	// Update presence for just the user fish
-	require.NoError(t, s.UpdatePresence("fish", now.Add(time.Hour)))
+	require.NoError(t, s.UpdatePresence("fish", "teleport-remote", now.Add(time.Hour)))
+	// Try to Update presence for user fish again, but with a different cluster.
+	require.Error(t, s.UpdatePresence("fish", "teleport-local", now.Add(time.Hour)))
 
 	// Verify that llama has not been active but that fish was
 	for _, participant := range s.GetParticipants() {

lib/auth/auth_with_roles.go

@@ -347,9 +347,10 @@ func (a *ServerWithRoles) filterSessionTracker(joinerRoles []types.Role, tracker
 		}
 
 		for _, participant := range tracker.GetParticipants() {
-			// We only need to fill in User here since other fields get discarded anyway.
+			// We only need to fill in User and cluster here since other fields get discarded anyway.
 			ruleCtx.SSHSession.Parties = append(ruleCtx.SSHSession.Parties, session.Party{
-				User: participant.User,
+				User:    participant.User,
+				Cluster: participant.Cluster,
 			})
 		}
 

lib/auth/authclient/clt.go

@@ -576,7 +576,7 @@ func (c *Client) DeleteClusterAuditConfig(ctx context.Context) error {
 	return trace.NotImplemented(notImplementedMessage)
 }
 
-func (c *Client) UpdatePresence(ctx context.Context, sessionID, user string) error {
+func (c *Client) UpdatePresence(ctx context.Context, _, _, _ string) error {
 	return trace.NotImplemented(notImplementedMessage)
 }
 

lib/auth/grpcserver.go

@@ -2480,12 +2480,28 @@ func doMFAPresenceChallenge(ctx context.Context, actx *grpcContext, stream authp
 		return trace.Wrap(err)
 	}
 
-	err = actx.authServer.UpdatePresence(ctx, challengeReq.SessionID, user)
-	if err != nil {
-		return trace.Wrap(err)
+	// We must use the real username associated with the identity.
+	// GetUser() can be remote-{user}-{cluster} if we are using
+	// a leaf cluster.
+	realUsername := actx.Identity.GetIdentity().Username
+	originCluster := actx.Identity.GetIdentity().OriginClusterName
+	origErr := actx.authServer.UpdatePresence(ctx, challengeReq.SessionID, realUsername, originCluster)
+	switch {
+	case trace.IsNotFound(origErr):
+		// If the user was not found, it may be because the session tracker
+		// was created with the .GetUser() value (remote-{user}-{cluster}).
+		// Try again with that username as a fallback.
+		// TODO(tigrato): DELETE IN 20.0.0
+		if fallbackErr := actx.authServer.UpdatePresence(ctx, challengeReq.SessionID, user, originCluster); fallbackErr != nil {
+			// Return the original error, as that is more relevant to the client.
+			return trace.Wrap(origErr)
+		}
+		return nil
+	case origErr != nil:
+		return trace.Wrap(origErr)
+	default:
+		return nil
 	}
-
-	return nil
 }
 
 // MaintainSessionPresence establishes a channel used to continuously verify the presence for a session.

lib/events/complete_test.go

@@ -409,7 +409,7 @@ func (m *mockSessionTrackerService) RemoveSessionTracker(ctx context.Context, se
 	return trace.NotImplemented("RemoveSessionTracker is not implemented")
 }
 
-func (m *mockSessionTrackerService) UpdatePresence(ctx context.Context, sessionID, user string) error {
+func (m *mockSessionTrackerService) UpdatePresence(_ context.Context, _, _, _ string) error {
 	return trace.NotImplemented("UpdatePresence is not implemented")
 }
 

lib/kube/proxy/sess.go

@@ -993,7 +993,8 @@ func (s *session) join(p *party, emitJoinEvent bool) error {
 	s.log.DebugContext(s.forwarder.ctx, "Tracking participant", "participant_id", p.ID)
 	participant := &types.Participant{
 		ID:         p.ID.String(),
-		User:       p.Ctx.User.GetName(),
+		User:       p.Ctx.Identity.GetIdentity().Username,
+		Cluster:    p.Ctx.Identity.GetIdentity().OriginClusterName,
 		Mode:       string(p.Mode),
 		LastActive: time.Now().UTC(),
 	}

lib/services/local/sessiontracker.go

@@ -62,7 +62,7 @@ func (s *sessionTracker) loadSession(ctx context.Context, sessionID string) (typ
 }
 
 // UpdatePresence updates the presence status of a user in a session.
-func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user string) error {
+func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user, userTeleportCluster string) error {
 	for range updateRetryLimit {
 		sessionItem, err := s.bk.Get(ctx, backend.NewKey(sessionPrefix, sessionID))
 		if err != nil {
@@ -74,7 +74,7 @@ func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user str
 			return trace.Wrap(err)
 		}
 
-		if err := session.UpdatePresence(user, s.bk.Clock().Now().UTC()); err != nil {
+		if err := session.UpdatePresence(user, userTeleportCluster, s.bk.Clock().Now().UTC()); err != nil {
 			return trace.Wrap(err)
 		}
 

lib/services/parser.go

@@ -566,7 +566,14 @@ func toCtxTracker(t types.SessionTracker) ctxTracker {
 		participants := s.GetParticipants()
 		names := make([]string, len(participants))
 		for i, participant := range participants {
-			names[i] = participant.User
+			// Participant for RBAC must be represented as `remote-{user}-{cluster}`.
+			// if they belong to a different cluster. This is because the user
+			// is also named like that when they authenticate.
+			names[i] = UsernameForCluster(UsernameForClusterConfig{
+				User:              participant.User,
+				OriginClusterName: participant.Cluster,
+				LocalClusterName:  s.GetClusterName(),
+			})
 		}
 
 		return names

lib/services/parser_test.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
 	"github.com/gravitational/trace"
 	"github.com/stretchr/testify/require"
 	"github.com/vulcand/predicate"
@@ -824,3 +825,98 @@ func TestCanView(t *testing.T) {
 		})
 	}
 }
+
+func TestSessionTrackerRBAC(t *testing.T) {
+	t.Parallel()
+	localCluster := "local-cluster"
+	remoteCluster := "remote-cluster"
+	tracker, err := types.NewSessionTracker(types.SessionTrackerSpecV1{
+		SessionID:   uuid.NewString(),
+		Kind:        types.KindSSHSession,
+		Hostname:    "hostname",
+		ClusterName: localCluster,
+		Login:       "root",
+		Participants: []types.Participant{
+			{
+				ID:      uuid.New().String(),
+				User:    "eve",
+				Cluster: localCluster,
+				Mode:    string(types.SessionPeerMode),
+			},
+			{
+				ID:      uuid.New().String(),
+				User:    "alice",
+				Cluster: remoteCluster,
+				Mode:    string(types.SessionObserverMode),
+			},
+			{
+				ID:      uuid.New().String(),
+				User:    "bob",
+				Cluster: localCluster,
+				Mode:    string(types.SessionObserverMode),
+			},
+		},
+		Expires: time.Now().UTC().Add(24 * time.Hour),
+	})
+	require.NoError(t, err)
+
+	testCases := []struct {
+		name        string
+		username    string
+		userCluster string
+		checkAccess require.BoolAssertionFunc
+	}{
+		{
+			name:        "same user and same cluster as peer participant",
+			username:    "eve",
+			userCluster: localCluster,
+			checkAccess: require.True,
+		},
+		{
+			name:        "same user but different cluster as peer participant",
+			username:    "eve",
+			userCluster: remoteCluster,
+			checkAccess: require.False,
+		},
+		{
+			name:        "different user but same cluster as peer participant",
+			username:    "alice",
+			userCluster: localCluster,
+			checkAccess: require.False,
+		},
+		{
+			name:        "observer participant, same cluster",
+			username:    "bob",
+			userCluster: localCluster,
+			checkAccess: require.True,
+		},
+		{
+			name:        "not a participant, same cluster",
+			username:    "mallory",
+			userCluster: localCluster,
+			checkAccess: require.False,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			user, err := types.NewUser(UsernameForCluster(UsernameForClusterConfig{
+				User:              tc.username,
+				LocalClusterName:  localCluster,
+				OriginClusterName: tc.userCluster,
+			}))
+			require.NoError(t, err)
+			ctx := &Context{
+				User:           user,
+				SessionTracker: tracker,
+			}
+			parser, err := NewWhereParser(ctx)
+			require.NoError(t, err)
+			expr, err := parser.Parse("contains(session_tracker.participants, user.metadata.name)")
+			require.NoError(t, err)
+			pred, ok := expr.(predicate.BoolPredicate)
+			require.True(t, ok)
+			tc.checkAccess(t, pred())
+		})
+	}
+}

lib/services/sessiontracker.go

@@ -50,7 +50,7 @@ type SessionTrackerService interface {
 	RemoveSessionTracker(ctx context.Context, sessionID string) error
 
 	// UpdatePresence updates the presence status of a user in a session.
-	UpdatePresence(ctx context.Context, sessionID, user string) error
+	UpdatePresence(ctx context.Context, sessionID, user, userCluster string) error
 }
 
 // UnmarshalSessionTracker unmarshals the Session resource from JSON.

lib/services/user.go

@@ -143,3 +143,26 @@ func MarshalUser(user types.User, opts ...MarshalOption) ([]byte, error) {
 func UsernameForRemoteCluster(localUsername, localClusterName string) string {
 	return fmt.Sprintf("remote-%v-%v", localUsername, localClusterName)
 }
+
+// UsernameForClusterConfig is a configuration struct for UsernameForCluster.
+type UsernameForClusterConfig struct {
+	// User is the username.
+	User string
+	// OriginClusterName is the cluster name where the user is authenticated.
+	OriginClusterName string
+	// LocalClusterName is the local cluster name.
+	LocalClusterName string
+}
+
+// UsernameForCluster returns an username that is prefixed with "remote-"
+// and suffixed with cluster name if the user is from a remote cluster,
+// otherwise returns the local username.
+func UsernameForCluster(cfg UsernameForClusterConfig) string {
+	// originClusterName == "" is a special case for backward compatibility
+	// with older clients that do not send origin cluster name.
+	// In this case we assume the user is local.
+	if cfg.OriginClusterName == cfg.LocalClusterName || cfg.OriginClusterName == "" {
+		return cfg.User
+	}
+	return UsernameForRemoteCluster(cfg.User, cfg.OriginClusterName)
+}

lib/services/user_test.go

@@ -369,3 +369,53 @@ func claimsToAttributes(claims map[string]any) saml2.AssertionInfo {
 	}
 	return info
 }
+
+func TestUsernameForCluster(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		username         string
+		originCluster    string
+		localClusterName string
+		expected         string
+	}{
+
+		{
+			username:         "alice",
+			originCluster:    "leaf",
+			localClusterName: "root",
+			expected:         "remote-alice-leaf",
+		},
+		{
+			username:         "bob",
+			originCluster:    "",
+			localClusterName: "root",
+			expected:         "bob",
+		},
+		{
+			username:         "carol",
+			originCluster:    "leaf.cluster",
+			localClusterName: "root.cluster",
+			expected:         "remote-carol-leaf.cluster",
+		},
+		{
+			username:         "dave",
+			originCluster:    "leaf-cluster",
+			localClusterName: "leaf-cluster",
+			expected:         "dave",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.username, func(t *testing.T) {
+			result := UsernameForCluster(
+				UsernameForClusterConfig{
+					User:              test.username,
+					OriginClusterName: test.originCluster,
+					LocalClusterName:  test.localClusterName,
+				},
+			)
+			require.Equal(t, test.expected, result)
+		})
+	}
+}

lib/session/session.go

@@ -170,6 +170,8 @@ type Party struct {
 	RemoteAddr string `json:"remote_addr"`
 	// User is a teleport user using this session
 	User string `json:"user"`
+	// Cluster is the cluster name the user is authenticated against.
+	Cluster string `json:"cluster"`
 	// ServerID is an address of the server
 	ServerID string `json:"server_id"`
 	// LastActive is a last time this party was active

lib/srv/app/session.go

@@ -343,7 +343,8 @@ func (c *ConnectionsHandler) createTracker(sess *sessionChunk, identity *tlsca.I
 		ClusterName: identity.RouteToApp.ClusterName,
 		Login:       identity.GetUserMetadata().Login,
 		Participants: []types.Participant{{
-			User: identity.Username,
+			User:    identity.Username,
+			Cluster: identity.OriginClusterName,
 		}},
 		HostUser:     identity.Username,
 		Created:      c.cfg.Clock.Now(),