Merge branch 'develop' of github.com:devtron-labs/devtron into cdworkflow-runner-read

Author: ayu-devtron

cmd/external-app/wire_gen.go

@@ -139,6 +139,8 @@ type CiArtifactRepository interface {
 	// MigrateToWebHookDataSourceType is used for backward compatibility. It'll migrate the deprecated DataSource type
 	MigrateToWebHookDataSourceType(id int) error
 	UpdateLatestTimestamp(artifactIds []int) error
+
+	Update(ciArtifact *CiArtifact) error
 }
 
 type CiArtifactRepositoryImpl struct {
@@ -858,3 +860,12 @@ func (impl CiArtifactRepositoryImpl) FindCiArtifactByImagePaths(images []string)
 	}
 	return ciArtifacts, nil
 }
+
+func (impl CiArtifactRepositoryImpl) Update(ciArtifact *CiArtifact) error {
+	err := impl.dbConnection.Update(ciArtifact)
+	if err != nil {
+		impl.logger.Errorw("error in updating ciArtifact", "ciArtifact", ciArtifact, "err", err)
+		return err
+	}
+	return nil
+}

internal/sql/repository/CiArtifactRepository.go

@@ -82,6 +82,7 @@ type CiCompleteEvent struct {
 	ImageDetailsFromCR            json.RawMessage          `json:"imageDetailsFromCR"`
 	PluginRegistryArtifactDetails map[string][]string      `json:"PluginRegistryArtifactDetails"`
 	PluginArtifactStage           string                   `json:"pluginArtifactStage"`
+	IsScanEnabled                 bool                     `json:"isScanEnabled"`
 	pluginImageDetails            *registry.ImageDetailsFromCR
 	PluginArtifacts               *PluginArtifacts `json:"pluginArtifacts"`
 }

pkg/eventProcessor/bean/workflowEventBean.go

@@ -685,6 +685,7 @@ func (impl *WorkflowEventProcessorImpl) BuildCiArtifactRequest(event bean.CiComp
 		IsArtifactUploaded:            event.IsArtifactUploaded,
 		PluginRegistryArtifactDetails: pluginArtifacts,
 		PluginArtifactStage:           event.PluginArtifactStage,
+		IsScanEnabled:                 event.IsScanEnabled,
 	}
 	// if DataSource is empty, repository.WEBHOOK is considered as default
 	if request.DataSource == "" {

pkg/eventProcessor/in/WorkflowEventProcessorService.go

@@ -50,6 +50,7 @@ type PipelineStageService interface {
 	// , there was a bug(https://github.com/devtron-labs/devtron/issues/3826) where we were not deleting pipeline stage entry even after deleting all the pipelineStageSteps
 	// , this will delete those pipelineStage entry
 	DeletePipelineStageIfReq(stageReq *bean.PipelineStageDto, userId int32) (error, bool)
+	IsScanPluginConfiguredAtPipelineStage(pipelineId int, pipelineStage repository.PipelineStageType, pluginName string) (bool, error)
 }
 
 func NewPipelineStageService(logger *zap.SugaredLogger,
@@ -2168,3 +2169,20 @@ func (impl *PipelineStageServiceImpl) extractAndMapScopedVariables(stageReq *bea
 	return impl.scopedVariableManager.ExtractAndMapVariables(string(requestJson), stageReq.Id, repository3.EntityTypePipelineStage, userId, tx)
 
 }
+
+func (impl *PipelineStageServiceImpl) IsScanPluginConfiguredAtPipelineStage(pipelineId int, pipelineStage repository.PipelineStageType, pluginName string) (bool, error) {
+	plugin, err := impl.globalPluginRepository.GetPluginByName(pluginName)
+	if err != nil {
+		impl.logger.Errorw("error in getting image scanning plugin, Vulnerability Scanning", "pipelineId", pipelineId, "pipelineStage", pipelineStage, "err", err)
+		return false, err
+	}
+	if len(plugin) == 0 {
+		return false, nil
+	}
+	isScanPluginConfigured, err := impl.pipelineStageRepository.CheckIfPluginExistsInPipelineStage(pipelineId, pipelineStage, plugin[0].Id)
+	if err != nil {
+		impl.logger.Errorw("error in getting ci pipeline plugin", "err", err, "pipelineId", pipelineId, "pluginId", plugin[0].Id)
+		return false, err
+	}
+	return isScanPluginConfigured, nil
+}

pkg/pipeline/PipelineStageService.go

@@ -53,6 +53,22 @@ const (
 	PIPELINE_STAGE_STEP_VARIABLE_FORMAT_TYPE_DATE    PipelineStageStepVariableFormatType = "DATE"
 )
 
+func (r PipelineStageType) ToString() string {
+	return string(r)
+}
+func (r PipelineStageType) IsStageTypePreCi() bool {
+	return r == PIPELINE_STAGE_TYPE_PRE_CI
+}
+func (r PipelineStageType) IsStageTypePreCd() bool {
+	return r == PIPELINE_STAGE_TYPE_PRE_CD
+}
+func (r PipelineStageType) IsStageTypePostCi() bool {
+	return r == PIPELINE_STAGE_TYPE_POST_CI
+}
+func (r PipelineStageType) IsStageTypePostCd() bool {
+	return r == PIPELINE_STAGE_TYPE_POST_CD
+}
+
 type PipelineStage struct {
 	tableName    struct{}          `sql:"pipeline_stage" pg:",discard_unknown_columns"`
 	Id           int               `sql:"id,pk"`
@@ -184,6 +200,7 @@ type PipelineStageRepository interface {
 	MarkStepsDeletedByStageId(stageId int) error
 	MarkStepsDeletedExcludingActiveStepsInUpdateReq(activeStepIdsPresentInReq []int, stageId int) error
 	GetActiveStepsByRefPluginId(refPluginId int) ([]*PipelineStageStep, error)
+	CheckIfPluginExistsInPipelineStage(pipelineId int, stageType PipelineStageType, pluginId int) (bool, error)
 
 	CreatePipelineScript(pipelineScript *PluginPipelineScript, tx *pg.Tx) (*PluginPipelineScript, error)
 	UpdatePipelineScript(pipelineScript *PluginPipelineScript) (*PluginPipelineScript, error)
@@ -873,3 +890,26 @@ func (impl *PipelineStageRepositoryImpl) MarkConditionsDeletedExcludingActiveVar
 	}
 	return nil
 }
+
+func (impl *PipelineStageRepositoryImpl) CheckIfPluginExistsInPipelineStage(pipelineId int, stageType PipelineStageType, pluginId int) (bool, error) {
+	var step PipelineStageStep
+	query := impl.dbConnection.Model(&step).
+		Column("pipeline_stage_step.*").
+		Join("INNER JOIN pipeline_stage ps on ps.id = pipeline_stage_step.pipeline_stage_id").
+		Where("pipeline_stage_step.ref_plugin_id = ?", pluginId).
+		Where("ps.type = ?", stageType).
+		Where("pipeline_stage_step.deleted=?", false).
+		Where("ps.deleted= ?", false)
+
+	if stageType.IsStageTypePostCi() || stageType.IsStageTypePreCi() {
+		query.Where("ps.ci_pipeline_id= ?", pipelineId)
+	} else if stageType.IsStageTypePostCd() || stageType.IsStageTypePreCd() {
+		query.Where("ps.cd_pipeline_id= ?", pipelineId)
+	}
+	exists, err := query.Exists()
+	if err != nil {
+		impl.logger.Errorw("error in getting plugin stage step by pipelineId, stageType nad plugin id", "pipelineId", pipelineId, "stageType", stageType.ToString(), "pluginId", pluginId, "err", err)
+		return false, err
+	}
+	return exists, nil
+}

pkg/pipeline/repository/PipelineStageRepository.go

@@ -24,6 +24,7 @@ import (
 	bean3 "github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning/bean"
 	repository3 "github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning/repository"
 	securityBean "github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning/repository/bean"
+	serverBean "github.com/devtron-labs/devtron/pkg/server/bean"
 	"go.opentelemetry.io/otel"
 	"time"
 
@@ -43,6 +44,7 @@ type ImageScanService interface {
 	FetchMinScanResultByAppIdAndEnvId(request *bean3.ImageScanRequest) (*bean3.ImageScanExecutionDetail, error)
 	VulnerabilityExposure(request *repository3.VulnerabilityRequest) (*repository3.VulnerabilityExposureListingResponse, error)
 	GetArtifactVulnerabilityStatus(ctx context.Context, request *bean2.VulnerabilityCheckRequest) (bool, error)
+	IsImageScanExecutionCompleted(image, imageDigest string) (bool, error)
 }
 
 type ImageScanServiceImpl struct {
@@ -644,3 +646,19 @@ func (impl ImageScanServiceImpl) updateCount(severity securityBean.Severity, cri
 	}
 	return criticalCount, highCount, moderateCount, lowCount, unkownCount
 }
+
+func (impl ImageScanServiceImpl) IsImageScanExecutionCompleted(image, imageDigest string) (bool, error) {
+	var isScanningCompleted bool
+	allScanHistoryMappings, err := impl.scanToolExecutionHistoryMappingRepository.FetchScanHistoryMappingsUsingImageAndImageDigest(image, imageDigest)
+	if err != nil {
+		impl.Logger.Errorw("error in fetching all scan execution history mapping", "image", image, "imageDigest", imageDigest, "err", err)
+		return false, err
+	}
+
+	for _, scanHistoryMapping := range allScanHistoryMappings {
+		if scanHistoryMapping.State == serverBean.ScanExecutionProcessStateCompleted {
+			isScanningCompleted = true
+		}
+	}
+	return isScanningCompleted, nil
+}

pkg/policyGovernance/security/imageScanning/ImageScanService.go

@@ -45,6 +45,7 @@ type ScanToolExecutionHistoryMappingRepository interface {
 	GetAllScanHistoriesByState(state serverBean.ScanExecutionProcessState) ([]*ScanToolExecutionHistoryMapping, error)
 	GetAllScanHistoriesByExecutionHistoryIdAndStates(executionHistoryId int, states []serverBean.ScanExecutionProcessState) ([]*ScanToolExecutionHistoryMapping, error)
 	GetAllScanHistoriesByExecutionHistoryIds(ids []int) ([]*ScanToolExecutionHistoryMapping, error)
+	FetchScanHistoryMappingsUsingImageAndImageDigest(image, imageDigest string) ([]*ScanToolExecutionHistoryMapping, error)
 }
 
 type ScanToolExecutionHistoryMappingRepositoryImpl struct {
@@ -142,3 +143,18 @@ func (repo *ScanToolExecutionHistoryMappingRepositoryImpl) GetAllScanHistoriesBy
 	}
 	return models, nil
 }
+
+func (repo *ScanToolExecutionHistoryMappingRepositoryImpl) FetchScanHistoryMappingsUsingImageAndImageDigest(image, imageDigest string) ([]*ScanToolExecutionHistoryMapping, error) {
+	var models []*ScanToolExecutionHistoryMapping
+	err := repo.dbConnection.Model(&models).
+		Column("scan_tool_execution_history_mapping.*").
+		Join("INNER JOIN image_scan_execution_history iseh on iseh.id=scan_tool_execution_history_mapping.image_scan_execution_history_id").
+		Where("iseh.image = ?", image).
+		Where("iseh.image_hash = ?", imageDigest).
+		Select()
+	if err != nil {
+		repo.logger.Errorw("error in getting ScanToolExecutionHistoryMapping using image and image hash", "err", err)
+		return nil, err
+	}
+	return models, nil
+}

pkg/policyGovernance/security/imageScanning/repository/ScanToolExecutionHistoryMapping.go

@@ -44,12 +44,17 @@ import (
 	"github.com/devtron-labs/devtron/pkg/deployment/trigger/devtronApps/userDeploymentRequest/service"
 	eventProcessorBean "github.com/devtron-labs/devtron/pkg/eventProcessor/bean"
 	"github.com/devtron-labs/devtron/pkg/pipeline"
+	constants2 "github.com/devtron-labs/devtron/pkg/pipeline/constants"
 	"github.com/devtron-labs/devtron/pkg/pipeline/executors"
 	repository2 "github.com/devtron-labs/devtron/pkg/plugin/repository"
+	"github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning"
+	repository3 "github.com/devtron-labs/devtron/pkg/policyGovernance/security/imageScanning/repository"
 	"github.com/devtron-labs/devtron/pkg/sql"
 	"github.com/devtron-labs/devtron/pkg/workflow/cd"
 	bean4 "github.com/devtron-labs/devtron/pkg/workflow/cd/bean"
+	"github.com/devtron-labs/devtron/pkg/workflow/dag/adaptor"
 	bean2 "github.com/devtron-labs/devtron/pkg/workflow/dag/bean"
+	"github.com/devtron-labs/devtron/pkg/workflow/dag/helper"
 	error2 "github.com/devtron-labs/devtron/util/error"
 	util2 "github.com/devtron-labs/devtron/util/event"
 	"strings"
@@ -124,6 +129,8 @@ type WorkflowDagExecutorImpl struct {
 	commonArtifactService   artifacts.CommonArtifactService
 	deploymentConfigService common2.DeploymentConfigService
 	asyncRunnable           *async.Runnable
+	scanHistoryRepository   repository3.ImageScanHistoryRepository
+	imageScanService        imageScanning.ImageScanService
 }
 
 func NewWorkflowDagExecutorImpl(Logger *zap.SugaredLogger, pipelineRepository pipelineConfig.PipelineRepository,
@@ -147,7 +154,10 @@ func NewWorkflowDagExecutorImpl(Logger *zap.SugaredLogger, pipelineRepository pi
 	manifestCreationService manifest.ManifestCreationService,
 	commonArtifactService artifacts.CommonArtifactService,
 	deploymentConfigService common2.DeploymentConfigService,
-	asyncRunnable *async.Runnable) *WorkflowDagExecutorImpl {
+	asyncRunnable *async.Runnable,
+	scanHistoryRepository repository3.ImageScanHistoryRepository,
+	imageScanService imageScanning.ImageScanService,
+) *WorkflowDagExecutorImpl {
 	wde := &WorkflowDagExecutorImpl{logger: Logger,
 		pipelineRepository:            pipelineRepository,
 		cdWorkflowRepository:          cdWorkflowRepository,
@@ -171,6 +181,8 @@ func NewWorkflowDagExecutorImpl(Logger *zap.SugaredLogger, pipelineRepository pi
 		commonArtifactService:         commonArtifactService,
 		deploymentConfigService:       deploymentConfigService,
 		asyncRunnable:                 asyncRunnable,
+		scanHistoryRepository:         scanHistoryRepository,
+		imageScanService:              imageScanService,
 	}
 	config, err := types.GetCdConfig()
 	if err != nil {
@@ -558,6 +570,22 @@ func (impl *WorkflowDagExecutorImpl) HandlePreStageSuccessEvent(triggerContext t
 		if err != nil {
 			return err
 		}
+		scanEnabled, scanned := ciArtifact.ScanEnabled, ciArtifact.Scanned
+		isScanPluginConfigured, isScanningDoneViaPlugin, err := impl.isArtifactScannedByPluginForPipeline(ciArtifact, cdStageCompleteEvent.CdPipelineId, repository4.PIPELINE_STAGE_TYPE_PRE_CD, bean2.ImageScanningPluginToCheckInPipelineStageStep)
+		if err != nil {
+			impl.logger.Errorw("error in checking if artifact scanned by plugin for a pipeline or not", "ciArtifact", ciArtifact, "err", err)
+			return err
+		}
+		helper.UpdateScanStatusInCiArtifact(ciArtifact, isScanPluginConfigured, isScanningDoneViaPlugin)
+
+		// if ciArtifact scanEnabled and scanned state changed from above func then update ciArtifact
+		if scanEnabled != ciArtifact.ScanEnabled || scanned != ciArtifact.Scanned {
+			err = impl.ciArtifactRepository.Update(ciArtifact)
+			if err != nil {
+				impl.logger.Errorw("error in updating ci artifact after handling scan event for this artifact", "ciArtifact", ciArtifact, "err", err)
+				return err
+			}
+		}
 		// Migration of deprecated DataSource Type
 		if ciArtifact.IsMigrationRequired() {
 			migrationErr := impl.ciArtifactRepository.MigrateToWebHookDataSourceType(ciArtifact.Id)
@@ -651,6 +679,22 @@ func (impl *WorkflowDagExecutorImpl) HandlePostStageSuccessEvent(triggerContext
 		impl.logger.Errorw("error in finding artifact by cd workflow id", "err", err, "cdWorkflowId", cdWorkflowId)
 		return err
 	}
+	scanEnabled, scanned := ciArtifact.ScanEnabled, ciArtifact.Scanned
+	isScanPluginConfigured, isScanningDoneViaPlugin, err := impl.isArtifactScannedByPluginForPipeline(ciArtifact, cdPipelineId, repository4.PIPELINE_STAGE_TYPE_POST_CD, bean2.ImageScanningPluginToCheckInPipelineStageStep)
+	if err != nil {
+		impl.logger.Errorw("error in checking if artifact scanned by plugin for a pipeline or not", "ciArtifact", ciArtifact, "err", err)
+		return err
+	}
+	helper.UpdateScanStatusInCiArtifact(ciArtifact, isScanPluginConfigured, isScanningDoneViaPlugin)
+
+	// if ciArtifact scanEnabled and scanned state changed from above func then update ciArtifact
+	if scanEnabled != ciArtifact.ScanEnabled || scanned != ciArtifact.Scanned {
+		err = impl.ciArtifactRepository.Update(ciArtifact)
+		if err != nil {
+			impl.logger.Errorw("error in updating ci artifact after handling scan event for this artifact", "ciArtifact", ciArtifact, "err", err)
+			return err
+		}
+	}
 	if len(pluginRegistryImageDetails) > 0 {
 		PostCDArtifacts, err := impl.commonArtifactService.SavePluginArtifacts(ciArtifact, pluginRegistryImageDetails, cdPipelineId, repository.POST_CD, triggeredBy)
 		if err != nil {
@@ -710,6 +754,24 @@ func (impl *WorkflowDagExecutorImpl) UpdateCiWorkflowForCiSuccess(request *bean2
 	return nil
 }
 
+func (impl *WorkflowDagExecutorImpl) isArtifactScannedByPluginForPipeline(ciArtifact *repository.CiArtifact, pipelineId int,
+	pipelineStage repository4.PipelineStageType, pluginName string) (bool, bool, error) {
+	var isScanningDone bool
+	isScanPluginConfigured, err := impl.pipelineStageService.IsScanPluginConfiguredAtPipelineStage(pipelineId, pipelineStage, pluginName)
+	if err != nil {
+		impl.logger.Errorw("error in fetching if a scan plugin is configured or not in a pipeline", "pipelineStage", pipelineStage, "ciArtifact", ciArtifact)
+		return false, false, err
+	}
+	if isScanPluginConfigured {
+		isScanningDone, err = impl.imageScanService.IsImageScanExecutionCompleted(ciArtifact.Image, ciArtifact.ImageDigest)
+		if err != nil {
+			impl.logger.Errorw("error in checking if image scanning is completed or not", "image", ciArtifact.Image, "imageDigest", ciArtifact.ImageDigest)
+			return false, false, err
+		}
+	}
+	return isScanPluginConfigured, isScanningDone, nil
+}
+
 func (impl *WorkflowDagExecutorImpl) HandleCiSuccessEvent(triggerContext triggerBean.TriggerContext, ciPipelineId int, request *bean2.CiArtifactWebhookRequest, imagePushedAt time.Time) (id int, err error) {
 	impl.logger.Infow("webhook for artifact save", "req", request)
 	pipelineModal, err := impl.ciPipelineRepository.FindByCiAndAppDetailsById(ciPipelineId)
@@ -720,48 +782,36 @@ func (impl *WorkflowDagExecutorImpl) HandleCiSuccessEvent(triggerContext trigger
 	if request.PipelineName == "" {
 		request.PipelineName = pipelineModal.Name
 	}
-	materialJson, err := request.MaterialInfo.MarshalJSON()
-	if err != nil {
-		impl.logger.Errorw("unable to marshal material metadata", "err", err)
-		return 0, err
-	}
-	dst := new(bytes.Buffer)
-	err = json.Compact(dst, materialJson)
+	materialJson, err := helper.GetMaterialInfoJson(request.MaterialInfo)
 	if err != nil {
+		impl.logger.Errorw("unable to get materialJson", "materialInfo", request.MaterialInfo, "err", err)
 		return 0, err
 	}
-	materialJson = dst.Bytes()
 	createdOn := time.Now()
 	updatedOn := time.Now()
 	if !imagePushedAt.IsZero() {
 		createdOn = imagePushedAt
 	}
-	buildArtifact := &repository.CiArtifact{
-		Image:              request.Image,
-		ImageDigest:        request.ImageDigest,
-		MaterialInfo:       string(materialJson),
-		DataSource:         request.DataSource,
-		PipelineId:         pipelineModal.Id,
-		WorkflowId:         request.WorkflowId,
-		ScanEnabled:        pipelineModal.ScanEnabled,
-		IsArtifactUploaded: request.IsArtifactUploaded, // for backward compatibility
-		Scanned:            false,
-		AuditLog:           sql.AuditLog{CreatedBy: request.UserId, UpdatedBy: request.UserId, CreatedOn: createdOn, UpdatedOn: updatedOn},
-	}
-	plugin, err := impl.globalPluginRepository.GetPluginByName(bean3.VULNERABILITY_SCANNING_PLUGIN)
-	if err != nil || len(plugin) == 0 {
-		impl.logger.Errorw("error in getting image scanning plugin", "err", err)
-		return 0, err
-	}
-	isScanPluginConfigured, err := impl.pipelineStageRepository.CheckPluginExistsInCiPipeline(pipelineModal.Id, string(repository4.PIPELINE_STAGE_TYPE_POST_CI), plugin[0].Id)
-	if err != nil {
-		impl.logger.Errorw("error in getting ci pipelineModal plugin", "err", err, "pipelineId", pipelineModal.Id, "pluginId", plugin[0].Id)
-		return 0, err
+	buildArtifact := adaptor.GetBuildArtifact(request, pipelineModal.Id, materialJson, createdOn, updatedOn)
+
+	// image scanning plugin can only be applied in Post-ci, scanning in pre-ci doesn't make sense
+	pipelineStage := repository4.PIPELINE_STAGE_TYPE_POST_CI
+	if pipelineModal.PipelineType == constants2.CI_JOB.ToString() {
+		pipelineStage = repository4.PIPELINE_STAGE_TYPE_PRE_CI
 	}
-	if pipelineModal.ScanEnabled || isScanPluginConfigured {
+	// this flag comes from ci-runner when scanning is enabled from ciPipeline modal
+	if request.IsScanEnabled {
 		buildArtifact.Scanned = true
 		buildArtifact.ScanEnabled = true
+	} else {
+		isScanPluginConfigured, isScanningDoneViaPlugin, err := impl.isArtifactScannedByPluginForPipeline(buildArtifact, pipelineModal.Id, pipelineStage, bean2.ImageScanningPluginToCheckInPipelineStageStep)
+		if err != nil {
+			impl.logger.Errorw("error in checking if artifact scanned by plugin for a pipeline or not", "ciArtifact", buildArtifact, "err", err)
+			return 0, err
+		}
+		helper.UpdateScanStatusInCiArtifact(buildArtifact, isScanPluginConfigured, isScanningDoneViaPlugin)
 	}
+
 	if err = impl.ciArtifactRepository.Save(buildArtifact); err != nil {
 		impl.logger.Errorw("error in saving material", "err", err)
 		return 0, err
@@ -815,12 +865,12 @@ func (impl *WorkflowDagExecutorImpl) HandleCiSuccessEvent(triggerContext trigger
 			PipelineId:         ci.Id,
 			ParentCiArtifact:   buildArtifact.Id,
 			IsArtifactUploaded: request.IsArtifactUploaded, // for backward compatibility
-			ScanEnabled:        ci.ScanEnabled,
+			ScanEnabled:        buildArtifact.ScanEnabled,
 			Scanned:            false,
 			AuditLog:           sql.AuditLog{CreatedBy: request.UserId, UpdatedBy: request.UserId, CreatedOn: time.Now(), UpdatedOn: time.Now()},
 		}
-		if ci.ScanEnabled {
-			ciArtifact.Scanned = true
+		if buildArtifact.ScanEnabled {
+			ciArtifact.Scanned = buildArtifact.Scanned
 		}
 		ciArtifactArr = append(ciArtifactArr, ciArtifact)
 	}