diff --git a/pkg/controlplane/apiserver/config.go b/pkg/controlplane/apiserver/config.go index dec22de82fe..7ba9958e77d 100644 --- a/pkg/controlplane/apiserver/config.go +++ b/pkg/controlplane/apiserver/config.go @@ -330,7 +330,9 @@ func CreateConfig( opts.PeerAdvertiseAddress, genericConfig.APIServerID, config.Extra.PeerEndpointLeaseReconciler, - config.Generic.Serializer) + config.Generic.Serializer, + config.Generic.EgressSelector, + ) if err != nil { return nil, nil, err } diff --git a/pkg/controlplane/apiserver/peer.go b/pkg/controlplane/apiserver/peer.go index 9806bf25701..3d1fa8a5418 100644 --- a/pkg/controlplane/apiserver/peer.go +++ b/pkg/controlplane/apiserver/peer.go @@ -28,6 +28,7 @@ import ( "k8s.io/client-go/transport" genericapiserver "k8s.io/apiserver/pkg/server" + "k8s.io/apiserver/pkg/server/egressselector" serverstorage "k8s.io/apiserver/pkg/server/storage" utilpeerproxy "k8s.io/apiserver/pkg/util/peerproxy" coordinationv1informers "k8s.io/client-go/informers/coordination/v1" @@ -52,7 +53,9 @@ func BuildPeerProxy( peerAdvertiseAddress reconcilers.PeerAdvertiseAddress, apiServerID string, reconciler reconcilers.PeerEndpointLeaseReconciler, - serializer runtime.NegotiatedSerializer) (utilpeerproxy.Interface, error) { + serializer runtime.NegotiatedSerializer, + es *egressselector.EgressSelector, +) (utilpeerproxy.Interface, error) { if proxyClientCertFile == "" { return nil, fmt.Errorf("error building peer proxy handler, proxy-cert-file not specified") } @@ -69,6 +72,19 @@ func BuildPeerProxy( ServerName: "kubernetes.default.svc", }} + if es != nil { + egressDialer, err := es.Lookup(egressselector.ControlPlane.AsNetworkContext()) + if err != nil { + return nil, err + } + + // We only set the DialHolder if an egress dialer is explicitly configured for the + // ControlPlane network context, otherwise the transport will fall back to the default direct dialer. + if egressDialer != nil { + proxyClientConfig.DialHolder = &transport.DialHolder{Dial: egressDialer} + } + } + return utilpeerproxy.NewPeerProxyHandler( apiServerID, IdentityLeaseComponentLabelKey+"="+KubeAPIServer, diff --git a/test/integration/apiserver/peerproxy/peer_proxy_test.go b/test/integration/apiserver/peerproxy/peer_proxy_test.go index 35e3e5b76f0..1ca46493c73 100644 --- a/test/integration/apiserver/peerproxy/peer_proxy_test.go +++ b/test/integration/apiserver/peerproxy/peer_proxy_test.go @@ -18,6 +18,13 @@ package peerproxy import ( "context" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "sync/atomic" "testing" "time" @@ -201,6 +208,77 @@ func TestPeerProxiedRequestToThirdServerAfterFirstDies(t *testing.T) { assert.Equal(t, job.Name, jobsB.Items[0].Name) } +func TestPeerProxy_EgressDialerIsUsed(t *testing.T) { + ktesting.SetDefaultVerbosity(1) + + // Start a HTTP CONNECT proxy to act as a fake egress dialer over UDS + dialerHit, udsName := runMockEgressProxy(t) + + // Write a temporary egress selector config file pointing to our proxy + egressSelectorConfigFile, err := os.CreateTemp("", "egress_selector_configuration.yaml") + require.NoError(t, err) + defer func() { _ = os.Remove(egressSelectorConfigFile.Name()) }() + + configYAML := fmt.Sprintf(` +apiVersion: apiserver.k8s.io/v1beta1 +kind: EgressSelectorConfiguration +egressSelections: +- name: controlplane + connection: + proxyProtocol: HTTPConnect + transport: + uds: + udsName: "%s" +`, udsName) + err = os.WriteFile(egressSelectorConfigFile.Name(), []byte(configYAML), 0644) + require.NoError(t, err) + + featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{ + features.APIServerIdentity: true, + features.UnknownVersionInteroperabilityProxy: true, + }) + + etcd := framework.SharedEtcd() + proxyCA, err := createProxyCertContent() + require.NoError(t, err) + + // Start two apiservers, one with batch/v1 disabled + server.SetHostnameFuncForTests("test-server-egress-a") + serverA := kastesting.StartTestServerOrDie(t, &kastesting.TestServerInstanceOptions{ + EnableCertAuth: true, + ProxyCA: &proxyCA}, + []string{"--runtime-config=api/all=true", "--egress-selector-config-file=" + egressSelectorConfigFile.Name()}, etcd) + t.Cleanup(serverA.TearDownFn) + + server.SetHostnameFuncForTests("test-server-egress-b") + serverB := kastesting.StartTestServerOrDie(t, &kastesting.TestServerInstanceOptions{ + EnableCertAuth: true, + ProxyCA: &proxyCA}, + []string{"--runtime-config=api/all=true,batch/v1=false", "--egress-selector-config-file=" + egressSelectorConfigFile.Name()}, etcd) + t.Cleanup(serverB.TearDownFn) + + kubeClientSetA, err := kubernetes.NewForConfig(serverA.ClientConfig) + require.NoError(t, err) + kubeClientSetB, err := kubernetes.NewForConfig(serverB.ClientConfig) + require.NoError(t, err) + + // create jobs resource using serverA + job := createJobResource() + _, err = kubeClientSetA.BatchV1().Jobs("default").Create(context.Background(), job, metav1.CreateOptions{}) + require.NoError(t, err) + + // List jobs using ServerB (should be proxied to ServerA, hitting our dialer) + jobsB, err := kubeClientSetB.BatchV1().Jobs("default").List(context.Background(), metav1.ListOptions{}) + require.NoError(t, err) + assert.NotEmpty(t, jobsB) + assert.Equal(t, job.Name, jobsB.Items[0].Name) + + // Assert that our dialer was hit at least once + if atomic.LoadInt32(dialerHit) == 0 { + t.Errorf("expected egress dialer to be used, but it was not hit") + } +} + func createProxyCertContent() (kastesting.ProxyCA, error) { result := kastesting.ProxyCA{} proxySigningKey, err := testutil.NewPrivateKey() @@ -240,3 +318,70 @@ func createJobResource() *v1.Job { }, } } + +func runMockEgressProxy(t *testing.T) (*int32, string) { + t.Helper() + udsFile, err := os.CreateTemp("/tmp", "egress-*.sock") + require.NoError(t, err) + udsName := udsFile.Name() + _ = udsFile.Close() + _ = os.Remove(udsName) // Must remove so we can bind + + listener, err := net.Listen("unix", udsName) + require.NoError(t, err) + t.Cleanup(func() { + _ = listener.Close() + _ = os.Remove(udsName) + }) + + dialerHit := new(int32) + proxy := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + atomic.AddInt32(dialerHit, 1) + + // A standard HTTP CONNECT proxy proxies opaque TCP streams. + if req.Method == http.MethodConnect { + // Actively establish a TCP connection to the destination API server + destConn, err := net.DialTimeout("tcp", req.Host, 5*time.Second) + if err != nil { + http.Error(w, err.Error(), http.StatusServiceUnavailable) + return + } + + // Acknowledge the connection stream has opened successfully + w.WriteHeader(http.StatusOK) + + // 'Hijack' the underlying network connection away from Go's + // generic HTTP Server so we can use it as a raw TCP proxy socket. + hijacker, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "Hijacking not supported", http.StatusInternalServerError) + return + } + clientConn, _, err := hijacker.Hijack() + if err != nil { + http.Error(w, err.Error(), http.StatusServiceUnavailable) + return + } + + // Spin up two concurrent forwarding pipes to pass traffic between + // the client and the destination over our UDS socket. + go func() { + _, _ = io.Copy(destConn, clientConn) + _ = destConn.Close() + _ = clientConn.Close() + }() + go func() { + _, _ = io.Copy(clientConn, destConn) + _ = destConn.Close() + _ = clientConn.Close() + }() + } else { + http.Error(w, "this is a proxy", http.StatusBadRequest) + } + })) + proxy.Listener = listener + proxy.Start() + t.Cleanup(proxy.Close) + + return dialerHit, udsName +}