Information disclosure / Improper routing due to stale connection of webhook client (misrouting)

PoC (Go test-style proof of concept) demonstrating the behavior: 1) Setup a fake service resolver returning two endpoints (ServerA and ServerB). 2) Enable the feature WebhookRoundTripLoadBalancing and issue two webhook requests. 3) Observe that the first request is sent to ServerA, and the second request can be routed to ServerB when the resolver exposes a new endpoint, illustrating that the client now follows the resolved IP rather than reusing a pre-established connection. Code sketch (adapted from the test in the patch): package webhook import ( "net/http" "net/url" "sync/atomic" "testing" "time" "k8s.io/apiserver/pkg/util/feature" utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/apimachinery/pkg/runtime/schema" ) func TestWebhookClientIdleConnectionIPReuse_PoC(t *testing.T) { // Pseudo-test showing the behavior with the fake resolver. // Two endpoints A and B to simulate IP changes behind a service. var aCalls, bCalls int32 serverA := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request){ atomic.AddInt32(&aCalls, 1); w.Write([]byte("ServerA")) })) defer serverA.Close() serverB := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request){ atomic.AddInt32(&bCalls, 1); w.Write([]byte("ServerB")) })) defer serverB.Close() urlA, _ := url.Parse(serverA.URL) urlB, _ := url.Parse(serverB.URL) resolver := &fakeDynamicServiceResolver{endpoints: []*url.URL{urlA, urlB}} cm, _ := NewClientManager([]schema.GroupVersion{}) cm.SetAuthenticationInfoResolver(&fakeAuthInfoResolver{}) cm.SetServiceResolver(resolver) cc := ClientConfig{ Name: "test-webhook", CABundle: nil, Service: &ClientConfigService{ Name: "test-service", Namespace: "default", Port: 443 } } // Enable feature gate for the round trip load balancing featuregate := utilfeature.DefaultFeatureGate // Assume we toggle the feature elsewhere in the test harness as in the repo client, err := cm.HookClient(cc) if err != nil { t.Fatal(err) } // First request should hit ServerA (resolver returns A first) req1 := client.Post().Body([]byte("test")) _, err = req1.DoRaw(context.Background()) if err != nil { t.Fatal(err) } // Second request should hit ServerB if feature gate is enabled; otherwise it may reuse the cached connection to A. req2 := client.Post().Body([]byte("test")) res2, err := req2.DoRaw(context.Background()) if err != nil { t.Fatal(err) } // Interpret res2 to determine whether it came from A or B _ = res2 // Verify endpoints were contacted accordingly per the feature gate state used in the test harness _ = aCalls; _ = bCalls; _ = time.Now } Note: The exact wiring of the test harness, feature gates, and client initialization mirrors the approach in the patch: a custom RoundTripper resolves the endpoint per request when the feature is enabled, otherwise it keeps dialing through the cached connection. The key observable effect is that with the feature gate enabled, the second request can be routed to a different endpoint (ServerB) after a simulated IP change; without the feature gate, the client may reuse the old connection (ServerA).

diff --git a/pkg/features/kube_features.go b/pkg/features/kube_features.go index a96849af298d7..e5e964e1b438a 100644 --- a/pkg/features/kube_features.go +++ b/pkg/features/kube_features.go @@ -2222,6 +2222,10 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate {Version: version.MustParse("1.34"), Default: true, PreRelease: featuregate.Beta}, }, + genericfeatures.WebhookRoundTripLoadBalancing: { + {Version: version.MustParse("1.37"), Default: true, PreRelease: featuregate.Beta}, + }, + kcmfeatures.CloudControllerManagerWatchBasedRoutesReconciliation: { {Version: version.MustParse("1.35"), Default: false, PreRelease: featuregate.Alpha}, }, @@ -2663,6 +2667,8 @@ var defaultKubernetesFeatureGateDependencies = map[featuregate.Feature][]feature genericfeatures.WatchList: {}, + genericfeatures.WebhookRoundTripLoadBalancing: {}, + kcmfeatures.CloudControllerManagerWatchBasedRoutesReconciliation: {}, kcmfeatures.CloudControllerManagerWebhook: {}, diff --git a/staging/src/k8s.io/apiserver/pkg/features/kube_features.go b/staging/src/k8s.io/apiserver/pkg/features/kube_features.go index 8e6cc4664edd9..201f1d4f30ebc 100644 --- a/staging/src/k8s.io/apiserver/pkg/features/kube_features.go +++ b/staging/src/k8s.io/apiserver/pkg/features/kube_features.go @@ -276,6 +276,14 @@ const ( // // Allow the API server to stream individual items instead of chunking WatchList featuregate.Feature = "WatchList" + + // owner: @aojea + // beta: v1.37 + // + // Enables using a custom resolving RoundTripper to load-balance admission + // webhook requests across service endpoints instead of caching connections + // by service name. + WebhookRoundTripLoadBalancing featuregate.Feature = "WebhookRoundTripLoadBalancing" ) func init() { @@ -452,4 +460,8 @@ var defaultVersionedKubernetesFeatureGates = map[featuregate.Feature]featuregate {Version: version.MustParse("1.33"), Default: false, PreRelease: featuregate.Beta}, {Version: version.MustParse("1.34"), Default: true, PreRelease: featuregate.Beta}, }, + + WebhookRoundTripLoadBalancing: { + {Version: version.MustParse("1.37"), Default: true, PreRelease: featuregate.Beta}, + }, } diff --git a/staging/src/k8s.io/apiserver/pkg/util/webhook/client.go b/staging/src/k8s.io/apiserver/pkg/util/webhook/client.go index 63ea4e2666e81..078e4377cb93f 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/webhook/client.go +++ b/staging/src/k8s.io/apiserver/pkg/util/webhook/client.go @@ -22,6 +22,7 @@ import ( "errors" "fmt" "net" + "net/http" "net/url" "strconv" "strings" @@ -30,6 +31,8 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/runtime/serializer" utilerrors "k8s.io/apimachinery/pkg/util/errors" + "k8s.io/apiserver/pkg/features" + utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/apiserver/pkg/util/x509metrics" "k8s.io/client-go/rest" "k8s.io/utils/lru" @@ -194,20 +197,36 @@ func (cm *ClientManager) hookClientConfig(cc ClientConfig) (*rest.Config, error) cfg.TLSClientConfig.ServerName = serverName } - delegateDialer := cfg.Dial - if delegateDialer == nil { - var d net.Dialer - delegateDialer = d.DialContext - } - cfg.Dial = func(ctx context.Context, network, addr string) (net.Conn, error) { - if addr == host { - u, err := cm.serviceResolver.ResolveEndpoint(cc.Service.Namespace, cc.Service.Name, port) - if err != nil { - return nil, err + if !utilfeature.DefaultFeatureGate.Enabled(features.WebhookRoundTripLoadBalancing) { + delegateDialer := cfg.Dial + if delegateDialer == nil { + var d net.Dialer + delegateDialer = d.DialContext + } + cfg.Dial = func(ctx context.Context, network, addr string) (net.Conn, error) { + if addr == host { + u, err := cm.serviceResolver.ResolveEndpoint(cc.Service.Namespace, cc.Service.Name, port) + if err != nil { + return nil, err + } + addr = u.Host } - addr = u.Host + return delegateDialer(ctx, network, addr) } - return delegateDialer(ctx, network, addr) + } else { + // Use a custom roundtripper since http transport caches + // the connections by the URL. Host. The service resolver + // provides the actual endpoint address, if we use a custom + // dialer then the cached connection may not be closed. + cfg.Wrap(func(rt http.RoundTripper) http.RoundTripper { + return &resolvingRoundTripper{ + delegate: rt, + serviceResolver: cm.serviceResolver, + namespace: cc.Service.Namespace, + serviceName: cc.Service.Name, + port: port, + } + }) } return complete(cfg) @@ -255,3 +274,36 @@ func isLocalHost(u *url.URL) bool { } return false } + +// resolvingRoundTripper is a roundtripper that resolves the endpoint address +// for the given service and updates the request URL to use the resolved endpoint address. +type resolvingRoundTripper struct { + delegate http.RoundTripper + serviceResolver ServiceResolver + namespace string + serviceName string + port int32 +} + +func (r *resolvingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + serverName := r.serviceName + "." + r.namespace + ".svc" + host := net.JoinHostPort(serverName, strconv.Itoa(int(r.port))) + if req.URL.Host != host { + return r.delegate.RoundTrip(req) + } + u, err := r.serviceResolver.ResolveEndpoint(r.namespace, r.serviceName, r.port) + if err != nil { + return nil, err + } + newReq := req.Clone(req.Context()) + // Preserve the original Host header + if len(newReq.Host) == 0 { + newReq.Host = req.URL.Host + } + newReq.URL.Host = u.Host + return r.delegate.RoundTrip(newReq) +} + +func (r *resolvingRoundTripper) WrappedRoundTripper() http.RoundTripper { + return r.delegate +} diff --git a/staging/src/k8s.io/apiserver/pkg/util/webhook/client_test.go b/staging/src/k8s.io/apiserver/pkg/util/webhook/client_test.go index dd00f238dc549..26e9c1fce68b6 100644 --- a/staging/src/k8s.io/apiserver/pkg/util/webhook/client_test.go +++ b/staging/src/k8s.io/apiserver/pkg/util/webhook/client_test.go @@ -17,10 +17,20 @@ limitations under the License. package webhook import ( + "context" + "encoding/pem" + "net/http" + "net/http/httptest" + "net/url" + "sync/atomic" "testing" "golang.org/x/net/http2" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apiserver/pkg/features" + utilfeature "k8s.io/apiserver/pkg/util/feature" + "k8s.io/client-go/rest" + featuregatetesting "k8s.io/component-base/featuregate/testing" ) func TestWebhookClientConfig(t *testing.T) { @@ -89,3 +99,155 @@ func allowHTTP2(nextProtos []string) bool { // the transport explicitly set NextProtos and excluded http/2 return false } + +type fakeAuthInfoResolver struct{} + +func (f *fakeAuthInfoResolver) ClientConfigFor(server string) (*rest.Config, error) { + return &rest.Config{ + TLSClientConfig: rest.TLSClientConfig{ + ServerName: "example.com", + }, + }, nil +} + +func (f *fakeAuthInfoResolver) ClientConfigForService(serviceName, namespace string, port int) (*rest.Config, error) { + return &rest.Config{ + TLSClientConfig: rest.TLSClientConfig{ + ServerName: "example.com", + }, + }, nil +} + +// fakeDynamicServiceResolver returns the next endpoint in the list for each request. +type fakeDynamicServiceResolver struct { + endpoints []*url.URL + counter int32 +} + +func (f *fakeDynamicServiceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) { + val := atomic.AddInt32(&f.counter, 1) - 1 + if val >= int32(len(f.endpoints)) { + val = int32(len(f.endpoints)) - 1 + } + return f.endpoints[val], nil +} + +// TestWebhookClientIdleConnectionIPReuse tests that the webhook client follow the resolver +// endpoint instead of reusing the previous endpoint when there are IP address changes. +func TestWebhookClientIdleConnectionIPReuse(t *testing.T) { + tests := []struct { + name string + enableFeatureGate bool + expectedServerACalls int32 + expectedServerBCalls int32 + expectedSecondResponse string + }{ + { + name: "feature gate enabled - round-trip load balancing routes to Server B", + enableFeatureGate: true, + expectedServerACalls: 1, + expectedServerBCalls: 1, + expectedSecondResponse: "ServerB", + }, + { + name: "feature gate disabled - dialer resolution caches to Server A", + enableFeatureGate: false, + expectedServerACalls: 2, + expectedServerBCalls: 0, + expectedSecondResponse: "ServerA", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.WebhookRoundTripLoadBalancing, tc.enableFeatureGate) + + var serverACalls int32 + serverA := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&serverACalls, 1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ServerA")) + })) + defer serverA.Close() + + var serverBCalls int32 + serverB := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&serverBCalls, 1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ServerB")) + })) + defer serverB.Close() + + urlA, err := url.Parse(serverA.URL) + if err != nil { + t.Fatal(err) + } + urlB, err := url.Parse(serverB.URL) + if err != nil { + t.Fatal(err) + } + + // Combine CAs from both test servers + var caBundle []byte + for _, cert := range serverA.TLS.Certificates[0].Certificate { + caBundle = append(caBundle, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert})...) + } + for _, cert := range serverB.TLS.Certificates[0].Certificate { + caBundle = append(caBundle, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert})...) + } + + resolver := &fakeDynamicServiceResolver{ + endpoints: []*url.URL{urlA, urlB}, + } + + cm, err := NewClientManager([]schema.GroupVersion{}) + if err != nil { + t.Fatal(err) + } + cm.SetAuthenticationInfoResolver(&fakeAuthInfoResolver{}) + cm.SetServiceResolver(resolver) + + cc := ClientConfig{ + Name: "test-webhook", + CABundle: caBundle, + Service: &ClientConfigService{ + Name: "test-service", + Namespace: "default", + Port: 443, + }, + } + + client, err := cm.HookClient(cc) + if err != nil { + t.Fatal(err) + } + + // Request 1: Resolves to Server A + req1 := client.Post().Body([]byte("test")) + res1, err := req1.DoRaw(context.Background()) + if err != nil { + t.Fatalf("First request failed: %v", err) + } + if string(res1) != "ServerA" { + t.Errorf("Expected Response ServerA, got %s", string(res1)) + } + + // Request 2: Resolves to Server B if feature gate enabled, Server A if disabled + req2 := client.Post().Body([]byte("test")) + res2, err := req2.DoRaw(context.Background()) + if err != nil { + t.Fatalf("Second request failed: %v", err) + } + if string(res2) != tc.expectedSecondResponse { + t.Errorf("Expected Response %s, got %s", tc.expectedSecondResponse, string(res2)) + } + + if callsA := atomic.LoadInt32(&serverACalls); callsA != tc.expectedServerACalls { + t.Errorf("Expected %d calls to Server A, got %d", tc.expectedServerACalls, callsA) + } + if callsB := atomic.LoadInt32(&serverBCalls); callsB != tc.expectedServerBCalls { + t.Errorf("Expected %d calls to Server B, got %d", tc.expectedServerBCalls, callsB) + } + }) + } +} diff --git a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml index 9399331d9e60c..e2209f346bb6c 100644 --- a/test/compatibility_lifecycle/reference/versioned_feature_list.yaml +++ b/test/compatibility_lifecycle/reference/versioned_feature_list.yaml @@ -2047,6 +2047,12 @@ lockToDefault: false preRelease: Beta version: "1.34" +- name: WebhookRoundTripLoadBalancing + versionedSpecs: + - default: true + lockToDefault: false + preRelease: Beta + version: "1.37" - name: WindowsCPUAndMemoryAffinity versionedSpecs: - default: false diff --git a/test/e2e/apimachinery/webhook.go b/test/e2e/apimachinery/webhook.go index ae3063ecbd86a..d1edbd42d4d13 100644 --- a/test/e2e/apimachinery/webhook.go +++ b/test/e2e/apimachinery/webhook.go @@ -392,6 +392,77 @@ var _ = SIGDescribe("AdmissionWebhook [Privileged:ClusterAdmin]", func() { slowWebhookCleanup(ctx) }) + /* + Release: v1.31 + Testname: Admission webhook, connection pool isolation on pod restart + Description: Verify that the admission webhook connection pool is isolated by resolved IP address. + When the backend pod behind the service is recreated and its IP address changes, subsequent admission + requests must successfully route to the new pod IP address without hitting the old connection pool. + */ + ginkgo.It("should isolate connection pool by resolved backend IP address on pod restart", func(ctx context.Context) { + client := f.ClientSet + + ginkgo.By("Registering a validating webhook") + registerWebhook(ctx, f, markersNamespaceName, f.UniqueName, certCtx, servicePort) + + // Get the active webhook pod details + ginkgo.By("Getting the current webhook pod details") + pods, err := client.CoreV1().Pods(f.Namespace.Name).List(ctx, metav1.ListOptions{LabelSelector: "app=sample-webhook"}) + framework.ExpectNoError(err) + gomega.Expect(pods.Items).To(gomega.HaveLen(1)) + oldPod := pods.Items[0] + oldIP := oldPod.Status.PodIP + framework.Logf("Webhook currently running on pod %s with IP %s", oldPod.Name, oldIP) + + // Recreate/restart the webhook pod by rolling out a change to the deployment + ginkgo.By("Triggering a rolling update of the webhook deployment") + deployment, err := client.AppsV1().Deployments(f.Namespace.Name).Get(ctx, deploymentName, metav1.GetOptions{}) + framework.ExpectNoError(err) + + // Update the deployment template labels or env to trigger a rolling update + if deployment.Spec.Template.Annotations == nil { + deployment.Spec.Template.Annotations = make(map[string]string) + } + deployment.Spec.Template.Annotations["restartedAt"] = time.Now().Format(time.RFC3339) + + deployment, err = client.AppsV1().Deployments(f.Namespace.Name).Update(ctx, deployment, metav1.UpdateOptions{}) + framework.ExpectNoError(err) + + ginkgo.By("Wait for the deployment rolling update to complete") + err = e2edeployment.WaitForDeploymentComplete(client, deployment) + framework.ExpectNoError(err) + + // Get the new webhook pod details + ginkgo.By("Getting the new webhook pod details") + newPods, err := client.CoreV1().Pods(f.Namespace.Name).List(ctx, metav1.ListOptions{LabelSelector: "app=sample-webhook"}) + framework.ExpectNoError(err) + gomega.Expect(newPods.Items).To(gomega.HaveLen(1)) + newPod := newPods ... [truncated]