Merge pull request #46126 from timstclair/forward-ip

Automatic merge from submit-queue (batch tested with PRs 42042, 46139, 46126, 46258, 46312)

Append X-Forwarded-For in proxy handler

Append the request sender's IP to the `X-Forwarded-For` header chain when proxying requests. This is important for audit logging (https://github.com/kubernetes/features/issues/22) in order to capture the client IP (specifically in the case of federation or kube-aggregator).

/cc @liggitt @deads2k @ericchiang @ihmccreery @soltysh

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

@@ -209,6 +209,21 @@ func GetClientIP(req *http.Request) net.IP {
 	return ips[0]
 }
 
+// Prepares the X-Forwarded-For header for another forwarding hop by appending the previous sender's
+// IP address to the X-Forwarded-For chain.
+func AppendForwardedForHeader(req *http.Request) {
+	// Copied from net/http/httputil/reverseproxy.go:
+	if clientIP, _, err := net.SplitHostPort(req.RemoteAddr); err == nil {
+		// If we aren't the first proxy retain prior
+		// X-Forwarded-For information as a comma+space
+		// separated list and fold multiple headers into one.
+		if prior, ok := req.Header["X-Forwarded-For"]; ok {
+			clientIP = strings.Join(prior, ", ") + ", " + clientIP
+		}
+		req.Header.Set("X-Forwarded-For", clientIP)
+	}
+}
+
 var defaultProxyFuncPointer = fmt.Sprintf("%p", http.ProxyFromEnvironment)
 
 // isDefault checks to see if the transportProxierFunc is pointing to the default one

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

@@ -108,6 +108,32 @@ func TestGetClientIP(t *testing.T) {
 	}
 }
 
+func TestAppendForwardedForHeader(t *testing.T) {
+	testCases := []struct {
+		addr, forwarded, expected string
+	}{
+		{"1.2.3.4:8000", "", "1.2.3.4"},
+		{"1.2.3.4:8000", "8.8.8.8", "8.8.8.8, 1.2.3.4"},
+		{"1.2.3.4:8000", "8.8.8.8, 1.2.3.4", "8.8.8.8, 1.2.3.4, 1.2.3.4"},
+		{"1.2.3.4:8000", "foo,bar", "foo,bar, 1.2.3.4"},
+	}
+	for i, test := range testCases {
+		req := &http.Request{
+			RemoteAddr: test.addr,
+			Header:     make(http.Header),
+		}
+		if test.forwarded != "" {
+			req.Header.Set("X-Forwarded-For", test.forwarded)
+		}
+
+		AppendForwardedForHeader(req)
+		actual := req.Header.Get("X-Forwarded-For")
+		if actual != test.expected {
+			t.Errorf("[%d] Expected %q, Got %q", i, test.expected, actual)
+		}
+	}
+}
+
 func TestProxierWithNoProxyCIDR(t *testing.T) {
 	testCases := []struct {
 		name    string

staging/src/k8s.io/apiserver/pkg/endpoints/handlers/proxy.go

@@ -17,6 +17,7 @@ limitations under the License.
 package handlers
 
 import (
+	"context"
 	"errors"
 	"io"
 	"math/rand"
@@ -156,17 +157,10 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	}
 	location.RawQuery = values.Encode()
 
-	newReq, err := http.NewRequest(req.Method, location.String(), req.Body)
+	// WithContext creates a shallow clone of the request with the new context.
-	if err != nil {
+	newReq := req.WithContext(context.Background())
-		httpCode = responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)
+	newReq.Header = net.CloneHeader(req.Header)
-		return
+	newReq.URL = location
-	}
-	httpCode = http.StatusOK
-	newReq.Header = req.Header
-	newReq.ContentLength = req.ContentLength
-	// Copy the TransferEncoding is for future-proofing. Currently Go only supports "chunked" and
-	// it can determine the TransferEncoding based on ContentLength and the Body.
-	newReq.TransferEncoding = req.TransferEncoding
 
 	// TODO convert this entire proxy to an UpgradeAwareProxy similar to
 	// https://github.com/openshift/origin/blob/master/pkg/util/httpproxy/upgradeawareproxy.go.
@@ -224,6 +218,10 @@ func (r *ProxyHandler) tryUpgrade(ctx request.Context, w http.ResponseWriter, re
 	if !httpstream.IsUpgradeRequest(req) {
 		return false
 	}
+	// Only append X-Forwarded-For in the upgrade path, since httputil.NewSingleHostReverseProxy
+	// handles this in the non-upgrade path.
+	net.AppendForwardedForHeader(newReq)
+
 	backendConn, err := proxyutil.DialURL(location, transport)
 	if err != nil {
 		responsewriters.ErrorNegotiated(ctx, err, r.Serializer, gv, w, req)

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

@@ -17,6 +17,7 @@ limitations under the License.
 package rest
 
 import (
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -116,16 +117,10 @@ func (h *UpgradeAwareProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Re
 		h.Transport = h.defaultProxyTransport(req.URL, h.Transport)
 	}
 
-	newReq, err := http.NewRequest(req.Method, loc.String(), req.Body)
+	// WithContext creates a shallow clone of the request with the new context.
-	if err != nil {
+	newReq := req.WithContext(context.Background())
-		h.Responder.Error(err)
+	newReq.Header = utilnet.CloneHeader(req.Header)
-		return
+	newReq.URL = &loc
-	}
-	newReq.Header = req.Header
-	newReq.ContentLength = req.ContentLength
-	// Copy the TransferEncoding is for future-proofing. Currently Go only supports "chunked" and
-	// it can determine the TransferEncoding based on ContentLength and the Body.
-	newReq.TransferEncoding = req.TransferEncoding
 
 	proxy := httputil.NewSingleHostReverseProxy(&url.URL{Scheme: h.Location.Scheme, Host: h.Location.Host})
 	proxy.Transport = h.Transport
@@ -145,10 +140,13 @@ func (h *UpgradeAwareProxyHandler) tryUpgrade(w http.ResponseWriter, req *http.R
 		err         error
 	)
 
+	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)
 	if h.InterceptRedirects && utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StreamingProxyRedirects) {
-		backendConn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, h.Location, req.Header, req.Body, h)
+		backendConn, rawResponse, err = utilnet.ConnectWithRedirects(req.Method, h.Location, clone.Header, req.Body, h)
 	} else {
-		clone := utilnet.CloneRequest(req)
 		clone.URL = h.Location
 		backendConn, err = h.Dial(clone)
 	}

staging/src/k8s.io/kube-aggregator/pkg/apiserver/BUILD

@@ -54,6 +54,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/httpstream:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/httpstream/spdy:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/net:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",

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

@@ -17,6 +17,7 @@ limitations under the License.
 package apiserver
 
 import (
+	"context"
 	"net/http"
 	"net/url"
 	"sync/atomic"
@@ -24,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/httpstream"
 	"k8s.io/apimachinery/pkg/util/httpstream/spdy"
+	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apiserver/pkg/endpoints/handlers/responsewriters"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericfeatures "k8s.io/apiserver/pkg/features"
@@ -110,21 +112,14 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	location.Path = req.URL.Path
 	location.RawQuery = req.URL.Query().Encode()
 
-	// make a new request object with the updated location and the body we already have
+	// WithContext creates a shallow clone of the request with the new context.
-	newReq, err := http.NewRequest(req.Method, location.String(), req.Body)
+	newReq := req.WithContext(context.Background())
-	if err != nil {
+	newReq.Header = utilnet.CloneHeader(req.Header)
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+	newReq.URL = location
-		return
-	}
-	mergeHeader(newReq.Header, req.Header)
-	newReq.ContentLength = req.ContentLength
-	// Copy the TransferEncoding is for future-proofing. Currently Go only supports "chunked" and
-	// it can determine the TransferEncoding based on ContentLength and the Body.
-	newReq.TransferEncoding = req.TransferEncoding
 
 	upgrade := false
 	// we need to wrap the roundtripper in another roundtripper which will apply the front proxy headers
-	proxyRoundTripper, upgrade, err = maybeWrapForConnectionUpgrades(handlingInfo.restConfig, proxyRoundTripper, req)
+	proxyRoundTripper, upgrade, err := maybeWrapForConnectionUpgrades(handlingInfo.restConfig, proxyRoundTripper, req)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
@@ -163,14 +158,6 @@ func maybeWrapForConnectionUpgrades(restConfig *restclient.Config, rt http.Round
 	return wrappedRT, true, nil
 }
 
-func mergeHeader(dst, src http.Header) {
-	for k, vv := range src {
-		for _, v := range vv {
-			dst.Add(k, v)
-		}
-	}
-}
-
 // responder implements rest.Responder for assisting a connector in writing objects or errors.
 type responder struct {
 	w http.ResponseWriter

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

@@ -128,6 +128,7 @@ func TestProxyHandler(t *testing.T) {
 			expectedHeaders: map[string][]string{
 				"X-Forwarded-Proto": {"https"},
 				"X-Forwarded-Uri":   {"/request/path"},
+				"X-Forwarded-For":   {"127.0.0.1"},
 				"X-Remote-User":     {"username"},
 				"User-Agent":        {"Go-http-client/1.1"},
 				"Accept-Encoding":   {"gzip"},