add proxying capability to kubernetes-discovery

cmd/kubernetes-discovery/artifacts/local-cluster-up/kubernetes-discover-pod.yaml

@@ -18,6 +18,8 @@ spec:
         image: kubernetes-discovery:latest
         imagePullPolicy: Never
         args:
+        - "--proxy-client-cert-file=/var/run/auth-proxy-client/tls.crt"
+        - "--proxy-client-key-file=/var/run/auth-proxy-client/tls.key"
         - "--tls-cert-file=/var/run/serving-cert/tls.crt"
         - "--tls-private-key-file=/var/run/serving-cert/tls.key"
         - "--tls-ca-file=/var/run/serving-ca/ca.crt"

cmd/kubernetes-discovery/pkg/apiserver/BUILD

@@ -16,6 +16,7 @@ go_library(
         "apiserver.go",
         "apiservice_controller.go",
         "handler_apis.go",
+        "handler_proxy.go",
     ],
     tags = ["automanaged"],
     deps = [
@@ -35,14 +36,17 @@ go_library(
         "//pkg/apiserver/filters:go_default_library",
         "//pkg/auth/handlers:go_default_library",
         "//pkg/client/cache:go_default_library",
+        "//pkg/client/restclient:go_default_library",
+        "//pkg/client/transport:go_default_library",
         "//pkg/controller:go_default_library",
         "//pkg/genericapiserver:go_default_library",
         "//pkg/genericapiserver/filters:go_default_library",
         "//pkg/labels:go_default_library",
         "//pkg/registry/generic:go_default_library",
+        "//pkg/registry/generic/rest:go_default_library",
         "//pkg/runtime:go_default_library",
+        "//pkg/util/httpstream/spdy:go_default_library",
         "//pkg/util/runtime:go_default_library",
-        "//pkg/util/sets:go_default_library",
         "//pkg/util/wait:go_default_library",
         "//pkg/util/workqueue:go_default_library",
         "//pkg/version:go_default_library",
@@ -52,7 +56,10 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["handler_apis_test.go"],
+    srcs = [
+        "handler_apis_test.go",
+        "handler_proxy_test.go",
+    ],
     library = "go_default_library",
     tags = ["automanaged"],
     deps = [
@@ -60,8 +67,11 @@ go_test(
         "//cmd/kubernetes-discovery/pkg/client/listers/apiregistration/internalversion:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/apis/meta/v1:go_default_library",
+        "//pkg/apiserver/request:go_default_library",
+        "//pkg/auth/user:go_default_library",
         "//pkg/client/cache:go_default_library",
         "//pkg/runtime:go_default_library",
         "//pkg/util/diff:go_default_library",
+        "//pkg/util/sets:go_default_library",
     ],
 )

cmd/kubernetes-discovery/pkg/apiserver/apiserver.go

@@ -28,7 +28,6 @@ import (
 	"k8s.io/kubernetes/pkg/genericapiserver"
 	genericfilters "k8s.io/kubernetes/pkg/genericapiserver/filters"
 	"k8s.io/kubernetes/pkg/registry/generic"
-	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/version"
 
@@ -47,6 +46,11 @@ const legacyAPIServiceName = "v1."
 type Config struct {
 	GenericConfig *genericapiserver.Config
 
+	// ProxyClientCert/Key are the client cert used to identify this proxy. Backing APIServices use
+	// this to confirm the proxy's identity
+	ProxyClientCert []byte
+	ProxyClientKey  []byte
+
 	// RESTOptionsGetter is used to construct storage for a particular resource
 	RESTOptionsGetter generic.RESTOptionsGetter
 }
@@ -55,13 +59,22 @@ type Config struct {
 type APIDiscoveryServer struct {
 	GenericAPIServer *genericapiserver.GenericAPIServer
 
-	// handledAPIServices tracks which APIServices have already been handled.  Once endpoints are added,
+	contextMapper api.RequestContextMapper
-	// the listers that are used keep bits in sync automatically.
+
-	handledAPIServices sets.String
+	// proxyClientCert/Key are the client cert used to identify this proxy. Backing APIServices use
+	// this to confirm the proxy's identity
+	proxyClientCert []byte
+	proxyClientKey  []byte
+
+	// proxyHandlers are the proxy handlers that are currently registered, keyed by apiservice.name
+	proxyHandlers map[string]*proxyHandler
 
 	// lister is used to add group handling for /apis/<group> discovery lookups based on
 	// controller state
 	lister listers.APIServiceLister
+
+	// proxyMux intercepts requests that need to be proxied to backing API servers
+	proxyMux *http.ServeMux
 }
 
 type completedConfig struct {
@@ -91,9 +104,12 @@ func (c completedConfig) New() (*APIDiscoveryServer, error) {
 		5*time.Minute, // this is effectively used as a refresh interval right now.  Might want to do something nicer later on.
 	)
 
+	proxyMux := http.NewServeMux()
+
 	// most API servers don't need to do this, but we need a custom handler chain to handle the special /apis handling here
 	c.Config.GenericConfig.BuildHandlerChainsFunc = (&handlerChainConfig{
 		informers: informerFactory,
+		proxyMux:  proxyMux,
 	}).handlerChain
 
 	genericServer, err := c.Config.GenericConfig.SkipComplete().New() // completion is done in Complete, no need for a second time
@@ -102,9 +118,13 @@ func (c completedConfig) New() (*APIDiscoveryServer, error) {
 	}
 
 	s := &APIDiscoveryServer{
-		GenericAPIServer:   genericServer,
+		GenericAPIServer: genericServer,
-		handledAPIServices: sets.String{},
+		contextMapper:    c.GenericConfig.RequestContextMapper,
-		lister:             informerFactory.Apiregistration().InternalVersion().APIServices().Lister(),
+		proxyClientCert:  c.ProxyClientCert,
+		proxyClientKey:   c.ProxyClientKey,
+		proxyHandlers:    map[string]*proxyHandler{},
+		lister:           informerFactory.Apiregistration().InternalVersion().APIServices().Lister(),
+		proxyMux:         proxyMux,
 	}
 
 	apiGroupInfo := genericapiserver.NewDefaultAPIGroupInfo(apiregistration.GroupName)
@@ -134,6 +154,7 @@ func (c completedConfig) New() (*APIDiscoveryServer, error) {
 // handlerChainConfig is the config used to build the custom handler chain for this api server
 type handlerChainConfig struct {
 	informers informers.SharedInformerFactory
+	proxyMux  *http.ServeMux
 }
 
 // handlerChain is a method to build the handler chain for this API server.  We need a custom handler chain so that we
@@ -144,6 +165,12 @@ func (h *handlerChainConfig) handlerChain(apiHandler http.Handler, c *genericapi
 	handler := WithAPIs(apiHandler, h.informers.Apiregistration().InternalVersion().APIServices())
 
 	handler = apiserverfilters.WithAuthorization(handler, c.RequestContextMapper, c.Authorizer)
+
+	// this mux is NOT protected by authorization, but DOES have authentication information
+	// this is so that everyone can hit the proxy and we can properly identify the user.  The backing
+	// API server will deal with authorization
+	handler = WithProxyMux(handler, h.proxyMux)
+
 	handler = apiserverfilters.WithImpersonation(handler, c.RequestContextMapper, c.Authorizer)
 	// audit to stdout to help with debugging as we get this started
 	handler = apiserverfilters.WithAudit(handler, c.RequestContextMapper, os.Stdout)
@@ -162,12 +189,34 @@ func (h *handlerChainConfig) handlerChain(apiHandler http.Handler, c *genericapi
 // AddAPIService adds an API service.  It is not thread-safe, so only call it on one thread at a time please.
 // It's a slow moving API, so its ok to run the controller on a single thread
 func (s *APIDiscoveryServer) AddAPIService(apiService *apiregistration.APIService) {
-	if s.handledAPIServices.Has(apiService.Name) {
+	// if the proxyHandler already exists, it needs to be updated. The discovery bits do not
+	// since they are wired against listers because they require multiple resources to respond
+	if proxyHandler, exists := s.proxyHandlers[apiService.Name]; exists {
+		proxyHandler.updateAPIService(apiService)
 		return
 	}
+
+	proxyPath := "/apis/" + apiService.Spec.Group + "/" + apiService.Spec.Version
+	// v1. is a special case for the legacy API.  It proxies to a wider set of endpoints.
+	if apiService.Name == "v1." {
+		proxyPath = "/api"
+	}
+
+	// register the proxy handler
+	proxyHandler := &proxyHandler{
+		contextMapper:          s.contextMapper,
+		proxyClientCert:        s.proxyClientCert,
+		proxyClientKey:         s.proxyClientKey,
+		transportBuildingError: nil,
+		proxyRoundTripper:      nil,
+	}
+	proxyHandler.updateAPIService(apiService)
+	s.proxyHandlers[apiService.Name] = proxyHandler
+	s.proxyMux.Handle(proxyPath, proxyHandler)
+	s.proxyMux.Handle(proxyPath+"/", proxyHandler)
+
 	// if we're dealing with the legacy group, we're done here
 	if apiService.Name == legacyAPIServiceName {
-		s.handledAPIServices.Insert(apiService.Name)
 		return
 	}
 
@@ -186,7 +235,23 @@ func (s *APIDiscoveryServer) AddAPIService(apiService *apiregistration.APIServic
 // RemoveAPIService removes the APIService from being handled.  Later on it will disable the proxy endpoint.
 // Right now it does nothing because our handler has to properly 404 itself since muxes don't unregister
 func (s *APIDiscoveryServer) RemoveAPIService(apiServiceName string) {
-	if !s.handledAPIServices.Has(apiServiceName) {
+	proxyHandler, exists := s.proxyHandlers[apiServiceName]
+	if !exists {
 		return
 	}
+	proxyHandler.removeAPIService()
+}
+
+func WithProxyMux(handler http.Handler, mux *http.ServeMux) http.Handler {
+	if mux == nil {
+		return handler
+	}
+
+	// register the handler at this stage against everything under slash.  More specific paths that get registered will take precedence
+	// this effectively delegates by default unless something specific gets registered.
+	mux.Handle("/", handler)
+
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		mux.ServeHTTP(w, req)
+	})
 }

cmd/kubernetes-discovery/pkg/apiserver/handler_apis.go

@@ -44,7 +44,7 @@ func WithAPIs(handler http.Handler, informer informers.APIServiceInformer) http.
 	})
 }
 
-// apisHandler servers the `/apis` endpoint.
+// apisHandler serves the `/apis` endpoint.
 // This is registered as a filter so that it never collides with any explictly registered endpoints
 type apisHandler struct {
 	lister listers.APIServiceLister
@@ -79,7 +79,8 @@ func (r *apisHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	}
 
 	discoveryGroupList := &metav1.APIGroupList{
-		// always add OUR api group to the list first
+		// always add OUR api group to the list first.  Since we'll never have a registered APIService for it
+		// and since this is the crux of the API, having this first will give our names priority.  It's good to be king.
 		Groups: []metav1.APIGroup{discoveryGroup},
 	}
 
@@ -130,7 +131,7 @@ func newDiscoveryAPIGroup(apiServices []*apiregistrationapi.APIService) *metav1.
 	return discoveryGroup
 }
 
-// apiGroupHandler servers the `/apis/<group>` endpoint.
+// apiGroupHandler serves the `/apis/<group>` endpoint.
 type apiGroupHandler struct {
 	groupName string
 

cmd/kubernetes-discovery/pkg/apiserver/handler_proxy.go

@@ -0,0 +1,198 @@
+/*
+Copyright 2016 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 apiserver
+
+import (
+	"net/http"
+	"net/url"
+	"sync"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apiserver"
+	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/client/transport"
+	genericrest "k8s.io/kubernetes/pkg/registry/generic/rest"
+	"k8s.io/kubernetes/pkg/runtime"
+	"k8s.io/kubernetes/pkg/util/httpstream/spdy"
+
+	apiregistrationapi "k8s.io/kubernetes/cmd/kubernetes-discovery/pkg/apis/apiregistration"
+)
+
+// proxyHandler provides a http.Handler which will proxy traffic to locations
+// specified by items implementing Redirector.
+type proxyHandler struct {
+	contextMapper api.RequestContextMapper
+
+	// proxyClientCert/Key are the client cert used to identify this proxy. Backing APIServices use
+	// this to confirm the proxy's identity
+	proxyClientCert []byte
+	proxyClientKey  []byte
+
+	// lock protects us for updates.
+	lock sync.RWMutex
+	// restConfig holds the information for building a roundtripper
+	restConfig *restclient.Config
+	// transportBuildingError is an error produced while building the transport.  If this
+	// is non-nil, it will be reported to clients.
+	transportBuildingError error
+	// proxyRoundTripper is the re-useable portion of the transport.  It does not vary with any request.
+	proxyRoundTripper http.RoundTripper
+	// destinationHost is the hostname of the backing API server
+	destinationHost string
+}
+
+func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	proxyRoundTripper, err := r.getRoundTripper()
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if proxyRoundTripper == nil {
+		http.Error(w, "", http.StatusNotFound)
+		return
+	}
+
+	ctx, ok := r.contextMapper.Get(req)
+	if !ok {
+		http.Error(w, "missing context", http.StatusInternalServerError)
+		return
+	}
+	user, ok := api.UserFrom(ctx)
+	if !ok {
+		http.Error(w, "missing user", http.StatusInternalServerError)
+		return
+	}
+
+	// write a new location based on the existing request pointed at the target service
+	location := &url.URL{}
+	location.Scheme = "https"
+	location.Host = r.getDestinationHost()
+	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
+	newReq, err := http.NewRequest(req.Method, location.String(), req.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		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 = transport.NewAuthProxyRoundTripper(user.GetName(), user.GetGroups(), user.GetExtra(), proxyRoundTripper)
+	proxyRoundTripper, upgrade, err = r.maybeWrapForConnectionUpgrades(proxyRoundTripper, req)
+
+	handler := genericrest.NewUpgradeAwareProxyHandler(location, proxyRoundTripper, true, upgrade, &responder{w: w})
+	handler.ServeHTTP(w, newReq)
+}
+
+// maybeWrapForConnectionUpgrades wraps the roundtripper for upgrades.  The bool indicates if it was wrapped
+func (r *proxyHandler) maybeWrapForConnectionUpgrades(rt http.RoundTripper, req *http.Request) (http.RoundTripper, bool, error) {
+	connectionHeader := req.Header.Get("Connection")
+	if len(connectionHeader) == 0 {
+		return rt, false, nil
+	}
+
+	cfg := r.getRESTConfig()
+	tlsConfig, err := restclient.TLSConfigFor(cfg)
+	if err != nil {
+		return nil, true, err
+	}
+	upgradeRoundTripper := spdy.NewRoundTripper(tlsConfig)
+	wrappedRT, err := restclient.HTTPWrappersForConfig(cfg, upgradeRoundTripper)
+	if err != nil {
+		return nil, true, err
+	}
+
+	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
+}
+
+// TODO this should properly handle content type negotiation
+// if the caller asked for protobuf and you write JSON bad things happen.
+func (r *responder) Object(statusCode int, obj runtime.Object) {
+	apiserver.WriteRawJSON(statusCode, obj, r.w)
+}
+
+func (r *responder) Error(err error) {
+	http.Error(r.w, err.Error(), http.StatusInternalServerError)
+}
+
+// these methods provide locked access to fields
+
+func (r *proxyHandler) updateAPIService(apiService *apiregistrationapi.APIService) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.transportBuildingError = nil
+	r.proxyRoundTripper = nil
+
+	r.destinationHost = apiService.Spec.Service.Name + "." + apiService.Spec.Service.Namespace + ".svc"
+	r.restConfig = &restclient.Config{
+		Insecure: apiService.Spec.InsecureSkipTLSVerify,
+		TLSClientConfig: restclient.TLSClientConfig{
+			CertData: r.proxyClientCert,
+			KeyData:  r.proxyClientKey,
+			CAData:   apiService.Spec.CABundle,
+		},
+	}
+	r.proxyRoundTripper, r.transportBuildingError = restclient.TransportFor(r.restConfig)
+}
+
+func (r *proxyHandler) removeAPIService() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.transportBuildingError = nil
+	r.proxyRoundTripper = nil
+}
+
+func (r *proxyHandler) getRoundTripper() (http.RoundTripper, error) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	return r.proxyRoundTripper, r.transportBuildingError
+}
+
+func (r *proxyHandler) getDestinationHost() string {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	return r.destinationHost
+}
+
+func (r *proxyHandler) getRESTConfig() *restclient.Config {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	return r.restConfig
+}

cmd/kubernetes-discovery/pkg/apiserver/handler_proxy_test.go

@@ -0,0 +1,198 @@
+/*
+Copyright 2016 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 apiserver
+
+import (
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"net/http/httputil"
+	"reflect"
+	"strings"
+	"testing"
+
+	"k8s.io/kubernetes/pkg/api"
+	"k8s.io/kubernetes/pkg/apiserver/request"
+	"k8s.io/kubernetes/pkg/auth/user"
+	"k8s.io/kubernetes/pkg/util/sets"
+
+	"k8s.io/kubernetes/cmd/kubernetes-discovery/pkg/apis/apiregistration"
+)
+
+type targetHTTPHandler struct {
+	called  bool
+	headers map[string][]string
+	path    string
+}
+
+func (d *targetHTTPHandler) Reset() {
+	d.path = ""
+	d.called = false
+	d.headers = nil
+}
+
+func (d *targetHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	d.path = r.URL.Path
+	d.called = true
+	d.headers = r.Header
+	w.WriteHeader(http.StatusOK)
+}
+
+type fakeRequestContextMapper struct {
+	user user.Info
+}
+
+func (m *fakeRequestContextMapper) Get(req *http.Request) (api.Context, bool) {
+	ctx := api.NewContext()
+	if m.user != nil {
+		ctx = api.WithUser(ctx, m.user)
+	}
+
+	resolver := &request.RequestInfoFactory{
+		APIPrefixes:          sets.NewString("api", "apis"),
+		GrouplessAPIPrefixes: sets.NewString("api"),
+	}
+	info, err := resolver.NewRequestInfo(req)
+	if err == nil {
+		ctx = request.WithRequestInfo(ctx, info)
+	}
+
+	return ctx, true
+}
+
+func (*fakeRequestContextMapper) Update(req *http.Request, context api.Context) error {
+	return nil
+}
+
+func TestProxyHandler(t *testing.T) {
+	target := &targetHTTPHandler{}
+	targetServer := httptest.NewTLSServer(target)
+	defer targetServer.Close()
+
+	handler := &proxyHandler{}
+
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	tests := map[string]struct {
+		user       user.Info
+		path       string
+		apiService *apiregistration.APIService
+
+		expectedStatusCode int
+		expectedBody       string
+		expectedCalled     bool
+		expectedHeaders    map[string][]string
+	}{
+		"no target": {
+			expectedStatusCode: http.StatusNotFound,
+		},
+		"no user": {
+			apiService: &apiregistration.APIService{
+				ObjectMeta: api.ObjectMeta{Name: "v1.foo"},
+				Spec: apiregistration.APIServiceSpec{
+					Group:   "foo",
+					Version: "v1",
+				},
+			},
+			expectedStatusCode: http.StatusInternalServerError,
+			expectedBody:       "missing user",
+		},
+		"proxy with user": {
+			user: &user.DefaultInfo{
+				Name:   "username",
+				Groups: []string{"one", "two"},
+			},
+			path: "/request/path",
+			apiService: &apiregistration.APIService{
+				ObjectMeta: api.ObjectMeta{Name: "v1.foo"},
+				Spec: apiregistration.APIServiceSpec{
+					Group:                 "foo",
+					Version:               "v1",
+					InsecureSkipTLSVerify: true,
+				},
+			},
+			expectedStatusCode: http.StatusOK,
+			expectedCalled:     true,
+			expectedHeaders: map[string][]string{
+				"X-Forwarded-Proto": {"https"},
+				"X-Forwarded-Uri":   {"/request/path"},
+				"X-Remote-User":     {"username"},
+				"User-Agent":        {"Go-http-client/1.1"},
+				"Accept-Encoding":   {"gzip"},
+				"X-Remote-Group":    {"one", "two"},
+			},
+		},
+		"fail on bad serving cert": {
+			user: &user.DefaultInfo{
+				Name:   "username",
+				Groups: []string{"one", "two"},
+			},
+			path: "/request/path",
+			apiService: &apiregistration.APIService{
+				ObjectMeta: api.ObjectMeta{Name: "v1.foo"},
+				Spec: apiregistration.APIServiceSpec{
+					Group:   "foo",
+					Version: "v1",
+				},
+			},
+			expectedStatusCode: http.StatusServiceUnavailable,
+		},
+	}
+
+	for name, tc := range tests {
+		target.Reset()
+		handler.contextMapper = &fakeRequestContextMapper{user: tc.user}
+		handler.removeAPIService()
+		if tc.apiService != nil {
+			handler.updateAPIService(tc.apiService)
+			handler.destinationHost = targetServer.Listener.Addr().String()
+		}
+
+		resp, err := http.Get(server.URL + tc.path)
+		if err != nil {
+			t.Errorf("%s: %v", name, err)
+			continue
+		}
+		if e, a := tc.expectedStatusCode, resp.StatusCode; e != a {
+			body, _ := httputil.DumpResponse(resp, true)
+			t.Logf("%s: %v", name, string(body))
+			t.Errorf("%s: expected %v, got %v", name, e, a)
+			continue
+		}
+		bytes, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			t.Errorf("%s: %v", name, err)
+			continue
+		}
+		if !strings.Contains(string(bytes), tc.expectedBody) {
+			t.Errorf("%s: expected %q, got %q", name, tc.expectedBody, string(bytes))
+			continue
+		}
+
+		if e, a := tc.expectedCalled, target.called; e != a {
+			t.Errorf("%s: expected %v, got %v", name, e, a)
+			continue
+		}
+		// this varies every test
+		delete(target.headers, "X-Forwarded-Host")
+		if e, a := tc.expectedHeaders, target.headers; !reflect.DeepEqual(e, a) {
+			t.Errorf("%s: expected %v, got %v", name, e, a)
+			continue
+		}
+	}
+}

cmd/kubernetes-discovery/pkg/cmd/server/BUILD

@@ -20,12 +20,14 @@ go_library(
         "//cmd/kubernetes-discovery/pkg/legacy:go_default_library",
         "//pkg/api:go_default_library",
         "//pkg/genericapiserver:go_default_library",
+        "//pkg/genericapiserver/filters:go_default_library",
         "//pkg/genericapiserver/options:go_default_library",
         "//pkg/kubectl/cmd/util:go_default_library",
         "//pkg/registry/generic:go_default_library",
         "//pkg/registry/generic/registry:go_default_library",
         "//pkg/runtime/schema:go_default_library",
         "//pkg/storage/storagebackend:go_default_library",
+        "//pkg/util/sets:go_default_library",
         "//pkg/util/wait:go_default_library",
         "//vendor:github.com/pborman/uuid",
         "//vendor:github.com/spf13/cobra",

cmd/kubernetes-discovery/pkg/cmd/server/start.go

@@ -19,6 +19,7 @@ package server
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 
 	"github.com/pborman/uuid"
 	"github.com/spf13/cobra"
@@ -27,12 +28,14 @@ import (
 	"k8s.io/kubernetes/cmd/kubernetes-discovery/pkg/legacy"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/genericapiserver"
+	"k8s.io/kubernetes/pkg/genericapiserver/filters"
 	genericoptions "k8s.io/kubernetes/pkg/genericapiserver/options"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
 	"k8s.io/kubernetes/pkg/registry/generic"
 	"k8s.io/kubernetes/pkg/registry/generic/registry"
 	"k8s.io/kubernetes/pkg/runtime/schema"
 	"k8s.io/kubernetes/pkg/storage/storagebackend"
+	"k8s.io/kubernetes/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/util/wait"
 
 	"k8s.io/kubernetes/cmd/kubernetes-discovery/pkg/apis/apiregistration/v1alpha1"
@@ -46,6 +49,11 @@ type DiscoveryServerOptions struct {
 	Authentication *genericoptions.DelegatingAuthenticationOptions
 	Authorization  *genericoptions.DelegatingAuthorizationOptions
 
+	// ProxyClientCert/Key are the client cert used to identify this proxy. Backing APIServices use
+	// this to confirm the proxy's identity
+	ProxyClientCertFile string
+	ProxyClientKeyFile  string
+
 	StdOut io.Writer
 	StdErr io.Writer
 }
@@ -81,6 +89,8 @@ func NewCommandStartDiscoveryServer(out, err io.Writer) *cobra.Command {
 	o.SecureServing.AddFlags(flags)
 	o.Authentication.AddFlags(flags)
 	o.Authorization.AddFlags(flags)
+	flags.StringVar(&o.ProxyClientCertFile, "proxy-client-cert-file", o.ProxyClientCertFile, "client certificate used identify the proxy to the API server")
+	flags.StringVar(&o.ProxyClientKeyFile, "proxy-client-key-file", o.ProxyClientKeyFile, "client certificate key used identify the proxy to the API server")
 
 	return cmd
 }
@@ -114,6 +124,10 @@ func (o DiscoveryServerOptions) RunDiscoveryServer() error {
 	if _, err := genericAPIServerConfig.ApplyDelegatingAuthorizationOptions(o.Authorization); err != nil {
 		return err
 	}
+	genericAPIServerConfig.LongRunningFunc = filters.BasicLongRunningRequestCheck(
+		sets.NewString("watch", "proxy"),
+		sets.NewString("attach", "exec", "proxy", "log", "portforward"),
+	)
 
 	var err error
 	privilegedLoopbackToken := uuid.NewRandom().String()
@@ -126,6 +140,15 @@ func (o DiscoveryServerOptions) RunDiscoveryServer() error {
 		RESTOptionsGetter: &restOptionsFactory{storageConfig: &o.Etcd.StorageConfig},
 	}
 
+	config.ProxyClientCert, err = ioutil.ReadFile(o.ProxyClientCertFile)
+	if err != nil {
+		return err
+	}
+	config.ProxyClientKey, err = ioutil.ReadFile(o.ProxyClientKeyFile)
+	if err != nil {
+		return err
+	}
+
 	server, err := config.Complete().New()
 	if err != nil {
 		return err

hack/verify-flags/known-flags.txt

@@ -469,6 +469,8 @@ prom-push-gateway
 protect-kernel-defaults
 proto-import
 proxy-bindall
+proxy-client-cert-file
+proxy-client-key-file
 proxy-kubeconfig
 proxy-logv
 proxy-mode