Add query param to filter decisions by scenarios and origin (#1294)

* Add query param to filter decisions by scenarios

pkg/apiclient/decisions_service.go

@@ -3,7 +3,6 @@ package apiclient
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/crowdsecurity/crowdsec/pkg/models"
 	qs "github.com/google/go-querystring/query"
@@ -18,10 +17,24 @@ type DecisionsListOpts struct {
 	IPEquals    *string `url:"ip,omitempty"`
 	RangeEquals *string `url:"range,omitempty"`
 	Contains    *bool   `url:"contains,omitempty"`
-
 	ListOpts
 }
 
+type DecisionsStreamOpts struct {
+	Startup                bool   `url:"startup,omitempty"`
+	Scopes                 string `url:"scopes,omitempty"`
+	ScenariosContaining    string `url:"scenarios_containing,omitempty"`
+	ScenariosNotContaining string `url:"scenarios_not_containing,omitempty"`
+}
+
+func (o *DecisionsStreamOpts) addQueryParamsToURL(url string) (string, error) {
+	params, err := qs.Values(o)
+	if err != nil {
+		return "", err
+	}
+	return fmt.Sprintf("%s?%s", url, params.Encode()), nil
+}
+
 type DecisionsDeleteOpts struct {
 	ScopeEquals *string `url:"scope,omitempty"`
 	ValueEquals *string `url:"value,omitempty"`
@@ -53,11 +66,11 @@ func (s *DecisionsService) List(ctx context.Context, opts DecisionsListOpts) (*m
 	return &decisions, resp, nil
 }
 
-func (s *DecisionsService) GetStream(ctx context.Context, startup bool, scopes []string) (*models.DecisionsStreamResponse, *Response, error) {
+func (s *DecisionsService) GetStream(ctx context.Context, opts DecisionsStreamOpts) (*models.DecisionsStreamResponse, *Response, error) {
 	var decisions models.DecisionsStreamResponse
-	u := fmt.Sprintf("%s/decisions/stream?startup=%t", s.client.URLPrefix, startup)
-	if len(scopes) > 0 {
-		u += "&scopes=" + strings.Join(scopes, ",")
+	u, err := opts.addQueryParamsToURL(s.client.URLPrefix + "/decisions/stream")
+	if err != nil {
+		return nil, nil, err
 	}
 	req, err := s.client.NewRequest("GET", u, nil)
 	if err != nil {

pkg/apiclient/decisions_service_test.go

@@ -160,7 +160,7 @@ func TestDecisionsStream(t *testing.T) {
 		},
 	}
 
-	decisions, resp, err := newcli.Decisions.GetStream(context.Background(), true, []string{})
+	decisions, resp, err := newcli.Decisions.GetStream(context.Background(), DecisionsStreamOpts{Startup: true})
 	require.NoError(t, err)
 
 	if resp.Response.StatusCode != http.StatusOK {
@@ -175,7 +175,7 @@ func TestDecisionsStream(t *testing.T) {
 	}
 
 	//and second call, we get empty lists
-	decisions, resp, err = newcli.Decisions.GetStream(context.Background(), false, []string{})
+	decisions, resp, err = newcli.Decisions.GetStream(context.Background(), DecisionsStreamOpts{Startup: false})
 	require.NoError(t, err)
 
 	if resp.Response.StatusCode != http.StatusOK {
@@ -234,6 +234,73 @@ func TestDeleteDecisions(t *testing.T) {
 	defer teardown()
 }
 
+func TestDecisionsStreamOpts_addQueryParamsToURL(t *testing.T) {
+	baseURLString := "http://localhost:8080/v1/decisions/stream"
+	type fields struct {
+		Startup                bool
+		Scopes                 string
+		ScenariosContaining    string
+		ScenariosNotContaining string
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "no filter",
+			want: baseURLString + "?",
+		},
+		{
+			name: "startup=true",
+			fields: fields{
+				Startup: true,
+			},
+			want: baseURLString + "?startup=true",
+		},
+		{
+			name: "set all params",
+			fields: fields{
+				Startup:                true,
+				Scopes:                 "ip,range",
+				ScenariosContaining:    "ssh",
+				ScenariosNotContaining: "bf",
+			},
+			want: baseURLString + "?scenarios_containing=ssh&scenarios_not_containing=bf&scopes=ip%2Crange&startup=true",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			o := &DecisionsStreamOpts{
+				Startup:                tt.fields.Startup,
+				Scopes:                 tt.fields.Scopes,
+				ScenariosContaining:    tt.fields.ScenariosContaining,
+				ScenariosNotContaining: tt.fields.ScenariosNotContaining,
+			}
+			got, err := o.addQueryParamsToURL(baseURLString)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("DecisionsStreamOpts.addQueryParamsToURL() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			gotURL, err := url.Parse(got)
+			if err != nil {
+				t.Errorf("DecisionsStreamOpts.addQueryParamsToURL() got error while parsing URL: %s", err)
+			}
+
+			expectedURL, err := url.Parse(tt.want)
+			if err != nil {
+				t.Errorf("DecisionsStreamOpts.addQueryParamsToURL() got error while parsing URL: %s", err)
+			}
+
+			if *gotURL != *expectedURL {
+				t.Errorf("DecisionsStreamOpts.addQueryParamsToURL() = %v, want %v", *gotURL, *expectedURL)
+			}
+		})
+	}
+}
+
 // func TestDeleteOneDecision(t *testing.T) {
 // 	mux, urlx, teardown := setup()
 // 	mux.HandleFunc("/watchers/login", func(w http.ResponseWriter, r *http.Request) {

pkg/apiserver/apic.go

@@ -275,7 +275,7 @@ func (a *apic) PullTop() error {
 		log.Printf("last CAPI pull is newer than 1h30, skip.")
 		return nil
 	}
-	data, _, err := a.apiClient.Decisions.GetStream(context.Background(), a.startup, []string{})
+	data, _, err := a.apiClient.Decisions.GetStream(context.Background(), apiclient.DecisionsStreamOpts{Startup: a.startup})
 	if err != nil {
 		return errors.Wrap(err, "get stream")
 	}

pkg/apiserver/controllers/v1/decisions.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
-	"strings"
 	"time"
 
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent"
@@ -127,10 +126,9 @@ func (c *Controller) StreamDecision(gctx *gin.Context) {
 		return
 	}
 
-	filters := make(map[string][]string)
-	filters["scope"] = []string{"ip", "range"}
-	if val, ok := gctx.Request.URL.Query()["scopes"]; ok {
-		filters["scope"] = strings.Split(val[0], ",")
+	filters := gctx.Request.URL.Query()
+	if _, ok := filters["scopes"]; !ok {
+		filters["scopes"] = []string{"ip,range"}
 	}
 
 	// if the blocker just start, return all decisions

pkg/apiserver/decisions_test.go

@@ -120,7 +120,7 @@ func TestDeleteDecisionFilter(t *testing.T) {
 
 	// delete by scope/value
 	w = httptest.NewRecorder()
-	req, _ = http.NewRequest("DELETE", "/v1/decisions?scope=Ip&value=91.121.79.178", strings.NewReader(""))
+	req, _ = http.NewRequest("DELETE", "/v1/decisions?scopes=Ip&value=91.121.79.178", strings.NewReader(""))
 	AddAuthHeaders(req, loginResp)
 	router.ServeHTTP(w, req)
 	assert.Equal(t, 200, w.Code)
@@ -185,7 +185,7 @@ func TestGetDecisionFilters(t *testing.T) {
 
 	// Get Decision : scope/value
 	w = httptest.NewRecorder()
-	req, _ = http.NewRequest("GET", "/v1/decisions?scope=Ip&value=91.121.79.179", strings.NewReader(""))
+	req, _ = http.NewRequest("GET", "/v1/decisions?scopes=Ip&value=91.121.79.179", strings.NewReader(""))
 	req.Header.Add("User-Agent", UserAgent)
 	req.Header.Add("X-Api-Key", APIKey)
 	router.ServeHTTP(w, req)
@@ -249,19 +249,19 @@ func TestGetDecision(t *testing.T) {
 		log.Fatalf("%s", err.Error())
 	}
 
-	// Get Decision with invalid filter
+	// Get Decision
 	w = httptest.NewRecorder()
-	req, _ = http.NewRequest("GET", "/v1/decisions?test=test", strings.NewReader(""))
+	req, _ = http.NewRequest("GET", "/v1/decisions", strings.NewReader(""))
 	req.Header.Add("User-Agent", UserAgent)
 	req.Header.Add("X-Api-Key", APIKey)
 	router.ServeHTTP(w, req)
 
-	assert.Equal(t, 500, w.Code)
-	assert.Equal(t, "{\"message\":\"'test' doesn't exist: invalid filter\"}", w.Body.String())
+	assert.Equal(t, 200, w.Code)
+	assert.Contains(t, w.Body.String(), "\"id\":1,\"origin\":\"test\",\"scenario\":\"crowdsecurity/test\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"}]")
 
-	// Get Decision
+	// Get Decision with invalid filter. It should ignore this filter
 	w = httptest.NewRecorder()
-	req, _ = http.NewRequest("GET", "/v1/decisions", strings.NewReader(""))
+	req, _ = http.NewRequest("GET", "/v1/decisions?test=test", strings.NewReader(""))
 	req.Header.Add("User-Agent", UserAgent)
 	req.Header.Add("X-Api-Key", APIKey)
 	router.ServeHTTP(w, req)
@@ -388,7 +388,7 @@ func TestStreamDecision(t *testing.T) {
 	}
 
 	// Create Valid Alert
-	alertContentBytes, err := ioutil.ReadFile("./tests/alert_sample.json")
+	alertContentBytes, err := ioutil.ReadFile("./tests/alert_stream_fixture.json")
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -432,5 +432,51 @@ func TestStreamDecision(t *testing.T) {
 	router.ServeHTTP(w, req)
 
 	assert.Equal(t, 200, w.Code)
-	assert.Contains(t, w.Body.String(), "\"id\":1,\"origin\":\"test\",\"scenario\":\"crowdsecurity/test\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"}]}")
+	assert.Contains(t, w.Body.String(), "\"id\":1,\"origin\":\"test1\",\"scenario\":\"crowdsecurity/http_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":2,\"origin\":\"test2\",\"scenario\":\"crowdsecurity/ssh_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":3,\"origin\":\"test3\",\"scenario\":\"crowdsecurity/ddos\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+
+	// test filter scenarios_not_containing
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequest("GET", "/v1/decisions/stream?startup=true&scenarios_not_containing=http", strings.NewReader(""))
+	req.Header.Add("X-Api-Key", APIKey)
+	router.ServeHTTP(w, req)
+
+	assert.Equal(t, 200, w.Code)
+	assert.NotContains(t, w.Body.String(), "\"id\":1,\"origin\":\"test1\",\"scenario\":\"crowdsecurity/http_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":2,\"origin\":\"test2\",\"scenario\":\"crowdsecurity/ssh_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":3,\"origin\":\"test3\",\"scenario\":\"crowdsecurity/ddos\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+
+	// test  filter scenarios_containing
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequest("GET", "/v1/decisions/stream?startup=true&scenarios_containing=http", strings.NewReader(""))
+	req.Header.Add("X-Api-Key", APIKey)
+	router.ServeHTTP(w, req)
+
+	assert.Equal(t, 200, w.Code)
+	assert.Contains(t, w.Body.String(), "\"id\":1,\"origin\":\"test1\",\"scenario\":\"crowdsecurity/http_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.NotContains(t, w.Body.String(), "\"id\":2,\"origin\":\"test2\",\"scenario\":\"crowdsecurity/ssh_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.NotContains(t, w.Body.String(), "\"id\":3,\"origin\":\"test3\",\"scenario\":\"crowdsecurity/ddos\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+
+	// test filters both by scenarios_not_containing and scenarios_containing
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequest("GET", "/v1/decisions/stream?startup=true&scenarios_not_containing=ssh&scenarios_containing=ddos", strings.NewReader(""))
+	req.Header.Add("X-Api-Key", APIKey)
+	router.ServeHTTP(w, req)
+
+	assert.Equal(t, 200, w.Code)
+	assert.NotContains(t, w.Body.String(), "\"id\":1,\"origin\":\"test1\",\"scenario\":\"crowdsecurity/http_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.NotContains(t, w.Body.String(), "\"id\":2,\"origin\":\"test2\",\"scenario\":\"crowdsecurity/ssh_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":3,\"origin\":\"test3\",\"scenario\":\"crowdsecurity/ddos\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+
+	// test filter by origin
+	w = httptest.NewRecorder()
+	req, _ = http.NewRequest("GET", "/v1/decisions/stream?startup=true&origins=test1,test2", strings.NewReader(""))
+	req.Header.Add("X-Api-Key", APIKey)
+	router.ServeHTTP(w, req)
+
+	assert.Equal(t, 200, w.Code)
+	assert.Contains(t, w.Body.String(), "\"id\":1,\"origin\":\"test1\",\"scenario\":\"crowdsecurity/http_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.Contains(t, w.Body.String(), "\"id\":2,\"origin\":\"test2\",\"scenario\":\"crowdsecurity/ssh_bf\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
+	assert.NotContains(t, w.Body.String(), "\"id\":3,\"origin\":\"test3\",\"scenario\":\"crowdsecurity/ddos\",\"scope\":\"Ip\",\"type\":\"ban\",\"value\":\"127.0.0.1\"")
 }

pkg/apiserver/tests/alert_stream_fixture.json

@@ -0,0 +1,173 @@
+[
+    {
+        "id": 42,
+        "machine_id": "test",
+        "capacity": 1,
+        "created_at": "2020-10-09T10:00:10Z",
+        "decisions": [
+            {
+                "id": 1,
+                "duration": "1h",
+                "origin": "test1",
+                "scenario": "crowdsecurity/http_bf",
+                "scope": "Ip",
+                "value": "127.0.0.1",
+                "type": "ban"
+            }
+        ],
+        "Events": [
+            {
+                "meta": [
+                    {
+                        "key": "test",
+                        "value": "test"
+                    }
+                ],
+                "timestamp": "2020-10-09T10:00:01Z"
+            }
+        ],
+        "events_count": 1,
+        "labels": [
+            "test"
+        ],
+        "leakspeed": "0.5s",
+        "message": "test",
+        "meta": [
+            {
+                "key": "test",
+                "value": "test"
+            }
+        ],
+        "scenario": "crowdsecurity/http_bf",
+        "scenario_hash": "hashtest",
+        "scenario_version": "v1",
+        "simulated": false,
+        "source": {
+            "as_name": "test",
+            "as_number": "0123456",
+            "cn": "france",
+            "ip": "127.0.0.1",
+            "latitude": 46.227638,
+            "logitude": 2.213749,
+            "range": "127.0.0.1/32",
+            "scope": "ip",
+            "value": "127.0.0.1"
+        },
+        "start_at": "2020-10-09T10:00:01Z",
+        "stop_at": "2020-10-09T10:00:05Z"
+    },
+    {
+        "id": 43,
+        "machine_id": "test",
+        "capacity": 1,
+        "created_at": "2020-10-09T10:00:10Z",
+        "decisions": [
+            {
+                "id": 2,
+                "duration": "1h",
+                "origin": "test2",
+                "scenario": "crowdsecurity/ssh_bf",
+                "scope": "Ip",
+                "value": "127.0.0.1",
+                "type": "ban"
+            }
+        ],
+        "Events": [
+            {
+                "meta": [
+                    {
+                        "key": "test",
+                        "value": "test"
+                    }
+                ],
+                "timestamp": "2020-10-09T10:00:01Z"
+            }
+        ],
+        "events_count": 1,
+        "labels": [
+            "test"
+        ],
+        "leakspeed": "0.5s",
+        "message": "test",
+        "meta": [
+            {
+                "key": "test",
+                "value": "test"
+            }
+        ],
+        "scenario": "crowdsecurity/ssh_bf",
+        "scenario_hash": "hashtest",
+        "scenario_version": "v1",
+        "simulated": false,
+        "source": {
+            "as_name": "test",
+            "as_number": "0123456",
+            "cn": "france",
+            "ip": "127.0.0.1",
+            "latitude": 46.227638,
+            "logitude": 2.213749,
+            "range": "127.0.0.1/32",
+            "scope": "ip",
+            "value": "127.0.0.1"
+        },
+        "start_at": "2020-10-09T10:00:01Z",
+        "stop_at": "2020-10-09T10:00:05Z"
+    },
+    {
+        "id": 44,
+        "machine_id": "test",
+        "capacity": 1,
+        "created_at": "2020-10-09T10:00:10Z",
+        "decisions": [
+            {
+                "id": 3,
+                "duration": "1h",
+                "origin": "test3",
+                "scenario": "crowdsecurity/ddos",
+                "scope": "Ip",
+                "value": "127.0.0.1",
+                "type": "ban"
+            }
+        ],
+        "Events": [
+            {
+                "meta": [
+                    {
+                        "key": "test",
+                        "value": "test"
+                    }
+                ],
+                "timestamp": "2020-10-09T10:00:01Z"
+            }
+        ],
+        "events_count": 1,
+        "labels": [
+            "test"
+        ],
+        "leakspeed": "0.5s",
+        "message": "test",
+        "meta": [
+            {
+                "key": "test",
+                "value": "test"
+            }
+        ],
+        "scenario": "crowdsecurity/ddos",
+        "scenario_hash": "hashtest",
+        "scenario_version": "v1",
+        "simulated": false,
+        "source": {
+            "as_name": "test",
+            "as_number": "0123456",
+            "cn": "france",
+            "ip": "127.0.0.1",
+            "latitude": 46.227638,
+            "logitude": 2.213749,
+            "range": "127.0.0.1/32",
+            "scope": "ip",
+            "value": "127.0.0.1"
+        },
+        "start_at": "2020-10-09T10:00:01Z",
+        "stop_at": "2020-10-09T10:00:05Z"
+    }
+]

pkg/database/decisions.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent"
 	"github.com/crowdsecurity/crowdsec/pkg/database/ent/decision"
+	"github.com/crowdsecurity/crowdsec/pkg/database/ent/predicate"
 	"github.com/crowdsecurity/crowdsec/pkg/types"
 	"github.com/pkg/errors"
 )
@@ -39,31 +40,44 @@ func BuildDecisionRequestWithFilter(query *ent.DecisionQuery, filter map[string]
 			if err != nil {
 				return nil, errors.Wrapf(InvalidFilter, "invalid contains value : %s", err)
 			}
-		case "scope":
-			for i, scope := range value {
+		case "scopes":
+			scopes := strings.Split(value[0], ",")
+			for i, scope := range scopes {
 				switch strings.ToLower(scope) {
 				case "ip":
-					value[i] = types.Ip
+					scopes[i] = types.Ip
 				case "range":
-					value[i] = types.Range
+					scopes[i] = types.Range
 				case "country":
-					value[i] = types.Country
+					scopes[i] = types.Country
 				case "as":
-					value[i] = types.AS
+					scopes[i] = types.AS
 				}
 			}
-			query = query.Where(decision.ScopeIn(value...))
+			query = query.Where(decision.ScopeIn(scopes...))
 		case "value":
 			query = query.Where(decision.ValueEQ(value[0]))
 		case "type":
 			query = query.Where(decision.TypeEQ(value[0]))
+		case "origins":
+			query = query.Where(
+				decision.OriginIn(strings.Split(value[0], ",")...),
+			)
+		case "scenarios_containing":
+			predicates := decisionPredicatesFromStr(value[0], decision.ScenarioContainsFold)
+			query = query.Where(decision.Or(predicates...))
+		case "scenarios_not_containing":
+			predicates := decisionPredicatesFromStr(value[0], decision.ScenarioContainsFold)
+			query = query.Where(decision.Not(
+				decision.Or(
+					predicates...,
+				),
+			))
 		case "ip", "range":
 			ip_sz, start_ip, start_sfx, end_ip, end_sfx, err = types.Addr2Ints(value[0])
 			if err != nil {
 				return nil, errors.Wrapf(InvalidIPOrRange, "unable to convert '%s' to int: %s", value[0], err)
 			}
-		default:
-			return query, errors.Wrapf(InvalidFilter, "'%s' doesn't exist", param)
 		}
 	}
 
@@ -367,7 +381,7 @@ func (c *Client) SoftDeleteDecisionsWithFilter(filter map[string][]string) (stri
 			if err != nil {
 				return "0", errors.Wrapf(InvalidFilter, "invalid contains value : %s", err)
 			}
-		case "scope":
+		case "scopes":
 			decisions = decisions.Where(decision.ScopeEQ(value[0]))
 		case "value":
 			decisions = decisions.Where(decision.ValueEQ(value[0]))
@@ -474,3 +488,12 @@ func (c *Client) SoftDeleteDecisionByID(decisionID int) error {
 	}
 	return nil
 }
+
+func decisionPredicatesFromStr(s string, predicateFunc func(string) predicate.Decision) []predicate.Decision {
+	words := strings.Split(s, ",")
+	predicates := make([]predicate.Decision, len(words))
+	for i, word := range words {
+		predicates[i] = predicateFunc(word)
+	}
+	return predicates
+}

pkg/models/localapi_swagger.yaml

@@ -45,6 +45,21 @@ paths:
           required: false
           type: string
           description: 'Comma separated scopes of decisions to fetch'
+        - name: origins
+          in: query
+          required: false
+          type: string
+          description: 'Comma separated name of origins. If provided, then only the decisions originating from provided origins would be returned.'
+        - name: scenarios_containing
+          in: query
+          required: false
+          type: string
+          description: 'Comma separated words. If provided, only the decisions created by scenarios containing any of the provided word would be returned.'
+        - name: scenarios_not_containing
+          in: query
+          required: false
+          type: string
+          description: 'Comma separated words. If provided, only the decisions created by scenarios, not containing any of the provided word would be returned.'
       responses:
         '200':
           description: successful operation