Join Kubernetes sessions in the web ui

Author: rosstimothy

integration/kube_integration_test.go

@@ -2098,7 +2098,7 @@ func kubeJoin(kubeConfig kube.ProxyConfig, tc *client.TeleportClient, meta types
 		return nil, trace.Wrap(err)
 	}
 
-	sess, err := client.NewKubeSession(context.TODO(), tc, meta, tc.KubeProxyAddr, "", mode, tlsConfig)
+	sess, err := client.NewKubeSession(context.TODO(), tc, meta, "", mode, tlsConfig)
 	if err != nil {
 		return nil, trace.Wrap(err)
 	}

lib/client/kubesession.go

@@ -49,16 +49,16 @@ type KubeSession struct {
 }
 
 // NewKubeSession joins a live kubernetes session.
-func NewKubeSession(ctx context.Context, tc *TeleportClient, meta types.SessionTracker, kubeAddr string, tlsServer string, mode types.SessionParticipantMode, tlsConfig *tls.Config) (*KubeSession, error) {
+func NewKubeSession(ctx context.Context, tc *TeleportClient, meta types.SessionTracker, tlsServer string, mode types.SessionParticipantMode, tlsConfig *tls.Config) (*KubeSession, error) {
 	ctx, cancel := context.WithCancel(ctx)
-	joinEndpoint := "wss://" + kubeAddr + "/api/v1/teleport/join/" + meta.GetSessionID()
+	joinEndpoint := "wss://" + tc.KubeProxyAddr + "/api/v1/teleport/join/" + meta.GetSessionID()
 
 	if tlsServer != "" {
 		tlsConfig.ServerName = tlsServer
 	}
 
 	dialer := &websocket.Dialer{
-		NetDialContext:  kubeSessionNetDialer(ctx, tc, kubeAddr).DialContext,
+		NetDialContext:  kubeSessionNetDialer(ctx, tc).DialContext,
 		TLSClientConfig: tlsConfig,
 	}
 
@@ -93,6 +93,10 @@ func NewKubeSession(ctx context.Context, tc *TeleportClient, meta types.SessionT
 		return nil, trace.Wrap(err)
 	}
 
+	context.AfterFunc(ctx, func() {
+		_ = stream.Close()
+	})
+
 	term, err := terminal.New(tc.Stdin, tc.Stdout, tc.Stderr)
 	if err != nil {
 		cancel()
@@ -108,26 +112,25 @@ func NewKubeSession(ctx context.Context, tc *TeleportClient, meta types.SessionT
 	stdout := utils.NewSyncWriter(term.Stdout())
 
 	go handleOutgoingResizeEvents(ctx, stream, term)
-	go handleIncomingResizeEvents(stream, term)
+	go handleIncomingResizeEvents(ctx, stream, term)
 
 	s := &KubeSession{stream, term, ctx, cancel, meta, sync.WaitGroup{}}
-	err = s.handleMFA(ctx, tc, mode, stdout)
-	if err != nil {
+	if err := s.handleMFA(ctx, tc, mode, stdout); err != nil {
 		return nil, trace.Wrap(err)
 	}
 
-	s.pipeInOut(stdout, tc.EnableEscapeSequences, mode)
+	s.pipeInOut(ctx, stdout, tc.EnableEscapeSequences, mode)
 	return s, nil
 }
 
-func kubeSessionNetDialer(ctx context.Context, tc *TeleportClient, kubeAddr string) client.ContextDialer {
+func kubeSessionNetDialer(ctx context.Context, tc *TeleportClient) client.ContextDialer {
 	dialOpts := []client.DialOption{
 		client.WithInsecureSkipVerify(tc.InsecureSkipVerify),
 	}
 
 	// Add options for ALPN connection upgrade only if kube is served at Proxy
 	// web address.
-	if tc.WebProxyAddr == kubeAddr && tc.TLSRoutingConnUpgradeRequired {
+	if tc.WebProxyAddr == tc.KubeProxyAddr && tc.TLSRoutingConnUpgradeRequired {
 		dialOpts = append(dialOpts,
 			client.WithALPNConnUpgrade(tc.TLSRoutingConnUpgradeRequired),
 			client.WithALPNConnUpgradePing(true), // Use Ping protocol for long-lived connections.
@@ -157,27 +160,30 @@ func handleOutgoingResizeEvents(ctx context.Context, stream *streamproto.Session
 	}
 }
 
-func handleIncomingResizeEvents(stream *streamproto.SessionStream, term *terminal.Terminal) {
+func handleIncomingResizeEvents(ctx context.Context, stream *streamproto.SessionStream, term *terminal.Terminal) {
 	events := term.Subscribe()
 
 	for {
-		event, more := <-events
-		_, ok := event.(terminal.ResizeEvent)
-		if ok {
-			w, h, err := term.Size()
-			if err != nil {
-				fmt.Printf("Error attempting to fetch terminal size: %v\n\r", err)
-			}
+		select {
+		case <-ctx.Done():
+			return
+		case event, more := <-events:
+			_, ok := event.(terminal.ResizeEvent)
+			if ok {
+				w, h, err := term.Size()
+				if err != nil {
+					fmt.Printf("Error attempting to fetch terminal size: %v\n\r", err)
+				}
 
-			size := remotecommand.TerminalSize{Width: uint16(w), Height: uint16(h)}
-			err = stream.Resize(&size)
-			if err != nil {
-				fmt.Printf("Error attempting to resize terminal: %v\n\r", err)
+				size := remotecommand.TerminalSize{Width: uint16(w), Height: uint16(h)}
+				if err := stream.Resize(&size); err != nil {
+					fmt.Printf("Error attempting to resize terminal: %v\n\r", err)
+				}
 			}
-		}
 
-		if !more {
-			break
+			if !more {
+				return
+			}
 		}
 	}
 }
@@ -205,14 +211,15 @@ func (s *KubeSession) handleMFA(ctx context.Context, tc *TeleportClient, mode ty
 }
 
 // pipeInOut starts background tasks that copy input to and from the terminal.
-func (s *KubeSession) pipeInOut(stdout io.Writer, enableEscapeSequences bool, mode types.SessionParticipantMode) {
+func (s *KubeSession) pipeInOut(ctx context.Context, stdout io.Writer, enableEscapeSequences bool, mode types.SessionParticipantMode) {
 	// wait for the session to copy everything
 	s.wg.Add(1)
 	go func() {
-		defer s.wg.Done()
-		defer s.cancel()
-		_, err := io.Copy(stdout, s.stream)
-		if err != nil {
+		defer func() {
+			s.wg.Done()
+			s.cancel()
+		}()
+		if _, err := io.Copy(stdout, s.stream); err != nil {
 			fmt.Printf("Error while reading remote stream: %v\n\r", err.Error())
 		}
 	}()
@@ -225,9 +232,8 @@ func (s *KubeSession) pipeInOut(stdout io.Writer, enableEscapeSequences bool, mo
 			handlePeerControls(s.term, enableEscapeSequences, s.stream)
 		default:
 			handleNonPeerControls(mode, s.term, func() {
-				err := s.stream.ForceTerminate()
-				if err != nil {
-					log.DebugContext(context.Background(), "Error sending force termination request", "error", err)
+				if err := s.stream.ForceTerminate(); err != nil {
+					log.DebugContext(ctx, "Error sending force termination request", "error", err)
 					fmt.Print("\n\rError while sending force termination request\n\r")
 				}
 			})

lib/web/apiserver.go

@@ -3724,6 +3724,10 @@ func (h *Handler) setDefaultConnectorHandle(w http.ResponseWriter, r *http.Reque
 type podConnectParams struct {
 	// Term is the initial PTY size.
 	Term session.TerminalParams `json:"term"`
+	// SessionID is a Teleport session ID to join as.
+	SessionID session.ID `json:"sid"`
+	// ParticipantMode is the mode that determines what you can do when you join an active session.
+	ParticipantMode types.SessionParticipantMode `json:"mode"`
 }
 
 func (h *Handler) podConnect(
@@ -3743,6 +3747,20 @@ func (h *Handler) podConnect(
 		return nil, trace.Wrap(err)
 	}
 
+	// If a session is provided, then join an existing session
+	// instead of creating a new one.
+	if !params.SessionID.IsZero() {
+		return nil, trace.Wrap(h.joinKubernetesSession(
+			r.Context(),
+			params.SessionID.String(),
+			params.ParticipantMode,
+			sctx,
+			site,
+			ws,
+		))
+	}
+
+	// Wait for the user to supply the pod information.
 	execReq, err := readPodExecRequestFromWS(ws)
 	if err != nil {
 		if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || terminal.IsOKWebsocketCloseError(trace.Unwrap(err)) {
@@ -3761,26 +3779,11 @@ func (h *Handler) podConnect(
 		return nil, trace.Wrap(err)
 	}
 
-	clt, err := sctx.GetUserClient(r.Context(), site)
-	if err != nil {
-		return nil, trace.Wrap(err)
-	}
-
-	clusterName := site.GetName()
-
-	accessChecker, err := sctx.GetUserAccessChecker()
-	if err != nil {
-		return session.Session{}, trace.Wrap(err)
-	}
-	policySets := accessChecker.SessionPolicySets()
-	accessEvaluator := auth.NewSessionAccessEvaluator(policySets, types.KubernetesSessionKind, sctx.GetUser())
-
 	sess := session.Session{
 		Kind:                  types.KubernetesSessionKind,
 		Login:                 "root",
-		ClusterName:           clusterName,
+		ClusterName:           site.GetName(),
 		KubernetesClusterName: execReq.KubeCluster,
-		Moderated:             accessEvaluator.IsModerated(),
 		ID:                    session.NewID(),
 		Created:               h.clock.Now().UTC(),
 		LastActive:            h.clock.Now().UTC(),
@@ -3807,8 +3810,6 @@ func (h *Handler) podConnect(
 		return nil, trace.Wrap(err)
 	}
 
-	keepAliveInterval := netConfig.GetKeepAliveInterval()
-
 	serverAddr, tlsServerName, err := h.getKubeExecClusterData(netConfig)
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -3822,13 +3823,18 @@ func (h *Handler) podConnect(
 		return nil, trace.Wrap(err)
 	}
 
-	ph := podHandler{
+	clt, err := sctx.GetUserClient(r.Context(), site)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	ph := podExecHandler{
 		req:                 execReq,
 		sess:                sess,
 		sctx:                sctx,
 		teleportCluster:     site.GetName(),
 		ws:                  ws,
-		keepAliveInterval:   keepAliveInterval,
+		keepAliveInterval:   netConfig.GetKeepAliveInterval(),
 		logger:              h.logger.With(teleport.ComponentKey, "pod"),
 		userClient:          clt,
 		localCA:             hostCA,