Merge pull request kubernetes#81443 from mikedanese/socks5

rest.Config: support configuring an explict proxy URL

Author: k8s-ci-robot

pkg/kubelet/server/server_test.go

@@ -1034,7 +1034,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, true)
+	upgradeRoundTripper := spdy.NewRoundTripper(nil, true, true)
 	c := &http.Client{Transport: upgradeRoundTripper}
 
 	resp, err := c.Do(makeReq(t, "POST", url, "v4.channel.k8s.io"))

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

@@ -76,19 +76,20 @@ var _ utilnet.TLSClientConfigHolder = &SpdyRoundTripper{}
 var _ httpstream.UpgradeRoundTripper = &SpdyRoundTripper{}
 var _ utilnet.Dialer = &SpdyRoundTripper{}
 
-// NewRoundTripper creates a new SpdyRoundTripper that will use
-// the specified tlsConfig.
-func NewRoundTripper(tlsConfig *tls.Config, followRedirects, requireSameHostRedirects bool) httpstream.UpgradeRoundTripper {
-	return NewSpdyRoundTripper(tlsConfig, followRedirects, requireSameHostRedirects)
+// NewRoundTripper creates a new SpdyRoundTripper that will use the specified
+// tlsConfig.
+func NewRoundTripper(tlsConfig *tls.Config, followRedirects, requireSameHostRedirects bool) *SpdyRoundTripper {
+	return NewRoundTripperWithProxy(tlsConfig, followRedirects, requireSameHostRedirects, utilnet.NewProxierWithNoProxyCIDR(http.ProxyFromEnvironment))
 }
 
-// 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, requireSameHostRedirects bool) *SpdyRoundTripper {
+// NewRoundTripperWithProxy creates a new SpdyRoundTripper that will use the
+// specified tlsConfig and proxy func.
+func NewRoundTripperWithProxy(tlsConfig *tls.Config, followRedirects, requireSameHostRedirects bool, proxier func(*http.Request) (*url.URL, error)) *SpdyRoundTripper {
 	return &SpdyRoundTripper{
 		tlsConfig:                tlsConfig,
 		followRedirects:          followRedirects,
 		requireSameHostRedirects: requireSameHostRedirects,
+		proxier:                  proxier,
 	}
 }
 
@@ -116,11 +117,7 @@ func (s *SpdyRoundTripper) Dial(req *http.Request) (net.Conn, error) {
 // dial dials the host specified by req, using TLS if appropriate, optionally
 // using a proxy server if one is configured via environment variables.
 func (s *SpdyRoundTripper) dial(req *http.Request) (net.Conn, error) {
-	proxier := s.proxier
-	if proxier == nil {
-		proxier = utilnet.NewProxierWithNoProxyCIDR(http.ProxyFromEnvironment)
-	}
-	proxyURL, err := proxier(req)
+	proxyURL, err := s.proxier(req)
 	if err != nil {
 		return nil, err
 	}

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, redirect)
+				spdyTransport := NewRoundTripper(testCase.clientTLS, redirect, redirect)
 
 				var proxierCalled bool
 				var proxyCalledWithHost string
@@ -425,7 +425,7 @@ func TestRoundTripRedirects(t *testing.T) {
 				t.Fatalf("Error creating request: %s", err)
 			}
 
-			spdyTransport := NewSpdyRoundTripper(nil, true, true)
+			spdyTransport := NewRoundTripper(nil, true, true)
 			client := &http.Client{Transport: spdyTransport}
 
 			resp, err := client.Do(req)

staging/src/k8s.io/client-go/rest/BUILD

@@ -38,6 +38,7 @@ go_test(
         "//staging/src/k8s.io/client-go/transport:go_default_library",
         "//staging/src/k8s.io/client-go/util/flowcontrol:go_default_library",
         "//staging/src/k8s.io/client-go/util/testing:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
         "//vendor/github.com/google/gofuzz:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
         "//vendor/k8s.io/klog:go_default_library",

staging/src/k8s.io/client-go/rest/client_test.go

@@ -18,16 +18,16 @@ package rest
 
 import (
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"net/http/httputil"
 	"net/url"
 	"os"
 	"reflect"
 	"testing"
 	"time"
 
-	"fmt"
-
 	v1 "k8s.io/api/core/v1"
 	v1beta1 "k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -37,6 +37,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/client-go/kubernetes/scheme"
 	utiltesting "k8s.io/client-go/util/testing"
+
+	"github.com/google/go-cmp/cmp"
 )
 
 type TestParam struct {
@@ -252,7 +254,7 @@ func validate(testParam TestParam, t *testing.T, body []byte, fakeHandler *utilt
 
 }
 
-func TestHttpMethods(t *testing.T) {
+func TestHTTPMethods(t *testing.T) {
 	testServer, _, _ := testServerEnv(t, 200)
 	defer testServer.Close()
 	c, _ := restClient(testServer)
@@ -283,6 +285,57 @@ func TestHttpMethods(t *testing.T) {
 	}
 }
 
+func TestHTTPProxy(t *testing.T) {
+	ctx := context.Background()
+	testServer, fh, _ := testServerEnv(t, 200)
+	fh.ResponseBody = "backend data"
+	defer testServer.Close()
+
+	testProxyServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		to, err := url.Parse(req.RequestURI)
+		if err != nil {
+			t.Fatalf("err: %v", err)
+		}
+		w.Write([]byte("proxied: "))
+		httputil.NewSingleHostReverseProxy(to).ServeHTTP(w, req)
+	}))
+	defer testProxyServer.Close()
+
+	t.Logf(testProxyServer.URL)
+
+	u, err := url.Parse(testProxyServer.URL)
+	if err != nil {
+		t.Fatalf("Failed to parse test proxy server url: %v", err)
+	}
+
+	c, err := RESTClientFor(&Config{
+		Host: testServer.URL,
+		ContentConfig: ContentConfig{
+			GroupVersion:         &v1.SchemeGroupVersion,
+			NegotiatedSerializer: scheme.Codecs.WithoutConversion(),
+		},
+		Proxy:    http.ProxyURL(u),
+		Username: "user",
+		Password: "pass",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+
+	request := c.Get()
+	if request == nil {
+		t.Fatalf("Get: Object returned should not be nil")
+	}
+
+	b, err := request.DoRaw(ctx)
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if got, want := string(b), "proxied: backend data"; !cmp.Equal(got, want) {
+		t.Errorf("unexpected body: %v", cmp.Diff(want, got))
+	}
+}
+
 func TestCreateBackoffManager(t *testing.T) {
 
 	theUrl, _ := url.Parse("http://localhost")

staging/src/k8s.io/client-go/rest/config.go

@@ -23,6 +23,7 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
 	gruntime "runtime"
@@ -128,6 +129,13 @@ type Config struct {
 	// Dial specifies the dial function for creating unencrypted TCP connections.
 	Dial func(ctx context.Context, network, address string) (net.Conn, error)
 
+	// Proxy is the the proxy func to be used for all requests made by this
+	// transport. If Proxy is nil, http.ProxyFromEnvironment is used. If Proxy
+	// returns a nil *URL, no proxy is used.
+	//
+	// socks5 proxying does not currently support spdy streaming endpoints.
+	Proxy func(*http.Request) (*url.URL, error)
+
 	// Version forces a specific version to be used (if registered)
 	// Do we need this?
 	// Version string
@@ -560,6 +568,7 @@ func AnonymousClientConfig(config *Config) *Config {
 		Burst:              config.Burst,
 		Timeout:            config.Timeout,
 		Dial:               config.Dial,
+		Proxy:              config.Proxy,
 	}
 }
 
@@ -601,5 +610,6 @@ func CopyConfig(config *Config) *Config {
 		RateLimiter:        config.RateLimiter,
 		Timeout:            config.Timeout,
 		Dial:               config.Dial,
+		Proxy:              config.Proxy,
 	}
 }

staging/src/k8s.io/client-go/rest/config_test.go

@@ -23,6 +23,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/url"
 	"path/filepath"
 	"reflect"
 	"strings"
@@ -32,12 +33,12 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/client-go/kubernetes/scheme"
 	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/client-go/transport"
 	"k8s.io/client-go/util/flowcontrol"
 
+	"github.com/google/go-cmp/cmp"
 	fuzz "github.com/google/gofuzz"
 	"github.com/stretchr/testify/assert"
 )
@@ -274,8 +275,13 @@ func (n *fakeNegotiatedSerializer) DecoderToVersion(serializer runtime.Decoder,
 var fakeDialFunc = func(ctx context.Context, network, addr string) (net.Conn, error) {
 	return nil, fakeDialerError
 }
+
 var fakeDialerError = errors.New("fakedialer")
 
+func fakeProxyFunc(*http.Request) (*url.URL, error) {
+	return nil, errors.New("fakeproxy")
+}
+
 type fakeAuthProviderConfigPersister struct{}
 
 func (fakeAuthProviderConfigPersister) Persist(map[string]string) error {
@@ -318,8 +324,12 @@ func TestAnonymousConfig(t *testing.T) {
 		func(r *clientcmdapi.AuthProviderConfig, f fuzz.Continue) {
 			r.Config = map[string]string{}
 		},
-		// Dial does not require fuzzer
-		func(r *func(ctx context.Context, network, addr string) (net.Conn, error), f fuzz.Continue) {},
+		func(r *func(ctx context.Context, network, addr string) (net.Conn, error), f fuzz.Continue) {
+			*r = fakeDialFunc
+		},
+		func(r *func(*http.Request) (*url.URL, error), f fuzz.Continue) {
+			*r = fakeProxyFunc
+		},
 	)
 	for i := 0; i < 20; i++ {
 		original := &Config{}
@@ -350,13 +360,22 @@ func TestAnonymousConfig(t *testing.T) {
 			if !reflect.DeepEqual(expectedError, actualError) {
 				t.Fatalf("AnonymousClientConfig dropped the Dial field")
 			}
-		} else {
-			actual.Dial = nil
-			expected.Dial = nil
 		}
+		actual.Dial = nil
+		expected.Dial = nil
 
-		if !reflect.DeepEqual(*actual, expected) {
-			t.Fatalf("AnonymousClientConfig dropped unexpected fields, identify whether they are security related or not: %s", diff.ObjectGoPrintDiff(expected, actual))
+		if actual.Proxy != nil {
+			_, actualError := actual.Proxy(nil)
+			_, expectedError := expected.Proxy(nil)
+			if !reflect.DeepEqual(expectedError, actualError) {
+				t.Fatalf("AnonymousClientConfig dropped the Proxy field")
+			}
+		}
+		actual.Proxy = nil
+		expected.Proxy = nil
+
+		if diff := cmp.Diff(*actual, expected); diff != "" {
+			t.Fatalf("AnonymousClientConfig dropped unexpected fields, identify whether they are security related or not (-got, +want): %s", diff)
 		}
 	}
 }
@@ -396,6 +415,9 @@ func TestCopyConfig(t *testing.T) {
 		func(r *func(ctx context.Context, network, addr string) (net.Conn, error), f fuzz.Continue) {
 			*r = fakeDialFunc
 		},
+		func(r *func(*http.Request) (*url.URL, error), f fuzz.Continue) {
+			*r = fakeProxyFunc
+		},
 	)
 	for i := 0; i < 20; i++ {
 		original := &Config{}
@@ -410,10 +432,10 @@ func TestCopyConfig(t *testing.T) {
 		// function return the expected object.
 		if actual.WrapTransport == nil || !reflect.DeepEqual(expected.WrapTransport(nil), &fakeRoundTripper{}) {
 			t.Fatalf("CopyConfig dropped the WrapTransport field")
-		} else {
-			actual.WrapTransport = nil
-			expected.WrapTransport = nil
 		}
+		actual.WrapTransport = nil
+		expected.WrapTransport = nil
+
 		if actual.Dial != nil {
 			_, actualError := actual.Dial(context.Background(), "", "")
 			_, expectedError := expected.Dial(context.Background(), "", "")
@@ -423,6 +445,7 @@ func TestCopyConfig(t *testing.T) {
 		}
 		actual.Dial = nil
 		expected.Dial = nil
+
 		if actual.AuthConfigPersister != nil {
 			actualError := actual.AuthConfigPersister.Persist(nil)
 			expectedError := expected.AuthConfigPersister.Persist(nil)
@@ -433,8 +456,18 @@ func TestCopyConfig(t *testing.T) {
 		actual.AuthConfigPersister = nil
 		expected.AuthConfigPersister = nil
 
-		if !reflect.DeepEqual(*actual, expected) {
-			t.Fatalf("CopyConfig  dropped unexpected fields, identify whether they are security related or not: %s", diff.ObjectReflectDiff(expected, *actual))
+		if actual.Proxy != nil {
+			_, actualError := actual.Proxy(nil)
+			_, expectedError := expected.Proxy(nil)
+			if !reflect.DeepEqual(expectedError, actualError) {
+				t.Fatalf("CopyConfig  dropped the Proxy field")
+			}
+		}
+		actual.Proxy = nil
+		expected.Proxy = nil
+
+		if diff := cmp.Diff(*actual, expected); diff != "" {
+			t.Fatalf("CopyConfig  dropped unexpected fields, identify whether they are security related or not (-got, +want): %s", diff)
 		}
 	}
 }
@@ -564,10 +597,11 @@ func TestConfigSprint(t *testing.T) {
 		RateLimiter:   &fakeLimiter{},
 		Timeout:       3 * time.Second,
 		Dial:          fakeDialFunc,
+		Proxy:         fakeProxyFunc,
 	}
 	want := fmt.Sprintf(
-		`&rest.Config{Host:"localhost:8080", APIPath:"v1", ContentConfig:rest.ContentConfig{AcceptContentTypes:"application/json", ContentType:"application/json", GroupVersion:(*schema.GroupVersion)(nil), NegotiatedSerializer:runtime.NegotiatedSerializer(nil)}, Username:"gopher", Password:"--- REDACTED ---", BearerToken:"--- REDACTED ---", BearerTokenFile:"", Impersonate:rest.ImpersonationConfig{UserName:"gopher2", Groups:[]string(nil), Extra:map[string][]string(nil)}, AuthProvider:api.AuthProviderConfig{Name: "gopher", Config: map[string]string{--- REDACTED ---}}, AuthConfigPersister:rest.AuthProviderConfigPersister(--- REDACTED ---), ExecProvider:api.AuthProviderConfig{Command: "sudo", Args: []string{"--- REDACTED ---"}, Env: []ExecEnvVar{--- REDACTED ---}, APIVersion: ""}, TLSClientConfig:rest.sanitizedTLSClientConfig{Insecure:false, ServerName:"", CertFile:"a.crt", KeyFile:"a.key", CAFile:"", CertData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x54, 0x52, 0x55, 0x4e, 0x43, 0x41, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, KeyData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x52, 0x45, 0x44, 0x41, 0x43, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, CAData:[]uint8(nil), NextProtos:[]string{"h2", "http/1.1"}}, UserAgent:"gobot", DisableCompression:false, Transport:(*rest.fakeRoundTripper)(%p), WrapTransport:(transport.WrapperFunc)(%p), QPS:1, Burst:2, RateLimiter:(*rest.fakeLimiter)(%p), Timeout:3000000000, Dial:(func(context.Context, string, string) (net.Conn, error))(%p)}`,
-		c.Transport, fakeWrapperFunc, c.RateLimiter, fakeDialFunc,
+		`&rest.Config{Host:"localhost:8080", APIPath:"v1", ContentConfig:rest.ContentConfig{AcceptContentTypes:"application/json", ContentType:"application/json", GroupVersion:(*schema.GroupVersion)(nil), NegotiatedSerializer:runtime.NegotiatedSerializer(nil)}, Username:"gopher", Password:"--- REDACTED ---", BearerToken:"--- REDACTED ---", BearerTokenFile:"", Impersonate:rest.ImpersonationConfig{UserName:"gopher2", Groups:[]string(nil), Extra:map[string][]string(nil)}, AuthProvider:api.AuthProviderConfig{Name: "gopher", Config: map[string]string{--- REDACTED ---}}, AuthConfigPersister:rest.AuthProviderConfigPersister(--- REDACTED ---), ExecProvider:api.AuthProviderConfig{Command: "sudo", Args: []string{"--- REDACTED ---"}, Env: []ExecEnvVar{--- REDACTED ---}, APIVersion: ""}, TLSClientConfig:rest.sanitizedTLSClientConfig{Insecure:false, ServerName:"", CertFile:"a.crt", KeyFile:"a.key", CAFile:"", CertData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x54, 0x52, 0x55, 0x4e, 0x43, 0x41, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, KeyData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x52, 0x45, 0x44, 0x41, 0x43, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, CAData:[]uint8(nil), NextProtos:[]string{"h2", "http/1.1"}}, UserAgent:"gobot", DisableCompression:false, Transport:(*rest.fakeRoundTripper)(%p), WrapTransport:(transport.WrapperFunc)(%p), QPS:1, Burst:2, RateLimiter:(*rest.fakeLimiter)(%p), Timeout:3000000000, Dial:(func(context.Context, string, string) (net.Conn, error))(%p), Proxy:(func(*http.Request) (*url.URL, error))(%p)}`,
+		c.Transport, fakeWrapperFunc, c.RateLimiter, fakeDialFunc, fakeProxyFunc,
 	)
 
 	for _, f := range []string{"%s", "%v", "%+v", "%#v"} {

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

@@ -85,7 +85,8 @@ func (c *Config) TransportConfig() (*transport.Config, error) {
 			Groups:   c.Impersonate.Groups,
 			Extra:    c.Impersonate.Extra,
 		},
-		Dial: c.Dial,
+		Dial:  c.Dial,
+		Proxy: c.Proxy,
 	}
 
 	if c.ExecProvider != nil && c.AuthProvider != nil {

staging/src/k8s.io/client-go/tools/clientcmd/api/types.go

@@ -82,6 +82,17 @@ type Cluster struct {
 	// CertificateAuthorityData contains PEM-encoded certificate authority certificates. Overrides CertificateAuthority
 	// +optional
 	CertificateAuthorityData []byte `json:"certificate-authority-data,omitempty"`
+	// ProxyURL is the URL to the proxy to be used for all requests made by this
+	// client. URLs with "http", "https", and "socks5" schemes are supported.  If
+	// this configuration is not provided or the empty string, the client
+	// attempts to construct a proxy configuration from http_proxy and
+	// https_proxy environment variables. If these environment variables are not
+	// set, the client does not attempt to proxy requests.
+	//
+	// socks5 proxying does not currently support spdy streaming endpoints (exec,
+	// attach, port forward).
+	// +optional
+	ProxyURL string `json:"proxy-url,omitempty"`
 	// Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields
 	// +optional
 	Extensions map[string]runtime.Object `json:"extensions,omitempty"`

staging/src/k8s.io/client-go/tools/clientcmd/api/v1/types.go

@@ -75,6 +75,17 @@ type Cluster struct {
 	// CertificateAuthorityData contains PEM-encoded certificate authority certificates. Overrides CertificateAuthority
 	// +optional
 	CertificateAuthorityData []byte `json:"certificate-authority-data,omitempty"`
+	// ProxyURL is the URL to the proxy to be used for all requests made by this
+	// client. URLs with "http", "https", and "socks5" schemes are supported.  If
+	// this configuration is not provided or the empty string, the client
+	// attempts to construct a proxy configuration from http_proxy and
+	// https_proxy environment variables. If these environment variables are not
+	// set, the client does not attempt to proxy requests.
+	//
+	// socks5 proxying does not currently support spdy streaming endpoints (exec,
+	// attach, port forward).
+	// +optional
+	ProxyURL string `json:"proxy-url,omitempty"`
 	// Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields
 	// +optional
 	Extensions []NamedExtension `json:"extensions,omitempty"`