Fix moderated session presence checking (#25988)

* Fix moderated session presence checking

Addresses all of the issues that were preventing presence checking during
moderated sessions from working as described in
[#18092](https://github.com/gravitational/teleport/issues/18092#issuecomment-1540900859).

Closes #18092

* make presence test clearer

* fix presence checking on the web ui

* Refactor web socket message handling

A single message processing loop handles retrieving the envelope
and passing it off to individual message handlers. This allows all
messages to be processed outside of `Read` which was dependent on
the terminal being active to process any messages.

The webauthn challenge response was also moved from a raw message
to a webauthn message. By sending it as a raw message it made
presence checking fail because the response has a `t` in it which
caused the session to be killed during moderated sessions.

* enforce mfa ceremony when joining and cluster wide mfa is enabled

* fix conflicts with master

* moderated tests

* Add moderated session tests for the UI

* Add moderated session integration test

* fix lints

* clean up presence test

* refactor envelope handling

* fix build

* fix: revert test debug timeout

* fix: use local context in tests

* simplify closing streams

* generalize waitForOutput to work with an io.Reader

* fix error handling in stream close

* unexport PresenceOptions

* Improve waitForOutput to match against output in successive reads

api/types/session_tracker.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/gravitational/trace"
+	"golang.org/x/exp/slices"
 
 	"github.com/gravitational/teleport/api/defaults"
 )
@@ -94,7 +95,7 @@ type SessionTracker interface {
 	RemoveParticipant(string) error
 
 	// UpdatePresence updates presence timestamp of a participant.
-	UpdatePresence(string) error
+	UpdatePresence(string, time.Time) error
 
 	// GetKubeCluster returns the name of the kubernetes cluster the session is running in.
 	GetKubeCluster() string
@@ -302,15 +303,17 @@ func (s *SessionTrackerV1) GetHostUser() string {
 }
 
 // UpdatePresence updates presence timestamp of a participant.
-func (s *SessionTrackerV1) UpdatePresence(user string) error {
-	for _, participant := range s.Spec.Participants {
-		if participant.User == user {
-			participant.LastActive = time.Now().UTC()
-			return nil
-		}
+func (s *SessionTrackerV1) UpdatePresence(user string, t time.Time) error {
+	idx := slices.IndexFunc(s.Spec.Participants, func(participant Participant) bool {
+		return participant.User == user
+	})
+
+	if idx < 0 {
+		return trace.NotFound("participant %v not found", user)
 	}
 
-	return trace.NotFound("participant %v not found", user)
+	s.Spec.Participants[idx].LastActive = t
+	return nil
 }
 
 // GetHostPolicySets returns a list of policy sets held by the host user at the time of session creation.

api/types/session_tracker_test.go

@@ -0,0 +1,66 @@
+// Copyright 2023 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 types
+
+import (
+	"testing"
+	"time"
+
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestSessionTrackerV1_UpdatePresence(t *testing.T) {
+	clock := clockwork.NewFakeClock()
+	now := clock.Now().UTC()
+
+	s, err := NewSessionTracker(SessionTrackerSpecV1{
+		SessionID: "123",
+		Participants: []Participant{
+			{
+				ID:         "1",
+				User:       "llama",
+				Mode:       string(SessionPeerMode),
+				LastActive: now,
+			},
+			{
+				ID:         "2",
+				User:       "fish",
+				Mode:       string(SessionModeratorMode),
+				LastActive: now,
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	// Presence cannot be updated for a non-existent user
+	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)))
+
+	// Verify that llama has not been active but that fish was
+	for _, participant := range s.GetParticipants() {
+		lastActive := now
+		if participant.User == "fish" {
+			lastActive = lastActive.Add(time.Hour)
+		}
+
+		assert.Equal(t, lastActive, participant.LastActive)
+	}
+}

integration/helpers/instance.go

@@ -1243,6 +1243,8 @@ func (i *TeleInstance) Start() error {
 
 // ClientConfig is a client configuration
 type ClientConfig struct {
+	// TeleportUser is Teleport username
+	TeleportUser string
 	// Login is SSH login name
 	Login string
 	// Cluster is a cluster name to connect to
@@ -1322,8 +1324,12 @@ func (i *TeleInstance) NewUnauthenticatedClient(cfg ClientConfig) (tc *client.Te
 		fwdAgentMode = client.ForwardAgentYes
 	}
 
+	if cfg.TeleportUser == "" {
+		cfg.TeleportUser = cfg.Login
+	}
+
 	cconf := &client.Config{
-		Username:                      cfg.Login,
+		Username:                      cfg.TeleportUser,
 		Host:                          cfg.Host,
 		HostPort:                      cfg.Port,
 		HostLogin:                     cfg.Login,
@@ -1554,21 +1560,22 @@ func (w *WebClient) SSH(termReq web.TerminalRequest) (*web.TerminalStream, error
 		return nil, trace.Wrap(err)
 	}
 
-	stream, err := web.NewTerminalStream(ws)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
+	stream := web.NewTerminalStream(context.Background(), ws, utils.NewLoggerForTests())
 	return stream, nil
 }
 
 // AddClientCredentials adds authenticated credentials to a client.
 // (server CAs and signed session key).
 func (i *TeleInstance) AddClientCredentials(tc *client.TeleportClient, cfg ClientConfig) (*client.TeleportClient, error) {
+	login := cfg.Login
+	if cfg.TeleportUser != "" {
+		login = cfg.TeleportUser
+	}
+
 	// Generate certificates for the user simulating login.
 	creds, err := GenerateUserCreds(UserCredsRequest{
 		Process:  i.Process,
-		Username: cfg.Login,
+		Username: login,
 		SourceIP: cfg.SourceIP,
 	})
 	if err != nil {

integration/integration_test.go

@@ -72,8 +72,11 @@ import (
 	"github.com/gravitational/teleport/integration/helpers"
 	"github.com/gravitational/teleport/lib"
 	"github.com/gravitational/teleport/lib/auth"
+	"github.com/gravitational/teleport/lib/auth/mocku2f"
 	"github.com/gravitational/teleport/lib/auth/native"
 	"github.com/gravitational/teleport/lib/auth/testauthority"
+	wanlib "github.com/gravitational/teleport/lib/auth/webauthn"
+	wancli "github.com/gravitational/teleport/lib/auth/webauthncli"
 	"github.com/gravitational/teleport/lib/bpf"
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/cloud"
@@ -90,6 +93,7 @@ import (
 	"github.com/gravitational/teleport/lib/sshutils"
 	"github.com/gravitational/teleport/lib/tlsca"
 	"github.com/gravitational/teleport/lib/utils"
+	"github.com/gravitational/teleport/lib/utils/prompt"
 	"github.com/gravitational/teleport/lib/web"
 )
 
@@ -149,6 +153,7 @@ func TestIntegrations(t *testing.T) {
 	t.Run("JumpTrustedClustersWithLabels", suite.bind(testJumpTrustedClustersWithLabels))
 	t.Run("List", suite.bind(testList))
 	t.Run("MapRoles", suite.bind(testMapRoles))
+	t.Run("ModeratedSessions", suite.bind(testModeratedSessions))
 	t.Run("MultiplexingTrustedClusters", suite.bind(testMultiplexingTrustedClusters))
 	t.Run("PAM", suite.bind(testPAM))
 	t.Run("PortForwarding", suite.bind(testPortForwarding))
@@ -8152,3 +8157,216 @@ func TestConnectivityWithoutAuth(t *testing.T) {
 		})
 	}
 }
+
+func testModeratedSessions(t *testing.T, suite *integrationTestSuite) {
+	modules.SetTestModules(t, &modules.TestModules{TestBuildType: modules.BuildEnterprise})
+
+	const password = "supersecretpassword"
+	inputReader := prompt.NewFakeReader().
+		AddString(password).
+		AddReply(func(ctx context.Context) (string, error) {
+			panic("this should not be called")
+		})
+
+	oldStdin, oldWebauthn := prompt.Stdin(), *client.PromptWebauthn
+	t.Cleanup(func() {
+		prompt.SetStdin(oldStdin)
+		*client.PromptWebauthn = oldWebauthn
+	})
+
+	device, err := mocku2f.Create()
+	require.NoError(t, err)
+	device.SetPasswordless()
+
+	prompt.SetStdin(inputReader)
+	*client.PromptWebauthn = func(ctx context.Context, realOrigin string, assertion *wanlib.CredentialAssertion, prompt wancli.LoginPrompt, _ *wancli.LoginOpts) (*proto.MFAAuthenticateResponse, string, error) {
+		car, err := device.SignAssertion("https://127.0.0.1", assertion) // use the fake origin to prevent a mismatch
+		if err != nil {
+			return nil, "", err
+		}
+		return &proto.MFAAuthenticateResponse{
+			Response: &proto.MFAAuthenticateResponse_Webauthn{
+				Webauthn: wanlib.CredentialAssertionResponseToProto(car),
+			},
+		}, "", nil
+	}
+
+	// Enable web service.
+	cfg := suite.defaultServiceConfig()
+	cfg.Auth.Enabled = true
+	cfg.Auth.Preference.SetSecondFactor("on")
+	cfg.Auth.Preference.(*types.AuthPreferenceV2).Spec.RequireMFAType = types.RequireMFAType_SESSION
+	cfg.Auth.Preference.SetWebauthn(&types.Webauthn{RPID: "127.0.0.1"})
+	cfg.Proxy.DisableWebService = false
+	cfg.Proxy.DisableWebInterface = true
+	cfg.Proxy.Enabled = true
+	cfg.SSH.Enabled = true
+
+	instance := suite.NewTeleportWithConfig(t, nil, nil, cfg)
+
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+
+	peerRole, err := types.NewRole("moderated", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			RequireSessionJoin: []*types.SessionRequirePolicy{
+				{
+					Name:   "moderated",
+					Filter: "contains(user.roles, \"moderator\")",
+					Kinds:  []string{string(types.SSHSessionKind)},
+					Count:  1,
+					Modes:  []string{string(types.SessionModeratorMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, instance.Process.GetAuthServer().UpsertRole(ctx, peerRole))
+
+	moderatorRole, err := types.NewRole("moderator", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			JoinSessions: []*types.SessionJoinPolicy{
+				{
+					Name:  "moderated",
+					Roles: []string{peerRole.GetName()},
+					Kinds: []string{string(types.SSHSessionKind)},
+					Modes: []string{string(types.SessionModeratorMode), string(types.SessionObserverMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, instance.Process.GetAuthServer().UpsertRole(ctx, moderatorRole))
+
+	setupUser := func(user, role string, asrv *auth.Server) {
+		u, err := types.NewUser(user)
+		require.NoError(t, err)
+		u.SetRoles([]string{"access", role})
+		u.SetLogins([]string{suite.Me.Username, user})
+
+		require.NoError(t, asrv.CreateUser(ctx, u))
+
+		token, err := asrv.CreateResetPasswordToken(ctx, auth.CreateUserTokenRequest{
+			Name: user,
+		})
+		require.NoError(t, err)
+		tokenID := token.GetName()
+		res, err := asrv.CreateRegisterChallenge(ctx, &proto.CreateRegisterChallengeRequest{
+			TokenID:     tokenID,
+			DeviceType:  proto.DeviceType_DEVICE_TYPE_WEBAUTHN,
+			DeviceUsage: proto.DeviceUsage_DEVICE_USAGE_PASSWORDLESS,
+		})
+		require.NoError(t, err)
+		cc := wanlib.CredentialCreationFromProto(res.GetWebauthn())
+
+		ccr, err := device.SignCredentialCreation("https://127.0.0.1", cc)
+		require.NoError(t, err)
+		_, err = asrv.ChangeUserAuthentication(ctx, &proto.ChangeUserAuthenticationRequest{
+			TokenID:     tokenID,
+			NewPassword: []byte(password),
+			NewMFARegisterResponse: &proto.MFARegisterResponse{
+				Response: &proto.MFARegisterResponse_Webauthn{
+					Webauthn: wanlib.CredentialCreationResponseToProto(ccr),
+				},
+			},
+		})
+		require.NoError(t, err)
+	}
+
+	setupUser("peer", peerRole.GetName(), instance.Process.GetAuthServer())
+	setupUser("moderator", moderatorRole.GetName(), instance.Process.GetAuthServer())
+
+	peerTerminal := NewTerminal(250)
+	moderatorTerminal := NewTerminal(250)
+
+	// get a reference to site obj:
+	site := instance.GetSiteAPI(helpers.Site)
+	require.NotNil(t, site)
+
+	// PersonA: SSH into the server, wait one second, then type some commands on stdin:
+	openSession := func() {
+		cl, err := instance.NewClient(helpers.ClientConfig{
+			TeleportUser: "peer",
+			Login:        suite.Me.Username,
+			Cluster:      helpers.Site,
+			Host:         Host,
+		})
+		if err != nil {
+			cancel(trace.Wrap(err, "failed to create peer client"))
+			return
+		}
+
+		cl.Stdout = peerTerminal
+		cl.Stdin = peerTerminal
+		if err := cl.SSH(ctx, []string{}, false); err != nil {
+			cancel(trace.Wrap(err, "peer session failed"))
+			return
+		}
+	}
+
+	// PersonB: wait for a session to become available, then join:
+	joinSession := func() {
+		sessionTimeoutCtx, sessionTimeoutCancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer sessionTimeoutCancel()
+		sessions, err := waitForSessionToBeEstablished(sessionTimeoutCtx, defaults.Namespace, site)
+		if err != nil {
+			cancel(trace.Wrap(err, "failed waiting for peer session to be established"))
+			return
+		}
+
+		sessionID := sessions[0].GetSessionID()
+		cl, err := instance.NewClient(helpers.ClientConfig{
+			TeleportUser: "moderator",
+			Login:        suite.Me.Username,
+			Cluster:      helpers.Site,
+			Host:         Host,
+		})
+		if err != nil {
+			cancel(trace.Wrap(err, "failed to create peer client"))
+			return
+		}
+
+		cl.Stdout = moderatorTerminal
+		cl.Stdin = moderatorTerminal
+		if err := cl.Join(ctx, types.SessionModeratorMode, defaults.Namespace, session.ID(sessionID), moderatorTerminal); err != nil {
+			cancel(trace.Wrap(err, "moderator session failed"))
+		}
+	}
+
+	go openSession()
+	go joinSession()
+
+	require.Eventually(t, func() bool {
+		ss, err := site.GetActiveSessionTrackers(ctx)
+		if err != nil {
+			return false
+		}
+
+		if len(ss) != 1 {
+			return false
+		}
+
+		return len(ss[0].GetParticipants()) == 2
+	}, 5*time.Second, 100*time.Millisecond)
+
+	peerTerminal.Type("echo llamas\n\r")
+
+	require.Eventually(t, func() bool {
+		return strings.Contains(moderatorTerminal.AllOutput(), "llamas")
+	}, 5*time.Second, 100*time.Millisecond)
+
+	// terminate the session
+	moderatorTerminal.Type("t\n\r")
+
+	select {
+	case <-ctx.Done():
+		err = context.Cause(ctx)
+		require.Contains(t, err.Error(), "Process exited with status 255")
+		p := peerTerminal.AllOutput()
+		require.Contains(t, p, "Forcefully terminating session...")
+		m := moderatorTerminal.AllOutput()
+		require.Contains(t, m, "Forcefully terminating session...")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Timeout waiting for session to be terminated.")
+	}
+}

lib/client/api.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"encoding/pem"
 	"errors"
 	"fmt"
@@ -1779,7 +1778,7 @@ func (tc *TeleportClient) runShellOrCommandOnSingleNode(ctx context.Context, clt
 		// Reuse the existing nodeClient we connected above.
 		return nodeClient.RunCommand(ctx, command)
 	}
-	return tc.runShell(ctx, nodeClient, types.SessionPeerMode, nil, nil)
+	return trace.Wrap(nodeClient.RunInteractiveShell(ctx, types.SessionPeerMode, nil, nil))
 }
 
 func (tc *TeleportClient) runShellOrCommandOnMultipleNodes(ctx context.Context, clt *ClusterClient, nodeAddrs []string, command []string) error {
@@ -1902,7 +1901,12 @@ func (tc *TeleportClient) Join(ctx context.Context, mode types.SessionParticipan
 	if mode == types.SessionModeratorMode {
 		beforeStart = func(out io.Writer) {
 			nc.OnMFA = func() {
-				runPresenceTask(presenceCtx, out, clt.AuthClient, tc, session.GetSessionID())
+				RunPresenceTask(presenceCtx, out, clt.AuthClient, session.GetSessionID(), func(ctx context.Context, proxyAddr string, c *proto.MFAAuthenticateChallenge) (*proto.MFAAuthenticateResponse, error) {
+					resp, err := tc.PromptMFAChallenge(ctx, proxyAddr, c, func(opts *PromptMFAChallengeOpts) {
+						opts.Quiet = true
+					})
+					return resp, trace.Wrap(err)
+				})
 			}
 		}
 	}
@@ -1910,7 +1914,7 @@ func (tc *TeleportClient) Join(ctx context.Context, mode types.SessionParticipan
 	fmt.Printf("Joining session with participant mode: %v. \n\n", mode)
 
 	// running shell with a given session means "join" it:
-	err = tc.runShell(ctx, nc, mode, session, beforeStart)
+	err = nc.RunInteractiveShell(ctx, mode, session, beforeStart)
 	return trace.Wrap(err)
 }
 
@@ -2677,49 +2681,6 @@ func (tc *TeleportClient) newSessionEnv() map[string]string {
 	return env
 }
 
-// runShell starts an interactive SSH session/shell.
-// sessionID : when empty, creates a new shell. otherwise it tries to join the existing session.
-func (tc *TeleportClient) runShell(ctx context.Context, nodeClient *NodeClient, mode types.SessionParticipantMode, sessToJoin types.SessionTracker, beforeStart func(io.Writer)) error {
-	ctx, span := tc.Tracer.Start(
-		ctx,
-		"teleportClient/runShell",
-		oteltrace.WithSpanKind(oteltrace.SpanKindClient),
-	)
-	defer span.End()
-
-	env := tc.newSessionEnv()
-	env[teleport.EnvSSHJoinMode] = string(mode)
-	env[teleport.EnvSSHSessionReason] = tc.Config.Reason
-	env[teleport.EnvSSHSessionDisplayParticipantRequirements] = strconv.FormatBool(tc.Config.DisplayParticipantRequirements)
-
-	encoded, err := json.Marshal(&tc.Config.Invited)
-	if err != nil {
-		return trace.Wrap(err)
-	}
-	env[teleport.EnvSSHSessionInvited] = string(encoded)
-
-	nodeSession, err := newSession(ctx, nodeClient, sessToJoin, env, tc.Stdin, tc.Stdout, tc.Stderr, tc.EnableEscapeSequences)
-	if err != nil {
-		return trace.Wrap(err)
-	}
-	if err = nodeSession.runShell(ctx, mode, beforeStart, tc.OnShellCreated); err != nil {
-		switch e := trace.Unwrap(err).(type) {
-		case *ssh.ExitError:
-			tc.ExitStatus = e.ExitStatus()
-		case *ssh.ExitMissingError:
-			tc.ExitStatus = 1
-		}
-
-		return trace.Wrap(err)
-	}
-	if nodeSession.ExitMsg == "" {
-		fmt.Fprintln(tc.Stderr, "the connection was closed on the remote side on ", time.Now().Format(time.RFC822))
-	} else {
-		fmt.Fprintln(tc.Stderr, nodeSession.ExitMsg)
-	}
-	return nil
-}
-
 // getProxyLogin determines which SSH principal to use when connecting to proxy.
 func (tc *TeleportClient) getProxySSHPrincipal() string {
 	if tc.ProxySSHPrincipal != "" {

lib/client/client.go

@@ -1597,7 +1597,7 @@ func NewNodeClient(ctx context.Context, sshConfig *ssh.ClientConfig, conn net.Co
 // RunInteractiveShell creates an interactive shell on the node and copies stdin/stdout/stderr
 // to and from the node and local shell. This will block until the interactive shell on the node
 // is terminated.
-func (c *NodeClient) RunInteractiveShell(ctx context.Context, mode types.SessionParticipantMode, sessToJoin types.SessionTracker) error {
+func (c *NodeClient) RunInteractiveShell(ctx context.Context, mode types.SessionParticipantMode, sessToJoin types.SessionTracker, beforeStart func(io.Writer)) error {
 	ctx, span := c.Tracer.Start(
 		ctx,
 		"nodeClient/RunInteractiveShell",
@@ -1624,7 +1624,7 @@ func (c *NodeClient) RunInteractiveShell(ctx context.Context, mode types.Session
 		return trace.Wrap(err)
 	}
 
-	if err = nodeSession.runShell(ctx, mode, nil, c.TC.OnShellCreated); err != nil {
+	if err = nodeSession.runShell(ctx, mode, beforeStart, c.TC.OnShellCreated); err != nil {
 		switch e := trace.Unwrap(err).(type) {
 		case *ssh.ExitError:
 			c.TC.ExitStatus = e.ExitStatus()
@@ -1703,7 +1703,7 @@ func (c *NodeClient) handleGlobalRequests(ctx context.Context, requestCh <-chan
 					continue
 				}
 
-				c.OnMFA()
+				go c.OnMFA()
 			case teleport.SessionEvent:
 				// Parse event and create events.EventFields that can be consumed directly
 				// by caller.

lib/client/kubesession.go

@@ -23,13 +23,13 @@ import (
 	"fmt"
 	"io"
 	"sync"
-	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/gravitational/trace"
 	"k8s.io/client-go/tools/remotecommand"
 
 	"github.com/gravitational/teleport/api/client"
+	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/defaults"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/lib/client/terminal"
@@ -37,8 +37,6 @@ import (
 	"github.com/gravitational/teleport/lib/utils"
 )
 
-const mfaChallengeInterval = time.Second * 30
-
 // KubeSession a joined kubernetes session from the client side.
 type KubeSession struct {
 	stream *streamproto.SessionStream
@@ -197,7 +195,12 @@ func (s *KubeSession) handleMFA(ctx context.Context, tc *TeleportClient, mode ty
 			return trace.Wrap(err)
 		}
 
-		go runPresenceTask(ctx, stdout, auth, tc, s.meta.GetSessionID())
+		go RunPresenceTask(ctx, stdout, auth, s.meta.GetSessionID(), func(ctx context.Context, proxyAddr string, c *proto.MFAAuthenticateChallenge) (*proto.MFAAuthenticateResponse, error) {
+			resp, err := tc.PromptMFAChallenge(ctx, proxyAddr, c, func(opts *PromptMFAChallengeOpts) {
+				opts.Quiet = true
+			})
+			return resp, trace.Wrap(err)
+		})
 	}
 
 	return nil

lib/client/presence.go

@@ -23,18 +23,51 @@ import (
 	"time"
 
 	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 
 	"github.com/gravitational/teleport/api/client/proto"
-	"github.com/gravitational/teleport/lib/auth"
 )
 
-func runPresenceTask(ctx context.Context, term io.Writer, auth auth.ClientI, tc *TeleportClient, sessionID string) error {
+// PresenceMaintainer allows maintaining presence with the Auth service.
+type PresenceMaintainer interface {
+	MaintainSessionPresence(ctx context.Context) (proto.AuthService_MaintainSessionPresenceClient, error)
+}
+
+const mfaChallengeInterval = time.Second * 30
+
+// presenceOptions allows passing optional overrides
+// to RunPresenceTask. Mainly used by tests.
+type presenceOptions struct {
+	Clock clockwork.Clock
+}
+
+// PresenceOption a functional option for RunPresenceTask.
+type PresenceOption func(p *presenceOptions)
+
+// WithPresenceClock sets the clock to be used by RunPresenceTask.
+func WithPresenceClock(clock clockwork.Clock) PresenceOption {
+	return func(p *presenceOptions) {
+		p.Clock = clock
+	}
+}
+
+// RunPresenceTask periodically performs and MFA ceremony to detect that a user is
+// still present and attentive.
+func RunPresenceTask(ctx context.Context, term io.Writer, maintainer PresenceMaintainer, sessionID string, promptMFAChallenge PromptMFAChallengeHandler, opts ...PresenceOption) error {
 	fmt.Fprintf(term, "\r\nTeleport > MFA presence enabled\r\n")
 
-	ticker := time.NewTicker(mfaChallengeInterval)
+	o := &presenceOptions{
+		Clock: clockwork.NewRealClock(),
+	}
+
+	for _, opt := range opts {
+		opt(o)
+	}
+
+	ticker := o.Clock.NewTicker(mfaChallengeInterval)
 	defer ticker.Stop()
 
-	stream, err := auth.MaintainSessionPresence(ctx)
+	stream, err := maintainer.MaintainSessionPresence(ctx)
 	if err != nil {
 		fmt.Fprintf(term, "\r\nstream error: %v\r\n", err)
 		return trace.Wrap(err)
@@ -42,7 +75,7 @@ func runPresenceTask(ctx context.Context, term io.Writer, auth auth.ClientI, tc
 
 	for {
 		select {
-		case <-ticker.C:
+		case <-ticker.Chan():
 			req := &proto.PresenceMFAChallengeSend{
 				Request: &proto.PresenceMFAChallengeSend_ChallengeRequest{
 					ChallengeRequest: &proto.PresenceMFAChallengeRequest{SessionID: sessionID},
@@ -59,11 +92,20 @@ func runPresenceTask(ctx context.Context, term io.Writer, auth auth.ClientI, tc
 				return trace.Wrap(err)
 			}
 
-			solution, err := solveMFA(ctx, term, tc, challenge)
+			fmt.Fprint(term, "\r\nTeleport > Please tap your MFA key\r\n")
+
+			// This is here to enforce the usage of a MFA device.
+			// We don't support TOTP for live presence.
+			challenge.TOTP = nil
+
+			solution, err := promptMFAChallenge(ctx, "" /* proxyAddr */, challenge)
 			if err != nil {
+				fmt.Fprintf(term, "\r\nTeleport > Failed to confirm presence: %v\r\n", err)
 				return trace.Wrap(err)
 			}
 
+			fmt.Fprint(term, "\r\nTeleport > Received MFA presence confirmation\r\n")
+
 			req = &proto.PresenceMFAChallengeSend{
 				Request: &proto.PresenceMFAChallengeSend_ChallengeResponse{
 					ChallengeResponse: solution,
@@ -79,22 +121,3 @@ func runPresenceTask(ctx context.Context, term io.Writer, auth auth.ClientI, tc
 		}
 	}
 }
-
-func solveMFA(ctx context.Context, term io.Writer, tc *TeleportClient, challenge *proto.MFAAuthenticateChallenge) (*proto.MFAAuthenticateResponse, error) {
-	fmt.Fprint(term, "\r\nTeleport > Please tap your MFA key\r\n")
-
-	// This is here to enforce the usage of a MFA device.
-	// We don't support TOTP for live presence.
-	challenge.TOTP = nil
-
-	response, err := tc.PromptMFAChallenge(ctx, "" /* proxyAddr */, challenge, func(opts *PromptMFAChallengeOpts) {
-		opts.Quiet = true
-	})
-	if err != nil {
-		fmt.Fprintf(term, "\r\nTeleport > Failed to confirm presence: %v\r\n", err)
-		return nil, trace.Wrap(err)
-	}
-
-	fmt.Fprint(term, "\r\nTeleport > Received MFA presence confirmation\r\n")
-	return response, nil
-}

lib/client/session.go

@@ -493,17 +493,41 @@ func (ns *NodeSession) updateTerminalSize(ctx context.Context, s *tracessh.Sessi
 	}
 }
 
+// sessionWriter wraps the [tracessh.Session]
+// stdout to prevent any panics that may occur
+// by trying to use it before it has been initialized.
+// In those cases output is written to the stdout that
+// is configured for tsh so that output is not lost entirely.
+type sessionWriter struct {
+	tshOut  io.Writer
+	session *tracessh.Session
+}
+
+func (s *sessionWriter) Write(p []byte) (int, error) {
+	if s.session.Stdout != nil {
+		return s.session.Stdout.Write(p)
+	}
+
+	return s.tshOut.Write(p)
+}
+
 // runShell executes user's shell on the remote node under an interactive session
 func (ns *NodeSession) runShell(ctx context.Context, mode types.SessionParticipantMode, beforeStart func(io.Writer), callback ShellCreatedCallback) error {
 	return ns.interactiveSession(ctx, mode, func(s *tracessh.Session, shell io.ReadWriteCloser) error {
+		w := &sessionWriter{
+			tshOut:  ns.nodeClient.TC.Stdout,
+			session: s,
+		}
+
 		if beforeStart != nil {
-			beforeStart(s.Stdout)
+			beforeStart(w)
 		}
 
 		// start the shell on the server:
 		if err := s.Shell(ctx); err != nil {
 			return trace.Wrap(err)
 		}
+
 		// call the client-supplied callback
 		if callback != nil {
 			exit, err := callback(s, ns.nodeClient.Client, shell)

lib/services/local/sessiontracker.go

@@ -72,7 +72,7 @@ func (s *sessionTracker) UpdatePresence(ctx context.Context, sessionID, user str
 			return trace.Wrap(err)
 		}
 
-		err = session.UpdatePresence(user)
+		err = session.UpdatePresence(user, s.bk.Clock().Now().UTC())
 		if err != nil {
 			return trace.Wrap(err)
 		}

lib/srv/authhandlers.go

@@ -583,17 +583,29 @@ func (a *ahLoginChecker) canLoginWithRBAC(cert *ssh.Certificate, ca types.CertAu
 		return trace.Wrap(err)
 	}
 
-	// we don't need to check the RBAC for the node if they are only allowed to join sessions
-	if osUser == teleport.SSHSessionJoinPrincipal && auth.RoleSupportsModeratedSessions(accessChecker.Roles()) {
-		return nil
-	}
-
 	authPref, err := a.c.AccessPoint.GetAuthPreference(ctx)
 	if err != nil {
 		return trace.Wrap(err)
 	}
 	state := accessChecker.GetAccessState(authPref)
 	_, state.MFAVerified = cert.Extensions[teleport.CertExtensionMFAVerified]
+
+	// we don't need to check the RBAC for the node if they are only allowed to join sessions
+	if osUser == teleport.SSHSessionJoinPrincipal &&
+		auth.RoleSupportsModeratedSessions(accessChecker.Roles()) {
+
+		// allow joining if cluster wide MFA is not required
+		if state.MFARequired != services.MFARequiredAlways {
+			return nil
+		}
+
+		// only allow joining if the MFA ceremony was completed
+		// first if cluster wide MFA is enabled
+		if state.MFAVerified {
+			return nil
+		}
+	}
+
 	state.EnableDeviceVerification = true
 	state.DeviceVerified = dtauthz.IsSSHDeviceVerified(cert)
 

lib/srv/sess.go

@@ -1065,19 +1065,18 @@ func (s *session) launch() {
 	// If the identity is verified with an MFA device, we enabled MFA-based presence for the session.
 	if s.presenceEnabled {
 		go func() {
-			ticker := time.NewTicker(PresenceVerifyInterval)
+			ticker := s.registry.clock.NewTicker(PresenceVerifyInterval)
 			defer ticker.Stop()
-		outer:
 			for {
 				select {
-				case <-ticker.C:
-					err := s.checkPresence()
+				case <-ticker.Chan():
+					err := s.checkPresence(s.serverCtx)
 					if err != nil {
 						s.log.WithError(err).Error("Failed to check presence, terminating session as a security measure")
 						s.Stop()
 					}
 				case <-s.stopC:
-					break outer
+					return
 				}
 			}
 		}()
@@ -1127,7 +1126,7 @@ func (s *session) startInteractive(ctx context.Context, scx *ServerContext, p *p
 	s.inWriter = inWriter
 	s.io.AddReader("reader", inReader)
 	s.io.AddWriter(sessionRecorderID, utils.WriteCloserWithContext(scx.srv.Context(), s.Recorder()))
-	s.BroadcastMessage("Creating session with ID: %v...", s.id)
+	s.BroadcastMessage("Creating session with ID: %v", s.id)
 	s.BroadcastMessage(SessionControlsInfoBroadcast)
 
 	if err := s.startTerminal(ctx, scx); err != nil {
@@ -1546,16 +1545,24 @@ func (s *session) lingerAndDie(ctx context.Context, party *party) {
 	}
 }
 
-func (s *session) checkPresence() error {
+func (s *session) checkPresence(ctx context.Context) error {
+	// We cannot check presence on the local tracker as that will not
+	// be updated in response to parties performing their presence
+	// checks. To prevent the stale version of the session tracker from
+	// terminating a session we must get the session tracker from Auth.
+	tracker, err := s.registry.SessionTrackerService.GetSessionTracker(ctx, s.tracker.tracker.GetSessionID())
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
 	s.mu.Lock()
 	defer s.mu.Unlock()
-
-	for _, participant := range s.tracker.GetParticipants() {
+	for _, participant := range tracker.GetParticipants() {
 		if participant.User == s.initiator {
 			continue
 		}
 
-		if participant.Mode == string(types.SessionModeratorMode) && time.Now().UTC().After(participant.LastActive.Add(PresenceMaxDifference)) {
+		if participant.Mode == string(types.SessionModeratorMode) && s.registry.clock.Now().UTC().After(participant.LastActive.Add(PresenceMaxDifference)) {
 			s.log.Warnf("Participant %v is not active, kicking.", participant.ID)
 			if party := s.parties[rsession.ID(participant.ID)]; party != nil {
 				if err := party.closeUnderSessionLock(); err != nil {
@@ -1839,7 +1846,7 @@ func (s *session) join(ch ssh.Channel, scx *ServerContext, mode types.SessionPar
 		}
 
 		if s.presenceEnabled {
-			_, err := ch.SendRequest(teleport.MFAPresenceRequest, false, nil)
+			_, _, err := scx.ServerConn.SendRequest(teleport.MFAPresenceRequest, false, nil)
 			if err != nil {
 				return trace.WrapWithMessage(err, "failed to send MFA presence request")
 			}
@@ -1970,13 +1977,13 @@ func (s *session) trackSession(ctx context.Context, teleportUser string, policyS
 		HostUser:     teleportUser,
 		Reason:       s.scx.env[teleport.EnvSSHSessionReason],
 		HostPolicies: policySet,
-		Created:      s.registry.clock.Now(),
+		Created:      s.registry.clock.Now().UTC(),
 		Participants: []types.Participant{
 			{
 				ID:         p.id.String(),
 				User:       p.user,
 				Mode:       string(p.mode),
-				LastActive: time.Now().UTC(),
+				LastActive: s.registry.clock.Now().UTC(),
 			},
 		},
 		HostID: s.registry.Srv.ID(),

lib/web/apiserver.go

@@ -2656,11 +2656,13 @@ func (h *Handler) siteNodeConnect(
 		return nil, trace.Wrap(err)
 	}
 
-	var sessionData session.Session
-	var displayLogin string
+	var (
+		sessionData  session.Session
+		displayLogin string
+		tracker      types.SessionTracker
+	)
 
 	clusterName := site.GetName()
-
 	if req.SessionID.IsZero() {
 		// An existing session ID was not provided so we need to create a new one.
 		sessionData, err = h.generateSession(ctx, clt, req, clusterName, sessionCtx)
@@ -2669,10 +2671,11 @@ func (h *Handler) siteNodeConnect(
 			return nil, trace.Wrap(err)
 		}
 	} else {
-		sessionData, displayLogin, err = h.fetchExistingSession(ctx, clt, req, clusterName)
+		sessionData, tracker, err = h.fetchExistingSession(ctx, clt, req, clusterName)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
+		displayLogin = tracker.GetLogin()
 	}
 
 	// If the participantMode is not specified, and the user is the one who created the session,
@@ -2715,6 +2718,8 @@ func (h *Handler) siteNodeConnect(
 		TracerProvider:     h.cfg.TracerProvider,
 		ParticipantMode:    req.ParticipantMode,
 		PROXYSigner:        h.cfg.PROXYSigner,
+		Tracker:            tracker,
+		Clock:              h.clock,
 	}
 
 	term, err := NewTerminal(ctx, terminalConfig)
@@ -2902,20 +2907,20 @@ func findByHost(ctx context.Context, clt auth.ClientI, serverName string) (*host
 }
 
 // fetchExistingSession fetches an active or pending SSH session by the SessionID passed in the TerminalRequest.
-func (h *Handler) fetchExistingSession(ctx context.Context, clt auth.ClientI, req *TerminalRequest, siteName string) (session.Session, string, error) {
+func (h *Handler) fetchExistingSession(ctx context.Context, clt auth.ClientI, req *TerminalRequest, siteName string) (session.Session, types.SessionTracker, error) {
 	sessionID, err := session.ParseID(req.SessionID.String())
 	if err != nil {
-		return session.Session{}, "", trace.Wrap(err)
+		return session.Session{}, nil, trace.Wrap(err)
 	}
 	h.log.Infof("Attempting to join existing session: %s\n", sessionID)
 
 	tracker, err := clt.GetSessionTracker(ctx, string(*sessionID))
 	if err != nil {
-		return session.Session{}, "", trace.Wrap(err)
+		return session.Session{}, nil, trace.Wrap(err)
 	}
 
 	if tracker.GetSessionKind() != types.SSHSessionKind || tracker.GetState() == types.SessionState_SessionStateTerminated {
-		return session.Session{}, "", trace.NotFound("SSH session %v not found", sessionID)
+		return session.Session{}, nil, trace.NotFound("SSH session %v not found", sessionID)
 	}
 
 	sessionData := trackerToLegacySession(tracker, siteName)
@@ -2925,12 +2930,8 @@ func (h *Handler) fetchExistingSession(ctx context.Context, clt auth.ClientI, re
 	// new ones themselves for auditing purposes. Otherwise, the user would
 	// fail the SSH lib username validation step.
 	sessionData.Login = teleport.SSHSessionJoinPrincipal
-	// Using the Login above will then display `-teleport-internal-join` as the
-	// username that the user is logging in as, so we need to instead show the
-	// session username in the UI.
-	displayLogin := tracker.GetLogin()
 
-	return sessionData, displayLogin, nil
+	return sessionData, tracker, nil
 }
 
 type siteSessionGenerateResponse struct {

lib/web/apiserver_test.go

@@ -212,8 +212,7 @@ func newWebSuiteWithConfig(t *testing.T, cfg webSuiteConfig) *WebSuite {
 			Dir:                     t.TempDir(),
 			Clock:                   s.clock,
 			ClusterNetworkingConfig: networkingConfig,
-
-			AuthPreferenceSpec: cfg.authPreferenceSpec,
+			AuthPreferenceSpec:      cfg.authPreferenceSpec,
 		},
 	}
 
@@ -622,11 +621,12 @@ type authPack struct {
 	session   *CreateSessionResponse
 	clt       *TestWebClient
 	cookies   []*http.Cookie
+	device    *auth.TestDevice
 }
 
 // authPack returns new authenticated package consisting of created valid
 // user, otp token, created web session and authenticated client.
-func (s *WebSuite) authPack(t *testing.T, user string) *authPack {
+func (s *WebSuite) authPack(t *testing.T, user string, roles ...string) *authPack {
 	login := s.user
 	pass := "abc123"
 	rawSecret := "def456"
@@ -640,7 +640,7 @@ func (s *WebSuite) authPack(t *testing.T, user string) *authPack {
 	err = s.server.Auth().SetAuthPreference(s.ctx, ap)
 	require.NoError(t, err)
 
-	s.createUser(t, user, login, pass, otpSecret)
+	s.createUser(t, user, login, pass, otpSecret, roles...)
 
 	// create a valid otp token
 	validToken, err := totp.GenerateCode(otpSecret, s.clock.Now())
@@ -679,7 +679,93 @@ func (s *WebSuite) authPack(t *testing.T, user string) *authPack {
 	}
 }
 
-func (s *WebSuite) createUser(t *testing.T, user string, login string, pass string, otpSecret string) {
+func (s *WebSuite) authPackWithMFA(t *testing.T, name string, roles ...types.Role) *authPack {
+	const password = "testing"
+	user, err := types.NewUser(name)
+	require.NoError(t, err)
+
+	userRole := services.RoleForUser(user)
+	userRole.SetLogins(types.Allow, []string{s.user})
+	err = s.server.Auth().UpsertRole(s.ctx, userRole)
+	require.NoError(t, err)
+
+	for _, role := range roles {
+		err = s.server.Auth().UpsertRole(s.ctx, role)
+		require.NoError(t, err)
+		user.AddRole(role.GetName())
+	}
+
+	user.AddRole(userRole.GetName())
+	err = s.server.Auth().CreateUser(s.ctx, user)
+	require.NoError(t, err)
+
+	clt := s.client(t)
+
+	// create register challenge
+	token, err := s.server.Auth().CreateResetPasswordToken(s.ctx, auth.CreateUserTokenRequest{
+		Name: name,
+	})
+	require.NoError(t, err)
+
+	res, err := s.server.Auth().CreateRegisterChallenge(s.ctx, &authproto.CreateRegisterChallengeRequest{
+		TokenID:     token.GetName(),
+		DeviceType:  authproto.DeviceType_DEVICE_TYPE_WEBAUTHN,
+		DeviceUsage: authproto.DeviceUsage_DEVICE_USAGE_PASSWORDLESS,
+	})
+	require.NoError(t, err)
+
+	cc := wanlib.CredentialCreationFromProto(res.GetWebauthn())
+
+	// use passwordless as auth method
+	device, err := mocku2f.Create()
+	require.NoError(t, err)
+
+	device.SetPasswordless()
+
+	ccr, err := device.SignCredentialCreation("https://localhost", cc)
+	require.NoError(t, err)
+
+	_, err = s.server.Auth().ChangeUserAuthentication(s.ctx, &authproto.ChangeUserAuthenticationRequest{
+		TokenID:     token.GetName(),
+		NewPassword: []byte(password),
+		NewMFARegisterResponse: &authproto.MFARegisterResponse{
+			Response: &authproto.MFARegisterResponse_Webauthn{
+				Webauthn: wanlib.CredentialCreationResponseToProto(ccr),
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	beginReq := &client.MFAChallengeRequest{
+		User: name,
+		Pass: password,
+	}
+	re, err := s.loginMFA(clt, beginReq, device)
+	require.NoError(t, err)
+
+	var rawSess *CreateSessionResponse
+	require.NoError(t, json.Unmarshal(re.Bytes(), &rawSess))
+
+	sess, err := rawSess.response()
+	require.NoError(t, err)
+
+	jar, err := cookiejar.New(nil)
+	require.NoError(t, err)
+
+	clt = s.client(t, roundtrip.BearerAuth(sess.Token), roundtrip.CookieJar(jar))
+	jar.SetCookies(s.url(), re.Cookies())
+
+	return &authPack{
+		user:    name,
+		login:   s.user,
+		session: sess,
+		clt:     clt,
+		cookies: re.Cookies(),
+		device:  &auth.TestDevice{Key: device},
+	}
+}
+
+func (s *WebSuite) createUser(t *testing.T, user string, login string, pass string, otpSecret string, roles ...string) {
 	teleUser, err := types.NewUser(user)
 	require.NoError(t, err)
 	role := services.RoleForUser(teleUser)
@@ -691,6 +777,10 @@ func (s *WebSuite) createUser(t *testing.T, user string, login string, pass stri
 	require.NoError(t, err)
 	teleUser.AddRole(role.GetName())
 
+	for _, r := range roles {
+		teleUser.AddRole(r)
+	}
+
 	teleUser.SetCreatedBy(types.CreatedBy{
 		User: types.UserRef{Name: "some-auth-user"},
 	})
@@ -1700,8 +1790,7 @@ func TestTerminalRouting(t *testing.T) {
 			require.NoError(t, err)
 			t.Cleanup(func() { tt.wsCloseAssertion(t, ws.Close()) })
 
-			stream, err := NewTerminalStream(ws)
-			require.NoError(t, err)
+			stream := NewTerminalStream(s.ctx, ws, utils.NewLoggerForTests())
 
 			// here we intentionally run a command where the output we're looking
 			// for is not present in the command itself
@@ -1820,12 +1909,20 @@ func TestTerminalRequireSessionMfa(t *testing.T) {
 				res, err := dev.SolveAuthn(&authproto.MFAAuthenticateChallenge{
 					WebauthnChallenge: wanlib.CredentialAssertionToProto(chals.WebauthnChallenge),
 				})
-				require.Nil(t, err)
+				require.NoError(t, err)
 
 				webauthnResBytes, err := json.Marshal(wanlib.CredentialAssertionResponseFromProto(res.GetWebauthn()))
-				require.Nil(t, err)
+				require.NoError(t, err)
 
-				return webauthnResBytes
+				envelope := &Envelope{
+					Version: defaults.WebsocketVersion,
+					Type:    defaults.WebsocketWebauthnChallenge,
+					Payload: string(webauthnResBytes),
+				}
+				protoBytes, err := proto.Marshal(envelope)
+				require.NoError(t, err)
+
+				return protoBytes
 			},
 		},
 	}
@@ -1851,15 +1948,14 @@ func TestTerminalRequireSessionMfa(t *testing.T) {
 			require.Nil(t, json.Unmarshal([]byte(env.Payload), &chals))
 
 			// Send response over ws.
-			stream, err := NewTerminalStream(ws)
-			require.NoError(t, err)
-			_, err = stream.Write(tc.getChallengeResponseBytes(chals, dev))
+			stream := NewTerminalStream(ctx, ws, utils.NewLoggerForTests())
+			err = stream.ws.WriteMessage(websocket.BinaryMessage, tc.getChallengeResponseBytes(chals, dev))
 			require.Nil(t, err)
 
 			// Test we can write.
-			_, err = io.WriteString(stream, "echo alpacas\r\n")
+			_, err = io.WriteString(stream, "echo txlxport | sed 's/x/e/g'\r\n")
 			require.Nil(t, err)
-			require.Nil(t, waitForOutput(stream, "alpacas"))
+			require.Nil(t, waitForOutput(stream, "teleport"))
 		})
 	}
 }
@@ -1955,7 +2051,7 @@ func TestDesktopAccessMFARequiresMfa(t *testing.T) {
 				},
 				RequireMFAType: types.RequireMFAType_SESSION,
 			},
-			mfaHandler: handleMFAWebauthnChallenge,
+			mfaHandler: handleDesktopMFAWebauthnChallenge,
 			registerDevice: func(t *testing.T, ctx context.Context, clt *auth.Client) *auth.TestDevice {
 				webauthnDev, err := auth.RegisterTestDevice(ctx, clt, "webauthn", authproto.DeviceType_DEVICE_TYPE_WEBAUTHN, nil /* authenticator */)
 				require.NoError(t, err)
@@ -2013,7 +2109,7 @@ func TestDesktopAccessMFARequiresMfa(t *testing.T) {
 	}
 }
 
-func handleMFAWebauthnChallenge(t *testing.T, ws *websocket.Conn, dev *auth.TestDevice) {
+func handleDesktopMFAWebauthnChallenge(t *testing.T, ws *websocket.Conn, dev *auth.TestDevice) {
 	br := bufio.NewReader(&WebsocketIO{Conn: ws})
 	mt, err := br.ReadByte()
 	require.NoError(t, err)
@@ -2043,8 +2139,7 @@ func TestWebAgentForward(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, ws.Close()) })
 
-	stream, err := NewTerminalStream(ws)
-	require.NoError(t, err)
+	stream := NewTerminalStream(s.ctx, ws, utils.NewLoggerForTests())
 
 	_, err = io.WriteString(stream, "echo $SSH_AUTH_SOCK\r\n")
 	require.NoError(t, err)
@@ -2156,8 +2251,7 @@ func TestCloseConnectionsOnLogout(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, ws.Close()) })
 
-	stream, err := NewTerminalStream(ws)
-	require.NoError(t, err)
+	stream := NewTerminalStream(s.ctx, ws, utils.NewLoggerForTests())
 
 	// to make sure we have a session
 	_, err = io.WriteString(stream, "expr 137 + 39\r\n")
@@ -6958,6 +7052,10 @@ func (mock authProviderMock) GenerateOpenSSHCert(ctx context.Context, req *authp
 	return nil, nil
 }
 
+func (mock authProviderMock) MaintainSessionPresence(ctx context.Context) (authproto.AuthService_MaintainSessionPresenceClient, error) {
+	return nil, nil
+}
+
 func (mock authProviderMock) GetUser(_ string, _ bool) (types.User, error) {
 	return nil, nil
 }
@@ -7055,9 +7153,11 @@ func (s *WebSuite) makeTerminal(t *testing.T, pack *authPack, opts ...terminalOp
 	return ws, &sessResp.Session, nil
 }
 
-func waitForOutput(stream *TerminalStream, substr string) error {
+func waitForOutput(r io.Reader, substr string) error {
 	timeoutCh := time.After(10 * time.Second)
 
+	var prev string
+	out := make([]byte, int64(len(substr)*2))
 	for {
 		select {
 		case <-timeoutCh:
@@ -7065,17 +7165,23 @@ func waitForOutput(stream *TerminalStream, substr string) error {
 		default:
 		}
 
-		out := make([]byte, 100)
-		n, err := stream.Read(out)
+		n, err := r.Read(out)
+		outStr := removeSpace(string(out[:n]))
 
-		// check for the string before checking the error,
-		// as it's valid for n > 0 even when there is an error
-		if n > 0 && strings.Contains(removeSpace(string(out[:n])), substr) {
+		// Check for [substr] before checking the error,
+		// as it's valid for n > 0 even when there is an error.
+		// The [substr] is checked against the current and previous
+		// output to account for scenarios where the [substr] is split
+		// across two reads. While we try to prevent this by reading
+		// twice the length of [substr] there are no guarantees the
+		// whole thing will arrive in a single read.
+		if n > 0 && strings.Contains(prev+outStr, substr) {
 			return nil
 		}
 		if err != nil {
 			return trace.Wrap(err)
 		}
+		prev = outStr
 	}
 }
 
@@ -7122,6 +7228,54 @@ func (s *WebSuite) login(clt *TestWebClient, cookieToken string, reqToken string
 	}))
 }
 
+func (s *WebSuite) loginMFA(clt *TestWebClient, reqData *client.MFAChallengeRequest, device *mocku2f.Key) (*roundtrip.Response, error) {
+	resp, err := httplib.ConvertResponse(clt.RoundTrip(func() (*http.Response, error) {
+		data, err := json.Marshal(reqData)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		req, err := http.NewRequest("POST", clt.Endpoint("webapi", "mfa", "login", "begin"), bytes.NewBuffer(data))
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		req.Header.Set("Content-Type", "application/json")
+		resp, err := clt.HTTPClient().Do(req)
+		return resp, trace.Wrap(err)
+	}))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	var challenge client.MFAAuthenticateChallenge
+	err = json.Unmarshal(resp.Bytes(), &challenge)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	car, err := device.SignAssertion("https://localhost", challenge.WebauthnChallenge)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return httplib.ConvertResponse(clt.RoundTrip(func() (*http.Response, error) {
+		respData := &client.AuthenticateWebUserRequest{
+			User:                      reqData.User,
+			WebauthnAssertionResponse: car,
+		}
+		data, err := json.Marshal(respData)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		req, err := http.NewRequest("POST", clt.Endpoint("webapi", "mfa", "login", "finishsession"), bytes.NewBuffer(data))
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		req.Header.Set("Content-Type", "application/json")
+		resp, err := clt.HTTPClient().Do(req)
+		return resp, trace.Wrap(err)
+	}))
+}
+
 func (s *WebSuite) url() *url.URL {
 	u, err := url.Parse("https://" + s.webServer.Listener.Addr().String())
 	if err != nil {
@@ -7759,12 +7913,11 @@ func login(t *testing.T, clt *TestWebClient, cookieToken, reqToken string, reqDa
 
 func validateTerminalStream(t *testing.T, ws *websocket.Conn) {
 	t.Helper()
-	stream, err := NewTerminalStream(ws)
-	require.NoError(t, err)
+	stream := NewTerminalStream(context.Background(), ws, utils.NewLoggerForTests())
 
 	// here we intentionally run a command where the output we're looking
 	// for is not present in the command itself
-	_, err = io.WriteString(stream, "echo txlxport | sed 's/x/e/g'\r\n")
+	_, err := io.WriteString(stream, "echo txlxport | sed 's/x/e/g'\r\n")
 	require.NoError(t, err)
 	require.NoError(t, waitForOutput(stream, "teleport"))
 }
@@ -8792,3 +8945,224 @@ type mockedPingTestProxy struct {
 func (m mockedPingTestProxy) Ping(ctx context.Context) (authproto.PingResponse, error) {
 	return m.mockedPing(ctx)
 }
+
+// TestModeratedSession validates that peers are able to start Moderated
+// Sessions and remain in the waiting room until the required number of
+// moderators are present. Only when the moderator is present the peer
+// is allowed to access the host and start entering input and receiving
+// output until the moderator terminates the session.
+func TestModeratedSession(t *testing.T) {
+	modules.SetTestModules(t, &modules.TestModules{TestBuildType: modules.BuildEnterprise})
+
+	ctx := context.Background()
+	s := newWebSuite(t)
+
+	peerRole, err := types.NewRole("moderated", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			RequireSessionJoin: []*types.SessionRequirePolicy{
+				{
+					Name:   "moderated",
+					Filter: "contains(user.roles, \"moderator\")",
+					Kinds:  []string{string(types.SSHSessionKind)},
+					Count:  1,
+					Modes:  []string{string(types.SessionModeratorMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, s.server.Auth().UpsertRole(s.ctx, peerRole))
+
+	moderatorRole, err := types.NewRole("moderator", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			JoinSessions: []*types.SessionJoinPolicy{
+				{
+					Name:  "moderated",
+					Roles: []string{peerRole.GetName()},
+					Kinds: []string{string(types.SSHSessionKind)},
+					Modes: []string{string(types.SessionModeratorMode), string(types.SessionObserverMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+	require.NoError(t, s.server.Auth().UpsertRole(s.ctx, moderatorRole))
+
+	peer := s.authPack(t, "foo", peerRole.GetName())
+
+	peerWS, sess, err := s.makeTerminal(t, peer)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, peerWS.Close()) })
+
+	peerStream := NewTerminalStream(ctx, peerWS, utils.NewLoggerForTests())
+
+	require.NoError(t, waitForOutput(peerStream, "Teleport > User foo joined the session with participant mode: peer."))
+
+	moderator := s.authPack(t, "bar", moderatorRole.GetName())
+	moderatorWS, _, err := s.makeTerminal(t, moderator, withSessionID(sess.ID), withParticipantMode(types.SessionModeratorMode))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, moderatorWS.Close()) })
+
+	moderatorStream := NewTerminalStream(ctx, moderatorWS, utils.NewLoggerForTests())
+
+	require.NoError(t, waitForOutput(peerStream, "Teleport > Connecting to node over SSH"))
+
+	// here we intentionally run a command where the output we're looking
+	// for is not present in the command itself
+	_, err = io.WriteString(peerStream, "echo llxmx | sed 's/x/a/g'\r\n")
+	require.NoError(t, err)
+	require.NoError(t, waitForOutput(peerStream, "llama"))
+	require.NoError(t, waitForOutput(moderatorStream, "llama"))
+
+	// the moderator terminates the session
+	_, err = io.WriteString(moderatorStream, "t")
+	require.NoError(t, err)
+
+	require.NoError(t, waitForOutput(moderatorStream, "Stopping session..."))
+	require.NoError(t, waitForOutput(peerStream, "Process exited with status 255"))
+}
+
+// TestModeratedSessionWithMFA validates the same behavior as TestModeratedSession while
+// also ensuring that MFA is performed prior to accessing the host and that periodic
+// presence checks are performed by the moderator. When presence checks are not performed
+// the session is aborted.
+func TestModeratedSessionWithMFA(t *testing.T) {
+	modules.SetTestModules(t, &modules.TestModules{TestBuildType: modules.BuildEnterprise})
+	ctx := context.Background()
+
+	const RPID = "localhost"
+
+	s := newWebSuiteWithConfig(t, webSuiteConfig{
+		authPreferenceSpec: &types.AuthPreferenceSpecV2{
+			Type:           constants.Local,
+			ConnectorName:  constants.PasswordlessConnector,
+			SecondFactor:   constants.SecondFactorOn,
+			RequireMFAType: types.RequireMFAType_SESSION,
+			Webauthn: &types.Webauthn{
+				RPID: RPID,
+			},
+		},
+	})
+
+	peerRole, err := types.NewRole("moderated", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			RequireSessionJoin: []*types.SessionRequirePolicy{
+				{
+					Name:   "moderated",
+					Filter: "contains(user.roles, \"moderator\")",
+					Kinds:  []string{string(types.SSHSessionKind)},
+					Count:  1,
+					Modes:  []string{string(types.SessionModeratorMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	moderatorRole, err := types.NewRole("moderator", types.RoleSpecV6{
+		Allow: types.RoleConditions{
+			JoinSessions: []*types.SessionJoinPolicy{
+				{
+					Name:  "moderated",
+					Roles: []string{peerRole.GetName()},
+					Kinds: []string{string(types.SSHSessionKind)},
+					Modes: []string{string(types.SessionModeratorMode), string(types.SessionObserverMode)},
+				},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	peer := s.authPackWithMFA(t, "foo", peerRole)
+	moderator := s.authPackWithMFA(t, "bar", moderatorRole)
+
+	peerWS, sess, err := s.makeTerminal(t, peer)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, peerWS.Close()) })
+
+	handleMFAWebauthnChallenge(t, peerWS, peer.device)
+
+	peerStream := NewTerminalStream(ctx, peerWS, utils.NewLoggerForTests())
+
+	require.NoError(t, waitForOutput(peerStream, "Teleport > User foo joined the session with participant mode: peer."))
+
+	moderatorWS, _, err := s.makeTerminal(t, moderator, withSessionID(sess.ID), withParticipantMode(types.SessionModeratorMode))
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, moderatorWS.Close()) })
+
+	handleMFAWebauthnChallenge(t, moderatorWS, moderator.device)
+
+	moderatorStream := NewTerminalStream(ctx, moderatorWS, utils.NewLoggerForTests())
+
+	require.NoError(t, waitForOutput(peerStream, "Teleport > Connecting to node over SSH"))
+
+	// here we intentionally run a command where the output we're looking
+	// for is not present in the command itself
+	_, err = io.WriteString(peerStream, "echo llxmx | sed 's/x/a/g'\r\n")
+	require.NoError(t, err)
+	require.NoError(t, waitForOutput(peerStream, "llama"))
+	require.NoError(t, waitForOutput(moderatorStream, "llama"))
+
+	// run the presence check a few times
+	for i := 0; i < 3; i++ {
+		s.clock.Advance(30 * time.Second)
+		require.NoError(t, waitForOutput(moderatorStream, "Teleport > Please tap your MFA key"))
+
+		challenge, err := moderatorStream.readChallenge(protobufMFACodec{})
+		require.NoError(t, err)
+
+		res, err := moderator.device.SolveAuthn(challenge)
+		require.NoError(t, err)
+
+		webauthnResBytes, err := json.Marshal(wanlib.CredentialAssertionResponseFromProto(res.GetWebauthn()))
+		require.NoError(t, err)
+
+		envelope := &Envelope{
+			Version: defaults.WebsocketVersion,
+			Type:    defaults.WebsocketWebauthnChallenge,
+			Payload: string(webauthnResBytes),
+		}
+		envelopeBytes, err := proto.Marshal(envelope)
+		require.NoError(t, err)
+
+		require.NoError(t, moderatorWS.WriteMessage(websocket.BinaryMessage, envelopeBytes))
+	}
+
+	// advance the clock far enough in the future to make the moderator stale
+	// which will terminate the session
+	s.clock.Advance(180 * time.Second)
+	require.NoError(t, waitForOutput(moderatorStream, "wait: remote command exited without exit status or exit signal"))
+	require.NoError(t, waitForOutput(peerStream, "Process exited with status 255"))
+}
+
+func handleMFAWebauthnChallenge(t *testing.T, ws *websocket.Conn, dev *auth.TestDevice) {
+	// Wait for websocket authn challenge event.
+	ty, raw, err := ws.ReadMessage()
+	require.NoError(t, err)
+	require.Equal(t, websocket.BinaryMessage, ty)
+
+	var env Envelope
+	require.NoError(t, proto.Unmarshal(raw, &env))
+
+	var challenge client.MFAAuthenticateChallenge
+	require.NoError(t, json.Unmarshal([]byte(env.Payload), &challenge))
+
+	res, err := dev.SolveAuthn(&authproto.MFAAuthenticateChallenge{
+		WebauthnChallenge: wanlib.CredentialAssertionToProto(challenge.WebauthnChallenge),
+	})
+	require.NoError(t, err)
+
+	webauthnResBytes, err := json.Marshal(wanlib.CredentialAssertionResponseFromProto(res.GetWebauthn()))
+	require.NoError(t, err)
+
+	envelope := &Envelope{
+		Version: defaults.WebsocketVersion,
+		Type:    defaults.WebsocketWebauthnChallenge,
+		Payload: string(webauthnResBytes),
+	}
+	envelopeBytes, err := proto.Marshal(envelope)
+	require.NoError(t, err)
+
+	require.NoError(t, ws.WriteMessage(websocket.BinaryMessage, envelopeBytes))
+
+}

lib/web/command.go

@@ -305,8 +305,7 @@ func getMFACacheFn() mfaFuncCache {
 			return authMethods, nil
 		}
 
-		var err error
-		authMethods, err = issueMfaAuthFn()
+		authMethods, err := issueMfaAuthFn()
 		return authMethods, trace.Wrap(err)
 	}
 }
@@ -485,7 +484,7 @@ func (t *commandHandler) ServeHTTP(_ http.ResponseWriter, r *http.Request) {
 }
 
 func (t *commandHandler) handler(r *http.Request) {
-	t.stream = NewWStream(t.ws)
+	t.stream = NewWStream(r.Context(), t.ws, t.log, nil)
 
 	// Create a Teleport client, if not able to, show the reason to the user in
 	// the terminal.
@@ -500,8 +499,7 @@ func (t *commandHandler) handler(r *http.Request) {
 
 	// Update the read deadline upon receiving a pong message.
 	t.ws.SetPongHandler(func(_ string) error {
-		t.ws.SetReadDeadline(deadlineForInterval(t.keepAliveInterval))
-		return nil
+		return trace.Wrap(t.ws.SetReadDeadline(deadlineForInterval(t.keepAliveInterval)))
 	})
 
 	// Start sending ping frames through websocket to the client.

lib/web/command_test.go

@@ -17,6 +17,7 @@
 package web
 
 import (
+	"context"
 	"crypto/tls"
 	"encoding/base64"
 	"encoding/json"
@@ -39,6 +40,7 @@ import (
 
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/session"
+	"github.com/gravitational/teleport/lib/utils"
 )
 
 func TestExecuteCommand(t *testing.T) {
@@ -49,7 +51,7 @@ func TestExecuteCommand(t *testing.T) {
 	require.NoError(t, err)
 	t.Cleanup(func() { require.NoError(t, ws.Close()) })
 
-	stream := NewWStream(ws)
+	stream := NewWStream(context.Background(), ws, utils.NewLoggerForTests(), nil)
 
 	require.NoError(t, waitForCommandOutput(stream, "teleport"))
 }

lib/web/command_utils.go

@@ -41,7 +41,7 @@ type WSConn interface {
 	WriteMessage(messageType int, data []byte) error
 	NextReader() (messageType int, r io.Reader, err error)
 	ReadMessage() (messageType int, p []byte, err error)
-
+	SetReadLimit(limit int64)
 	SetReadDeadline(t time.Time) error
 	PingHandler() func(appData string) error
 	SetPingHandler(h func(appData string) error)

lib/web/desktop.go

@@ -43,7 +43,9 @@ import (
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/utils/keys"
 	"github.com/gravitational/teleport/lib/auth"
+	wanlib "github.com/gravitational/teleport/lib/auth/webauthn"
 	"github.com/gravitational/teleport/lib/client"
+	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/httplib"
 	"github.com/gravitational/teleport/lib/multiplexer"
 	"github.com/gravitational/teleport/lib/reversetunnel"
@@ -248,10 +250,8 @@ func desktopTLSConfig(ctx context.Context, ws *websocket.Conn, pc *client.ProxyC
 		return nil, trace.Wrap(err)
 	}
 
-	stream, err := NewTerminalStream(ws)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
+	codec := tdpMFACodec{}
+
 	key, err := pc.IssueUserCertsWithMFA(ctx, client.ReissueParams{
 		RouteToWindowsDesktop: proto.RouteToWindowsDesktop{
 			WindowsDesktop: desktopName,
@@ -264,7 +264,33 @@ func desktopTLSConfig(ctx context.Context, ws *websocket.Conn, pc *client.ProxyC
 			TLSCert:             sessCtx.cfg.Session.GetTLSCert(),
 			WindowsDesktopCerts: make(map[string][]byte),
 		},
-	}, promptMFAChallenge(&stream.WSStream, tdpMFACodec{}))
+	}, func(ctx context.Context, proxyAddr string, c *proto.MFAAuthenticateChallenge) (*proto.MFAAuthenticateResponse, error) {
+		challenge := &client.MFAAuthenticateChallenge{
+			WebauthnChallenge: wanlib.CredentialAssertionFromProto(c.WebauthnChallenge),
+		}
+
+		// Send the challenge over the socket.
+		msg, err := codec.encode(challenge, defaults.WebsocketWebauthnChallenge)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		if err := ws.WriteMessage(websocket.BinaryMessage, msg); err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		ty, buf, err := ws.ReadMessage()
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		if ty != websocket.BinaryMessage {
+			return nil, trace.BadParameter("received unexpected web socket message type %d", ty)
+		}
+
+		resp, err := codec.decodeResponse(buf, defaults.WebsocketWebauthnChallenge)
+		return resp, trace.Wrap(err)
+	})
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/web/fuzz_test.go

@@ -28,7 +28,7 @@ func FuzzTdpMFACodecDecode(f *testing.F) {
 	f.Fuzz(func(t *testing.T, buf []byte) {
 		require.NotPanics(t, func() {
 			codec := tdpMFACodec{}
-			_, _ = codec.decode(buf, "")
+			_, _ = codec.decodeResponse(buf, "")
 		})
 	})
 }

lib/web/mfa_codec.go

@@ -24,6 +24,7 @@ import (
 	"github.com/gravitational/trace"
 
 	authproto "github.com/gravitational/teleport/api/client/proto"
+	wanlib "github.com/gravitational/teleport/lib/auth/webauthn"
 	"github.com/gravitational/teleport/lib/client"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/srv/desktop/tdp"
@@ -36,8 +37,11 @@ type mfaCodec interface {
 	// encode converts an MFA challenge to wire format
 	encode(chal *client.MFAAuthenticateChallenge, envelopeType string) ([]byte, error)
 
-	// decode parses an MFA authentication response
-	decode(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error)
+	// decodeChallenge parses an MFA authentication challenge
+	decodeChallenge(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateChallenge, error)
+
+	// decodeResponse parses an MFA authentication response
+	decodeResponse(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error)
 }
 
 // protobufMFACodec converts MFA challenges and responses to the protobuf
@@ -61,13 +65,19 @@ func (protobufMFACodec) encode(chal *client.MFAAuthenticateChallenge, envelopeTy
 	return protoBytes, nil
 }
 
-func (protobufMFACodec) decode(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error) {
-	envelope := &Envelope{}
-	if err := proto.Unmarshal(bytes, envelope); err != nil {
+func (protobufMFACodec) decodeResponse(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error) {
+	return mfajson.Decode(bytes, envelopeType)
+}
+
+func (protobufMFACodec) decodeChallenge(bytes []byte, envelopeType string) (*authproto.MFAAuthenticateChallenge, error) {
+	var challenge client.MFAAuthenticateChallenge
+	if err := json.Unmarshal(bytes, &challenge); err != nil {
 		return nil, trace.Wrap(err)
 	}
 
-	return mfajson.Decode([]byte(envelope.Payload), envelopeType)
+	return &authproto.MFAAuthenticateChallenge{
+		WebauthnChallenge: wanlib.CredentialAssertionToProto(challenge.WebauthnChallenge),
+	}, nil
 }
 
 // tdpMFACodec converts MFA challenges and responses to Teleport Desktop
@@ -89,7 +99,7 @@ func (tdpMFACodec) encode(chal *client.MFAAuthenticateChallenge, envelopeType st
 	return tdpMsg.Encode()
 }
 
-func (tdpMFACodec) decode(buf []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error) {
+func (tdpMFACodec) decodeResponse(buf []byte, envelopeType string) (*authproto.MFAAuthenticateResponse, error) {
 	if len(buf) == 0 {
 		return nil, trace.BadParameter("empty MFA message received")
 	}
@@ -102,3 +112,20 @@ func (tdpMFACodec) decode(buf []byte, envelopeType string) (*authproto.MFAAuthen
 	}
 	return msg.MFAAuthenticateResponse, nil
 }
+
+func (tdpMFACodec) decodeChallenge(buf []byte, envelopeType string) (*authproto.MFAAuthenticateChallenge, error) {
+	if len(buf) == 0 {
+		return nil, trace.BadParameter("empty MFA message received")
+	}
+	if tdp.MessageType(buf[0]) != tdp.TypeMFA {
+		return nil, trace.BadParameter("expected MFA message type %v, got %v", tdp.TypeMFA, buf[0])
+	}
+	msg, err := tdp.DecodeMFAChallenge(bytes.NewReader(buf[1:]))
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &authproto.MFAAuthenticateChallenge{
+		WebauthnChallenge: wanlib.CredentialAssertionToProto(msg.WebauthnChallenge),
+	}, nil
+}

lib/web/terminal.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
@@ -31,6 +32,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/gorilla/websocket"
 	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 	"github.com/sirupsen/logrus"
 	oteltrace "go.opentelemetry.io/otel/trace"
 	"golang.org/x/crypto/ssh"
@@ -96,6 +98,7 @@ type AuthProvider interface {
 	GetSessionTracker(ctx context.Context, sessionID string) (types.SessionTracker, error)
 	IsMFARequired(ctx context.Context, req *authproto.IsMFARequiredRequest) (*authproto.IsMFARequiredResponse, error)
 	GenerateUserSingleUseCerts(ctx context.Context) (authproto.AuthService_GenerateUserSingleUseCertsClient, error)
+	MaintainSessionPresence(ctx context.Context) (authproto.AuthService_MaintainSessionPresenceClient, error)
 }
 
 // NewTerminal creates a web-based terminal based on WebSockets and returns a
@@ -130,11 +133,13 @@ func NewTerminal(ctx context.Context, cfg TerminalHandlerConfig) (*TerminalHandl
 		term:            cfg.Term,
 		proxySigner:     cfg.PROXYSigner,
 		participantMode: cfg.ParticipantMode,
+		tracker:         cfg.Tracker,
+		clock:           cfg.Clock,
 	}, nil
 }
 
 // TerminalHandlerConfig contains the configuration options necessary to
-// correctly setup the TerminalHandler
+// correctly set up the TerminalHandler
 type TerminalHandlerConfig struct {
 	// Term is the initial PTY size.
 	Term session.TerminalParams
@@ -169,6 +174,11 @@ type TerminalHandlerConfig struct {
 	tracer oteltrace.Tracer
 	// ParticipantMode is the mode that determines what you can do when you join an active session.
 	ParticipantMode types.SessionParticipantMode
+	// Tracker is the session tracker of the session being joined. May be nil
+	// if the user is not joining a session.
+	Tracker types.SessionTracker
+	// Clock used for presence checking.
+	Clock clockwork.Clock
 }
 
 func (t *TerminalHandlerConfig) CheckAndSetDefaults() error {
@@ -253,15 +263,6 @@ type TerminalHandler struct {
 	// displayLogin is the login name to display in the UI.
 	displayLogin string
 
-	// sshSession holds the "shell" SSH channel to the node.
-	sshSession *tracessh.Session
-
-	// terminalContext is used to signal when the terminal sesson is closing.
-	terminalContext context.Context
-
-	// terminalCancel is used to signal when the terminal session is closing.
-	terminalCancel context.CancelFunc
-
 	closeOnce sync.Once
 
 	// term is the initial PTY size.
@@ -276,6 +277,12 @@ type TerminalHandler struct {
 	// stream manages sending and receiving [Envelope] to the UI
 	// for the duration of the session
 	stream *TerminalStream
+	// tracker is the session tracker of the session being joined. May be nil
+	// if the user is not joining a session.
+	tracker types.SessionTracker
+
+	// clock to use for presence checking
+	clock clockwork.Clock
 }
 
 // ServeHTTP builds a connection to the remote node and then pumps back two types of
@@ -348,17 +355,15 @@ func (t *TerminalHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 // Close the websocket stream.
 func (t *TerminalHandler) Close() error {
+	var err error
 	t.closeOnce.Do(func() {
-		// Close the SSH connection to the remote node.
-		if t.sshSession != nil {
-			t.sshSession.Close()
-		}
-
 		// If the terminal handler was closed (most likely due to the *SessionContext
 		// closing) then the stream should be closed as well.
-		t.terminalCancel()
+		if t.stream != nil {
+			err = t.stream.Close()
+		}
 	})
-	return nil
+	return trace.Wrap(err)
 }
 
 // handler is the main websocket loop. It creates a Teleport client and then
@@ -367,61 +372,42 @@ func (t *TerminalHandler) Close() error {
 func (t *TerminalHandler) handler(ws *websocket.Conn, r *http.Request) {
 	defer ws.Close()
 
-	// Create a terminal stream that wraps/unwraps the envelope used to
-	// communicate over the websocket.
-	resizeC := make(chan *session.TerminalParams, 1)
-	fileTransferRequestC := make(chan *session.FileTransferRequestParams, 1)
-	fileTransferDecisionC := make(chan *session.FileTransferDecisionParams, 1)
-	stream, err := NewTerminalStream(ws, WithTerminalStreamResizeHandler(resizeC), WithTerminalStreamFileTransferHandlers(fileTransferRequestC, fileTransferDecisionC))
-	if err != nil {
-		t.log.WithError(err).Info("Failed creating a terminal stream for session")
-		t.writeError(err)
-		return
-	}
-	t.stream = stream
+	// Update the read deadline upon receiving a pong message.
+	ws.SetPongHandler(func(_ string) error {
+		return trace.Wrap(ws.SetReadDeadline(deadlineForInterval(t.keepAliveInterval)))
+	})
 
 	// Create a context for signaling when the terminal session is over and
 	// link it first with the trace context from the request context
 	tctx := oteltrace.ContextWithRemoteSpanContext(context.Background(), oteltrace.SpanContextFromContext(r.Context()))
-	t.terminalContext, t.terminalCancel = context.WithCancel(tctx)
+	ctx, cancel := context.WithCancel(tctx)
+	defer cancel()
+	t.stream = NewTerminalStream(ctx, ws, t.log)
 
 	// Create a Teleport client, if not able to, show the reason to the user in
 	// the terminal.
-	tc, err := t.makeClient(r.Context(), ws)
+	tc, err := t.makeClient(ctx, t.stream, ws.RemoteAddr().String())
 	if err != nil {
 		t.log.WithError(err).Info("Failed creating a client for session")
-		t.writeError(err)
+		t.stream.writeError(err.Error())
 		return
 	}
 
 	t.log.Debug("Creating websocket stream")
 
-	// Update the read deadline upon receiving a pong message.
-	ws.SetPongHandler(func(_ string) error {
-		ws.SetReadDeadline(deadlineForInterval(t.keepAliveInterval))
-		return nil
-	})
-
 	// Start sending ping frames through websocket to client.
-	go startPingLoop(t.terminalContext, ws, t.keepAliveInterval, t.log, t.Close)
+	go startPingLoop(ctx, ws, t.keepAliveInterval, t.log, t.Close)
 
 	// Pump raw terminal in/out and audit events into the websocket.
-	go t.streamTerminal(ws, tc)
-	go t.streamEvents(tc)
-
-	// process window resizing
-	go t.handleWindowResize(resizeC)
-
-	// process file transfer requests/responses
-	go t.handleFileTransfer(fileTransferRequestC, fileTransferDecisionC)
+	go t.streamEvents(ctx, tc)
 
 	// Block until the terminal session is complete.
-	<-t.terminalContext.Done()
+	t.streamTerminal(ctx, tc)
 	t.log.Debug("Closing websocket stream")
 }
 
 // makeClient builds a *client.TeleportClient for the connection.
-func (t *TerminalHandler) makeClient(ctx context.Context, ws *websocket.Conn) (*client.TeleportClient, error) {
+func (t *TerminalHandler) makeClient(ctx context.Context, stream *TerminalStream, clientAddr string) (*client.TeleportClient, error) {
 	ctx, span := tracing.DefaultProvider().Tracer("terminal").Start(ctx, "terminal/makeClient")
 	defer span.End()
 
@@ -433,9 +419,9 @@ func (t *TerminalHandler) makeClient(ctx context.Context, ws *websocket.Conn) (*
 	clientConfig.HostLogin = t.sessionData.Login
 	clientConfig.ForwardAgent = client.ForwardAgentLocal
 	clientConfig.Namespace = apidefaults.Namespace
-	clientConfig.Stdout = t.stream
-	clientConfig.Stderr = t.stream
-	clientConfig.Stdin = t.stream
+	clientConfig.Stdout = stream
+	clientConfig.Stderr = stream
+	clientConfig.Stdin = stream
 	clientConfig.SiteName = t.sessionData.ClusterName
 	if err := clientConfig.ParseProxyHost(t.proxyHostPort); err != nil {
 		return nil, trace.BadParameter("failed to parse proxy address: %v", err)
@@ -443,7 +429,7 @@ func (t *TerminalHandler) makeClient(ctx context.Context, ws *websocket.Conn) (*
 	clientConfig.Host = t.sessionData.ServerHostname
 	clientConfig.HostPort = t.sessionData.ServerHostPort
 	clientConfig.SessionID = t.sessionData.ID.String()
-	clientConfig.ClientAddr = ws.RemoteAddr().String()
+	clientConfig.ClientAddr = clientAddr
 	clientConfig.Tracer = t.tracer
 
 	if len(t.interactiveCommand) > 0 {
@@ -459,8 +445,10 @@ func (t *TerminalHandler) makeClient(ctx context.Context, ws *websocket.Conn) (*
 	// used to update all other parties window size to that of the web client and
 	// to allow future window changes.
 	tc.OnShellCreated = func(s *tracessh.Session, c *tracessh.Client, _ io.ReadWriteCloser) (bool, error) {
-		t.sshSession = s
-		t.windowChange(ctx, &t.term)
+		t.stream.sessionCreated(s)
+		if err := s.WindowChange(ctx, t.term.H, t.term.W); err != nil {
+			t.log.Error(err)
+		}
 
 		return false, nil
 	}
@@ -614,7 +602,7 @@ func promptMFAChallenge(
 			return nil, trace.Wrap(err)
 		}
 
-		resp, err := stream.readChallenge(codec)
+		resp, err := stream.readChallengeResponse(codec)
 		return resp, trace.Wrap(err)
 	}
 }
@@ -626,9 +614,7 @@ type connectWithMFAFn = func(ctx context.Context, ws WSConn, tc *client.Teleport
 // certs and with single use certs after completing the mfa ceremony. Only one of
 // the operations will succeed, and if per session mfa will not gain access to the
 // target it will abort before prompting a user to perform the ceremony.
-func (t *sshBaseHandler) connectToHost(ctx context.Context, ws WSConn,
-	tc *client.TeleportClient, connectToNodeWithMFA connectWithMFAFn,
-) (*client.NodeClient, error) {
+func (t *sshBaseHandler) connectToHost(ctx context.Context, ws WSConn, tc *client.TeleportClient, connectToNodeWithMFA connectWithMFAFn) (*client.NodeClient, error) {
 	ctx, span := t.tracer.Start(ctx, "terminal/connectToHost")
 	defer span.End()
 
@@ -721,26 +707,40 @@ func (t *sshBaseHandler) connectToHost(ctx context.Context, ws WSConn,
 
 // streamTerminal opens a SSH connection to the remote host and streams
 // events back to the web client.
-func (t *TerminalHandler) streamTerminal(ws *websocket.Conn, tc *client.TeleportClient) {
-	ctx, span := t.tracer.Start(t.terminalContext, "terminal/streamTerminal")
+func (t *TerminalHandler) streamTerminal(ctx context.Context, tc *client.TeleportClient) {
+	ctx, span := t.tracer.Start(ctx, "terminal/streamTerminal")
 	defer span.End()
 
-	defer t.terminalCancel()
-
-	nc, err := t.connectToHost(ctx, ws, tc, t.connectToNodeWithMFA)
+	nc, err := t.connectToHost(ctx, t.stream.ws, tc, t.connectToNodeWithMFA)
 	if err != nil {
 		t.log.WithError(err).Warn("Unable to stream terminal - failure connecting to host")
-		t.writeError(err)
+		t.stream.writeError(err.Error())
 		return
 	}
 
 	defer nc.Close()
 
+	var beforeStart func(io.Writer)
+	if t.participantMode == types.SessionModeratorMode {
+		beforeStart = func(out io.Writer) {
+			nc.OnMFA = func() {
+				prompt := func(ctx context.Context, proxyAddr string, c *authproto.MFAAuthenticateChallenge) (*authproto.MFAAuthenticateResponse, error) {
+					resp, err := promptMFAChallenge(t.stream.WSStream, protobufMFACodec{})(ctx, proxyAddr, c)
+					return resp, trace.Wrap(err)
+				}
+				if err := client.RunPresenceTask(ctx, out, t.authProvider, t.sessionData.ID.String(), prompt, client.WithPresenceClock(t.clock)); err != nil {
+					t.log.WithError(err).Warn("Unable to stream terminal - failure performing presence checks")
+					return
+				}
+			}
+		}
+	}
+
 	// Establish SSH connection to the server. This function will block until
 	// either an error occurs or it completes successfully.
-	if err = nc.RunInteractiveShell(ctx, t.participantMode, nil); err != nil {
+	if err = nc.RunInteractiveShell(ctx, t.participantMode, t.tracker, beforeStart); err != nil {
 		t.log.WithError(err).Warn("Unable to stream terminal - failure running interactive shell")
-		t.writeError(err)
+		t.stream.writeError(err.Error())
 		return
 	}
 
@@ -790,7 +790,7 @@ func (t *sshBaseHandler) connectToNode(ctx context.Context, ws WSConn, tc *clien
 // host with the retrieved single use certs.
 func (t *TerminalHandler) connectToNodeWithMFA(ctx context.Context, ws WSConn, tc *client.TeleportClient, accessChecker services.AccessChecker, getAgent teleagent.Getter, signer agentless.SignerCreator) (*client.NodeClient, error) {
 	// perform mfa ceremony and retrieve new certs
-	authMethods, err := t.issueSessionMFACerts(ctx, tc, &t.stream.WSStream)
+	authMethods, err := t.issueSessionMFACerts(ctx, tc, t.stream.WSStream)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}
@@ -828,7 +828,7 @@ func (t *sshBaseHandler) connectToNodeWithMFABase(ctx context.Context, ws WSConn
 
 // streamEvents receives events over the SSH connection and forwards them to
 // the web client.
-func (t *TerminalHandler) streamEvents(tc *client.TeleportClient) {
+func (t *TerminalHandler) streamEvents(ctx context.Context, tc *client.TeleportClient) {
 	for {
 		select {
 		// Send push events that come over the events channel to the web client.
@@ -856,87 +856,12 @@ func (t *TerminalHandler) streamEvents(tc *client.TeleportClient) {
 
 		// Once the terminal stream is over (and the close envelope has been sent),
 		// close stop streaming envelopes.
-		case <-t.terminalContext.Done():
+		case <-ctx.Done():
 			return
 		}
 	}
 }
 
-// handleFileTransfer receives file transfer requests and responses and forwards them
-// to the SSH session
-func (t *TerminalHandler) handleFileTransfer(fileTransferRequestC <-chan *session.FileTransferRequestParams, fileTransferDecisionC <-chan *session.FileTransferDecisionParams) {
-	for {
-		select {
-		case <-t.terminalContext.Done():
-			return
-		case transferRequest := <-fileTransferRequestC:
-			if transferRequest == nil {
-				// channel closed
-				continue
-			}
-			if err := t.sshSession.RequestFileTransfer(t.terminalContext, tracessh.FileTransferReq{
-				Download: transferRequest.Download,
-				Location: transferRequest.Location,
-				Filename: transferRequest.Filename,
-			}); err != nil {
-				t.log.WithError(err).Error("Unable to request file transfer")
-			}
-		case transferResponse := <-fileTransferDecisionC:
-			if transferResponse == nil {
-				// channel closed
-				continue
-			}
-
-			var err error
-			if transferResponse.Approved {
-				err = t.sshSession.ApproveFileTransferRequest(t.terminalContext, transferResponse.RequestID)
-			} else {
-				err = t.sshSession.DenyFileTransferRequest(t.terminalContext, transferResponse.RequestID)
-			}
-			if err != nil {
-				t.log.WithError(err).Error("Unable to respond to file transfer request")
-			}
-		}
-
-	}
-}
-
-// handleWindowResize receives window resize events and forwards
-// them to the SSH session.
-func (t *TerminalHandler) handleWindowResize(resizeC <-chan *session.TerminalParams) {
-	for {
-		select {
-		case <-t.terminalContext.Done():
-			return
-		case params := <-resizeC:
-			// nil params indicates the channel was closed
-			if params == nil {
-				return
-			}
-			// process window change
-			t.windowChange(t.terminalContext, params)
-		}
-	}
-}
-
-// writeError displays an error in the terminal window.
-func (t *TerminalHandler) windowChange(ctx context.Context, params *session.TerminalParams) {
-	if t.sshSession == nil {
-		return
-	}
-
-	if err := t.sshSession.WindowChange(ctx, params.H, params.W); err != nil {
-		t.log.Error(err)
-	}
-}
-
-// writeError displays an error in the terminal window.
-func (t *TerminalHandler) writeError(err error) {
-	if writeErr := t.stream.writeError(err); writeErr != nil {
-		t.log.WithError(writeErr).Warnf("Unable to send error to terminal: %v", err)
-	}
-}
-
 // the defaultPort of 0 indicates that the port is
 // unknown or was not provided and should be guessed
 const defaultPort = 0
@@ -979,74 +904,60 @@ func serverHostPort(servername string) (string, int, error) {
 	return host, port, nil
 }
 
-// WithTerminalStreamEncoder overrides the default stream encoder
-func WithTerminalStreamEncoder(enc *encoding.Encoder) func(stream *TerminalStream) {
-	return func(stream *TerminalStream) {
-		stream.encoder = enc
+func NewWStream(ctx context.Context, ws WSConn, log logrus.FieldLogger, handlers map[string]WSHandlerFunc) *WSStream {
+	w := &WSStream{
+		log:        log,
+		ws:         ws,
+		encoder:    unicode.UTF8.NewEncoder(),
+		decoder:    unicode.UTF8.NewDecoder(),
+		completedC: make(chan struct{}),
+		rawC:       make(chan Envelope, 100),
+		challengeC: make(chan Envelope, 1),
+		handlers:   handlers,
 	}
-}
 
-// WithTerminalStreamDecoder overrides the default stream decoder
-func WithTerminalStreamDecoder(dec *encoding.Decoder) func(stream *TerminalStream) {
-	return func(stream *TerminalStream) {
-		stream.decoder = dec
-	}
-}
+	go w.processMessages(ctx)
 
-// WithTerminalStreamResizeHandler provides a channel to subscribe to
-// terminal resize events
-func WithTerminalStreamResizeHandler(resizeC chan<- *session.TerminalParams) func(stream *TerminalStream) {
-	return func(stream *TerminalStream) {
-		stream.resizeC = resizeC
-	}
-}
-
-// WithTerminalStreamFileTransferHandlers provides two channels, one to subscribe to new file transfer requests, and
-// one to subscribe to file transfer decision requests, such as approve/deny
-func WithTerminalStreamFileTransferHandlers(fileTransferRequestC chan<- *session.FileTransferRequestParams, fileTransferDecisionC chan<- *session.FileTransferDecisionParams) func(stream *TerminalStream) {
-	return func(stream *TerminalStream) {
-		stream.fileTransferRequestC = fileTransferRequestC
-		stream.fileTransferDecisionC = fileTransferDecisionC
-	}
-}
-
-func NewWStream(ws WSConn) *WSStream {
-	return &WSStream{
-		ws:      ws,
-		encoder: unicode.UTF8.NewEncoder(),
-		decoder: unicode.UTF8.NewDecoder(),
-	}
+	return w
 }
 
 // NewTerminalStream creates a stream that manages reading and writing
 // data over the provided [websocket.Conn]
-func NewTerminalStream(ws *websocket.Conn, opts ...func(*TerminalStream)) (*TerminalStream, error) {
-	switch {
-	case ws == nil:
-		return nil, trace.BadParameter("required parameter ws not provided")
-	}
-
+func NewTerminalStream(ctx context.Context, ws *websocket.Conn, log logrus.FieldLogger) *TerminalStream {
 	t := &TerminalStream{
-		WSStream: WSStream{
-			ws:      ws,
-			encoder: unicode.UTF8.NewEncoder(),
-			decoder: unicode.UTF8.NewDecoder(),
-		},
+		sessionReadyC: make(chan struct{}),
 	}
 
-	for _, opt := range opts {
-		opt(t)
+	handlers := map[string]WSHandlerFunc{
+		defaults.WebsocketResize:               t.handleWindowResize,
+		defaults.WebsocketFileTransferRequest:  t.handleFileTransferRequest,
+		defaults.WebsocketFileTransferDecision: t.handleFileTransferDecision,
 	}
 
-	return t, nil
+	t.WSStream = NewWStream(ctx, ws, log, handlers)
+
+	return t
 }
 
+// WSHandlerFunc specifies a handler that processes received a specific
+// [Envelope] received via a web socket.
+type WSHandlerFunc func(context.Context, Envelope)
+
+// WSStream handles web socket communication with
+// the frontend.
 type WSStream struct {
 	// encoder is used to encode UTF-8 strings.
 	encoder *encoding.Encoder
 	// decoder is used to decode UTF-8 strings.
 	decoder *encoding.Decoder
 
+	handlers map[string]WSHandlerFunc
+	// once ensures that all channels are closed at most one time.
+	once       sync.Once
+	challengeC chan Envelope
+	rawC       chan Envelope
+	completedC chan struct{}
+
 	// buffer is a buffer used to store the remaining payload data if it did not
 	// fit into the buffer provided by the callee to Read method
 	buffer []byte
@@ -1055,32 +966,208 @@ type WSStream struct {
 	mu sync.Mutex
 	// ws the connection to the UI
 	ws WSConn
+
+	// log holds the structured logger.
+	log logrus.FieldLogger
 }
 
 // TerminalStream manages the [websocket.Conn] to the web UI
 // for a terminal session.
 type TerminalStream struct {
-	WSStream
+	*WSStream
 
-	// once ensures that all channels are closed at most one time.
-	once sync.Once
-	// resizeC a channel to forward resize events so that
-	// they happen out of band and don't block reads
-	resizeC chan<- *session.TerminalParams
-	// fileTransferRequestC is a channel to facilitate requesting a file transfer
-	fileTransferRequestC chan<- *session.FileTransferRequestParams
-	// fileTransferDecisionC is a channel to facilitate responding to a file transfer
-	// with an approval or denial
-	fileTransferDecisionC chan<- *session.FileTransferDecisionParams
+	// sshSession holds the "shell" SSH channel to the node.
+	sshSession    *tracessh.Session
+	sessionReadyC chan struct{}
 }
 
 // Replace \n with \r\n so the message is correctly aligned.
 var replacer = strings.NewReplacer("\r\n", "\r\n", "\n", "\r\n")
 
 // writeError displays an error in the terminal window.
-func (t *WSStream) writeError(err error) error {
-	_, writeErr := replacer.WriteString(t, err.Error())
-	return trace.Wrap(writeErr)
+func (t *WSStream) writeError(msg string) {
+	if _, writeErr := replacer.WriteString(t, msg); writeErr != nil {
+		t.log.WithError(writeErr).Warnf("Unable to send error to terminal: %v", msg)
+	}
+}
+
+func (t *WSStream) processMessages(ctx context.Context) {
+	defer func() {
+		close(t.completedC)
+	}()
+	t.ws.SetReadLimit(teleport.MaxHTTPRequestSize)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			ty, bytes, err := t.ws.ReadMessage()
+			if err != nil {
+				if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) ||
+					websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
+					return
+				}
+
+				msg := err.Error()
+				if len(bytes) > 0 {
+					msg = string(bytes)
+				}
+				select {
+				case <-ctx.Done():
+				default:
+					t.writeError(msg)
+					return
+				}
+			}
+
+			if ty != websocket.BinaryMessage {
+				t.writeError(fmt.Sprintf("Expected binary message, got %v", ty))
+				return
+			}
+
+			var envelope Envelope
+			if err := proto.Unmarshal(bytes, &envelope); err != nil {
+				t.writeError(fmt.Sprintf("Unable to parse message payload %v", err))
+				return
+			}
+
+			switch envelope.Type {
+			case defaults.WebsocketClose:
+				return
+			case defaults.WebsocketWebauthnChallenge:
+				select {
+				case <-ctx.Done():
+					return
+				case t.challengeC <- envelope:
+				default:
+				}
+			case defaults.WebsocketRaw:
+				select {
+				case <-ctx.Done():
+					return
+				case t.rawC <- envelope:
+				default:
+				}
+			default:
+				if t.handlers == nil {
+					continue
+				}
+
+				handler, ok := t.handlers[envelope.Type]
+				if !ok {
+					t.log.Warnf("Received web socket envelope with unknown type %v", envelope.Type)
+					continue
+				}
+
+				go handler(ctx, envelope)
+			}
+		}
+	}
+}
+
+// handleWindowResize receives window resize events and forwards
+// them to the SSH session.
+func (t *TerminalStream) handleWindowResize(ctx context.Context, envelope Envelope) {
+	select {
+	case <-ctx.Done():
+		return
+	case <-t.sessionReadyC:
+	}
+
+	if t.sshSession == nil {
+		return
+	}
+
+	var e map[string]string
+	err := json.Unmarshal([]byte(envelope.Payload), &e)
+	if err != nil {
+		t.log.Warnf("Failed to parse resize payload: %v", err)
+		return
+	}
+
+	params, err := session.UnmarshalTerminalParams(e["size"])
+	if err != nil {
+		t.log.Warnf("Failed to retrieve terminal size: %v", err)
+		return
+	}
+
+	// nil params indicates the channel was closed
+	if params == nil {
+		return
+	}
+
+	if err := t.sshSession.WindowChange(ctx, params.H, params.W); err != nil {
+		t.log.Error(err)
+	}
+}
+
+func (t *TerminalStream) handleFileTransferDecision(ctx context.Context, envelope Envelope) {
+	select {
+	case <-ctx.Done():
+		return
+	case <-t.sessionReadyC:
+	}
+
+	if t.sshSession == nil {
+		return
+	}
+
+	var e utils.Fields
+	err := json.Unmarshal([]byte(envelope.Payload), &e)
+	if err != nil {
+		return
+	}
+	approved, ok := e["approved"].(bool)
+	if !ok {
+		t.log.Error("Unable to find approved status on response")
+		return
+	}
+
+	if approved {
+		err = t.sshSession.ApproveFileTransferRequest(ctx, e.GetString("requestId"))
+	} else {
+		err = t.sshSession.DenyFileTransferRequest(ctx, e.GetString("requestId"))
+	}
+	if err != nil {
+		t.log.WithError(err).Error("Unable to respond to file transfer request")
+	}
+}
+
+func (t *TerminalStream) handleFileTransferRequest(ctx context.Context, envelope Envelope) {
+	select {
+	case <-ctx.Done():
+		return
+	case <-t.sessionReadyC:
+	}
+
+	if t.sshSession == nil {
+		return
+	}
+
+	var e utils.Fields
+	err := json.Unmarshal([]byte(envelope.Payload), &e)
+	if err != nil {
+		return
+	}
+	download, ok := e["download"].(bool)
+	if !ok {
+		t.log.Error("Unable to find download param in response")
+		return
+	}
+
+	if err := t.sshSession.RequestFileTransfer(ctx, tracessh.FileTransferReq{
+		Download: download,
+		Location: e.GetString("location"),
+		Filename: e.GetString("filename"),
+	}); err != nil {
+		t.log.WithError(err).Error("Unable to request file transfer")
+	}
+}
+
+func (t *TerminalStream) sessionCreated(s *tracessh.Session) {
+	t.sshSession = s
+	close(t.sessionReadyC)
 }
 
 // writeChallenge encodes and writes the challenge to the
@@ -1097,21 +1184,28 @@ func (t *WSStream) writeChallenge(challenge *client.MFAAuthenticateChallenge, co
 	return trace.Wrap(t.ws.WriteMessage(websocket.BinaryMessage, msg))
 }
 
-// readChallenge reads and decodes the challenge response from the
+// readChallengeResponse reads and decodes the challenge response from the
 // websocket in the correct format.
-func (t *WSStream) readChallenge(codec mfaCodec) (*authproto.MFAAuthenticateResponse, error) {
-	// Read the challenge response.
-	ty, bytes, err := t.ws.ReadMessage()
-	if err != nil {
-		return nil, trace.Wrap(err)
+func (t *WSStream) readChallengeResponse(codec mfaCodec) (*authproto.MFAAuthenticateResponse, error) {
+	select {
+	case <-t.completedC:
+		return nil, io.EOF
+	case envelope := <-t.challengeC:
+		resp, err := codec.decodeResponse([]byte(envelope.Payload), defaults.WebsocketWebauthnChallenge)
+		return resp, trace.Wrap(err)
 	}
+}
 
-	if ty != websocket.BinaryMessage {
-		return nil, trace.BadParameter("expected websocket.BinaryMessage, got %v", ty)
+// readChallenge reads and decodes the challenge from the
+// websocket in the correct format.
+func (t *WSStream) readChallenge(codec mfaCodec) (*authproto.MFAAuthenticateChallenge, error) {
+	select {
+	case <-t.completedC:
+		return nil, io.EOF
+	case envelope := <-t.challengeC:
+		challenge, err := codec.decodeChallenge([]byte(envelope.Payload), defaults.WebsocketWebauthnChallenge)
+		return challenge, trace.Wrap(err)
 	}
-
-	resp, err := codec.decode(bytes, defaults.WebsocketWebauthnChallenge)
-	return resp, trace.Wrap(err)
 }
 
 // writeAuditEvent encodes and writes the audit event to the
@@ -1168,7 +1262,10 @@ func (t *WSStream) Write(data []byte) (n int, err error) {
 	return len(data), nil
 }
 
-func (t *WSStream) readMessage(out []byte) (string, []byte, int, error) {
+// Read provides data received from [defaults.WebsocketRaw] envelopes. If
+// the previous envelope was not consumed in the last read any remaining data
+// is returned prior to processing the next envelope.
+func (t *WSStream) Read(out []byte) (n int, err error) {
 	if len(t.buffer) > 0 {
 		n := copy(out, t.buffer)
 		if n == len(t.buffer) {
@@ -1176,53 +1273,18 @@ func (t *WSStream) readMessage(out []byte) (string, []byte, int, error) {
 		} else {
 			t.buffer = t.buffer[n:]
 		}
-		return "", nil, n, nil
+		return n, nil
 	}
 
-	ty, bytes, err := t.ws.ReadMessage()
-	if err != nil {
-		// if the connection has closed, we must return io.EOF in order to abort
-		// the websocket copy loop
-		if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) ||
-			websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
-			return "", nil, 0, io.EOF
+	select {
+	case <-t.completedC:
+		return 0, io.EOF
+	case envelope := <-t.rawC:
+		data, err := t.decoder.Bytes([]byte(envelope.Payload))
+		if err != nil {
+			return 0, trace.Wrap(err)
 		}
 
-		return "", nil, 0, trace.Wrap(err)
-	}
-
-	if ty != websocket.BinaryMessage {
-		return "", nil, 0, trace.BadParameter("expected binary message, got %v", ty)
-	}
-
-	var envelope Envelope
-	err = proto.Unmarshal(bytes, &envelope)
-	if err != nil {
-		return "", nil, 0, trace.Wrap(err)
-	}
-
-	var data []byte
-	data, err = t.decoder.Bytes([]byte(envelope.GetPayload()))
-	if err != nil {
-		return "", nil, 0, trace.Wrap(err)
-	}
-
-	return envelope.Type, data, 0, nil
-}
-
-// Read unwraps the envelope and either fills out the passed in bytes or
-// performs an action on the connection (sending window-change request).
-func (t *WSStream) Read(out []byte) (n int, err error) {
-	messageType, data, n, err := t.readMessage(out)
-	if err != nil || n > 0 {
-		return n, err
-	}
-
-	switch messageType {
-	// the session was closed
-	case defaults.WebsocketClose:
-		return 0, io.EOF
-	case defaults.WebsocketRaw:
 		n := copy(out, data)
 		// if payload size is greater than [out], store the remaining
 		// part in the buffer to be processed on the next Read call
@@ -1230,136 +1292,37 @@ func (t *WSStream) Read(out []byte) (n int, err error) {
 			t.buffer = data[n:]
 		}
 		return n, nil
-	default:
-		return 0, trace.BadParameter("unknown prefix type: %v", messageType)
-	}
-}
-
-func (t *TerminalStream) Read(out []byte) (n int, err error) {
-	messageType, data, n, err := t.readMessage(out)
-	if err != nil || n > 0 {
-		return n, err
-	}
-
-	switch messageType {
-	// the session was closed
-	case defaults.WebsocketClose:
-		return 0, io.EOF
-	case defaults.WebsocketRaw:
-		n := copy(out, data)
-		// if payload size is greater than [out], store the remaining
-		// part in the buffer to be processed on the next Read call
-		if len(data) > n {
-			t.buffer = data[n:]
-		}
-		return n, nil
-	case defaults.WebsocketResize:
-		if t.resizeC == nil {
-			return n, nil
-		}
-
-		var e events.EventFields
-		err := json.Unmarshal(data, &e)
-		if err != nil {
-			return 0, trace.Wrap(err)
-		}
-
-		params, err := session.UnmarshalTerminalParams(e.GetString("size"))
-		if err != nil {
-			return 0, trace.Wrap(err)
-		}
-
-		// Send the window change request in a goroutine so reads are not blocked
-		// by network connectivity issues.
-		select {
-		case t.resizeC <- params:
-		default:
-		}
-
-		return 0, nil
-	case defaults.WebsocketFileTransferDecision:
-		if t.fileTransferDecisionC == nil {
-			return n, nil
-		}
-		var e events.EventFields
-		err := json.Unmarshal(data, &e)
-		if err != nil {
-			return 0, trace.Wrap(err)
-		}
-		approved, ok := e["approved"].(bool)
-		if !ok {
-			return 0, trace.BadParameter("Unable to find approved status on response")
-		}
-		select {
-		case t.fileTransferDecisionC <- &session.FileTransferDecisionParams{
-			RequestID: e.GetString("requestId"),
-			Approved:  approved,
-		}:
-		default:
-		}
-		return 0, nil
-	case defaults.WebsocketFileTransferRequest:
-		if t.fileTransferRequestC == nil {
-			return n, nil
-		}
-		var e events.EventFields
-		err := json.Unmarshal(data, &e)
-		if err != nil {
-			return 0, trace.Wrap(err)
-		}
-		download, ok := e["download"].(bool)
-		if !ok {
-			return 0, trace.BadParameter("Unable to find download param in response")
-		}
-		select {
-		case t.fileTransferRequestC <- &session.FileTransferRequestParams{
-			Location: e.GetString("location"),
-			Download: download,
-			Filename: e.GetString("filename"),
-		}:
-		default:
-		}
-		return 0, nil
-	default:
-		return 0, trace.BadParameter("unknown prefix type: %v", messageType)
 	}
 }
 
 // Close sends a close message on the web socket and closes the web socket.
 func (t *WSStream) Close() error {
-	// Send close envelope to web terminal upon exit without an error.
-	envelope := &Envelope{
-		Version: defaults.WebsocketVersion,
-		Type:    defaults.WebsocketClose,
-	}
-	envelopeBytes, err := proto.Marshal(envelope)
-	if err != nil {
-		return trace.NewAggregate(err, t.ws.Close())
-	}
-
-	t.mu.Lock()
-	defer t.mu.Unlock()
-	return trace.NewAggregate(t.ws.WriteMessage(websocket.BinaryMessage, envelopeBytes), t.ws.Close())
-}
-
-// Close sends a close message on the web socket
-// prior to closing the web socket altogether.
-func (t *TerminalStream) Close() error {
+	var closeErr error
 	t.once.Do(func() {
-		if t.resizeC != nil {
-			close(t.resizeC)
+		defer func() {
+			<-t.completedC
+
+			close(t.rawC)
+			close(t.challengeC)
+		}()
+
+		// Send close envelope to web terminal upon exit without an error.
+		envelope := &Envelope{
+			Version: defaults.WebsocketVersion,
+			Type:    defaults.WebsocketClose,
+		}
+		envelopeBytes, err := proto.Marshal(envelope)
+		if err != nil {
+			closeErr = trace.NewAggregate(err, t.ws.Close())
+			return
 		}
 
-		if t.fileTransferRequestC != nil {
-			close(t.fileTransferRequestC)
-		}
-
-		if t.fileTransferDecisionC != nil {
-			close(t.fileTransferDecisionC)
-		}
+		t.mu.Lock()
+		defer t.mu.Unlock()
+		closeErr = trace.NewAggregate(t.ws.WriteMessage(websocket.BinaryMessage, envelopeBytes), t.ws.Close())
 	})
 
-	return t.WSStream.Close()
+	return trace.Wrap(closeErr)
 }
 
 // deadlineForInterval returns a suitable network read deadline for a given ping interval.

lib/web/terminal_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package web_test
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -28,6 +29,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/gravitational/teleport/lib/defaults"
+	"github.com/gravitational/teleport/lib/utils"
 	"github.com/gravitational/teleport/lib/web"
 )
 
@@ -62,12 +64,11 @@ func TestTerminalReadFromClosedConn(t *testing.T) {
 	require.NoError(t, err)
 	defer resp.Body.Close()
 
-	stream, err := web.NewTerminalStream(conn)
-	require.NoError(t, err)
+	stream := web.NewTerminalStream(context.Background(), conn, utils.NewLoggerForTests())
 
 	// close the stream before we attempt to read from it,
 	// this will produce a net.ErrClosed error on the read
-	stream.Close()
+	require.NoError(t, stream.Close())
 
 	_, err = io.Copy(io.Discard, stream)
 	require.NoError(t, err)

tool/tsh/common/tsh.go

@@ -3223,7 +3223,7 @@ func onJoin(cf *CLIConf) error {
 		return trace.BadParameter("'%v' is not a valid session ID (must be GUID)", cf.SessionID)
 	}
 	err = client.RetryWithRelogin(cf.Context, tc, func() error {
-		return tc.Join(context.TODO(), types.SessionParticipantMode(cf.JoinMode), cf.Namespace, *sid, nil)
+		return tc.Join(cf.Context, types.SessionParticipantMode(cf.JoinMode), cf.Namespace, *sid, nil)
 	})
 	if err != nil {
 		return trace.Wrap(err)

web/packages/teleport/src/lib/term/protobuf.js

@@ -56,6 +56,7 @@ export const messageFields = {
       data: MessageTypeEnum.RAW.charCodeAt(0),
       event: MessageTypeEnum.AUDIT.charCodeAt(0),
       close: MessageTypeEnum.SESSION_END.charCodeAt(0),
+      challengeResponse: MessageTypeEnum.WEBAUTHN_CHALLENGE.charCodeAt(0),
     },
   },
 };
@@ -73,6 +74,10 @@ export class Protobuf {
     return this.encode(messageFields.type.values.resize, message);
   }
 
+  encodeChallengeResponse(message) {
+    return this.encode(messageFields.type.values.challengeResponse, message);
+  }
+
   encodeFileTransferRequest(message) {
     return this.encode(messageFields.type.values.fileTransferRequest, message);
   }

web/packages/teleport/src/lib/term/tty.ts

@@ -78,7 +78,9 @@ class Tty extends EventEmitterWebAuthnSender {
   }
 
   sendWebAuthn(data: WebauthnAssertionResponse) {
-    this.send(JSON.stringify(data));
+    const encoded = this._proto.encodeChallengeResponse(JSON.stringify(data));
+    const bytearray = new Uint8Array(encoded);
+    this.socket.send(bytearray);
   }
 
   _sendFileTransferRequest(message: string) {