diff --git a/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream.go b/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream.go index a32fce5a0c1..8054b98676b 100644 --- a/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream.go +++ b/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream.go @@ -116,6 +116,15 @@ func IsUpgradeFailure(err error) bool { return errors.As(err, &upgradeErr) } +// isHTTPSProxyError returns true if error is Gorilla/Websockets HTTPS Proxy dial error; +// false otherwise (see https://github.com/kubernetes/kubernetes/issues/126134). +func IsHTTPSProxyError(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "proxy: unknown scheme: https") +} + // IsUpgradeRequest returns true if the given request is a connection upgrade request func IsUpgradeRequest(req *http.Request) bool { for _, h := range req.Header[http.CanonicalHeaderKey(HeaderConnection)] { diff --git a/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream_test.go b/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream_test.go index 11fb928634e..6254e97096a 100644 --- a/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream_test.go +++ b/staging/src/k8s.io/apimachinery/pkg/util/httpstream/httpstream_test.go @@ -168,3 +168,32 @@ func TestIsUpgradeFailureError(t *testing.T) { }) } } + +func TestIsHTTPSProxyError(t *testing.T) { + testCases := map[string]struct { + err error + expected bool + }{ + "nil error should return false": { + err: nil, + expected: false, + }, + "Not HTTPS proxy error should return false": { + err: errors.New("this is not an upgrade error"), + expected: false, + }, + "HTTPS proxy error should return true": { + err: errors.New("proxy: unknown scheme: https"), + expected: true, + }, + } + + for name, test := range testCases { + t.Run(name, func(t *testing.T) { + actual := IsHTTPSProxyError(test.err) + if test.expected != actual { + t.Errorf("expected HTTPS proxy error %t, got %t", test.expected, actual) + } + }) + } +} diff --git a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go index 1a6805f1221..70583958ea4 100644 --- a/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go +++ b/staging/src/k8s.io/client-go/tools/portforward/fallback_dialer_test.go @@ -17,10 +17,12 @@ limitations under the License. package portforward import ( + "errors" "fmt" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/util/httpstream" ) @@ -36,7 +38,7 @@ func TestFallbackDialer(t *testing.T) { assert.True(t, primary.dialed, "no fallback; primary should have dialed") assert.False(t, secondary.dialed, "no fallback; secondary should *not* have dialed") assert.Equal(t, primaryProtocol, negotiated, "primary negotiated protocol returned") - assert.Nil(t, err, "error from primary dialer should be nil") + require.NoError(t, err, "error from primary dialer should be nil") // If primary dialer error is upgrade error, then fallback returning secondary dial response. primary = &fakeDialer{dialed: false, negotiatedProtocol: primaryProtocol, err: &httpstream.UpgradeFailureError{}} secondary = &fakeDialer{dialed: false, negotiatedProtocol: secondaryProtocol} @@ -45,7 +47,18 @@ func TestFallbackDialer(t *testing.T) { assert.True(t, primary.dialed, "fallback; primary should have dialed") assert.True(t, secondary.dialed, "fallback; secondary should have dialed") assert.Equal(t, secondaryProtocol, negotiated, "negotiated protocol is from secondary dialer") - assert.Nil(t, err, "error from secondary dialer should be nil") + require.NoError(t, err, "error from secondary dialer should be nil") + // If primary dialer error is https proxy dialing error, then fallback returning secondary dial response. + primary = &fakeDialer{negotiatedProtocol: primaryProtocol, err: errors.New("proxy: unknown scheme: https")} + secondary = &fakeDialer{negotiatedProtocol: secondaryProtocol} + fallbackDialer = NewFallbackDialer(primary, secondary, func(err error) bool { + return httpstream.IsUpgradeFailure(err) || httpstream.IsHTTPSProxyError(err) + }) + _, negotiated, err = fallbackDialer.Dial(protocols...) + assert.True(t, primary.dialed, "fallback; primary should have dialed") + assert.True(t, secondary.dialed, "fallback; secondary should have dialed") + assert.Equal(t, secondaryProtocol, negotiated, "negotiated protocol is from secondary dialer") + require.NoError(t, err, "error from secondary dialer should be nil") // If primary dialer returns non-upgrade error, then primary error is returned. nonUpgradeErr := fmt.Errorf("This is a non-upgrade error") primary = &fakeDialer{dialed: false, err: nonUpgradeErr} diff --git a/staging/src/k8s.io/client-go/tools/remotecommand/fallback_test.go b/staging/src/k8s.io/client-go/tools/remotecommand/fallback_test.go index 52e1f7b160d..b82c7bedf7a 100644 --- a/staging/src/k8s.io/client-go/tools/remotecommand/fallback_test.go +++ b/staging/src/k8s.io/client-go/tools/remotecommand/fallback_test.go @@ -20,15 +20,19 @@ import ( "bytes" "context" "crypto/rand" + "crypto/tls" "io" "net/http" "net/http/httptest" "net/url" + "sync/atomic" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/util/httpstream" + 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" @@ -225,3 +229,175 @@ func TestFallbackClient_PrimaryAndSecondaryFail(t *testing.T) { require.Error(t, err) } } + +// localhostCert was generated from crypto/tls/generate_cert.go with the following command: +// +// go run generate_cert.go --rsa-bits 2048 --host 127.0.0.1,::1,example.com --ca --start-date "Jan 1 00:00:00 1970" --duration=1000000h +var localhostCert = []byte(`-----BEGIN CERTIFICATE----- +MIIDGTCCAgGgAwIBAgIRALL5AZcefF4kkYV1SEG6YrMwDQYJKoZIhvcNAQELBQAw +EjEQMA4GA1UEChMHQWNtZSBDbzAgFw03MDAxMDEwMDAwMDBaGA8yMDg0MDEyOTE2 +MDAwMFowEjEQMA4GA1UEChMHQWNtZSBDbzCCASIwDQYJKoZIhvcNAQEBBQADggEP +ADCCAQoCggEBALQ/FHcyVwdFHxARbbD2KBtDUT7Eni+8ioNdjtGcmtXqBv45EC1C +JOqqGJTroFGJ6Q9kQIZ9FqH5IJR2fOOJD9kOTueG4Vt1JY1rj1Kbpjefu8XleZ5L +SBwIWVnN/lEsEbuKmj7N2gLt5AH3zMZiBI1mg1u9Z5ZZHYbCiTpBrwsq6cTlvR9g +dyo1YkM5hRESCzsrL0aUByoo0qRMD8ZsgANJwgsiO0/M6idbxDwv1BnGwGmRYvOE +Hxpy3v0Jg7GJYrvnpnifJTs4nw91N5X9pXxR7FFzi/6HTYDWRljvTb0w6XciKYAz +bWZ0+cJr5F7wB7ovlbm7HrQIR7z7EIIu2d8CAwEAAaNoMGYwDgYDVR0PAQH/BAQD +AgKkMBMGA1UdJQQMMAoGCCsGAQUFBwMBMA8GA1UdEwEB/wQFMAMBAf8wLgYDVR0R +BCcwJYILZXhhbXBsZS5jb22HBH8AAAGHEAAAAAAAAAAAAAAAAAAAAAEwDQYJKoZI +hvcNAQELBQADggEBAFPPWopNEJtIA2VFAQcqN6uJK+JVFOnjGRoCrM6Xgzdm0wxY +XCGjsxY5dl+V7KzdGqu858rCaq5osEBqypBpYAnS9C38VyCDA1vPS1PsN8SYv48z +DyBwj+7R2qar0ADBhnhWxvYO9M72lN/wuCqFKYMeFSnJdQLv3AsrrHe9lYqOa36s +8wxSwVTFTYXBzljPEnSaaJMPqFD8JXaZK1ryJPkO5OsCNQNGtatNiWAf3DcmwHAT +MGYMzP0u4nw47aRz9shB8w+taPKHx2BVwE1m/yp3nHVioOjXqA1fwRQVGclCJSH1 +D2iq3hWVHRENgjTjANBPICLo9AZ4JfN6PH19mnU= +-----END CERTIFICATE-----`) + +// localhostKey is the private key for localhostCert. +var localhostKey = []byte(`-----BEGIN RSA PRIVATE KEY----- +MIIEogIBAAKCAQEAtD8UdzJXB0UfEBFtsPYoG0NRPsSeL7yKg12O0Zya1eoG/jkQ +LUIk6qoYlOugUYnpD2RAhn0WofkglHZ844kP2Q5O54bhW3UljWuPUpumN5+7xeV5 +nktIHAhZWc3+USwRu4qaPs3aAu3kAffMxmIEjWaDW71nllkdhsKJOkGvCyrpxOW9 +H2B3KjViQzmFERILOysvRpQHKijSpEwPxmyAA0nCCyI7T8zqJ1vEPC/UGcbAaZFi +84QfGnLe/QmDsYliu+emeJ8lOzifD3U3lf2lfFHsUXOL/odNgNZGWO9NvTDpdyIp +gDNtZnT5wmvkXvAHui+VubsetAhHvPsQgi7Z3wIDAQABAoIBAGmw93IxjYCQ0ncc +kSKMJNZfsdtJdaxuNRZ0nNNirhQzR2h403iGaZlEpmdkhzxozsWcto1l+gh+SdFk +bTUK4MUZM8FlgO2dEqkLYh5BcMT7ICMZvSfJ4v21E5eqR68XVUqQKoQbNvQyxFk3 +EddeEGdNrkb0GDK8DKlBlzAW5ep4gjG85wSTjR+J+muUv3R0BgLBFSuQnIDM/IMB +LWqsja/QbtB7yppe7jL5u8UCFdZG8BBKT9fcvFIu5PRLO3MO0uOI7LTc8+W1Xm23 +uv+j3SY0+v+6POjK0UlJFFi/wkSPTFIfrQO1qFBkTDQHhQ6q/7GnILYYOiGbIRg2 +NNuP52ECgYEAzXEoy50wSYh8xfFaBuxbm3ruuG2W49jgop7ZfoFrPWwOQKAZS441 +VIwV4+e5IcA6KkuYbtGSdTYqK1SMkgnUyD/VevwAqH5TJoEIGu0pDuKGwVuwqioZ +frCIAV5GllKyUJ55VZNbRr2vY2fCsWbaCSCHETn6C16DNuTCe5C0JBECgYEA4JqY +5GpNbMG8fOt4H7hU0Fbm2yd6SHJcQ3/9iimef7xG6ajxsYrIhg1ft+3IPHMjVI0+ +9brwHDnWg4bOOx/VO4VJBt6Dm/F33bndnZRkuIjfSNpLM51P+EnRdaFVHOJHwKqx +uF69kihifCAG7YATgCveeXImzBUSyZUz9UrETu8CgYARNBimdFNG1RcdvEg9rC0/ +p9u1tfecvNySwZqU7WF9kz7eSonTueTdX521qAHowaAdSpdJMGODTTXaywm6cPhQ +jIfj9JZZhbqQzt1O4+08Qdvm9TamCUB5S28YLjza+bHU7nBaqixKkDfPqzCyilpX +yVGGL8SwjwmN3zop/sQXAQKBgC0JMsESQ6YcDsRpnrOVjYQc+LtW5iEitTdfsaID +iGGKihmOI7B66IxgoCHMTws39wycKdSyADVYr5e97xpR3rrJlgQHmBIrz+Iow7Q2 +LiAGaec8xjl6QK/DdXmFuQBKqyKJ14rljFODP4QuE9WJid94bGqjpf3j99ltznZP +4J8HAoGAJb4eb4lu4UGwifDzqfAPzLGCoi0fE1/hSx34lfuLcc1G+LEu9YDKoOVJ +9suOh0b5K/bfEy9KrVMBBriduvdaERSD8S3pkIQaitIz0B029AbE4FLFf9lKQpP2 +KR8NJEkK99Vh/tew6jAMll70xFrE7aF8VLXJVE7w4sQzuvHxl9Q= +-----END RSA PRIVATE KEY----- +`) + +// See (https://github.com/kubernetes/kubernetes/issues/126134). +func TestFallbackClient_WebSocketHTTPSProxyCausesSPDYFallback(t *testing.T) { + cert, err := tls.X509KeyPair(localhostCert, localhostKey) + if err != nil { + t.Errorf("https (valid hostname): proxy_test: %v", err) + } + + var proxyCalled atomic.Int64 + proxyHandler := utilnettesting.NewHTTPProxyHandler(t, func(req *http.Request) bool { + proxyCalled.Add(1) + return true + }) + defer proxyHandler.Wait() + + proxyServer := httptest.NewUnstartedServer(proxyHandler) + proxyServer.TLS = &tls.Config{Certificates: []tls.Certificate{cert}} + proxyServer.StartTLS() + defer proxyServer.Close() //nolint:errcheck + + 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, + }) + if err != nil { + w.WriteHeader(http.StatusForbidden) + return + } + defer ctx.conn.Close() //nolint:errcheck + _, err = io.Copy(ctx.stdoutStream, ctx.stdinStream) + if err != nil { + t.Fatalf("error copying STDIN to STDOUT: %v", err) + } + })) + defer spdyServer.Close() //nolint:errcheck + + backendLocation, err := url.Parse(spdyServer.URL) + require.NoError(t, err) + + clientConfig := &rest.Config{ + Host: spdyServer.URL, + 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()) + require.NoError(t, err) + spdyExecutor, err := NewSPDYExecutor(clientConfig, "POST", backendLocation) + require.NoError(t, err) + // Fallback to spdyExecutor with websocket https proxy error; spdyExecutor succeeds against fake spdy server. + sawHTTPSProxyError := false + exec, err := NewFallbackExecutor(websocketExecutor, spdyExecutor, func(err error) bool { + if httpstream.IsUpgradeFailure(err) { + t.Errorf("saw upgrade failure: %v", err) + return true + } + if httpstream.IsHTTPSProxyError(err) { + sawHTTPSProxyError = true + t.Logf("saw https proxy error: %v", err) + return true + } + return false + }) + 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() { + 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.Errorf("unexpected error") + } + } + + data, err := io.ReadAll(bytes.NewReader(stdout.Bytes())) + if err != nil { + t.Errorf("error reading the stream: %v", err) + return + } + // Check the random data sent on STDIN was the same returned on STDOUT. + if !bytes.Equal(randomData, data) { + 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 proxy was called once + if e, a := int64(1), proxyCalled.Load(); e != a { + t.Errorf("expected %d proxy call, got %d", e, a) + } +} diff --git a/staging/src/k8s.io/client-go/tools/remotecommand/websocket_test.go b/staging/src/k8s.io/client-go/tools/remotecommand/websocket_test.go index 4a333b0b24e..b70afcacbeb 100644 --- a/staging/src/k8s.io/client-go/tools/remotecommand/websocket_test.go +++ b/staging/src/k8s.io/client-go/tools/remotecommand/websocket_test.go @@ -39,6 +39,7 @@ import ( 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" "k8s.io/apimachinery/pkg/util/remotecommand" "k8s.io/apimachinery/pkg/util/wait" @@ -1340,3 +1341,39 @@ 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) + }() + + 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") + } + if !httpstream.IsHTTPSProxyError(err) { + t.Errorf("expected https proxy error, got (%s)", err) + } + } +} diff --git a/staging/src/k8s.io/kubectl/pkg/cmd/attach/attach.go b/staging/src/k8s.io/kubectl/pkg/cmd/attach/attach.go index c1e8d83dc54..b38760a1c86 100644 --- a/staging/src/k8s.io/kubectl/pkg/cmd/attach/attach.go +++ b/staging/src/k8s.io/kubectl/pkg/cmd/attach/attach.go @@ -184,7 +184,9 @@ func createExecutor(url *url.URL, config *restclient.Config) (remotecommand.Exec if err != nil { return nil, err } - exec, err = remotecommand.NewFallbackExecutor(websocketExec, exec, httpstream.IsUpgradeFailure) + exec, err = remotecommand.NewFallbackExecutor(websocketExec, exec, func(err error) bool { + return httpstream.IsUpgradeFailure(err) || httpstream.IsHTTPSProxyError(err) + }) if err != nil { return nil, err } diff --git a/staging/src/k8s.io/kubectl/pkg/cmd/exec/exec.go b/staging/src/k8s.io/kubectl/pkg/cmd/exec/exec.go index 40580766418..a6ec2f4304d 100644 --- a/staging/src/k8s.io/kubectl/pkg/cmd/exec/exec.go +++ b/staging/src/k8s.io/kubectl/pkg/cmd/exec/exec.go @@ -147,7 +147,9 @@ func createExecutor(url *url.URL, config *restclient.Config) (remotecommand.Exec if err != nil { return nil, err } - exec, err = remotecommand.NewFallbackExecutor(websocketExec, exec, httpstream.IsUpgradeFailure) + exec, err = remotecommand.NewFallbackExecutor(websocketExec, exec, func(err error) bool { + return httpstream.IsUpgradeFailure(err) || httpstream.IsHTTPSProxyError(err) + }) if err != nil { return nil, err } diff --git a/staging/src/k8s.io/kubectl/pkg/cmd/portforward/portforward.go b/staging/src/k8s.io/kubectl/pkg/cmd/portforward/portforward.go index 66d48589512..027bffd4a48 100644 --- a/staging/src/k8s.io/kubectl/pkg/cmd/portforward/portforward.go +++ b/staging/src/k8s.io/kubectl/pkg/cmd/portforward/portforward.go @@ -148,7 +148,9 @@ func (f *defaultPortForwarder) ForwardPorts(method string, url *url.URL, opts Po return err } // First attempt tunneling (websocket) dialer, then fallback to spdy dialer. - dialer = portforward.NewFallbackDialer(tunnelingDialer, dialer, httpstream.IsUpgradeFailure) + dialer = portforward.NewFallbackDialer(tunnelingDialer, dialer, func(err error) bool { + return httpstream.IsUpgradeFailure(err) || httpstream.IsHTTPSProxyError(err) + }) } fw, err := portforward.NewOnAddresses(dialer, opts.Address, opts.Ports, opts.StopChannel, opts.ReadyChannel, f.Out, f.ErrOut) if err != nil {