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

WebSocket HTTPS Proxy support

Author: k8s-ci-robot

Diff for: go.mod

@@ -159,7 +159,7 @@ require (
 	github.com/google/btree v1.1.3 // indirect
 	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect

Diff for: go.sum

@@ -302,8 +302,8 @@ github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 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.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-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/grpc-ecosystem/go-grpc-middleware v1.3.0 h1:+9834+KizmvFV7pXQGSXQTsaWhq2GjuNUt0aUU0YBYw=

Diff for: staging/src/k8s.io/apiextensions-apiserver/go.mod

@@ -62,7 +62,7 @@ require (
 	github.com/golang-jwt/jwt/v4 v4.5.1 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.1.3 // indirect
-	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect

Diff for: staging/src/k8s.io/apiextensions-apiserver/go.sum

@@ -19,7 +19,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/grpc-ecosystem/go-grpc-prometheus v1.2.0
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f

Diff for: staging/src/k8s.io/apiserver/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

Diff for: staging/src/k8s.io/apiserver/go.sum

@@ -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 {

Diff for: staging/src/k8s.io/cli-runtime/go.sum

@@ -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,38 +1345,110 @@ 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)
-	if err != nil {
-		t.Fatalf("Unable to parse WebSocket server URL: %s", urlStr)
-	}
-	// 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)
-	if err != nil {
-		t.Errorf("unexpected error creating websocket executor: %v", err)
-	}
-	var stdout bytes.Buffer
-	options := &StreamOptions{
-		Stdout: &stdout,
-	}
-	errorChan := make(chan error)
-	go func() {
-		// Start the streaming on the WebSocket "exec" client.
-		errorChan <- exec.StreamWithContext(context.Background(), *options)
-	}()
+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.Errorf("https (valid hostname): proxy_test: %v", err)
+			}
+			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())
 
-	select {
-	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")
+		// 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.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)
+		go func() {
+			// Start the streaming on the WebSocket "exec" client.
+			errorChan <- exec.StreamWithContext(context.Background(), *options)
+		}()
+
+		select {
+		case <-time.After(wait.ForeverTestTimeout):
+			t.Fatalf("expect stream to be closed after connection is closed.")
+		case err := <-errorChan:
+			if err != nil {
+				t.Fatalf("unexpected error: %v", 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")
 		}
-		if !httpstream.IsHTTPSProxyError(err) {
-			t.Errorf("expected https proxy error, got (%s)", err)
+		// 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)
 		}
 	}
 }