Merge pull request #4020 from XiShanYongYe-Chang/fix-exec-proxy

karmadactl exec command not work with clusters that enabled proxy

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.5.9
 	github.com/kr/pretty v0.3.1
+	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/onsi/ginkgo/v2 v2.11.0
 	github.com/onsi/gomega v1.27.8
@@ -23,6 +24,7 @@ require (
 	github.com/vektra/mockery/v2 v2.10.0
 	github.com/yuin/gopher-lua v0.0.0-20220504180219-658193537a64
 	go.uber.org/atomic v1.9.0
+	golang.org/x/net v0.12.0
 	golang.org/x/term v0.10.0
 	golang.org/x/text v0.11.0
 	golang.org/x/time v0.3.0
@@ -125,7 +127,6 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pelletier/go-toml v1.9.4 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
@@ -166,7 +167,6 @@ require (
 	go.uber.org/zap v1.24.0 // indirect
 	golang.org/x/crypto v0.11.0 // indirect
 	golang.org/x/mod v0.10.0 // indirect
-	golang.org/x/net v0.12.0 // indirect
 	golang.org/x/oauth2 v0.10.0 // indirect
 	golang.org/x/sync v0.2.0 // indirect
 	golang.org/x/sys v0.10.0 // indirect

pkg/util/proxy/proxy.go

@@ -8,27 +8,29 @@ import (
 	"net/http"
 	"net/url"
 	"path"
+	"strings"
+	"time"
 
 	authenticationv1 "k8s.io/api/authentication/v1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/httpstream"
 	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/proxy"
+	proxyutil "k8s.io/apimachinery/pkg/util/proxy"
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	"k8s.io/apiserver/pkg/endpoints/request"
-	"k8s.io/apiserver/pkg/registry/rest"
+	registryrest "k8s.io/apiserver/pkg/registry/rest"
 
 	clusterapis "github.com/karmada-io/karmada/pkg/apis/cluster"
 )
 
 // ConnectCluster returns a handler for proxy cluster.
-func ConnectCluster(ctx context.Context, cluster *clusterapis.Cluster, proxyPath string,
+func ConnectCluster(ctx context.Context, cluster *clusterapis.Cluster, proxyPath string, secretGetter func(context.Context, string, string) (*corev1.Secret, error), responder registryrest.Responder) (http.Handler, error) {
-	secretGetter func(context.Context, string, string) (*corev1.Secret, error), responder rest.Responder) (http.Handler, error) {
+	location, proxyTransport, err := Location(cluster)
-	location, transport, err := Location(cluster)
 	if err != nil {
 		return nil, err
 	}
-
 	location.Path = path.Join(location.Path, proxyPath)
 
 	if cluster.Spec.ImpersonatorSecretRef == nil {
@@ -45,12 +47,50 @@ func ConnectCluster(ctx context.Context, cluster *clusterapis.Cluster, proxyPath
 		return nil, fmt.Errorf("failed to get impresonateToken for cluster %s: %v", cluster.Name, err)
 	}
 
-	return newProxyHandler(location, transport, impersonateToken, responder)
+	return newProxyHandler(location, proxyTransport, cluster, impersonateToken, responder)
+}
+
+func newProxyHandler(location *url.URL, proxyTransport http.RoundTripper, cluster *clusterapis.Cluster, impersonateToken string, responder registryrest.Responder) (http.Handler, error) {
+	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		requester, exist := request.UserFrom(req.Context())
+		if !exist {
+			responsewriters.InternalError(rw, req, errors.New("no user found for request"))
+			return
+		}
+
+		req.Header.Set(authenticationv1.ImpersonateUserHeader, requester.GetName())
+		for _, group := range requester.GetGroups() {
+			if !skipGroup(group) {
+				req.Header.Add(authenticationv1.ImpersonateGroupHeader, group)
+			}
+		}
+		req.Header.Set("Authorization", fmt.Sprintf("bearer %s", impersonateToken))
+
+		var proxyURL *url.URL
+		if proxyURLStr := cluster.Spec.ProxyURL; proxyURLStr != "" {
+			proxyURL, _ = url.Parse(proxyURLStr)
+		}
+
+		// Retain RawQuery in location because upgrading the request will use it.
+		// See https://github.com/karmada-io/karmada/issues/1618#issuecomment-1103793290 for more info.
+		location.RawQuery = req.URL.RawQuery
+
+		upgradeDialer := NewUpgradeDialerWithConfig(UpgradeDialerWithConfig{
+			TLS:        &tls.Config{InsecureSkipVerify: true}, //nolint:gosec
+			Proxier:    http.ProxyURL(proxyURL),
+			PingPeriod: time.Second * 5,
+			Header:     ParseProxyHeaders(cluster.Spec.ProxyHeader),
+		})
+
+		handler := NewUpgradeAwareHandler(location, proxyTransport, false, httpstream.IsUpgradeRequest(req), proxyutil.NewErrorResponder(responder))
+		handler.UpgradeDialer = upgradeDialer
+		handler.ServeHTTP(rw, req)
+	}), nil
 }
 
 // NewThrottledUpgradeAwareProxyHandler creates a new proxy handler with a default flush interval. Responder is required for returning
 // errors to the caller.
-func NewThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder rest.Responder) *proxy.UpgradeAwareHandler {
+func NewThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder registryrest.Responder) *proxy.UpgradeAwareHandler {
 	return proxy.NewUpgradeAwareHandler(location, transport, wrapTransport, upgradeRequired, proxy.NewErrorResponder(responder))
 }
 
@@ -61,12 +101,12 @@ func Location(cluster *clusterapis.Cluster) (*url.URL, http.RoundTripper, error)
 		return nil, nil, err
 	}
 
-	transport, err := createProxyTransport(cluster)
+	proxyTransport, err := createProxyTransport(cluster)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	return location, transport, nil
+	return location, proxyTransport, nil
 }
 
 func constructLocation(cluster *clusterapis.Cluster) (*url.URL, error) {
@@ -96,10 +136,24 @@ func createProxyTransport(cluster *clusterapis.Cluster) (*http.Transport, error)
 			return nil, fmt.Errorf("failed to parse url of proxy url %s: %v", proxyURL, err)
 		}
 		trans.Proxy = http.ProxyURL(u)
+		trans.ProxyConnectHeader = ParseProxyHeaders(cluster.Spec.ProxyHeader)
 	}
 	return trans, nil
 }
 
+func ParseProxyHeaders(proxyHeaders map[string]string) http.Header {
+	if len(proxyHeaders) == 0 {
+		return nil
+	}
+
+	header := http.Header{}
+	for headerKey, headerValues := range proxyHeaders {
+		values := strings.Split(headerValues, ",")
+		header[headerKey] = values
+	}
+	return header
+}
+
 func getImpersonateToken(clusterName string, secret *corev1.Secret) (string, error) {
 	token, found := secret.Data[clusterapis.SecretTokenKey]
 	if !found {
@@ -108,31 +162,6 @@ func getImpersonateToken(clusterName string, secret *corev1.Secret) (string, err
 	return string(token), nil
 }
 
-func newProxyHandler(location *url.URL, transport http.RoundTripper, impersonateToken string, responder rest.Responder) (http.Handler, error) {
-	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
-		requester, exist := request.UserFrom(req.Context())
-		if !exist {
-			responsewriters.InternalError(rw, req, errors.New("no user found for request"))
-			return
-		}
-		req.Header.Set(authenticationv1.ImpersonateUserHeader, requester.GetName())
-		for _, group := range requester.GetGroups() {
-			if !skipGroup(group) {
-				req.Header.Add(authenticationv1.ImpersonateGroupHeader, group)
-			}
-		}
-
-		req.Header.Set("Authorization", fmt.Sprintf("bearer %s", impersonateToken))
-
-		// Retain RawQuery in location because upgrading the request will use it.
-		// See https://github.com/karmada-io/karmada/issues/1618#issuecomment-1103793290 for more info.
-		location.RawQuery = req.URL.RawQuery
-
-		handler := NewThrottledUpgradeAwareProxyHandler(location, transport, true, false, responder)
-		handler.ServeHTTP(rw, req)
-	}), nil
-}
-
 func skipGroup(group string) bool {
 	switch group {
 	case user.AllAuthenticated, user.AllUnauthenticated:

pkg/util/proxy/upgrade_dialer.go

@@ -0,0 +1,271 @@
+/*
+Copyright 2015 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 proxy
+
+import (
+	"context"
+	"crypto/tls"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"time"
+
+	"golang.org/x/net/proxy"
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+	"k8s.io/apimachinery/third_party/forked/golang/netutil"
+)
+
+// UpgradeDialer knows how to upgrade an HTTP request to one that supports
+// multiplexed streams. After Dial() is invoked, Conn will be usable.
+type UpgradeDialer struct {
+	//tlsConfig holds the TLS configuration settings to use when connecting
+	//to the remote server.
+	tlsConfig *tls.Config
+
+	// Dialer is the dialer used to connect.  Used if non-nil.
+	Dialer *net.Dialer
+
+	// header holds the HTTP request headers for dialing.
+	header http.Header
+
+	// proxier knows which proxy to use given a request.
+	proxier func(req *http.Request) (*url.URL, error)
+
+	// pingPeriod is a period for sending Ping frames over established
+	// connections.
+	pingPeriod time.Duration
+}
+
+var _ utilnet.TLSClientConfigHolder = &UpgradeDialer{}
+var _ utilnet.Dialer = &UpgradeDialer{}
+
+// NewUpgradeDialerWithConfig creates a new UpgradeRoundTripper with the specified
+// configuration.
+func NewUpgradeDialerWithConfig(cfg UpgradeDialerWithConfig) *UpgradeDialer {
+	if cfg.Proxier == nil {
+		cfg.Proxier = utilnet.NewProxierWithNoProxyCIDR(http.ProxyFromEnvironment)
+	}
+	return &UpgradeDialer{
+		tlsConfig:  cfg.TLS,
+		proxier:    cfg.Proxier,
+		pingPeriod: cfg.PingPeriod,
+		header:     cfg.Header,
+	}
+}
+
+// UpgradeDialerWithConfig is a set of options for an UpgradeDialer.
+type UpgradeDialerWithConfig struct {
+	// TLS configuration used by the round tripper.
+	TLS *tls.Config
+	// Header holds the HTTP request headers for dialing. Optional.
+	Header http.Header
+	// Proxier is a proxy function invoked on each request. Optional.
+	Proxier func(*http.Request) (*url.URL, error)
+	// PingPeriod is a period for sending Pings on the connection.
+	// Optional.
+	PingPeriod time.Duration
+}
+
+// TLSClientConfig implements pkg/util/net.TLSClientConfigHolder for proper TLS checking during
+// proxying with a roundtripper.
+func (u *UpgradeDialer) TLSClientConfig() *tls.Config {
+	return u.tlsConfig
+}
+
+// Dial implements k8s.io/apimachinery/pkg/util/net.Dialer.
+func (u *UpgradeDialer) Dial(req *http.Request) (net.Conn, error) {
+	conn, err := u.dial(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := req.Write(conn); err != nil {
+		conn.Close()
+		return nil, err
+	}
+
+	return conn, nil
+}
+
+// dial dials the host specified by req, using TLS if appropriate.
+func (u *UpgradeDialer) dial(req *http.Request) (net.Conn, error) {
+	proxyURL, err := u.proxier(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if proxyURL == nil {
+		return u.dialWithoutProxy(req.Context(), req.URL)
+	}
+
+	switch proxyURL.Scheme {
+	case "socks5":
+		return u.dialWithSocks5Proxy(req, proxyURL)
+	case "https", "http", "":
+		return u.dialWithHttpProxy(req, proxyURL)
+	}
+
+	return nil, fmt.Errorf("proxy URL scheme not supported: %s", proxyURL.Scheme)
+}
+
+// dialWithHttpProxy dials the host specified by url through an http or an https proxy.
+func (u *UpgradeDialer) dialWithHttpProxy(req *http.Request, proxyURL *url.URL) (net.Conn, error) {
+	// ensure we use a canonical host with proxyReq
+	targetHost := netutil.CanonicalAddr(req.URL)
+
+	// proxying logic adapted from http://blog.h6t.eu/post/74098062923/golang-websocket-with-http-proxy-support
+	proxyReq := http.Request{
+		Method: "CONNECT",
+		URL:    &url.URL{},
+		Host:   targetHost,
+		Header: u.header,
+	}
+
+	proxyReq = *proxyReq.WithContext(req.Context())
+
+	if pa := u.proxyAuth(proxyURL); pa != "" {
+		proxyReq.Header.Set("Proxy-Authorization", pa)
+	}
+
+	proxyDialConn, err := u.dialWithoutProxy(proxyReq.Context(), proxyURL)
+	if err != nil {
+		return nil, err
+	}
+
+	//nolint:staticcheck // SA1019 ignore deprecated httputil.NewProxyClientConn
+	proxyClientConn := httputil.NewProxyClientConn(proxyDialConn, nil)
+	response, err := proxyClientConn.Do(&proxyReq)
+	//nolint:staticcheck // SA1019 ignore deprecated httputil.ErrPersistEOF: it might be
+	// returned from the invocation of proxyClientConn.Do
+	if err != nil && err != httputil.ErrPersistEOF {
+		return nil, err
+	}
+	if response != nil && response.StatusCode >= 300 || response.StatusCode < 200 {
+		return nil, fmt.Errorf("CONNECT request to %s returned response: %s", proxyURL.Redacted(), response.Status)
+	}
+
+	rwc, _ := proxyClientConn.Hijack()
+
+	if req.URL.Scheme == "https" {
+		return u.tlsConn(proxyReq.Context(), rwc, targetHost)
+	}
+	return rwc, nil
+}
+
+// dialWithSocks5Proxy dials the host specified by url through a socks5 proxy.
+func (u *UpgradeDialer) dialWithSocks5Proxy(req *http.Request, proxyURL *url.URL) (net.Conn, error) {
+	// ensure we use a canonical host with proxyReq
+	targetHost := netutil.CanonicalAddr(req.URL)
+	proxyDialAddr := netutil.CanonicalAddr(proxyURL)
+
+	var auth *proxy.Auth
+	if proxyURL.User != nil {
+		pass, _ := proxyURL.User.Password()
+		auth = &proxy.Auth{
+			User:     proxyURL.User.Username(),
+			Password: pass,
+		}
+	}
+
+	dialer := u.Dialer
+	if dialer == nil {
+		dialer = &net.Dialer{
+			Timeout: 30 * time.Second,
+		}
+	}
+
+	proxyDialer, err := proxy.SOCKS5("tcp", proxyDialAddr, auth, dialer)
+	if err != nil {
+		return nil, err
+	}
+
+	// According to the implementation of proxy.SOCKS5, the type assertion will always succeed
+	contextDialer, ok := proxyDialer.(proxy.ContextDialer)
+	if !ok {
+		return nil, errors.New("SOCKS5 Dialer must implement ContextDialer")
+	}
+
+	proxyDialConn, err := contextDialer.DialContext(req.Context(), "tcp", targetHost)
+	if err != nil {
+		return nil, err
+	}
+
+	if req.URL.Scheme == "https" {
+		return u.tlsConn(req.Context(), proxyDialConn, targetHost)
+	}
+	return proxyDialConn, nil
+}
+
+// tlsConn returns a TLS client side connection using rwc as the underlying transport.
+func (u *UpgradeDialer) tlsConn(ctx context.Context, rwc net.Conn, targetHost string) (net.Conn, error) {
+	host, _, err := net.SplitHostPort(targetHost)
+	if err != nil {
+		return nil, err
+	}
+
+	tlsConfig := u.tlsConfig
+	switch {
+	case tlsConfig == nil:
+		tlsConfig = &tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}
+	case len(tlsConfig.ServerName) == 0:
+		tlsConfig = tlsConfig.Clone()
+		tlsConfig.ServerName = host
+	}
+
+	tlsConn := tls.Client(rwc, tlsConfig)
+
+	if err := tlsConn.HandshakeContext(ctx); err != nil {
+		tlsConn.Close()
+		return nil, err
+	}
+
+	return tlsConn, nil
+}
+
+// dialWithoutProxy dials the host specified by url, using TLS if appropriate.
+func (u *UpgradeDialer) dialWithoutProxy(ctx context.Context, url *url.URL) (net.Conn, error) {
+	dialAddr := netutil.CanonicalAddr(url)
+	dialer := u.Dialer
+	if dialer == nil {
+		dialer = &net.Dialer{}
+	}
+
+	if url.Scheme == "http" {
+		return dialer.DialContext(ctx, "tcp", dialAddr)
+	}
+
+	tlsDialer := tls.Dialer{
+		NetDialer: dialer,
+		Config:    u.tlsConfig,
+	}
+	return tlsDialer.DialContext(ctx, "tcp", dialAddr)
+}
+
+// proxyAuth returns, for a given proxy URL, the value to be used for the Proxy-Authorization header
+func (u *UpgradeDialer) proxyAuth(proxyURL *url.URL) string {
+	if proxyURL == nil || proxyURL.User == nil {
+		return ""
+	}
+	username := proxyURL.User.Username()
+	password, _ := proxyURL.User.Password()
+	auth := username + ":" + password
+	return "Basic " + base64.StdEncoding.EncodeToString([]byte(auth))
+}

pkg/util/proxy/upgrader.go

@@ -0,0 +1,419 @@
+/*
+Copyright 2017 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 proxy
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/mxk/go-flowrate/flowrate"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/httpstream"
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+	proxyutil "k8s.io/apimachinery/pkg/util/proxy"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/klog/v2"
+)
+
+type UpgradeAwareHandler struct {
+	proxyutil.UpgradeAwareHandler
+
+	UpgradeDialer *UpgradeDialer
+}
+
+const defaultFlushInterval = 200 * time.Millisecond
+
+// normalizeLocation returns the result of parsing the full URL, with scheme set to http if missing
+func normalizeLocation(location *url.URL) *url.URL {
+	normalized, _ := url.Parse(location.String())
+	if len(normalized.Scheme) == 0 {
+		normalized.Scheme = "http"
+	}
+	return normalized
+}
+
+// NewUpgradeAwareHandler creates a new proxy handler with a default flush interval. Responder is required for returning
+// errors to the caller.
+func NewUpgradeAwareHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired bool, responder proxyutil.ErrorResponder) *UpgradeAwareHandler {
+	return &UpgradeAwareHandler{
+		UpgradeAwareHandler: proxyutil.UpgradeAwareHandler{
+			Location:        normalizeLocation(location),
+			Transport:       transport,
+			WrapTransport:   wrapTransport,
+			UpgradeRequired: upgradeRequired,
+			FlushInterval:   defaultFlushInterval,
+			Responder:       responder,
+		},
+	}
+}
+
+func proxyRedirectsforRootPath(path string, w http.ResponseWriter, req *http.Request) bool {
+	redirect := false
+	method := req.Method
+
+	// From pkg/genericapiserver/endpoints/handlers/proxy.go#ServeHTTP:
+	// Redirect requests with an empty path to a location that ends with a '/'
+	// This is essentially a hack for https://issue.k8s.io/4958.
+	// Note: Keep this code after tryUpgrade to not break that flow.
+	if len(path) == 0 && (method == http.MethodGet || method == http.MethodHead) {
+		var queryPart string
+		if len(req.URL.RawQuery) > 0 {
+			queryPart = "?" + req.URL.RawQuery
+		}
+		w.Header().Set("Location", req.URL.Path+"/"+queryPart)
+		w.WriteHeader(http.StatusMovedPermanently)
+		redirect = true
+	}
+	return redirect
+}
+
+// ServeHTTP handles the proxy request
+//
+//nolint:gocyclo
+func (h *UpgradeAwareHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	if h.tryUpgrade(w, req) {
+		return
+	}
+	if h.UpgradeRequired {
+		h.Responder.Error(w, req, apierrors.NewBadRequest("Upgrade request required"))
+		return
+	}
+
+	loc := *h.Location
+	loc.RawQuery = req.URL.RawQuery
+
+	// If original request URL ended in '/', append a '/' at the end of the
+	// of the proxy URL
+	if !strings.HasSuffix(loc.Path, "/") && strings.HasSuffix(req.URL.Path, "/") {
+		loc.Path += "/"
+	}
+
+	proxyRedirect := proxyRedirectsforRootPath(loc.Path, w, req)
+	if proxyRedirect {
+		return
+	}
+
+	if h.Transport == nil || h.WrapTransport {
+		h.Transport = h.defaultProxyTransport(req.URL, h.Transport)
+	}
+
+	// WithContext creates a shallow clone of the request with the same context.
+	newReq := req.WithContext(req.Context())
+	newReq.Header = utilnet.CloneHeader(req.Header)
+	if !h.UseRequestLocation {
+		newReq.URL = &loc
+	}
+	if h.UseLocationHost {
+		// exchanging req.Host with the backend location is necessary for backends that act on the HTTP host header (e.g. API gateways),
+		// because req.Host has preference over req.URL.Host in filling this header field
+		newReq.Host = h.Location.Host
+	}
+
+	// create the target location to use for the reverse proxy
+	reverseProxyLocation := &url.URL{Scheme: h.Location.Scheme, Host: h.Location.Host}
+	if h.AppendLocationPath {
+		reverseProxyLocation.Path = h.Location.Path
+	}
+
+	proxy := httputil.NewSingleHostReverseProxy(reverseProxyLocation)
+	proxy.Transport = h.Transport
+	proxy.FlushInterval = h.FlushInterval
+	proxy.ErrorLog = log.New(noSuppressPanicError{}, "", log.LstdFlags)
+	if h.RejectForwardingRedirects {
+		oldModifyResponse := proxy.ModifyResponse
+		proxy.ModifyResponse = func(response *http.Response) error {
+			code := response.StatusCode
+			if code >= 300 && code <= 399 && len(response.Header.Get("Location")) > 0 {
+				// close the original response
+				response.Body.Close()
+				msg := "the backend attempted to redirect this request, which is not permitted"
+				// replace the response
+				*response = http.Response{
+					StatusCode:    http.StatusBadGateway,
+					Status:        fmt.Sprintf("%d %s", response.StatusCode, http.StatusText(response.StatusCode)),
+					Body:          io.NopCloser(strings.NewReader(msg)),
+					ContentLength: int64(len(msg)),
+				}
+			} else {
+				if oldModifyResponse != nil {
+					if err := oldModifyResponse(response); err != nil {
+						return err
+					}
+				}
+			}
+			return nil
+		}
+	}
+	if h.Responder != nil {
+		// if an optional error interceptor/responder was provided wire it
+		// the custom responder might be used for providing a unified error reporting
+		// or supporting retry mechanisms by not sending non-fatal errors to the clients
+		proxy.ErrorHandler = h.Responder.Error
+	}
+	proxy.ServeHTTP(w, newReq)
+}
+
+type noSuppressPanicError struct{}
+
+func (noSuppressPanicError) Write(p []byte) (n int, err error) {
+	// skip "suppressing panic for copyResponse error in test; copy error" error message
+	// that ends up in CI tests on each kube-apiserver termination as noise and
+	// everybody thinks this is fatal.
+	if strings.Contains(string(p), "suppressing panic") {
+		return len(p), nil
+	}
+	return os.Stderr.Write(p)
+}
+
+// tryUpgrade returns true if the request was handled.
+//
+//nolint:gocyclo
+func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Request) bool {
+	if !httpstream.IsUpgradeRequest(req) {
+		klog.V(6).Infof("Request was not an upgrade")
+		return false
+	}
+
+	var (
+		backendConn net.Conn
+		rawResponse []byte
+		err         error
+	)
+
+	location := *h.Location
+	if h.UseRequestLocation {
+		location = *req.URL
+		location.Scheme = h.Location.Scheme
+		location.Host = h.Location.Host
+		if h.AppendLocationPath {
+			location.Path = singleJoiningSlash(h.Location.Path, location.Path)
+		}
+	}
+
+	clone := utilnet.CloneRequest(req)
+	// Only append X-Forwarded-For in the upgrade path, since httputil.NewSingleHostReverseProxy
+	// handles this in the non-upgrade path.
+	utilnet.AppendForwardedForHeader(clone)
+	klog.V(6).Infof("Connecting to backend proxy (direct dial) %s\n  Headers: %v", &location, clone.Header)
+	if h.UseLocationHost {
+		clone.Host = h.Location.Host
+	}
+	clone.URL = &location
+
+	//backendConn, err = h.DialForUpgrade(clone)
+	backendConn, err = h.UpgradeDialer.Dial(clone)
+	if err != nil {
+		klog.V(6).Infof("Proxy connection error: %v", err)
+		h.Responder.Error(w, req, err)
+		return true
+	}
+	defer backendConn.Close()
+
+	// determine the http response code from the backend by reading from rawResponse+backendConn
+	backendHTTPResponse, headerBytes, err := getResponse(io.MultiReader(bytes.NewReader(rawResponse), backendConn))
+	if err != nil {
+		klog.V(6).Infof("Proxy connection error: %v", err)
+		h.Responder.Error(w, req, err)
+		return true
+	}
+	if len(headerBytes) > len(rawResponse) {
+		// we read beyond the bytes stored in rawResponse, update rawResponse to the full set of bytes read from the backend
+		rawResponse = headerBytes
+	}
+
+	// If the backend did not upgrade the request, return an error to the client. If the response was
+	// an error, the error is forwarded directly after the connection is hijacked. Otherwise, just
+	// return a generic error here.
+	if backendHTTPResponse.StatusCode != http.StatusSwitchingProtocols && backendHTTPResponse.StatusCode < 400 {
+		err := fmt.Errorf("invalid upgrade response: status code %d", backendHTTPResponse.StatusCode)
+		klog.Errorf("Proxy upgrade error: %v", err)
+		h.Responder.Error(w, req, err)
+		return true
+	}
+
+	// Once the connection is hijacked, the ErrorResponder will no longer work, so
+	// 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)
+		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)
+		h.Responder.Error(w, req, fmt.Errorf("error hijacking connection: %v", err))
+		return true
+	}
+	defer requestHijackedConn.Close()
+
+	if backendHTTPResponse.StatusCode != http.StatusSwitchingProtocols {
+		// If the backend did not upgrade the request, echo the response from the backend to the client and return, closing the connection.
+		klog.V(6).Infof("Proxy upgrade error, status code %d", backendHTTPResponse.StatusCode)
+		// set read/write deadlines
+		deadline := time.Now().Add(10 * time.Second)
+		_ = backendConn.SetReadDeadline(deadline)
+		_ = requestHijackedConn.SetWriteDeadline(deadline)
+		// write the response to the client
+		err := backendHTTPResponse.Write(requestHijackedConn)
+		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			klog.Errorf("Error proxying data from backend to client: %v", err)
+		}
+		// Indicate we handled the request
+		return true
+	}
+
+	// Forward raw response bytes back to client.
+	if len(rawResponse) > 0 {
+		klog.V(6).Infof("Writing %d bytes to hijacked connection", len(rawResponse))
+		if _, err = requestHijackedConn.Write(rawResponse); err != nil {
+			utilruntime.HandleError(fmt.Errorf("Error proxying response from backend to client: %v", err))
+		}
+	}
+
+	// Proxy the connection. This is bidirectional, so we need a goroutine
+	// to copy in each direction. Once one side of the connection exits, we
+	// exit the function which performs cleanup and in the process closes
+	// the other half of the connection in the defer.
+	writerComplete := make(chan struct{})
+	readerComplete := make(chan struct{})
+
+	go func() {
+		var writer io.WriteCloser
+		if h.MaxBytesPerSec > 0 {
+			writer = flowrate.NewWriter(backendConn, h.MaxBytesPerSec)
+		} else {
+			writer = backendConn
+		}
+		_, err := io.Copy(writer, requestHijackedConn)
+		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			klog.Errorf("Error proxying data from client to backend: %v", err)
+		}
+		close(writerComplete)
+	}()
+
+	go func() {
+		var reader io.ReadCloser
+		if h.MaxBytesPerSec > 0 {
+			reader = flowrate.NewReader(backendConn, h.MaxBytesPerSec)
+		} else {
+			reader = backendConn
+		}
+		_, err := io.Copy(requestHijackedConn, reader)
+		if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+			klog.Errorf("Error proxying data from backend to client: %v", err)
+		}
+		close(readerComplete)
+	}()
+
+	// Wait for one half the connection to exit. Once it does the defer will
+	// clean up the other half of the connection.
+	select {
+	case <-writerComplete:
+	case <-readerComplete:
+	}
+	klog.V(6).Infof("Disconnecting from backend proxy %s\n  Headers: %v", &location, clone.Header)
+
+	return true
+}
+
+// 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 {
+	aslash := strings.HasSuffix(a, "/")
+	bslash := strings.HasPrefix(b, "/")
+	switch {
+	case aslash && bslash:
+		return a + b[1:]
+	case !aslash && !bslash:
+		return a + "/" + b
+	}
+	return a + b
+}
+
+// getResponseCode reads a http response from the given reader, returns the response,
+// the bytes read from the reader, and any error encountered
+func getResponse(r io.Reader) (*http.Response, []byte, error) {
+	rawResponse := bytes.NewBuffer(make([]byte, 0, 256))
+	// Save the bytes read while reading the response headers into the rawResponse buffer
+	resp, err := http.ReadResponse(bufio.NewReader(io.TeeReader(r, rawResponse)), nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	// return the http response and the raw bytes consumed from the reader in the process
+	return resp, rawResponse.Bytes(), nil
+}
+
+func (h *UpgradeAwareHandler) defaultProxyTransport(url *url.URL, internalTransport http.RoundTripper) http.RoundTripper {
+	scheme := url.Scheme
+	host := url.Host
+	suffix := h.Location.Path
+	if strings.HasSuffix(url.Path, "/") && !strings.HasSuffix(suffix, "/") {
+		suffix += "/"
+	}
+	pathPrepend := strings.TrimSuffix(url.Path, suffix)
+	rewritingTransport := &proxyutil.Transport{
+		Scheme:       scheme,
+		Host:         host,
+		PathPrepend:  pathPrepend,
+		RoundTripper: internalTransport,
+	}
+	return &corsRemovingTransport{
+		RoundTripper: rewritingTransport,
+	}
+}
+
+// corsRemovingTransport is a wrapper for an internal transport. It removes CORS headers
+// from the internal response.
+// Implements pkg/util/net.RoundTripperWrapper
+type corsRemovingTransport struct {
+	http.RoundTripper
+}
+
+var _ = utilnet.RoundTripperWrapper(&corsRemovingTransport{})
+
+func (rt *corsRemovingTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := rt.RoundTripper.RoundTrip(req)
+	if err != nil {
+		return nil, err
+	}
+	removeCORSHeaders(resp)
+	return resp, nil
+}
+
+func (rt *corsRemovingTransport) WrappedRoundTripper() http.RoundTripper {
+	return rt.RoundTripper
+}
+
+// removeCORSHeaders strip CORS headers sent from the backend
+// This should be called on all responses before returning
+func removeCORSHeaders(resp *http.Response) {
+	resp.Header.Del("Access-Control-Allow-Credentials")
+	resp.Header.Del("Access-Control-Allow-Headers")
+	resp.Header.Del("Access-Control-Allow-Methods")
+	resp.Header.Del("Access-Control-Allow-Origin")
+}