Bugfix: Hunt dispatcher did not expire hunts

services/hunt_dispatcher/hunt_dispatcher.go

@@ -194,8 +194,6 @@ func (self *HuntDispatcher) ApplyFuncOnHunts(
 	ctx context.Context, options services.HuntSearchOptions,
 	cb func(hunt *api_proto.Hunt) error) (res_error error) {
 
-	now := uint64(utils.GetTime().Now().UnixNano() / 1000)
-
 	// Page through the hunts table and apply the function on each
 	// page.
 	var offset, length int64
@@ -209,7 +207,7 @@ func (self *HuntDispatcher) ApplyFuncOnHunts(
 
 		for _, hunt := range hunts {
 			if options == services.OnlyRunningHunts &&
-				(hunt.State != api_proto.Hunt_RUNNING || now > hunt.Expires) {
+				hunt.State != api_proto.Hunt_RUNNING {
 				continue
 			}
 
@@ -269,7 +267,6 @@ func (self *HuntDispatcher) MutateHunt(
 			Set("hunt_id", mutation.HuntId).
 			Set("mutation", mutation),
 		"Server.Internal.HuntModification")
-
 	return nil
 }
 
@@ -549,7 +546,7 @@ func NewHuntDispatcher(
 
 			case <-time.After(time.Duration(refresh) * time.Second):
 				// Re-read the hunts from the data store.
-				err := service.Store.Refresh(ctx, config_obj)
+				err := service.Refresh(ctx, config_obj)
 				if err != nil {
 					logger.Error("Unable to sync hunts: %v", err)
 				}

services/hunt_dispatcher/hunt_dispatcher_test.go

@@ -11,10 +11,13 @@ import (
 	api_proto "www.velocidex.com/golang/velociraptor/api/proto"
 	"www.velocidex.com/golang/velociraptor/datastore"
 	"www.velocidex.com/golang/velociraptor/file_store/test_utils"
+	flows_proto "www.velocidex.com/golang/velociraptor/flows/proto"
+	"www.velocidex.com/golang/velociraptor/json"
 	"www.velocidex.com/golang/velociraptor/paths"
 	"www.velocidex.com/golang/velociraptor/services"
 	"www.velocidex.com/golang/velociraptor/services/hunt_dispatcher"
 	"www.velocidex.com/golang/velociraptor/utils"
+	"www.velocidex.com/golang/velociraptor/vql/acl_managers"
 	"www.velocidex.com/golang/velociraptor/vtesting"
 	"www.velocidex.com/golang/velociraptor/vtesting/assert"
 
@@ -36,6 +39,10 @@ func (self *HuntDispatcherTestSuite) SetupTest() {
 	self.ConfigObj = self.TestSuite.LoadConfig()
 	self.ConfigObj.Services.FrontendServer = true
 	self.ConfigObj.Services.HuntDispatcher = true
+	self.ConfigObj.Services.HuntManager = true
+	self.ConfigObj.Services.RepositoryManager = true
+	self.ConfigObj.Services.JournalService = true
+	self.ConfigObj.Defaults.HuntDispatcherRefreshSec = 1
 
 	self.LoadArtifactsIntoConfig([]string{`
 name: Server.Internal.HuntUpdate
@@ -196,6 +203,54 @@ func (self *HuntDispatcherTestSuite) getAllHunts() []*api_proto.Hunt {
 	return hunts
 }
 
+func (self *HuntDispatcherTestSuite) TestExpiringHunts() {
+	closer := utils.MockTime(utils.RealClock{})
+	defer closer()
+
+	hunt_id := "H.121222"
+
+	now := utils.GetTime().Now().Unix()
+	json.Dump(now)
+	acl_manager := acl_managers.NullACLManager{}
+
+	hunt_obj, err := self.master_dispatcher.CreateHunt(self.Ctx, self.ConfigObj,
+		acl_manager, &api_proto.Hunt{
+			HuntId:    hunt_id,
+			State:     api_proto.Hunt_RUNNING,
+			Version:   now,
+			StartTime: uint64(now),
+			Expires:   uint64(now+1) * 1000000,
+			StartRequest: &flows_proto.ArtifactCollectorArgs{
+				Artifacts: []string{"Generic.Client.Info"},
+			},
+		})
+	assert.NoError(self.T(), err)
+	assert.Equal(self.T(), hunt_obj.State, api_proto.Hunt_RUNNING)
+
+	// Fast forward the time
+	closer = utils.MockTime(utils.RealClockWithOffset{Duration: 600 * time.Second})
+	defer closer()
+
+	vtesting.WaitUntil(500*time.Second, self.T(), func() bool {
+		hunt_obj, pres := self.master_dispatcher.GetHunt(self.Ctx, hunt_id)
+		assert.True(self.T(), pres)
+
+		return hunt_obj.State == api_proto.Hunt_STOPPED
+	})
+
+	// Delete the new hunt
+	self.master_dispatcher.MutateHunt(self.Ctx, self.ConfigObj,
+		&api_proto.HuntMutation{
+			HuntId: hunt_id,
+			State:  api_proto.Hunt_ARCHIVED,
+		})
+
+	vtesting.WaitUntil(500*time.Second, self.T(), func() bool {
+		_, pres := self.master_dispatcher.GetHunt(self.Ctx, hunt_id)
+		return pres == false
+	})
+}
+
 func TestHuntDispatcherTestSuite(t *testing.T) {
 	suite.Run(t, &HuntDispatcherTestSuite{
 		hunt_id: "H.1234",

services/hunt_dispatcher/storage.go

@@ -197,6 +197,18 @@ func (self *HuntStorageManagerImpl) SetHunt(
 		return utils.InvalidArgError
 	}
 
+	db, err := datastore.GetDB(self.config_obj)
+	if err != nil {
+		return err
+	}
+
+	hunt_path_manager := paths.NewHuntPathManager(hunt.HuntId)
+
+	if hunt.State == api_proto.Hunt_ARCHIVED {
+		delete(self.hunts, hunt.HuntId)
+		return db.DeleteSubject(self.config_obj, hunt_path_manager.Path())
+	}
+
 	// The hunts start time could have been modified - we need to
 	// update ours then (and also the metrics).
 	if hunt.StartTime > self.GetLastTimestamp() {
@@ -210,12 +222,6 @@ func (self *HuntStorageManagerImpl) SetHunt(
 	}
 	self.dirty = true
 
-	db, err := datastore.GetDB(self.config_obj)
-	if err != nil {
-		return err
-	}
-
-	hunt_path_manager := paths.NewHuntPathManager(hunt.HuntId)
 	return db.SetSubject(self.config_obj, hunt_path_manager.Path(), hunt)
 }
 

utils/clock.go

@@ -53,6 +53,24 @@ func (self RealClock) Now() time.Time {
 	return time.Now()
 }
 
+// A clock that behaves like a real one with sleeps but can be moved
+// forward arbitrarily.
+type RealClockWithOffset struct {
+	Duration time.Duration
+}
+
+func (self RealClockWithOffset) Sleep(d time.Duration) {
+	time.Sleep(d)
+}
+
+func (self RealClockWithOffset) After(d time.Duration) <-chan time.Time {
+	return time.After(d)
+}
+
+func (self RealClockWithOffset) Now() time.Time {
+	return time.Now().Add(self.Duration)
+}
+
 type MockClock struct {
 	mu      sync.Mutex
 	mockNow time.Time