pkg/types -> new imports pt 3 (#4014)

Author: mmetc

cmd/crowdsec/crowdsec.go

@@ -19,7 +19,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
 	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 // initCrowdsec prepares the log processor service
@@ -172,8 +172,8 @@ func startLPMetrics(ctx context.Context, cConfig *csconfig.Config, apiClient *ap
 
 // runCrowdsec starts the log processor service
 func runCrowdsec(ctx context.Context, cConfig *csconfig.Config, parsers *parser.Parsers, hub *cwhub.Hub, datasources []acquisition.DataSource) error {
-	inputEventChan = make(chan types.Event)
-	inputLineChan = make(chan types.Event)
+	inputEventChan = make(chan pipeline.Event)
+	inputLineChan = make(chan pipeline.Event)
 
 	startParserRoutines(cConfig, parsers)
 

cmd/crowdsec/main.go

@@ -27,7 +27,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 	"github.com/crowdsecurity/crowdsec/pkg/logging"
 	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 var (
@@ -49,9 +49,9 @@ var (
 	holders []leakybucket.BucketFactory
 	buckets *leakybucket.Buckets
 
-	inputLineChan   chan types.Event
-	inputEventChan  chan types.Event
-	outputEventChan chan types.Event // the buckets init returns its own chan that is used for multiplexing
+	inputLineChan   chan pipeline.Event
+	inputEventChan  chan pipeline.Event
+	outputEventChan chan pipeline.Event // the buckets init returns its own chan that is used for multiplexing
 	// settings
 	lastProcessedItem time.Time // keep track of last item timestamp in time-machine. it is used to GC buckets when we dump them.
 	pluginBroker      csplugin.PluginBroker

cmd/crowdsec/output.go

@@ -12,10 +12,10 @@ import (
 	leaky "github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 	"github.com/crowdsecurity/crowdsec/pkg/models"
 	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func dedupAlerts(alerts []types.RuntimeAlert) ([]*models.Alert, error) {
+func dedupAlerts(alerts []pipeline.RuntimeAlert) ([]*models.Alert, error) {
 	var dedupCache []*models.Alert
 
 	for idx, alert := range alerts {
@@ -44,7 +44,7 @@ func dedupAlerts(alerts []types.RuntimeAlert) ([]*models.Alert, error) {
 	return dedupCache, nil
 }
 
-func PushAlerts(ctx context.Context, alerts []types.RuntimeAlert, client *apiclient.ApiClient) error {
+func PushAlerts(ctx context.Context, alerts []pipeline.RuntimeAlert, client *apiclient.ApiClient) error {
 	alertsToPush, err := dedupAlerts(alerts)
 	if err != nil {
 		return fmt.Errorf("failed to transform alerts for api: %w", err)
@@ -58,12 +58,12 @@ func PushAlerts(ctx context.Context, alerts []types.RuntimeAlert, client *apicli
 	return nil
 }
 
-var bucketOverflows []types.Event
+var bucketOverflows []pipeline.Event
 
-func runOutput(ctx context.Context, input chan types.Event, overflow chan types.Event, buckets *leaky.Buckets, postOverflowCTX parser.UnixParserCtx,
+func runOutput(ctx context.Context, input chan pipeline.Event, overflow chan pipeline.Event, buckets *leaky.Buckets, postOverflowCTX parser.UnixParserCtx,
 	postOverflowNodes []parser.Node, client *apiclient.ApiClient) error {
 	var (
-		cache      []types.RuntimeAlert
+		cache      []pipeline.RuntimeAlert
 		cacheMutex sync.Mutex
 	)
 
@@ -76,7 +76,7 @@ func runOutput(ctx context.Context, input chan types.Event, overflow chan types.
 			if len(cache) > 0 {
 				cacheMutex.Lock()
 				cachecopy := cache
-				newcache := make([]types.RuntimeAlert, 0)
+				newcache := make([]pipeline.RuntimeAlert, 0)
 				cache = newcache
 				cacheMutex.Unlock()
 				/*
@@ -121,7 +121,7 @@ func runOutput(ctx context.Context, input chan types.Event, overflow chan types.
 			// dump after postoveflow processing to avoid missing whitelist info
 			if dumpStates && event.Overflow.Alert != nil {
 				if bucketOverflows == nil {
-					bucketOverflows = make([]types.Event, 0)
+					bucketOverflows = make([]pipeline.Event, 0)
 				}
 				bucketOverflows = append(bucketOverflows, event)
 			}

cmd/crowdsec/parse.go

@@ -8,10 +8,10 @@ import (
 
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
 	"github.com/crowdsecurity/crowdsec/pkg/parser"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func runParse(input chan types.Event, output chan types.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) error {
+func runParse(input chan pipeline.Event, output chan pipeline.Event, parserCTX parser.UnixParserCtx, nodes []parser.Node) error {
 	for {
 		select {
 		case <-parsersTomb.Dying():
@@ -24,7 +24,7 @@ func runParse(input chan types.Event, output chan types.Event, parserCTX parser.
 			/*Application security engine is going to generate 2 events:
 			- one that is treated as a log and can go to scenarios
 			- another one that will go directly to LAPI*/
-			if event.Type == types.APPSEC {
+			if event.Type == pipeline.APPSEC {
 				outputEventChan <- event
 				continue
 			}

cmd/crowdsec/pour.go

@@ -10,10 +10,10 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
 	leaky "github.com/crowdsecurity/crowdsec/pkg/leakybucket"
 	"github.com/crowdsecurity/crowdsec/pkg/metrics"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
-func maybeGC(parsed types.Event, buckets *leaky.Buckets, cConfig *csconfig.Config) error {
+func maybeGC(parsed pipeline.Event, buckets *leaky.Buckets, cConfig *csconfig.Config) error {
 	log.Infof("%d existing buckets", leaky.LeakyRoutineCount)
 	// when in forensics mode, garbage collect buckets
 	if cConfig.Crowdsec.BucketsGCEnabled {
@@ -34,7 +34,7 @@ func maybeGC(parsed types.Event, buckets *leaky.Buckets, cConfig *csconfig.Confi
 	return nil
 }
 
-func runPour(input chan types.Event, holders []leaky.BucketFactory, buckets *leaky.Buckets, cConfig *csconfig.Config) error {
+func runPour(input chan pipeline.Event, holders []leaky.BucketFactory, buckets *leaky.Buckets, cConfig *csconfig.Config) error {
 	count := 0
 
 	for {

cmd/crowdsec/serve.go

@@ -21,7 +21,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
 	"github.com/crowdsecurity/crowdsec/pkg/database"
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 func reloadHandler(ctx context.Context, _ os.Signal) (*csconfig.Config, error) {
@@ -205,7 +205,7 @@ func shutdown(_ os.Signal, cConfig *csconfig.Config) error {
 	return nil
 }
 
-func drainChan(c chan types.Event) {
+func drainChan(c chan pipeline.Event) {
 	time.Sleep(500 * time.Millisecond)
 	// delay to avoid draining chan before the acquisition/parser
 	// get a chance to push its event

pkg/alertcontext/alertcontext.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
 	"github.com/crowdsecurity/crowdsec/pkg/models"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 const MaxContextValueLen = 4000
@@ -30,8 +30,8 @@ type Context struct {
 func ValidateContextExpr(key string, expressions []string) error {
 	for _, expression := range expressions {
 		_, err := expr.Compile(expression, exprhelpers.GetExprOptions(map[string]interface{}{
-			"evt":   &types.Event{},
-			"match": &types.MatchedRule{},
+			"evt":   &pipeline.Event{},
+			"match": &pipeline.MatchedRule{},
 			"req":   &http.Request{},
 		})...)
 		if err != nil {
@@ -70,8 +70,8 @@ func NewAlertContext(contextToSend map[string][]string, valueLength int) error {
 
 		for _, value := range values {
 			valueCompiled, err := expr.Compile(value, exprhelpers.GetExprOptions(map[string]interface{}{
-				"evt":   &types.Event{},
-				"match": &types.MatchedRule{},
+				"evt":   &pipeline.Event{},
+				"match": &pipeline.MatchedRule{},
 				"req":   &http.Request{},
 			})...)
 			if err != nil {
@@ -144,13 +144,13 @@ func TruncateContext(values []string, contextValueLen int) (string, error) {
 	return ret, nil
 }
 
-func EvalAlertContextRules(evt types.Event, match *types.MatchedRule, request *http.Request, tmpContext map[string][]string) []error {
+func EvalAlertContextRules(evt pipeline.Event, match *pipeline.MatchedRule, request *http.Request, tmpContext map[string][]string) []error {
 	var errors []error
 
 	// if we're evaluating context for appsec event, match and request will be present.
 	// otherwise, only evt will be.
 	if match == nil {
-		match = types.NewMatchedRule()
+		match = pipeline.NewMatchedRule()
 	}
 
 	if request == nil {
@@ -212,12 +212,12 @@ func EvalAlertContextRules(evt types.Event, match *types.MatchedRule, request *h
 }
 
 // Iterate over the individual appsec matched rules to create the needed alert context.
-func AppsecEventToContext(event types.AppsecEvent, request *http.Request) (models.Meta, []error) {
+func AppsecEventToContext(event pipeline.AppsecEvent, request *http.Request) (models.Meta, []error) {
 	var errors []error
 
 	tmpContext := make(map[string][]string)
 
-	evt := types.MakeEvent(false, types.LOG, false)
+	evt := pipeline.MakeEvent(false, pipeline.LOG, false)
 	for _, matched_rule := range event.MatchedRules {
 		tmpErrors := EvalAlertContextRules(evt, &matched_rule, request, tmpContext)
 		errors = append(errors, tmpErrors...)
@@ -232,7 +232,7 @@ func AppsecEventToContext(event types.AppsecEvent, request *http.Request) (model
 }
 
 // Iterate over the individual events to create the needed alert context.
-func EventToContext(events []types.Event) (models.Meta, []error) {
+func EventToContext(events []pipeline.Event) (models.Meta, []error) {
 	var errors []error
 
 	tmpContext := make(map[string][]string)

pkg/alertcontext/alertcontext_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/crowdsecurity/go-cs-lib/ptr"
 
 	"github.com/crowdsecurity/crowdsec/pkg/models"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 func TestNewAlertContext(t *testing.T) {
@@ -43,7 +43,7 @@ func TestEventToContext(t *testing.T) {
 		name           string
 		contextToSend  map[string][]string
 		valueLength    int
-		events         []types.Event
+		events         []pipeline.Event
 		expectedResult models.Meta
 	}{
 		{
@@ -53,7 +53,7 @@ func TestEventToContext(t *testing.T) {
 				"nonexistent_field": {"evt.Parsed.nonexist"},
 			},
 			valueLength: 100,
-			events: []types.Event{
+			events: []pipeline.Event{
 				{
 					Parsed: map[string]string{
 						"source_ip":      "1.2.3.4",
@@ -76,7 +76,7 @@ func TestEventToContext(t *testing.T) {
 				"cve":            {"evt.Parsed.cve"},
 			},
 			valueLength: 100,
-			events: []types.Event{
+			events: []pipeline.Event{
 				{
 					Parsed: map[string]string{
 						"source_ip":      "1.2.3.4",
@@ -122,7 +122,7 @@ func TestEventToContext(t *testing.T) {
 				"uri":            {"evt.Parsed.uri"},
 			},
 			valueLength: 100,
-			events: []types.Event{
+			events: []pipeline.Event{
 				{
 					Parsed: map[string]string{
 						"source_ip":      "1.2.3.4",
@@ -168,7 +168,7 @@ func TestEventToContext(t *testing.T) {
 				"uri":            {"evt.Parsed.uri"},
 			},
 			valueLength: 100,
-			events: []types.Event{
+			events: []pipeline.Event{
 				{
 					Parsed: map[string]string{
 						"source_ip":      "1.2.3.4",
@@ -244,7 +244,7 @@ func TestAppsecEventToContext(t *testing.T) {
 	tests := []struct {
 		name           string
 		contextToSend  map[string][]string
-		match          types.AppsecEvent
+		match          pipeline.AppsecEvent
 		req            *http.Request
 		expectedResult models.Meta
 		expectedErrLen int
@@ -254,8 +254,8 @@ func TestAppsecEventToContext(t *testing.T) {
 			contextToSend: map[string][]string{
 				"id": {"match.id"},
 			},
-			match: types.AppsecEvent{
-				MatchedRules: types.MatchedRules{
+			match: pipeline.AppsecEvent{
+				MatchedRules: pipeline.MatchedRules{
 					{
 						"id": "test",
 					},
@@ -275,8 +275,8 @@ func TestAppsecEventToContext(t *testing.T) {
 			contextToSend: map[string][]string{
 				"ua": {"req.UserAgent()"},
 			},
-			match: types.AppsecEvent{
-				MatchedRules: types.MatchedRules{
+			match: pipeline.AppsecEvent{
+				MatchedRules: pipeline.MatchedRules{
 					{
 						"id": "test",
 					},
@@ -300,8 +300,8 @@ func TestAppsecEventToContext(t *testing.T) {
 			contextToSend: map[string][]string{
 				"foobarxx": {"req.Header.Values('Foobar')"},
 			},
-			match: types.AppsecEvent{
-				MatchedRules: types.MatchedRules{
+			match: pipeline.AppsecEvent{
+				MatchedRules: pipeline.MatchedRules{
 					{
 						"id": "test",
 					},
@@ -326,8 +326,8 @@ func TestAppsecEventToContext(t *testing.T) {
 			contextToSend: map[string][]string{
 				"foobarxx": {"len(req.Header.Values('Foobar'))"},
 			},
-			match: types.AppsecEvent{
-				MatchedRules: types.MatchedRules{
+			match: pipeline.AppsecEvent{
+				MatchedRules: pipeline.MatchedRules{
 					{
 						"id": "test",
 					},
@@ -352,8 +352,8 @@ func TestAppsecEventToContext(t *testing.T) {
 			contextToSend: map[string][]string{
 				"ja4h": {"JA4H(req)"},
 			},
-			match: types.AppsecEvent{
-				MatchedRules: types.MatchedRules{
+			match: pipeline.AppsecEvent{
+				MatchedRules: pipeline.MatchedRules{
 					{
 						"id": "test",
 					},
@@ -404,8 +404,8 @@ func TestEvalAlertContextRules(t *testing.T) {
 	tests := []struct {
 		name           string
 		contextToSend  map[string][]string
-		event          types.Event
-		match          types.MatchedRule
+		event          pipeline.Event
+		match          pipeline.MatchedRule
 		req            *http.Request
 		expectedResult map[string][]string
 		expectedErrLen int
@@ -416,7 +416,7 @@ func TestEvalAlertContextRules(t *testing.T) {
 				"source_ip": {"evt.Parsed.source_ip"},
 				"id":        {"match.id"},
 			},
-			event: types.Event{
+			event: pipeline.Event{
 				Parsed: map[string]string{
 					"source_ip":      "1.2.3.4",
 					"source_machine": "mymachine",

pkg/appsec/appsec.go

@@ -13,7 +13,7 @@ import (
 
 	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
 	"github.com/crowdsecurity/crowdsec/pkg/exprhelpers"
-	"github.com/crowdsecurity/crowdsec/pkg/types"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
 )
 
 type Hook struct {
@@ -49,7 +49,7 @@ func (h *Hook) Build(hookStage int) error {
 	case hookPostEval:
 		ctx = GetPostEvalEnv(&AppsecRuntimeConfig{}, &ParsedRequest{})
 	case hookOnMatch:
-		ctx = GetOnMatchEnv(&AppsecRuntimeConfig{}, &ParsedRequest{}, types.Event{})
+		ctx = GetOnMatchEnv(&AppsecRuntimeConfig{}, &ParsedRequest{}, pipeline.Event{})
 	}
 
 	opts := exprhelpers.GetExprOptions(ctx)
@@ -445,7 +445,7 @@ func (w *AppsecRuntimeConfig) ProcessOnLoadRules() error {
 	return nil
 }
 
-func (w *AppsecRuntimeConfig) ProcessOnMatchRules(request *ParsedRequest, evt types.Event) error {
+func (w *AppsecRuntimeConfig) ProcessOnMatchRules(request *ParsedRequest, evt pipeline.Event) error {
 	has_match := false
 
 	for _, rule := range w.CompiledOnMatch {