Merge pull request #6648 from devtron-labs/develop-main-sync-4jun-1

sync: Develop main sync

Author: kartik-579

App.go

@@ -56,6 +56,7 @@ type App struct {
 	server        *http.Server
 	db            *pg.DB
 	posthogClient *posthogTelemetry.PosthogClient
+	userService   user.UserService
 	// eventProcessor.CentralEventProcessor is used to register event processors
 	centralEventProcessor *eventProcessor.CentralEventProcessor // do not remove this.
 	// used for local dev only
@@ -79,6 +80,7 @@ func NewApp(router *router.MuxRouter,
 	pubSubClient *pubsub.PubSubClientServiceImpl,
 	workflowEventProcessorImpl *in.WorkflowEventProcessorImpl,
 	enforcerV2 *casbinv2.SyncedEnforcer,
+	userService user.UserService,
 ) *App {
 	//check argo connection
 	//todo - check argo-cd version on acd integration installation
@@ -97,6 +99,7 @@ func NewApp(router *router.MuxRouter,
 		centralEventProcessor:      centralEventProcessor,
 		pubSubClient:               pubSubClient,
 		workflowEventProcessorImpl: workflowEventProcessorImpl,
+		userService:                userService,
 	}
 	return app
 }
@@ -112,7 +115,7 @@ func (app *App) Start() {
 	app.MuxRouter.Init()
 	//authEnforcer := casbin2.Create()
 
-	server := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: authMiddleware.Authorizer(app.sessionManager2, user.WhitelistChecker, nil)(app.MuxRouter.Router)}
+	server := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: authMiddleware.Authorizer(app.sessionManager2, user.WhitelistChecker, app.userService.CheckUserStatusAndUpdateLoginAudit)(app.MuxRouter.Router)}
 	app.MuxRouter.Router.Use(app.loggingMiddleware.LoggingMiddleware)
 	app.MuxRouter.Router.Use(middleware.PrometheusMiddleware)
 	app.MuxRouter.Router.Use(middlewares.Recovery)

cmd/external-app/externalApp.go

@@ -42,21 +42,24 @@ type App struct {
 	server         *http.Server
 	telemetry      telemetry.TelemetryEventClient
 	posthogClient  *posthogTelemetry.PosthogClient
+	userService    user.UserService
 }
 
 func NewApp(db *pg.DB,
 	sessionManager *authMiddleware.SessionManager,
 	MuxRouter *MuxRouter,
 	telemetry telemetry.TelemetryEventClient,
 	posthogClient *posthogTelemetry.PosthogClient,
-	Logger *zap.SugaredLogger) *App {
+	Logger *zap.SugaredLogger,
+	userService user.UserService) *App {
 	return &App{
 		db:             db,
 		sessionManager: sessionManager,
 		MuxRouter:      MuxRouter,
 		Logger:         Logger,
 		telemetry:      telemetry,
 		posthogClient:  posthogClient,
+		userService:    userService,
 	}
 }
 func (app *App) Start() {
@@ -73,7 +76,7 @@ func (app *App) Start() {
 	if err != nil {
 		app.Logger.Warnw("telemetry installation success event failed", "err", err)
 	}
-	server := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: authMiddleware.Authorizer(app.sessionManager, user.WhitelistChecker, nil)(app.MuxRouter.Router)}
+	server := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: authMiddleware.Authorizer(app.sessionManager, user.WhitelistChecker, app.userService.CheckUserStatusAndUpdateLoginAudit)(app.MuxRouter.Router)}
 	app.MuxRouter.Router.Use(middleware.PrometheusMiddleware)
 	app.MuxRouter.Router.Use(middlewares.Recovery)
 	app.server = server

cmd/external-app/wire_gen.go

@@ -607,6 +607,10 @@ func (impl AppWorkflowServiceImpl) FindAppWorkflowsByEnvironmentId(request resou
 	}
 	appResults, envResults := request.CheckAuthBatch(token, appObjectArr, envObjectArr)
 	for _, pipeline := range pipelines {
+		if _, ok := objects[pipeline.Id]; !ok {
+			impl.Logger.Warnw("pipeline not found in objects map", "pipelineId", pipeline.Id)
+			continue
+		}
 		appObject := objects[pipeline.Id][0]
 		envObject := objects[pipeline.Id][1]
 		if !(appResults[appObject] && envResults[envObject]) {

pkg/appWorkflow/AppWorkflowService.go

@@ -126,20 +126,15 @@ func (impl *UserSelfRegistrationServiceImpl) CheckAndCreateUserIfConfigured(clai
 		emailId = sub
 	}
 	exists := impl.userService.UserExists(emailId)
-	var id int32
 	if !exists {
 		impl.logger.Infow("self registering user,  ", "email", emailId)
 		user, err := impl.SelfRegister(emailId)
 		if err != nil {
 			impl.logger.Errorw("error while register user", "error", err)
 		} else if user != nil && user.Id > 0 {
-			id = user.Id
 			exists = true
 		}
 	}
-	if exists {
-		impl.userService.SaveLoginAudit(emailId, "localhost", id)
-	}
 	impl.logger.Infow("user status", "email", emailId, "status", exists)
 	return exists
 }

pkg/auth/user/UserSelfRegistrationService.go

@@ -86,6 +86,7 @@ type UserService interface {
 	GetRoleFiltersByUserRoleGroups(userRoleGroups []userBean.UserRoleGroup) ([]userBean.RoleFilter, error)
 	SaveLoginAudit(emailId, clientIp string, id int32)
 	CheckIfTokenIsValid(email string, version string) error
+	CheckUserStatusAndUpdateLoginAudit(token string) (bool, int32, string, error)
 }
 
 type UserServiceImpl struct {
@@ -1826,3 +1827,34 @@ func (impl *UserServiceImpl) GetUserBasicDataByEmailId(emailId string) (*userBea
 	}
 	return response, nil
 }
+
+func (impl *UserServiceImpl) CheckUserStatusAndUpdateLoginAudit(token string) (bool, int32, string, error) {
+	emailId, version, err := impl.GetEmailAndVersionFromToken(token)
+	if err != nil {
+		impl.logger.Error("unable to fetch user by token")
+		err = &util.ApiError{HttpStatusCode: 401, UserMessage: "Invalid User", InternalMessage: "unable to fetch user by token"}
+		return false, 0, "", err
+	}
+	userId, isInactive, err := impl.getUserWithTimeoutWindowConfiguration(emailId)
+	if err != nil {
+		err = &util.ApiError{HttpStatusCode: 401, UserMessage: "Invalid User", InternalMessage: err.Error()}
+		impl.logger.Errorw("unable to fetch user by email, %s", token)
+		return isInactive, userId, "", err
+	}
+	//if user is inactive, no need to store audit log
+	if !isInactive {
+		impl.SaveLoginAudit(emailId, "localhost", userId)
+	}
+	// checking length of version, to ensure backward compatibility as earlier we did not
+	// have version for api-tokens
+	// therefore, for tokens without version we will skip the below part
+	if strings.HasPrefix(emailId, userBean.API_TOKEN_USER_EMAIL_PREFIX) && len(version) > 0 {
+		err := impl.CheckIfTokenIsValid(emailId, version)
+		if err != nil {
+			impl.logger.Errorw("token is not valid", "error", err, "token", token)
+			return isInactive, userId, "", err
+		}
+	}
+
+	return isInactive, userId, emailId, nil
+}

pkg/auth/user/UserService.go

@@ -8,6 +8,7 @@ import (
 	userBean "github.com/devtron-labs/devtron/pkg/auth/user/bean"
 	userrepo "github.com/devtron-labs/devtron/pkg/auth/user/repository"
 	"github.com/go-pg/pg"
+	"github.com/juju/errors"
 	"strings"
 )
 
@@ -204,3 +205,14 @@ func (impl *UserServiceImpl) checkValidationAndPerformOperationsForUpdate(token
 	}
 	return false, isUserSuperAdmin, nil
 }
+
+func (impl *UserServiceImpl) getUserWithTimeoutWindowConfiguration(emailId string) (int32, bool, error) {
+	user, err := impl.userRepository.FetchActiveUserByEmail(emailId)
+	if err != nil {
+		impl.logger.Errorw("error while fetching user from db", "error", err)
+		errMsg := fmt.Sprintf("failed to fetch user by email id, err: %s", err.Error())
+		return 0, false, errors.New(errMsg)
+	}
+	// here false is always returned to match signature of authoriser function.
+	return user.Id, false, nil
+}

pkg/auth/user/UserService_ent.go

@@ -1336,6 +1336,13 @@ func (impl BulkUpdateServiceImpl) BulkDeploy(request *bean4.BulkApplicationForEn
 			pResponse[pipelineKey] = false
 			response[appKey] = pResponse
 		}
+		if _, ok := objects[pipeline.Id]; !ok {
+			//if user unauthorized, skip items
+			pipelineResponse := response[appKey]
+			pipelineResponse[pipelineKey] = false
+			response[appKey] = pipelineResponse
+			continue
+		}
 		appObject := objects[pipeline.Id][0]
 		envObject := objects[pipeline.Id][1]
 		if !(appResults[appObject] && envResults[envObject]) {

pkg/bulkAction/service/BulkUpdateService.go

@@ -771,7 +771,8 @@ func (impl *ClusterServiceImpl) updateConnectionStatusForVirtualCluster(respMap
 func (impl *ClusterServiceImpl) ConnectClustersInBatch(clusters []*bean.ClusterBean, clusterExistInDb bool) {
 	var wg sync.WaitGroup
 	respMap := &sync.Map{}
-	for idx, cluster := range clusters {
+	for idx := range clusters {
+		cluster := clusters[idx]
 		if cluster.IsVirtualCluster {
 			impl.updateConnectionStatusForVirtualCluster(respMap, cluster.Id, cluster.ClusterName)
 			continue

pkg/cluster/ClusterService.go

@@ -404,17 +404,18 @@ func (impl *K8sCommonServiceImpl) getManifestsByBatch(ctx context.Context, reque
 		var wg sync.WaitGroup
 		for j := 0; j < batchSize; j++ {
 			wg.Add(1)
-			runnableFunc := func(j int) {
+			runnableFunc := func(index int) {
 				resp := bean5.BatchResourceResponse{}
-				response, err := impl.GetResource(ctx, &requests[i+j])
+				response, err := impl.GetResource(ctx, &requests[index])
 				if response != nil {
 					resp.ManifestResponse = response.ManifestResponse
 				}
 				resp.Err = err
-				res[i+j] = resp
+				res[index] = resp
 				wg.Done()
 			}
-			impl.asyncRunnable.Execute(func() { runnableFunc(j) })
+			index := i + j
+			impl.asyncRunnable.Execute(func() { runnableFunc(index) })
 		}
 		wg.Wait()
 		i += batchSize