Merge pull request #66516 from tallclair/redirect

Add verification to apiserver redirect following

pkg/kubelet/server/server_test.go

@@ -51,6 +51,7 @@ import (
 	api "k8s.io/kubernetes/pkg/apis/core"
 	runtimeapi "k8s.io/kubernetes/pkg/kubelet/apis/cri/runtime/v1alpha2"
 	statsapi "k8s.io/kubernetes/pkg/kubelet/apis/stats/v1alpha1"
+
 	// Do some initialization to decode the query parameters correctly.
 	_ "k8s.io/kubernetes/pkg/apis/core/install"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
@@ -1166,7 +1167,7 @@ func TestServeExecInContainerIdleTimeout(t *testing.T) {
 
 	url := fw.testHTTPServer.URL + "/exec/" + podNamespace + "/" + podName + "/" + expectedContainerName + "?c=ls&c=-a&" + api.ExecStdinParam + "=1"
 
-	upgradeRoundTripper := spdy.NewSpdyRoundTripper(nil, true)
+	upgradeRoundTripper := spdy.NewSpdyRoundTripper(nil, true, true)
 	c := &http.Client{Transport: upgradeRoundTripper}
 
 	resp, err := c.Post(url, "", nil)
@@ -1332,7 +1333,7 @@ func testExecAttach(t *testing.T, verb string) {
 					return http.ErrUseLastResponse
 				}
 			} else {
-				upgradeRoundTripper = spdy.NewRoundTripper(nil, true)
+				upgradeRoundTripper = spdy.NewRoundTripper(nil, true, true)
 				c = &http.Client{Transport: upgradeRoundTripper}
 			}
 
@@ -1429,7 +1430,7 @@ func TestServePortForwardIdleTimeout(t *testing.T) {
 
 	url := fw.testHTTPServer.URL + "/portForward/" + podNamespace + "/" + podName
 
-	upgradeRoundTripper := spdy.NewRoundTripper(nil, true)
+	upgradeRoundTripper := spdy.NewRoundTripper(nil, true, true)
 	c := &http.Client{Transport: upgradeRoundTripper}
 
 	resp, err := c.Post(url, "", nil)
@@ -1536,7 +1537,7 @@ func TestServePortForward(t *testing.T) {
 					return http.ErrUseLastResponse
 				}
 			} else {
-				upgradeRoundTripper = spdy.NewRoundTripper(nil, true)
+				upgradeRoundTripper = spdy.NewRoundTripper(nil, true, true)
 				c = &http.Client{Transport: upgradeRoundTripper}
 			}
 

pkg/registry/core/pod/rest/log.go

@@ -80,6 +80,7 @@ func (r *LogREST) Get(ctx context.Context, name string, opts runtime.Object) (ru
 		ContentType:     "text/plain",
 		Flush:           logOpts.Follow,
 		ResponseChecker: genericrest.NewGenericHttpResponseChecker(api.Resource("pods/log"), name),
+		RedirectChecker: genericrest.PreventRedirects,
 	}, nil
 }
 

pkg/registry/core/pod/rest/subresources.go

@@ -194,6 +194,7 @@ func (r *PortForwardREST) Connect(ctx context.Context, name string, opts runtime
 func newThrottledUpgradeAwareProxyHandler(location *url.URL, transport http.RoundTripper, wrapTransport, upgradeRequired, interceptRedirects bool, responder rest.Responder) *proxy.UpgradeAwareHandler {
 	handler := proxy.NewUpgradeAwareHandler(location, transport, wrapTransport, upgradeRequired, proxy.NewErrorResponder(responder))
 	handler.InterceptRedirects = interceptRedirects && utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects)
+	handler.RequireSameHostRedirects = utilfeature.DefaultFeatureGate.Enabled(genericfeatures.ValidateProxyRedirects)
 	handler.MaxBytesPerSec = capabilities.Get().PerConnectionBandwidthLimitBytesPerSec
 	return handler
 }

staging/src/k8s.io/apimachinery/pkg/util/httpstream/spdy/roundtripper.go

@@ -67,6 +67,9 @@ type SpdyRoundTripper struct {
 	// followRedirects indicates if the round tripper should examine responses for redirects and
 	// follow them.
 	followRedirects bool
+	// requireSameHostRedirects restricts redirect following to only follow redirects to the same host
+	// as the original request.
+	requireSameHostRedirects bool
 }
 
 var _ utilnet.TLSClientConfigHolder = &SpdyRoundTripper{}
@@ -75,14 +78,18 @@ var _ utilnet.Dialer = &SpdyRoundTripper{}
 
 // NewRoundTripper creates a new SpdyRoundTripper that will use
 // the specified tlsConfig.
-func NewRoundTripper(tlsConfig *tls.Config, followRedirects bool) httpstream.UpgradeRoundTripper {
+func NewRoundTripper(tlsConfig *tls.Config, followRedirects, requireSameHostRedirects bool) httpstream.UpgradeRoundTripper {
-	return NewSpdyRoundTripper(tlsConfig, followRedirects)
+	return NewSpdyRoundTripper(tlsConfig, followRedirects, requireSameHostRedirects)
 }
 
 // NewSpdyRoundTripper creates a new SpdyRoundTripper that will use
 // the specified tlsConfig. This function is mostly meant for unit tests.
-func NewSpdyRoundTripper(tlsConfig *tls.Config, followRedirects bool) *SpdyRoundTripper {
+func NewSpdyRoundTripper(tlsConfig *tls.Config, followRedirects, requireSameHostRedirects bool) *SpdyRoundTripper {
-	return &SpdyRoundTripper{tlsConfig: tlsConfig, followRedirects: followRedirects}
+	return &SpdyRoundTripper{
+		tlsConfig:                tlsConfig,
+		followRedirects:          followRedirects,
+		requireSameHostRedirects: requireSameHostRedirects,
+	}
 }
 
 // TLSClientConfig implements pkg/util/net.TLSClientConfigHolder for proper TLS checking during
@@ -257,7 +264,7 @@ func (s *SpdyRoundTripper) RoundTrip(req *http.Request) (*http.Response, error)
 	)
 
 	if s.followRedirects {
-		conn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, req.URL, header, req.Body, s)
+		conn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, req.URL, header, req.Body, s, s.requireSameHostRedirects)
 	} else {
 		clone := utilnet.CloneRequest(req)
 		clone.Header = header

staging/src/k8s.io/apimachinery/pkg/util/httpstream/spdy/roundtripper_test.go

@@ -282,7 +282,7 @@ func TestRoundTripAndNewConnection(t *testing.T) {
 					t.Fatalf("%s: Error creating request: %s", k, err)
 				}
 
-				spdyTransport := NewSpdyRoundTripper(testCase.clientTLS, redirect)
+				spdyTransport := NewSpdyRoundTripper(testCase.clientTLS, redirect, redirect)
 
 				var proxierCalled bool
 				var proxyCalledWithHost string
@@ -391,8 +391,8 @@ func TestRoundTripRedirects(t *testing.T) {
 	}{
 		{0, true},
 		{1, true},
-		{10, true},
+		{9, true},
-		{11, false},
+		{10, false},
 	}
 	for _, test := range tests {
 		t.Run(fmt.Sprintf("with %d redirects", test.redirects), func(t *testing.T) {
@@ -425,7 +425,7 @@ func TestRoundTripRedirects(t *testing.T) {
 				t.Fatalf("Error creating request: %s", err)
 			}
 
-			spdyTransport := NewSpdyRoundTripper(nil, true)
+			spdyTransport := NewSpdyRoundTripper(nil, true, true)
 			client := &http.Client{Transport: spdyTransport}
 
 			resp, err := client.Do(req)

staging/src/k8s.io/apimachinery/pkg/util/net/BUILD

@@ -16,7 +16,12 @@ go_test(
         "util_test.go",
     ],
     embed = [":go_default_library"],
-    deps = ["//vendor/github.com/spf13/pflag:go_default_library"],
+    deps = [
+        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/github.com/spf13/pflag:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/github.com/stretchr/testify/require:go_default_library",
+    ],
 )
 
 go_library(

staging/src/k8s.io/apimachinery/pkg/util/net/http.go

@@ -321,9 +321,10 @@ type Dialer interface {
 
 // ConnectWithRedirects uses dialer to send req, following up to 10 redirects (relative to
 // originalLocation). It returns the opened net.Conn and the raw response bytes.
-func ConnectWithRedirects(originalMethod string, originalLocation *url.URL, header http.Header, originalBody io.Reader, dialer Dialer) (net.Conn, []byte, error) {
+// If requireSameHostRedirects is true, only redirects to the same host are permitted.
+func ConnectWithRedirects(originalMethod string, originalLocation *url.URL, header http.Header, originalBody io.Reader, dialer Dialer, requireSameHostRedirects bool) (net.Conn, []byte, error) {
 	const (
-		maxRedirects    = 10
+		maxRedirects    = 9     // Fail on the 10th redirect
 		maxResponseSize = 16384 // play it safe to allow the potential for lots of / large headers
 	)
 
@@ -387,10 +388,6 @@ redirectLoop:
 
 		resp.Body.Close() // not used
 
-		// Reset the connection.
-		intermediateConn.Close()
-		intermediateConn = nil
-
 		// Prepare to follow the redirect.
 		redirectStr := resp.Header.Get("Location")
 		if redirectStr == "" {
@@ -404,6 +401,15 @@ redirectLoop:
 		if err != nil {
 			return nil, nil, fmt.Errorf("malformed Location header: %v", err)
 		}
+
+		// Only follow redirects to the same host. Otherwise, propagate the redirect response back.
+		if requireSameHostRedirects && location.Hostname() != originalLocation.Hostname() {
+			break redirectLoop
+		}
+
+		// Reset the connection.
+		intermediateConn.Close()
+		intermediateConn = nil
 	}
 
 	connToReturn := intermediateConn

staging/src/k8s.io/apimachinery/pkg/util/net/http_test.go

@@ -19,14 +19,23 @@ limitations under the License.
 package net
 
 import (
+	"bufio"
+	"bytes"
 	"crypto/tls"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"os"
 	"reflect"
+	"strings"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 func TestGetClientIP(t *testing.T) {
@@ -280,3 +289,153 @@ func TestJoinPreservingTrailingSlash(t *testing.T) {
 		})
 	}
 }
+
+func TestConnectWithRedirects(t *testing.T) {
+	tests := []struct {
+		desc              string
+		redirects         []string
+		method            string // initial request method, empty == GET
+		expectError       bool
+		expectedRedirects int
+		newPort           bool // special case different port test
+	}{{
+		desc:              "relative redirects allowed",
+		redirects:         []string{"/ok"},
+		expectedRedirects: 1,
+	}, {
+		desc:              "redirects to the same host are allowed",
+		redirects:         []string{"http://HOST/ok"}, // HOST replaced with server address in test
+		expectedRedirects: 1,
+	}, {
+		desc:              "POST redirects to GET",
+		method:            http.MethodPost,
+		redirects:         []string{"/ok"},
+		expectedRedirects: 1,
+	}, {
+		desc:              "PUT redirects to GET",
+		method:            http.MethodPut,
+		redirects:         []string{"/ok"},
+		expectedRedirects: 1,
+	}, {
+		desc:              "DELETE redirects to GET",
+		method:            http.MethodDelete,
+		redirects:         []string{"/ok"},
+		expectedRedirects: 1,
+	}, {
+		desc:              "9 redirects are allowed",
+		redirects:         []string{"/1", "/2", "/3", "/4", "/5", "/6", "/7", "/8", "/9"},
+		expectedRedirects: 9,
+	}, {
+		desc:        "10 redirects are forbidden",
+		redirects:   []string{"/1", "/2", "/3", "/4", "/5", "/6", "/7", "/8", "/9", "/10"},
+		expectError: true,
+	}, {
+		desc:              "redirect to different host are prevented",
+		redirects:         []string{"http://example.com/foo"},
+		expectedRedirects: 0,
+	}, {
+		desc:              "multiple redirect to different host forbidden",
+		redirects:         []string{"/1", "/2", "/3", "http://example.com/foo"},
+		expectedRedirects: 3,
+	}, {
+		desc:              "redirect to different port is allowed",
+		redirects:         []string{"http://HOST/foo"},
+		expectedRedirects: 1,
+		newPort:           true,
+	}}
+
+	const resultString = "Test output"
+	for _, test := range tests {
+		t.Run(test.desc, func(t *testing.T) {
+			redirectCount := 0
+			s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+				// Verify redirect request.
+				if redirectCount > 0 {
+					expectedURL, err := url.Parse(test.redirects[redirectCount-1])
+					require.NoError(t, err, "test URL error")
+					assert.Equal(t, req.URL.Path, expectedURL.Path, "unknown redirect path")
+					assert.Equal(t, http.MethodGet, req.Method, "redirects must always be GET")
+				}
+				if redirectCount < len(test.redirects) {
+					http.Redirect(w, req, test.redirects[redirectCount], http.StatusFound)
+					redirectCount++
+				} else if redirectCount == len(test.redirects) {
+					w.Write([]byte(resultString))
+				} else {
+					t.Errorf("unexpected number of redirects %d to %s", redirectCount, req.URL.String())
+				}
+			}))
+			defer s.Close()
+
+			u, err := url.Parse(s.URL)
+			require.NoError(t, err, "Error parsing server URL")
+			host := u.Host
+
+			// Special case new-port test with a secondary server.
+			if test.newPort {
+				s2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+					w.Write([]byte(resultString))
+				}))
+				defer s2.Close()
+				u2, err := url.Parse(s2.URL)
+				require.NoError(t, err, "Error parsing secondary server URL")
+
+				// Sanity check: secondary server uses same hostname, different port.
+				require.Equal(t, u.Hostname(), u2.Hostname(), "sanity check: same hostname")
+				require.NotEqual(t, u.Port(), u2.Port(), "sanity check: different port")
+
+				// Redirect to the secondary server.
+				host = u2.Host
+
+			}
+
+			// Update redirect URLs with actual host.
+			for i := range test.redirects {
+				test.redirects[i] = strings.Replace(test.redirects[i], "HOST", host, 1)
+			}
+
+			method := test.method
+			if method == "" {
+				method = http.MethodGet
+			}
+
+			netdialer := &net.Dialer{
+				Timeout:   wait.ForeverTestTimeout,
+				KeepAlive: wait.ForeverTestTimeout,
+			}
+			dialer := DialerFunc(func(req *http.Request) (net.Conn, error) {
+				conn, err := netdialer.Dial("tcp", req.URL.Host)
+				if err != nil {
+					return conn, err
+				}
+				if err = req.Write(conn); err != nil {
+					require.NoError(t, conn.Close())
+					return nil, fmt.Errorf("error sending request: %v", err)
+				}
+				return conn, err
+			})
+			conn, rawResponse, err := ConnectWithRedirects(method, u, http.Header{} /*body*/, nil, dialer, true)
+			if test.expectError {
+				require.Error(t, err, "expected request error")
+				return
+			}
+
+			require.NoError(t, err, "unexpected request error")
+			assert.NoError(t, conn.Close(), "error closing connection")
+
+			resp, err := http.ReadResponse(bufio.NewReader(bytes.NewReader(rawResponse)), nil)
+			require.NoError(t, err, "unexpected request error")
+
+			result, err := ioutil.ReadAll(resp.Body)
+			require.NoError(t, resp.Body.Close())
+			if test.expectedRedirects < len(test.redirects) {
+				// Expect the last redirect to be returned.
+				assert.Equal(t, http.StatusFound, resp.StatusCode, "Final response is not a redirect")
+				assert.Equal(t, test.redirects[len(test.redirects)-1], resp.Header.Get("Location"))
+				assert.NotEqual(t, resultString, string(result), "wrong content")
+			} else {
+				assert.Equal(t, resultString, string(result), "stream content does not match")
+			}
+		})
+	}
+}

staging/src/k8s.io/apimachinery/pkg/util/proxy/upgradeaware.go

@@ -68,6 +68,8 @@ type UpgradeAwareHandler struct {
 	// InterceptRedirects determines whether the proxy should sniff backend responses for redirects,
 	// following them as necessary.
 	InterceptRedirects bool
+	// RequireSameHostRedirects only allows redirects to the same host. It is only used if InterceptRedirects=true.
+	RequireSameHostRedirects bool
 	// UseRequestLocation will use the incoming request URL when talking to the backend server.
 	UseRequestLocation bool
 	// FlushInterval controls how often the standard HTTP proxy will flush content from the upstream.
@@ -256,7 +258,7 @@ func (h *UpgradeAwareHandler) tryUpgrade(w http.ResponseWriter, req *http.Reques
 	utilnet.AppendForwardedForHeader(clone)
 	if h.InterceptRedirects {
 		glog.V(6).Infof("Connecting to backend proxy (intercepting redirects) %s\n  Headers: %v", &location, clone.Header)
-		backendConn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, &location, clone.Header, req.Body, utilnet.DialerFunc(h.DialForUpgrade))
+		backendConn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, &location, clone.Header, req.Body, utilnet.DialerFunc(h.DialForUpgrade), h.RequireSameHostRedirects)
 	} else {
 		glog.V(6).Infof("Connecting to backend proxy (direct dial) %s\n  Headers: %v", &location, clone.Header)
 		clone.URL = &location

staging/src/k8s.io/apiserver/pkg/features/kube_features.go

@@ -29,11 +29,19 @@ const (
 
 	// owner: @tallclair
 	// alpha: v1.5
+	// beta: v1.6
 	//
 	// StreamingProxyRedirects controls whether the apiserver should intercept (and follow)
 	// redirects from the backend (Kubelet) for streaming requests (exec/attach/port-forward).
 	StreamingProxyRedirects utilfeature.Feature = "StreamingProxyRedirects"
 
+	// owner: @tallclair
+	// alpha: v1.10
+	//
+	// ValidateProxyRedirects controls whether the apiserver should validate that redirects are only
+	// followed to the same host. Only used if StreamingProxyRedirects is enabled.
+	ValidateProxyRedirects utilfeature.Feature = "ValidateProxyRedirects"
+
 	// owner: @tallclair
 	// alpha: v1.7
 	// beta: v1.8
@@ -83,6 +91,7 @@ func init() {
 // available throughout Kubernetes binaries.
 var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureSpec{
 	StreamingProxyRedirects: {Default: true, PreRelease: utilfeature.Beta},
+	ValidateProxyRedirects:  {Default: false, PreRelease: utilfeature.Alpha},
 	AdvancedAuditing:        {Default: true, PreRelease: utilfeature.GA},
 	APIResponseCompression:  {Default: false, PreRelease: utilfeature.Alpha},
 	Initializers:            {Default: false, PreRelease: utilfeature.Alpha},

staging/src/k8s.io/apiserver/pkg/registry/generic/rest/BUILD

@@ -17,6 +17,8 @@ go_test(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//vendor/github.com/stretchr/testify/assert:go_default_library",
+        "//vendor/github.com/stretchr/testify/require:go_default_library",
     ],
 )
 

staging/src/k8s.io/apiserver/pkg/registry/generic/rest/streamer.go

@@ -18,6 +18,7 @@ package rest
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net/http"
 	"net/url"
@@ -29,13 +30,14 @@ import (
 )
 
 // LocationStreamer is a resource that streams the contents of a particular
-// location URL
+// location URL.
 type LocationStreamer struct {
 	Location        *url.URL
 	Transport       http.RoundTripper
 	ContentType     string
 	Flush           bool
 	ResponseChecker HttpResponseChecker
+	RedirectChecker func(req *http.Request, via []*http.Request) error
 }
 
 // a LocationStreamer must implement a rest.ResourceStreamer
@@ -59,7 +61,10 @@ func (s *LocationStreamer) InputStream(ctx context.Context, apiVersion, acceptHe
 	if transport == nil {
 		transport = http.DefaultTransport
 	}
-	client := &http.Client{Transport: transport}
+	client := &http.Client{
+		Transport:     transport,
+		CheckRedirect: s.RedirectChecker,
+	}
 	req, err := http.NewRequest("GET", s.Location.String(), nil)
 	// Pass the parent context down to the request to ensure that the resources
 	// will be release properly.
@@ -87,3 +92,8 @@ func (s *LocationStreamer) InputStream(ctx context.Context, apiVersion, acceptHe
 	stream = resp.Body
 	return
 }
+
+// PreventRedirects is a redirect checker that prevents the client from following a redirect.
+func PreventRedirects(_ *http.Request, _ []*http.Request) error {
+	return errors.New("redirects forbidden")
+}

staging/src/k8s.io/apiserver/pkg/registry/generic/rest/streamer_test.go

@@ -28,6 +28,8 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
@@ -147,3 +149,23 @@ func TestInputStreamInternalServerErrorTransport(t *testing.T) {
 		t.Errorf("StreamInternalServerError does not match. Got: %s. Expected: %s.", err, expectedError)
 	}
 }
+
+func TestInputStreamRedirects(t *testing.T) {
+	const redirectPath = "/redirect"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		if req.URL.Path == redirectPath {
+			t.Fatal("Redirects should not be followed")
+		} else {
+			http.Redirect(w, req, redirectPath, http.StatusFound)
+		}
+	}))
+	loc, err := url.Parse(s.URL)
+	require.NoError(t, err, "Error parsing server URL")
+
+	streamer := &LocationStreamer{
+		Location:        loc,
+		RedirectChecker: PreventRedirects,
+	}
+	_, _, _, err = streamer.InputStream(context.Background(), "", "")
+	assert.Error(t, err, "Redirect should trigger an error")
+}

staging/src/k8s.io/client-go/transport/spdy/spdy.go

@@ -38,7 +38,7 @@ func RoundTripperFor(config *restclient.Config) (http.RoundTripper, Upgrader, er
 	if err != nil {
 		return nil, nil, err
 	}
-	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, true)
+	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, true, false)
 	wrapper, err := restclient.HTTPWrappersForConfig(config, upgradeRoundTripper)
 	if err != nil {
 		return nil, nil, err

staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go

@@ -161,7 +161,8 @@ func maybeWrapForConnectionUpgrades(restConfig *restclient.Config, rt http.Round
 		return nil, true, err
 	}
 	followRedirects := utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects)
-	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, followRedirects)
+	requireSameHostRedirects := utilfeature.DefaultFeatureGate.Enabled(genericfeatures.ValidateProxyRedirects)
+	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig, followRedirects, requireSameHostRedirects)
 	wrappedRT, err := restclient.HTTPWrappersForConfig(restConfig, upgradeRoundTripper)
 	if err != nil {
 		return nil, true, err