Allow specifying scheme when proxying

cmd/kube-apiserver/app/server.go

@@ -376,6 +376,30 @@ func (s *APIServer) Run(_ []string) error {
 		glog.Fatalf("Cloud provider could not be initialized: %v", err)
 	}
 
+	// Setup tunneler if needed
+	var tunneler master.Tunneler
+	var proxyDialerFn apiserver.ProxyDialerFunc
+	if len(s.SSHUser) > 0 {
+		// Get ssh key distribution func, if supported
+		var installSSH master.InstallSSHKey
+		if cloud != nil {
+			if instances, supported := cloud.Instances(); supported {
+				installSSH = instances.AddSSHKeyToAllInstances
+			}
+		}
+
+		// Set up the tunneler
+		tunneler = master.NewSSHTunneler(s.SSHUser, s.SSHKeyfile, installSSH)
+
+		// Use the tunneler's dialer to connect to the kubelet
+		s.KubeletConfig.Dial = tunneler.Dial
+		// Use the tunneler's dialer when proxying to pods, services, and nodes
+		proxyDialerFn = tunneler.Dial
+	}
+
+	// Proxying to pods and services is IP-based... don't expect to be able to verify the hostname
+	proxyTLSClientConfig := &tls.Config{InsecureSkipVerify: true}
+
 	kubeletClient, err := client.NewKubeletClient(&s.KubeletConfig)
 	if err != nil {
 		glog.Fatalf("Failure to start kubelet client: %v", err)
@@ -508,12 +532,7 @@ func (s *APIServer) Run(_ []string) error {
 			}
 		}
 	}
-	var installSSH master.InstallSSHKey
-	if cloud != nil {
-		if instances, supported := cloud.Instances(); supported {
-			installSSH = instances.AddSSHKeyToAllInstances
-		}
-	}
+
 	config := &master.Config{
 		StorageDestinations:       storageDestinations,
 		StorageVersions:           storageVersions,
@@ -542,9 +561,9 @@ func (s *APIServer) Run(_ []string) error {
 		ClusterName:               s.ClusterName,
 		ExternalHost:              s.ExternalHost,
 		MinRequestTimeout:         s.MinRequestTimeout,
-		SSHUser:                   s.SSHUser,
-		SSHKeyfile:                s.SSHKeyfile,
-		InstallSSHKey:             installSSH,
+		ProxyDialer:               proxyDialerFn,
+		ProxyTLSClientConfig:      proxyTLSClientConfig,
+		Tunneler:                  tunneler,
 		ServiceNodePortRange:      s.ServiceNodePortRange,
 		KubernetesServiceNodePort: s.KubernetesServiceNodePort,
 	}

pkg/apiserver/api_installer.go

@@ -41,7 +41,6 @@ type APIInstaller struct {
 	info              *APIRequestInfoResolver
 	prefix            string // Path prefix where API resources are to be registered.
 	minRequestTimeout time.Duration
-	proxyDialerFn     ProxyDialerFunc
 }
 
 // Struct capturing information about an action ("GET", "POST", "WATCH", PROXY", etc).
@@ -64,7 +63,7 @@ var errEmptyName = errors.NewBadRequest("name must be provided")
 func (a *APIInstaller) Install(ws *restful.WebService) (apiResources []api.APIResource, errors []error) {
 	errors = make([]error, 0)
 
-	proxyHandler := (&ProxyHandler{a.prefix + "/proxy/", a.group.Storage, a.group.Codec, a.group.Context, a.info, a.proxyDialerFn})
+	proxyHandler := (&ProxyHandler{a.prefix + "/proxy/", a.group.Storage, a.group.Codec, a.group.Context, a.info})
 
 	// Register the paths in a deterministic (sorted) order to get a deterministic swagger spec.
 	paths := make([]string, len(a.group.Storage))

pkg/apiserver/apiserver.go

@@ -105,7 +105,6 @@ type APIGroupVersion struct {
 	Admit   admission.Interface
 	Context api.RequestContextMapper
 
-	ProxyDialerFn     ProxyDialerFunc
 	MinRequestTimeout time.Duration
 }
 
@@ -164,7 +163,6 @@ func (g *APIGroupVersion) newInstaller() *APIInstaller {
 		info:              g.APIRequestInfoResolver,
 		prefix:            prefix,
 		minRequestTimeout: g.MinRequestTimeout,
-		proxyDialerFn:     g.ProxyDialerFn,
 	}
 	return installer
 }

pkg/apiserver/proxy.go

@@ -17,11 +17,8 @@ limitations under the License.
 package apiserver
 
 import (
-	"crypto/tls"
-	"fmt"
 	"io"
 	"math/rand"
-	"net"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
@@ -40,7 +37,6 @@ import (
 	proxyutil "k8s.io/kubernetes/pkg/util/proxy"
 
 	"github.com/golang/glog"
-	"k8s.io/kubernetes/third_party/golang/netutil"
 )
 
 // ProxyHandler provides a http.Handler which will proxy traffic to locations
@@ -51,8 +47,6 @@ type ProxyHandler struct {
 	codec                  runtime.Codec
 	context                api.RequestContextMapper
 	apiRequestInfoResolver *APIRequestInfoResolver
-
-	dial func(network, addr string) (net.Conn, error)
 }
 
 func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
@@ -125,11 +119,8 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		httpCode = http.StatusNotFound
 		return
 	}
-	// If we have a custom dialer, and no pre-existing transport, initialize it to use the dialer.
-	if roundTripper == nil && r.dial != nil {
-		glog.V(5).Infof("[%x: %v] making a dial-only transport...", proxyHandlerTraceID, req.URL)
-		roundTripper = &http.Transport{Dial: r.dial}
-	} else if roundTripper != nil {
+
+	if roundTripper != nil {
 		glog.V(5).Infof("[%x: %v] using transport %T...", proxyHandlerTraceID, req.URL, roundTripper)
 	}
 
@@ -217,7 +208,7 @@ func (r *ProxyHandler) tryUpgrade(w http.ResponseWriter, req, newReq *http.Reque
 	if !httpstream.IsUpgradeRequest(req) {
 		return false
 	}
-	backendConn, err := dialURL(location, transport)
+	backendConn, err := proxyutil.DialURL(location, transport)
 	if err != nil {
 		status := errToAPIStatus(err)
 		writeJSON(status.Code, r.codec, status, w, true)
@@ -264,46 +255,6 @@ func (r *ProxyHandler) tryUpgrade(w http.ResponseWriter, req, newReq *http.Reque
 	return true
 }
 
-func dialURL(url *url.URL, transport http.RoundTripper) (net.Conn, error) {
-	dialAddr := netutil.CanonicalAddr(url)
-
-	switch url.Scheme {
-	case "http":
-		return net.Dial("tcp", dialAddr)
-	case "https":
-		// Get the tls config from the transport if we recognize it
-		var tlsConfig *tls.Config
-		if transport != nil {
-			httpTransport, ok := transport.(*http.Transport)
-			if ok {
-				tlsConfig = httpTransport.TLSClientConfig
-			}
-		}
-
-		// Dial
-		tlsConn, err := tls.Dial("tcp", dialAddr, tlsConfig)
-		if err != nil {
-			return nil, err
-		}
-
-		// Return if we were configured to skip validation
-		if tlsConfig != nil && tlsConfig.InsecureSkipVerify {
-			return tlsConn, nil
-		}
-
-		// Verify
-		host, _, _ := net.SplitHostPort(dialAddr)
-		if err := tlsConn.VerifyHostname(host); err != nil {
-			tlsConn.Close()
-			return nil, err
-		}
-
-		return tlsConn, nil
-	default:
-		return nil, fmt.Errorf("Unknown scheme: %s", url.Scheme)
-	}
-}
-
 // borrowed from net/http/httputil/reverseproxy.go
 func singleJoiningSlash(a, b string) string {
 	aslash := strings.HasSuffix(a, "/")

pkg/apiserver/proxy_test.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -171,6 +172,21 @@ func TestProxyUpgrade(t *testing.T) {
 			},
 			ProxyTransport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: localhostPool}},
 		},
+		"https (valid hostname + RootCAs + custom dialer)": {
+			ServerFunc: func(h http.Handler) *httptest.Server {
+				cert, err := tls.X509KeyPair(localhostCert, localhostKey)
+				if err != nil {
+					t.Errorf("https (valid hostname): proxy_test: %v", err)
+				}
+				ts := httptest.NewUnstartedServer(h)
+				ts.TLS = &tls.Config{
+					Certificates: []tls.Certificate{cert},
+				}
+				ts.StartTLS()
+				return ts
+			},
+			ProxyTransport: &http.Transport{Dial: net.Dial, TLSClientConfig: &tls.Config{RootCAs: localhostPool}},
+		},
 	}
 
 	for k, tc := range testcases {

pkg/client/chaosclient/chaosclient.go

@@ -28,6 +28,8 @@ import (
 	"net/http"
 	"reflect"
 	"runtime"
+
+	"k8s.io/kubernetes/pkg/util"
 )
 
 // chaosrt provides the ability to perform simulations of HTTP client failures
@@ -86,6 +88,12 @@ func (rt *chaosrt) RoundTrip(req *http.Request) (*http.Response, error) {
 	return rt.rt.RoundTrip(req)
 }
 
+var _ = util.RoundTripperWrapper(&chaosrt{})
+
+func (rt *chaosrt) WrappedRoundTripper() http.RoundTripper {
+	return rt.rt
+}
+
 // Seed represents a consistent stream of chaos.
 type Seed struct {
 	*rand.Rand

pkg/client/unversioned/debugging.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/golang/glog"
 
+	"k8s.io/kubernetes/pkg/util"
 	"k8s.io/kubernetes/pkg/util/sets"
 )
 
@@ -133,3 +134,9 @@ func (rt *DebuggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 
 	return response, err
 }
+
+var _ = util.RoundTripperWrapper(&DebuggingRoundTripper{})
+
+func (rt *DebuggingRoundTripper) WrappedRoundTripper() http.RoundTripper {
+	return rt.delegatedRoundTripper
+}

pkg/client/unversioned/transport.go

@@ -22,6 +22,8 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
+
+	"k8s.io/kubernetes/pkg/util"
 )
 
 type userAgentRoundTripper struct {
@@ -42,6 +44,12 @@ func (rt *userAgentRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 	return rt.rt.RoundTrip(req)
 }
 
+var _ = util.RoundTripperWrapper(&userAgentRoundTripper{})
+
+func (rt *userAgentRoundTripper) WrappedRoundTripper() http.RoundTripper {
+	return rt.rt
+}
+
 type basicAuthRoundTripper struct {
 	username string
 	password string
@@ -63,6 +71,12 @@ func (rt *basicAuthRoundTripper) RoundTrip(req *http.Request) (*http.Response, e
 	return rt.rt.RoundTrip(req)
 }
 
+var _ = util.RoundTripperWrapper(&basicAuthRoundTripper{})
+
+func (rt *basicAuthRoundTripper) WrappedRoundTripper() http.RoundTripper {
+	return rt.rt
+}
+
 type bearerAuthRoundTripper struct {
 	bearer string
 	rt     http.RoundTripper
@@ -84,6 +98,12 @@ func (rt *bearerAuthRoundTripper) RoundTrip(req *http.Request) (*http.Response,
 	return rt.rt.RoundTrip(req)
 }
 
+var _ = util.RoundTripperWrapper(&bearerAuthRoundTripper{})
+
+func (rt *bearerAuthRoundTripper) WrappedRoundTripper() http.RoundTripper {
+	return rt.rt
+}
+
 // TLSConfigFor returns a tls.Config that will provide the transport level security defined
 // by the provided Config. Will return nil if no transport level security is requested.
 func TLSConfigFor(config *Config) (*tls.Config, error) {