Authorize create run requests

backend/src/apiserver/client/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -8,10 +8,13 @@ go_library(
         "scheduled_workflow.go",
         "sql.go",
         "workflow.go",
+        "kfam.go",
+        "kfam_fake.go",
     ],
     importpath = "github.com/kubeflow/pipelines/backend/src/apiserver/client",
     visibility = ["//visibility:public"],
     deps = [
+        "//backend/src/common/util:go_default_library",
         "//backend/src/crd/pkg/client/clientset/versioned:go_default_library",
         "//backend/src/crd/pkg/client/clientset/versioned/typed/scheduledworkflow/v1beta1:go_default_library",
         "@com_github_argoproj_argo//pkg/client/clientset/versioned:go_default_library",
@@ -26,3 +29,16 @@ go_library(
         "@io_k8s_client_go//rest:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = [
+        "kfam_test.go",
+    ],
+    data = glob(["test/**/*"]),  # keep
+    embed = [":go_default_library"],
+    deps = [
+        "//backend/src/common/util:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
+)

backend/src/apiserver/client/kfam.go

@@ -0,0 +1,100 @@
+// Copyright 2019 Google LLC
+//
+// 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 client
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/kubeflow/pipelines/backend/src/common/util"
+	"github.com/pkg/errors"
+)
+
+type KFAMClientInterface interface {
+	IsAuthorized(userIdentity string, namespace string) (bool, error)
+}
+
+type KFAMClient struct {
+	kfamServiceUrl string
+}
+
+type User struct {
+	Kind string
+	Name string
+}
+
+type RoleRef struct {
+	ApiGroup string
+	Kind     string
+	Name     string
+}
+
+type Binding struct {
+	User              User
+	ReferredNamespace string
+	RoleRef           RoleRef
+}
+
+type Bindings struct {
+	Bindings []Binding
+}
+
+const (
+	HTTP_TIMEOUT_SECONDS = 10
+)
+
+func (c *KFAMClient) IsAuthorized(userIdentity string, namespace string) (bool, error) {
+	req, err := http.NewRequest("GET", c.kfamServiceUrl, nil)
+	if err != nil {
+		return false, util.NewInternalServerError(err, "Failed to create a KFAM http request.")
+	}
+	q := req.URL.Query()
+	q.Add("user", userIdentity)
+	req.URL.RawQuery = q.Encode()
+
+	var httpClient = &http.Client{Timeout: HTTP_TIMEOUT_SECONDS * time.Second}
+
+	resp, err := httpClient.Get(req.URL.String())
+	if err != nil {
+		return false, util.NewInternalServerError(err, "Failed to connect to the KFAM service.")
+	}
+	if resp.StatusCode != http.StatusOK {
+		return false, util.NewInternalServerError(errors.New("Requests to the KFAM service failed."), resp.Status)
+	}
+	defer resp.Body.Close()
+
+	jsonBindings := new(Bindings)
+	err = json.NewDecoder(resp.Body).Decode(jsonBindings)
+
+	if err != nil {
+		return false, util.NewInternalServerError(err, "Failed to parse KFAM response.")
+	}
+
+	nsFound := false
+	for _, jsonBinding := range jsonBindings.Bindings {
+		if jsonBinding.ReferredNamespace == namespace {
+			nsFound = true
+			break
+		}
+	}
+	return nsFound, nil
+}
+
+func NewKFAMClient(kfamServiceHost string, kfamServicePort string) *KFAMClient {
+	kfamServiceUrl := fmt.Sprintf("http://%s:%s/kfam/v1/bindings", kfamServiceHost, kfamServicePort)
+	return &KFAMClient{kfamServiceUrl}
+}

backend/src/apiserver/client/kfam_fake.go

@@ -0,0 +1,37 @@
+// Copyright 2019 Google LLC
+//
+// 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 client
+
+type FakeKFAMClientAuthorized struct {
+}
+
+func NewFakeKFAMClientAuthorized() *FakeKFAMClientAuthorized {
+	return &FakeKFAMClientAuthorized{}
+}
+
+func (c *FakeKFAMClientAuthorized) IsAuthorized(userIdentity string, namespace string) (bool, error) {
+	return true, nil
+}
+
+type FakeKFAMClientUnauthorized struct {
+}
+
+func NewFakeKFAMClientUnauthorized() *FakeKFAMClientUnauthorized {
+	return &FakeKFAMClientUnauthorized{}
+}
+
+func (c *FakeKFAMClientUnauthorized) IsAuthorized(userIdentity string, namespace string) (bool, error) {
+	return false, nil
+}

backend/src/apiserver/client/kfam_test.go

@@ -0,0 +1,46 @@
+// Copyright 2019 Google LLC
+//
+// 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 client
+import(
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsAuthorize(t *testing.T) {
+	expect_response := []byte (`{"bindings":[{"user": {"kind": "User","name": "userA@google.com"},"referredNamespace": "nsA","RoleRef": {"apiGroup": "","kind": "ClusterRole", "name":"edit"}},{"user": {"kind": "User","name": "userA@google.com"},"referredNamespace": "nsB","RoleRef": {"apiGroup": "","kind": "ClusterRole", "name":"admin"}}]}`)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+		w.Write(expect_response)
+	}))
+	defer srv.Close()
+	fmt.Println(srv.URL)
+	kfam_client := NewKFAMClient("","")
+	kfam_client.kfamServiceUrl = srv.URL
+	authorized, err := kfam_client.IsAuthorized("user", "nsA")
+	assert.Nil(t, err)
+	assert.True(t, authorized)
+
+	authorized, err = kfam_client.IsAuthorized("user", "nsB")
+	assert.Nil(t, err)
+	assert.True(t, authorized)
+
+	authorized, err = kfam_client.IsAuthorized("user", "nsC")
+	assert.Nil(t, err)
+	assert.False(t, authorized)
+}

backend/src/apiserver/client_manager.go

@@ -45,6 +45,8 @@ const (
 	mysqlPassword          = "DBConfig.Password"
 	mysqlDBName            = "DBConfig.DBName"
 	mysqlGroupConcatMaxLen = "DBConfig.GroupConcatMaxLen"
+	kfamServiceHost        = "KFAM_SERVICE_HOST"
+	kfamServicePort        = "KFAM_SERVICE_PORT"
 
 	visualizationServiceHost = "ML_PIPELINE_VISUALIZATIONSERVER_SERVICE_HOST"
 	visualizationServicePort = "ML_PIPELINE_VISUALIZATIONSERVER_SERVICE_PORT"
@@ -67,6 +69,7 @@ type ClientManager struct {
 	wfClient               workflowclient.WorkflowInterface
 	swfClient              scheduledworkflowclient.ScheduledWorkflowInterface
 	podClient              v1.PodInterface
+	kfamClient             client.KFAMClientInterface
 	time                   util.TimeInterface
 	uuid                   util.UUIDGeneratorInterface
 }
@@ -115,6 +118,10 @@ func (c *ClientManager) PodClient() v1.PodInterface {
 	return c.podClient
 }
 
+func (c *ClientManager) KFAMClient() client.KFAMClientInterface {
+	return c.kfamClient
+}
+
 func (c *ClientManager) Time() util.TimeInterface {
 	return c.time
 }
@@ -154,6 +161,9 @@ func (c *ClientManager) init() {
 	runStore := storage.NewRunStore(db, c.time)
 	c.runStore = runStore
 
+	if common.IsMultiUserMode() {
+		c.kfamClient = client.NewKFAMClient(common.GetStringConfig(kfamServiceHost), common.GetStringConfig(kfamServicePort))
+	}
 	glog.Infof("Client manager initialized successfully")
 }
 

backend/src/apiserver/common/config.go

@@ -22,6 +22,10 @@ import (
 	"github.com/spf13/viper"
 )
 
+const (
+	MultiUserMode     string = "MULTIUSER"
+)
+
 func GetStringConfig(configName string) string {
 	if !viper.IsSet(configName) {
 		glog.Fatalf("Please specify flag %s", configName)
@@ -53,3 +57,7 @@ func GetDurationConfig(configName string) time.Duration {
 	}
 	return viper.GetDuration(configName)
 }
+
+func IsMultiUserMode() bool {
+	return GetBoolConfigWithDefault(MultiUserMode, false)
+}

backend/src/apiserver/common/const.go

@@ -36,6 +36,10 @@ const (
 	Creator Relationship = "Creator"
 )
 
+const (
+	GoogleIAPUserIdentityHeader	string = "x-goog-authenticated-user-email"
+)
+
 func ToModelResourceType(apiType api.ResourceType) (ResourceType, error) {
 	switch apiType {
 	case api.ResourceType_EXPERIMENT:

backend/src/apiserver/main.go

@@ -71,6 +71,17 @@ func main() {
 	clientManager.Close()
 }
 
+// A custom http request header matcher to pass on the user identity
+// Reference: https://github.com/grpc-ecosystem/grpc-gateway/blob/master/docs/_docs/customizingyourgateway.md#mapping-from-http-request-headers-to-grpc-client-metadata
+func grpcCustomMatcher(key string) (string, bool) {
+	switch strings.ToLower(key) {
+	case common.GoogleIAPUserIdentityHeader:
+		return strings.ToLower(key), true
+	default:
+		return strings.ToLower(key), false
+	}
+}
+
 func startRpcServer(resourceManager *resource.ResourceManager) {
 	glog.Info("Starting RPC server")
 	listener, err := net.Listen("tcp", *rpcPortFlag)
@@ -107,7 +118,7 @@ func startHttpProxy(resourceManager *resource.ResourceManager) {
 	defer cancel()
 
 	// Create gRPC HTTP MUX and register services.
-	mux := runtime.NewServeMux()
+	mux := runtime.NewServeMux(runtime.WithIncomingHeaderMatcher(grpcCustomMatcher))
 	registerHttpHandlerFromEndpoint(api.RegisterPipelineServiceHandlerFromEndpoint, "PipelineService", ctx, mux)
 	registerHttpHandlerFromEndpoint(api.RegisterExperimentServiceHandlerFromEndpoint, "ExperimentService", ctx, mux)
 	registerHttpHandlerFromEndpoint(api.RegisterJobServiceHandlerFromEndpoint, "JobService", ctx, mux)

backend/src/apiserver/resource/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
         "//backend/src/apiserver/list:go_default_library",
         "//backend/src/apiserver/model:go_default_library",
         "//backend/src/apiserver/storage:go_default_library",
+        "//backend/src/apiserver/client:go_default_library",
         "//backend/src/common/util:go_default_library",
         "//backend/src/crd/pkg/apis/scheduledworkflow/v1beta1:go_default_library",
         "//backend/src/crd/pkg/client/clientset/versioned/typed/scheduledworkflow/v1beta1:go_default_library",
@@ -52,6 +53,7 @@ go_test(
         "//backend/src/apiserver/common:go_default_library",
         "//backend/src/apiserver/model:go_default_library",
         "//backend/src/apiserver/storage:go_default_library",
+        "//backend/src/apiserver/client:go_default_library",
         "//backend/src/common/util:go_default_library",
         "//backend/src/crd/pkg/apis/scheduledworkflow/v1beta1:go_default_library",
         "@com_github_argoproj_argo//pkg/apis/workflow/v1alpha1:go_default_library",

backend/src/apiserver/resource/client_manager_fake.go

@@ -17,6 +17,7 @@ package resource
 import (
 	workflowclient "github.com/argoproj/argo/pkg/client/clientset/versioned/typed/workflow/v1alpha1"
 	"github.com/golang/glog"
+	"github.com/kubeflow/pipelines/backend/src/apiserver/client"
 	"github.com/kubeflow/pipelines/backend/src/apiserver/storage"
 	"github.com/kubeflow/pipelines/backend/src/common/util"
 	scheduledworkflowclient "github.com/kubeflow/pipelines/backend/src/crd/pkg/client/clientset/versioned/typed/scheduledworkflow/v1beta1"
@@ -41,6 +42,7 @@ type FakeClientManager struct {
 	workflowClientFake          *FakeWorkflowClient
 	scheduledWorkflowClientFake *FakeScheduledWorkflowClient
 	podClientFake               v1.PodInterface
+	KfamClientFake              client.KFAMClientInterface
 	time                        util.TimeInterface
 	uuid                        util.UUIDGeneratorInterface
 }
@@ -76,6 +78,7 @@ func NewFakeClientManager(time util.TimeInterface, uuid util.UUIDGeneratorInterf
 		objectStore:                 storage.NewFakeObjectStore(),
 		scheduledWorkflowClientFake: NewScheduledWorkflowClientFake(),
 		podClientFake:               FakePodClient{},
+		KfamClientFake:              client.NewFakeKFAMClientAuthorized(),
 		time:                        time,
 		uuid:                        uuid,
 	}, nil
@@ -146,6 +149,10 @@ func (f *FakeClientManager) PodClient() v1.PodInterface {
 	return f.podClientFake
 }
 
+func (f *FakeClientManager) KFAMClient() client.KFAMClientInterface {
+	return f.KfamClientFake
+}
+
 func (f *FakeClientManager) Close() error {
 	return f.db.Close()
 }

backend/src/apiserver/resource/model_converter.go

@@ -199,6 +199,7 @@ func (r *ResourceManager) toModelResourceReferences(
 		if err != nil {
 			return nil, util.Wrap(err, "Failed to find the referred resource")
 		}
+
 		//TODO(gaoning777) further investigation: Is the plain namespace a good option?  maybe uuid for distinctness even with namespace deletion/recreation.
 		modelRef := &model.ResourceReference{
 			ResourceUUID:  resourceId,