#592 feat(container-details): split websocket backend code into more files and add attach handler

api/http/handler/websocket/attach.go

@@ -0,0 +1,125 @@
+package websocket
+
+import (
+	"bytes"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"time"
+
+	"github.com/asaskevich/govalidator"
+	"github.com/gorilla/websocket"
+	httperror "github.com/portainer/libhttp/error"
+	"github.com/portainer/libhttp/request"
+	"github.com/portainer/portainer/api"
+	"github.com/portainer/portainer/api/http/handler/websocket/netdial"
+)
+
+
+// websocketAttach handles GET requests on /websocket/attach?id=<attachID>&endpointId=<endpointID>&nodeName=<nodeName>&token=<token>
+// If the nodeName query parameter is present, the request will be proxied to the underlying agent endpoint.
+// If the nodeName query parameter is not specified, the request will be upgraded to the websocket protocol and
+// an AttachStart operation HTTP request will be created and hijacked.
+// Authentication and access is controled via the mandatory token query parameter.
+func (handler *Handler) websocketAttach(w http.ResponseWriter, r *http.Request) *httperror.HandlerError {
+	attachID, err := request.RetrieveQueryParameter(r, "id", false)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusBadRequest, "Invalid query parameter: id", err}
+	}
+	if !govalidator.IsHexadecimal(attachID) {
+		return &httperror.HandlerError{http.StatusBadRequest, "Invalid query parameter: id (must be hexadecimal identifier)", err}
+	}
+
+	endpointID, err := request.RetrieveNumericQueryParameter(r, "endpointId", false)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusBadRequest, "Invalid query parameter: endpointId", err}
+	}
+
+	endpoint, err := handler.EndpointService.Endpoint(portainer.EndpointID(endpointID))
+	if err == portainer.ErrObjectNotFound {
+		return &httperror.HandlerError{http.StatusNotFound, "Unable to find the endpoint 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}
+	}
+
+	err = handler.requestBouncer.EndpointAccess(r, endpoint)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusForbidden, "Permission denied to access endpoint", portainer.ErrEndpointAccessDenied}
+	}
+
+	params := &webSocketRequestParams{
+		endpoint: endpoint,
+		ID:   attachID,
+		nodeName: r.FormValue("nodeName"),
+	}
+
+	err = handler.handleAttachRequest(w, r, params)
+	if err != nil {
+		return &httperror.HandlerError{http.StatusInternalServerError, "An error occured during websocket attach operation", err}
+	}
+
+	return nil
+}
+
+func (handler *Handler) handleAttachRequest(w http.ResponseWriter, r *http.Request, params *webSocketRequestParams) error {
+
+	r.Header.Del("Origin")
+
+	if params.nodeName != "" || params.endpoint.Type == portainer.AgentOnDockerEnvironment {
+		return handler.proxyWebsocketRequest(w, r, params)
+	}
+
+	websocketConn, err := handler.connectionUpgrader.Upgrade(w, r, nil)
+	if err != nil {
+		return err
+	}
+	defer websocketConn.Close()
+
+	return hijackAttachStartOperation(websocketConn, params.endpoint, params.ID)
+}
+
+func hijackAttachStartOperation(websocketConn *websocket.Conn, endpoint *portainer.Endpoint, attachID string) error {
+	dial, err := netdial.InitDial(endpoint)
+	if err != nil {
+		return err
+	}
+
+	// When we set up a TCP connection for hijack, there could be long periods
+	// of inactivity (a long running command with no output) that in certain
+	// network setups may cause ECONNTIMEOUT, leaving the client in an unknown
+	// state. Setting TCP KeepAlive on the socket connection will prohibit
+	// ECONNTIMEOUT unless the socket connection truly is broken
+	if tcpConn, ok := dial.(*net.TCPConn); ok {
+		tcpConn.SetKeepAlive(true)
+		tcpConn.SetKeepAlivePeriod(30 * time.Second)
+	}
+
+	httpConn := httputil.NewClientConn(dial, nil)
+	defer httpConn.Close()
+
+	attachStartRequest, err := createAttachStartRequest(attachID)
+	if err != nil {
+		return err
+	}
+
+	err = hijackRequest(websocketConn, httpConn, attachStartRequest)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func createAttachStartRequest(attachID string) (*http.Request, error) {
+
+	request, err := http.NewRequest("POST", "/containers/"+attachID+"/attach?stdin=1&stdout=1&stderr=1&stream=1", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	request.Header.Set("Content-Type", "application/json")
+	request.Header.Set("Connection", "Upgrade")
+	request.Header.Set("Upgrade", "tcp")
+
+	return request, nil
+}

api/http/handler/websocket/exec.go

@@ -1,31 +1,21 @@
 package websocket
 
 import (
-	"bufio"
 	"bytes"
-	"crypto/tls"
 	"encoding/json"
-	"fmt"
 	"net"
 	"net/http"
 	"net/http/httputil"
-	"net/url"
 	"time"
 
 	"github.com/asaskevich/govalidator"
 	"github.com/gorilla/websocket"
-	"github.com/koding/websocketproxy"
 	httperror "github.com/portainer/libhttp/error"
 	"github.com/portainer/libhttp/request"
 	"github.com/portainer/portainer/api"
-	"github.com/portainer/portainer/api/crypto"
+	"github.com/portainer/portainer/api/http/handler/websocket/netdial"
 )
 
-type webSocketExecRequestParams struct {
-	execID   string
-	nodeName string
-	endpoint *portainer.Endpoint
-}
 
 type execStartOperationPayload struct {
 	Tty    bool
@@ -63,13 +53,13 @@ func (handler *Handler) websocketExec(w http.ResponseWriter, r *http.Request) *h
 		return &httperror.HandlerError{http.StatusForbidden, "Permission denied to access endpoint", portainer.ErrEndpointAccessDenied}
 	}
 
-	params := &webSocketExecRequestParams{
+	params := &webSocketRequestParams{
 		endpoint: endpoint,
-		execID:   execID,
+		ID:   execID,
 		nodeName: r.FormValue("nodeName"),
 	}
 
-	err = handler.handleRequest(w, r, params)
+	err = handler.handleExecRequest(w, r, params)
 	if err != nil {
 		return &httperror.HandlerError{http.StatusInternalServerError, "An error occured during websocket exec operation", err}
 	}
@@ -77,7 +67,7 @@ func (handler *Handler) websocketExec(w http.ResponseWriter, r *http.Request) *h
 	return nil
 }
 
-func (handler *Handler) handleRequest(w http.ResponseWriter, r *http.Request, params *webSocketExecRequestParams) error {
+func (handler *Handler) handleExecRequest(w http.ResponseWriter, r *http.Request, params *webSocketRequestParams) error {
 	r.Header.Del("Origin")
 
 	if params.nodeName != "" || params.endpoint.Type == portainer.AgentOnDockerEnvironment {
@@ -90,45 +80,11 @@ func (handler *Handler) handleRequest(w http.ResponseWriter, r *http.Request, pa
 	}
 	defer websocketConn.Close()
 
-	return hijackExecStartOperation(websocketConn, params.endpoint, params.execID)
+	return hijackExecStartOperation(websocketConn, params.endpoint, params.ID)
-}
-
-func (handler *Handler) proxyWebsocketRequest(w http.ResponseWriter, r *http.Request, params *webSocketExecRequestParams) error {
-	agentURL, err := url.Parse(params.endpoint.URL)
-	if err != nil {
-		return err
-	}
-
-	agentURL.Scheme = "ws"
-	proxy := websocketproxy.NewProxy(agentURL)
-
-	if params.endpoint.TLSConfig.TLS || params.endpoint.TLSConfig.TLSSkipVerify {
-		agentURL.Scheme = "wss"
-		proxy.Dialer = &websocket.Dialer{
-			TLSClientConfig: &tls.Config{
-				InsecureSkipVerify: params.endpoint.TLSConfig.TLSSkipVerify,
-			},
-		}
-	}
-
-	signature, err := handler.SignatureService.CreateSignature(portainer.PortainerAgentSignatureMessage)
-	if err != nil {
-		return err
-	}
-
-	proxy.Director = func(incoming *http.Request, out http.Header) {
-		out.Set(portainer.PortainerAgentPublicKeyHeader, handler.SignatureService.EncodedPublicKey())
-		out.Set(portainer.PortainerAgentSignatureHeader, signature)
-		out.Set(portainer.PortainerAgentTargetHeader, params.nodeName)
-	}
-
-	proxy.ServeHTTP(w, r)
-
-	return nil
 }
 
 func hijackExecStartOperation(websocketConn *websocket.Conn, endpoint *portainer.Endpoint, execID string) error {
-	dial, err := initDial(endpoint)
+	dial, err := netdial.InitDial(endpoint)
 	if err != nil {
 		return err
 	}
@@ -159,30 +115,6 @@ func hijackExecStartOperation(websocketConn *websocket.Conn, endpoint *portainer
 	return nil
 }
 
-func initDial(endpoint *portainer.Endpoint) (net.Conn, error) {
-	url, err := url.Parse(endpoint.URL)
-	if err != nil {
-		return nil, err
-	}
-
-	host := url.Host
-
-	if url.Scheme == "unix" || url.Scheme == "npipe" {
-		host = url.Path
-	}
-
-	if endpoint.TLSConfig.TLS {
-		tlsConfig, err := crypto.CreateTLSConfigurationFromDisk(endpoint.TLSConfig.TLSCACertPath, endpoint.TLSConfig.TLSCertPath, endpoint.TLSConfig.TLSKeyPath, endpoint.TLSConfig.TLSSkipVerify)
-		if err != nil {
-			return nil, err
-		}
-
-		return tls.Dial(url.Scheme, host, tlsConfig)
-	}
-
-	return createDial(url.Scheme, host)
-}
-
 func createExecStartRequest(execID string) (*http.Request, error) {
 	execStartOperationPayload := &execStartOperationPayload{
 		Tty:    true,
@@ -206,64 +138,3 @@ func createExecStartRequest(execID string) (*http.Request, error) {
 
 	return request, nil
 }
-
-func hijackRequest(websocketConn *websocket.Conn, httpConn *httputil.ClientConn, request *http.Request) error {
-	// Server hijacks the connection, error 'connection closed' expected
-	resp, err := httpConn.Do(request)
-	if err != httputil.ErrPersistEOF {
-		if err != nil {
-			return err
-		}
-		if resp.StatusCode != http.StatusSwitchingProtocols {
-			resp.Body.Close()
-			return fmt.Errorf("unable to upgrade to tcp, received %d", resp.StatusCode)
-		}
-	}
-
-	tcpConn, brw := httpConn.Hijack()
-	defer tcpConn.Close()
-
-	errorChan := make(chan error, 1)
-	go streamFromTCPConnToWebsocketConn(websocketConn, brw, errorChan)
-	go streamFromWebsocketConnToTCPConn(websocketConn, tcpConn, errorChan)
-
-	err = <-errorChan
-	if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
-		return err
-	}
-
-	return nil
-}
-
-func streamFromWebsocketConnToTCPConn(websocketConn *websocket.Conn, tcpConn net.Conn, errorChan chan error) {
-	for {
-		_, in, err := websocketConn.ReadMessage()
-		if err != nil {
-			errorChan <- err
-			break
-		}
-
-		_, err = tcpConn.Write(in)
-		if err != nil {
-			errorChan <- err
-			break
-		}
-	}
-}
-
-func streamFromTCPConnToWebsocketConn(websocketConn *websocket.Conn, br *bufio.Reader, errorChan chan error) {
-	for {
-		out := make([]byte, 2048)
-		_, err := br.Read(out)
-		if err != nil {
-			errorChan <- err
-			break
-		}
-
-		err = websocketConn.WriteMessage(websocket.TextMessage, out)
-		if err != nil {
-			errorChan <- err
-			break
-		}
-	}
-}

api/http/handler/websocket/handler.go

@@ -26,5 +26,7 @@ func NewHandler(bouncer *security.RequestBouncer) *Handler {
 	}
 	h.PathPrefix("/websocket/exec").Handler(
 		bouncer.AuthenticatedAccess(httperror.LoggerHandler(h.websocketExec)))
+	h.PathPrefix("/websocket/attach").Handler(
+		bouncer.AuthenticatedAccess(httperror.LoggerHandler(h.websocketAttach)))
 	return h
 }

api/http/handler/websocket/hijack.go

@@ -0,0 +1,36 @@
+package websocket
+
+import (
+	"fmt"
+	"github.com/gorilla/websocket"
+	"net/http"
+	"net/http/httputil"
+)
+
+func hijackRequest(websocketConn *websocket.Conn, httpConn *httputil.ClientConn, request *http.Request) error {
+	// Server hijacks the connection, error 'connection closed' expected
+	resp, err := httpConn.Do(request)
+	if err != httputil.ErrPersistEOF {
+		if err != nil {
+			return err
+		}
+		if resp.StatusCode != http.StatusSwitchingProtocols {
+			resp.Body.Close()
+			return fmt.Errorf("unable to upgrade to tcp, received %d", resp.StatusCode)
+		}
+	}
+
+	tcpConn, brw := httpConn.Hijack()
+	defer tcpConn.Close()
+
+	errorChan := make(chan error, 1)
+	go streamFromTCPConnToWebsocketConn(websocketConn, brw, errorChan)
+	go streamFromWebsocketConnToTCPConn(websocketConn, tcpConn, errorChan)
+
+	err = <-errorChan
+	if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
+		return err
+	}
+
+	return nil
+}

api/http/handler/websocket/netdial/dial.go

@@ -1,6 +1,6 @@
 // +build !windows
 
-package websocket
+package netdial
 
 import (
 	"net"

api/http/handler/websocket/netdial/dial_windows.go

@@ -1,11 +1,9 @@
 // +build windows
 
-package websocket
+package netdial
 
 import (
 	"net"
-
-	"github.com/Microsoft/go-winio"
 )
 
 func createDial(scheme, host string) (net.Conn, error) {

api/http/handler/websocket/netdial/initdial.go

@@ -0,0 +1,35 @@
+package netdial
+
+import (
+	"crypto/tls"
+	"net"
+	"net/url"
+	"github.com/portainer/portainer/api"
+	"github.com/portainer/portainer/api/crypto"
+)
+
+func InitDial(endpoint *portainer.Endpoint) (net.Conn, error) {
+	url, err := url.Parse(endpoint.URL)
+	if err != nil {
+		return nil, err
+	}
+
+	host := url.Host
+
+	if url.Scheme == "unix" || url.Scheme == "npipe" {
+		host = url.Path
+	}
+
+	if endpoint.TLSConfig.TLS {
+		tlsConfig, err := crypto.CreateTLSConfigurationFromDisk(endpoint.TLSConfig.TLSCACertPath, endpoint.TLSConfig.TLSCertPath, endpoint.TLSConfig.TLSKeyPath, endpoint.TLSConfig.TLSSkipVerify)
+		if err != nil {
+			return nil, err
+		}
+
+		return tls.Dial(url.Scheme, host, tlsConfig)
+	}
+
+	con, err := createDial(url.Scheme, host)
+
+	return con, err
+}

api/http/handler/websocket/proxy.go

@@ -0,0 +1,44 @@
+package websocket
+
+import (
+	"crypto/tls"
+	"github.com/gorilla/websocket"
+	"github.com/koding/websocketproxy"
+	"net/http"
+	"net/url"
+	"github.com/portainer/portainer/api"
+)
+
+func (handler *Handler) proxyWebsocketRequest(w http.ResponseWriter, r *http.Request, params *webSocketRequestParams) error {
+	agentURL, err := url.Parse(params.endpoint.URL)
+	if err != nil {
+		return err
+	}
+
+	agentURL.Scheme = "ws"
+	proxy := websocketproxy.NewProxy(agentURL)
+
+	if params.endpoint.TLSConfig.TLS || params.endpoint.TLSConfig.TLSSkipVerify {
+		agentURL.Scheme = "wss"
+		proxy.Dialer = &websocket.Dialer{
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: params.endpoint.TLSConfig.TLSSkipVerify,
+			},
+		}
+	}
+
+	signature, err := handler.SignatureService.CreateSignature(portainer.PortainerAgentSignatureMessage)
+	if err != nil {
+		return err
+	}
+
+	proxy.Director = func(incoming *http.Request, out http.Header) {
+		out.Set(portainer.PortainerAgentPublicKeyHeader, handler.SignatureService.EncodedPublicKey())
+		out.Set(portainer.PortainerAgentSignatureHeader, signature)
+		out.Set(portainer.PortainerAgentTargetHeader, params.nodeName)
+	}
+
+	proxy.ServeHTTP(w, r)
+
+	return nil
+}

api/http/handler/websocket/stream.go

@@ -0,0 +1,40 @@
+package websocket
+
+import (
+	"bufio"
+	"github.com/gorilla/websocket"
+	"net"
+)
+
+func streamFromWebsocketConnToTCPConn(websocketConn *websocket.Conn, tcpConn net.Conn, errorChan chan error) {
+	for {
+		_, in, err := websocketConn.ReadMessage()
+		if err != nil {
+			errorChan <- err
+			break
+		}
+
+		_, err = tcpConn.Write(in)
+		if err != nil {
+			errorChan <- err
+			break
+		}
+	}
+}
+
+func streamFromTCPConnToWebsocketConn(websocketConn *websocket.Conn, br *bufio.Reader, errorChan chan error) {
+	for {
+		out := make([]byte, 2048)
+		_, err := br.Read(out)
+		if err != nil {
+			errorChan <- err
+			break
+		}
+
+		err = websocketConn.WriteMessage(websocket.TextMessage, out)
+		if err != nil {
+			errorChan <- err
+			break
+		}
+	}
+}

api/http/handler/websocket/types.go

@@ -0,0 +1,12 @@
+package websocket
+
+import (
+	"github.com/portainer/portainer/api"
+)
+
+
+type webSocketRequestParams struct {
+	ID   string
+	nodeName string
+	endpoint *portainer.Endpoint
+}