Merge branch 'master' into releases/1.6.x

Author: mmetc

.github/workflows/publish-tarball-release.yml

@@ -1,4 +1,3 @@
-# .github/workflows/build-docker-image.yml
 name: Release
 
 on:

.golangci.yml

@@ -403,10 +403,6 @@ issues:
       path: pkg/acquisition/modules/victorialogs/internal/vlclient/vl_client.go
       text: "confusing-naming: Method 'QueryRange' differs only by capitalization to method 'queryRange' in the same source file"
 
-    - linters:
-        - revive
-      path: cmd/crowdsec-cli/copyfile.go
-
     - linters:
         - revive
       path: pkg/hubtest/hubtest_item.go

config/crowdsec.cron.daily

@@ -5,7 +5,7 @@ test -x /usr/bin/cscli || exit 0
 # splay hub upgrade and crowdsec reload
 sleep "$(seq 1 300 | shuf -n 1)"
 
-/usr/bin/cscli --error hub update
+/usr/bin/cscli --error hub update >/dev/null
 
 upgraded=$(/usr/bin/cscli --error hub upgrade)
 if [ -n "$upgraded" ]; then

pkg/acquisition/modules/appsec/appsec.go

@@ -63,7 +63,6 @@ type AppsecSource struct {
 	lapiURL               string
 	AuthCache             AuthCache
 	AppsecRunners         []AppsecRunner // one for each go-routine
-	apiClient             *apiclient.ApiClient
 	appsecAllowlistClient *allowlists.AppsecAllowlist
 }
 
@@ -226,12 +225,7 @@ func (w *AppsecSource) Configure(yamlConfig []byte, logger *log.Entry, metricsLe
 
 	w.AppsecRunners = make([]AppsecRunner, w.config.Routines)
 
-	w.apiClient, err = apiclient.GetLAPIClient()
-	if err != nil {
-		return fmt.Errorf("unable to get authenticated LAPI client: %w", err)
-	}
-
-	w.appsecAllowlistClient = allowlists.NewAppsecAllowlist(w.apiClient, w.logger)
+	w.appsecAllowlistClient = allowlists.NewAppsecAllowlist(w.logger)
 
 	for nbRoutine := range w.config.Routines {
 		appsecRunnerUUID := uuid.New().String()
@@ -282,7 +276,16 @@ func (w *AppsecSource) OneShotAcquisition(_ context.Context, _ chan types.Event,
 func (w *AppsecSource) StreamingAcquisition(ctx context.Context, out chan types.Event, t *tomb.Tomb) error {
 	w.outChan = out
 
-	w.appsecAllowlistClient.StartRefresh(t)
+	apiClient, err := apiclient.GetLAPIClient()
+	if err != nil {
+		return fmt.Errorf("unable to get authenticated LAPI client: %w", err)
+	}
+
+	err = w.appsecAllowlistClient.Start(ctx, apiClient)
+	if err != nil {
+		return fmt.Errorf("failed to fetch allowlists: %w", err)
+	}
+	w.appsecAllowlistClient.StartRefresh(ctx, t)
 
 	t.Go(func() error {
 		defer trace.CatchPanic("crowdsec/acquis/appsec/live")

pkg/acquisition/modules/appsec/appsec_test.go

@@ -146,10 +146,9 @@ func loadAppSecEngine(test appsecRuleTest, t *testing.T) {
 		assert.NoError(t, err)
 	})
 
-	allowlistClient := allowlists.NewAppsecAllowlist(client, logger)
-	// In real life, allowlists updater is started by the acquisition
-	// Do it manually here as we are simulating the appsec itself
-	err = allowlistClient.FetchAllowlists()
+	allowlistClient := allowlists.NewAppsecAllowlist(logger)
+
+	err = allowlistClient.Start(t.Context(), client)
 	require.NoError(t, err)
 	runner := AppsecRunner{
 		inChan:                 InChan,

pkg/acquisition/modules/syslog/syslog.go

@@ -1,6 +1,7 @@
 package syslogacquisition
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -27,6 +28,7 @@ type SyslogConfiguration struct {
 	Port                              int    `yaml:"listen_port,omitempty"`
 	Addr                              string `yaml:"listen_addr,omitempty"`
 	MaxMessageLen                     int    `yaml:"max_message_len,omitempty"`
+	DisableRFCParser                  bool   `yaml:"disable_rfc_parser,omitempty"` // if true, we don't try to be smart and just remove the PRI
 	configuration.DataSourceCommonCfg `yaml:",inline"`
 }
 
@@ -182,6 +184,66 @@ func (s *SyslogSource) buildLogFromSyslog(ts time.Time, hostname string,
 	return ret
 }
 
+func (s *SyslogSource) parseLine(syslogLine syslogserver.SyslogMessage) string {
+	var line string
+
+	logger := s.logger.WithField("client", syslogLine.Client)
+	logger.Tracef("raw: %s", syslogLine)
+	if s.metricsLevel != configuration.METRICS_NONE {
+		linesReceived.With(prometheus.Labels{"source": syslogLine.Client}).Inc()
+	}
+	if !s.config.DisableRFCParser {
+		p := rfc3164.NewRFC3164Parser(rfc3164.WithCurrentYear())
+		err := p.Parse(syslogLine.Message)
+		if err != nil {
+			logger.Debugf("could not parse as RFC3164 (%s)", err)
+			p2 := rfc5424.NewRFC5424Parser()
+			err = p2.Parse(syslogLine.Message)
+			if err != nil {
+				logger.Errorf("could not parse message: %s", err)
+				logger.Debugf("could not parse as RFC5424 (%s) : %s", err, syslogLine.Message)
+				return ""
+			}
+			line = s.buildLogFromSyslog(p2.Timestamp, p2.Hostname, p2.Tag, p2.PID, p2.Message)
+			if s.metricsLevel != configuration.METRICS_NONE {
+				linesParsed.With(prometheus.Labels{"source": syslogLine.Client, "type": "rfc5424"}).Inc()
+			}
+		} else {
+			line = s.buildLogFromSyslog(p.Timestamp, p.Hostname, p.Tag, p.PID, p.Message)
+			if s.metricsLevel != configuration.METRICS_NONE {
+				linesParsed.With(prometheus.Labels{"source": syslogLine.Client, "type": "rfc3164"}).Inc()
+			}
+		}
+	} else {
+		if len(syslogLine.Message) < 3 {
+			logger.Errorf("malformated message, missing PRI (message too short)")
+			return ""
+		}
+		if syslogLine.Message[0] != '<' {
+			logger.Errorf("malformated message, missing PRI beginning")
+			return ""
+		}
+		priEnd := bytes.Index(syslogLine.Message, []byte(">"))
+		if priEnd == -1 {
+			logger.Errorf("malformated message, missing PRI end")
+			return ""
+		}
+		if priEnd > 4 {
+			logger.Errorf("malformated message, PRI too long")
+			return ""
+		}
+		for i := 1; i < priEnd; i++ {
+			if syslogLine.Message[i] < '0' || syslogLine.Message[i] > '9' {
+				logger.Errorf("malformated message, PRI not a number")
+				return ""
+			}
+		}
+		line = string(syslogLine.Message[priEnd+1:])
+	}
+
+	return strings.TrimSuffix(line, "\n")
+}
+
 func (s *SyslogSource) handleSyslogMsg(out chan types.Event, t *tomb.Tomb, c chan syslogserver.SyslogMessage) error {
 	killed := false
 	for {
@@ -196,37 +258,12 @@ func (s *SyslogSource) handleSyslogMsg(out chan types.Event, t *tomb.Tomb, c cha
 			s.logger.Info("Syslog server has exited")
 			return nil
 		case syslogLine := <-c:
-			var line string
-			var ts time.Time
-
-			logger := s.logger.WithField("client", syslogLine.Client)
-			logger.Tracef("raw: %s", syslogLine)
-			if s.metricsLevel != configuration.METRICS_NONE {
-				linesReceived.With(prometheus.Labels{"source": syslogLine.Client}).Inc()
-			}
-			p := rfc3164.NewRFC3164Parser(rfc3164.WithCurrentYear())
-			err := p.Parse(syslogLine.Message)
-			if err != nil {
-				logger.Debugf("could not parse as RFC3164 (%s)", err)
-				p2 := rfc5424.NewRFC5424Parser()
-				err = p2.Parse(syslogLine.Message)
-				if err != nil {
-					logger.Errorf("could not parse message: %s", err)
-					logger.Debugf("could not parse as RFC5424 (%s) : %s", err, syslogLine.Message)
-					continue
-				}
-				line = s.buildLogFromSyslog(p2.Timestamp, p2.Hostname, p2.Tag, p2.PID, p2.Message)
-				if s.metricsLevel != configuration.METRICS_NONE {
-					linesParsed.With(prometheus.Labels{"source": syslogLine.Client, "type": "rfc5424"}).Inc()
-				}
-			} else {
-				line = s.buildLogFromSyslog(p.Timestamp, p.Hostname, p.Tag, p.PID, p.Message)
-				if s.metricsLevel != configuration.METRICS_NONE {
-					linesParsed.With(prometheus.Labels{"source": syslogLine.Client, "type": "rfc3164"}).Inc()
-				}
+			line := s.parseLine(syslogLine)
+			if line == "" {
+				continue
 			}
 
-			line = strings.TrimSuffix(line, "\n")
+			var ts time.Time
 
 			l := types.Line{}
 			l.Raw = line

pkg/acquisition/modules/syslog/syslog_test.go

@@ -120,6 +120,26 @@ listen_addr: 127.0.0.1`,
 				`<13>May 18 12:37:56 mantis sshd`,
 			},
 		},
+		{
+			name: "RFC3164 - no parsing",
+			config: `source: syslog
+listen_port: 4242
+listen_addr: 127.0.0.1
+disable_rfc_parser: true`,
+			expectedLines: 5,
+			logs: []string{
+				`<13>May 18 12:37:56 mantis sshd[49340]: blabla2[foobar]`,
+				`<13>May 18 12:37:56 mantis sshd[49340]: blabla2`,
+				`<13>May 18 12:37:56 mantis sshd: blabla2`,
+				`<13>May 18 12:37:56 mantis sshd`,
+				`<999>May 18 12:37:56 mantis sshd`,
+				`<1000>May 18 12:37:56 mantis sshd`,
+				`>?> asd`,
+				`<asd>asdasd`,
+				`<1a asd`,
+				`<123123>asdasd`,
+			},
+		},
 	}
 	if runtime.GOOS != "windows" {
 		tests = append(tests, struct {

pkg/appsec/allowlists/allowlists.go

@@ -36,25 +36,26 @@ type AppsecAllowlist struct {
 	tomb       *tomb.Tomb
 }
 
-func NewAppsecAllowlist(client *apiclient.ApiClient, logger *log.Entry) *AppsecAllowlist {
+func NewAppsecAllowlist(logger *log.Entry) *AppsecAllowlist {
 	a := &AppsecAllowlist{
-		LAPIClient: client,
-		logger:     logger.WithField("component", "appsec-allowlist"),
-		ips:        []ipAllowlist{},
-		ranges:     []rangeAllowlist{},
-	}
-
-	if err := a.FetchAllowlists(); err != nil {
-		a.logger.Errorf("failed to fetch allowlists: %s", err)
+		logger: logger.WithField("component", "appsec-allowlist"),
+		ips:    []ipAllowlist{},
+		ranges: []rangeAllowlist{},
 	}
 
 	return a
 }
 
-func (a *AppsecAllowlist) FetchAllowlists() error {
+func (a *AppsecAllowlist) Start(ctx context.Context, client *apiclient.ApiClient) error {
+	a.LAPIClient = client
+	err := a.FetchAllowlists(ctx)
+	return err
+}
+
+func (a *AppsecAllowlist) FetchAllowlists(ctx context.Context) error {
 	a.logger.Debug("fetching allowlists")
 
-	allowlists, _, err := a.LAPIClient.Allowlists.List(context.TODO(), apiclient.AllowlistListOpts{WithContent: true})
+	allowlists, _, err := a.LAPIClient.Allowlists.List(ctx, apiclient.AllowlistListOpts{WithContent: true})
 	if err != nil {
 		return err
 	}
@@ -92,32 +93,38 @@ func (a *AppsecAllowlist) FetchAllowlists() error {
 		}
 	}
 
+	if len(a.ips) != 0 || len(a.ranges) != 0 {
+		a.logger.Infof("fetched %d IPs and %d ranges", len(a.ips), len(a.ranges))
+	}
 	a.logger.Debugf("fetched %d IPs and %d ranges", len(a.ips), len(a.ranges))
 	a.logger.Tracef("allowlisted ips: %+v", a.ips)
 	a.logger.Tracef("allowlisted ranges: %+v", a.ranges)
 
 	return nil
 }
 
-func (a *AppsecAllowlist) updateAllowlists() error {
+func (a *AppsecAllowlist) updateAllowlists(ctx context.Context) {
 	ticker := time.NewTicker(allowlistRefreshInterval)
 
 	for {
 		select {
 		case <-ticker.C:
-			if err := a.FetchAllowlists(); err != nil {
+			if err := a.FetchAllowlists(ctx); err != nil {
 				a.logger.Errorf("failed to fetch allowlists: %s", err)
 			}
 		case <-a.tomb.Dying():
 			ticker.Stop()
-			return nil
+			return
 		}
 	}
 }
 
-func (a *AppsecAllowlist) StartRefresh(t *tomb.Tomb) {
+func (a *AppsecAllowlist) StartRefresh(ctx context.Context, t *tomb.Tomb) {
 	a.tomb = t
-	a.tomb.Go(a.updateAllowlists)
+	a.tomb.Go(func() error {
+		a.updateAllowlists(ctx)
+		return nil
+	})
 }
 
 func (a *AppsecAllowlist) IsAllowlisted(sourceIP string) (bool, string) {

pkg/appsec/allowlists/allowlists_test.go

@@ -64,9 +64,13 @@ func TestAppsecAllowlist(t *testing.T) {
 		assert.NoError(t, err)
 	})
 
-	allowlistClient := NewAppsecAllowlist(client, log.NewEntry(log.StandardLogger()))
+	ctx := t.Context()
+	allowlistClient := NewAppsecAllowlist(log.NewEntry(log.StandardLogger()))
 
-	err = allowlistClient.FetchAllowlists()
+	err = allowlistClient.Start(ctx, client)
+	require.NoError(t, err)
+
+	err = allowlistClient.FetchAllowlists(ctx)
 	require.NoError(t, err)
 
 	res, reason := allowlistClient.IsAllowlisted("1.2.3.4")
@@ -84,7 +88,7 @@ func TestAppsecAllowlist(t *testing.T) {
 	assert.Len(t, allowlistClient.ips, 1)
 	assert.Len(t, allowlistClient.ranges, 1)
 
-	err = allowlistClient.FetchAllowlists()
+	err = allowlistClient.FetchAllowlists(ctx)
 	require.NoError(t, err)
 
 	// No duplicates should be added

test/bats-detect/apache2-deb.bats

@@ -1,5 +1,4 @@
 #!/usr/bin/env bats
-# vim: ft=bats:list:ts=8:sts=4:sw=4:et:ai:si:
 
 set -u