Suricata alerts (#9322)

plugins/inputs/suricata/README.md

@@ -4,6 +4,7 @@ This plugin reports internal performance counters of the Suricata IDS/IPS
 engine, such as captured traffic volume, memory usage, uptime, flow counters,
 and much more. It provides a socket for the Suricata log output to write JSON
 stats output to, and processes the incoming data to fit Telegraf's format.
+It can also report for triggered Suricata IDS/IPS alerts.
 
 ### Configuration
 
@@ -17,6 +18,9 @@ stats output to, and processes the incoming data to fit Telegraf's format.
   # Delimiter for flattening field keys, e.g. subitem "alert" of "detect"
   # becomes "detect_alert" when delimiter is "_".
   delimiter = "_"
+
+  # Detect alert logs 
+  alerts = false
 ```
 
 ### Metrics
@@ -26,7 +30,7 @@ stats output.
 See http://suricata.readthedocs.io/en/latest/performance/statistics.html for
 more information.
 
-All fields are numeric.
+All fields for Suricata stats are numeric.
 - suricata
   - tags:
     - thread: `Global` for global statistics (if enabled), thread IDs (e.g. `W#03-enp0s31f6`) for thread-specific statistics
@@ -94,6 +98,19 @@ All fields are numeric.
     - tcp_synack
     - ...
 
+Some fields of the Suricata alerts are strings, for example the signatures. See https://suricata.readthedocs.io/en/suricata-6.0.0/output/eve/eve-json-format.html?highlight=priority#event-type-alert for more information.
+
+- suricata_alert
+  - fields:
+    - action
+    - gid
+    - severity
+    - signature
+    - source_ip
+    - source_port
+    - target_port
+    - target_port
+    - ...
 
 #### Suricata configuration
 

plugins/inputs/suricata/suricata.go

@@ -25,6 +25,7 @@ const (
 type Suricata struct {
 	Source    string `toml:"source"`
 	Delimiter string `toml:"delimiter"`
+	Alerts    bool   `toml:"alerts"`
 
 	inputListener *net.UnixListener
 	cancel        context.CancelFunc
@@ -36,11 +37,11 @@ type Suricata struct {
 
 // Description returns the plugin description.
 func (s *Suricata) Description() string {
-	return "Suricata stats plugin"
+	return "Suricata stats and alerts plugin"
 }
 
 const sampleConfig = `
-  ## Data sink for Suricata stats log
+  ## Data sink for Suricata stats and alerts logs
   # This is expected to be a filename of a
   # unix socket to be created for listening.
   source = "/var/run/suricata-stats.sock"
@@ -48,6 +49,9 @@ const sampleConfig = `
   # Delimiter for flattening field keys, e.g. subitem "alert" of "detect"
   # becomes "detect_alert" when delimiter is "_".
   delimiter = "_"
+  
+  ## Detect alert logs 
+  # alerts = false 
 `
 
 // SampleConfig returns a sample TOML section to illustrate configuration
@@ -100,8 +104,12 @@ func (s *Suricata) readInput(ctx context.Context, acc telegraf.Accumulator, conn
 			line, rerr := reader.ReadBytes('\n')
 			if rerr != nil {
 				return rerr
-			} else if len(line) > 0 {
+			}
-				s.parse(acc, line)
+			if len(line) > 0 {
+				err := s.parse(acc, line)
+				if err != nil {
+					acc.AddError(err)
+				}
 			}
 		}
 	}
@@ -158,28 +166,35 @@ func flexFlatten(outmap map[string]interface{}, field string, v interface{}, del
 	case string:
 		outmap[field] = v
 	case float64:
-		outmap[field] = v.(float64)
+		outmap[field] = t
 	default:
 		return fmt.Errorf("unsupported type %T encountered", t)
 	}
 	return nil
 }
 
-func (s *Suricata) parse(acc telegraf.Accumulator, sjson []byte) {
+func (s *Suricata) parseAlert(acc telegraf.Accumulator, result map[string]interface{}) {
-	// initial parsing
+	if _, ok := result["alert"].(map[string]interface{}); !ok {
-	var result map[string]interface{}
+		s.Log.Debug("'alert' sub-object does not have required structure")
-	err := json.Unmarshal(sjson, &result)
+		return
+	}
+
+	totalmap := make(map[string]interface{})
+	for k, v := range result["alert"].(map[string]interface{}) {
+		//source and target fields are maps
+		err := flexFlatten(totalmap, k, v, s.Delimiter)
 		if err != nil {
-		acc.AddError(err)
+			s.Log.Debugf("Flattening alert failed: %v", err)
-		return
+			// we skip this subitem as something did not parse correctly
+			continue
+		}
 	}
 
-	// check for presence of relevant stats
+	//threads field do not exist in alert output, always global
-	if _, ok := result["stats"]; !ok {
+	acc.AddFields("suricata_alert", totalmap, nil)
-		s.Log.Debug("Input does not contain necessary 'stats' sub-object")
-		return
 }
 
+func (s *Suricata) parseStats(acc telegraf.Accumulator, result map[string]interface{}) {
 	if _, ok := result["stats"].(map[string]interface{}); !ok {
 		s.Log.Debug("The 'stats' sub-object does not have required structure")
 		return
@@ -193,9 +208,9 @@ func (s *Suricata) parse(acc telegraf.Accumulator, sjson []byte) {
 				for k, t := range v {
 					outmap := make(map[string]interface{})
 					if threadStruct, ok := t.(map[string]interface{}); ok {
-						err = flexFlatten(outmap, "", threadStruct, s.Delimiter)
+						err := flexFlatten(outmap, "", threadStruct, s.Delimiter)
 						if err != nil {
-							s.Log.Debug(err)
+							s.Log.Debugf("Flattening alert failed: %v", err)
 							// we skip this thread as something did not parse correctly
 							continue
 						}
@@ -206,10 +221,11 @@ func (s *Suricata) parse(acc telegraf.Accumulator, sjson []byte) {
 				s.Log.Debug("The 'threads' sub-object does not have required structure")
 			}
 		} else {
-			err = flexFlatten(totalmap, k, v, s.Delimiter)
+			err := flexFlatten(totalmap, k, v, s.Delimiter)
 			if err != nil {
-				s.Log.Debug(err.Error())
+				s.Log.Debugf("Flattening alert failed: %v", err)
 				// we skip this subitem as something did not parse correctly
+				continue
 			}
 		}
 	}
@@ -224,6 +240,28 @@ func (s *Suricata) parse(acc telegraf.Accumulator, sjson []byte) {
 	}
 }
 
+func (s *Suricata) parse(acc telegraf.Accumulator, sjson []byte) error {
+	// initial parsing
+	var result map[string]interface{}
+	err := json.Unmarshal(sjson, &result)
+	if err != nil {
+		return err
+	}
+	// check for presence of relevant stats or alert
+	_, ok := result["stats"]
+	_, ok2 := result["alert"]
+	if !ok && !ok2 {
+		s.Log.Debugf("Invalid input without 'stats' or 'alert' object: %v", result)
+		return fmt.Errorf("input does not contain 'stats' or 'alert' object")
+	}
+	if ok {
+		s.parseStats(acc, result)
+	} else if ok2 && s.Alerts {
+		s.parseAlert(acc, result)
+	}
+	return nil
+}
+
 // Gather measures and submits one full set of telemetry to Telegraf.
 // Not used here, submission is completely input-driven.
 func (s *Suricata) Gather(_ telegraf.Accumulator) error {

plugins/inputs/suricata/suricata_test.go

@@ -29,6 +29,7 @@ func TestSuricataLarge(t *testing.T) {
 	s := Suricata{
 		Source:    tmpfn,
 		Delimiter: ".",
+		Alerts:    true,
 		Log: testutil.Logger{
 			Name: "inputs.suricata",
 		},
@@ -46,11 +47,74 @@ func TestSuricataLarge(t *testing.T) {
 	require.NoError(t, err)
 	_, err = c.Write([]byte("\n"))
 	require.NoError(t, err)
+
+	//test suricata alerts
+	data2, err := ioutil.ReadFile("testdata/test2.json")
+	require.NoError(t, err)
+	_, err = c.Write(data2)
+	require.NoError(t, err)
+	_, err = c.Write([]byte("\n"))
+	require.NoError(t, err)
 	require.NoError(t, c.Close())
 
 	acc.Wait(1)
 }
 
+func TestSuricataAlerts(t *testing.T) {
+	dir, err := ioutil.TempDir("", "test")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+	tmpfn := filepath.Join(dir, fmt.Sprintf("t%d", rand.Int63()))
+
+	s := Suricata{
+		Source:    tmpfn,
+		Delimiter: ".",
+		Alerts:    true,
+		Log: testutil.Logger{
+			Name: "inputs.suricata",
+		},
+	}
+	acc := testutil.Accumulator{}
+	require.NoError(t, s.Start(&acc))
+	defer s.Stop()
+
+	data, err := ioutil.ReadFile("testdata/test3.json")
+	require.NoError(t, err)
+
+	c, err := net.Dial("unix", tmpfn)
+	require.NoError(t, err)
+	_, err = c.Write(data)
+	require.NoError(t, err)
+	_, err = c.Write([]byte("\n"))
+	require.NoError(t, err)
+	require.NoError(t, c.Close())
+
+	acc.Wait(1)
+
+	expected := []telegraf.Metric{
+		testutil.MustMetric(
+			"suricata_alert",
+			map[string]string{},
+			map[string]interface{}{
+				"action":       "allowed",
+				"category":     "Misc activity",
+				"gid":          float64(1),
+				"rev":          float64(0),
+				"signature":    "Corrupted HTTP body",
+				"signature_id": float64(6),
+				"severity":     float64(3),
+				"source.ip":    "10.0.0.5",
+				"target.ip":    "179.60.192.3",
+				"source.port":  float64(18715),
+				"target.port":  float64(80),
+			},
+			time.Unix(0, 0),
+		),
+	}
+
+	testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), testutil.IgnoreTime())
+}
+
 func TestSuricata(t *testing.T) {
 	dir, err := ioutil.TempDir("", "test")
 	require.NoError(t, err)

plugins/inputs/suricata/testdata/test3.json

@@ -0,0 +1 @@
+{"timestamp":"2021-05-30T20:07:13.208777+0200","flow_id":1696236471136137,"in_iface":"s1-suricata","event_type":"alert","src_ip":"10.0.0.5","src_port":18715,"dest_ip":"179.60.192.3","dest_port":80,"proto":"TCP","alert":{"action":"allowed","gid":1,"source":{"ip":"10.0.0.5","port":18715},"target":{"ip":"179.60.192.3","port":80},"signature_id":6,"rev":0,"signature":"Corrupted HTTP body","severity": 3,"category":"Misc activity","severity":3},"flow":{"pkts_toserver":1,"pkts_toclient":0,"bytes_toserver":174,"bytes_toclient":0,"start":"2021-05-30T20:07:13.208777+0200"}}