chore(integration): whitelist certain user emails for certain integra…

…tions (#845)

Because

- Certain OAuth integrations like Google Drive pass a review process. We
need to test the scopes internally before making the app public.

This commit

- Adds a way to hide the OAuth integration for certain components and
emails.

Makefile

@@ -13,7 +13,7 @@ GOTEST_FLAGS := CFG_DATABASE_HOST=${TEST_DBHOST} CFG_DATABASE_NAME=${TEST_DBNAME
 .PHONY: dev
 dev:							## Run dev container
 	@docker compose ls -q | grep -q "instill-core" && true || \
-		(echo "Error: Run \"make latest PROFILE=pipeline\" in vdp repository (https://github.com/instill-ai/instill-core) in your local machine first." && exit 1)
+		(echo "Error: Run \"make latest PROFILE=exclude-pipeline\" in vdp repository (https://github.com/instill-ai/instill-core) in your local machine first." && exit 1)
 	@docker inspect --type container ${SERVICE_NAME} >/dev/null 2>&1 && echo "A container named ${SERVICE_NAME} is already running." || \
 		echo "Run dev container ${SERVICE_NAME}. To stop it, run \"make stop\"."
 	@docker run -d --rm \
@@ -24,11 +24,10 @@ dev:							## Run dev container
 		--network instill-network \
 		--name ${SERVICE_NAME} \
 		instill/${SERVICE_NAME}:dev >/dev/null 2>&1
-
 .PHONY: latest
 latest:							## Run latest container
 	@docker compose ls -q | grep -q "instill-core" && true || \
-		(echo "Error: Run \"make latest PROFILE=pipeline\" in vdp repository (https://github.com/instill-ai/instill-core) in your local machine first." && exit 1)
+		(echo "Error: Run \"make latest PROFILE=exclude-pipeline\" in vdp repository (https://github.com/instill-ai/instill-core) in your local machine first." && exit 1)
 	@docker inspect --type container ${SERVICE_NAME} >/dev/null 2>&1 && echo "A container named ${SERVICE_NAME} is already running." || \
 		echo "Run latest container ${SERVICE_NAME} and ${SERVICE_NAME}-worker. To stop it, run \"make stop\"."
 	@docker run --network=instill-network \

cmd/main/main.go

@@ -242,6 +242,10 @@ func main() {
 	if pipelinePublicServiceClientConn != nil {
 		defer pipelinePublicServiceClientConn.Close()
 	}
+	mgmtPublicServiceClient, mgmtPublicServiceClientConn := external.InitMgmtPublicServiceClient(ctx)
+	if mgmtPublicServiceClientConn != nil {
+		defer mgmtPublicServiceClientConn.Close()
+	}
 	mgmtPrivateServiceClient, mgmtPrivateServiceClientConn := external.InitMgmtPrivateServiceClient(ctx)
 	if mgmtPrivateServiceClientConn != nil {
 		defer mgmtPrivateServiceClientConn.Close()
@@ -289,6 +293,7 @@ func main() {
 				InstillCoreHost: config.Config.Server.InstillCoreHost,
 				ComponentStore:  compStore,
 			}),
+			MgmtPublicServiceClient:  mgmtPublicServiceClient,
 			MgmtPrivateServiceClient: mgmtPrivateServiceClient,
 			MinioClient:              minioClient,
 			ComponentStore:           compStore,

config/config.go

@@ -94,7 +94,9 @@ type ServerConfig struct {
 // default API key when no setup is specified, or to connect with a 3rd party
 // vendor via OAuth.
 type ComponentConfig struct {
-	Secrets ComponentSecrets
+	Secrets                     ComponentSecrets
+	InternalUserEmails          []string
+	ComponentsWithInternalUsers []string
 }
 
 // ComponentSecrets contains the global config secrets of each

pkg/external/external.go

@@ -25,6 +25,30 @@ import (
 	usagepb "github.com/instill-ai/protogen-go/core/usage/v1beta"
 )
 
+// InitMgmtPublicServiceClient initialises a MgmtPublicServiceClient instance
+func InitMgmtPublicServiceClient(ctx context.Context) (mgmtpb.MgmtPublicServiceClient, *grpc.ClientConn) {
+	logger, _ := logger.GetZapLogger(ctx)
+
+	var clientDialOpts grpc.DialOption
+	if config.Config.MgmtBackend.HTTPS.Cert != "" && config.Config.MgmtBackend.HTTPS.Key != "" {
+		creds, err := credentials.NewServerTLSFromFile(config.Config.MgmtBackend.HTTPS.Cert, config.Config.MgmtBackend.HTTPS.Key)
+		if err != nil {
+			logger.Fatal(err.Error())
+		}
+		clientDialOpts = grpc.WithTransportCredentials(creds)
+	} else {
+		clientDialOpts = grpc.WithTransportCredentials(insecure.NewCredentials())
+	}
+
+	clientConn, err := grpc.NewClient(fmt.Sprintf("%v:%v", config.Config.MgmtBackend.Host, config.Config.MgmtBackend.PublicPort), clientDialOpts, grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(constant.MaxPayloadSize), grpc.MaxCallSendMsgSize(constant.MaxPayloadSize)))
+	if err != nil {
+		logger.Error(err.Error())
+		return nil, nil
+	}
+
+	return mgmtpb.NewMgmtPublicServiceClient(clientConn), clientConn
+}
+
 // InitMgmtPrivateServiceClient initialises a MgmtPrivateServiceClient instance
 func InitMgmtPrivateServiceClient(ctx context.Context) (mgmtpb.MgmtPrivateServiceClient, *grpc.ClientConn) {
 	logger, _ := logger.GetZapLogger(ctx)

pkg/service/integration.go

@@ -6,12 +6,14 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
 
 	"github.com/gofrs/uuid"
 	"github.com/iancoleman/strcase"
 	"github.com/santhosh-tekuri/jsonschema/v5"
 	"go.einride.tech/aip/filtering"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/fieldmaskpb"
 	"google.golang.org/protobuf/types/known/structpb"
@@ -20,14 +22,18 @@ import (
 
 	fieldmaskutil "github.com/mennanov/fieldmask-utils"
 
+	"github.com/instill-ai/pipeline-backend/config"
+	"github.com/instill-ai/pipeline-backend/pkg/constant"
 	"github.com/instill-ai/pipeline-backend/pkg/datamodel"
 	"github.com/instill-ai/pipeline-backend/pkg/recipe"
 	"github.com/instill-ai/pipeline-backend/pkg/repository"
+	"github.com/instill-ai/pipeline-backend/pkg/resource"
 	"github.com/instill-ai/x/checkfield"
 	"github.com/instill-ai/x/errmsg"
 
 	componentbase "github.com/instill-ai/pipeline-backend/pkg/component/base"
 	errdomain "github.com/instill-ai/pipeline-backend/pkg/errors"
+	mgmtpb "github.com/instill-ai/protogen-go/core/mgmt/v1beta"
 	pb "github.com/instill-ai/protogen-go/vdp/pipeline/v1beta"
 )
 
@@ -43,7 +49,7 @@ func (s *service) GetIntegration(ctx context.Context, id string, view pb.View) (
 		return nil, fmt.Errorf("fetching component information: %w", err)
 	}
 
-	integration, err := s.componentDefinitionToIntegration(cd, view)
+	integration, err := s.componentDefinitionToIntegration(ctx, cd, view)
 	if err != nil {
 		if errors.Is(err, errIntegrationConversion) {
 			return nil, errIntegrationNotFound
@@ -93,7 +99,7 @@ func (s *service) ListIntegrations(ctx context.Context, req *pb.ListIntegrations
 			return nil, fmt.Errorf("fetching component definition: %w", err)
 		}
 
-		integrations[i], err = s.componentDefinitionToIntegration(cd, pb.View_VIEW_BASIC)
+		integrations[i], err = s.componentDefinitionToIntegration(ctx, cd, pb.View_VIEW_BASIC)
 		if err != nil {
 			return nil, fmt.Errorf("converting component definition: %w", err)
 		}
@@ -109,6 +115,7 @@ func (s *service) ListIntegrations(ctx context.Context, req *pb.ListIntegrations
 var errIntegrationConversion = fmt.Errorf("component definition has no integration configuration")
 
 func (s *service) componentDefinitionToIntegration(
+	ctx context.Context,
 	cd *pb.ComponentDefinition,
 	view pb.View,
 ) (*pb.Integration, error) {
@@ -157,9 +164,26 @@ func (s *service) componentDefinitionToIntegration(
 	if err != nil {
 		return nil, fmt.Errorf("checking OAuth support: %w", err)
 	}
+	if !supportsOAuth {
+		return integration, nil
+	}
+
+	// This check verifies that the user can see the OAuth details. This is a
+	// mechanism to publish the OAuth feature for internal users in cases
+	// where, e.g., the OAuth flow of a vendor requires a review before being
+	// made public.
+	// TODO jvallesm: in the future we should use feature flags instead of this
+	// kind of condition.
+	supportsOAuth, err = s.isOAuthVisible(ctx, integration.Id)
+	if err != nil {
+		return nil, fmt.Errorf("checking OAuth visibility: %w", err)
+	}
+	if !supportsOAuth {
+		return integration, nil
+	}
 
 	oAuthConfig, hasOAuthConfig := schemaFields["instillOAuthConfig"]
-	if !(supportsOAuth && hasOAuthConfig) {
+	if !hasOAuthConfig {
 		return integration, nil
 	}
 
@@ -183,6 +207,39 @@ func (s *service) componentDefinitionToIntegration(
 
 }
 
+func (s *service) isOAuthVisible(ctx context.Context, integrationID string) (bool, error) {
+	integrationsWithInternalUsers := config.Config.Component.ComponentsWithInternalUsers
+	if !slices.Contains(integrationsWithInternalUsers, integrationID) {
+		return true, nil
+	}
+
+	authUserUID := resource.GetRequestSingleHeader(ctx, constant.HeaderUserUIDKey)
+	if authUserUID == "" {
+		return false, nil
+	}
+
+	// TODO: instead of using the public API, we should have a private endpoint
+	// that returns the whole user info. The private API only exposes
+	// `mgmtpb.User`, which we can't extend with the email because it's an
+	// entity used by other public endpoints and we'd break the user's privacy.
+	md := metadata.AppendToOutgoingContext(ctx,
+		constant.HeaderAuthTypeKey, "user",
+		constant.HeaderUserUIDKey, authUserUID,
+	)
+
+	resp, err := s.mgmtPublicServiceClient.GetAuthenticatedUser(md, new(mgmtpb.GetAuthenticatedUserRequest))
+	if err != nil {
+		return false, fmt.Errorf("fetching user info: %w", err)
+	}
+
+	whitelistedEmails := config.Config.Component.InternalUserEmails
+	if !slices.Contains(whitelistedEmails, resp.GetUser().GetEmail()) {
+		return false, nil
+	}
+
+	return true, nil
+}
+
 var outputOnlyConnectionFields = []string{
 	"uid",
 	"namespace_id",

pkg/service/main.go

@@ -103,6 +103,7 @@ type service struct {
 	redisClient              *redis.Client
 	temporalClient           client.Client
 	component                *componentstore.Store
+	mgmtPublicServiceClient  mgmtpb.MgmtPublicServiceClient
 	mgmtPrivateServiceClient mgmtpb.MgmtPrivateServiceClient
 	aclClient                acl.ACLClientInterface
 	converter                Converter
@@ -121,6 +122,7 @@ type ServiceConfig struct {
 	TemporalClient           client.Client
 	ACLClient                acl.ACLClientInterface
 	Converter                Converter
+	MgmtPublicServiceClient  mgmtpb.MgmtPublicServiceClient
 	MgmtPrivateServiceClient mgmtpb.MgmtPrivateServiceClient
 	MinioClient              miniox.MinioI
 	ComponentStore           *componentstore.Store
@@ -143,6 +145,7 @@ func NewService(
 		repository:               cfg.Repository,
 		redisClient:              cfg.RedisClient,
 		temporalClient:           cfg.TemporalClient,
+		mgmtPublicServiceClient:  cfg.MgmtPublicServiceClient,
 		mgmtPrivateServiceClient: cfg.MgmtPrivateServiceClient,
 		component:                cfg.ComponentStore,
 		aclClient:                cfg.ACLClient,