Network direction processor supports dynamic internal networks specification (#68712)

docs/reference/ingest/processors/network-direction.asciidoc

@@ -21,8 +21,9 @@ only the `internal_networks` option must be specified.
 | `source_ip`        | no       | `source.ip`   | Field containing the source IP address.
 | `destination_ip`   | no       | `destination.ip` | Field containing the destination IP address.
 | `target_field`     | no       | `network.direction` | Output field for the network direction.
-| `internal_networks`| yes      |               | List of internal networks. Supports IPv4 and
-IPv6 addresses and ranges in CIDR notation. Also supports the named ranges listed below.
+| `internal_networks`| yes *    |               | List of internal networks. Supports IPv4 and
+IPv6 addresses and ranges in CIDR notation. Also supports the named ranges listed below. These may be constructed with <<template-snippets,template snippets>>. * Must specify only one of `internal_networks` or `internal_networks_field`.
+| `internal_networks_field`| no      |               | A field on the given document to read the `internal_networks` configuration from.
 | `ignore_missing`   | no       | `true`        | If `true` and any required fields are missing,
 the processor quietly exits without modifying the document.
 
@@ -30,6 +31,8 @@ the processor quietly exits without modifying the document.
 include::common-options.asciidoc[]
 |======
 
+One of either `internal_networks` or `internal_networks_field` must be specified. If `internal_networks_field` is specified, it follows the behavior specified by `ignore_missing`.
+
 [float]
 [[supported-named-network-ranges]]
 ===== Supported named network ranges

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/IngestCommonPlugin.java

@@ -79,7 +79,7 @@ public class IngestCommonPlugin extends Plugin implements ActionPlugin, IngestPl
                 entry(HtmlStripProcessor.TYPE, new HtmlStripProcessor.Factory()),
                 entry(CsvProcessor.TYPE, new CsvProcessor.Factory()),
                 entry(UriPartsProcessor.TYPE, new UriPartsProcessor.Factory()),
-                entry(NetworkDirectionProcessor.TYPE, new NetworkDirectionProcessor.Factory()),
+                entry(NetworkDirectionProcessor.TYPE, new NetworkDirectionProcessor.Factory(parameters.scriptService)),
                 entry(CommunityIdProcessor.TYPE, new CommunityIdProcessor.Factory()),
                 entry(FingerprintProcessor.TYPE, new FingerprintProcessor.Factory()),
                 entry(RegisteredDomainProcessor.TYPE, new RegisteredDomainProcessor.Factory())

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/NetworkDirectionProcessor.java

@@ -14,12 +14,17 @@ import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.script.TemplateScript;
 
 import java.net.InetAddress;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
+import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
 import static org.elasticsearch.ingest.ConfigurationUtils.readBooleanProperty;
 
 public class NetworkDirectionProcessor extends AbstractProcessor {
@@ -48,7 +53,8 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
     private final String sourceIpField;
     private final String destinationIpField;
     private final String targetField;
-    private final List<String> internalNetworks;
+    private final List<TemplateScript.Factory> internalNetworks;
+    private final String internalNetworksField;
     private final boolean ignoreMissing;
 
     NetworkDirectionProcessor(
@@ -57,7 +63,8 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
         String sourceIpField,
         String destinationIpField,
         String targetField,
-        List<String> internalNetworks,
+        List<TemplateScript.Factory> internalNetworks,
+        String internalNetworksField,
         boolean ignoreMissing
     ) {
         super(tag, description);
@@ -65,6 +72,7 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
         this.destinationIpField = destinationIpField;
         this.targetField = targetField;
         this.internalNetworks = internalNetworks;
+        this.internalNetworksField = internalNetworksField;
         this.ignoreMissing = ignoreMissing;
     }
 
@@ -80,10 +88,14 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
         return targetField;
     }
 
-    public List<String> getInternalNetworks() {
+    public List<TemplateScript.Factory> getInternalNetworks() {
         return internalNetworks;
     }
 
+    public String getInternalNetworksField() {
+        return internalNetworksField;
+    }
+
     public boolean getIgnoreMissing() {
         return ignoreMissing;
     }
@@ -103,9 +115,18 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
         return ingestDocument;
     }
 
-    private String getDirection(IngestDocument d) {
-        if (internalNetworks == null) {
-            return null;
+    private String getDirection(IngestDocument d) throws Exception {
+        List<String> networks = new ArrayList<>();
+
+        if (internalNetworksField != null) {
+            @SuppressWarnings("unchecked")
+            List<String> stringList = d.getFieldValue(internalNetworksField, networks.getClass(), ignoreMissing);
+            if (stringList == null) {
+                return null;
+            }
+            networks.addAll(stringList);
+        } else {
+            networks = internalNetworks.stream().map(network -> d.renderTemplate(network)).collect(Collectors.toList());
         }
 
         String sourceIpAddrString = d.getFieldValue(sourceIpField, String.class, ignoreMissing);
@@ -118,8 +139,8 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
             return null;
         }
 
-        boolean sourceInternal = isInternal(sourceIpAddrString);
-        boolean destinationInternal = isInternal(destIpAddrString);
+        boolean sourceInternal = isInternal(networks, sourceIpAddrString);
+        boolean destinationInternal = isInternal(networks, destIpAddrString);
 
         if (sourceInternal && destinationInternal) {
             return DIRECTION_INTERNAL;
@@ -133,8 +154,8 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
         return DIRECTION_EXTERNAL;
     }
 
-    private boolean isInternal(String ip) {
-        for (String network : internalNetworks) {
+    private boolean isInternal(List<String> networks, String ip) {
+        for (String network : networks) {
             if (inNetwork(ip, network)) {
                 return true;
             }
@@ -227,11 +248,15 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
     }
 
     public static final class Factory implements Processor.Factory {
-
+        private final ScriptService scriptService;
         static final String DEFAULT_SOURCE_IP = "source.ip";
         static final String DEFAULT_DEST_IP = "destination.ip";
         static final String DEFAULT_TARGET = "network.direction";
 
+        public Factory(ScriptService scriptService) {
+            this.scriptService = scriptService;
+        }
+
         @Override
         public NetworkDirectionProcessor create(
             Map<String, Processor.Factory> registry,
@@ -239,19 +264,44 @@ public class NetworkDirectionProcessor extends AbstractProcessor {
             String description,
             Map<String, Object> config
         ) throws Exception {
-            String sourceIpField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "source_ip", DEFAULT_SOURCE_IP);
-            String destIpField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "destination_ip", DEFAULT_DEST_IP);
-            String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field", DEFAULT_TARGET);
-            List<String> internalNetworks = ConfigurationUtils.readList(TYPE, processorTag, config, "internal_networks");
-            boolean ignoreMissing = readBooleanProperty(TYPE, processorTag, config, "ignore_missing", true);
+            final String sourceIpField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "source_ip", DEFAULT_SOURCE_IP);
+            final String destIpField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "destination_ip", DEFAULT_DEST_IP);
+            final String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field", DEFAULT_TARGET);
+            final boolean ignoreMissing = readBooleanProperty(TYPE, processorTag, config, "ignore_missing", true);
+
+            final List<String> internalNetworks = ConfigurationUtils.readOptionalList(TYPE, processorTag, config, "internal_networks");
+            final String internalNetworksField = ConfigurationUtils.readOptionalStringProperty(
+                TYPE,
+                processorTag,
+                config,
+                "internal_networks_field"
+            );
 
+            if (internalNetworks == null && internalNetworksField == null) {
+                throw newConfigurationException(TYPE, processorTag, "internal_networks", "or [internal_networks_field] must be specified");
+            }
+            if (internalNetworks != null && internalNetworksField != null) {
+                throw newConfigurationException(
+                    TYPE,
+                    processorTag,
+                    "internal_networks", "and [internal_networks_field] cannot both be used in the same processor"
+                );
+            }
+
+            List<TemplateScript.Factory> internalNetworkTemplates = null;
+            if (internalNetworks != null) {
+                internalNetworkTemplates = internalNetworks.stream()
+                    .map(n -> ConfigurationUtils.compileTemplate(TYPE, processorTag, "internal_networks", n, scriptService))
+                    .collect(Collectors.toList());
+            }
             return new NetworkDirectionProcessor(
                 processorTag,
                 description,
                 sourceIpField,
                 destIpField,
                 targetField,
-                internalNetworks,
+                internalNetworkTemplates,
+                internalNetworksField,
                 ignoreMissing
             );
         }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/NetworkDirectionProcessorFactoryTests.java

@@ -10,10 +10,12 @@ package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.ingest.TestTemplateService;
 import org.junit.Before;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
@@ -21,6 +23,7 @@ import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.
 import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.DEFAULT_SOURCE_IP;
 import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.DEFAULT_TARGET;
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
 
 public class NetworkDirectionProcessorFactoryTests extends ESTestCase {
 
@@ -28,7 +31,7 @@ public class NetworkDirectionProcessorFactoryTests extends ESTestCase {
 
     @Before
     public void init() {
-        factory = new NetworkDirectionProcessor.Factory();
+        factory = new NetworkDirectionProcessor.Factory(TestTemplateService.instance());
     }
 
     public void testCreate() throws Exception {
@@ -52,7 +55,32 @@ public class NetworkDirectionProcessorFactoryTests extends ESTestCase {
         assertThat(networkProcessor.getSourceIpField(), equalTo(sourceIpField));
         assertThat(networkProcessor.getDestinationIpField(), equalTo(destIpField));
         assertThat(networkProcessor.getTargetField(), equalTo(targetField));
-        assertThat(networkProcessor.getInternalNetworks(), equalTo(internalNetworks));
+        assertThat(networkProcessor.getInternalNetworks().size(), greaterThan(0));
+        assertThat(networkProcessor.getInternalNetworks().get(0).newInstance(Collections.emptyMap()).execute(), equalTo("10.0.0.0/8"));
+        assertThat(networkProcessor.getIgnoreMissing(), equalTo(ignoreMissing));
+    }
+
+    public void testCreateInternalNetworksField() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+
+        String sourceIpField = randomAlphaOfLength(6);
+        config.put("source_ip", sourceIpField);
+        String destIpField = randomAlphaOfLength(6);
+        config.put("destination_ip", destIpField);
+        String targetField = randomAlphaOfLength(6);
+        config.put("target_field", targetField);
+        String internalNetworksField = randomAlphaOfLength(6);
+        config.put("internal_networks_field", internalNetworksField);
+        boolean ignoreMissing = randomBoolean();
+        config.put("ignore_missing", ignoreMissing);
+
+        String processorTag = randomAlphaOfLength(10);
+        NetworkDirectionProcessor networkProcessor = factory.create(null, processorTag, null, config);
+        assertThat(networkProcessor.getTag(), equalTo(processorTag));
+        assertThat(networkProcessor.getSourceIpField(), equalTo(sourceIpField));
+        assertThat(networkProcessor.getDestinationIpField(), equalTo(destIpField));
+        assertThat(networkProcessor.getTargetField(), equalTo(targetField));
+        assertThat(networkProcessor.getInternalNetworksField(), equalTo(internalNetworksField));
         assertThat(networkProcessor.getIgnoreMissing(), equalTo(ignoreMissing));
     }
 
@@ -63,7 +91,7 @@ public class NetworkDirectionProcessorFactoryTests extends ESTestCase {
             factory.create(null, processorTag, null, config);
             fail("factory create should have failed");
         } catch (ElasticsearchParseException e) {
-            assertThat(e.getMessage(), equalTo("[internal_networks] required property is missing"));
+            assertThat(e.getMessage(), equalTo("[internal_networks] or [internal_networks_field] must be specified"));
         }
     }
 

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/NetworkDirectionProcessorTests.java

@@ -10,14 +10,16 @@ package org.elasticsearch.ingest.common;
 
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.ingest.TestTemplateService;
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.ingest.TestTemplateService;
 
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.ArrayList;
 import java.util.Map;
 
-import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.DEFAULT_DEST_IP;
-import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.DEFAULT_SOURCE_IP;
 import static org.elasticsearch.ingest.common.NetworkDirectionProcessor.Factory.DEFAULT_TARGET;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -49,8 +51,11 @@ public class NetworkDirectionProcessorTests extends ESTestCase {
     }
 
     public void testNoInternalNetworks() throws Exception {
-        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> testNetworkDirectionProcessor(buildEvent(), null));
-        assertThat(e.getMessage(), containsString("unable to calculate network direction from document"));
+        ElasticsearchParseException e = expectThrows(
+            ElasticsearchParseException.class,
+            () -> testNetworkDirectionProcessor(buildEvent(), null)
+        );
+        assertThat(e.getMessage(), containsString("[internal_networks] or [internal_networks_field] must be specified"));
     }
 
     public void testNoSource() throws Exception {
@@ -130,6 +135,50 @@ public class NetworkDirectionProcessorTests extends ESTestCase {
         testNetworkDirectionProcessor(source, internalNetworks, expectedDirection, false);
     }
 
+    public void testReadFromField() throws Exception {
+        String processorTag = randomAlphaOfLength(10);
+        Map<String, Object> source = buildEvent("192.168.1.1", "192.168.1.2");
+        ArrayList<String> networks = new ArrayList<>();
+        networks.add("public");
+        source.put("some_field", networks);
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("internal_networks_field", "some_field");
+        NetworkDirectionProcessor processor = new NetworkDirectionProcessor.Factory(TestTemplateService.instance()).create(
+            null,
+            processorTag,
+            null,
+            config
+        );
+        IngestDocument input = new IngestDocument(source, Map.of());
+        IngestDocument output = processor.execute(input);
+        String hash = output.getFieldValue(DEFAULT_TARGET, String.class);
+        assertThat(hash, equalTo("external"));
+    }
+
+    public void testInternalNetworksAndField() throws Exception {
+        String processorTag = randomAlphaOfLength(10);
+        Map<String, Object> source = buildEvent("192.168.1.1", "192.168.1.2");
+        ArrayList<String> networks = new ArrayList<>();
+        networks.add("public");
+        source.put("some_field", networks);
+        Map<String, Object> config = new HashMap<>();
+        config.put("internal_networks_field", "some_field");
+        config.put("internal_networks", networks);
+        ElasticsearchParseException e = expectThrows(
+            ElasticsearchParseException.class,
+            () -> new NetworkDirectionProcessor.Factory(TestTemplateService.instance()).create(
+                null,
+                processorTag,
+                null,
+                config
+            )
+        );
+        assertThat(e.getMessage(), containsString(
+            "[internal_networks] and [internal_networks_field] cannot both be used in the same processor"
+        ));
+    }
+
     private void testNetworkDirectionProcessor(
         Map<String, Object> source,
         String[] internalNetworks,
@@ -140,14 +189,15 @@ public class NetworkDirectionProcessorTests extends ESTestCase {
 
         if (internalNetworks != null) networks = Arrays.asList(internalNetworks);
 
-        var processor = new NetworkDirectionProcessor(
+        String processorTag = randomAlphaOfLength(10);
+        Map<String, Object> config = new HashMap<>();
+        config.put("internal_networks", networks);
+        config.put("ignore_missing", ignoreMissing);
+        NetworkDirectionProcessor processor = new NetworkDirectionProcessor.Factory(TestTemplateService.instance()).create(
             null,
+            processorTag,
             null,
-            DEFAULT_SOURCE_IP,
-            DEFAULT_DEST_IP,
-            DEFAULT_TARGET,
-            networks,
-            ignoreMissing
+            config
         );
 
         IngestDocument input = new IngestDocument(source, Map.of());