portforward: tunnel spdy through websockets

Kubernetes-commit: 8b447d8c97e8823b4308eb91cf7d75693e867c61

Author: seans3

pkg/util/httpstream/wsstream/conn.go

@@ -27,6 +27,7 @@ import (
 	"golang.org/x/net/websocket"
 
 	"k8s.io/apimachinery/pkg/util/httpstream"
+	"k8s.io/apimachinery/pkg/util/portforward"
 	"k8s.io/apimachinery/pkg/util/remotecommand"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/klog/v2"
@@ -106,6 +107,23 @@ func IsWebSocketRequestWithStreamCloseProtocol(req *http.Request) bool {
 	return false
 }
 
+// IsWebSocketRequestWithTunnelingProtocol returns true if the request contains headers
+// identifying that it is requesting a websocket upgrade with a tunneling protocol;
+// false otherwise.
+func IsWebSocketRequestWithTunnelingProtocol(req *http.Request) bool {
+	if !IsWebSocketRequest(req) {
+		return false
+	}
+	requestedProtocols := strings.TrimSpace(req.Header.Get(WebSocketProtocolHeader))
+	for _, requestedProtocol := range strings.Split(requestedProtocols, ",") {
+		if protocolSupportsWebsocketTunneling(strings.TrimSpace(requestedProtocol)) {
+			return true
+		}
+	}
+
+	return false
+}
+
 // IgnoreReceives reads from a WebSocket until it is closed, then returns. If timeout is set, the
 // read and write deadlines are pushed every time a new message is received.
 func IgnoreReceives(ws *websocket.Conn, timeout time.Duration) {
@@ -301,6 +319,12 @@ func protocolSupportsStreamClose(protocol string) bool {
 	return protocol == remotecommand.StreamProtocolV5Name
 }
 
+// protocolSupportsWebsocketTunneling returns true if the passed protocol
+// is a tunneled Kubernetes spdy protocol; false otherwise.
+func protocolSupportsWebsocketTunneling(protocol string) bool {
+	return strings.HasPrefix(protocol, portforward.WebsocketsSPDYTunnelingPrefix) && strings.HasSuffix(protocol, portforward.KubernetesSuffix)
+}
+
 // handle implements a websocket handler.
 func (conn *Conn) handle(ws *websocket.Conn) {
 	conn.initialize(ws)

pkg/util/portforward/constants.go

@@ -0,0 +1,24 @@
+/*
+Copyright 2016 The Kubernetes Authors.
+
+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 portforward
+
+const (
+	PortForwardV1Name                    = "portforward.k8s.io"
+	WebsocketsSPDYTunnelingPrefix        = "SPDY/3.1+"
+	KubernetesSuffix                     = ".k8s.io"
+	WebsocketsSPDYTunnelingPortForwardV1 = WebsocketsSPDYTunnelingPrefix + PortForwardV1Name
+)

pkg/util/proxy/upgradeaware.go

@@ -36,6 +36,7 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
 	"github.com/mxk/go-flowrate/flowrate"
+
 	"k8s.io/klog/v2"
 )
 
@@ -336,6 +337,7 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 		clone.Host = h.Location.Host
 	}
 	clone.URL = &location
+	klog.V(6).Infof("UpgradeAwareProxy: dialing for SPDY upgrade with headers: %v", clone.Header)
 	backendConn, err = h.DialForUpgrade(clone)
 	if err != nil {
 		klog.V(6).Infof("Proxy connection error: %v", err)
@@ -370,13 +372,13 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 	// hijacking should be the last step in the upgrade.
 	requestHijacker, ok := w.(http.Hijacker)
 	if !ok {
-		klog.V(6).Infof("Unable to hijack response writer: %T", w)
+		klog.Errorf("Unable to hijack response writer: %T", w)
 		h.Responder.Error(w, req, fmt.Errorf("request connection cannot be hijacked: %T", w))
 		return true
 	}
 	requestHijackedConn, _, err := requestHijacker.Hijack()
 	if err != nil {
-		klog.V(6).Infof("Unable to hijack response: %v", err)
+		klog.Errorf("Unable to hijack response: %v", err)
 		h.Responder.Error(w, req, fmt.Errorf("error hijacking connection: %v", err))
 		return true
 	}
@@ -420,7 +422,7 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 		} else {
 			writer = backendConn
 		}
-		_, err := io.Copy(writer, requestHijackedConn)
+		_, err := io.Copy(writer, &loggingReader{name: "client->backend", delegate: requestHijackedConn})
 		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
 			klog.Errorf("Error proxying data from client to backend: %v", err)
 		}
@@ -434,7 +436,7 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 		} else {
 			reader = backendConn
 		}
-		_, err := io.Copy(requestHijackedConn, reader)
+		_, err := io.Copy(requestHijackedConn, &loggingReader{name: "backend->client", delegate: reader})
 		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
 			klog.Errorf("Error proxying data from backend to client: %v", err)
 		}
@@ -452,6 +454,18 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 	return true
 }
 
+// loggingReader logs the bytes read from the "delegate" with a "name" prefix.
+type loggingReader struct {
+	name     string
+	delegate io.Reader
+}
+
+func (l *loggingReader) Read(p []byte) (int, error) {
+	n, err := l.delegate.Read(p)
+	klog.V(8).Infof("%s: %d bytes, err=%v, bytes=% X", l.name, n, err, p[:n])
+	return n, err
+}
+
 // FIXME: Taken from net/http/httputil/reverseproxy.go as singleJoiningSlash is not exported to be re-used.
 // See-also: https://github.com/golang/go/issues/44290
 func singleJoiningSlash(a, b string) string {