diff --git a/api/types/session_tracker.go b/api/types/session_tracker.go index 8c2866c1782..b96428a601c 100644 --- a/api/types/session_tracker.go +++ b/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. diff --git a/api/types/session_tracker_test.go b/api/types/session_tracker_test.go new file mode 100644 index 00000000000..59cbb7fbd84 --- /dev/null +++ b/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) + } +} diff --git a/integration/helpers/instance.go b/integration/helpers/instance.go index 056e938180a..e7ac22bc041 100644 --- a/integration/helpers/instance.go +++ b/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 { diff --git a/integration/integration_test.go b/integration/integration_test.go index b8d32d1911b..1b31ff8a6f2 100644 --- a/integration/integration_test.go +++ b/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.") + } +} diff --git a/lib/client/api.go b/lib/client/api.go index 6bc3177952c..54808d5a1f8 100644 --- a/lib/client/api.go +++ b/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 != "" { diff --git a/lib/client/client.go b/lib/client/client.go index d56037f5b5b..e19c8b525eb 100644 --- a/lib/client/client.go +++ b/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. diff --git a/lib/client/kubesession.go b/lib/client/kubesession.go index f31d99add02..b2ab3e0a367 100644 --- a/lib/client/kubesession.go +++ b/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 diff --git a/lib/client/presence.go b/lib/client/presence.go index 41705c084af..df6fdfcfb8a 100644 --- a/lib/client/presence.go +++ b/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 -} diff --git a/lib/client/session.go b/lib/client/session.go index cbb2341c072..69173486a7b 100644 --- a/lib/client/session.go +++ b/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) diff --git a/lib/services/local/sessiontracker.go b/lib/services/local/sessiontracker.go index c2165eb8a2f..07fbe6f6e71 100644 --- a/lib/services/local/sessiontracker.go +++ b/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) } diff --git a/lib/srv/authhandlers.go b/lib/srv/authhandlers.go index 11846bf41b1..694ab0908fb 100644 --- a/lib/srv/authhandlers.go +++ b/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) diff --git a/lib/srv/sess.go b/lib/srv/sess.go index ca04ecf6736..4a48fc808fd 100644 --- a/lib/srv/sess.go +++ b/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(), diff --git a/lib/web/apiserver.go b/lib/web/apiserver.go index ee2cc88a24b..946a56d2538 100644 --- a/lib/web/apiserver.go +++ b/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 { diff --git a/lib/web/apiserver_test.go b/lib/web/apiserver_test.go index 43e370b86a3..fb50be86d6b 100644 --- a/lib/web/apiserver_test.go +++ b/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)) + +} diff --git a/lib/web/command.go b/lib/web/command.go index 5e758f4f97f..fd38c44ea29 100644 --- a/lib/web/command.go +++ b/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. diff --git a/lib/web/command_test.go b/lib/web/command_test.go index 7eb5860de1e..a21e41a82a3 100644 --- a/lib/web/command_test.go +++ b/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")) } diff --git a/lib/web/command_utils.go b/lib/web/command_utils.go index 8da217940c6..cc3a0835cc9 100644 --- a/lib/web/command_utils.go +++ b/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) diff --git a/lib/web/desktop.go b/lib/web/desktop.go index 69afce3bc3e..3f47aa6cbe5 100644 --- a/lib/web/desktop.go +++ b/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) } diff --git a/lib/web/fuzz_test.go b/lib/web/fuzz_test.go index 8448abcfca2..f421f6faeb2 100644 --- a/lib/web/fuzz_test.go +++ b/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, "") }) }) } diff --git a/lib/web/mfa_codec.go b/lib/web/mfa_codec.go index 589fe1cf560..5222a2c1623 100644 --- a/lib/web/mfa_codec.go +++ b/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 +} diff --git a/lib/web/terminal.go b/lib/web/terminal.go index d5a17f2ff39..09d1c3f96b9 100644 --- a/lib/web/terminal.go +++ b/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. diff --git a/lib/web/terminal_test.go b/lib/web/terminal_test.go index 816a07e5017..884613d0a04 100644 --- a/lib/web/terminal_test.go +++ b/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) diff --git a/tool/tsh/common/tsh.go b/tool/tsh/common/tsh.go index 5ccdb0271af..5de728d1179 100644 --- a/tool/tsh/common/tsh.go +++ b/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) diff --git a/web/packages/teleport/src/lib/term/protobuf.js b/web/packages/teleport/src/lib/term/protobuf.js index 96f67ef6cd4..a259070ae78 100644 --- a/web/packages/teleport/src/lib/term/protobuf.js +++ b/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); } diff --git a/web/packages/teleport/src/lib/term/tty.ts b/web/packages/teleport/src/lib/term/tty.ts index 602cba36932..852898ef020 100644 --- a/web/packages/teleport/src/lib/term/tty.ts +++ b/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) {