Merge branch 'release/2.9'

api/http/handler/websocket/pod.go

@@ -2,12 +2,13 @@ package websocket
 
 import (
 	"fmt"
-	"github.com/portainer/portainer/api/http/security"
 	"io"
 	"log"
 	"net/http"
 	"strings"
 
+	"github.com/portainer/portainer/api/http/security"
+
 	"github.com/gorilla/websocket"
 	httperror "github.com/portainer/libhttp/error"
 	"github.com/portainer/libhttp/request"
@@ -23,12 +24,12 @@ import (
 // @tags websocket
 // @accept json
 // @produce json
-// @param endpointId query int true "endpoint ID of the endpoint where the resource is located"
+// @param endpointId query int true "environment(endpoint) ID of the environment(endpoint) where the resource is located"
 // @param namespace query string true "namespace where the container is located"
 // @param podName query string true "name of the pod containing the container"
 // @param containerName query string true "name of the container"
 // @param command query string true "command to execute in the container"
-// @param token query string true "JWT token used for authentication against this endpoint"
+// @param token query string true "JWT token used for authentication against this environment(endpoint)"
 // @success 200
 // @failure 400
 // @failure 403
@@ -63,14 +64,19 @@ func (handler *Handler) websocketPodExec(w http.ResponseWriter, r *http.Request)
 
 	endpoint, err := handler.DataStore.Endpoint().Endpoint(portainer.EndpointID(endpointID))
 	if err == bolterrors.ErrObjectNotFound {
-		return &httperror.HandlerError{http.StatusNotFound, "Unable to find the endpoint associated to the stack inside the database", err}
+		return &httperror.HandlerError{http.StatusNotFound, "Unable to find the environment associated to the stack inside the database", err}
 	} else if err != nil {
-		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to find the endpoint associated to the stack inside the database", err}
+		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to find the environment associated to the stack inside the database", err}
 	}
 
 	err = handler.requestBouncer.AuthorizedEndpointOperation(r, endpoint)
 	if err != nil {
-		return &httperror.HandlerError{http.StatusForbidden, "Permission denied to access endpoint", err}
+		return &httperror.HandlerError{http.StatusForbidden, "Permission denied to access environment", err}
+	}
+
+	serviceAccountToken, isAdminToken, err := handler.getToken(r, endpoint, false)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to get user service account token", err}
 	}
 
 	token, useAdminToken, err := handler.getToken(r, endpoint, false)
@@ -80,7 +86,7 @@ func (handler *Handler) websocketPodExec(w http.ResponseWriter, r *http.Request)
 
 	params := &webSocketRequestParams{
 		endpoint: endpoint,
-		token:    token,
+		token:    serviceAccountToken,
 	}
 
 	r.Header.Del("Origin")
@@ -99,6 +105,28 @@ func (handler *Handler) websocketPodExec(w http.ResponseWriter, r *http.Request)
 		return nil
 	}
 
+	cli, err := handler.KubernetesClientFactory.GetKubeClient(endpoint)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to create Kubernetes client", err}
+	}
+
+	handlerErr := handler.hijackPodExecStartOperation(w, r, cli, serviceAccountToken, isAdminToken, endpoint, namespace, podName, containerName, command)
+	if handlerErr != nil {
+		return handlerErr
+	}
+
+	return nil
+}
+
+func (handler *Handler) hijackPodExecStartOperation(
+	w http.ResponseWriter,
+	r *http.Request,
+	cli portainer.KubeClient,
+	serviceAccountToken string,
+	isAdminToken bool,
+	endpoint *portainer.Endpoint,
+	namespace, podName, containerName, command string,
+) *httperror.HandlerError {
 	commandArray := strings.Split(command, " ")
 
 	websocketConn, err := handler.connectionUpgrader.Upgrade(w, r, nil)
@@ -112,26 +140,59 @@ func (handler *Handler) websocketPodExec(w http.ResponseWriter, r *http.Request)
 	stdoutReader, stdoutWriter := io.Pipe()
 	defer stdoutWriter.Close()
 
+	// errorChan is used to propagate errors from the go routines to the caller.
 	errorChan := make(chan error, 1)
 	go streamFromWebsocketToWriter(websocketConn, stdinWriter, errorChan)
 	go streamFromReaderToWebsocket(websocketConn, stdoutReader, errorChan)
 
-	cli, err := handler.KubernetesClientFactory.GetKubeClient(endpoint)
-	if err != nil {
-		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to create Kubernetes client", err}
-	}
-
-	err = cli.StartExecProcess(token, useAdminToken, namespace, podName, containerName, commandArray, stdinReader, stdoutWriter)
-	if err != nil {
-		return &httperror.HandlerError{http.StatusInternalServerError, "Unable to start exec process inside container", err}
-	}
+	// StartExecProcess is a blocking operation which streams IO to/from pod;
+	// this must execute in asynchronously, since the websocketConn could return errors (e.g. client disconnects) before
+	// the blocking operation is completed.
+	go cli.StartExecProcess(serviceAccountToken, isAdminToken, namespace, podName, containerName, commandArray, stdinReader, stdoutWriter, errorChan)
 
 	err = <-errorChan
+
+	// websocket client successfully disconnected
 	if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
 		log.Printf("websocket error: %s \n", err.Error())
+		return nil
 	}
 
-	return nil
+	return &httperror.HandlerError{http.StatusInternalServerError, "Unable to start exec process inside container", err}
+}
+
+func (handler *Handler) getToken(request *http.Request, endpoint *portainer.Endpoint, setLocalAdminToken bool) (string, bool, error) {
+	tokenData, err := security.RetrieveTokenData(request)
+	if err != nil {
+		return "", false, err
+	}
+
+	kubecli, err := handler.KubernetesClientFactory.GetKubeClient(endpoint)
+	if err != nil {
+		return "", false, err
+	}
+
+	tokenCache := handler.kubernetesTokenCacheManager.GetOrCreateTokenCache(int(endpoint.ID))
+
+	tokenManager, err := kubernetes.NewTokenManager(kubecli, handler.DataStore, tokenCache, setLocalAdminToken)
+	if err != nil {
+		return "", false, err
+	}
+
+	if tokenData.Role == portainer.AdministratorRole {
+		return tokenManager.GetAdminServiceAccountToken(), true, nil
+	}
+
+	token, err := tokenManager.GetUserServiceAccountToken(int(tokenData.ID), endpoint.ID)
+	if err != nil {
+		return "", false, err
+	}
+
+	if token == "" {
+		return "", false, fmt.Errorf("can not get a valid user service account token")
+	}
+
+	return token, false, nil
 }
 
 func (handler *Handler) getToken(request *http.Request, endpoint *portainer.Endpoint, setLocalAdminToken bool) (string, bool, error) {