diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 7b8008ce..922bdced 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -291,34 +291,49 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { ctx := shuffle.GetContext(request) env, err := shuffle.GetEnvironment(ctx, orgId, "") timeNow := time.Now().Unix() - if err == nil && len(env.Id) > 0 && len(env.Name) > 0 { + if err == nil && len(env.Id) > 0 && len(env.Name) > 0 && request.Method == "POST" { // Updates every 60 seconds~ if time.Now().Unix() > env.Edited+60 { env.RunningIp = shuffle.GetRequestIp(request) + + // Orborus label = custom label for Orborus if len(orborusLabel) > 0 { env.RunningIp = orborusLabel } - if request.Method == "POST" { - body, err := ioutil.ReadAll(request.Body) - if err == nil { - var envData shuffle.OrborusStats - err = json.Unmarshal(body, &envData) - if err == nil { - if envData.Swarm { - env.Licensed = true - env.RunType = "docker" - } + // Set the checkin cache - if envData.Kubernetes { - env.RunType = "k8s" - } + + body, err := ioutil.ReadAll(request.Body) + if err == nil { + var envData shuffle.OrborusStats + err = json.Unmarshal(body, &envData) + if err == nil { + envData.RunningIp = env.RunningIp + + marshalled, err := json.Marshal(envData) + if err == nil { + cacheKey := fmt.Sprintf("queueconfig-%s-%s", env.Name, env.OrgId) + go shuffle.SetCache(context.Background(), cacheKey, marshalled, 2) } + + + + if envData.Swarm { + env.Licensed = true + env.RunType = "docker" + } + + if envData.Kubernetes { + env.RunType = "k8s" + } + + envData.DataLake = env.DataLake } } env.Checkin = timeNow - err = shuffle.SetEnvironment(ctx, env) + err = shuffle.SetEnvironment(ctx, &env) if err != nil { log.Printf("[ERROR] Failed updating environment: %s", err) } diff --git a/functions/onprem/orborus/go.mod b/functions/onprem/orborus/go.mod index 74471ede..06ae0659 100644 --- a/functions/onprem/orborus/go.mod +++ b/functions/onprem/orborus/go.mod @@ -4,7 +4,7 @@ go 1.22.0 toolchain go1.22.2 -//replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared +replace github.com/shuffle/shuffle-shared => ../../../../shuffle-shared require ( github.com/docker/docker v27.0.2+incompatible diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 3246fb4a..dfa645b7 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -30,8 +30,8 @@ import ( "strings" "sync" "time" - "math/rand" + "math/rand" //"os/signal" //"syscall" @@ -103,11 +103,12 @@ var swarmNetworkName = os.Getenv("SHUFFLE_SWARM_NETWORK_NAME") var orborusLabel = os.Getenv("SHUFFLE_ORBORUS_LABEL") var memcached = os.Getenv("SHUFFLE_MEMCACHED") -// For it to download from Sigma? -var apiKey = os.Getenv("AUTH_FOR_ORBORUS") +// For it to download from Sigma? +var apiKey = os.Getenv("AUTH_FOR_ORBORUS") var pipelineUrl = os.Getenv("SHUFFLE_PIPELINE_URL") var executionIds = []string{} +var pipelines = []shuffle.PipelineInfoMini{} var namespacemade = false // For K8s var dockercli *dockerclient.Client @@ -725,7 +726,6 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar { func handleBackendImageDownload(ctx context.Context, images string) error { - // Replicate images with lowercase, as the name may be wrong // Most of the time lowercase is correct. Swapping to have that first originalImages := images @@ -756,7 +756,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { resp, err := dockercli.ImageRemove(ctx, image, removeOptions) if err != nil { log.Printf("[ERROR] Failed removing image: %s. Resp: %#v", err, resp) - + // Goroutining images that don't already exist, as they are most likely not the correct one go shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, image) } else { @@ -806,10 +806,10 @@ func handleBackendImageDownload(ctx context.Context, images string) error { //docker service update --image username/imagename:latest servicename --force serviceUpdateOptions := types.ServiceUpdateOptions{} resp, err := dockercli.ServiceUpdate( - ctx, - service.ID, - service.Version, - service.Spec, + ctx, + service.ID, + service.Version, + service.Spec, serviceUpdateOptions, ) @@ -824,7 +824,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { } } } - + } } @@ -1522,7 +1522,6 @@ func checkSwarmService(ctx context.Context) { log.Printf("[DEBUG] Swarm info: %s\n\n", ret) } - func getContainerResourceUsage(ctx context.Context, cli *dockerclient.Client, containerID string) (float64, float64, error) { // Get container stats stats, err := cli.ContainerStats(ctx, containerID, false) @@ -2019,8 +2018,6 @@ func main() { log.Printf("[INFO] Waiting for executions at %s with Environment %#v", fullUrl, environment) hasStarted := false for { - _ = sendTenzirHealthStatus() - if req.Method == "POST" { // Should find data to send (memory etc.) @@ -2030,6 +2027,13 @@ func main() { // Marshal and set body orborusStats := getOrborusStats(ctx) + pipelinePayload, pipelineerr := sendPipelineHealthStatus() + if pipelineerr != nil { + // Too verbose to be enabled. + //log.Printf("[ERROR] Failed sending pipeline health status: %s", pipelineerr) + } + + orborusStats.DataLake = pipelinePayload jsonData, err := json.Marshal(orborusStats) if err == nil { req.Body = ioutil.NopCloser(bytes.NewBuffer(jsonData)) @@ -2115,23 +2119,52 @@ func main() { var toBeRemoved shuffle.ExecutionRequestWrapper if len(executionRequests.Data) > 0 { newrequests := []shuffle.ExecutionRequest{} + + // Deduplicating in case same job shows up multiple times + // This is specifically to handle data pipelines better + deduplicatedJobs := []shuffle.ExecutionRequest{} for _, incRequest := range executionRequests.Data { + if !strings.Contains(incRequest.Type, "DOCKER") && !strings.Contains(incRequest.Type, "PIPELINE") && !strings.Contains(incRequest.Type, "SIGMA") && !strings.Contains(incRequest.Type, "TENZIR") { + deduplicatedJobs = append(deduplicatedJobs, incRequest) + continue + } + + found := false + for _, dedupRequest := range deduplicatedJobs { + if incRequest.ExecutionArgument == dedupRequest.ExecutionArgument && incRequest.Type == dedupRequest.Type { + found = true + break + } + } + + if found { + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + continue + } + + deduplicatedJobs = append(deduplicatedJobs, incRequest) + } + + executionRequests.Data = deduplicatedJobs + for _, incRequest := range executionRequests.Data { + // Looking for specific jobs if incRequest.Type == "PIPELINE_CREATE" || incRequest.Type == "PIPELINE_START" || incRequest.Type == "PIPELINE_STOP" || incRequest.Type == "PIPELINE_DELETE" { err := handlePipeline(incRequest) if err != nil { - log.Printf("[ERROR] Failed handling pipeline: %s", err) - } else { - toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + + log.Printf("[ERROR] Failed handling pipeline (%s %s): %s. Deleting job anyway.", incRequest.Type, incRequest.ExecutionSource, err) } + + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) } else if incRequest.Type == "DOCKER_IMAGE_DOWNLOAD" { log.Printf("[INFO] Should delete -> download new images: %#v", incRequest.ExecutionArgument) if len(incRequest.ExecutionArgument) > 0 { // FIXME: Wait X seconds before running this as the image build may not be done yet. This is shitty, but may be ok to do in Orborus. Easy fix for the future: Just let it run through jobs 5-10 times before actually picking it up - // Run after 25 seconds in the goroutine instead + // Run after 25 seconds in the goroutine instead go handleBackendImageDownload(ctx, incRequest.ExecutionArgument) } else { log.Printf("[ERROR] No image name provided for download. Removing job from queue.") @@ -2143,7 +2176,7 @@ func main() { err := deployTenzirNode() if err != nil { - log.Printf("[ERROR] Failed to deploy CATEGORY UPDATE, reason: %s", err) + log.Printf("[ERROR] Failed to run CATEGORY UPDATE, reason: %s", err) } else { continue } @@ -2159,7 +2192,7 @@ func main() { fileName := incRequest.ExecutionArgument err := deployTenzirNode() if err != nil { - log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FILE, reason: %s", err) + log.Printf("[ERROR] Failed to run DISABLE SIGMA FILE, reason: %s", err) } else { continue } @@ -2175,11 +2208,11 @@ func main() { fileName := incRequest.ExecutionArgument err := deployTenzirNode() if err != nil { - log.Printf("[ERROR] Failed to deploy ENABLE SIGMA FILE, reason: %s", err) + log.Printf("[ERROR] Failed to run ENABLE SIGMA FILE, reason: %s", err) } else { continue } - + err = enableRule(fileName) if err != nil { log.Printf("[ERROR] Failed to disable the sigma file %s, reason: %s", fileName, err) @@ -2190,7 +2223,7 @@ func main() { } else if incRequest.Type == "DISABLE_SIGMA_FOLDER" { err := deployTenzirNode() if err != nil { - log.Printf("[ERROR] Failed to deploy DISABLE SIGMA FOLDER, reason: %s", err) + log.Printf("[ERROR] Failed to run DISABLE SIGMA FOLDER, reason: %s", err) } err = removeAllFiles() @@ -2201,9 +2234,28 @@ func main() { } } else if incRequest.Type == "START_TENZIR" { log.Printf("[INFO] Got job to start tenzir") + err := deployTenzirNode() if err != nil { - log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err) + if strings.Contains(fmt.Sprintf("%s", err), "node available") { + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) + } else { + log.Printf("[ERROR] Failed to start tenzir, reason: %s", err) + err = shuffle.CreateOrgNotification( + ctx, + fmt.Sprintf("Failed to start Tenzir: %s", err), + fmt.Sprintf("Tenzir failed to start due to: %s", err), + fmt.Sprintf("/detections/Sigma"), + org, + true, + ) + + if err != nil { + log.Printf("[ERROR] Failed to send notification: %s", err) + return + } + } + } else { toBeRemoved.Data = append(toBeRemoved.Data, incRequest) } @@ -2557,6 +2609,8 @@ func main() { // 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 { + log.Printf("[INFO] Pipeline: %s to %s", incRequest.Type, incRequest.ExecutionSource) + err := deployTenzirNode() if err != nil { log.Printf("[ERROR] Failed to deploy the pipeline, reason: %s", err) @@ -2569,10 +2623,13 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { 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 + identifier := strings.ToLower(strings.ReplaceAll(incRequest.ExecutionSource, " ", "-")) + if !strings.HasPrefix(strings.ToLower(incRequest.ExecutionSource), "shuffle") { + 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 with id %#v", identifier) //err := deployPipeline(image, identifier, command) @@ -2581,23 +2638,28 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { log.Printf("[ERROR] Failed to create pipeline: %s", err) return err } - } else if incRequest.Type == "PIPELINE_DELETE" { + } else if incRequest.Type == "PIPELINE_DELETE" || incRequest.Type == "PIPELINE_STOP" { { 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) return err } + err = deletePipeline(pipelineId) if err != nil { log.Printf("[ERROR] Failed Deleting Pipeline %s", err) return err } + } + + /* } else if incRequest.Type == "PIPELINE_STOP" { log.Printf("[INFO] Should stop the pipeline %#v", identifier) pipelineId, err := searchPipeline(identifier) if err != nil { log.Printf("[ERROR] Failed searching for Pipeline with name %s reason:%s ", identifier, err) + toBeRemoved.Data = append(toBeRemoved.Data, incRequest) return err } _, err = updatePipelineState(command, pipelineId, "stop") @@ -2605,18 +2667,20 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { log.Printf("[ERROR] Failed to stop Pipeline: %s reason:%s ", pipelineId, err) return err } else { - log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId) + log.Printf("[INFO] Successfully stopped the Pipeline: %s", pipelineId) } + */ } 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) + 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 } @@ -2625,7 +2689,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error { log.Printf("[ERROR] Failed to start Pipeline: %s reason:%s ", pipelineId, err) return err } else { - log.Printf("[INFO] successfully started the Pipeline: %s", pipelineId) + log.Printf("[INFO] Successfully started the Pipeline: %s", pipelineId) } } else { @@ -2743,6 +2807,7 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri hostConfig := &container.HostConfig{ PortBindings: nat.PortMap{ + "514/udp": []nat.PortBinding{{HostPort: "514"}}, "5160/tcp": []nat.PortBinding{{HostPort: "5160"}}, }, Mounts: []mount.Mount{ @@ -2775,14 +2840,22 @@ func createAndStartTenzirNode(ctx context.Context, containerName, imageName stri 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 ...") + log.Printf("[INFO] Tenzir Node container started successfully. Waiting for it to become available..") + time.Sleep(10 * time.Second) err = checkTenzirNode() if err != nil { return err } - log.Printf("[INFO] Successfully deployed Tenzir Node!") + + log.Printf("[INFO] Successfully deployed Tenzir Node! Setting up default syslog listener on UDP 514") + + command := "from udp://0.0.0.0:514 read syslog | import" + _, err = createPipeline(command, "default-syslog-514") + if err != nil { + log.Printf("[ERROR] Failed to create default syslog pipeline: %s", err) + return nil + } return nil } @@ -2853,22 +2926,31 @@ func checkTenzirNode() error { func createPipeline(command, identifier string) (string, error) { - toBeDeleted := false - pipelineId, err := searchPipeline(identifier) + //toBeDeleted := false + /* + // Pre-checked. No point here + pipelineId, err := searchPipeline(identifier) + if err != nil { + return "", err + } + */ url := fmt.Sprintf("%s/api/v0/pipeline/create", pipelineUrl) forwardMethod := "POST" - if err != nil { - if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") { - log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier) + /* + if err != nil { + if strings.Contains(fmt.Sprintf("%s", err), "no existing pipeline found") { + log.Printf("[INFO] No existing pipeline found with id: %s. Creating a new one!", identifier) + } else { + log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err) + } } else { - log.Printf("[ERROR] Failed to search for existing pipeline but continuing anyway : %s", err) + log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId) + toBeDeleted = true } - } else { - log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId) - toBeDeleted = true - } + */ + // if strings.Contains(command, "shuffler.io") { // } else { @@ -2957,9 +3039,9 @@ func createPipeline(command, identifier string) (string, error) { id := response.ID - if toBeDeleted { - go deletePipeline(pipelineId) - } + //if toBeDeleted { + // go deletePipeline(pipelineId) + //} return id, nil } @@ -3071,44 +3153,48 @@ func deletePipeline(pipelineId string) error { return fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) } - log.Printf("[INFO] pipeline with ID: %s deleted successfully", pipelineId) + log.Printf("[INFO] Pipeline with ID: %s deleted successfully", pipelineId) + + pipelines = []shuffle.PipelineInfoMini{} return nil } -func searchPipeline(identifier string) (string, error) { - - type pipelineInfo struct { - ID string `json:"id"` - Name string `json:"name"` - } +// Lists the pipelines from the API exactly as they are. Definition is set up in Shuffle structs +func listPipelines() ([]shuffle.PipelineInfo, error) { + responseData := shuffle.PipelineInfoWrapper{} var reqBody []byte - url := fmt.Sprintf("%s/api/v0/pipeline/list", pipelineUrl) - resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody)) - if err != nil { - return "", err - } - defer resp.Body.Close() + if err != nil { + return responseData.Pipelines, err + } + + defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return "", fmt.Errorf("got the status code %d instead of 200", resp.StatusCode) + return responseData.Pipelines, fmt.Errorf("Got the status code %d instead of 200 from Pipeline node", resp.StatusCode) } body, err := ioutil.ReadAll(resp.Body) if err != nil { - return "", err + return responseData.Pipelines, err } - var responseData struct { - Pipelines []pipelineInfo `json:"pipelines"` - } if err := json.Unmarshal(body, &responseData); err != nil { + return responseData.Pipelines, err + } + + return responseData.Pipelines, nil +} + +func searchPipeline(identifier string) (string, error) { + allPipelines, err := listPipelines() + if err != nil { return "", err } - for _, pipeline := range responseData.Pipelines { + for _, pipeline := range allPipelines { if pipeline.Name == identifier { return pipeline.ID, nil } @@ -3317,72 +3403,44 @@ func removePath(containerName, path string) error { return nil } -func sendTenzirHealthStatus() error { - // Check one in every 10 times only - randint := rand.Intn(10) - _ = randint - //if randint != 0 { - // return nil - //} +func sendPipelineHealthStatus() (shuffle.LakeConfig, error) { + pipelinePayload := shuffle.LakeConfig{ + Enabled: false, + Pipelines: []shuffle.PipelineInfoMini{}, + } - var status string - url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl) + // To not spam down the list API too much + randint := rand.Intn(5) + if len(pipelines) == 0 || randint == 0 { + pipelineDef, err := listPipelines() + + if err == nil { + for _, pipeline := range pipelineDef { + pipelinePayload.Pipelines = append(pipelinePayload.Pipelines, shuffle.PipelineInfoMini{ + ID: pipeline.ID, + Name: pipeline.Name, + Definition: pipeline.Definition, + TotalRuns: pipeline.TotalRuns, + CreatedAt: pipeline.CreatedAt, + }) + } + + pipelines = pipelinePayload.Pipelines + } + } else { + pipelinePayload.Pipelines = pipelines + } + + //url := fmt.Sprintf("%s/api/v1/detections/siem/health", baseUrl) err := checkTenzirNode() if err != nil { - return err - } else { - status = "active" + return pipelinePayload, err } - //log.Printf("[DEBUG] Sending Tenzir health update to backend url '%s'", baseUrl) - forwardMethod := "POST" - payload := map[string]interface{}{ - "status": status, - "environment": environment, - "authorization": "", - "org_id": "", + pipelinePayload.Enabled = true - } - - if len(auth) > 0 { - payload["authorization"] = auth - } - - if len(org) > 0 { - payload["org_id"] = org - } - - payloadBytes, err := json.Marshal(payload) - if err != nil { - log.Printf("[ERROR] Failed to marshal payload: %s", err) - return err - } - 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") - 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] Pipeline: status for URL %s: %d", url, resp.StatusCode) - return fmt.Errorf("unexpected HTTP status code: %d", resp.StatusCode) - } - - return nil + // No direct sending. + return pipelinePayload, nil } func disableRule(fileName string) error { @@ -3438,7 +3496,7 @@ func enableRule(fileName string) error { return fmt.Errorf("error moving file: %v", err) } - fmt.Printf("File %s moved to %s successfully.\n", fileName, destDir) + fmt.Printf("[DEBUG] File %s moved to %s successfully.\n", fileName, destDir) return nil } @@ -3575,7 +3633,7 @@ func zombiecheck(ctx context.Context, workerTimeout int) error { // Check image name if !shuffleFound { - log.Printf("[WARNING] Zombie container skip: %#v, %s", container.Labels, container.Image) + //log.Printf("[WARNING] Zombie container skip: %#v, %s", container.Labels, container.Image) continue } //} else { @@ -3675,7 +3733,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string, if strings.Contains(streamUrl, "shuffler.io") || strings.Contains(streamUrl, "localhost") || strings.Contains(streamUrl, "127.0.0.1") || strings.Contains(streamUrl, "shuffle-backend") { - // Specific to debugging + // Specific to debugging if len(workerServerUrl) == 0 { log.Printf("[INFO] Using default worker server url as previous is invalid: %s", streamUrl) }