diff --git a/cmd/crowdsec/metrics.go b/cmd/crowdsec/metrics.go index 6cd31ab9e..22a1faf9c 100644 --- a/cmd/crowdsec/metrics.go +++ b/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()) diff --git a/pkg/leakybucket/bucket.go b/pkg/leakybucket/bucket.go index df1c0048a..b4f5b34d1 100644 --- a/pkg/leakybucket/bucket.go +++ b/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 diff --git a/pkg/leakybucket/buckets_test.go b/pkg/leakybucket/buckets_test.go index ce4be406f..d743c82b8 100644 --- a/pkg/leakybucket/buckets_test.go +++ b/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) } diff --git a/pkg/leakybucket/manager_load.go b/pkg/leakybucket/manager_load.go index be6f9817b..0cd03a9a0 100644 --- a/pkg/leakybucket/manager_load.go +++ b/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) diff --git a/pkg/leakybucket/reset_filter.go b/pkg/leakybucket/reset_filter.go new file mode 100644 index 000000000..3567e4ca6 --- /dev/null +++ b/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 +} diff --git a/pkg/leakybucket/tests/simple-leaky-cancel_on/bucket.yaml b/pkg/leakybucket/tests/simple-leaky-cancel_on/bucket.yaml new file mode 100644 index 000000000..f2d3716dd --- /dev/null +++ b/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 + diff --git a/pkg/leakybucket/tests/simple-leaky-cancel_on/scenarios.yaml b/pkg/leakybucket/tests/simple-leaky-cancel_on/scenarios.yaml new file mode 100644 index 000000000..f45f7be12 --- /dev/null +++ b/pkg/leakybucket/tests/simple-leaky-cancel_on/scenarios.yaml @@ -0,0 +1,2 @@ + - filename: {{.TestDirectory}}/bucket.yaml + diff --git a/pkg/leakybucket/tests/simple-leaky-cancel_on/test.json b/pkg/leakybucket/tests/simple-leaky-cancel_on/test.json new file mode 100644 index 000000000..1e3023fbc --- /dev/null +++ b/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 + } + } + } + + ] +} +