crowdstrike: fix handling of network direction (#12508)

Assume that network direction that is not inbound can be validly
semantically represented as outbound using the ECS fields available.
This is probably not true; documented values of the ConnectionDirection
are 0 - outbound, 1 - inbound, 2 - neither, and 3 - both[1].

Adhering strictly to inbound/outbound makes it impossible to map the
data to ECS since neither and both would only be expressible as unknown.

[1]https://docs.panther.com/data-onboarding/supported-logs/crowdstrike/falcon-data-replicator#crowdstrike.networkconnect

packages/crowdstrike/changelog.yml

@@ -1,4 +1,12 @@
 # newer versions go on top
+- version: "1.49.1"
+  changes:
+    - description: Fix network direction handling for FDR data stream.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/12508
+    - description: Handle invalid IP addresses robustly.
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/12508
 - version: "1.49.0"
   changes:
     - description: Add "preserve_original_event" tag to documents with `event.kind` manually set to "pipeline_error".

packages/crowdstrike/data_stream/fdr/_dev/test/pipeline/test-data.log-expected.json

@@ -316,4 +316,4 @@
             }
         }
     ]
-}
+}

packages/crowdstrike/data_stream/fdr/_dev/test/pipeline/test-fdr.log-expected.json

@@ -593,21 +593,9 @@
                 "name": "NetworkReceiveAcceptIP4LinV5"
             },
             "destination": {
-                "address": "67.43.156.14",
-                "as": {
-                    "number": 35908
-                },
-                "geo": {
-                    "continent_name": "Asia",
-                    "country_iso_code": "BT",
-                    "country_name": "Bhutan",
-                    "location": {
-                        "lat": 27.5,
-                        "lon": 90.5
-                    }
-                },
-                "ip": "67.43.156.14",
-                "port": 53
+                "address": "0.0.0.0",
+                "ip": "0.0.0.0",
+                "port": 39920
             },
             "ecs": {
                 "version": "8.11.0"
@@ -673,9 +661,21 @@
                 ]
             },
             "source": {
-                "address": "0.0.0.0",
-                "ip": "0.0.0.0",
-                "port": 39920
+                "address": "67.43.156.14",
+                "as": {
+                    "number": 35908
+                },
+                "geo": {
+                    "continent_name": "Asia",
+                    "country_iso_code": "BT",
+                    "country_name": "Bhutan",
+                    "location": {
+                        "lat": 27.5,
+                        "lon": 90.5
+                    }
+                },
+                "ip": "67.43.156.14",
+                "port": 53
             },
             "tags": [
                 "preserve_original_event"
@@ -7698,21 +7698,9 @@
                 "name": "NetworkConnectIP4MacV5"
             },
             "destination": {
-                "address": "67.43.156.14",
-                "as": {
-                    "number": 35908
-                },
-                "geo": {
-                    "continent_name": "Asia",
-                    "country_iso_code": "BT",
-                    "country_name": "Bhutan",
-                    "location": {
-                        "lat": 27.5,
-                        "lon": 90.5
-                    }
-                },
-                "ip": "67.43.156.14",
-                "port": 443
+                "address": "0.0.0.0",
+                "ip": "0.0.0.0",
+                "port": 0
             },
             "ecs": {
                 "version": "8.11.0"
@@ -7777,9 +7765,21 @@
                 ]
             },
             "source": {
-                "address": "0.0.0.0",
-                "ip": "0.0.0.0",
-                "port": 0
+                "address": "67.43.156.14",
+                "as": {
+                    "number": 35908
+                },
+                "geo": {
+                    "continent_name": "Asia",
+                    "country_iso_code": "BT",
+                    "country_name": "Bhutan",
+                    "location": {
+                        "lat": 27.5,
+                        "lon": 90.5
+                    }
+                },
+                "ip": "67.43.156.14",
+                "port": 443
             },
             "tags": [
                 "preserve_original_event"
@@ -12092,4 +12092,4 @@
             }
         }
     ]
-}
+}

packages/crowdstrike/data_stream/fdr/_dev/test/pipeline/test-fdrv2-notmanaged.log-expected.json

@@ -68,4 +68,4 @@
             }
         }
     ]
-}
+}

packages/crowdstrike/data_stream/fdr/elasticsearch/ingest_pipeline/default.yml

@@ -1925,16 +1925,21 @@ processors:
 
   ## Networking fields.
   - set:
-      field: source.ip
-      if: ctx.source?.ip == null && ctx.crowdstrike?.CurrentLocalIP != null
-      value: "{{{crowdstrike.CurrentLocalIP}}}"
+      field: network.direction
+      value: outbound
+      if: ctx.crowdstrike?.ConnectionDirection == "0"
   - set:
-      field: source.ip
-      if: ctx.source?.ip == null && ctx.crowdstrike.LocalIP != null
-      value: "{{{crowdstrike.LocalIP}}}"
+      field: network.direction
+      value: inbound
+      if: ctx.crowdstrike?.ConnectionDirection == "1"
+  - set:
+      field: network.direction
+      value: unknown
+      if: ctx.network?.direction == null && ctx.crowdstrike?.ConnectionDirection != null && ctx.crowdstrike.ConnectionDirection != ""
+
   - split:
       field: crowdstrike.LocalAddressIP4
-      separator: "\\s+"
+      separator: '\s+'
       if: ctx.crowdstrike?.LocalAddressIP4 != null
   - convert:
       tag: convert_LocalAddressIP4_ip
@@ -1945,21 +1950,9 @@ processors:
         - append:
             field: error.message
             value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
-  - set:
-      field: source.ip
-      if: ctx.source?.ip == null && ctx.crowdstrike?.LocalAddressIP4 instanceof List && ctx.crowdstrike.LocalAddressIP4.length > 0
-      value: "{{{crowdstrike.LocalAddressIP4.0}}}"
-  # - foreach:
-  #     if: ctx.crowdstrike?.LocalAddressIP4 instanceof List && ctx.crowdstrike.LocalAddressIP4.length > 0
-  #     field: crowdstrike.LocalAddressIP4
-  #     processor:
-  #       append:
-  #         field: source.ip
-  #         value: '{{{_ingest._value}}}'
-  #         allow_duplicates: false
   - split:
       field: crowdstrike.LocalAddressIP6
-      separator: "\\s+"
+      separator: '\s+'
       if: ctx.crowdstrike?.LocalAddressIP6 != null
   - convert:
       tag: convert_LocalAddressIP6_ip
@@ -1970,56 +1963,31 @@ processors:
         - append:
             field: error.message
             value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
-  - set:
-      field: source.ip
-      if: ctx.source?.ip == null && ctx.crowdstrike?.LocalAddressIP6 instanceof List && ctx.crowdstrike.LocalAddressIP6.length > 0
-      value: "{{{crowdstrike.LocalAddressIP6.0}}}"
-  # - foreach:
-  #     if: ctx.crowdstrike?.LocalAddressIP6 != null && ctx.crowdstrike.LocalAddressIP6 instanceof List && ctx.crowdstrike.LocalAddressIP6.length > 0
-  #     field: crowdstrike.LocalAddressIP6
-  #     processor:
-  #       append:
-  #         field: source.ip
-  #         value: '{{{_ingest._value}}}'
-  #         allow_duplicates: false
-  - set:
-      field: source.address
-      copy_from: source.ip
-      ignore_empty_value: true
-  - rename:
-      field: crowdstrike.LocalPort
-      target_field: source.port
-      ignore_missing: true
   - convert:
       tag: convert_RemoteAddressIP4_ip
       field: crowdstrike.RemoteAddressIP4
       type: ip
       ignore_missing: true
-  - rename:
-      field: crowdstrike.RemoteAddressIP4
-      target_field: destination.ip
-      ignore_missing: true
-  - set:
-      field: destination.address
-      copy_from: destination.ip
-      ignore_empty_value: true
   - convert:
       tag: convert_RemoteAddressIP6_ip
       field: crowdstrike.RemoteAddressIP6
       type: ip
       ignore_missing: true
-  - rename:
-      field: crowdstrike.RemoteAddressIP6
-      target_field: destination.ip
-      ignore_missing: true
-  - set:
-      field: destination.address
-      copy_from: destination.ip
-      ignore_empty_value: true
-  - rename:
-      field: crowdstrike.RemotePort
-      target_field: destination.port
-      ignore_missing: true
+
+  - pipeline:
+      tag: pipeline_outbound_network
+      # The condition is all non-inbound, but the pipeline operates assuming the traffic is outbound.
+      # In cases where there is no information we make this assumption rather than dropping the data
+      # on the floor.
+      if: ctx.network?.direction != 'inbound'
+      name: '{{ IngestPipeline "outbound_network" }}'
+      ignore_missing_pipeline: true
+  - pipeline:
+      tag: pipeline_inbound_network
+      if: ctx.network?.direction == 'inbound'
+      name: '{{ IngestPipeline "inbound_network" }}'
+      ignore_missing_pipeline: true
+
   - rename:
       field: crowdstrike.Protocol
       target_field: network.iana_number
@@ -2054,18 +2022,6 @@ processors:
         } else if (iana_number == '132') {
             ctx.network.transport = 'sctp';
         }
-  - set:
-      field: network.direction
-      value: outbound
-      if: ctx.crowdstrike?.ConnectionDirection == "0"
-  - set:
-      field: network.direction
-      value: inbound
-      if: ctx.crowdstrike?.ConnectionDirection == "1"
-  - set:
-      field: network.direction
-      value: unknown
-      if: ctx.network?.direction == null && ctx.crowdstrike?.ConnectionDirection != null && ctx.crowdstrike.ConnectionDirection != ""
   - community_id:
       ignore_missing: true
       ignore_failure: true

packages/crowdstrike/data_stream/fdr/elasticsearch/ingest_pipeline/inbound_network.yml

@@ -0,0 +1,90 @@
+---
+description: Pipeline for processing inbound network details
+processors:
+  - set:
+      field: destination.ip
+      if: ctx.destination?.ip == null && ctx.crowdstrike?.CurrentLocalIP != null
+      value: '{{{crowdstrike.CurrentLocalIP}}}'
+  - set:
+      field: destination.ip
+      if: ctx.destination?.ip == null && ctx.crowdstrike.LocalIP != null
+      value: '{{{crowdstrike.LocalIP}}}'
+  - set:
+      field: destination.ip
+      if: ctx.destination?.ip == null && ctx.crowdstrike?.LocalAddressIP4 instanceof List && ctx.crowdstrike.LocalAddressIP4.length > 0
+      value: '{{{crowdstrike.LocalAddressIP4.0}}}'
+  - set:
+      field: destination.ip
+      if: ctx.destination?.ip == null && ctx.crowdstrike?.LocalAddressIP6 instanceof List && ctx.crowdstrike.LocalAddressIP6.length > 0
+      value: '{{{crowdstrike.LocalAddressIP6.0}}}'
+  - convert:
+      tag: convert_destination_ip
+      field: destination.ip
+      type: ip
+      ignore_missing: true
+      on_failure:
+        - remove:
+            field: destination.ip
+            ignore_missing: true
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - set:
+      field: destination.address
+      copy_from: destination.ip
+      ignore_empty_value: true
+  - rename:
+      field: crowdstrike.LocalPort
+      target_field: destination.port
+      ignore_missing: true
+
+  - rename:
+      field: crowdstrike.MAC
+      target_field: destination.mac
+      ignore_missing: true
+  - rename:
+      if: ctx.destination?.mac == null
+      field: crowdstrike.PhysicalAddress
+      target_field: destination.mac
+      ignore_missing: true
+
+  - convert:
+      tag: convert_RemoteAddressIP4_ip
+      field: crowdstrike.RemoteAddressIP4
+      type: ip
+      ignore_missing: true
+      on_failure:
+        - remove:
+            field: crowdstrike.RemoteAddressIP4
+            ignore_missing: true
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - rename:
+      field: crowdstrike.RemoteAddressIP4
+      target_field: source.ip
+      ignore_missing: true
+  - convert:
+      tag: convert_RemoteAddressIP6_ip
+      field: crowdstrike.RemoteAddressIP6
+      type: ip
+      ignore_missing: true
+      on_failure:
+        - remove:
+            field: crowdstrike.RemoteAddressIP6
+            ignore_missing: true
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - rename:
+      field: crowdstrike.RemoteAddressIP6
+      target_field: source.ip
+      ignore_missing: true
+  - set:
+      field: source.address
+      copy_from: source.ip
+      ignore_empty_value: true
+  - rename:
+      field: crowdstrike.RemotePort
+      target_field: source.port
+      ignore_missing: true