Merge pull request #129872 from seans3/websocket-https-proxy

WebSocket HTTPS Proxy support Kubernetes-commit: 6f13ba03dac1865174b4edf2b43f6a0f453a8ac7
2025-07-21 10:30:24 +00:00 · 2025-03-20 15:34:31 -07:00 · 2025-03-20 15:34:31 -07:00 · a27e26debd
commit a27e26debd
parent e782ad64ac c77f9e4f9e
4 changed files with 134 additions and 61 deletions
--- a/go.mod
+++ b/go.mod
@ -12,7 +12,7 @@ require (
 	github.com/google/gnostic-models v0.6.9
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
-	github.com/gorilla/websocket v1.5.3
+	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
 	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 	github.com/peterbourgon/diskv v2.0.1+incompatible
@ -25,7 +25,7 @@ require (
 	golang.org/x/time v0.9.0
 	google.golang.org/protobuf v1.36.5
 	gopkg.in/evanphx/json-patch.v4 v4.12.0
-	k8s.io/api v0.0.0-20250320031247-741ca7705f8e
+	k8s.io/api v0.0.0-20250320173055-71f613bc3510
 	k8s.io/apimachinery v0.0.0-20250319092800-e8a77bd768fd
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff
--- a/go.sum
+++ b/go.sum
@ -34,8 +34,8 @@ github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db h1:097atOisP2aRj7vFgY
 github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
-github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo=
+github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 h1:+ngKgrYPPJrOjhax5N+uePQ0Fh1Z7PheYoUI/0nzkPA=
 github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
@ -146,8 +146,8 @@ gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.0.0-20250320031247-741ca7705f8e h1:O3P0nY3fp1Hj0yurtChVY3bTcCVY0QVwgtEVjCxZu9E=
-k8s.io/api v0.0.0-20250320031247-741ca7705f8e/go.mod h1:JO0tyTI0qSXXaGVhLdqwfi3RMbS2g9hcYvzBmZP5wVk=
+k8s.io/api v0.0.0-20250320173055-71f613bc3510 h1:iBS+3o4br4KHBsDGKYGCSaQylhR+r4Opj98akgLcjII=
+k8s.io/api v0.0.0-20250320173055-71f613bc3510/go.mod h1:JO0tyTI0qSXXaGVhLdqwfi3RMbS2g9hcYvzBmZP5wVk=
 k8s.io/apimachinery v0.0.0-20250319092800-e8a77bd768fd h1:KoXgjwEokLM8o95kMxowg5vp5iQ4v46Kk+zobsqeTgU=
 k8s.io/apimachinery v0.0.0-20250319092800-e8a77bd768fd/go.mod h1:D2UW665TVSpInyOuG6C+PMtC1MZheP0KQz65UPQEiI4=
 k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
--- a/tools/remotecommand/fallback_test.go
+++ b/tools/remotecommand/fallback_test.go
@ -288,7 +288,7 @@ KR8NJEkK99Vh/tew6jAMll70xFrE7aF8VLXJVE7w4sQzuvHxl9Q=
 `)

 // See (https://github.com/kubernetes/kubernetes/issues/126134).
-func TestFallbackClient_WebSocketHTTPSProxyCausesSPDYFallback(t *testing.T) {
+func TestFallbackClient_WebSocketHTTPSProxyNoFallback(t *testing.T) {
 	cert, err := tls.X509KeyPair(localhostCert, localhostKey)
 	if err != nil {
 		t.Errorf("https (valid hostname): proxy_test: %v", err)
@ -309,42 +309,40 @@ func TestFallbackClient_WebSocketHTTPSProxyCausesSPDYFallback(t *testing.T) {
 	proxyLocation, err := url.Parse(proxyServer.URL)
 	require.NoError(t, err)

-	// Create fake SPDY server. Copy received STDIN data back onto STDOUT stream.
-	spdyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		var stdin, stdout bytes.Buffer
-		ctx, err := createHTTPStreams(w, req, &StreamOptions{
-			Stdin:  &stdin,
-			Stdout: &stdout,
-		})
+	// Create fake WebSocket server. Copy received STDIN data back onto STDOUT stream.
+	websocketServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		conns, err := webSocketServerStreams(req, w, streamOptionsFromRequest(req))
 		if err != nil {
 			w.WriteHeader(http.StatusForbidden)
 			return
 		}
-		defer ctx.conn.Close() //nolint:errcheck
-		_, err = io.Copy(ctx.stdoutStream, ctx.stdinStream)
+		defer conns.conn.Close() //nolint:errcheck
+		// Loopback the STDIN stream onto the STDOUT stream.
+		_, err = io.Copy(conns.stdoutStream, conns.stdinStream)
 		if err != nil {
-			t.Fatalf("error copying STDIN to STDOUT: %v", err)
+			t.Fatalf("websocket copy error: %v", err)
 		}
 	}))
-	defer spdyServer.Close() //nolint:errcheck
+	defer websocketServer.Close() //nolint:errcheck

-	backendLocation, err := url.Parse(spdyServer.URL)
+	// Now create the WebSocket client (executor), and point it to the TLS proxy server.
+	// The proxy server should open a websocket connection to the fake websocket server.
+	websocketServer.URL = websocketServer.URL + "?" + "stdin=true" + "&" + "stdout=true"
+	websocketLocation, err := url.Parse(websocketServer.URL)
 	require.NoError(t, err)
-
 	clientConfig := &rest.Config{
-		Host:            spdyServer.URL,
+		Host:            websocketLocation.Host,
 		TLSClientConfig: rest.TLSClientConfig{CAData: localhostCert},
 		Proxy: func(req *http.Request) (*url.URL, error) {
 			return proxyLocation, nil
 		},
 	}
-
-	// Websocket with https proxy will fail in dialing (falling back to SPDY).
-	websocketExecutor, err := NewWebSocketExecutor(clientConfig, "GET", backendLocation.String())
+	websocketExecutor, err := NewWebSocketExecutor(clientConfig, "GET", websocketServer.URL)
 	require.NoError(t, err)
-	spdyExecutor, err := NewSPDYExecutor(clientConfig, "POST", backendLocation)
+	emptyURL, _ := url.Parse("")
+	spdyExecutor, err := NewSPDYExecutor(clientConfig, "POST", emptyURL)
 	require.NoError(t, err)
-	// Fallback to spdyExecutor with websocket https proxy error; spdyExecutor succeeds against fake spdy server.
+	// No fallback to spdyExecutor with websocket.
 	sawHTTPSProxyError := false
 	exec, err := NewFallbackExecutor(websocketExecutor, spdyExecutor, func(err error) bool {
 		if httpstream.IsUpgradeFailure(err) {
@ -396,9 +394,9 @@ func TestFallbackClient_WebSocketHTTPSProxyCausesSPDYFallback(t *testing.T) {
 		t.Errorf("unexpected data received: %d sent: %d", len(data), len(randomData))
 	}

-	// Ensure the https proxy error was observed
-	if !sawHTTPSProxyError {
-		t.Errorf("expected to see https proxy error")
+	// Ensure the https proxy error was *not* observed
+	if sawHTTPSProxyError {
+		t.Errorf("expected to *not* see https proxy error")
 	}
 	// Ensure the proxy was called once
 	if e, a := int64(1), proxyCalled.Load(); e != a {
--- a/tools/remotecommand/websocket_test.go
+++ b/tools/remotecommand/websocket_test.go
@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/rand"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
@ -31,16 +32,18 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"

 	gwebsocket "github.com/gorilla/websocket"
+	"github.com/stretchr/testify/require"

 	v1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/httpstream/wsstream"
+	utilnettesting "k8s.io/apimachinery/pkg/util/net/testing"
 	"k8s.io/apimachinery/pkg/util/remotecommand"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
@ -1342,21 +1345,75 @@ func createWebSocketStreams(req *http.Request, w http.ResponseWriter, opts *opti
 	return wsStreams, nil
 }

-// See (https://github.com/kubernetes/kubernetes/issues/126134).
-func TestWebSocketClient_HTTPSProxyErrorExpected(t *testing.T) {
-	urlStr := "http://127.0.0.1/never-used" + "?" + "stdin=true" + "&" + "stdout=true"
-	websocketLocation, err := url.Parse(urlStr)
+func TestWebSocketClient_ProxySucceeds(t *testing.T) {
+	// Validate websocket proxy succeeds for each of the enumerated schemes.
+	proxySchemes := []string{"http", "https"}
+	for _, proxyScheme := range proxySchemes {
+		// Create the proxy handler, keeping track of how many times it was called.
+		var proxyCalled atomic.Int64
+		proxyHandler := utilnettesting.NewHTTPProxyHandler(t, func(req *http.Request) bool {
+			proxyCalled.Add(1)
+			return true
+		})
+		defer proxyHandler.Wait()
+		// Create/Start the proxy server, adding TLS functionality depending on scheme.
+		proxyServer := httptest.NewUnstartedServer(proxyHandler)
+		if proxyScheme == "https" {
+			cert, err := tls.X509KeyPair(localhostCert, localhostKey)
 			if err != nil {
-		t.Fatalf("Unable to parse WebSocket server URL: %s", urlStr)
+				t.Errorf("https (valid hostname): proxy_test: %v", err)
 			}
-	// proxy url with https scheme will trigger websocket dialing error.
-	httpsProxyFunc := func(req *http.Request) (*url.URL, error) { return url.Parse("https://127.0.0.1") }
-	exec, err := NewWebSocketExecutor(&rest.Config{Host: websocketLocation.Host, Proxy: httpsProxyFunc}, "GET", urlStr)
+			proxyServer.TLS = &tls.Config{Certificates: []tls.Certificate{cert}}
+			proxyServer.StartTLS()
+		} else {
+			proxyServer.Start()
+		}
+		defer proxyServer.Close() //nolint:errcheck
+		proxyLocation, err := url.Parse(proxyServer.URL)
+		require.NoError(t, err)
+		t.Logf("Proxy URL: %s", proxyLocation.String())
+
+		// Create fake WebSocket server. Copy received STDIN data back onto STDOUT stream.
+		websocketServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+			conns, err := webSocketServerStreams(req, w, streamOptionsFromRequest(req))
 			if err != nil {
-		t.Errorf("unexpected error creating websocket executor: %v", err)
+				t.Fatalf("error on webSocketServerStreams: %v", err)
+			}
+			defer conns.conn.Close() //nolint:errcheck
+			// Loopback the STDIN stream onto the STDOUT stream.
+			_, err = io.Copy(conns.stdoutStream, conns.stdinStream)
+			if err != nil {
+				t.Fatalf("error copying STDIN to STDOUT: %v", err)
+			}
+		}))
+		defer websocketServer.Close() //nolint:errcheck
+
+		// Now create the WebSocket client (executor), and point it to the TLS proxy server.
+		// The proxy server should open a websocket connection to the fake websocket server.
+		websocketServer.URL = websocketServer.URL + "?" + "stdin=true" + "&" + "stdout=true"
+		websocketLocation, err := url.Parse(websocketServer.URL)
+		require.NoError(t, err)
+		clientConfig := &rest.Config{
+			Host: websocketLocation.Host,
+			// Unused if "http" scheme.
+			TLSClientConfig: rest.TLSClientConfig{CAData: localhostCert},
+			Proxy: func(req *http.Request) (*url.URL, error) {
+				return proxyLocation, nil
+			},
+		}
+		exec, err := NewWebSocketExecutor(clientConfig, "GET", websocketServer.URL)
+		require.NoError(t, err)
+
+		// Generate random data, and set it up to stream on STDIN. The data will be
+		// returned on the STDOUT buffer.
+		randomSize := 1024 * 1024
+		randomData := make([]byte, randomSize)
+		if _, err := rand.Read(randomData); err != nil {
+			t.Errorf("unexpected error reading random data: %v", err)
 		}
 		var stdout bytes.Buffer
 		options := &StreamOptions{
+			Stdin:  bytes.NewReader(randomData),
 			Stdout: &stdout,
 		}
 		errorChan := make(chan error)
@ -1369,11 +1426,29 @@ func TestWebSocketClient_HTTPSProxyErrorExpected(t *testing.T) {
 		case <-time.After(wait.ForeverTestTimeout):
 			t.Fatalf("expect stream to be closed after connection is closed.")
 		case err := <-errorChan:
-		if err == nil {
-			t.Errorf("expected error but received none")
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
 			}
-		if !httpstream.IsHTTPSProxyError(err) {
-			t.Errorf("expected https proxy error, got (%s)", err)
+			// Validate remote command v5 protocol was negotiated.
+			streamExec := exec.(*wsStreamExecutor)
+			if remotecommand.StreamProtocolV5Name != streamExec.negotiated {
+				t.Fatalf("expected remote command v5 protocol, got (%s)", streamExec.negotiated)
+			}
+		}
+		data, err := io.ReadAll(bytes.NewReader(stdout.Bytes()))
+		if err != nil {
+			t.Fatalf("error reading the stream: %v", err)
+		}
+		// Check the random data sent on STDIN was the same returned on STDOUT.
+		t.Logf("comparing %d random bytes sent data versus received", len(randomData))
+		if !bytes.Equal(randomData, data) {
+			t.Errorf("unexpected data received: %d sent: %d", len(data), len(randomData))
+		} else {
+			t.Log("success--random bytes are the same")
+		}
+		// Ensure the proxy was called once
+		if e, a := int64(1), proxyCalled.Load(); e != a {
+			t.Errorf("expected %d proxy call, got %d", e, a)
 		}
 	}
 }