[Filebeat] Add network.direction by specifying internal_networks to gcp module

CHANGELOG.next.asciidoc

@@ -759,6 +759,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add top_level_domain enrichment for suricata/eve fileset. {pull}23046[23046]
 - Add top_level_domain enrichment for zeek/dns fileset. {pull}23046[23046]
 - Add `network.direction` to netflow/log fileset. {pull}23052[23052]
+- Add `network.direction` override by specifying `internal_networks` in gcp module. {pull}23081[23081]
 
 *Heartbeat*
 

libbeat/docs/processors-list.asciidoc

@@ -138,6 +138,9 @@ endif::[]
 ifndef::no_add_locale_processor[]
 include::{libbeat-processors-dir}/add_locale/docs/add_locale.asciidoc[]
 endif::[]
+ifndef::no_add_network_direction_processor[]
+include::{libbeat-processors-dir}/actions/docs/add_network_direction.asciidoc[]
+endif::[]
 ifndef::no_add_observer_metadata_processor[]
 include::{libbeat-processors-dir}/add_observer_metadata/docs/add_observer_metadata.asciidoc[]
 endif::[]

libbeat/processors/actions/add_network_direction.go

@@ -28,13 +28,15 @@ import (
 	"github.com/elastic/beats/v7/libbeat/conditions"
 	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/beats/v7/libbeat/processors/checks"
+	jsprocessor "github.com/elastic/beats/v7/libbeat/processors/script/javascript/module/processor"
 )
 
 func init() {
 	processors.RegisterPlugin("add_network_direction",
 		checks.ConfigChecked(NewAddNetworkDirection,
 			checks.RequireFields("source", "destination", "target", "internal_networks"),
 			checks.AllowedFields("source", "destination", "target", "internal_networks")))
+	jsprocessor.RegisterPlugin("AddNetworkDirection", NewAddNetworkDirection)
 }
 
 const (

x-pack/filebeat/filebeat.reference.yml

@@ -843,6 +843,16 @@ filebeat.modules:
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # direction is determined by whether it is between source and destination
+    # instance information rather than IP.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   firewall:
     enabled: true
 
@@ -861,6 +871,15 @@ filebeat.modules:
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # is taken from the direction data in the rule_details event payload.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   audit:
     enabled: true
 

x-pack/filebeat/module/gcp/_meta/config.yml

@@ -17,6 +17,16 @@
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # direction is determined by whether it is between source and destination
+    # instance information rather than IP.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   firewall:
     enabled: true
 
@@ -35,6 +45,15 @@
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # is taken from the direction data in the rule_details event payload.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   audit:
     enabled: true
 

x-pack/filebeat/module/gcp/firewall/config/input.yml

@@ -31,6 +31,9 @@ processors:
       params:
         debug: {{ .debug }}
         keep_original_message: {{ .keep_original_message }}
+        {{ if .internal_networks }}
+        internal_networks: {{ .internal_networks | tojson }}
+        {{ end }}
       file: ${path.home}/module/gcp/firewall/config/pipeline.js
   - add_fields:
       target: ''

x-pack/filebeat/module/gcp/firewall/config/pipeline.js

@@ -65,7 +65,7 @@ function PipelineBuilder(pipelineName, debug) {
     }
 }
 
-function FirewallProcessor(keep_original_message, debug) {
+function FirewallProcessor(keep_original_message, debug, internalNetworks) {
     var builder = new PipelineBuilder("firewall", debug);
 
     // The pub/sub input writes the Stackdriver LogEntry object into the message
@@ -313,17 +313,25 @@ function FirewallProcessor(keep_original_message, debug) {
         event.AppendTo("related.ip", event.Get("destination.ip"));
     });
 
-    var chain = builder.Build();
+    if (internalNetworks) {
+        builder.Add("addNetworkDirection", processor.AddNetworkDirection({
+            source: "source.ip",
+            destination: "destination.ip",
+            target: "network.direction",
+            internal_networks: internalNetworks,
+        }))
+    }
+
     return {
-        process: chain.Run
+        process: builder.Build().Run
     };
 }
 
 var firewall;
 
 // Register params from configuration.
 function register(params) {
-    firewall = new FirewallProcessor(params.keep_original_message, params.debug);
+    firewall = new FirewallProcessor(params.keep_original_message, params.debug, params.internal_networks);
 }
 
 function process(evt) {

x-pack/filebeat/module/gcp/firewall/manifest.yml

@@ -17,6 +17,7 @@ var:
     default: false
   - name: tags
     default: [forwarded]
+  - name: internal_networks
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/input.yml

x-pack/filebeat/module/gcp/vpcflow/config/input.yml

@@ -31,6 +31,9 @@ processors:
       file: ${path.home}/module/gcp/vpcflow/config/pipeline.js
       params:
         keep_original_message: {{ .keep_original_message }}
+        {{ if .internal_networks }}
+        internal_networks: {{ .internal_networks | tojson }}
+        {{ end }}
   - add_fields:
       target: ''
       fields:

x-pack/filebeat/module/gcp/vpcflow/config/pipeline.js

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-function VPCFlow(keep_original_message) {
+function VPCFlow(keep_original_message, internalNetworks) {
     var processor = require("processor");
 
     // The pub/sub input writes the Stackdriver LogEntry object into the message
@@ -239,19 +239,27 @@ function VPCFlow(keep_original_message) {
         .Add(setNetworkTransport)
         .Add(setNetworkDirection)
         .Add(setNetworkType)
-        .Add(setRelatedIP)
-        .Build();
+        .Add(setRelatedIP);
+
+    if (internalNetworks) {
+        pipeline = pipeline.AddNetworkDirection({
+            source: "source.ip",
+            destination: "destination.ip",
+            target: "network.direction",
+            internal_networks: internalNetworks,
+        })
+    }
 
     return {
-        process: pipeline.Run,
+        process: pipeline.Build().Run,
     };
 }
 
 var vpcflow;
 
 // Register params from configuration.
 function register(params) {
-    vpcflow = new VPCFlow(params.keep_original_message);
+    vpcflow = new VPCFlow(params.keep_original_message, params.internal_networks);
 }
 
 function process(evt) {

x-pack/filebeat/module/gcp/vpcflow/manifest.yml

@@ -15,6 +15,7 @@ var:
     default: false
   - name: tags
     default: [forwarded]
+  - name: internal_networks
 
 ingest_pipeline: ingest/pipeline.yml
 input: config/input.yml

x-pack/filebeat/modules.d/gcp.yml.disabled

@@ -20,6 +20,16 @@
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # direction is determined by whether it is between source and destination
+    # instance information rather than IP.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   firewall:
     enabled: true
 
@@ -38,6 +48,15 @@
     # the subscription.
     var.credentials_file: ${path.config}/gcp-service-account-xyz.json
 
+    # Set internal networks. This is used to classify network.direction based
+    # off of what networks are considered "internal" either base off of a CIDR
+    # block or named network conditions. If this is not specified, then traffic
+    # is taken from the direction data in the rule_details event payload.
+    #
+    # For a full list of network conditions see:
+    # https://www.elastic.co/guide/en/beats/filebeat/current/defining-processors.html#condition-network
+    #var.internal_networks: [ "private" ]
+
   audit:
     enabled: true