Merge pull request #1398 from satti-hari-krishna-reddy/comTenz

A simple Modal for pipeline trigger and made improvements in tenzir deployement

.env

@@ -101,7 +101,7 @@ SHUFFLE_OPENSEARCH_INDEX_PREFIX=
 SHUFFLE_OPENSEARCH_SKIPSSL_VERIFY=true
 
 #Tenzir related 
-SHUFFLE_TENZIR_URL=http://localhost:5160
+SHUFFLE_TENZIR_URL=
 
 
 DEBUG_MODE=false

backend/go-app/main.go

@@ -1975,6 +1975,144 @@ func handleWebhookCallback(resp http.ResponseWriter, request *http.Request) {
 
 }
 
+func handlePipelineCallback(resp http.ResponseWriter, request *http.Request) {
+
+	if request.Method != "POST" {
+		request.Method = "POST"
+	}
+
+	if request.Body == nil {
+		stringReader := strings.NewReader("")
+		request.Body = ioutil.NopCloser(stringReader)
+	}
+
+	path := strings.Split(request.URL.String(), "/")
+	if len(path) < 4 {
+		resp.WriteHeader(403)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	ctx := context.Background()
+	location := strings.Split(request.URL.String(), "/")
+
+	var pipelineId string
+	
+	if location[1] == "api" {
+		if len(location) <= 4 {
+			log.Printf("[INFO] Couldn't handle location. Too short in pipeline: %d", len(location))
+			resp.WriteHeader(401)
+			resp.Write([]byte(`{"success": false}`))
+			return
+		}
+
+		pipelineId = location[4]
+	}
+
+	userAgent := request.Header.Get("User-Agent")
+	if strings.Contains(strings.ToLower(userAgent), "microsoftpreview") || strings.Contains(strings.ToLower(userAgent), "googlebot") {
+		log.Printf("[AUDIT] Blocking googlebot and microsoftbot for pielines. UA: '%s'", userAgent)
+		resp.WriteHeader(400)
+		resp.Write([]byte(`{"success": false, "reason": "Google/Microsoft preview bots not allowed. Please change the useragent."}`))
+		return
+	}
+
+	if len(pipelineId) != 45 {
+		log.Printf("[INFO] Couldn't handle pipeline. Too short in pipeline: %d", len(pipelineId))
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false, "reason": "pipeline ID not valid"}`))
+		return
+	}
+
+	pipelineId = pipelineId[9:]
+
+	pipeline, err := shuffle.GetPipeline(ctx, pipelineId)
+	if err != nil {
+		log.Printf("[WARNING] Failed getting pipeline %s (callback): %s", pipelineId, err)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	if pipeline.Status != "running" {
+		log.Printf("[WARNING] Not running %s because pipeline status is %s", pipeline.TriggerId, pipeline.Status)
+		resp.WriteHeader(401)
+		resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "The pipeline isn't running"}`)))
+		return
+	}
+
+	if pipeline.WorkflowId == "" {
+		log.Printf("[DEBUG] Not running because pipeline isn't connected to any workflows")
+		resp.WriteHeader(401)
+		resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No workflows are defined"}`)))
+		return
+	}
+
+	body, err := ioutil.ReadAll(request.Body)
+	if err != nil {
+		log.Printf("[DEBUG] Body data error: %s", err)
+		resp.WriteHeader(401)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	parsedBody := shuffle.GetExecutionbody(body)
+	newBody := shuffle.ExecutionStruct{
+		Start:             pipeline.StartNode,
+		ExecutionSource:   "pipeline",
+		ExecutionArgument: parsedBody,
+	}
+
+	workflow, err := shuffle.GetWorkflow(ctx, pipeline.WorkflowId)
+	if err == nil {
+		for _, branch := range workflow.Branches {
+			if branch.SourceID == pipeline.TriggerId {
+				log.Printf("[DEBUG] Found ID %s for pipeline", pipeline.TriggerId)
+				if branch.DestinationID != pipeline.StartNode {
+					newBody.Start = branch.DestinationID
+					break
+				}
+			}
+		}
+	}
+
+	b, err := json.Marshal(newBody)
+	if err != nil {
+		log.Printf("[ERROR] Failed newBody marshaling for pipeline: %s", err)
+		resp.WriteHeader(500)
+		resp.Write([]byte(`{"success": false}`))
+		return
+	}
+
+	log.Printf("[INFO] Running pipeline for workflow %s with startnode %s", pipeline.WorkflowId, pipeline.StartNode)
+
+	newWorkflow := shuffle.Workflow{
+		ID: "",
+	}
+
+	if len(pipeline.StartNode) == 0 {
+		log.Printf("[WARNING] No start node for pipeline %s - running with workflow default.", pipeline.TriggerId)
+
+	}
+
+	newRequest := &http.Request{
+		URL:    &url.URL{},
+		Method: "POST",
+		Body:   ioutil.NopCloser(bytes.NewReader(b)),
+	}
+
+	workflowExecution, executionResp, err := handleExecution(pipeline.WorkflowId, newWorkflow, newRequest, pipeline.OrgId)
+
+	if err == nil {
+		resp.WriteHeader(200)
+		resp.Write([]byte(fmt.Sprintf(`{"success": true, "execution_id": "%s"}`, workflowExecution.ExecutionId)))
+		return
+	}
+
+	resp.WriteHeader(500)
+	resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, executionResp)))
+}
+
 func executeCloudAction(action shuffle.CloudSyncJob, apikey string) error {
 	data, err := json.Marshal(action)
 	if err != nil {
@@ -4911,7 +5049,8 @@ func initHandlers() {
 	r.HandleFunc("/api/v1/triggers/gmail/register", shuffle.HandleNewGmailRegister).Methods("GET", "OPTIONS")
 	r.HandleFunc("/api/v1/triggers/gmail/getFolders", shuffle.HandleGetGmailFolders).Methods("GET", "OPTIONS")
 	r.HandleFunc("/api/v1/triggers/pipeline", shuffle.HandleNewPipelineRegister).Methods("POST", "OPTIONS")
-  r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS")
+    //r.HandleFunc("/api/v1/triggers/pipeline/save", shuffle.HandleSavePipelineInfo).Methods("PUT", "OPTIONS")
+	r.HandleFunc("/api/v1/pipelines/{key}", handlePipelineCallback).Methods("POST", "GET", "PATCH", "PUT", "DELETE", "OPTIONS")
 	r.HandleFunc("/api/v1/triggers", shuffle.HandleGetTriggers).Methods("GET", "OPTIONS")
 	//r.HandleFunc("/api/v1/triggers/gmail/routing", handleGmailRouting).Methods("POST", "OPTIONS")
 

frontend/src/views/AngularWorkflow.jsx

@@ -68,7 +68,8 @@ import {
   ListItemAvatar,
   Badge,
   AvatarGroup,
-  Autocomplete, 
+  Autocomplete,
+  Radio,
 } from "@mui/material";
 
 import {
@@ -220,7 +221,6 @@ export const triggers = [
 	  id: "",
     },
   ];
-	
 // Adds specific text to items
 
 // https://stackoverflow.com/questions/19014250/rerender-view-on-browser-resize-with-react
@@ -521,6 +521,10 @@ const AngularWorkflow = (defaultprops) => {
   const [highlightedApp, setHighlightedApp] = React.useState("")
 
   const [listCache, setListCache] = React.useState([]);
+
+  const [selectedOption, setSelectedOption] = React.useState("");
+  const [tenzirConfigModalOpen, setTenzirConfigModalOpen] = React.useState(false);
+
   const [suggestionBox, setSuggestionBox] = React.useState({
   	"position": {
   		"top": 500,
@@ -1342,6 +1346,56 @@ const AngularWorkflow = (defaultprops) => {
       });
   };
 
+  const handleKafkaSubmit = (trigger) => {
+    if (trigger.trigger_type !== "PIPELINE") {
+      toast("Unable to save the configuration");
+      return;
+    }
+  
+    trigger.parameters = [];
+
+    const topic = document.getElementById('topic')?.value;
+    const bootstrapServers = document.getElementById('bootstrap_servers')?.value;
+    const groupId = document.getElementById('group_id')?.value;
+    const autoOffsetReset = document.getElementById('auto_offset_reset')?.value;
+
+    if(topic) {
+      trigger.parameters.push({
+        name: "topic",
+        value: topic
+      });
+    } else {
+      toast("please enter the topic name");
+      return;
+    }
+  
+    if (bootstrapServers) {
+      trigger.parameters.push({
+        name: "bootstrap_servers",
+        value: bootstrapServers
+      });
+    } else {
+      toast("please enter bootstrap server details");
+      return;
+    }
+  
+    if (groupId) {
+      trigger.parameters.push({
+        name: "group_id",
+        value: groupId
+      });
+    }
+  
+    if (autoOffsetReset) {
+      trigger.parameters.push({
+        name: "auto_offset_reset",
+        value: autoOffsetReset
+      });
+    }
+  
+    setTenzirConfigModalOpen(false);
+  };
+  
 	const handleColoring = (actionId, status, label) => {
 		if (cy === undefined) {
 			return
@@ -6963,6 +7017,20 @@ const AngularWorkflow = (defaultprops) => {
       } else if (selectedNode.data().trigger_type === "EMAIL") {
         setSelectedTrigger(selectedNode.data());
         stopMailSub(selectedTrigger, triggerindex);
+      } else if (selectedNode.data().trigger_type === "PIPELINE") {
+        setSelectedTrigger(selectedNode.data());
+
+        const pipelineConfig = {
+          command: "",
+          name: selectedNode.data().label,
+          type: "delete",
+          environment: selectedNode.data().environment,
+          workflow_id: workflow.id,
+          trigger_id: selectedNode.data().id,
+          start_node: "",
+        };
+
+        submitPipeline(selectedNode.data(), triggerindex, pipelineConfig);
       }
     }
 
@@ -7348,47 +7416,67 @@ const AngularWorkflow = (defaultprops) => {
       toast("Error: name can't be empty");
       return;
     }
-
-	var mappedStartnode = ""
-	const alledges = cy.edges().jsons()
+  
+    var mappedStartnode = "";
+    const alledges = cy.edges().jsons();
     if (alledges !== undefined && alledges !== null && alledges.length > 0) {
-		for (let edgekey in alledges) {
-			const tmp = alledges[edgekey]
-			console.log("TMP: ", tmp, tmp.data.source)
-			if (tmp.data.source === trigger.id) {
-				mappedStartnode = tmp.data.target
-				break
-			}
-		}
+      for (let edgekey in alledges) {
+        const tmp = alledges[edgekey];
+        console.log("TMP: ", tmp, tmp.data.source);
+        if (tmp.data.source === trigger.id) {
+          mappedStartnode = tmp.data.target;
+          break;
+        }
+      }
+    }
+    const data = usecase;
+    data.start_node = mappedStartnode
+
+    if (data.type === "create") {
+      toast("Creating pipeline");
+    } else if (data.type === "stop") {
+      toast("stopping pipeline");
     }
 
-    toast("Creating pipeline") 
-    const data = usecase
-	const url = `${globalUrl}/api/v1/triggers/pipeline`
+    const url = `${globalUrl}/api/v1/triggers/pipeline`;
     fetch(url, {
-        method: "POST",
-        headers: {
-          "Content-Type": "application/json",
-          Accept: "application/json",
-        },
-        body: JSON.stringify(data),
-        credentials: "include",
-      }
-    )
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+        Accept: "application/json",
+      },
+      body: JSON.stringify(data),
+      credentials: "include",
+    })
       .then((response) => {
         if (response.status !== 200) {
           console.log("Status not 200 for stream results :O!");
-        } 
-
+        }
+  
         return response.json();
       })
       .then((responseJson) => {
         if (!responseJson.success) {
           toast("Failed to set pipeline: " + responseJson.reason);
         } else {
-          toast("Successfully created pipeline");
-          workflow.triggers[triggerindex].status = "running";
-          trigger.status = "running";
+          if (data.type === "create") {
+            toast("Pipeline will be created!");
+          } else if (data.type === "stop") {
+             toast("Pipeline will be stopped!");
+          } else {
+            toast("Pipeline deleted!")
+            return
+          }
+
+          trigger.parameters.push({
+            name: data.name,
+            value: data.command,
+          });
+          
+          if (data.type === "stop") trigger.status = "stopped";
+          else trigger.status = "running";
+          workflow.triggers[triggerindex] = trigger;
+  
           setSelectedTrigger(trigger);
           setWorkflow(workflow);
           console.log("Should set the status to running and save");
@@ -7396,11 +7484,10 @@ const AngularWorkflow = (defaultprops) => {
         }
       })
       .catch((error) => {
-        //toast(error.toString());
-        console.log("Get schedule error: ", error.toString());
+        console.log("Get pipeline error: ", error.toString());
       });
-  }
-
+  };
+  
   const submitSchedule = (trigger, triggerindex) => {
     if (trigger.name.length <= 0) {
       toast("Error: name can't be empty");
@@ -13825,225 +13912,270 @@ const AngularWorkflow = (defaultprops) => {
   }
 
   const PipelineSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined && selectedTrigger.trigger_type !== "SCHEDULE" ? null : !userdata.support === true ? null : 
-        <div style={appApiViewStyle}>
-		  <h3 style={{ marginBottom: "5px" }}>
-			{selectedTrigger.app_name}: {selectedTrigger.status}
-		  </h3>
-		  <a
-			rel="noopener noreferrer"
-			target="_blank"
-			href="https://shuffler.io/docs/triggers#pipelines"
-			style={{ textDecoration: "none", color: "#f85a3e" }}
-		  >
-			What are pipelines?
-		  </a>
-          <Divider
-            style={{
-              marginBottom: "10px",
-              marginTop: "10px",
-              height: "1px",
-              width: "100%",
-              backgroundColor: "rgb(91, 96, 100)",
-            }}
-          />
-          <div>Name</div>
-          <TextField
-            style={{
-              backgroundColor: theme.palette.inputColor,
-              borderRadius: theme.palette.borderRadius,
-            }}
-            InputProps={{
-              style: {
-              },
-            }}
-            fullWidth
-            color="primary"
-            placeholder={selectedTrigger.label}
-            onChange={selectedTriggerChange}
-          />
-
-          <div style={{ marginTop: "20px" }}>
-            <Typography>Environment</Typography>
-            <Select
-              MenuProps={{
-                disableScrollLock: true,
-              }}
-              value={selectedTrigger.environment}
-              disabled={selectedTrigger.status === "running"}
-              SelectDisplayProps={{}}
-              fullWidth
-              onChange={(e) => {
-                selectedTrigger.environment = e.target.value
-                setSelectedTrigger(selectedTrigger)
-
-                setWorkflow(workflow)
-                setUpdate(Math.random())
+          <div style={appApiViewStyle}>
+            <h3 style={{ marginBottom: "5px" }}>
+              {selectedTrigger.app_name}: {selectedTrigger.status}
+            </h3>
+            <a
+              rel="noopener noreferrer"
+              target="_blank"
+              href="https://shuffler.io/docs/triggers#pipelines"
+              style={{ textDecoration: "none", color: "#f85a3e" }}
+            >
+              What are pipelines?
+            </a>
+            <Divider
+              style={{
+                marginBottom: "10px",
+                marginTop: "10px",
+                height: "1px",
+                width: "100%",
+                backgroundColor: "rgb(91, 96, 100)",
               }}
+            />
+            <div>Name</div>
+            <TextField
               style={{
                 backgroundColor: theme.palette.inputColor,
-                color: "white",
-                height: 50,
+                borderRadius: theme.palette.borderRadius,
               }}
-            >
-              {environments.map((data) => {
-                if (data.archived) {
-                    return null
-                }
+              InputProps={{
+                style: {},
+              }}
+              fullWidth
+              color="primary"
+              placeholder={selectedTrigger.label}
+              onChange={selectedTriggerChange}
+            />
 
-				if (data.Name.toLowerCase() === "cloud") {
-					return null
-				}
+            <div style={{ marginTop: "20px" }}>
+              <Typography>Environment</Typography>
+              <Select
+                MenuProps={{
+                  disableScrollLock: true,
+                }}
+                value={selectedTrigger.environment}
+                disabled={selectedTrigger.status === "running"}
+                SelectDisplayProps={{}}
+                fullWidth
+                onChange={(e) => {
+                  selectedTrigger.environment = e.target.value;
+                  setSelectedTrigger(selectedTrigger);
 
-                return (
-                  <MenuItem
-                    key={data.id}
-                    style={{ backgroundColor: theme.palette.inputColor, color: "white" }}
-                    value={data.Name}
-                  >
-                    {data.Name}
-                  </MenuItem>
-                );
-              })}
-            </Select>
-          </div>
-          <Divider
-            style={{
-              marginTop: "20px",
-              height: "1px",
-              width: "100%",
-              backgroundColor: "rgb(91, 96, 100)",
-            }}
-          />
-          <div style={{ flex: "6", marginTop: 20, }}>
-            <div>
-              <b>Parameters</b>
-
-			  {/*
-			  <div 
-				style={{
-					border: "1px solid rgba(255,255,255,0.3)",
-					borderRadius: theme.palette.borderRadius,
-					padding: 10,
-					cursor: "pointer",
-
-				}}
-				onClick={() => {
-					const pipelineConfig = {
-						"name": "HTTP Testing",
-						"type": "create",
-						"command": "from http://192.168.86.44:5002/api/v1/orgs/7e9b9007-5df2-4b47-bca5-c4d267ef2943/cache/CIDR%20ranges?type=text&authorization=cec9d01f-09b2-4419-8a0a-76c6046e3fef read lines | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines",
-						"environment": selectedTrigger.environment,
-					}
-
-                    submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig)
-				}}
-			  >
-				Run HTTP Request
-			  </div>
-			  */}
-
-			  <div 
-				style={{
-					border: "1px solid rgba(255,255,255,0.3)",
-					borderRadius: theme.palette.borderRadius,
-					padding: 10,
-					cursor: "pointer",
-					marginTop: 5, 
-
-				}}
-				onClick={() => {
-					const pipelineConfig = {
-						"name": selectedTrigger.label,
-						"type": "create",
-						"command": "load tcp://0.0.0.0:514 | read syslog | export",
-						"environment": selectedTrigger.environment,
-					}
-
-                    submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig)
-				}}
-			  >
-				Start Syslog listener
-			  </div>
-
-			  <div 
-				style={{
-					border: "1px solid rgba(255,255,255,0.3)",
-					borderRadius: theme.palette.borderRadius,
-					padding: 10,
-					cursor: "pointer",
-					marginTop: 5, 
-
-				}}
-				onClick={() => {
-					const pipelineConfig = {
-						"name": selectedTrigger.label,
-						"type": "create",
-						"command": "export --live | sigma /path/to/rules | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines",
-						"environment": selectedTrigger.environment,
-					}
-
-                    submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig)
-				}}
-			  >
-				Run Sigma Rulesearch 
-			  </div>
-
-			  <div 
-				style={{
-					border: "1px solid rgba(255,255,255,0.3)",
-					borderRadius: theme.palette.borderRadius,
-					padding: 10,
-					cursor: "pointer",
-					marginTop: 5, 
-
-				}}
-				onClick={() => {
-					const pipelineConfig = {
-						"name": selectedTrigger.label,
-						"type": "create",
-						"command": "from kafka://1.2.3.4 --topic foo | to http://api.com X-Token:Secret",
-						"environment": selectedTrigger.environment,
-					}
-
-                    submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig)
-				}}
-			  >
-				Follow Kafka Queue
-			  </div>
-
-              <div
+                  setWorkflow(workflow);
+                  setUpdate(Math.random());
+                }}
                 style={{
-                  marginTop: "20px",
-                  marginBottom: "7px",
-                  display: "flex",
+                  backgroundColor: theme.palette.inputColor,
+                  color: "white",
+                  height: 50,
                 }}
               >
-                <Button
-                  style={{ flex: "1" }}
-                  variant="contained"
-                  disabled={selectedTrigger.status === "running"}
+                {environments.map((data) => {
+                  if (data.archived) {
+                    return null;
+                  }
+
+                  if (data.Name.toLowerCase() === "cloud") {
+                    return null;
+                  }
+
+                  return (
+                    <MenuItem
+                      key={data.id}
+                      style={{
+                        backgroundColor: theme.palette.inputColor,
+                        color: "white",
+                      }}
+                      value={data.Name}
+                    >
+                      {data.Name}
+                    </MenuItem>
+                  );
+                })}
+              </Select>
+            </div>
+            <Divider
+              style={{
+                marginTop: "20px",
+                height: "1px",
+                width: "100%",
+                backgroundColor: "rgb(91, 96, 100)",
+              }}
+            />
+            <div style={{ flex: 6, marginTop: 20 }}>
+              <div>
+                <b>Parameters</b>
+                <div
+                  key="syslogListener"
                   onClick={() => {
-					  toast("Should start. But it doesn't")
+                    // setSelectedOption("Syslog listener")
+                    // setTenzirConfigModalOpen(true);
+                  }}
+                  style={{
+                    border: "1px solid rgba(255,255,255,0.3)",
+                    borderRadius: theme.palette.borderRadius,
+                    padding: 10,
+                    cursor: "not-allowed",
+                    marginTop: 5,
+                    display: "flex",
+                    alignItems: "center",
                   }}
-                  color="primary"
                 >
-                  Start
-                </Button>
-                <Button
-                  style={{ flex: "1" }}
-                  variant="contained"
-                  disabled={selectedTrigger.status !== "running"}
+                  <FormControlLabel
+                    control={
+                      <Radio
+                        checked={selectedOption === "Syslog listener"}
+                        onChange={() => setSelectedOption("Syslog listener")}
+                        value={"Syslog listener"}
+                        name="option"
+                        disabled={true}
+                      />
+                    }
+                    label="Start Syslog listener"
+                  />
+                </div>
+
+                <div
+                  key="sigmaRulesearch"
                   onClick={() => {
-					  toast("Should stop triggert")
+                    // setSelectedOption("Sigma Rulesearch")
+                    // setTenzirConfigModalOpen(true);
+                  }}
+                  style={{
+                    border: "1px solid rgba(255,255,255,0.3)",
+                    borderRadius: theme.palette.borderRadius,
+                    padding: 10,
+                    cursor: "not-allowed",
+                    marginTop: 5,
+                    display: "flex",
+                    alignItems: "center",
                   }}
-                  color="primary"
                 >
-                  Stop
-                </Button>
+                  <FormControlLabel
+                    control={
+                      <Radio
+                        checked={selectedOption === "Sigma Rulesearch"}
+                        onChange={() => setSelectedOption("Sigma Rulesearch")}
+                        value={"Sigma Rulesearch"}
+                        name="option"
+                        disabled={true}
+                      />
+                    }
+                    label="Run Sigma Rulesearch"
+                  />
+                </div>
+
+                <div
+                  key="kafkaQueue"
+                  onClick={() => {
+                    if(selectedTrigger.status === "running"){
+                      toast("please stop the trigger to edit the configuration");
+                      return;
+                    } else {
+                    setSelectedOption("Kafka Queue");
+                    setTenzirConfigModalOpen(true);
+                  }}}
+                  style={{
+                    border: "1px solid rgba(255,255,255,0.3)",
+                    borderRadius: theme.palette.borderRadius,
+                    padding: 10,
+                    cursor: "pointer",
+                    marginTop: 5,
+                    display: "flex",
+                    alignItems: "center",
+                  }}
+                >
+                  <FormControlLabel
+                    control={
+                      <Radio
+                        checked={selectedOption === "Kafka Queue"}
+                        onChange={() => setSelectedOption("Kafka Queue")}
+                        value={"Kafka Queue"}
+                        name="option"
+                      />
+                    }
+                    label="Follow Kafka Queue"
+                  />
+                </div>
+
+                <div style={{ marginTop: 20, marginBottom: 7, display: "flex" }}>
+                  <Button
+                    style={{ flex: 1, marginRight: 7 }}
+                    variant="contained"
+                    disabled={selectedTrigger.status === "running"}
+                    onClick={() => {
+
+                      const topic = (selectedTrigger?.parameters?.find(param => param.name === "topic")?.value) || ''
+                      const bootstrapServers = (selectedTrigger?.parameters?.find(param => param.name === "bootstrap_servers")?.value) || ''
+                      const groupId = (selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || ''
+                      const autoOffsetReset = (selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || ''
+                      let command = "from kafka"
+                      
+                      if(topic) {
+                        command = `${command} -t ${topic}`
+                      } else {
+                        toast("please enter the topic name")
+                        return;
+                      }
+                      if(bootstrapServers) {
+                        command = `${command} -e -o stored -X bootstrap.servers=${bootstrapServers}`
+                      } else {
+                        toast("please enter the bootstrap servers details")
+                        return;
+                      }
+
+                      if(groupId) {
+                         command = `${command},group.id=${groupId}`
+                      } else {
+                        command = `${command},group.id=${selectedTrigger.id}`
+                      }
+                      if(autoOffsetReset) {
+                        command = `${command},auto.offset.reset=${autoOffsetReset}`
+                      } else {
+                        command = `${command},auto.offset.reset=earliest`
+                      }
+                      command = `${command},client.id=${selectedTrigger.id},enable.auto.commit=true,auto.commit.interval.ms=1`
+                      command = `${command} read json | to ${globalUrl}/api/v1/pipelines/pipeline_${selectedTrigger.id}`
+
+                      const pipelineConfig = {
+                        command: command,
+                        name: selectedTrigger.label,
+                        type: "create",
+                        environment: selectedTrigger.environment,
+                        workflow_id: workflow.id,
+                        trigger_id: selectedTrigger.id,
+                        start_node: "",
+                      };
+                      submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig);
+                    }}
+                    color="primary"
+                  >
+                    Start
+                  </Button>
+                  <Button
+                    style={{ flex: 1 }}
+                    variant="contained"
+                    disabled={selectedTrigger.status !== "running"}
+                    onClick={() => {
+                      const pipelineConfig = {
+                        name: selectedTrigger.label,
+                        type: "stop",
+                        environment: selectedTrigger.environment,
+                        workflow_id: workflow.id,
+                        trigger_id: selectedTrigger.id,
+                        start_node: "",
+                      };
+                      submitPipeline(selectedTrigger, selectedTriggerIndex, pipelineConfig);
+                    }}
+                    color="primary"
+                  >
+                    Stop
+                  </Button>
+                </div>
               </div>
             </div>
           </div>
-        </div>
 
   const ScheduleSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined && selectedTrigger.trigger_type !== "SCHEDULE" ? null :
         <div style={appApiViewStyle}>
@@ -14180,7 +14312,7 @@ const AngularWorkflow = (defaultprops) => {
 				  selectedTrigger.status === "running"
                 }
                 defaultValue={
-					selectedTrigger.parameters === undefined ? "" : selectedTrigger.parameters[0].value
+					selectedTrigger.parameters === undefined ? "" : selectedTrigger.parameters[0]?.value
                 }
                 color="primary"
                 placeholder=""
@@ -15997,7 +16129,18 @@ const AngularWorkflow = (defaultprops) => {
 		  style={{paddingTop: 8, paddingLeft: 4, height: 25, width: 25, }}
 		/>
       );
-    }
+    } else if (execution.execution_source === "pipeline") {
+      return (
+        <img
+          alt={"pipeline"}
+          src={
+            triggers.find((trigger) => trigger.trigger_type === "PIPELINE")
+              .large_image
+          }
+          style={{ width: size, height: size }}
+        />
+      );
+    } 
 
     if (
       execution.execution_parent !== null &&
@@ -18978,6 +19121,146 @@ const AngularWorkflow = (defaultprops) => {
     </Dialog>
   ) : null;
 
+  const tenzirConfigModal = tenzirConfigModalOpen ? (
+        <Dialog
+          PaperComponent={PaperComponent}
+          hideBackdrop={true}
+          disableEnforceFocus={true}
+          disableBackdropClick={true}
+          style={{ pointerEvents: "none" }}
+          open={tenzirConfigModalOpen}
+          PaperProps={{
+            style: {
+              pointerEvents: "auto",
+              color: "white",
+              minWidth: 600,
+              minHeight: 500,
+              maxHeight: 500,
+              padding: 15,
+              overflow: "hidden",
+              zIndex: 10012,
+              border: theme.palette.defaultBorder,
+            },
+          }}
+        >
+          <div
+            style={{
+              flex: 2,
+              padding: 0,
+              minHeight: isMobile ? "90%" : 700,
+              maxHeight: isMobile ? "90%" : 700,
+              overflowY: "auto",
+              overflowX: isMobile ? "auto" : "hidden",
+            }}
+          >
+            <DialogTitle id="tenzir-config-modal" style={{ cursor: "move" }}>
+              <div style={{ color: "white" }}>Configuration options for {selectedOption}</div>
+            </DialogTitle>
+            <DialogContent>
+              {selectedOption === "Kafka Queue" && (
+                <>
+                  <b>Topic</b>
+                  <TextField
+                    id="topic"
+                    style={{
+                      backgroundColor: theme.palette.inputColor,
+                      borderRadius: theme.palette.borderRadius,
+                    }}
+                    InputProps={{
+                      style: {},
+                    }}
+                    fullWidth
+                    color="primary"
+                    placeholder={"topic name"}
+                    defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "topic")?.value) || ''}
+                  />
+                  <b>bootstrap.servers</b>
+                  <TextField
+                    id="bootstrap_servers"
+                    style={{
+                      backgroundColor: theme.palette.inputColor,
+                      borderRadius: theme.palette.borderRadius,
+                    }}
+                    InputProps={{
+                      style: {},
+                    }}
+                    fullWidth
+                    color="primary"
+                    placeholder={"broker1.example.com:9092,192.168.1.100:9092"}
+                    defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "bootstrap_servers")?.value) || ''}
+                  />
+                  <b>group.id</b>
+                  <TextField
+                    id="group_id"
+                    style={{
+                      backgroundColor: theme.palette.inputColor,
+                      borderRadius: theme.palette.borderRadius,
+                    }}
+                    InputProps={{
+                      style: {},
+                    }}
+                    fullWidth
+                    color="primary"
+                    placeholder={"tenzir"}
+                    defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "group_id")?.value) || ''}    
+                  />
+                  <b>auto.offest.reset</b>
+                  <TextField
+                    id="auto_offset_reset"
+                    style={{
+                      backgroundColor: theme.palette.inputColor,
+                      borderRadius: theme.palette.borderRadius,
+                    }}
+                    InputProps={{
+                      style: {},
+                    }}
+                    fullWidth
+                    color="primary"
+                    placeholder={"earliest"}
+                    defaultValue={(selectedTrigger?.parameters?.find(param => param.name === "auto_offset_reset")?.value) || ''}
+                  />
+                </>
+              )}
+            </DialogContent>
+            <DialogActions>
+              <Button
+                style={{ borderRadius: "0px" }}
+                onClick={() => {
+                  setTenzirConfigModalOpen(false);
+                }}
+                color="primary"
+              >
+                Cancel
+              </Button>
+              <Button
+                style={{ borderRadius: "0px" }}
+                onClick={() => {
+                  handleKafkaSubmit(selectedTrigger);
+                }}
+                color="primary"
+              >
+                Submit
+              </Button>
+            </DialogActions>
+          </div>
+
+          <IconButton
+            style={{
+              zIndex: 5000,
+              position: "absolute",
+              top: 14,
+              right: 18,
+              color: "grey",
+            }}
+            onClick={() => {
+              setTenzirConfigModalOpen(false);
+            }}
+          >
+            <CloseIcon />
+          </IconButton>
+        </Dialog>
+      ) : null;
+
 	// Should get AI autocompletes
 	const aiSubmit = (value, setResponseMsg, setSuggestionLoading, inputAction) => {
 		if (setResponseMsg !== undefined) {
@@ -19794,6 +20077,7 @@ const AngularWorkflow = (defaultprops) => {
         {codePopoutModal}
 		{workflowRevisions}
         {authenticationModal}
+        {tenzirConfigModal}
         {/*editWorkflowModal*/}
   		{executionArgumentModal}
         {configureWorkflowModal}

functions/onprem/orborus/orborus.go

@@ -1667,7 +1667,7 @@ func main() {
 			newrequests := []shuffle.ExecutionRequest{}
 			for _, incRequest := range executionRequests.Data {
 				// Looking for specific jobs
-				if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" {
+				if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" {
 
 					err := handlePipeline(incRequest)
 					if err != nil {
@@ -2027,40 +2027,38 @@ func main() {
 // Read from Cache and send it to a webhook
 // docker run tenzir/tenzir:latest 'from http://192.168.86.44:5002/api/v1/orgs/7e9b9007-5df2-4b47-bca5-c4d267ef2943/cache/CIDR%20ranges?type=text&authorization=cec9d01f-09b2-4419-8a0a-76c6046e3fef read lines | to http://192.168.86.44:5002/api/v1/hooks/webhook_665ace5f-f27b-496a-a365-6e07eb61078c write lines'
 func handlePipeline(incRequest shuffle.ExecutionRequest) error {
+	
+	if tenzirUrl == "" {
+		tenzirUrl = "http://localhost:5160"
+		log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl)
+	}
+
 	err := deployTenzirNode()
 	if err != nil{
 		log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
+		return err
 	}
 
-	// no need of execution arguments for state updates
-	if incRequest.Type != "PIPELINE_STOP" && len(incRequest.ExecutionArgument) == 0 {
+	// no need of execution arguments for STOP and DELETE
+	if (incRequest.Type != "PIPELINE_STOP" && incRequest.Type != "PIPELINE_DELETE") && len(incRequest.ExecutionArgument) == 0 {
 		log.Printf("[ERROR] No execution argument found for pipeline create. Skipping")
 
 		return errors.New("no execution argument found for pipeline create. Skipping")
 	}
-
 	//image := "tenzir/tenzir:latest"
 	identifier := fmt.Sprintf("shuffle-%s", strings.ToLower(strings.ReplaceAll(incRequest.ExecutionSource, " ", "-")))
 	command := incRequest.ExecutionArgument
 
 	if incRequest.Type == "PIPELINE_CREATE" {
-		log.Printf("[INFO] Should delete -> recreate new pipeline %#v. Name: %#v", incRequest.ExecutionArgument, identifier)
+		log.Printf("[INFO] Should delete -> recreate new pipeline with id %#v", identifier)
 		//err := deployPipeline(image, identifier, command)
-		pipelineId, err := createPipeline(command, identifier)
+		_, err := createPipeline(command, identifier)
 		if err != nil {
 			log.Printf("[ERROR] Failed to create pipeline: %s", err)
 			return err
-		} else {
-			log.Printf("[INFO] Pipeline created successfully with Id: %s", pipelineId)
-			newErr := savePipelineData(pipelineId, identifier, "running")
-			if newErr != nil {
-				log.Printf("[DEBUG] failed to save the pipeline data: %s", newErr)
-			} else {
-				log.Printf("[INFO] succesfully saved the pipeline info ")
-			}
 		}
 	} else if incRequest.Type == "PIPELINE_DELETE" {
-		log.Printf("[INFO] Should delete pipeline %#v", incRequest.ExecutionArgument)
+		log.Printf("[INFO] Should delete pipeline %#v", identifier)
 		pipelineId, err := searchPipeline(identifier)
 		if err != nil {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
@@ -2070,6 +2068,8 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 		if err != nil {
 			log.Printf("[ERROR] Failed Deleting Pipeline %s", err)
 			return err
+		} else {
+			log.Printf("[INFO] successfully deleted the Pipeline: %s", pipelineId)
 		}
 	} else if incRequest.Type == "PIPELINE_STOP" {
 		log.Printf("[INFO] Should stop the pipeline %#v", identifier)
@@ -2078,18 +2078,32 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
 			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
 			return err
 		}
-		state, err := updatePipelineState(pipelineId, "stop")
+		_, err = updatePipelineState(pipelineId, "stop")
 		if err != nil {
 			log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err)
 			return err
 		} else {
 			log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId)
 		}
-		err = savePipelineData(pipelineId, identifier, state)
+
+	}  else if incRequest.Type == "PIPELINE_START" {
+		log.Printf("[INFO] Should start the pipeline %#v", identifier)
+		pipelineId, err := searchPipeline(identifier)
+		if err != nil { 
+			if err.Error() == "no existing pipeline found with name" {
+				log.Printf("[WARNING] no pipeline found for %s, creating a new one", identifier)
+				_, CreateErr := createPipeline(command, identifier)
+				return CreateErr
+			}
+			log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err)
+			return err
+		}
+		_, err = updatePipelineState(pipelineId, "start")
 		if err != nil {
-			log.Printf("[DEBUG] failed to save the pipeline data: %s", err)
+			log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err)
+			return err
 		} else {
-			log.Printf("[INFO] succesfully saved the pipeline info ")
+			log.Printf("[INFO] successfully started the Pipeline: %s", pipelineId)
 		}
 
 	} else {
@@ -2106,7 +2120,7 @@ func deployTenzirNode() error {
     }
 
     ctx := context.Background()
-	cacheKey := "tenzir-key"
+    cacheKey := "tenzir-key"
 
     imageName := "tenzir/tenzir:latest"
     containerName := "tenzir-node"
@@ -2114,19 +2128,29 @@ func deployTenzirNode() error {
 
     _, err := shuffle.GetCache(ctx, cacheKey)
     if err == nil {
-		return nil
-	}
+        return nil
+    }
 
     containerInfo, err := dockercli.ContainerInspect(ctx, containerName)
     if err != nil {
         if dockerclient.IsErrNotFound(err) {
-            pullOptions := types.ImagePullOptions{}
-            out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
-            if err != nil {
-                log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err)
+
+            // Check if image exists
+            _, _, err := dockercli.ImageInspectWithRaw(ctx, imageName)
+            if dockerclient.IsErrNotFound(err) {
+                log.Printf("[DEBUG] pulling image %s", imageName)
+                pullOptions := types.ImagePullOptions{}
+                out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
+                if err != nil {
+                    log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err)
+                    return err
+                }
+                defer out.Close()
+
+                io.Copy(io.Discard, out)
+            } else if err != nil {
                 return err
             }
-            defer out.Close()
 
             err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions)
             if err != nil {
@@ -2137,37 +2161,35 @@ func deployTenzirNode() error {
         }
     } else {
         if !containerInfo.State.Running {
-            log.Printf("[DEBUG] Tenzir Node exists but is not running, starting it")
+            log.Printf("[DEBUG] Tenzir Node exists but is not running")
             err := dockercli.ContainerStart(ctx, containerName, containerStartOptions)
             if err != nil {
                 log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
                 return err
             }
-            log.Printf("[INFO] Tenzir Node container started successfully")
 
             log.Printf("[INFO] Waiting for Tenzir to become available ...")
             err = checkTenzirNode()
             if err != nil {
                 return err
             }
-            log.Printf("[INFO] Successfully deployed Tenzir Node!")
         }
     }
 
-	tenzirStatus := struct {
-		ContainerStatus   string `json:"container_status"`
-	}{
-		ContainerStatus: "running",
-	}
+    tenzirStatus := struct {
+        ContainerStatus string `json:"container_status"`
+    }{
+        ContainerStatus: "running",
+    }
 
-	cacheData, err := json.Marshal(tenzirStatus)
-	if err != nil {
-		log.Printf("[WARNING] Failed marshalling execution: %s", err)
-	}
-	err = shuffle.SetCache(ctx, cacheKey, cacheData, 1)
-	if err != nil {
+    cacheData, err := json.Marshal(tenzirStatus)
+    if err != nil {
+        log.Printf("[WARNING] Failed marshalling execution: %s", err)
+    }
+    err = shuffle.SetCache(ctx, cacheKey, cacheData, 1)
+    if err != nil {
         log.Printf("[WARNING] Failed updating cache for tenzir: %s", err)
-	}
+    }
 
     return nil
 }
@@ -2264,6 +2286,24 @@ func createPipeline(command, identifier string) (string, error) {
 		log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId)
 		toBeDeleted = true
 	}
+    if strings.Contains(command, "shuffler.io") {
+
+	} else {
+		var scheme string
+		if strings.Contains(command, "http://") {
+			scheme = "http://"
+		} else if strings.Contains(command, "https://") {
+			scheme = "https://"
+		}
+
+		startIndex := strings.Index(command, scheme)
+		if startIndex != -1 {
+			endIndex := startIndex + len(scheme)
+			endIndex += strings.Index(command[endIndex:], "/")
+			
+			command = command[:startIndex] + baseUrl + command[endIndex:]
+		}
+	}
 
 	requestBody := map[string]interface{}{
 		"definition": command,
@@ -2271,12 +2311,12 @@ func createPipeline(command, identifier string) (string, error) {
 		"hidden":     false,
 		"autostart": map[string]bool{
 			"created":   true,
-			"completed": false,
-			"failed":    false,
+			"completed": true,
+			"failed":    true,
 		},
 		"autodelete": map[string]bool{
 			"completed": false,
-			"failed":    true,
+			"failed":    false,
 			"stopped":   false,
 		},
 		"retry_delay": "500.0ms",
@@ -2348,12 +2388,12 @@ func updatePipelineState(pipelineId, action string) (string, error) {
 		"action": action,
 		"autostart": map[string]bool{
 			"created":   true,
-			"completed": false,
-			"failed":    false,
+			"completed": true,
+			"failed":    true,
 		},
 		"autodelete": map[string]bool{
 			"completed": false,
-			"failed":    true,
+			"failed":    false,
 			"stopped":   false,
 		},
 	}
@@ -2490,53 +2530,53 @@ func searchPipeline(identifier string) (string, error) {
 	return "", errors.New("no existing pipeline found with name")
 }
 
-func savePipelineData(pipelineId, identifier, status string) error {
+// func savePipelineData(pipelineId, identifier, status string) error {
 
-	url :=  fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl)
-	identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-")
+// 	url :=  fmt.Sprintf("%s/api/v1/triggers/pipeline/save", baseUrl)
+// 	identifierWithoutPrefix := strings.TrimPrefix(identifier, "shuffle-")
 
-	forwardMethod := "PUT"
+// 	forwardMethod := "PUT"
 
-	payload := map[string]interface{}{
-		"pipeline_id": pipelineId,
-		"trigger_id":  identifierWithoutPrefix,
-		"status":      status,
-	}
+// 	payload := map[string]interface{}{
+// 		"pipeline_id": pipelineId,
+// 		"trigger_id":  identifierWithoutPrefix,
+// 		"status":      status,
+// 	}
 
-	payloadBytes, err := json.Marshal(payload)
-	if err != nil {
-		log.Printf("[ERROR] Failed to marshal payload: %s", err)
-		return err
-	}
+// 	payloadBytes, err := json.Marshal(payload)
+// 	if err != nil {
+// 		log.Printf("[ERROR] Failed to marshal payload: %s", err)
+// 		return err
+// 	}
 
-	forwardData := bytes.NewBuffer(payloadBytes)
+// 	forwardData := bytes.NewBuffer(payloadBytes)
 
-	req, err := http.NewRequest(
-		forwardMethod,
-		url,
-		forwardData,
-	)
-	if err != nil {
-		log.Printf("[ERROR] Failed to create HTTP request: %s", err)
-		return err
-	}
-	req.Header.Set("Content-Type", "application/json")
+// 	req, err := http.NewRequest(
+// 		forwardMethod,
+// 		url,
+// 		forwardData,
+// 	)
+// 	if err != nil {
+// 		log.Printf("[ERROR] Failed to create HTTP request: %s", err)
+// 		return err
+// 	}
+// 	req.Header.Set("Content-Type", "application/json")
 
-	client := &http.Client{Timeout: 10 * time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		log.Printf("[ERROR] Failed to send HTTP request: %s", err)
-		return err
-	}
-	defer resp.Body.Close()
+// 	client := &http.Client{Timeout: 10 * time.Second}
+// 	resp, err := client.Do(req)
+// 	if err != nil {
+// 		log.Printf("[ERROR] Failed to send HTTP request: %s", err)
+// 		return err
+// 	}
+// 	defer resp.Body.Close()
 
-	if resp.StatusCode != 200 {
-		log.Printf("[ERROR] Received non-successful HTTP status code: %d", resp.StatusCode)
-		return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode)
-	}
+// 	if resp.StatusCode != 200 {
+// 		log.Printf("[ERROR] Received non-successful HTTP status code: %d", resp.StatusCode)
+// 		return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode)
+// 	}
 
-	return nil
-}
+// 	return nil
+// }
 
 // Is this ok to do with Docker? idk :)
 func getRunningWorkers(ctx context.Context, workerTimeout int) int {