Add configuration to conditionally disable shared pipeline resources

Signed-off-by: Krishan Bhasin <[email protected]>

Author: krishanbhasin-px

backend/src/apiserver/common/config.go

@@ -32,12 +32,17 @@ const (
 	KubeflowUserIDPrefix                    string = "KUBEFLOW_USERID_PREFIX"
 	UpdatePipelineVersionByDefault          string = "AUTO_UPDATE_PIPELINE_DEFAULT_VERSION"
 	TokenReviewAudience                     string = "TOKEN_REVIEW_AUDIENCE"
+	RequireNamespaceForPipelines            string = "REQUIRE_NAMESPACE_FOR_PIPELINES"
 )
 
 func IsPipelineVersionUpdatedByDefault() bool {
 	return GetBoolConfigWithDefault(UpdatePipelineVersionByDefault, true)
 }
 
+func IsNamespaceRequiredForPipelines() bool {
+	return GetBoolConfigWithDefault(RequireNamespaceForPipelines, false)
+}
+
 func GetStringConfig(configName string) string {
 	if !viper.IsSet(configName) {
 		glog.Fatalf("Please specify flag %s", configName)

backend/src/apiserver/server/pipeline_server.go

@@ -28,6 +28,7 @@ import (
 	"github.com/kubeflow/pipelines/backend/src/apiserver/list"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/model"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/resource"
+	"github.com/kubeflow/pipelines/backend/src/apiserver/validation"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -121,6 +122,10 @@ type PipelineServerV1 struct {
 // Applies common logic on v1beta1 and v2beta1 API.
 func (s *BasePipelineServer) createPipeline(ctx context.Context, pipeline *model.Pipeline) (*model.Pipeline, error) {
 	pipeline.Namespace = s.resourceManager.ReplaceNamespace(pipeline.Namespace)
+	err := validation.ValidateNamespaceRequired(pipeline.Namespace)
+	if err != nil {
+		return nil, err
+	}
 
 	if pipeline.Name == "" {
 		return nil, util.NewInvalidInputError("name is required")
@@ -151,6 +156,10 @@ func (s *BasePipelineServer) createPipelineAndPipelineVersion(ctx context.Contex
 	}
 
 	pipeline.Namespace = s.resourceManager.ReplaceNamespace(pipeline.Namespace)
+	err := validation.ValidateNamespaceRequired(pipeline.Namespace)
+	if err != nil {
+		return nil, nil, err
+	}
 
 	// Check authorization
 	resourceAttributes := &authorizationv1.ResourceAttributes{
@@ -328,6 +337,9 @@ func (s *PipelineServer) GetPipeline(ctx context.Context, request *apiv2beta1.Ge
 // Applies common logic on v1beta1 and v2beta1 API.
 func (s *BasePipelineServer) getPipelineByName(ctx context.Context, name string, namespace string, apiRequestVersion string) (*model.Pipeline, *model.PipelineVersion, error) {
 	namespace = s.resourceManager.ReplaceNamespace(namespace)
+	if err := validation.ValidateNamespaceRequired(namespace); err != nil {
+		return nil, nil, err
+	}
 	resourceAttributes := &authorizationv1.ResourceAttributes{
 		Namespace: namespace,
 		Name:      name,
@@ -389,6 +401,9 @@ func (s *PipelineServer) GetPipelineByName(ctx context.Context, request *apiv2be
 func (s *BasePipelineServer) listPipelines(ctx context.Context, namespace string, pageToken string, pageSize int32, sortBy string, opts *list.Options, apiRequestVersion string) ([]*model.Pipeline, []*model.PipelineVersion, int, string, error) {
 	// Fill in the default namespace
 	namespace = s.resourceManager.ReplaceNamespace(namespace)
+	if err := validation.ValidateNamespaceRequired(namespace); err != nil {
+		return nil, nil, 0, "", err
+	}
 	resourceAttributes := &authorizationv1.ResourceAttributes{
 		Namespace: namespace,
 		Verb:      common.RbacResourceVerbList,

backend/src/apiserver/server/pipeline_upload_server.go

@@ -115,6 +115,11 @@ func (s *PipelineUploadServer) uploadPipeline(api_version string, w http.Respons
 		return
 	}
 	pipelineNamespace = s.resourceManager.ReplaceNamespace(pipelineNamespace)
+	err = validation.ValidateNamespaceRequired(pipelineNamespace)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
 	resourceAttributes := &authorizationv1.ResourceAttributes{
 		Namespace: pipelineNamespace,
 		Verb:      common.RbacResourceVerbCreate,
@@ -255,6 +260,12 @@ func (s *PipelineUploadServer) uploadPipelineVersion(api_version string, w http.
 		return
 	}
 
+	err = validation.ValidateNamespaceRequired(namespace)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
 	resourceAttributes := &authorizationv1.ResourceAttributes{
 		Namespace: namespace,
 		Verb:      common.RbacResourceVerbCreate,

backend/src/apiserver/validation/namespace.go

@@ -0,0 +1,30 @@
+// Copyright 2025 The Kubeflow 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
+//
+// https://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 validation
+
+import (
+	"github.com/kubeflow/pipelines/backend/src/apiserver/common"
+	"github.com/kubeflow/pipelines/backend/src/common/util"
+)
+
+// ValidateNamespaceRequired validates that a namespace is provided when required by configuration.
+// This ensures consistent authorization behavior across all KFP resources.
+// Only validates in multi-user mode, since single-user mode always uses empty namespaces by design.
+func ValidateNamespaceRequired(namespace string) error {
+	if common.IsMultiUserMode() && common.IsNamespaceRequiredForPipelines() && namespace == "" {
+		return util.NewInvalidInputError("Namespace is required for pipeline operations")
+	}
+	return nil
+}