Merge branch 'master' into snyk-upgrade-68da608d7a7b848ba032beb278694e6b

gui/velociraptor/package.json

@@ -13,7 +13,7 @@
         "@fortawesome/react-fontawesome": "0.2.2",
         "@popperjs/core": "^2.11.8",
         "axios": ">=1.6.8",
-        "ace-builds": "1.33.3",
+        "ace-builds": "1.34.0",
         "axios-retry": "3.9.1",
         "bootstrap": "^4.6.2",
         "classnames": "^2.5.1",

services/launcher.go

@@ -70,6 +70,7 @@ var (
 )
 
 type DeleteFlowResponse struct {
+	Id    int               `json:"-"`
 	Type  string            `json:"type"`
 	Data  *ordereddict.Dict `json:"data"`
 	Error string            `json:"error"`

services/launcher/delete.go

@@ -4,9 +4,12 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"sort"
+	"sync"
 	"time"
 
 	"github.com/Velocidex/ordereddict"
+	"github.com/alitto/pond"
 	config_proto "www.velocidex.com/golang/velociraptor/config/proto"
 	"www.velocidex.com/golang/velociraptor/datastore"
 	"www.velocidex.com/golang/velociraptor/file_store"
@@ -54,11 +57,13 @@ func (self *FlowStorageManager) DeleteFlow(
 	flow_path_manager := paths.NewFlowPathManager(client_id, flow_id)
 
 	upload_metadata_path := flow_path_manager.UploadMetadata()
+
 	r := &reporter{
 		really_do_it: really_do_it,
 		ctx:          ctx,
 		config_obj:   config_obj,
 		seen:         make(map[string]bool),
+		pool:         pond.New(100, 1000),
 	}
 	file_store_factory := file_store.GetFileStore(config_obj)
 	reader, err := result_sets.NewResultSetReader(
@@ -145,13 +150,18 @@ func (self *FlowStorageManager) DeleteFlow(
 			r.emit_fs("NotebookItem", path)
 			return nil
 		})
-
 	// Rebuild the flow index to ensure GUI paging works
 	// properly. This is pretty slow but we do not expect to delete
 	// flows that often.
 	if really_do_it {
 		err = self.buildFlowIndexFromLegacy(ctx, config_obj, client_id)
 	}
+	r.pool.StopAndWait()
+
+	// Sort responses to keep output stable
+	sort.Slice(r.responses, func(i, j int) bool {
+		return r.responses[i].Id < r.responses[j].Id
+	})
 
 	return r.responses, err
 }
@@ -162,65 +172,95 @@ type reporter struct {
 	seen         map[string]bool
 	config_obj   *config_proto.Config
 	really_do_it bool
+	mu           sync.Mutex
+	id           int
+	pool         *pond.WorkerPool
 }
 
 func (self *reporter) emit_ds(
 	item_type string, target api.DSPathSpec) {
+
 	client_path := target.String()
 	var error_message string
 
+	self.mu.Lock()
+	defer self.mu.Unlock()
+
 	if self.seen[client_path] {
 		return
 	}
 	self.seen[client_path] = true
 
-	if self.really_do_it {
-		db, err := datastore.GetDB(self.config_obj)
-		if err == nil {
-			err = db.DeleteSubject(self.config_obj, target)
-			if err != nil {
-				error_message = fmt.Sprintf(
-					"Error deleting %v: %v", client_path, err)
+	id := self.id
+	self.id++
+
+	self.pool.Submit(func() {
+		self.mu.Lock()
+		defer self.mu.Unlock()
+
+		if self.really_do_it {
+			db, err := datastore.GetDB(self.config_obj)
+			if err == nil {
+				err = db.DeleteSubject(self.config_obj, target)
+				if err != nil {
+					error_message = fmt.Sprintf(
+						"Error deleting %v: %v", client_path, err)
+				}
 			}
 		}
-	}
 
-	self.responses = append(self.responses, &services.DeleteFlowResponse{
-		Type:  item_type,
-		Data:  ordereddict.NewDict().Set("VFSPath", client_path),
-		Error: error_message,
+		self.responses = append(self.responses, &services.DeleteFlowResponse{
+			Id:    id,
+			Type:  item_type,
+			Data:  ordereddict.NewDict().Set("VFSPath", client_path),
+			Error: error_message,
+		})
 	})
+
 }
 
 func (self *reporter) emit_fs(
 	item_type string, target api.FSPathSpec) {
 	client_path := target.String()
 	var error_message string
 
+	self.mu.Lock()
+	defer self.mu.Unlock()
+
 	if self.seen[client_path] {
 		return
 	}
 	self.seen[client_path] = true
 
-	if self.really_do_it {
-		file_store_factory := file_store.GetFileStore(self.config_obj)
-		err := file_store_factory.Delete(target)
-		if err != nil {
-			error_message = fmt.Sprintf(
-				"Error deleting %v: %v", client_path, err)
+	self.id++
+	id := self.id
+
+	self.pool.Submit(func() {
+		self.mu.Lock()
+		defer self.mu.Unlock()
+
+		if self.really_do_it {
+			file_store_factory := file_store.GetFileStore(self.config_obj)
+			err := file_store_factory.Delete(target)
+			if err != nil {
+				error_message = fmt.Sprintf(
+					"Error deleting %v: %v", client_path, err)
+			}
 		}
-	}
 
-	self.responses = append(self.responses, &services.DeleteFlowResponse{
-		Type:  item_type,
-		Data:  ordereddict.NewDict().Set("VFSPath", client_path),
-		Error: error_message,
+		self.responses = append(self.responses, &services.DeleteFlowResponse{
+			Id:    id,
+			Type:  item_type,
+			Data:  ordereddict.NewDict().Set("VFSPath", client_path),
+			Error: error_message,
+		})
 	})
 }
 
-/* For now we do not bisect the event log files - we just remove the
-   entire file if the time stamp requested is in it. Since files are
-   split by day this will remove the entire day's worth of data.
+/*
+For now we do not bisect the event log files - we just remove the
+entire file if the time stamp requested is in it. Since files are
+split by day this will remove the entire day's worth of data.
 */
 func (self *Launcher) DeleteEvents(
 	ctx context.Context,