support for `cancel_on` (#1105)

* cancel_on filter

* tests

cmd/crowdsec/metrics.go

@@ -82,7 +82,7 @@ func registerPrometheus(config *csconfig.PrometheusCfg) {
 		log.Infof("Loading aggregated prometheus collectors")
 		prometheus.MustRegister(globalParserHits, globalParserHitsOk, globalParserHitsKo,
 			globalCsInfo,
-			leaky.BucketsUnderflow, leaky.BucketsInstanciation, leaky.BucketsOverflow,
+			leaky.BucketsUnderflow, leaky.BucketsCanceled, leaky.BucketsInstanciation, leaky.BucketsOverflow,
 			v1.LapiRouteHits,
 			leaky.BucketsCurrentCount)
 	} else {
@@ -91,7 +91,7 @@ func registerPrometheus(config *csconfig.PrometheusCfg) {
 			parser.NodesHits, parser.NodesHitsOk, parser.NodesHitsKo,
 			globalCsInfo,
 			v1.LapiRouteHits, v1.LapiMachineHits, v1.LapiBouncerHits, v1.LapiNilDecisions, v1.LapiNonNilDecisions,
-			leaky.BucketsPour, leaky.BucketsUnderflow, leaky.BucketsInstanciation, leaky.BucketsOverflow, leaky.BucketsCurrentCount)
+			leaky.BucketsPour, leaky.BucketsUnderflow, leaky.BucketsCanceled, leaky.BucketsInstanciation, leaky.BucketsOverflow, leaky.BucketsCurrentCount)
 
 	}
 	http.Handle("/metrics", promhttp.Handler())

pkg/leakybucket/bucket.go

@@ -48,6 +48,7 @@ type Leaky struct {
 	Mapkey string
 	// chan for signaling
 	Signal       chan bool `json:"-"`
+	Suicide      chan bool `json:"-"`
 	Reprocess    bool
 	Simulated    bool
 	Uuid         string
@@ -88,6 +89,14 @@ var BucketsOverflow = prometheus.NewCounterVec(
 	[]string{"name"},
 )
 
+var BucketsCanceled = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "cs_bucket_canceled_total",
+		Help: "Total buckets canceled.",
+	},
+	[]string{"name"},
+)
+
 var BucketsUnderflow = prometheus.NewCounterVec(
 	prometheus.CounterOpts{
 		Name: "cs_bucket_underflowed_total",
@@ -153,6 +162,7 @@ func FromFactory(bucketFactory BucketFactory) *Leaky {
 		Queue:           NewQueue(Qsize),
 		CacheSize:       bucketFactory.CacheSize,
 		Out:             make(chan *Queue, 1),
+		Suicide:         make(chan bool, 1),
 		AllOut:          bucketFactory.ret,
 		Capacity:        bucketFactory.Capacity,
 		Leakspeed:       bucketFactory.leakspeed,
@@ -237,7 +247,15 @@ func LeakRoutine(leaky *Leaky) error {
 		case ofw := <-leaky.Out:
 			leaky.overflow(ofw)
 			return nil
-			/*we underflow or reach bucket deadline (timers)*/
+		/*suiciiiide*/
+		case <-leaky.Suicide:
+			close(leaky.Signal)
+			BucketsCanceled.With(prometheus.Labels{"name": leaky.Name}).Inc()
+			leaky.logger.Debugf("Suicide triggered")
+			leaky.AllOut <- types.Event{Type: types.OVFLW, Overflow: types.RuntimeAlert{Mapkey: leaky.Mapkey}}
+			leaky.logger.Tracef("Returning from leaky routine.")
+			return nil
+		/*we underflow or reach bucket deadline (timers)*/
 		case <-durationTicker:
 			var (
 				alert types.RuntimeAlert

pkg/leakybucket/buckets_test.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"os"
 	"reflect"
+	"sync"
 	"testing"
 	"time"
 
@@ -42,6 +43,7 @@ func TestBucket(t *testing.T) {
 			t.Fatalf("Test '%s' failed : %s", envSetting, err)
 		}
 	} else {
+		wg := new(sync.WaitGroup)
 		fds, err := ioutil.ReadDir("./tests/")
 		if err != nil {
 			t.Fatalf("Unable to read test directory : %s", err)
@@ -50,12 +52,27 @@ func TestBucket(t *testing.T) {
 			fname := "./tests/" + fd.Name()
 			log.Infof("Running test on %s", fname)
 			tomb.Go(func() error {
+				wg.Add(1)
+				defer wg.Done()
 				if err := testOneBucket(t, fname, tomb); err != nil {
 					t.Fatalf("Test '%s' failed : %s", fname, err)
 				}
 				return nil
 			})
 		}
+		wg.Wait()
+	}
+}
+
+//during tests, we're likely to have only one scenario, and thus only one holder.
+//we want to avoid the death of the tomb because all existing buckets have been destroyed.
+func watchTomb(tomb *tomb.Tomb) {
+	for {
+		if tomb.Alive() == false {
+			log.Warningf("Tomb is dead")
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
 	}
 }
 
@@ -102,6 +119,10 @@ func testOneBucket(t *testing.T, dir string, tomb *tomb.Tomb) error {
 	if err != nil {
 		t.Fatalf("failed loading bucket : %s", err)
 	}
+	tomb.Go(func() error {
+		watchTomb(tomb)
+		return nil
+	})
 	if !testFile(t, dir+"/test.json", dir+"/in-buckets_state.json", holders, response, buckets) {
 		return fmt.Errorf("tests from %s failed", dir)
 	}

pkg/leakybucket/manager_load.go

@@ -60,6 +60,7 @@ type BucketFactory struct {
 	RunTimeGroupBy  *vm.Program               `json:"-"`
 	Data            []*types.DataSource       `yaml:"data,omitempty"`
 	DataDir         string                    `yaml:"-"`
+	CancelOnFilter  string                    `yaml:"cancel_on,omitempty"` //a filter that, if matched, kills the bucket
 	leakspeed       time.Duration             //internal representation of `Leakspeed`
 	duration        time.Duration             //internal representation of `Duration`
 	ret             chan types.Event          //the bucket-specific output chan for overflows
@@ -292,6 +293,11 @@ func LoadBucket(bucketFactory *BucketFactory, tomb *tomb.Tomb) error {
 		bucketFactory.processors = append(bucketFactory.processors, &Uniq{})
 	}
 
+	if bucketFactory.CancelOnFilter != "" {
+		bucketFactory.logger.Tracef("Adding a cancel_on filter on %s.", bucketFactory.Name)
+		bucketFactory.processors = append(bucketFactory.processors, &CancelOnFilter{})
+	}
+
 	if bucketFactory.OverflowFilter != "" {
 		bucketFactory.logger.Tracef("Adding an overflow filter")
 		filovflw, err := NewOverflowFilter(bucketFactory)

pkg/leakybucket/reset_filter.go

@@ -0,0 +1,76 @@
+package leakybucket
+
+import (
+	"github.com/antonmedv/expr"
+	"github.com/antonmedv/expr/vm"
+
+	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
+	"github.com/crowdsecurity/crowdsec/pkg/types"
+)
+
+// ResetFilter allows to kill the bucket (without overflowing), if a particular condition is met.
+// An example would be a scenario to detect aggressive crawlers that *do not* fetch any static ressources :
+// type : leaky
+// filter: filter: "evt.Meta.log_type == 'http_access-log'
+// reset_filter: evt.Parsed.request endswith '.css'
+// ....
+// Thus, if the bucket receives a request that matches fetching a static ressource (here css), it cancels itself
+
+type CancelOnFilter struct {
+	CancelOnFilter      *vm.Program
+	CancelOnFilterDebug *exprhelpers.ExprDebugger
+}
+
+func (u *CancelOnFilter) OnBucketPour(bucketFactory *BucketFactory) func(types.Event, *Leaky) *types.Event {
+	return func(msg types.Event, leaky *Leaky) *types.Event {
+		var condition, ok bool
+		if u.CancelOnFilter != nil {
+			leaky.logger.Tracef("running cancel_on filter")
+			output, err := expr.Run(u.CancelOnFilter, exprhelpers.GetExprEnv(map[string]interface{}{"evt": &msg}))
+			if err != nil {
+				leaky.logger.Warningf("cancel_on error : %s", err)
+				return &msg
+			}
+			//only run debugger expression if condition is false
+			if u.CancelOnFilterDebug != nil {
+				u.CancelOnFilterDebug.Run(leaky.logger, condition, exprhelpers.GetExprEnv(map[string]interface{}{"evt": &msg}))
+			}
+			if condition, ok = output.(bool); !ok {
+				leaky.logger.Warningf("cancel_on, unexpected non-bool return : %T", output)
+				return &msg
+			}
+			if condition {
+				leaky.logger.Debugf("reset_filter matched, kill bucket")
+				leaky.Suicide <- true
+				return nil //counter intuitively, we need to keep the message so that it doesn't trigger an endless loop
+			} else {
+				leaky.logger.Debugf("reset_filter didn't match")
+			}
+		}
+		return &msg
+	}
+}
+
+func (u *CancelOnFilter) OnBucketOverflow(bucketFactory *BucketFactory) func(*Leaky, types.RuntimeAlert, *Queue) (types.RuntimeAlert, *Queue) {
+	return func(leaky *Leaky, alert types.RuntimeAlert, queue *Queue) (types.RuntimeAlert, *Queue) {
+		return alert, queue
+	}
+}
+
+func (u *CancelOnFilter) OnBucketInit(bucketFactory *BucketFactory) error {
+	var err error
+
+	u.CancelOnFilter, err = expr.Compile(bucketFactory.CancelOnFilter, expr.Env(exprhelpers.GetExprEnv(map[string]interface{}{"evt": &types.Event{}})))
+	if err != nil {
+		bucketFactory.logger.Errorf("reset_filter compile error : %s", err)
+		return err
+	}
+	if bucketFactory.Debug {
+		u.CancelOnFilterDebug, err = exprhelpers.NewDebugger(bucketFactory.CancelOnFilter, expr.Env(exprhelpers.GetExprEnv(map[string]interface{}{"evt": &types.Event{}})))
+		if err != nil {
+			bucketFactory.logger.Errorf("reset_filter debug error : %s", err)
+			return err
+		}
+	}
+	return err
+}

pkg/leakybucket/tests/simple-leaky-cancel_on/bucket.yaml

@@ -0,0 +1,13 @@
+type: leaky
+debug: true
+name: test/simple-leaky-cancel
+description: "Simple leaky"
+filter: "evt.Line.Labels.type =='testlog'"
+cancel_on: evt.Parsed.random_value == '42'
+leakspeed: "10s"
+blackhole: 1m
+capacity: 1
+groupby: evt.Meta.source_ip
+labels:
+ type: overflow_1
+

pkg/leakybucket/tests/simple-leaky-cancel_on/scenarios.yaml

@@ -0,0 +1,2 @@
+ - filename: {{.TestDirectory}}/bucket.yaml
+

pkg/leakybucket/tests/simple-leaky-cancel_on/test.json

@@ -0,0 +1,117 @@
+{
+  "lines": [
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE1 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:00+00:00",
+      "Meta": {
+        "source_ip": "1.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "41"
+      }
+    },
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE2 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:05+00:00",
+      "Meta": {
+        "source_ip": "1.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "42"
+      }
+    },
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE1 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:00+00:00",
+      "Meta": {
+        "source_ip": "1.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "41"
+      }
+    },
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE1 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:00+00:00",
+      "Meta": {
+        "source_ip": "2.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "41"
+      }
+    },
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE1 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:00+00:00",
+      "Meta": {
+        "source_ip": "2.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "41"
+      }
+    },
+    {
+      "Line": {
+        "Labels": {
+          "type": "testlog"
+        },
+        "Raw": "xxheader VALUE1 trailing stuff"
+      },
+      "MarshaledTime": "2020-01-01T10:00:00+00:00",
+      "Meta": {
+        "source_ip": "2.2.3.4"
+      },
+      "Parsed": {
+        "random_value" : "41"
+      }
+    }
+  ],
+  "results": [
+    {
+      "Alert": {
+      }
+    },
+    {
+      "Alert": {
+        "sources": {
+          "2.2.3.4": {
+              "scope": "Ip",
+              "value": "2.2.3.4",
+            "ip": "2.2.3.4"
+          }
+        },
+        "Alert" : {
+        "scenario": "test/simple-leaky-cancel",
+        "events_count": 2
+        }
+      }
+    }
+
+  ]
+}
+