Handle bi-directional flows from Cisco ASA

Author: robcowart

logstash/conf.d/20_filter.logstash.conf

@@ -184,7 +184,15 @@ filter {
                 id => "netflow-v9-normalize-bytes-from-in_permanent_bytes"
                 rename => { "[netflow][in_permanent_bytes]" => "[netflow][bytes]" }
             }
+        } else if [netflow][fwd_flow_delta_bytes] or [netflow][rev_flow_delta_bytes] {
+            ruby {
+                id => "netflow-v9-normalize-bytes-from-fwd-rev-bytes"
+                code => "
+                    event.set( '[netflow][bytes]', event.get('[netflow][fwd_flow_delta_bytes]').to_i + event.get('[netflow][rev_flow_delta_bytes]').to_i )
+                "
+            }
         }
+
         if [netflow][bytes] {
             mutate {
                 id => "netflow-v9-normalize-convert-bytes"
@@ -208,7 +216,15 @@ filter {
                 id => "netflow-v9-normalize-packets-from-in_permanent_pkts"
                 rename => { "[netflow][in_permanent_pkts]" => "[netflow][packets]" }
             }
+        } else if [netflow][initiatorPackets] or [netflow][responderPackets] {
+            ruby {
+                id => "netflow-v9-normalize-packets-from-init-resp-pkts"
+                code => "
+                    event.set( '[netflow][packets]', event.get('[netflow][initiatorPackets]').to_i + event.get('[netflow][responderPackets]').to_i )
+                "
+           }
         }
+
         if [netflow][packets] {
             mutate {
                 id => "netflow-v9-normalize-convert-packets"
@@ -262,6 +278,31 @@ filter {
         }
     }
 
+    # Attempt to populate netflow.last_switched and netflow.first_switched if not provided in raw data. Usually this is necessary for data from Cisco ASA.
+    if ![netflow][last_switched] {
+        if [netflow][event_time_msec] {
+            date {
+                id => "netflow-v9-normalize-lastsw-from-event_time_msec"
+                match => ["[netflow][event_time_msec]", "UNIX_MS"]
+                target => "[netflow][last_switched]"
+            }
+        } else {
+            mutate {
+                id => "netflow-v9-normalize-lastsw-from-timestamp"
+                add_field => { "[netflow][last_switched]" => "%{[@timestamp]}" }
+            }
+        }
+    }
+    if ![netflow][first_switched] {
+        if [netflow][flow_start_msec] {
+            date {
+                id => "netflow-v9-normalize-firstsw-from-flow_start_msec"
+                match => ["[netflow][flow_start_msec]", "UNIX_MS"]
+                target => "[netflow][first_switched]"
+            }
+        }
+    }
+
     #--------------------
     # We now have a normalized flow record. The rest of the logic works
     # regardless of the Netflow version.

logstash/templates/netflow.template.json

@@ -89,7 +89,7 @@
           "event_time_msec": {
             "path_match": "netflow.event_time_msec",
             "mapping": {
-              "type": "long"
+              "type": "date"
             }
           }
         },
@@ -153,7 +153,7 @@
           "flow_start_msec": {
             "path_match": "netflow.flow_start_msec",
             "mapping": {
-              "type": "long"
+              "type": "date"
             }
           }
         },