Support for NetFlow Version 9 (#11)

Closes #1
Closes #6

Author: joschi

README.md

@@ -9,7 +9,10 @@ This plugin provides a NetFlow UDP input to act as a Flow collector that receive
 
 ## Supported NetFlow Versions
 
-The plugin only supports NetFlow V5 at the moment.
+The version of the plugin now supports NetFlow V9.  It can support IPv6 addresses without 
+conversion and handles all of the fields from the fixed V5 format.  In addition this plugin supports
+events from a CISCO ASA 5500, including firewall and routing events.  Beware, there is significant
+duplication of typical syslog reporting in the v9 reporting. 
 
 ## Installation
 

pom.xml

@@ -62,6 +62,30 @@
             <version>${junit.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj-core.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+            <version>${jackson.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.pkts</groupId>
+            <artifactId>pkts-core</artifactId>
+            <version>2.0.7</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.jayway.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${awaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -75,6 +99,15 @@
             </resource>
         </resources>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <annotationProcessors>
+                        <annotationProcessor>com.google.auto.value.processor.AutoValueProcessor</annotationProcessor>
+                    </annotationProcessors>
+                </configuration>
+            </plugin>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-shade-plugin</artifactId>

src/main/java/org/graylog/plugins/netflow/NetFlowPluginModule.java

@@ -6,21 +6,33 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+/*
+* Modified by Benjamin H. Klimkowski, bhklimk@gmail.com
+*/
 package org.graylog.plugins.netflow;
 
 import org.graylog.plugins.netflow.codecs.NetFlowCodec;
 import org.graylog.plugins.netflow.inputs.NetFlowUdpInput;
+import org.graylog2.plugin.PluginConfigBean;
 import org.graylog2.plugin.PluginModule;
 
+import java.util.Collections;
+import java.util.Set;
+
 public class NetFlowPluginModule extends PluginModule {
+    @Override
+    public Set<? extends PluginConfigBean> getConfigBeans() {
+        return Collections.emptySet();
+    }
+
     @Override
     protected void configure() {
         addMessageInput(NetFlowUdpInput.class);

src/main/java/org/graylog/plugins/netflow/codecs/NetFlowCodec.java

@@ -1,12 +1,12 @@
 /**
  * Copyright (C) 2012, 2013, 2014 wasted.io Ltd <really@wasted.io>
- * Copyright (C) 2015 Graylog, Inc. (hello@graylog.org)
+ * Copyright (C) 2015-2017 Graylog, Inc. (hello@graylog.org)
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -16,15 +16,19 @@
  */
 package org.graylog.plugins.netflow.codecs;
 
-import com.google.common.collect.Lists;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.inject.assistedinject.Assisted;
+import org.apache.commons.lang3.SystemUtils;
 import org.graylog.plugins.netflow.flows.FlowException;
 import org.graylog.plugins.netflow.flows.NetFlowParser;
-import org.graylog.plugins.netflow.flows.NetFlow;
-import org.graylog.plugins.netflow.flows.NetFlowPacket;
+import org.graylog.plugins.netflow.v9.NetFlowV9FieldTypeRegistry;
+import org.graylog.plugins.netflow.v9.NetFlowV9TemplateCache;
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.configuration.Configuration;
 import org.graylog2.plugin.configuration.ConfigurationRequest;
+import org.graylog2.plugin.configuration.fields.NumberField;
+import org.graylog2.plugin.configuration.fields.TextField;
 import org.graylog2.plugin.inputs.annotations.Codec;
 import org.graylog2.plugin.inputs.annotations.ConfigClass;
 import org.graylog2.plugin.inputs.annotations.FactoryClass;
@@ -38,16 +42,42 @@
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import javax.inject.Inject;
+import javax.inject.Named;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Collection;
-import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
 
 @Codec(name = "netflow", displayName = "NetFlow")
 public class NetFlowCodec extends AbstractCodec implements MultiMessageCodec {
     private static final Logger LOG = LoggerFactory.getLogger(NetFlowCodec.class);
 
+    @VisibleForTesting
+    static final String CK_CACHE_SIZE = "cache_size";
+    @VisibleForTesting
+    static final String CK_CACHE_PATH = "cache_path";
+    @VisibleForTesting
+    static final String CK_CACHE_SAVE_INTERVAL = "cache_save_interval";
+
+    private static final int DEFAULT_CACHE_SIZE = 1000;
+    private static final String DEFAULT_CACHE_PATH = SystemUtils.getJavaIoTmpDir().toPath().resolve("netflow-templates.json").toString();
+    private static final int DEFAULT_CACHE_SAVE_INTERVAL = 15 * 60;
+
+    private final NetFlowV9TemplateCache templateCache;
+    private final NetFlowV9FieldTypeRegistry typeRegistry = new NetFlowV9FieldTypeRegistry();
+
     @Inject
-    protected NetFlowCodec(@Assisted Configuration configuration) {
+    protected NetFlowCodec(@Assisted Configuration configuration,
+                           @Named("daemonScheduler") ScheduledExecutorService scheduler,
+                           ObjectMapper objectMapper) {
         super(configuration);
+
+        final int cacheSize = configuration.getInt(CK_CACHE_SIZE, DEFAULT_CACHE_SIZE);
+        final int cacheSaveInterval = configuration.getInt(CK_CACHE_SAVE_INTERVAL, DEFAULT_CACHE_SAVE_INTERVAL);
+        final String configCachePath = configuration.getString(CK_CACHE_PATH, DEFAULT_CACHE_PATH);
+        final Path cachePath = Paths.get(configCachePath);
+
+        templateCache = new NetFlowV9TemplateCache(cacheSize, cachePath, cacheSaveInterval, scheduler, objectMapper);
     }
 
     @Nullable
@@ -60,19 +90,7 @@ public Message decode(@Nonnull RawMessage rawMessage) {
     @Override
     public Collection<Message> decodeMessages(@Nonnull RawMessage rawMessage) {
         try {
-            final NetFlowPacket packet = NetFlowParser.parse(rawMessage);
-
-            if (packet == null) {
-                return null;
-            }
-
-            final List<Message> messages = Lists.newArrayListWithCapacity(packet.getFlows().size());
-
-            for (NetFlow flow : packet.getFlows()) {
-                messages.add(flow.toMessage());
-            }
-
-            return messages;
+            return NetFlowParser.parse(rawMessage, templateCache, typeRegistry);
         } catch (FlowException e) {
             LOG.error("Error parsing NetFlow packet", e);
             return null;
@@ -96,5 +114,16 @@ public void overrideDefaultValues(@Nonnull ConfigurationRequest cr) {
                 cr.getField(NettyTransport.CK_PORT).setDefaultValue(2055);
             }
         }
+
+        @Override
+        public ConfigurationRequest getRequestedConfiguration() {
+            final ConfigurationRequest configuration = super.getRequestedConfiguration();
+
+            configuration.addField(new NumberField(CK_CACHE_SIZE, "Maximum cache size", DEFAULT_CACHE_SIZE, "Maximum number of elements in the NetFlow9 template cache"));
+            configuration.addField(new TextField(CK_CACHE_PATH, "Cache file path", DEFAULT_CACHE_PATH, "Path to the file persisting the the NetFlow9 template cache"));
+            configuration.addField(new NumberField(CK_CACHE_SAVE_INTERVAL, "Cache save interval (seconds)", DEFAULT_CACHE_SAVE_INTERVAL, "Interval in seconds for persisting the cache contents"));
+
+            return configuration;
+        }
     }
 }

src/main/java/org/graylog/plugins/netflow/flows/CorruptFlowPacketException.java

@@ -17,4 +17,11 @@
 package org.graylog.plugins.netflow.flows;
 
 public class CorruptFlowPacketException extends FlowException {
+    public CorruptFlowPacketException() {
+        super();
+    }
+
+    public CorruptFlowPacketException(String message) {
+        super(message);
+    }
 }

…plugins/netflow/flows/NetFlowPacket.java …etflow/flows/EmptyTemplateException.javasrc/main/java/org/graylog/plugins/netflow/flows/NetFlowPacket.java renamed to src/main/java/org/graylog/plugins/netflow/flows/EmptyTemplateException.java src/main/java/org/graylog/plugins/netflow/flows/NetFlowPacket.java renamed to src/main/java/org/graylog/plugins/netflow/flows/EmptyTemplateException.java

@@ -14,10 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.graylog.plugins.netflow.flows;
 
-import java.util.Collection;
+/*
+* Created by Benjamin H. Klimkowski, bhklimk@gmail.com
+*/
+
+package org.graylog.plugins.netflow.flows;
 
-public interface NetFlowPacket {
-    Collection<NetFlow> getFlows();
+public class EmptyTemplateException extends FlowException {
 }

src/main/java/org/graylog/plugins/netflow/flows/FlowException.java

@@ -6,7 +6,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -16,5 +16,12 @@
  */
 package org.graylog.plugins.netflow.flows;
 
-public class FlowException extends Exception {
+public class FlowException extends RuntimeException {
+    public FlowException() {
+        super();
+    }
+
+    public FlowException(String message) {
+        super(message);
+    }
 }

src/main/java/org/graylog/plugins/netflow/flows/InvalidFlowVersionException.java

@@ -6,7 +6,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *         http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -18,5 +18,6 @@
 
 public class InvalidFlowVersionException extends FlowException {
     public InvalidFlowVersionException(int version) {
+        super("Invalid NetFlow version " + version);
     }
 }