Merge branch 'master' into cef-issue-21

docs/netflow/README.md

@@ -0,0 +1,86 @@
+NetFlow Plugin for Graylog
+==========================
+
+[![Build Status](https://travis-ci.org/Graylog2/graylog-plugin-netflow.svg?branch=master)](https://travis-ci.org/Graylog2/graylog-plugin-netflow)
+
+This plugin provides a NetFlow UDP input to act as a Flow collector that receives data from Flow exporters. Each received Flow will be converted to a Graylog message.
+
+**Required Graylog version:** 2.3.0 and later
+
+## Supported NetFlow Versions
+
+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
+> Since Graylog Version 2.4.0 this plugin is already included in the Graylog server installation package as default plugin.
+
+[Download the plugin](https://github.com/Graylog2/graylog-plugin-netflow/releases)
+and place the `.jar` file in your Graylog plugin directory. The plugin directory
+is the `plugins/` folder relative from your `graylog-server` directory by default
+and can be configured in your `graylog.conf` file.
+
+Restart `graylog-server` and you are done.
+
+## Setup
+
+In the Graylog web interface, go to System/Inputs and create a new NetFlow input like this:
+
+![NetFlow input creation dialog](https://github.com/Graylog2/graylog-plugin-netflow/blob/master/images/netflow-udp-input-1.png)
+
+## Example Message
+
+This is an example NetFlow message in Graylog:
+
+![NetFlow example fields screenshot](https://github.com/Graylog2/graylog-plugin-netflow/blob/master/images/netflow-example.png)
+
+## Example Dashboard
+
+This is an example of a dashboard with NetFlow data:
+
+![NetFlow example dashboard screenshot](https://github.com/Graylog2/graylog-plugin-netflow/blob/master/images/netflow-dashboard.png)
+
+## Credits
+
+The NetFlow parsing code is based on the https://github.com/wasted/netflow project and has been ported from Scala to Java.
+
+## Plugin Development
+
+### Testing
+
+To generate some NetFlow data for debugging and testing you can use softflowd.
+
+Example command and output:
+
+```
+# softflowd -D -i eth0 -v 5 -t maxlife=1 -n 10.0.2.2:2055
+
+Using eth0 (idx: 0)
+softflowd v0.9.9 starting data collection
+Exporting flows to [10.0.2.2]:2055
+ADD FLOW seq:1 [10.0.2.2]:48164 <> [10.0.2.15]:22 proto:6
+ADD FLOW seq:2 [10.0.2.2]:51428 <> [10.0.2.15]:22 proto:6
+Starting expiry scan: mode 0
+Queuing flow seq:1 (0x7fef0318bc70) for expiry reason 6
+Finished scan 1 flow(s) to be evicted
+Sending v5 flow packet len = 120
+sent 1 netflow packets
+EXPIRED: seq:1 [10.0.2.2]:48164 <> [10.0.2.15]:22 proto:6 octets>:322 packets>:7 octets<:596 packets<:7 start:2015-07-21T13:18:01.236 finish:2015-07-21T13:18:27.718 tcp>:10 tcp<:18 flowlabel>:00000000 flo
+wlabel<:00000000  (0x7fef0318bc70)
+ADD FLOW seq:3 [10.0.2.2]:2055 <> [10.0.2.15]:48363 proto:17
+ADD FLOW seq:4 [10.0.2.2]:48164 <> [10.0.2.15]:22 proto:6
+```
+
+## Plugin Release
+
+We are using the Maven release plugin:
+
+```
+$ mvn release:prepare
+[...]
+$ mvn release:perform
+```
+
+This sets the version numbers, creates a tag and pushes to GitHub. Travis CI will build the release artifacts and upload to GitHub automatically.

graylog-project-parent/pom.xml

@@ -639,6 +639,12 @@
                 <version>${okhttp.version}</version>
                 <scope>test</scope>
             </dependency>
+            <dependency>
+                <groupId>io.pkts</groupId>
+                <artifactId>pkts-core</artifactId>
+                <version>${pkts.version}</version>
+                <scope>test</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

graylog2-server/pom.xml

@@ -177,6 +177,10 @@
             <groupId>com.fasterxml.jackson.jaxrs</groupId>
             <artifactId>jackson-jaxrs-json-provider</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>io.dropwizard.metrics</groupId>
@@ -556,6 +560,10 @@
             <groupId>com.jayway.awaitility</groupId>
             <artifactId>awaitility</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.pkts</groupId>
+            <artifactId>pkts-core</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

graylog2-server/src/main/java/org/graylog/plugins/netflow/NetFlowPluginModule.java

@@ -0,0 +1,40 @@
+/**
+ * This file is part of Graylog.
+ *
+ * Graylog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Graylog is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Graylog.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.graylog.plugins.netflow;
+
+import org.graylog.plugins.netflow.codecs.NetFlowCodec;
+import org.graylog.plugins.netflow.inputs.NetFlowUdpInput;
+import org.graylog.plugins.netflow.transport.NetFlowUdpTransport;
+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);
+        addCodec("netflow", NetFlowCodec.class);
+        addTransport("netflow-udp", NetFlowUdpTransport.class);
+    }
+}

graylog2-server/src/main/java/org/graylog/plugins/netflow/codecs/NetFlowCodec.java

@@ -0,0 +1,216 @@
+/**
+ * This file is part of Graylog.
+ *
+ * Graylog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Graylog is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Graylog.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.graylog.plugins.netflow.codecs;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
+import com.google.inject.assistedinject.Assisted;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import org.graylog.plugins.netflow.flows.FlowException;
+import org.graylog.plugins.netflow.flows.NetFlowFormatter;
+import org.graylog.plugins.netflow.v5.NetFlowV5Packet;
+import org.graylog.plugins.netflow.v5.NetFlowV5Parser;
+import org.graylog.plugins.netflow.v9.NetFlowV9FieldTypeRegistry;
+import org.graylog.plugins.netflow.v9.NetFlowV9Journal;
+import org.graylog.plugins.netflow.v9.NetFlowV9OptionTemplate;
+import org.graylog.plugins.netflow.v9.NetFlowV9Packet;
+import org.graylog.plugins.netflow.v9.NetFlowV9Parser;
+import org.graylog.plugins.netflow.v9.NetFlowV9Record;
+import org.graylog.plugins.netflow.v9.NetFlowV9Template;
+import org.graylog2.plugin.Message;
+import org.graylog2.plugin.ResolvableInetSocketAddress;
+import org.graylog2.plugin.configuration.Configuration;
+import org.graylog2.plugin.configuration.ConfigurationRequest;
+import org.graylog2.plugin.configuration.fields.ConfigurationField;
+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;
+import org.graylog2.plugin.inputs.codecs.AbstractCodec;
+import org.graylog2.plugin.inputs.codecs.CodecAggregator;
+import org.graylog2.plugin.inputs.codecs.MultiMessageCodec;
+import org.graylog2.plugin.inputs.transports.NettyTransport;
+import org.graylog2.plugin.journal.RawMessage;
+import org.graylog2.shared.utilities.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.inject.Inject;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Codec(name = "netflow", displayName = "NetFlow")
+public class NetFlowCodec extends AbstractCodec implements MultiMessageCodec {
+    /**
+     * Marker byte which signals that the contained netflow packet should be parsed as is.
+     */
+    public static final byte PASSTHROUGH_MARKER = 0x00;
+    /**
+     * Marker byte which signals that the contained netflow v9 packet is non-RFC:
+     * It contains all necessary template flows before any data flows and can be completely parsed without a template cache.
+     */
+    public static final byte ORDERED_V9_MARKER = 0x01;
+    @VisibleForTesting
+    static final String CK_NETFLOW9_DEFINITION_PATH = "netflow9_definitions_Path";
+    private static final Logger LOG = LoggerFactory.getLogger(NetFlowCodec.class);
+    private final NetFlowV9FieldTypeRegistry typeRegistry;
+    private final NetflowV9CodecAggregator netflowV9CodecAggregator;
+
+    @Inject
+    protected NetFlowCodec(@Assisted Configuration configuration, NetflowV9CodecAggregator netflowV9CodecAggregator) throws IOException {
+        super(configuration);
+        this.netflowV9CodecAggregator = netflowV9CodecAggregator;
+
+        final String netFlow9DefinitionsPath = configuration.getString(CK_NETFLOW9_DEFINITION_PATH);
+        if (netFlow9DefinitionsPath == null || netFlow9DefinitionsPath.trim().isEmpty()) {
+            this.typeRegistry = NetFlowV9FieldTypeRegistry.create();
+        } else {
+            try (InputStream inputStream = new FileInputStream(netFlow9DefinitionsPath)) {
+                this.typeRegistry = NetFlowV9FieldTypeRegistry.create(inputStream);
+            }
+        }
+    }
+
+    @Nullable
+    @Override
+    public CodecAggregator getAggregator() {
+        return netflowV9CodecAggregator;
+    }
+
+    @Nullable
+    @Override
+    public Message decode(@Nonnull RawMessage rawMessage) {
+        throw new UnsupportedOperationException("MultiMessageCodec " + getClass() + " does not support decode()");
+    }
+
+    @Nullable
+    @Override
+    public Collection<Message> decodeMessages(@Nonnull RawMessage rawMessage) {
+        try {
+            final ResolvableInetSocketAddress remoteAddress = rawMessage.getRemoteAddress();
+            final InetSocketAddress sender = remoteAddress != null ? remoteAddress.getInetSocketAddress() : null;
+
+            final byte[] payload = rawMessage.getPayload();
+            if (payload.length < 3) {
+                LOG.debug("NetFlow message (source: {}) doesn't even fit the NetFlow version (size: {} bytes)",
+                        sender, payload.length);
+                return null;
+            }
+
+            final ByteBuf buffer = Unpooled.wrappedBuffer(payload);
+            switch (buffer.readByte()) {
+                case PASSTHROUGH_MARKER:
+                    final NetFlowV5Packet netFlowV5Packet = NetFlowV5Parser.parsePacket(buffer);
+
+                    return netFlowV5Packet.records().stream()
+                            .map(record -> NetFlowFormatter.toMessage(netFlowV5Packet.header(), record, sender))
+                            .collect(Collectors.toList());
+                case ORDERED_V9_MARKER:
+                    // our "custom" netflow v9 that has all the templates in the same packet
+                    return decodeV9(sender, buffer);
+                default:
+                    final List<RawMessage.SourceNode> sourceNodes = rawMessage.getSourceNodes();
+                    final RawMessage.SourceNode sourceNode = sourceNodes.isEmpty() ? null : sourceNodes.get(sourceNodes.size() - 1);
+                    final String inputId = sourceNode == null ? "<unknown>" : sourceNode.inputId;
+                    LOG.warn("Unsupported NetFlow packet on input {} (source: {})", inputId, sender);
+                    return null;
+            }
+        } catch (FlowException e) {
+            LOG.error("Error parsing NetFlow packet <{}> received from <{}>", rawMessage.getId(), rawMessage.getRemoteAddress(), e);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("NetFlow packet hexdump:\n{}", ByteBufUtil.prettyHexDump(Unpooled.wrappedBuffer(rawMessage.getPayload())));
+            }
+            return null;
+        } catch (InvalidProtocolBufferException e) {
+            LOG.error("Invalid NetFlowV9 entry found, cannot parse the messages", ExceptionUtils.getRootCause(e));
+            return null;
+        }
+    }
+
+    @VisibleForTesting
+    Collection<Message> decodeV9(InetSocketAddress sender, ByteBuf buffer) throws InvalidProtocolBufferException {
+        final List<NetFlowV9Packet> netFlowV9Packets = decodeV9Packets(buffer);
+
+        return netFlowV9Packets.stream().map(netFlowV9Packet -> netFlowV9Packet.records().stream()
+                .filter(record -> record instanceof NetFlowV9Record)
+                .map(record -> NetFlowFormatter.toMessage(netFlowV9Packet.header(), record, sender))
+                .collect(Collectors.toList())
+        ).flatMap(Collection::stream)
+         .collect(Collectors.toList());
+    }
+
+    @VisibleForTesting
+    List<NetFlowV9Packet> decodeV9Packets(ByteBuf buffer) throws InvalidProtocolBufferException {
+        byte[] v9JournalEntry = new byte[buffer.readableBytes()];
+        buffer.readBytes(v9JournalEntry);
+        final NetFlowV9Journal.RawNetflowV9 rawNetflowV9 = NetFlowV9Journal.RawNetflowV9.parseFrom(v9JournalEntry);
+
+        // parse all templates used in the packet
+        final Map<Integer, NetFlowV9Template> templateMap = Maps.newHashMap();
+        rawNetflowV9.getTemplatesMap().forEach((templateId, byteString) -> {
+            final NetFlowV9Template netFlowV9Template = NetFlowV9Parser.parseTemplate(
+                    Unpooled.wrappedBuffer(byteString.toByteArray()), typeRegistry);
+            templateMap.put(templateId, netFlowV9Template);
+        });
+        final NetFlowV9OptionTemplate[] optionTemplate = {null};
+        rawNetflowV9.getOptionTemplateMap().forEach((templateId, byteString) -> {
+            optionTemplate[0] = NetFlowV9Parser.parseOptionTemplate(Unpooled.wrappedBuffer(byteString.toByteArray()), typeRegistry);
+        });
+
+        return rawNetflowV9.getPacketsList().stream()
+                .map(bytes -> Unpooled.wrappedBuffer(bytes.toByteArray()))
+                .map(buf -> NetFlowV9Parser.parsePacket(buf, typeRegistry, templateMap, optionTemplate[0]))
+                .collect(Collectors.toList());
+    }
+
+    @FactoryClass
+    public interface Factory extends AbstractCodec.Factory<NetFlowCodec> {
+        @Override
+        NetFlowCodec create(Configuration configuration);
+
+        @Override
+        Config getConfig();
+    }
+
+    @ConfigClass
+    public static class Config extends AbstractCodec.Config {
+        @Override
+        public void overrideDefaultValues(@Nonnull ConfigurationRequest cr) {
+            if (cr.containsField(NettyTransport.CK_PORT)) {
+                cr.getField(NettyTransport.CK_PORT).setDefaultValue(2055);
+            }
+        }
+
+        @Override
+        public ConfigurationRequest getRequestedConfiguration() {
+            final ConfigurationRequest configuration = super.getRequestedConfiguration();
+            configuration.addField(new TextField(CK_NETFLOW9_DEFINITION_PATH, "Netflow 9 field definitions", "", "Path to the YAML file containing Netflow 9 field definitions", ConfigurationField.Optional.OPTIONAL));
+            return configuration;
+        }
+    }
+}