Merge pull request #58644 from yguo0905/webhooks

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Use SSH tunnel for webhook communication iff the webhook is deployed as a service

**What this PR does / why we need it**:

We are getting the following error when the apiserver connects the webhook on localhost (configured via URL). We should only use the SSL tunnel for the connections to nodes when the webhooks are running as services.

```
I0119 17:41:18.678436       1 ssh.go:400] [4cdf44753cc3705d: localhost:10258] Dialing...
W0119 17:41:18.678483       1 ssh.go:424] SSH tunnel not found for address "localhost", picking random node
I0119 17:41:18.679810       1 ssh.go:402] [4cdf44753cc3705d: localhost:10258] Dialed in 1.398691ms.
W0119 17:41:18.679928       1 admission.go:256] Failed calling webhook, failing closed xxx: failed calling admission webhook "xxx": Post xxx: ssh: rejected: connect failed (Connection refused)
I0119 17:41:18.680346       1 wrap.go:42] POST /api/v1/namespaces/kube-system/pods: (5.725588ms) 500
```

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes # https://github.com/kubernetes/kubernetes/issues/58779

**Special notes for your reviewer**:

**Release note**:

```release-note
kube-apiserver is changed to use SSH tunnels for webhook iff the webhook is not directly routable from apiserver's network environment.
```

/assign @lavalamp @caesarxuchao @cheftako

cmd/kube-apiserver/app/server.go

@@ -463,27 +463,35 @@ func BuildGenericConfig(s *options.ServerRunOptions, proxyTransport *http.Transp
 		genericConfig.DisabledPostStartHooks.Insert(rbacrest.PostStartHookName)
 	}
 
-	webhookAuthResolver := func(delegate webhookconfig.AuthenticationInfoResolver) webhookconfig.AuthenticationInfoResolver {
-		return webhookconfig.AuthenticationInfoResolverFunc(func(server string) (*rest.Config, error) {
-			if server == "kubernetes.default.svc" {
-				return genericConfig.LoopbackClientConfig, nil
-			}
-			ret, err := delegate.ClientConfigFor(server)
-			if err != nil {
-				return nil, err
-			}
-			if proxyTransport != nil && proxyTransport.Dial != nil {
-				ret.Dial = proxyTransport.Dial
-			}
-			return ret, err
-		})
+	webhookAuthResolverWrapper := func(delegate webhookconfig.AuthenticationInfoResolver) webhookconfig.AuthenticationInfoResolver {
+		return &webhookconfig.AuthenticationInfoResolverDelegator{
+			ClientConfigForFunc: func(server string) (*rest.Config, error) {
+				if server == "kubernetes.default.svc" {
+					return genericConfig.LoopbackClientConfig, nil
+				}
+				return delegate.ClientConfigFor(server)
+			},
+			ClientConfigForServiceFunc: func(serviceName, serviceNamespace string) (*rest.Config, error) {
+				if serviceName == "kubernetes" && serviceNamespace == "default" {
+					return genericConfig.LoopbackClientConfig, nil
+				}
+				ret, err := delegate.ClientConfigForService(serviceName, serviceNamespace)
+				if err != nil {
+					return nil, err
+				}
+				if proxyTransport != nil && proxyTransport.Dial != nil {
+					ret.Dial = proxyTransport.Dial
+				}
+				return ret, err
+			},
+		}
 	}
 	pluginInitializers, err := BuildAdmissionPluginInitializers(
 		s,
 		client,
 		sharedInformers,
 		serviceResolver,
-		webhookAuthResolver,
+		webhookAuthResolverWrapper,
 	)
 	if err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("failed to create admission plugin initializer: %v", err)

staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/config/authentication.go

@@ -31,17 +31,28 @@ import (
 // rest.Config generated by the resolver.
 type AuthenticationInfoResolverWrapper func(AuthenticationInfoResolver) AuthenticationInfoResolver
 
-// AuthenticationInfoResolver builds rest.Config base on the server name.
+// AuthenticationInfoResolver builds rest.Config base on the server or service
+// name and service namespace.
 type AuthenticationInfoResolver interface {
+	// ClientConfigFor builds rest.Config based on the server.
 	ClientConfigFor(server string) (*rest.Config, error)
+	// ClientConfigForService builds rest.Config based on the serviceName and
+	// serviceNamespace.
+	ClientConfigForService(serviceName, serviceNamespace string) (*rest.Config, error)
 }
 
-// AuthenticationInfoResolverFunc implements AuthenticationInfoResolver.
-type AuthenticationInfoResolverFunc func(server string) (*rest.Config, error)
+// AuthenticationInfoResolverDelegator implements AuthenticationInfoResolver.
+type AuthenticationInfoResolverDelegator struct {
+	ClientConfigForFunc        func(server string) (*rest.Config, error)
+	ClientConfigForServiceFunc func(serviceName, serviceNamespace string) (*rest.Config, error)
+}
 
-//ClientConfigFor implements AuthenticationInfoResolver.
-func (a AuthenticationInfoResolverFunc) ClientConfigFor(server string) (*rest.Config, error) {
-	return a(server)
+func (a *AuthenticationInfoResolverDelegator) ClientConfigFor(server string) (*rest.Config, error) {
+	return a.ClientConfigForFunc(server)
+}
+
+func (a *AuthenticationInfoResolverDelegator) ClientConfigForService(serviceName, serviceNamespace string) (*rest.Config, error) {
+	return a.ClientConfigForServiceFunc(serviceName, serviceNamespace)
 }
 
 type defaultAuthenticationInfoResolver struct {
@@ -68,13 +79,21 @@ func NewDefaultAuthenticationInfoResolver(kubeconfigFile string) (Authentication
 }
 
 func (c *defaultAuthenticationInfoResolver) ClientConfigFor(server string) (*rest.Config, error) {
+	return c.clientConfig(server)
+}
+
+func (c *defaultAuthenticationInfoResolver) ClientConfigForService(serviceName, serviceNamespace string) (*rest.Config, error) {
+	return c.clientConfig(serviceName + "." + serviceNamespace + ".svc")
+}
+
+func (c *defaultAuthenticationInfoResolver) clientConfig(target string) (*rest.Config, error) {
 	// exact match
-	if authConfig, ok := c.kubeconfig.AuthInfos[server]; ok {
+	if authConfig, ok := c.kubeconfig.AuthInfos[target]; ok {
 		return restConfigFromKubeconfig(authConfig)
 	}
 
 	// star prefixed match
-	serverSteps := strings.Split(server, ".")
+	serverSteps := strings.Split(target, ".")
 	for i := 1; i < len(serverSteps); i++ {
 		nickName := "*." + strings.Join(serverSteps[i:], ".")
 		if authConfig, ok := c.kubeconfig.AuthInfos[nickName]; ok {
@@ -83,7 +102,7 @@ func (c *defaultAuthenticationInfoResolver) ClientConfigFor(server string) (*res
 	}
 
 	// if we're trying to hit the kube-apiserver and there wasn't an explicit config, use the in-cluster config
-	if server == "kubernetes.default.svc" {
+	if target == "kubernetes.default.svc" {
 		// if we can find an in-cluster-config use that.  If we can't, fall through.
 		inClusterConfig, err := rest.InClusterConfig()
 		if err == nil {

staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/config/client.go

@@ -122,12 +122,12 @@ func (cm *ClientManager) HookClient(h *v1beta1.Webhook) (*rest.RESTClient, error
 	}
 
 	if svc := h.ClientConfig.Service; svc != nil {
-		serverName := svc.Name + "." + svc.Namespace + ".svc"
-		restConfig, err := cm.authInfoResolver.ClientConfigFor(serverName)
+		restConfig, err := cm.authInfoResolver.ClientConfigForService(svc.Name, svc.Namespace)
 		if err != nil {
 			return nil, err
 		}
 		cfg := rest.CopyConfig(restConfig)
+		serverName := svc.Name + "." + svc.Namespace + ".svc"
 		host := serverName + ":443"
 		cfg.Host = "https://" + host
 		if svc.Path != nil {

staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/mutating/admission_test.go

@@ -636,6 +636,11 @@ func (c *fakeAuthenticationInfoResolver) ClientConfigFor(server string) (*rest.C
 	return c.restConfig, nil
 }
 
+func (c *fakeAuthenticationInfoResolver) ClientConfigForService(serviceName, serviceNamespace string) (*rest.Config, error) {
+	atomic.AddInt32(c.cachedCount, 1)
+	return c.restConfig, nil
+}
+
 func newMatchEverythingRules() []registrationv1beta1.RuleWithOperations {
 	return []registrationv1beta1.RuleWithOperations{{
 		Operations: []registrationv1beta1.OperationType{registrationv1beta1.OperationAll},

staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/validating/admission_test.go

@@ -661,6 +661,11 @@ func (c *fakeAuthenticationInfoResolver) ClientConfigFor(server string) (*rest.C
 	return c.restConfig, nil
 }
 
+func (c *fakeAuthenticationInfoResolver) ClientConfigForService(serviceName, serviceNamespace string) (*rest.Config, error) {
+	atomic.AddInt32(c.cachedCount, 1)
+	return c.restConfig, nil
+}
+
 func newMatchEverythingRules() []registrationv1beta1.RuleWithOperations {
 	return []registrationv1beta1.RuleWithOperations{{
 		Operations: []registrationv1beta1.OperationType{registrationv1beta1.OperationAll},