Merge pull request #84 from satta/multi-forward

Implement selective event forwarding to multiple outputs

Author: 0mbi

cmd/fever/cmds/run.go

@@ -140,12 +140,39 @@ func mainfunc(cmd *cobra.Command, args []string) {
 		}()
 	}
 
-	// Configure forwarding
-	outputSocket := viper.GetString("output.socket")
-	forward = (outputSocket != "")
-	eventTypes := viper.GetStringSlice("forward.types")
-	allTypes := viper.GetBool("forward.all")
-	util.PrepareEventFilter(eventTypes, allTypes)
+	// Get config from viper
+	var multiForwardConf processing.MultiForwardConfiguration
+	err = viper.Unmarshal(&multiForwardConf)
+	if err != nil {
+		log.Fatal(err)
+	}
+	// Keep supporting legacy config ("output" and "forward" sections)
+	if len(multiForwardConf.Outputs) == 0 {
+		log.Info("no multi-forwarder configuration, found, using legacy config")
+		outSocketPath := viper.GetString("output.socket")
+		if len(outSocketPath) > 0 {
+			multiForwardConf.Outputs = make(map[string]processing.MultiForwardOutput)
+			multiForwardConf.Outputs["default"] =
+				processing.MultiForwardOutput{
+					Socket: viper.GetString("output.socket"),
+					All:    viper.GetBool("forward.all"),
+					Types:  viper.GetStringSlice("forward.types"),
+				}
+		}
+	} else {
+		log.Info("found multi-forwarder configuration, ignoring legacy config")
+	}
+
+	if len(multiForwardConf.Outputs) > 0 {
+		forward = true
+	}
+
+	if pse != nil {
+		multiForwardConf.SubmitStats(pse)
+	}
+	multiFwdChan := make(chan types.Entry)
+	reconnectTimes := viper.GetInt("reconnect-retries")
+	multiForwardConf.Run(multiFwdChan, reconnectTimes)
 
 	// Optional profiling
 	profileFile := viper.GetString("profile")
@@ -185,14 +212,10 @@ func mainfunc(cmd *cobra.Command, args []string) {
 	s.Run(eventChan)
 
 	var forwardHandler processing.Handler
-	reconnectTimes := viper.GetInt("reconnect-retries")
 	// start forwarding
 	if forward {
-		forwardHandler = processing.MakeForwardHandler(int(reconnectTimes), outputSocket)
+		forwardHandler = processing.MakeForwardHandler(multiFwdChan)
 		fh := forwardHandler.(*processing.ForwardHandler)
-		if pse != nil {
-			fh.SubmitStats(pse)
-		}
 		rdns := viper.GetBool("active.rdns")
 		if rdns {
 			expiryPeriod := viper.GetDuration("active.rdns-cache-expiry")
@@ -202,7 +225,6 @@ func mainfunc(cmd *cobra.Command, args []string) {
 				fh.RDNSHandler.EnableOnlyPrivateIPRanges()
 			}
 		}
-		fh.Run()
 
 		stenosis := viper.GetBool("stenosis.enable")
 		if stenosis {
@@ -236,12 +258,6 @@ func mainfunc(cmd *cobra.Command, args []string) {
 
 		addFields := viper.GetStringMapString("add-fields")
 		fh.AddFields(addFields)
-
-		defer func() {
-			c := make(chan bool)
-			fh.Stop(c)
-			<-c
-		}()
 	} else {
 		// in this case we use a void handler that does nothing
 		forwardHandler = processing.MakeVoidHandler()

fever.yaml

@@ -54,18 +54,23 @@ input:
   #  # Disables Redis pipelining.
   #  nopipe: true
 
-# Definition what event types to forward. Set 'all' to true to forward 
-# everything received from Suricata, otherwise use the 'types' list to choose.
-# By default, we only forward alerts and stats events.
-forward:
-  all: false
-  types: 
-    - alert
-    - stats
-
-# Configuration for output of forwarded events (socket to e.g. Logstash).
-output:
-  socket: /tmp/suri-forward.sock
+# Configure forwarding of events processed by FEVER, i.e. define what event
+# types to forward.
+multi-forward:
+  # Set 'all' to true to forward everything received from Suricata, otherwise
+  # use the 'types' list to choose. Example:
+  # socketall:
+  #  socket: /tmp/out-all.sock
+  #  buffer-length: 100000
+  #  all: true
+  #  types: []
+  socketalerts:
+    socket: /tmp/suri-forward.sock
+    all: false
+    buffer-length: 1000
+    types:
+      - alert
+      - stats
 
 # Configuration for flow report submission.
 flowreport:

processing/bloom_handler.go

@@ -31,7 +31,6 @@ type BloomHandler struct {
 	BloomFileIsCompressed bool
 	DatabaseEventChan     chan types.Entry
 	ForwardHandler        Handler
-	DoForwardAlert        bool
 	AlertPrefix           string
 	Alertifier            *util.Alertifier
 	BlacklistIOCs         map[string]struct{}
@@ -75,7 +74,6 @@ func MakeBloomHandler(iocBloom *bloom.BloomFilter,
 		IocBloom:          iocBloom,
 		DatabaseEventChan: databaseChan,
 		ForwardHandler:    forwardHandler,
-		DoForwardAlert:    (util.ForwardAllEvents || util.AllowType("alert")),
 		AlertPrefix:       alertPrefix,
 		Alertifier:        util.MakeAlertifier(alertPrefix),
 		BlacklistIOCs:     make(map[string]struct{}),

processing/bloom_handler_test.go

@@ -278,9 +278,6 @@ func (h *CollectorHandler) GetEntries() map[string]bool {
 }
 
 func TestBloomHandler(t *testing.T) {
-	// make sure that alerts are forwarded
-	util.PrepareEventFilter([]string{"alert"}, false)
-
 	// initalize Bloom filter and fill with 'interesting' values
 	bf := bloom.Initialize(100000, 0.0000001)
 	fillBloom(&bf)
@@ -394,9 +391,9 @@ func TestBloomHandler(t *testing.T) {
 				e = makeBloomTLSEvent(s, ":::")
 				bh.Consume(&e)
 			case 3:
-				f := fmt.Sprintf("%s", util.RndStringFromAlpha(6))
+				f := util.RndStringFromAlpha(6)
 				for bf.Check([]byte(f)) {
-					f = fmt.Sprintf("%s", util.RndStringFromAlpha(6))
+					f = util.RndStringFromAlpha(6)
 				}
 				e = makeBloomTLSEvent("foo.com", f)
 				bh.Consume(&e)
@@ -695,8 +692,6 @@ func TestBloomHandlerURL(t *testing.T) {
 		close(consumeWaitChan)
 	}()
 
-	util.PrepareEventFilter([]string{"alert"}, false)
-
 	// initalize Bloom filter and fill with 'interesting' values
 	bf := bloom.Initialize(100000, 0.0000001)
 	bf.Add([]byte("/oddlyspecific"))
@@ -950,8 +945,6 @@ func TestBloomHandlerBlacklistedSkip(t *testing.T) {
 		close(consumeWaitChan)
 	}()
 
-	util.PrepareEventFilter([]string{"alert"}, false)
-
 	// handler to receive forwarded events
 	fwhandler := &CollectorHandler{
 		Entries: make(map[string]bool),
@@ -981,9 +974,6 @@ func TestBloomHandlerBlacklistedSkip(t *testing.T) {
 }
 
 func TestBloomHandlerInvalidDNS(t *testing.T) {
-	// make sure that alerts are forwarded
-	util.PrepareEventFilter([]string{"alert"}, false)
-
 	// initalize Bloom filter and fill with 'interesting' values
 	bf := bloom.Initialize(100000, 0.0000001)
 

processing/flow_extractor_test.go

@@ -35,8 +35,8 @@ func makeFlowExtractorEvent(ipv6 bool) types.Entry {
 		srcIP = fmt.Sprintf("10.0.0.%d", rand.Intn(50))
 		destIP = fmt.Sprintf("10.0.0.%d", rand.Intn(50))
 	} else {
-		srcIP = fmt.Sprintf("2001:0db8:85a3:0000:0000:8a2e:0370:7334")
-		destIP = fmt.Sprintf("2001:0db8:85a3:0000:0000:8a2e:0370:7334")
+		srcIP = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
+		destIP = "2001:0db8:85a3:0000:0000:8a2e:0370:7334"
 	}
 
 	e := types.Entry{