feat: File notification plugin (#2932)

* wip: basic impl of file notification no log rotate but might now do it 🤷

* wip: ticker to 2 seconds and lower some log levels

* wip: remove redundant logrus formatter

* wip: the plugin should not handle it own data queue since the plugin process may timeout, so instead have a function that uses said context and loop whilst locking the filewriter this may not be the best way 🤷, however, I dont want multiple notifications to attempt to reopen the file if it has been rotated outside of the plugin context

* wip: impl log rotation which checks on check append, however, this may cause some issues in slow systems as the mutex lock doesnt give up until the file is rotated, however, the plugin looks for context and will give up if the plugin broker decides its timeout and will retry once the plugin has pushed again

* wip: update yaml dep

* wip: me no english great

* wip: even if the file has been rotated outside our control we should still compute the file size

* wip: improve context handling with creating a custom io writer struct which checks the context before attempting to write

* wip: used return byte count instead of calling a conversion again

* wip: actually check the enabled flag on log rotate

* wip: changed my mind, we check when we check file size

* wip: use io copy instead for memory alloc

* fix: add notification file to deb/rpm build

cmd/notification-file/Makefile

@@ -0,0 +1,17 @@
+ifeq ($(OS), Windows_NT)
+	SHELL := pwsh.exe
+	.SHELLFLAGS := -NoProfile -Command
+	EXT = .exe
+endif
+
+GO = go
+GOBUILD = $(GO) build
+
+BINARY_NAME = notification-file$(EXT)
+
+build: clean
+	$(GOBUILD) $(LD_OPTS) -o $(BINARY_NAME)
+
+.PHONY: clean
+clean:
+	@$(RM) $(BINARY_NAME) $(WIN_IGNORE_ERR)

cmd/notification-file/file.yaml

@@ -0,0 +1,23 @@
+# Don't change this
+type: file
+
+name: file_default # this must match with the registered plugin in the profile
+log_level: info # Options include: trace, debug, info, warn, error, off
+
+# This template render all events as ndjson
+format: |
+  {{range . -}}
+   { "time": "{{.StopAt}}", "program": "crowdsec", "alert": {{. | toJson }} }
+  {{ end -}}
+
+# group_wait: # duration to wait collecting alerts before sending to this plugin, eg "30s"
+# group_threshold: # if alerts exceed this, then the plugin will be sent the message. eg "10"
+
+#Use full path EG /tmp/crowdsec_alerts.json or %TEMP%\crowdsec_alerts.json
+log_path: "/tmp/crowdsec_alerts.json"
+rotate:
+  enabled: true # Change to false if you want to handle log rotate on system basis
+  max_size: 500 # in MB
+  max_files: 5
+  max_age: 5
+  compress: true

cmd/notification-file/main.go

@@ -0,0 +1,250 @@
+package main
+
+import (
+	"compress/gzip"
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/crowdsecurity/crowdsec/pkg/protobufs"
+	"github.com/hashicorp/go-hclog"
+	plugin "github.com/hashicorp/go-plugin"
+	"gopkg.in/yaml.v3"
+)
+
+var (
+	FileWriter     *os.File
+	FileWriteMutex *sync.Mutex
+	FileSize       int64
+)
+
+type FileWriteCtx struct {
+	Ctx    context.Context
+	Writer io.Writer
+}
+
+func (w *FileWriteCtx) Write(p []byte) (n int, err error) {
+	if err := w.Ctx.Err(); err != nil {
+		return 0, err
+	}
+	return w.Writer.Write(p)
+}
+
+type PluginConfig struct {
+	Name      string    `yaml:"name"`
+	LogLevel  string    `yaml:"log_level"`
+	LogPath   string    `yaml:"log_path"`
+	LogRotate LogRotate `yaml:"rotate"`
+}
+
+type LogRotate struct {
+	MaxSize  int  `yaml:"max_size"`
+	MaxAge   int  `yaml:"max_age"`
+	MaxFiles int  `yaml:"max_files"`
+	Enabled  bool `yaml:"enabled"`
+	Compress bool `yaml:"compress"`
+}
+
+type FilePlugin struct {
+	PluginConfigByName map[string]PluginConfig
+}
+
+var logger hclog.Logger = hclog.New(&hclog.LoggerOptions{
+	Name:       "file-plugin",
+	Level:      hclog.LevelFromString("INFO"),
+	Output:     os.Stderr,
+	JSONFormat: true,
+})
+
+func (r *LogRotate) rotateLogs(cfg PluginConfig) {
+	// Rotate the log file
+	err := r.rotateLogFile(cfg.LogPath, r.MaxFiles)
+	if err != nil {
+		logger.Error("Failed to rotate log file", "error", err)
+	}
+	// Reopen the FileWriter
+	FileWriter.Close()
+	FileWriter, err = os.OpenFile(cfg.LogPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		logger.Error("Failed to reopen log file", "error", err)
+	}
+	// Reset the file size
+	FileInfo, err := FileWriter.Stat()
+	if err != nil {
+		logger.Error("Failed to get file info", "error", err)
+	}
+	FileSize = FileInfo.Size()
+}
+
+func (r *LogRotate) rotateLogFile(logPath string, maxBackups int) error {
+	// Rename the current log file
+	backupPath := logPath + "." + time.Now().Format("20060102-150405")
+	err := os.Rename(logPath, backupPath)
+	if err != nil {
+		return err
+	}
+	glob := logPath + ".*"
+	if r.Compress {
+		glob = logPath + ".*.gz"
+		err = compressFile(backupPath)
+		if err != nil {
+			return err
+		}
+	}
+
+	// Remove old backups
+	files, err := filepath.Glob(glob)
+	if err != nil {
+		return err
+	}
+
+	sort.Sort(sort.Reverse(sort.StringSlice(files)))
+
+	for i, file := range files {
+		logger.Trace("Checking file", "file", file, "index", i, "maxBackups", maxBackups)
+		if i >= maxBackups {
+			logger.Trace("Removing file as over max backup count", "file", file)
+			os.Remove(file)
+		} else {
+			// Check the age of the file
+			fileInfo, err := os.Stat(file)
+			if err != nil {
+				return err
+			}
+			age := time.Since(fileInfo.ModTime()).Hours()
+			if age > float64(r.MaxAge*24) {
+				logger.Trace("Removing file as age was over configured amount", "file", file, "age", age)
+				os.Remove(file)
+			}
+		}
+	}
+
+	return nil
+}
+
+func compressFile(src string) error {
+	// Open the source file for reading
+	srcFile, err := os.Open(src)
+	if err != nil {
+		return err
+	}
+	defer srcFile.Close()
+
+	// Create the destination file
+	dstFile, err := os.Create(src + ".gz")
+	if err != nil {
+		return err
+	}
+	defer dstFile.Close()
+
+	// Create a gzip writer
+	gw := gzip.NewWriter(dstFile)
+	defer gw.Close()
+
+	// Read the source file and write its contents to the gzip writer
+	_, err = io.Copy(gw, srcFile)
+	if err != nil {
+		return err
+	}
+
+	// Delete the original (uncompressed) backup file
+	err = os.Remove(src)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func WriteToFileWithCtx(ctx context.Context, cfg PluginConfig, log string) error {
+	FileWriteMutex.Lock()
+	defer FileWriteMutex.Unlock()
+	originalFileInfo, err := FileWriter.Stat()
+	if err != nil {
+		logger.Error("Failed to get file info", "error", err)
+	}
+	currentFileInfo, _ := os.Stat(cfg.LogPath)
+	if !os.SameFile(originalFileInfo, currentFileInfo) {
+		// The file has been rotated outside our control
+		logger.Info("Log file has been rotated or missing attempting to reopen it")
+		FileWriter.Close()
+		FileWriter, err = os.OpenFile(cfg.LogPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			return err
+		}
+		FileInfo, err := FileWriter.Stat()
+		if err != nil {
+			return err
+		}
+		FileSize = FileInfo.Size()
+		logger.Info("Log file has been reopened successfully")
+	}
+	n, err := io.WriteString(&FileWriteCtx{Ctx: ctx, Writer: FileWriter}, log)
+	if err == nil {
+		FileSize += int64(n)
+		if FileSize > int64(cfg.LogRotate.MaxSize)*1024*1024 && cfg.LogRotate.Enabled {
+			logger.Debug("Rotating log file", "file", cfg.LogPath)
+			// Rotate the log file
+			cfg.LogRotate.rotateLogs(cfg)
+		}
+	}
+	return err
+}
+
+func (s *FilePlugin) Notify(ctx context.Context, notification *protobufs.Notification) (*protobufs.Empty, error) {
+	if _, ok := s.PluginConfigByName[notification.Name]; !ok {
+		return nil, fmt.Errorf("invalid plugin config name %s", notification.Name)
+	}
+	cfg := s.PluginConfigByName[notification.Name]
+
+	return &protobufs.Empty{}, WriteToFileWithCtx(ctx, cfg, notification.Text)
+}
+
+func (s *FilePlugin) Configure(ctx context.Context, config *protobufs.Config) (*protobufs.Empty, error) {
+	d := PluginConfig{}
+	err := yaml.Unmarshal(config.Config, &d)
+	if err != nil {
+		logger.Error("Failed to unmarshal config", "error", err)
+		return &protobufs.Empty{}, err
+	}
+	FileWriteMutex = &sync.Mutex{}
+	FileWriter, err = os.OpenFile(d.LogPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		logger.Error("Failed to open log file", "error", err)
+		return &protobufs.Empty{}, err
+	}
+	FileInfo, err := FileWriter.Stat()
+	if err != nil {
+		logger.Error("Failed to get file info", "error", err)
+		return &protobufs.Empty{}, err
+	}
+	FileSize = FileInfo.Size()
+	s.PluginConfigByName[d.Name] = d
+	logger.SetLevel(hclog.LevelFromString(d.LogLevel))
+	return &protobufs.Empty{}, err
+}
+
+func main() {
+	var handshake = plugin.HandshakeConfig{
+		ProtocolVersion:  1,
+		MagicCookieKey:   "CROWDSEC_PLUGIN_KEY",
+		MagicCookieValue: os.Getenv("CROWDSEC_PLUGIN_KEY"),
+	}
+
+	sp := &FilePlugin{PluginConfigByName: make(map[string]PluginConfig)}
+	plugin.Serve(&plugin.ServeConfig{
+		HandshakeConfig: handshake,
+		Plugins: map[string]plugin.Plugin{
+			"file": &protobufs.NotifierPlugin{
+				Impl: sp,
+			},
+		},
+		GRPCServer: plugin.DefaultGRPCServer,
+		Logger:     logger,
+	})
+}

debian/install

@@ -11,3 +11,4 @@ cmd/notification-http/http.yaml          etc/crowdsec/notifications/
 cmd/notification-splunk/splunk.yaml      etc/crowdsec/notifications/
 cmd/notification-email/email.yaml        etc/crowdsec/notifications/
 cmd/notification-sentinel/sentinel.yaml  etc/crowdsec/notifications/
+cmd/notification-file/file.yaml          etc/crowdsec/notifications/

debian/rules

@@ -31,6 +31,7 @@ override_dh_auto_install:
 	install -m 551 cmd/notification-splunk/notification-splunk debian/crowdsec/usr/lib/crowdsec/plugins/
 	install -m 551 cmd/notification-email/notification-email debian/crowdsec/usr/lib/crowdsec/plugins/
 	install -m 551 cmd/notification-sentinel/notification-sentinel debian/crowdsec/usr/lib/crowdsec/plugins/
+	install -m 551 cmd/notification-file/notification-file debian/crowdsec/usr/lib/crowdsec/plugins/
 
 	cp cmd/crowdsec/crowdsec debian/crowdsec/usr/bin
 	cp cmd/crowdsec-cli/cscli debian/crowdsec/usr/bin

rpm/SPECS/crowdsec.spec

@@ -67,13 +67,14 @@ install -m 551 cmd/notification-http/notification-http %{buildroot}%{_libdir}/%{
 install -m 551 cmd/notification-splunk/notification-splunk %{buildroot}%{_libdir}/%{name}/plugins/
 install -m 551 cmd/notification-email/notification-email %{buildroot}%{_libdir}/%{name}/plugins/
 install -m 551 cmd/notification-sentinel/notification-sentinel %{buildroot}%{_libdir}/%{name}/plugins/
+install -m 551 cmd/notification-file/notification-file %{buildroot}%{_libdir}/%{name}/plugins/
 
 install -m 600 cmd/notification-slack/slack.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
 install -m 600 cmd/notification-http/http.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
 install -m 600 cmd/notification-splunk/splunk.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
 install -m 600 cmd/notification-email/email.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
 install -m 600 cmd/notification-sentinel/sentinel.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
-
+install -m 600 cmd/notification-file/file.yaml %{buildroot}%{_sysconfdir}/crowdsec/notifications/
 
 %clean
 rm -rf %{buildroot}
@@ -88,6 +89,7 @@ rm -rf %{buildroot}
 %{_libdir}/%{name}/plugins/notification-splunk
 %{_libdir}/%{name}/plugins/notification-email
 %{_libdir}/%{name}/plugins/notification-sentinel
+%{_libdir}/%{name}/plugins/notification-file
 %{_sysconfdir}/%{name}/patterns/linux-syslog
 %{_sysconfdir}/%{name}/patterns/ruby
 %{_sysconfdir}/%{name}/patterns/nginx
@@ -123,6 +125,7 @@ rm -rf %{buildroot}
 %config(noreplace) %{_sysconfdir}/%{name}/notifications/splunk.yaml
 %config(noreplace) %{_sysconfdir}/%{name}/notifications/email.yaml
 %config(noreplace) %{_sysconfdir}/%{name}/notifications/sentinel.yaml
+%config(noreplace) %{_sysconfdir}/%{name}/notifications/file.yaml
 %config(noreplace) %{_sysconfdir}/cron.daily/%{name}
 
 %{_unitdir}/%{name}.service

wizard.sh

@@ -82,12 +82,14 @@ SLACK_PLUGIN_BINARY="./cmd/notification-slack/notification-slack"
 SPLUNK_PLUGIN_BINARY="./cmd/notification-splunk/notification-splunk"
 EMAIL_PLUGIN_BINARY="./cmd/notification-email/notification-email"
 SENTINEL_PLUGIN_BINARY="./cmd/notification-sentinel/notification-sentinel"
+FILE_PLUGIN_BINARY="./cmd/notification-file/notification-file"
 
 HTTP_PLUGIN_CONFIG="./cmd/notification-http/http.yaml"
 SLACK_PLUGIN_CONFIG="./cmd/notification-slack/slack.yaml"
 SPLUNK_PLUGIN_CONFIG="./cmd/notification-splunk/splunk.yaml"
 EMAIL_PLUGIN_CONFIG="./cmd/notification-email/email.yaml"
 SENTINEL_PLUGIN_CONFIG="./cmd/notification-sentinel/sentinel.yaml"
+FILE_PLUGIN_CONFIG="./cmd/notification-file/file.yaml"
 
 
 BACKUP_DIR=$(mktemp -d)
@@ -525,6 +527,7 @@ install_plugins(){
     cp ${HTTP_PLUGIN_BINARY} ${CROWDSEC_PLUGIN_DIR}
     cp ${EMAIL_PLUGIN_BINARY} ${CROWDSEC_PLUGIN_DIR}
     cp ${SENTINEL_PLUGIN_BINARY} ${CROWDSEC_PLUGIN_DIR}
+    cp ${FILE_PLUGIN_BINARY} ${CROWDSEC_PLUGIN_DIR}
 
     if [[ ${DOCKER_MODE} == "false" ]]; then
         cp -n ${SLACK_PLUGIN_CONFIG} /etc/crowdsec/notifications/
@@ -532,6 +535,7 @@ install_plugins(){
         cp -n ${HTTP_PLUGIN_CONFIG} /etc/crowdsec/notifications/
         cp -n ${EMAIL_PLUGIN_CONFIG} /etc/crowdsec/notifications/
         cp -n ${SENTINEL_PLUGIN_CONFIG} /etc/crowdsec/notifications/
+        cp -n ${FILE_PLUGIN_CONFIG} /etc/crowdsec/notifications/
     fi
 }