Merge branch 'master' into releases/1.6.x

Author: mmetc

.github/release-drafter.yml

@@ -23,6 +23,8 @@ categories:
       - 'kind/dependencies'
       - 'kind/chore'
 tag-template: "- $TITLE @$AUTHOR (#$NUMBER)"
+exclude-labels:
+  - 'skip-changelog'
 template: |
   ## Changes
 
@@ -31,8 +33,8 @@ template: |
   ## Geolite2 notice
 
   This product includes GeoLite2 data created by MaxMind, available from <a href="https://www.maxmind.com">https://www.maxmind.com</a>.
-  
+
   ## Installation
 
   Take a look at the [installation instructions](https://doc.crowdsec.net/docs/getting_started/install_crowdsec).
-  
+

.github/workflows/ci_release-drafter.yml

@@ -4,18 +4,24 @@ on:
   push:
     # branches to consider in the event; optional, defaults to all
     branches:
+      - main
       - master
       - releases/**
 
+permissions:
+  contents: read
+
 jobs:
   update_release_draft:
+    permissions:
+      contents: write
+      pull-requests: read
     runs-on: ubuntu-latest
     steps:
       # Drafts your next Release notes as Pull Requests are merged into "master"
       - uses: release-drafter/release-drafter@v6
         with:
-          config-name: release-drafter.yml
           # (Optional) specify config name to use, relative to .github/. Default: release-drafter.yml
-          # config-name: my-config.yml
+          config-name: release-drafter.yml
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

Makefile

@@ -317,7 +317,7 @@ check_golangci-lint:
 ifeq ($(OS),Windows_NT)
 	@where golangci-lint >nul || (echo "Error: golangci-lint is not installed. Install it from https://github.com/golangci/golangci-lint" && exit 1)
 else
-	@command -v galangci-lint > /dev/null 2>&1 || (echo "Error: golangci-lint is not installed. Install it from https://github.com/golangci/golangci-lint" && exit 1)
+	@command -v golangci-lint > /dev/null 2>&1 || (echo "Error: golangci-lint is not installed. Install it from https://github.com/golangci/golangci-lint" && exit 1)
 endif
 
 .PHONY: lint

cmd/crowdsec-cli/clicapi/capi.go

@@ -21,6 +21,7 @@ import (
 	"github.com/crowdsecurity/crowdsec/pkg/apiclient"
 	"github.com/crowdsecurity/crowdsec/pkg/csconfig"
 	"github.com/crowdsecurity/crowdsec/pkg/cwhub"
+	"github.com/crowdsecurity/crowdsec/pkg/database"
 	"github.com/crowdsecurity/crowdsec/pkg/models"
 	"github.com/crowdsecurity/crowdsec/pkg/types"
 )
@@ -120,7 +121,7 @@ func (cli *cliCapi) register(ctx context.Context, capiUserPrefix string, outputF
 
 		log.Infof("Central API credentials written to '%s'", dumpFile)
 	} else {
-		fmt.Println(string(apiConfigDump))
+		fmt.Fprintln(os.Stdout, string(apiConfigDump))
 	}
 
 	if msg := reload.UserMessage(); msg != "" {
@@ -154,8 +155,8 @@ func (cli *cliCapi) newRegisterCmd() *cobra.Command {
 	return cmd
 }
 
-// queryCAPIStatus checks if the Central API is reachable, and if the credentials are correct. It then checks if the instance is enrolle in the console.
-func queryCAPIStatus(ctx context.Context, hub *cwhub.Hub, credURL string, login string, password string) (bool, bool, error) {
+// queryCAPIStatus checks if the Central API is reachable, and if the credentials are correct. It then checks if the instance is enrolled in the console.
+func queryCAPIStatus(ctx context.Context, db *database.Client, hub *cwhub.Hub, credURL string, login string, password string) (bool, bool, error) {
 	apiURL, err := url.Parse(credURL)
 	if err != nil {
 		return false, false, err
@@ -172,7 +173,6 @@ func queryCAPIStatus(ctx context.Context, hub *cwhub.Hub, credURL string, login
 	client, err := apiclient.NewClient(&apiclient.Config{
 		MachineID: login,
 		Password:  passwd,
-		Scenarios: itemsForAPI,
 		URL:       apiURL,
 		// I don't believe papi is neede to check enrollement
 		// PapiURL:       papiURL,
@@ -198,6 +198,10 @@ func queryCAPIStatus(ctx context.Context, hub *cwhub.Hub, credURL string, login
 		return false, false, err
 	}
 
+	if err := db.SaveAPICToken(ctx, apiclient.TokenDBField, authResp.Token); err != nil {
+		return false, false, err
+	}
+
 	client.GetClient().Transport.(*apiclient.JWTTransport).Token = authResp.Token
 
 	if client.IsEnrolled() {
@@ -207,7 +211,7 @@ func queryCAPIStatus(ctx context.Context, hub *cwhub.Hub, credURL string, login
 	return true, false, nil
 }
 
-func (cli *cliCapi) Status(ctx context.Context, out io.Writer, hub *cwhub.Hub) error {
+func (cli *cliCapi) Status(ctx context.Context, db *database.Client, out io.Writer, hub *cwhub.Hub) error {
 	cfg := cli.cfg()
 
 	if err := require.CAPIRegistered(cfg); err != nil {
@@ -219,7 +223,7 @@ func (cli *cliCapi) Status(ctx context.Context, out io.Writer, hub *cwhub.Hub) e
 	fmt.Fprintf(out, "Loaded credentials from %s\n", cfg.API.Server.OnlineClient.CredentialsFilePath)
 	fmt.Fprintf(out, "Trying to authenticate with username %s on %s\n", cred.Login, cred.URL)
 
-	auth, enrolled, err := queryCAPIStatus(ctx, hub, cred.URL, cred.Login, cred.Password)
+	auth, enrolled, err := queryCAPIStatus(ctx, db, hub, cred.URL, cred.Login, cred.Password)
 	if err != nil {
 		return fmt.Errorf("failed to authenticate to Central API (CAPI): %w", err)
 	}
@@ -263,12 +267,20 @@ func (cli *cliCapi) newStatusCmd() *cobra.Command {
 		Args:              args.NoArgs,
 		DisableAutoGenTag: true,
 		RunE: func(cmd *cobra.Command, _ []string) error {
-			hub, err := require.Hub(cli.cfg(), nil)
+			cfg := cli.cfg()
+			ctx := cmd.Context()
+
+			hub, err := require.Hub(cfg, nil)
+			if err != nil {
+				return err
+			}
+
+			db, err := require.DBClient(ctx, cfg.DbConfig)
 			if err != nil {
 				return err
 			}
 
-			return cli.Status(cmd.Context(), color.Output, hub)
+			return cli.Status(ctx, db, color.Output, hub)
 		},
 	}
 

cmd/crowdsec-cli/cliconsole/console.go

@@ -9,8 +9,8 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"strconv"
 	"slices"
+	"strconv"
 	"strings"
 
 	"github.com/fatih/color"
@@ -85,9 +85,11 @@ func (cli *cliConsole) enroll(ctx context.Context, key string, name string, over
 	c, _ := apiclient.NewClient(&apiclient.Config{
 		MachineID:     cli.cfg().API.Server.OnlineClient.Credentials.Login,
 		Password:      password,
-		Scenarios:     hub.GetInstalledListForAPI(),
 		URL:           apiURL,
 		VersionPrefix: "v3",
+		UpdateScenario: func(_ context.Context) ([]string, error) {
+			return hub.GetInstalledListForAPI(), nil
+		},
 	})
 
 	resp, err := c.Auth.EnrollWatcher(ctx, key, name, tags, overwrite)
@@ -157,12 +159,14 @@ func optionFilterDisable(opts []string, disableOpts []string) ([]string, error)
 		// discard all elements == opt
 
 		j := 0
+
 		for _, o := range opts {
 			if o != opt {
 				opts[j] = o
 				j++
 			}
 		}
+
 		opts = opts[:j]
 	}
 
@@ -323,7 +327,7 @@ func (cli *cliConsole) newStatusCmd() *cobra.Command {
 				if err != nil {
 					return fmt.Errorf("failed to serialize configuration: %w", err)
 				}
-				fmt.Println(string(data))
+				fmt.Fprintln(os.Stdout, string(data))
 			case "raw":
 				csvwriter := csv.NewWriter(os.Stdout)
 				err := csvwriter.Write([]string{"option", "enabled"})

cmd/crowdsec-cli/climetrics/metrics.go

@@ -28,7 +28,7 @@ func (cli *cliMetrics) NewCommand() *cobra.Command {
 		Use:   "metrics",
 		Short: "Display crowdsec prometheus metrics.",
 		Long:  `Fetch metrics from a Local API server and display them`,
-		Example: `# Show all Metrics, skip empty tables (same as "cecli metrics show")
+		Example: `# Show all Metrics, skip empty tables (same as "cscli metrics show")
 cscli metrics
 
 # Show only some metrics, connect to a different url

cmd/crowdsec-cli/clisupport/support.go

@@ -256,13 +256,13 @@ func (cli *cliSupport) dumpLAPIStatus(ctx context.Context, zw *zip.Writer, hub *
 	return nil
 }
 
-func (cli *cliSupport) dumpCAPIStatus(ctx context.Context, zw *zip.Writer, hub *cwhub.Hub) error {
+func (cli *cliSupport) dumpCAPIStatus(ctx context.Context, zw *zip.Writer, hub *cwhub.Hub, db *database.Client) error {
 	log.Info("Collecting CAPI status")
 
 	out := new(bytes.Buffer)
 	cc := clicapi.New(cli.cfg)
 
-	err := cc.Status(ctx, out, hub)
+	err := cc.Status(ctx, db, out, hub)
 	if err != nil {
 		fmt.Fprintf(out, "%s\n", err)
 	}
@@ -534,7 +534,7 @@ func (cli *cliSupport) dump(ctx context.Context, outFile string) error {
 	}
 
 	if !skipCAPI {
-		if err = cli.dumpCAPIStatus(ctx, zipWriter, hub); err != nil {
+		if err = cli.dumpCAPIStatus(ctx, zipWriter, hub, db); err != nil {
 			log.Warnf("could not collect CAPI status: %s", err)
 		}
 

cmd/crowdsec/api.go

@@ -38,7 +38,7 @@ func initAPIServer(ctx context.Context, cConfig *csconfig.Config) (*apiserver.AP
 
 		err = pluginBroker.Init(ctx, cConfig.PluginConfig, cConfig.API.Server.Profiles, cConfig.ConfigPaths)
 		if err != nil {
-			return nil, fmt.Errorf("unable to run plugin broker: %w", err)
+			return nil, fmt.Errorf("plugin broker: %w", err)
 		}
 
 		log.Info("initiated plugin broker")

cmd/crowdsec/lapiclient.go

@@ -33,12 +33,13 @@ var linesRead = prometheus.NewCounterVec(
 	[]string{"topic"})
 
 type KafkaConfiguration struct {
-	Brokers                           []string   `yaml:"brokers"`
-	Topic                             string     `yaml:"topic"`
-	GroupID                           string     `yaml:"group_id"`
-	Partition                         int        `yaml:"partition"`
-	Timeout                           string     `yaml:"timeout"`
-	TLS                               *TLSConfig `yaml:"tls"`
+	Brokers                           []string                `yaml:"brokers"`
+	Topic                             string                  `yaml:"topic"`
+	GroupID                           string                  `yaml:"group_id"`
+	Partition                         int                     `yaml:"partition"`
+	Timeout                           string                  `yaml:"timeout"`
+	TLS                               *TLSConfig              `yaml:"tls"`
+	BatchConfiguration                KafkaBatchConfiguration `yaml:"batch"`
 	configuration.DataSourceCommonCfg `yaml:",inline"`
 }
 
@@ -49,6 +50,14 @@ type TLSConfig struct {
 	CaCert             string `yaml:"ca_cert"`
 }
 
+type KafkaBatchConfiguration struct {
+	BatchMinBytes  int           `yaml:"min_bytes"`
+	BatchMaxBytes  int           `yaml:"max_bytes"`
+	BatchMaxWait   time.Duration `yaml:"max_wait"`
+	BatchQueueSize int           `yaml:"queue_size"`
+	CommitInterval time.Duration `yaml:"commit_interval"`
+}
+
 type KafkaSource struct {
 	metricsLevel int
 	Config       KafkaConfiguration
@@ -294,6 +303,22 @@ func (kc *KafkaConfiguration) NewReader(dialer *kafka.Dialer, logger *log.Entry)
 		logger.Warnf("no group_id specified, crowdsec will only read from the 1st partition of the topic")
 	}
 
+	if kc.BatchConfiguration.BatchMinBytes != 0 {
+		rConf.MinBytes = kc.BatchConfiguration.BatchMinBytes
+	}
+	if kc.BatchConfiguration.BatchMaxBytes != 0 {
+		rConf.MaxBytes = kc.BatchConfiguration.BatchMaxBytes
+	}
+	if kc.BatchConfiguration.BatchMaxWait != 0 {
+		rConf.MaxWait = kc.BatchConfiguration.BatchMaxWait
+	}
+	if kc.BatchConfiguration.BatchQueueSize != 0 {
+		rConf.QueueCapacity = kc.BatchConfiguration.BatchQueueSize
+	}
+	if kc.BatchConfiguration.CommitInterval != 0 {
+		rConf.CommitInterval = kc.BatchConfiguration.CommitInterval
+	}
+
 	if err := rConf.Validate(); err != nil {
 		return &kafka.Reader{}, fmt.Errorf("while validating reader configuration: %w", err)
 	}