feat(backend): Upload namespaced pipeline definitions. Part of #4197 (#…

…8511) * feat(backend) Fix authentication in upload requests Fix the way KFP API server authenticates pipeline upload requests. We leverage 'isAuthenticated()` function which requires proper initialization of the context object to include user identity. * feat(backend): Add namespace field in pipeline upload swagger definition Extend swagger defintion of the pipeline upload API with a namespace parameter in order to support uploading namespaced pipelines. * chore(backend): Generate Go & Python clients Autogenerate the Go and Python clients after extending the swagger definitions of upload pipeline APIs with a namespace field.
kubeflow · Dec 1, 2022 · 931c14a · 931c14a
1 parent 6cd7cbc
commit 931c14a
Show file tree

Hide file tree

Showing 5 changed files with 121 additions and 13 deletions.
diff --git a/..._http_client/pipeline_upload_client/pipeline_upload_service/upload_pipeline_parameters.go b/..._http_client/pipeline_upload_client/pipeline_upload_service/upload_pipeline_parameters.go
diff --git a/...ient/pipeline_upload_client/pipeline_upload_service/upload_pipeline_version_parameters.go b/...ient/pipeline_upload_client/pipeline_upload_service/upload_pipeline_version_parameters.go
diff --git a/backend/api/v1beta1/swagger/kfp_api_single_file.swagger.json b/backend/api/v1beta1/swagger/kfp_api_single_file.swagger.json
@@ -1406,6 +1406,12 @@
             "in": "query",
             "required": false,
             "type": "string"
+          },
+          {
+            "name": "namespace",
+            "in": "query",
+            "required": false,
+            "type": "string"
           }
         ],
         "tags": [
@@ -1461,6 +1467,12 @@
             "in": "query",
             "required": false,
             "type": "string"
+          },
+          {
+            "name": "namespace",
+            "in": "query",
+            "required": false,
+            "type": "string"
           }
         ],
         "tags": [

diff --git a/backend/api/v1beta1/swagger/pipeline.upload.swagger.json b/backend/api/v1beta1/swagger/pipeline.upload.swagger.json
@@ -51,6 +51,12 @@
               "in": "query",
               "required": false,
               "type": "string"
+            },
+            {
+              "name": "namespace",
+              "in": "query",
+              "required": false,
+              "type": "string"
             }
           ],
           "tags": [
@@ -106,6 +112,12 @@
               "in": "query",
               "required": false,
               "type": "string"
+            },
+            {
+              "name": "namespace",
+              "in": "query",
+              "required": false,
+              "type": "string"
             }
           ],
           "tags": [
@@ -298,4 +310,4 @@
         "Bearer": []
       }
     ]
-  }
+  }
diff --git a/backend/src/apiserver/server/pipeline_upload_server.go b/backend/src/apiserver/server/pipeline_upload_server.go
@@ -21,6 +21,7 @@ import (
 	"net/http"
 	"net/url"
 
+	"google.golang.org/grpc/metadata"
 	"github.com/golang/glog"
 	"github.com/golang/protobuf/jsonpb"
 	api "github.com/kubeflow/pipelines/backend/api/v1beta1/go_client"
@@ -99,7 +100,11 @@ func (s *PipelineUploadServer) UploadPipeline(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	err = s.canUploadVersionedPipeline(r, pipelineNamespace)
+	resourceAttributes := &authorizationv1.ResourceAttributes{
+		Namespace: pipelineNamespace,
+		Verb: common.RbacResourceVerbCreate,
+	}
+	err = s.canUploadVersionedPipeline(r, "", resourceAttributes)
 	if err != nil {
 		s.writeErrorToResponse(w, http.StatusBadRequest, util.Wrap(err, "Authorization to namespace failed."))
 		return
@@ -178,7 +183,11 @@ func (s *PipelineUploadServer) UploadPipelineVersion(w http.ResponseWriter, r *h
 		return
 	}
 
-	err = s.canUploadVersionedPipeline(r, namespace)
+	resourceAttributes := &authorizationv1.ResourceAttributes{
+		Namespace: namespace,
+		Verb:      common.RbacResourceVerbCreate,
+	}
+	err = s.canUploadVersionedPipeline(r, pipelineId, resourceAttributes)
 	if err != nil {
 		s.writeErrorToResponse(w, http.StatusBadRequest, util.Wrap(err, "Authorization to namespace failed."))
 		return
@@ -221,19 +230,36 @@ func (s *PipelineUploadServer) UploadPipelineVersion(w http.ResponseWriter, r *h
 	}
 }
 
-func (s *PipelineUploadServer) canUploadVersionedPipeline(r *http.Request, namespace string) error {
-	if namespace == "" {
+func (s *PipelineUploadServer) canUploadVersionedPipeline(r *http.Request, pipelineId string, resourceAttributes *authorizationv1.ResourceAttributes) error {
+	if !common.IsMultiUserMode() {
+		// Skip authorization if not multi-user mode.
 		return nil
 	}
-	userIdentityHeader := r.Header.Get(common.GetKubeflowUserIDHeader())
-	resourceAttributes := &authorizationv1.ResourceAttributes{
-		Namespace: namespace,
-		Verb:      common.RbacResourceVerbCreate,
-		Group:     common.RbacPipelinesGroup,
-		Version:   common.RbacPipelinesVersion,
-		Resource:  common.RbacResourceTypePipelines,
+	if len(pipelineId) > 0 {
+		namespace, err := s.resourceManager.GetNamespaceFromPipelineID(pipelineId)
+		if err != nil {
+			return util.Wrap(err, "Failed to authorize with the Pipeline ID.")
+		}
+		if len(resourceAttributes.Namespace) == 0 {
+		    resourceAttributes.Namespace = namespace
+		}
+	}
+	if resourceAttributes.Namespace == "" {
+		return nil
 	}
-	err := s.resourceManager.IsRequestAuthorized(context.TODO(), userIdentityHeader, resourceAttributes)
+
+	resourceAttributes.Group = common.RbacPipelinesGroup
+	resourceAttributes.Version = common.RbacPipelinesVersion
+	resourceAttributes.Resource = common.RbacResourceTypePipelines
+
+	ctx := context.Background()
+	md := metadata.MD{}
+	for key, values := range r.Header {
+		md.Set(key, values...)
+	}
+	ctx = metadata.NewIncomingContext(ctx, md)
+
+	err := isAuthorized(s.resourceManager, ctx, resourceAttributes)
 	if err != nil {
 		return util.Wrap(err, "Authorization Failure.")
 	}