allow to set a template for the nats/stan subjects

Signed-off-by: Thomas Labarussias <[email protected]>
falcosecurity · Jan 23, 2025 · 9892800 · 9892800
1 parent 2424be2
commit 9892800
Show file tree

Hide file tree

Showing 7 changed files with 60 additions and 30 deletions.
diff --git a/config.go b/config.go
@@ -4,7 +4,6 @@ package main
 
 import (
 	"fmt"
-	"github.com/falcosecurity/falcosidekick/outputs/otlpmetrics"
 	"log"
 	"net"
 	"os"
@@ -17,6 +16,8 @@ import (
 	"text/template"
 	"time"
 
+	"github.com/falcosecurity/falcosidekick/outputs/otlpmetrics"
+
 	kingpin "github.com/alecthomas/kingpin/v2"
 	"github.com/spf13/viper"
 
@@ -156,14 +157,14 @@ var httpOutputDefaults = map[string]map[string]any{
 		"Name":            "",
 	},
 	"STAN": {
-		"HostPort":  "",
-		"ClusterID": "",
-		"ClientID":  "",
+		"HostPort":        "",
+		"ClusterID":       "",
+		"ClientID":        "",
+		"SubjectTemplate": "falco.<priority>.<rule>",
 	},
 	"NATS": {
-		"HostPort":  "",
-		"ClusterID": "",
-		"ClientID":  "",
+		"HostPort":        "",
+		"SubjectTemplate": "falco.<priority>.<rule>",
 	},
 	"Opsgenie": {
 		"Region":          "us",
@@ -589,8 +590,6 @@ func getConfig() *types.Configuration {
 	v.SetDefault("OTLP.Metrics.CheckCert", true)
 	v.SetDefault("OTLP.Metrics.ExtraAttributes", "")
 
-	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
-	v.AutomaticEnv()
 	if *configFile != "" {
 		d, f := path.Split(*configFile)
 		if d == "" {
@@ -603,6 +602,8 @@ func getConfig() *types.Configuration {
 			log.Printf("[ERROR] : Error when reading config file : %v\n", err)
 		}
 	}
+	v.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
+	v.AutomaticEnv()
 
 	v.GetStringSlice("TLSServer.NoTLSPaths")
 	v.GetStringSlice("Customtags")

diff --git a/config_example.yaml b/config_example.yaml
@@ -163,6 +163,7 @@ loki:
 
 nats:
   # hostport: "" # nats://{domain or ip}:{port}, if not empty, NATS output is enabled
+  # subjecttemplate: "falco.<priority>.<rule>" # template for the subject, tokens <priority> and <rule> will be automatically replaced (default: falco.<priority>.<rule>)
   # minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|error|warning|notice|informational|debug or "" (default)
   # mutualtls: false # if true, checkcert flag will be ignored (server cert will always be checked)
   # checkcert: true # check if ssl certificate of the output is valid (default: true)
@@ -171,6 +172,7 @@ stan:
   # hostport: "" # nats://{domain or ip}:{port}, if not empty, STAN output is enabled
   # clusterid: "" # Cluster name, if not empty, STAN output is enabled
   # clientid: "" # Client ID, if not empty, STAN output is enabled
+  # subjecttemplate: "falco.<priority>.<rule>" # template for the subject, tokens <priority> and <rule> will be automatically replaced (default: falco.<priority>.<rule>)
   # minimumpriority: "" # minimum priority of event for using this output, order is emergency|alert|critical|error|warning|notice|informational|debug or "" (default)
   # mutualtls: false # if true, checkcert flag will be ignored (server cert will always be checked)
   # checkcert: true # check if ssl certificate of the output is valid (default: true)

diff --git a/docs/outputs/nats.md b/docs/outputs/nats.md
@@ -8,18 +8,23 @@
 - [NATS](#nats)
   - [Table of content](#table-of-content)
   - [Configuration](#configuration)
+- [subjecttemplate: "falco.." # template for the subject, tokens  and  will be automatically replaced (default: falco..)](#subjecttemplate-falco--template-for-the-subject-tokens--and--will-be-automatically-replaced-default-falco)
   - [Example of config.yaml](#example-of-configyaml)
   - [Additional info](#additional-info)
   - [Screenshots](#screenshots)
 
 ## Configuration
 
-| Setting                | Env var                | Default value    | Description                                                                                                                         |
-| ---------------------- | ---------------------- | ---------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
-| `nats.hostport`        | `NATS_HOSTPORT`        |                  | nats://{domain or ip}:{port}, if not empty, NATS output is **enabled**                                                              |
-| `nats.mutualtls`       | `NATS_MUTUALTLS`       | `false`          | Authenticate to the output with TLS, if true, checkcert flag will be ignored (server cert will always be checked)                   |
-| `nats.checkcert`       | `NATS_CHECKCERT`       | `true`           | Check if ssl certificate of the output is valid                                                                                     |
-| `nats.minimumpriority` | `NATS_MINIMUMPRIORITY` | `""` (= `debug`) | Minimum priority of event for using this output, order is `emergency,alert,critical,error,warning,notice,informational,debug or ""` |
+  # subjecttemplate: "falco.<priority>.<rule>" # template for the subject, tokens <priority> and <rule> will be automatically replaced (default: falco.<priority>.<rule>)
+
+
+|        Setting         |        Env var         |       Default value       |                                                             Description                                                             |
+| ---------------------- | ---------------------- | ------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
+| `nats.hostport`        | `NATS_HOSTPORT`        |                           | nats://{domain or ip}:{port}, if not empty, NATS output is **enabled**                                                              |
+| `nats.subjecttemplate` | `NATS_SUBJECTTEMPLATE` | `falco.<priority>.<rule>` | Template for the subject, tokens <priority> and <rule> will be automatically replaced                                               |
+| `nats.mutualtls`       | `NATS_MUTUALTLS`       | `false`                   | Authenticate to the output with TLS, if true, checkcert flag will be ignored (server cert will always be checked)                   |
+| `nats.checkcert`       | `NATS_CHECKCERT`       | `true`                    | Check if ssl certificate of the output is valid                                                                                     |
+| `nats.minimumpriority` | `NATS_MINIMUMPRIORITY` | `""` (= `debug`)          | Minimum priority of event for using this output, order is `emergency,alert,critical,error,warning,notice,informational,debug or ""` |
 
 > [!NOTE]
 The Env var values override the settings from yaml file.

diff --git a/docs/outputs/stan.md b/docs/outputs/stan.md
@@ -14,13 +14,14 @@
 
 ## Configuration
 
-| Setting                | Env var                | Default value    | Description                                                                                                                         |
-| ---------------------- | ---------------------- | ---------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
-| `stan.hostport`        | `STAN_HOSTPORT`        |                  | stan://{domain or ip}:{port}, if not empty, STAN output is **enabled**                                                              |
-| `stan.clusterid`       | `STAN_CLUSTERID`       |                  | Cluster name (mandatory)                                                                                                            |
-| `stan.clientid`        | `STAN_CLIENTID`        |                  | Client ID (mandatory)                                                                                                               |
-| `stan.checkcert`       | `STAN_CHECKCERT`       | `true`           | Check if ssl certificate of the output is valid                                                                                     |
-| `stan.minimumpriority` | `STAN_MINIMUMPRIORITY` | `""` (= `debug`) | Minimum priority of event for using this output, order is `emergency,alert,critical,error,warning,notice,informational,debug or ""` |
+|        Setting         |        Env var         |       Default value       |                                                             Description                                                             |
+| ---------------------- | ---------------------- | ------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
+| `stan.hostport`        | `STAN_HOSTPORT`        |                           | stan://{domain or ip}:{port}, if not empty, STAN output is **enabled**                                                              |
+| `stan.subjecttemplate` | `STAN_SUBJECTTEMPLATE` | `falco.<priority>.<rule>` | Template for the subject, tokens <priority> and <rule> will be automatically replaced                                               |
+| `stan.clusterid`       | `STAN_CLUSTERID`       |                           | Cluster name (mandatory)                                                                                                            |
+| `stan.clientid`        | `STAN_CLIENTID`        |                           | Client ID (mandatory)                                                                                                               |
+| `stan.checkcert`       | `STAN_CHECKCERT`       | `true`                    | Check if ssl certificate of the output is valid                                                                                     |
+| `stan.minimumpriority` | `STAN_MINIMUMPRIORITY` | `""` (= `debug`)          | Minimum priority of event for using this output, order is `emergency,alert,critical,error,warning,notice,informational,debug or ""` |
 
 > [!NOTE]
 The Env var values override the settings from yaml file.

diff --git a/outputs/nats.go b/outputs/nats.go
@@ -4,22 +4,33 @@ package outputs
 
 import (
 	"encoding/json"
-	"go.opentelemetry.io/otel/attribute"
 	"log"
 	"regexp"
 	"strings"
 
+	"go.opentelemetry.io/otel/attribute"
+
 	nats "github.com/nats-io/nats.go"
 
 	"github.com/falcosecurity/falcosidekick/types"
 )
 
-var slugRegularExpression = regexp.MustCompile("[^a-z0-9]+")
+var slugRegExp = regexp.MustCompile("[^a-z0-9]+")
+
+const defaultNatsSubjects = "falco.<priority>.<rule>"
 
 // NatsPublish publishes event to NATS
 func (c *Client) NatsPublish(falcopayload types.FalcoPayload) {
 	c.Stats.Nats.Add(Total, 1)
 
+	subject := c.Config.Nats.SubjectTemplate
+	if len(subject) == 0 {
+		subject = defaultNatsSubjects
+	}
+
+	subject = strings.ReplaceAll(subject, "<priority>", strings.ToLower(falcopayload.Priority.String()))
+	subject = strings.ReplaceAll(subject, "<rule>", strings.Trim(slugRegExp.ReplaceAllString(strings.ToLower(falcopayload.Rule), "_"), "_"))
+
 	nc, err := nats.Connect(c.EndpointURL.String())
 	if err != nil {
 		c.setNatsErrorMetrics()
@@ -29,15 +40,14 @@ func (c *Client) NatsPublish(falcopayload types.FalcoPayload) {
 	defer nc.Flush()
 	defer nc.Close()
 
-	r := strings.Trim(slugRegularExpression.ReplaceAllString(strings.ToLower(falcopayload.Rule), "_"), "_")
 	j, err := json.Marshal(falcopayload)
 	if err != nil {
 		c.setStanErrorMetrics()
 		log.Printf("[ERROR] : STAN - %v\n", err.Error())
 		return
 	}
 
-	err = nc.Publish("falco."+strings.ToLower(falcopayload.Priority.String())+"."+r, j)
+	err = nc.Publish(subject, j)
 	if err != nil {
 		c.setNatsErrorMetrics()
 		log.Printf("[ERROR] : NATS - %v\n", err)

diff --git a/outputs/stan.go b/outputs/stan.go
@@ -4,10 +4,11 @@ package outputs
 
 import (
 	"encoding/json"
-	"go.opentelemetry.io/otel/attribute"
 	"log"
 	"strings"
 
+	"go.opentelemetry.io/otel/attribute"
+
 	stan "github.com/nats-io/stan.go"
 
 	"github.com/falcosecurity/falcosidekick/types"
@@ -17,6 +18,14 @@ import (
 func (c *Client) StanPublish(falcopayload types.FalcoPayload) {
 	c.Stats.Stan.Add(Total, 1)
 
+	subject := c.Config.Stan.SubjectTemplate
+	if len(subject) == 0 {
+		subject = defaultNatsSubjects
+	}
+
+	subject = strings.ReplaceAll(subject, "<priority>", strings.ToLower(falcopayload.Priority.String()))
+	subject = strings.ReplaceAll(subject, "<rule>", strings.Trim(slugRegExp.ReplaceAllString(strings.ToLower(falcopayload.Rule), "_"), "_"))
+
 	nc, err := stan.Connect(c.Config.Stan.ClusterID, c.Config.Stan.ClientID, stan.NatsURL(c.EndpointURL.String()))
 	if err != nil {
 		c.setStanErrorMetrics()
@@ -25,15 +34,14 @@ func (c *Client) StanPublish(falcopayload types.FalcoPayload) {
 	}
 	defer nc.Close()
 
-	r := strings.Trim(slugRegularExpression.ReplaceAllString(strings.ToLower(falcopayload.Rule), "_"), "_")
 	j, err := json.Marshal(falcopayload)
 	if err != nil {
 		c.setStanErrorMetrics()
 		log.Printf("[ERROR] : STAN - %v\n", err.Error())
 		return
 	}
 
-	err = nc.Publish("falco."+strings.ToLower(falcopayload.Priority.String())+"."+r, j)
+	err = nc.Publish(subject, j)
 	if err != nil {
 		c.setStanErrorMetrics()
 		log.Printf("[ERROR] : STAN - %v\n", err)

diff --git a/types/types.go b/types/types.go
@@ -6,10 +6,11 @@ import (
 	"context"
 	"encoding/json"
 	"expvar"
-	"github.com/falcosecurity/falcosidekick/outputs/otlpmetrics"
 	"text/template"
 	"time"
 
+	"github.com/falcosecurity/falcosidekick/outputs/otlpmetrics"
+
 	"github.com/DataDog/datadog-go/statsd"
 	"github.com/embano1/memlog"
 	"github.com/prometheus/client_golang/prometheus"
@@ -365,6 +366,7 @@ type prometheusOutputConfig struct {
 type natsOutputConfig struct {
 	CommonConfig    `mapstructure:",squash"`
 	HostPort        string
+	SubjectTemplate string
 	MinimumPriority string
 }
 
@@ -373,6 +375,7 @@ type stanOutputConfig struct {
 	HostPort        string
 	ClusterID       string
 	ClientID        string
+	SubjectTemplate string
 	MinimumPriority string
 }