Update SO sigma pipeline

server/modules/elastalert/elastalert.go

@@ -635,19 +635,6 @@ func (e *ElastAlertEngine) Sync(logger *log.Entry, forceSync bool) error {
 	// announce the beginning of the sync
 	e.EngineState.Syncing = true
 
-	// Check to see if the sigma processing pipelines have changed.
-	// If they have, set forceSync to true to regenerate the elastalert rule files.
-	regenNeeded, sigmaPipelineNewHash, err := e.checkSigmaPipelines()
-	if err != nil {
-		logger.WithField("sigmaPipelineError", err).Error("failed to check the sigma processing pipelines")
-	} else {
-		logger.Info("successfully checked the sigma processing pipelines")
-	}
-
-	if regenNeeded {
-		forceSync = true
-	}
-
 	var zips map[string][]byte
 	var errMap map[string]error
 
@@ -697,6 +684,25 @@ func (e *ElastAlertEngine) Sync(logger *log.Entry, forceSync bool) error {
 		zipHashes[pkg] = base64.StdEncoding.EncodeToString(h[:])
 	}
 
+	// Check to see if the SO Sigma Processing Pipeline needs to be updated
+	pipelineUpdated, err := e.updateSigmaPipeline()
+	if err != nil {
+		logger.WithFields(log.Fields{"sigmaPipelineUpdateError": err.Error()}).Error("failed to update SO sigma processing pipeline")
+	}
+
+	// Check to see if the sigma processing pipelines have changed.
+	// If they have, set forceSync to true to regenerate the elastalert rule files.
+	regenNeeded, sigmaPipelineNewHash, err := e.checkSigmaPipelines()
+	if err != nil {
+		logger.WithField("sigmaPipelineError", err).Error("failed to check the sigma processing pipelines")
+	} else {
+		logger.Info("successfully checked the sigma processing pipelines")
+	}
+
+	if pipelineUpdated || regenNeeded {
+		forceSync = true
+	}
+
 	if !forceSync {
 		// if we're not forcing a sync, check to see if anything has changed
 		// if nothing has changed, the sync is finished
@@ -2107,3 +2113,44 @@ func (e *ElastAlertEngine) getDeployedPublicIds() (publicIds []string, err error
 
 	return publicIds, nil
 }
+
+func (e *ElastAlertEngine) updateSigmaPipeline() (bool, error) {
+	sourcePath := filepath.Join(e.reposFolder, "securityonion-resources", "so_sigma_processing_pipeline.yaml")
+
+	// Check if source exists
+	if _, err := os.Stat(sourcePath); os.IsNotExist(err) {
+		return false, fmt.Errorf("source pipeline file does not exist: %s", sourcePath)
+	}
+
+	// Get hash of source file
+	sourceHash, err := e.hashFile(sourcePath)
+	if err != nil {
+		return false, fmt.Errorf("failed to hash source pipeline file: %v", err)
+	}
+
+	// If destination exists, compare hashes
+	if _, err := os.Stat(e.sigmaPipelineSO); err == nil {
+		destHash, err := e.hashFile(e.sigmaPipelineSO)
+		if err != nil {
+			return false, fmt.Errorf("failed to hash destination pipeline file: %v", err)
+		}
+
+		// If hashes match, no update needed
+		if sourceHash == destHash {
+			return false, nil
+		}
+	}
+
+	// Copy the file
+	sourceData, err := os.ReadFile(sourcePath)
+	if err != nil {
+		return false, fmt.Errorf("failed to read source pipeline file: %v", err)
+	}
+
+	err = os.WriteFile(e.sigmaPipelineSO, sourceData, 0644)
+	if err != nil {
+		return false, fmt.Errorf("failed to write destination pipeline file: %v", err)
+	}
+
+	return true, nil
+}