[apiserversdk] check service belongs to kuberay

apiserversdk/go.mod

@@ -13,16 +13,21 @@ require (
 replace github.com/ray-project/kuberay/ray-operator => ../ray-operator
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blang/semver/v4 v4.0.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.12.1 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.11 // indirect
+	github.com/fsnotify/fsnotify v1.8.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.21.0 // indirect
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/btree v1.1.3 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20250403155104-27863c87afa6 // indirect
@@ -34,21 +39,31 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/prometheus/client_golang v1.22.0 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.62.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/spf13/pflag v1.0.6 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	go.opentelemetry.io/otel v1.33.0 // indirect
+	go.opentelemetry.io/otel/trace v1.33.0 // indirect
 	go.uber.org/automaxprocs v1.6.0 // indirect
 	golang.org/x/net v0.38.0 // indirect
 	golang.org/x/oauth2 v0.27.0 // indirect
+	golang.org/x/sync v0.13.0 // indirect
 	golang.org/x/sys v0.32.0 // indirect
 	golang.org/x/term v0.31.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
 	golang.org/x/time v0.10.0 // indirect
 	golang.org/x/tools v0.31.0 // indirect
+	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.33.0 // indirect
+	k8s.io/apiserver v0.33.0 // indirect
+	k8s.io/component-base v0.33.0 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20250318190949-c8a335a9a2ff // indirect
 	k8s.io/utils v0.0.0-20241210054802-24370beab758 // indirect

apiserversdk/proxy.go

@@ -6,7 +6,12 @@ import (
 	"net/url"
 	"strings"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/net"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
+
+	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 )
 
 type MuxConfig struct {
@@ -32,8 +37,13 @@ func NewMux(config MuxConfig) (*http.ServeMux, error) {
 	// TODO: add template features to specify routes.
 	mux.Handle("/apis/ray.io/v1/", handler)                                                                                    // forward KubeRay CR requests.
 	mux.Handle("GET /api/v1/namespaces/{namespace}/events", WithFieldSelector(handler, "involvedObject.apiVersion=ray.io/v1")) // allow querying KubeRay CR events.
-	// TODO: check whether the service belongs to KubeRay first.
-	mux.Handle("/api/v1/namespaces/{namespace}/services/{service}/proxy", handler) // allow accessing KubeRay dashboards and job submissions.
+
+	kuberayServiceFilterHandler, err := requireKuberayService(handler, config.KubernetesConfig)
+	if err != nil {
+		return nil, err
+	}
+	mux.Handle("/api/v1/namespaces/{namespace}/services/{service}/proxy/", kuberayServiceFilterHandler) // allow accessing KubeRay dashboards and job submissions.
+
 	return mux, nil
 }
 
@@ -46,3 +56,37 @@ func WithFieldSelector(handler http.Handler, selectors ...string) http.Handler {
 		handler.ServeHTTP(w, r)
 	})
 }
+
+func requireKuberayService(baseHandler http.Handler, config *rest.Config) (http.Handler, error) {
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		parts := strings.Split(r.URL.Path, "/")
+		if len(parts) < 7 {
+			http.Error(w, "invalid service proxy URL", http.StatusBadRequest)
+			return
+		}
+		namespace, rawService := parts[4], parts[6]
+		_, serviceName, _, valid := net.SplitSchemeNamePort(rawService)
+		if !valid {
+			http.Error(w, "invalid service proxy URL", http.StatusBadRequest)
+			return
+		}
+
+		svc, err := clientset.CoreV1().Services(namespace).Get(r.Context(), serviceName, metav1.GetOptions{})
+		if err != nil {
+			http.Error(w, "kuberay service not found", http.StatusNotFound)
+			return
+		}
+
+		if name, ok := svc.Labels[utils.KubernetesApplicationNameLabelKey]; !ok || name != utils.ApplicationName {
+			http.Error(w, "kuberay service not found", http.StatusNotFound)
+			return
+		}
+
+		baseHandler.ServeHTTP(w, r)
+	}), nil
+}

apiserversdk/proxy_test.go

@@ -2,6 +2,7 @@ package apiserversdk
 
 import (
 	"context"
+	"errors"
 	"net"
 	"net/http"
 	"path/filepath"
@@ -12,22 +13,26 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	k8sclient "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
 	rayclient "github.com/ray-project/kuberay/ray-operator/pkg/client/clientset/versioned/typed/ray/v1"
 )
 
 var (
-	ln        net.Listener
-	cfg       *rest.Config
-	rayClient *rayclient.RayV1Client
-	k8sClient *k8sclient.CoreV1Client
-	testEnv   *envtest.Environment
-	lastReq   atomic.Pointer[http.Request]
+	ln                    net.Listener
+	cfg                   *rest.Config
+	rayClient             *rayclient.RayV1Client
+	k8sClient             *k8sclient.CoreV1Client
+	k8sClientWithoutProxy *k8sclient.CoreV1Client
+	testEnv               *envtest.Environment
+	lastReq               atomic.Pointer[http.Request]
 )
 
 func TestProxy(t *testing.T) {
@@ -71,6 +76,7 @@ var _ = BeforeSuite(func(_ SpecContext) {
 	proxyCfg := &rest.Config{Host: "http://" + ln.Addr().String()}
 	rayClient = rayclient.NewForConfigOrDie(proxyCfg)
 	k8sClient = k8sclient.NewForConfigOrDie(proxyCfg)
+	k8sClientWithoutProxy = k8sclient.NewForConfigOrDie(cfg)
 })
 
 var _ = AfterSuite(func() {
@@ -164,10 +170,9 @@ var _ = Describe("events", Ordered, func() {
 		}
 		testEvent2 := testEvent.DeepCopy()
 		testEvent2.InvolvedObject.APIVersion = ""
-		tmpK8sClient := k8sclient.NewForConfigOrDie(cfg)
-		_, err := tmpK8sClient.Events("default").Create(context.Background(), testEvent, metav1.CreateOptions{})
+		_, err := k8sClientWithoutProxy.Events("default").Create(context.Background(), testEvent, metav1.CreateOptions{})
 		Expect(err).ToNot(HaveOccurred())
-		_, err = tmpK8sClient.Events("default").Create(context.Background(), testEvent2, metav1.CreateOptions{})
+		_, err = k8sClientWithoutProxy.Events("default").Create(context.Background(), testEvent2, metav1.CreateOptions{})
 		Expect(err).ToNot(HaveOccurred())
 		events, err := k8sClient.Events("default").List(context.Background(), metav1.ListOptions{})
 		Expect(err).ToNot(HaveOccurred())
@@ -188,3 +193,56 @@ var _ = Describe("not match", Ordered, func() {
 		Expect(err).To(MatchError(ContainSubstring("the server could not find the requested resource")))
 	})
 })
+
+var _ = Describe("kuberay service", Ordered, func() {
+	svcName := "head-svc"
+	AfterEach(func() {
+		err := k8sClientWithoutProxy.Services("default").Delete(context.Background(), svcName, metav1.DeleteOptions{})
+		Expect(err).ToNot(HaveOccurred())
+	})
+
+	It("should allow proxying to KubeRay-labeled service", func() {
+		_, err := k8sClientWithoutProxy.Services("default").Create(context.Background(), &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: svcName,
+				Labels: map[string]string{
+					utils.KubernetesApplicationNameLabelKey: utils.ApplicationName,
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{Name: "http", Port: 80, TargetPort: intstr.FromInt(80)},
+				},
+			},
+		}, metav1.CreateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		r := k8sClient.Services("default").ProxyGet("http", svcName, "80", "foo/bar", nil)
+		// Since envtest doesn't create a full K8s cluster but only the control plane, we cannot actually hit the pod.
+		// So we just check the request and skip checking the error which is always a 404.
+		_, _ = r.DoRaw(context.Background())
+		Expect(lastReq.Load().Method).To(Equal(http.MethodGet))
+		Expect(lastReq.Load().RequestURI).To(Equal("/api/v1/namespaces/default/services/http:head-svc:80/proxy/foo/bar"))
+	})
+	It("should reject proxying to service without KubeRay label", func() {
+		_, err := k8sClientWithoutProxy.Services("default").Create(context.Background(), &corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: svcName,
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{Name: "http", Port: 80, TargetPort: intstr.FromInt(80)},
+				},
+			},
+		}, metav1.CreateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		r := k8sClient.Services("default").ProxyGet("http", svcName, "80", "foo/bar", nil)
+		_, err = r.DoRaw(context.Background())
+		Expect(err).To(HaveOccurred())
+		var statusErr *apierrors.StatusError
+		ok := errors.As(err, &statusErr)
+		Expect(ok).To(BeTrue())
+		Expect(statusErr.Status().Details.Causes[0].Message).To(Equal("kuberay service not found"))
+	})
+})