From ad92e5e72272e15363001cd1e5d5fb0c3c854111 Mon Sep 17 00:00:00 2001
From: Hari Krishna <hari.100.hks@outlook.com>
Date: Fri, 21 Jun 2024 10:28:49 +0000
Subject: [PATCH] made the pipeline to parse json logs

---
 backend/go-app/main.go              | 51 ++++++++++-------------------
 functions/onprem/orborus/orborus.go | 15 ++++++---
 2 files changed, 27 insertions(+), 39 deletions(-)

diff --git a/backend/go-app/main.go b/backend/go-app/main.go
index 713abda8..d5fd7c6b 100755
--- a/backend/go-app/main.go
+++ b/backend/go-app/main.go
@@ -2065,8 +2065,8 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 		resp.Write([]byte(`{"success": false}`))
 		return
 	}
-// fix this
-	parsedBody, err := string(jsonList)
+
+	parsedBody, err := json.Marshal(jsonList)
 	if err != nil {
 		log.Printf("[ERROR] Failed to marshal jsonList: %s", err)
 		resp.WriteHeader(500)
@@ -2077,7 +2077,7 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 	newBody := shuffle.ExecutionStruct{
 		Start:             pipeline.StartNode,
 		ExecutionSource:   "pipeline",
-		ExecutionArgument: parsedBody,
+		ExecutionArgument: string(parsedBody),
 	}
 
 	workflow, err := shuffle.GetWorkflow(ctx, pipeline.WorkflowId)
@@ -2130,42 +2130,25 @@ func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
 }
 
 func parseConcatenatedJSONLogs(logs string) ([]map[string]interface{}, error) {
-	var jsonList []map[string]interface{}
-	var currentObject []rune
-	var depth int
+    var jsonList []map[string]interface{}
+    decoder := json.NewDecoder(strings.NewReader(logs))
 
-	for _, char := range logs {
-		if char == '{' {
-			depth++
-		}
-		if char == '}' {
-			depth--
-		}
+    for decoder.More() {
+        var jsonObject map[string]interface{}
+        if err := decoder.Decode(&jsonObject); err != nil {
+            log.Printf("[WARNING] JSON decoding error: %s. Skipping this object.", err)
+            continue
+        }
+        jsonList = append(jsonList, jsonObject)
+    }
 
-		currentObject = append(currentObject, char)
+    if err := decoder.Decode(&struct{}{}); err != io.EOF {
+        return nil, fmt.Errorf("error after decoding all JSON objects: %v", err)
+    }
 
-		// When depth is 0, it means we have a complete JSON object but will this work ??
-		if depth == 0 && len(currentObject) > 0 {
-			var jsonObject map[string]interface{}
-			err := json.Unmarshal([]byte(string(currentObject)), &jsonObject)
-			if err != nil {
-				log.Printf("[WARNING] JSON unmarshal error: %s. Skipping this object.", err)
-			} else {
-				jsonList = append(jsonList, jsonObject)
-			}
-			currentObject = nil
-		}
-	}
-
-	currentObject = []rune(strings.TrimSpace(string(currentObject)))
-	if len(currentObject) > 0 {
-		log.Printf("[WARNING] Incomplete JSON object found: %s. Skipping this object.", string(currentObject))
-	}
-
-	return jsonList, nil
+    return jsonList, nil
 }
 
-
 func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error {
 	data, err := json.Marshal(action)
 	if err != nil {
diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go
index 9c84c03d..b2ec0e80 100755
--- a/functions/onprem/orborus/orborus.go
+++ b/functions/onprem/orborus/orborus.go
@@ -2462,7 +2462,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
-		_, err = updatePipelineState(pipelineId, "stop")
+		_, err = updatePipelineState(command, pipelineId, "stop")
 		if err != nil {
 			log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
 			return err
@@ -2482,7 +2482,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
-		_, err = updatePipelineState(pipelineId, "start")
+		_, err = updatePipelineState(command, pipelineId, "start")
 		if err != nil {
 			log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err)
 			return err
@@ -2689,7 +2689,11 @@ func createPipeline(command, identifier string) (string, error) {
 	// 	}
 	// }
 
-	command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | sigma /var/lib/tenzir/rule.yaml | to https://shuffler.io/api/v1/hooks/webhook_d295c43a-e322-4afc-9a59-af167ae7c190"
+	//command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | sigma /var/lib/tenzir/rule.yaml"
+	//command = "from file /var/lib/tenzir/sysmon_logs.ndjson read json | import"
+	//command = "export | to https://expert-acorn-v6vg4j4j5w7q2wg6g-5001.app.github.dev/api/v1/hooks/webhook_623eab3f-0af4-4d40-abb9-699d9a493411"
+   log.Printf("[HARI] this is the command %s", command)
+
 	requestBody := map[string]interface{}{
 		"definition": command,
 		"name":       identifier,
@@ -2697,7 +2701,7 @@ func createPipeline(command, identifier string) (string, error) {
 		"autostart": map[string]bool{
 			"created":   true,
 			"completed": false,
-			"failed":    true,
+			"failed":    false,
 		},
 		"autodelete": map[string]bool{
 			"completed": false,
@@ -2763,13 +2767,14 @@ func createPipeline(command, identifier string) (string, error) {
 	return id, nil
 }
 
-func updatePipelineState(pipelineId, action string) (string, error) {
+func updatePipelineState(command, pipelineId, action string) (string, error) {
 
 	url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl)
 	forwardMethod := "POST"
 
 	requestBody := map[string]interface{}{
 		"id":     pipelineId,
+		"definition": command,
 		"action": action,
 		"autostart": map[string]bool{
 			"created":   true,