idaholab#321, more consistently differentiate between uploaded and li…

…ve-captured traffic
mmguero-dev · Jan 3, 2024 · 355601d · 355601d
1 parent b673818
commit 355601d
Show file tree

Hide file tree

Showing 7 changed files with 59 additions and 13 deletions.
diff --git a/filebeat/filebeat.yml b/filebeat/filebeat.yml
@@ -32,7 +32,7 @@ filebeat.inputs:
     - ${FILEBEAT_ZEEK_LOG_LIVE_PATH:/zeek/live}/logs/current/*.log
   symlinks: true
   fields_under_root: true
-  tags: ["_filebeat_zeek_live"]
+  tags: ["_filebeat_zeek_malcolm_live"]
   compression_level: 0
   exclude_lines: ['^\s*#']
   scan_frequency: ${FILEBEAT_SCAN_FREQUENCY:10s}
@@ -91,7 +91,7 @@ filebeat.inputs:
     - ${FILEBEAT_SURICATA_LOG_PATH:/suricata}/live/eve.json
   symlinks: true
   fields_under_root: true
-  tags: ["_filebeat_suricata_live"]
+  tags: ["_filebeat_suricata_malcolm_live"]
   compression_level: 0
   scan_frequency: ${FILEBEAT_SCAN_FREQUENCY:10s}
   clean_inactive: ${FILEBEAT_CLEAN_INACTIVE:180m}

diff --git a/logstash/pipelines/suricata/01_input_suricata.conf b/logstash/pipelines/suricata/01_input_suricata.conf
@@ -5,11 +5,28 @@ input {
 }
 
 filter {
+
   # this pipeline only needs to see suricata logs forwarded from filebeat
-  if ("_filebeat_suricata" in [tags]) or ("_filebeat_suricata_live" in [tags]) {
+  if ("_filebeat_suricata" in [tags]) or
+     ("_filebeat_suricata_live" in [tags]) or
+     ("_filebeat_suricata_hedgehog_live" in [tags]) or
+     ("_filebeat_suricata_malcolm_live" in [tags]) {
+
+    if ("_filebeat_suricata_malcolm_live" in [tags]) {
+      # the shipper name comes from PCAP_NODE_NAME in filebeat.yml, but for consistency
+      #   with Arkime live capture append -live to the name for traffic captured live
+      if ([beat][name]) { mutate { id => "mutate_replace_beat_name_suricata_live"
+                                   replace => {"[beat][name]" => "%{[beat][name]}-live" } } }
+      if ([host][name]) { mutate { id => "mutate_replace_host_name_suricata_live"
+                                   replace => {"[host][name]" => "%{[host][name]}-live" } } }
+    }
+
     mutate { id => "mutate_filebeat_suricata_forward_tag_remove"
              remove_tag => [ "_filebeat_suricata",
-                             "_filebeat_suricata_live" ] }
+                             "_filebeat_suricata_live",
+                             "_filebeat_suricata_hedgehog_live",
+                             "_filebeat_suricata_malcolm_live" ] }
+
   } else {
     drop { id => "drop_not_filebeat_suricata" }
   }

diff --git a/logstash/pipelines/zeek/01_input_zeek.conf b/logstash/pipelines/zeek/01_input_zeek.conf
@@ -5,11 +5,28 @@ input {
 }
 
 filter {
-  # this pipeline only needs to see Zeek logs forwarded from filebeat
-  if ("_filebeat_zeek" in [tags]) or ("_filebeat_zeek_live" in [tags]) {
+
+  # this pipeline only needs to see zeek logs forwarded from filebeat
+  if ("_filebeat_zeek" in [tags]) or
+     ("_filebeat_zeek_live" in [tags]) or
+     ("_filebeat_zeek_hedgehog_live" in [tags]) or
+     ("_filebeat_zeek_malcolm_live" in [tags]) {
+
+    if ("_filebeat_zeek_malcolm_live" in [tags]) {
+      # the shipper name comes from PCAP_NODE_NAME in filebeat.yml, but for consistency
+      #   with Arkime live capture append -live to the name for traffic captured live
+      if ([beat][name]) { mutate { id => "mutate_replace_beat_name_zeek_live"
+                                   replace => {"[beat][name]" => "%{[beat][name]}-live" } } }
+      if ([host][name]) { mutate { id => "mutate_replace_host_name_zeek_live"
+                                   replace => {"[host][name]" => "%{[host][name]}-live" } } }
+    }
+
     mutate { id => "mutate_filebeat_zeek_forward_tag_remove"
              remove_tag => [ "_filebeat_zeek",
-                             "_filebeat_zeek_live" ] }
+                             "_filebeat_zeek_live",
+                             "_filebeat_zeek_hedgehog_live",
+                             "_filebeat_zeek_malcolm_live" ] }
+
   } else {
     drop { id => "drop_not_filebeat_zeek" }
   }

diff --git a/sensor-iso/interface/sensor_ctl/filebeat/filebeat.yml b/sensor-iso/interface/sensor_ctl/filebeat/filebeat.yml
@@ -9,7 +9,7 @@ filebeat.inputs:
     - ${BEAT_STATIC_LOG_PATTERN:/home/sensor/bro_logs/static/*.log}
   symlinks: true
   fields_under_root: true
-  tags: ["_filebeat_zeek_live"]
+  tags: ["_filebeat_zeek_hedgehog_live"]
   compression_level: 0
   exclude_lines: ['^\s*#']
   scan_frequency: ${BEAT_SCAN_FREQUENCY:10s}
@@ -27,7 +27,7 @@ filebeat.inputs:
     - ${BEAT_SURICATA_LOG_PATTERN:/home/sensor/bro_logs/suricata/eve*.json}
   symlinks: true
   fields_under_root: true
-  tags: ["_filebeat_suricata_live"]
+  tags: ["_filebeat_suricata_hedgehog_live"]
   compression_level: 0
   scan_frequency: ${BEAT_SCAN_FREQUENCY:10s}
   clean_inactive: ${BEAT_CLEAN_INACTIVE:180m}

diff --git a/shared/bin/pcap_processor.py b/shared/bin/pcap_processor.py
@@ -23,6 +23,7 @@
 import zmq
 
 from pcap_utils import (
+    FILE_INFO_DICT_LIVE,
     FILE_INFO_DICT_NAME,
     FILE_INFO_DICT_NODE,
     FILE_INFO_DICT_SIZE,
@@ -165,13 +166,19 @@ def arkimeCaptureFileWorker(arkimeWorkerArgs):
                         )
                         logger.info(f"{scriptName}[{scanWorkerId}]:\t🔎\t{fileInfo}")
 
+                        # if this is a "live" rotated PCAP captured by netsniff-ng or tcpdump,
+                        #   append -live to the node name used (which originates from PCAP_NODE_NAME)
+                        tmpNodeName = fileInfo[FILE_INFO_DICT_NODE] if (FILE_INFO_DICT_NODE in fileInfo) else nodeName
+                        if (FILE_INFO_DICT_LIVE in fileInfo) and fileInfo[FILE_INFO_DICT_LIVE]:
+                            tmpNodeName = tmpNodeName + '-live'
+
                         # put together arkime execution command
                         cmd = [
                             arkimeBin,
                             '--quiet',
                             '--insecure',
                             '--node',
-                            fileInfo[FILE_INFO_DICT_NODE] if (FILE_INFO_DICT_NODE in fileInfo) else nodeName,
+                            tmpNodeName,
                             '-o',
                             f'ecsEventProvider={arkimeProvider}',
                             '-o',

diff --git a/shared/bin/pcap_utils.py b/shared/bin/pcap_utils.py
@@ -16,12 +16,13 @@
 
 PCAP_MIME_TYPES = ['application/vnd.tcpdump.pcap', 'application/x-pcapng']
 
+FILE_INFO_DICT_LIVE = "live"
 FILE_INFO_DICT_NAME = "name"
-FILE_INFO_DICT_TAGS = "tags"
+FILE_INFO_DICT_NODE = "node"
 FILE_INFO_DICT_SIZE = "size"
-FILE_INFO_FILE_TYPE = "type"
+FILE_INFO_DICT_TAGS = "tags"
 FILE_INFO_FILE_MIME = "mime"
-FILE_INFO_DICT_NODE = "node"
+FILE_INFO_FILE_TYPE = "type"
 
 
 ###################################################################################################

diff --git a/shared/bin/pcap_watcher.py b/shared/bin/pcap_watcher.py
@@ -23,6 +23,7 @@
 import zmq
 
 from pcap_utils import (
+    FILE_INFO_DICT_LIVE,
     FILE_INFO_DICT_NAME,
     FILE_INFO_DICT_NODE,
     FILE_INFO_DICT_SIZE,
@@ -232,6 +233,9 @@ def processFile(self, pathname):
                             FILE_INFO_FILE_MIME: fileMime,
                             FILE_INFO_FILE_TYPE: fileType,
                             FILE_INFO_DICT_NODE: args.nodeName,
+                            FILE_INFO_DICT_LIVE: any(
+                                os.path.basename(pathname).startswith(prefix) for prefix in ('mnetsniff', 'mtcpdump')
+                            ),
                             FILE_INFO_DICT_TAGS: tags_from_filename(relativePath),
                         }
                         self.topic_socket.send_string(json.dumps(fileInfo))