Added a cache VQL plugin which stores data for a time. (Velocidex#633)

The cache function produces a scope lifetime cache. It takes 2
parameters, a func and a key. The key is used to index into the cache,
while the func will be evaluated when the cache misses.

The following is an example of how to use the cache:

```
  LET Foo(X) = if(condition=log(message=format(format="I Ran with %v", args=X)),
                    then=X + 5)
  SELECT cache(func=Foo(X=5), key=5),
           cache(func=Foo(X=10), key=5),
           cache(func=Foo(X=10), key=10)
  FROM scope()
```

This will produce a row with (10, 10, 15) - the middle item is cached
because its key is the same as the first item.

api/api.go

@@ -189,8 +189,12 @@ func (self *ApiServer) CollectArtifact(
 	if err != nil {
 		return nil, err
 	}
+	launcher, err := services.GetLauncher()
+	if err != nil {
+		return nil, err
+	}
 
-	flow_id, err := services.GetLauncher().ScheduleArtifactCollection(
+	flow_id, err := launcher.ScheduleArtifactCollection(
 		ctx, self.config, acl_manager, repository, in)
 	if err != nil {
 		return nil, err
@@ -813,7 +817,12 @@ func (self *ApiServer) WriteEvent(
 
 		// Only return the first row
 		if true {
-			err := services.GetJournal().PushRowsToArtifact(self.config,
+			journal, err := services.GetJournal()
+			if err != nil {
+				return nil, err
+			}
+
+			err = journal.PushRowsToArtifact(self.config,
 				rows, in.Query.Name, peer_name, "")
 			return &empty.Empty{}, err
 		}

artifacts/definitions/Generic/System/Pstree.yaml

@@ -24,8 +24,23 @@ sources:
       // through it many times.
       LET processes <= SELECT Name, Pid, Ppid FROM pslist()
 
-      // Recursive function to find process parent.
-      LET pstree(LookupPid) = SELECT * FROM foreach(
+      LET m <= memoize(query={ SELECT Name, Pid, Ppid FROM pslist()}, key="Pid", period=1)
+
+      // Recursive function to find process parent (clients > 0.4.9).
+      LET pstree_memoized(LookupPid) = SELECT * FROM foreach(
+      row=get(item=m, field=LookupPid),
+        query={
+           SELECT * FROM chain(
+           a={
+             SELECT Name, Pid, Ppid FROM scope()
+           },
+           b={
+             SELECT Name, Pid, Ppid FROM pstree(LookupPid=Ppid)
+           })
+        })
+
+      // Recursive function to find process parent (clients < 0.4.9).
+      LET pstree_legacy(LookupPid) = SELECT * FROM foreach(
          row={
            SELECT Name, Pid, Ppid FROM processes
            WHERE Pid = LookupPid
@@ -40,6 +55,12 @@ sources:
            })
       })
 
+      // Select which version we should use
+      LET pstree(LookupPid) = SELECT * FROM if(
+        condition=version(function="memoize") >= 0,
+        then=pstree_memoized,
+        else=pstree_legacy)
+
       SELECT Name, Pid, Ppid,
              join(array=pstree(LookupPid=Pid).Name, sep=CallChainSep) AS CallChain
       FROM processes

artifacts/definitions/Windows/Events/ProcessCreation.yaml

@@ -13,18 +13,38 @@ parameters:
 sources:
   - precondition:
       SELECT OS From info() where OS = 'windows'
-    queries:
-      - |
+    query: |
+        // Get information about the process
+        LET get_pid_query(Lpid) = SELECT Pid, Ppid, Name FROM if(condition=Lpid > 0,
+        then={
+            SELECT Pid, Ppid, Name FROM pslist(pid=Lpid)
+        })
+
+        // Build the call chain. Cache the results for a short time.
+        LET pstree(LookupPid) = SELECT * FROM foreach(
+          row=cache(func=get_pid_query(Lpid=LookupPid), key=str(str=LookupPid)),
+          query={
+              SELECT * FROM chain(
+            a={
+              SELECT Pid, Ppid, Name FROM scope()
+            }, b={
+              SELECT Pid, Ppid, Name FROM pstree(LookupPid=Ppid)
+            })
+        })
+
+        LET call_chain(LookupPid) = SELECT Pid, Ppid, Name FROM pstree(LookupPid=LookupPid)
+
         // Convert the timestamp from WinFileTime to Epoch.
-        SELECT timestamp(epoch=atoi(string=Parse.TIME_CREATED) / 10000000 - 11644473600 ) as Timestamp,
+        SELECT timestamp(winfiletime=atoi(string=Parse.TIME_CREATED)) as Timestamp,
                Parse.ParentProcessID as PPID,
                Parse.ProcessID as PID,
                Parse.ProcessName as Name, {
                  SELECT CommandLine FROM pslist(pid=Parse.ProcessID)
                } AS CommandLine,
                {
                  SELECT CommandLine FROM pslist(pid=Parse.ParentProcessID)
-               } AS ParentInfo
+               } AS ParentInfo,
+               join(array=call_chain(LookupPid=Parse.ProcessID).Name, sep=" <- ") AS CallChain
         FROM wmi_events(
            query=eventQuery,
            wait=5000000,   // Do not time out.

artifacts/testdata/server/testcases/functions.in.yaml

@@ -15,3 +15,11 @@ Queries:
            WHERE _value = 3
        })
     }) AS Filtered FROM scope()
+
+  # Test cache functions (first 2 should be same value due to caching)
+  - LET Foo(X) = if(condition=log(message=format(format="I Ran with %v", args=X)),
+                    then=X + 5)
+  - SELECT cache(func=Foo(X=5), key=5),
+           cache(func=Foo(X=10), key=5),
+           cache(func=Foo(X=10), key=10)
+    FROM scope()

artifacts/testdata/server/testcases/functions.out.yaml

@@ -11,4 +11,10 @@ LET rows <= SELECT * FROM scope()[]SELECT len(list=["a", "b"]), len(list="hello"
    "Z": 3
   }
  }
+]LET Foo(X) = if(condition=log(message=format(format="I Ran with %v", args=X)), then=X + 5)[]SELECT cache(func=Foo(X=5), key=5), cache(func=Foo(X=10), key=5), cache(func=Foo(X=10), key=10) FROM scope()[
+ {
+  "cache(func=Foo(X=5), key=5)": 10,
+  "cache(func=Foo(X=10), key=5)": 10,
+  "cache(func=Foo(X=10), key=10)": 15
+ }
 ]

bin/artifacts.go

@@ -242,7 +242,10 @@ func doArtifactList() {
 			continue
 		}
 
-		request, err := services.GetLauncher().CompileCollectorArgs(
+		launcher, err := services.GetLauncher()
+		kingpin.FatalIfError(err, "GetLauncher")
+
+		request, err := launcher.CompileCollectorArgs(
 			ctx, config_obj, vql_subsystem.NullACLManager{}, repository,
 			&flows_proto.ArtifactCollectorArgs{
 				Artifacts: []string{artifact.Name},

bin/golden.go

@@ -110,14 +110,16 @@ func makeCtxWithTimeout(duration int) (context.Context, func()) {
 
 				p := pprof.Lookup("goroutines")
 				if p != nil {
-					p.WriteTo(os.Stderr, 1)
+					p.WriteTo(os.Stdout, 1)
 				}
 
 				p = pprof.Lookup("mutex")
 				if p != nil {
-					p.WriteTo(os.Stderr, 1)
+					p.WriteTo(os.Stdout, 1)
 				}
 
+				os.Stdout.Close()
+
 				// Hard exit with an error.
 				os.Exit(-1)
 			}

flows/api.go

@@ -266,9 +266,14 @@ func ArchiveFlow(
 		Set("Timestamp", time.Now().UTC().Unix()).
 		Set("Flow", collection_context)
 
+	journal, err := services.GetJournal()
+	if err != nil {
+		return nil, err
+	}
+
 	return &api_proto.StartFlowResponse{
 			FlowId: flow_id,
-		}, services.GetJournal().PushRowsToArtifact(config_obj,
+		}, journal.PushRowsToArtifact(config_obj,
 			[]*ordereddict.Dict{row},
 			"System.Flow.Archive", client_id, flow_id)
 }

flows/artifacts.go

@@ -125,7 +125,12 @@ func closeContext(
 			Set("FlowId", collection_context.SessionId).
 			Set("ClientId", collection_context.ClientId)
 
-		return services.GetJournal().PushRowsToArtifact(config_obj,
+		journal, err := services.GetJournal()
+		if err != nil {
+			return err
+		}
+
+		return journal.PushRowsToArtifact(config_obj,
 			[]*ordereddict.Dict{row},
 			"System.Flow.Completion", collection_context.ClientId,
 			collection_context.SessionId,
@@ -540,7 +545,12 @@ func appendUploadDataToFile(
 			Set("Size", file_buffer.Size).
 			Set("UploadedSize", file_buffer.StoredSize)
 
-		return services.GetJournal().PushRowsToArtifact(config_obj,
+		journal, err := services.GetJournal()
+		if err != nil {
+			return err
+		}
+
+		return journal.PushRowsToArtifact(config_obj,
 			[]*ordereddict.Dict{row},
 			"System.Upload.Completion", message.Source, collection_context.SessionId,
 		)

flows/artifacts_test.go

@@ -94,7 +94,10 @@ func (self *TestSuite) TestRetransmission() {
 
 	// Schedule a new flow.
 	ctx := context.Background()
-	flow_id, err := services.GetLauncher().ScheduleArtifactCollection(
+	launcher, err := services.GetLauncher()
+	assert.NoError(self.T(), err)
+
+	flow_id, err := launcher.ScheduleArtifactCollection(
 		ctx, self.config_obj,
 		vql_subsystem.NullACLManager{},
 		repository, request)
@@ -153,7 +156,10 @@ func (self *TestSuite) TestResourceLimits() {
 
 	// Schedule a new flow.
 	ctx := context.Background()
-	flow_id, err := services.GetLauncher().ScheduleArtifactCollection(
+	launcher, err := services.GetLauncher()
+	assert.NoError(self.T(), err)
+
+	flow_id, err := launcher.ScheduleArtifactCollection(
 		ctx,
 		self.config_obj,
 		vql_subsystem.NullACLManager{},

flows/foreman.go

@@ -99,8 +99,13 @@ func ForemanProcessMessage(
 			return nil
 		}
 
+		journal, err := services.GetJournal()
+		if err != nil {
+			return err
+		}
+
 		// Notify the hunt manager that we need to hunt this client.
-		err := services.GetJournal().PushRowsToArtifact(config_obj,
+		err = journal.PushRowsToArtifact(config_obj,
 			[]*ordereddict.Dict{ordereddict.NewDict().
 				Set("HuntId", hunt.HuntId).
 				Set("ClientId", client_id).

flows/hunts.go

@@ -126,9 +126,13 @@ func CreateHunt(
 	// time. This ensures that if the artifact definition is
 	// changed after this point, the hunt will continue to
 	// schedule consistent VQL on the clients.
-	compiled, err := services.GetLauncher().
-		CompileCollectorArgs(
-			ctx, config_obj, acl_manager, repository, hunt.StartRequest)
+	launcher, err := services.GetLauncher()
+	if err != nil {
+		return "", err
+	}
+
+	compiled, err := launcher.CompileCollectorArgs(
+		ctx, config_obj, acl_manager, repository, hunt.StartRequest)
 	if err != nil {
 		return "", err
 	}
@@ -272,7 +276,12 @@ func ModifyHunt(
 					Set("Hunt", hunt).
 					Set("User", user)
 
-				err := services.GetJournal().PushRowsToArtifact(config_obj,
+				journal, err := services.GetJournal()
+				if err != nil {
+					return err
+				}
+
+				err = journal.PushRowsToArtifact(config_obj,
 					[]*ordereddict.Dict{row}, "System.Hunt.Archive",
 					"server", hunt_modification.HuntId)
 				if err != nil {

flows/monitoring.go

@@ -54,7 +54,12 @@ func MonitoringProcessMessage(
 		for _, row := range rows {
 			row.Set("ClientId", message.Source)
 		}
-		return services.GetJournal().PushRowsToArtifact(
+		journal, err := services.GetJournal()
+		if err != nil {
+			return err
+		}
+
+		return journal.PushRowsToArtifact(
 			config_obj, rows, response.Query.Name,
 			message.Source, message.SessionId)
 

go.mod

@@ -114,7 +114,7 @@ require (
 	www.velocidex.com/golang/go-prefetch v0.0.0-20200722101157-37e4751dd5ca
 	www.velocidex.com/golang/oleparse v0.0.0-20190327031422-34195d413196
 	www.velocidex.com/golang/regparser v0.0.0-20190625082115-b02dc43c2500
-	www.velocidex.com/golang/vfilter v0.0.0-20200903172553-c41828760309
+	www.velocidex.com/golang/vfilter v0.0.0-20200911062452-cf0fe8a4de78
 	www.velocidex.com/golang/vtypes v0.0.0-20180924145839-b0d509f8925b
 )
 

go.sum

@@ -70,7 +70,6 @@ github.com/Velocidex/go-yara v1.1.10-0.20200414034554-457848df11f9 h1:zatH9PNN7I
 github.com/Velocidex/go-yara v1.1.10-0.20200414034554-457848df11f9/go.mod h1:N1A2MzBKorQm3WixuPUSm4gmzGA6i5sYrwHyUBvY5lg=
 github.com/Velocidex/json v0.0.0-20200724131328-8f5c7b0a25ec h1:ep1GlQfQtVZsn8yaUwILZxlK1Sw13dYhsOT3bEhZ8HE=
 github.com/Velocidex/json v0.0.0-20200724131328-8f5c7b0a25ec/go.mod h1:ukJBuruT9b24pdgZwWDvOaCYHeS03B7oQPCUWh25bwM=
-github.com/Velocidex/ordereddict v0.0.0-20191103011020-3b5a5f6957d4/go.mod h1:pxJpvN5ISMtDwrdIdqnJ3ZrjIngCw+WT6gfNil6Zjvo=
 github.com/Velocidex/ordereddict v0.0.0-20191106020901-97c468e5e403/go.mod h1:pxJpvN5ISMtDwrdIdqnJ3ZrjIngCw+WT6gfNil6Zjvo=
 github.com/Velocidex/ordereddict v0.0.0-20200723153557-9460a6764ab8 h1:wRt5CLsj5hS3OpJjQYJMuUrmxKwRfUn76+4sqjfmNbA=
 github.com/Velocidex/ordereddict v0.0.0-20200723153557-9460a6764ab8/go.mod h1:pxJpvN5ISMtDwrdIdqnJ3ZrjIngCw+WT6gfNil6Zjvo=
@@ -762,9 +761,7 @@ www.velocidex.com/golang/oleparse v0.0.0-20190327031422-34195d413196 h1:3oYZ7hPN
 www.velocidex.com/golang/oleparse v0.0.0-20190327031422-34195d413196/go.mod h1:i7M+d4Vxir8nmDACh+c6CsUU1r1Wcj00aRgNp8mXcPQ=
 www.velocidex.com/golang/regparser v0.0.0-20190625082115-b02dc43c2500 h1:XqZddiAbjPIsTZcEPbqqqABS/ZV5SB7j33eczNsqD60=
 www.velocidex.com/golang/regparser v0.0.0-20190625082115-b02dc43c2500/go.mod h1:DVzloLH8L+oF3zma1Jisaat5bGF+4VLggDcYlIp00ns=
-www.velocidex.com/golang/vfilter v0.0.0-20200903081613-6b20dc963172 h1:tmKvMO1zsjMfThcdyqWXoHkoOdvfY/iHhzaa93GnBOw=
-www.velocidex.com/golang/vfilter v0.0.0-20200903081613-6b20dc963172/go.mod h1:mABF6rGkfq9qwvo2SppBxYonhnd8OPSA6rxzzl75A4Y=
-www.velocidex.com/golang/vfilter v0.0.0-20200903172553-c41828760309 h1:mq63LAXMGM41TZAm9YTkC+N8ic+U+kUkvJDSJwj6OHc=
-www.velocidex.com/golang/vfilter v0.0.0-20200903172553-c41828760309/go.mod h1:XlUeViBwZxeefhxbkxW2oGUVcB/oQfxtBgnxL9jLryg=
+www.velocidex.com/golang/vfilter v0.0.0-20200911062452-cf0fe8a4de78 h1:9M+BvRRVYm4TCpA4UVGI/bf5vBhsGHnzYklH2OgigtI=
+www.velocidex.com/golang/vfilter v0.0.0-20200911062452-cf0fe8a4de78/go.mod h1:XlUeViBwZxeefhxbkxW2oGUVcB/oQfxtBgnxL9jLryg=
 www.velocidex.com/golang/vtypes v0.0.0-20180924145839-b0d509f8925b h1:z5v5o1dhtzaxvlWm6qSTYZ4OTr56Ol2JpM1Y5Wu9zQE=
 www.velocidex.com/golang/vtypes v0.0.0-20180924145839-b0d509f8925b/go.mod h1:tXxIx8UJuI81Hoxcv0DTq2a1Pi1H6l1uCf4dhqUSUkw=

gui/static/angular-components/artifact/linechart-directive.js

@@ -76,7 +76,7 @@ exports.LineChartDirective = function() {
           elem = $(elem);
 
           var plot = null;
-          var placeholder = $("<div>").appendTo(elem);
+          var placeholder = $("<div id='plot'>").appendTo(elem);
           placeholder.bind("plotselected", function (event, ranges) {
               $.each(plot.getXAxes(), function(_, axis) {
                   var opts = axis.options;

gui/static/angular-components/core/api-service.js

@@ -288,7 +288,10 @@ ApiService.prototype.poll = function(apiPath, intervalMs, opt_params,
       if (cancelled) {
         return;
       }
-      result.reject(response);
+        // Do not cancel the polling if there is a server
+        // error. Polling normally updates the display and
+        // intermittant failures will cause the display to break.
+        // result.reject(response);
     }.bind(this)).finally(function() {
       if (cancelled) {
         return;

gui/static/package.json

@@ -20,7 +20,7 @@
     "datatables.net-buttons": "^1.6.3",
     "datatables.net-colreorder": "^1.5.2",
     "del": "^5.1.0",
-    "flot": "^4.2.1",
+    "flot": "^3.2.10",
     "font-awesome": "^4.7.0",
     "gulp-cli": "^2.3.0",
     "gulp-uglify": "^3.0.2",

result_sets/result_sets.go

@@ -16,6 +16,7 @@
 package result_sets
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/Velocidex/json"
@@ -29,7 +30,11 @@ import (
 )
 
 func GetArtifactMode(config_obj *config_proto.Config, artifact_name string) (int, error) {
-	repository, _ := services.GetRepositoryManager().GetGlobalRepository(config_obj)
+	manager := services.GetRepositoryManager()
+	if manager == nil {
+		return 0, errors.New("No Repository Manager")
+	}
+	repository, _ := manager.GetGlobalRepository(config_obj)
 
 	artifact, pres := repository.Get(config_obj, artifact_name)
 	if !pres {

server/enroll.go

@@ -44,7 +44,12 @@ func enroll(
 		return err
 	}
 
-	return services.GetJournal().PushRowsToArtifact(config_obj,
+	journal, err := services.GetJournal()
+	if err != nil {
+		return err
+	}
+
+	return journal.PushRowsToArtifact(config_obj,
 		[]*ordereddict.Dict{ordereddict.NewDict().Set("ClientId", client_id)},
 		"Server.Internal.Enrollment", "server" /* client_id */, "",
 	)