Merge branch 'master' into 54-1-7-19-Fixes

Merged master back in to feature branch in order to
bring in fixes needed to test the new installer.

README.md

@@ -19,7 +19,7 @@ often practical to install everything in one place.
 **You must ensure the RITA database can be contacted on an IP address other
 than localhost**. This can be done by editing the `bindIP` setting in
 `/etc/mongod.conf`. The installer will prompt you to ensure this change is made
-before continuing on. If you intend to install IPFIX-RITA on the same machine 
+before continuing on. If you intend to install IPFIX-RITA on the same machine
 as RITA and MongoDB, please add the IP address suggested by the installer.
 \
 NOTE: if you want multiple bind IP addresses in your MongoDB config file you
@@ -31,7 +31,7 @@ for IPFIX-RITA to access) your bind IP line should look like the following.
 Also if your RITA config file (`/etc/rita/config.yaml`) connects to MongoDB on
 localhost you will need to change that to the same value as MongoDB is listening
 on. For example if you change the bindIP in your MongoDB config file to 10.0.0.5
-and you check your RITA config file and the connection string is 
+and you check your RITA config file and the connection string is
 `mongodb://localhost:27017` you'll need to change it to ` mongodb://10.0.0.5:27017`.
 
 #### How to [Install RITA](https://github.com/activecm/rita#automatic-installation)
@@ -53,7 +53,7 @@ $ wget $(curl --silent "https://api.github.com/repos/activecm/ipfix-rita/release
 | grep '"browser_download_url":' | cut -d \" -f 4 ) -O ipfix-rita.tgz
 ```
 
-#### Upack the installer
+#### Unpack the installer
 ```
 $ tar -zxf ipfix-rita.tgz
 ```
@@ -67,7 +67,7 @@ You will be prompted for configuration details regarding the RITA database
 connection and the names of the resulting datasets. Further configuration
 options can be set in `/etc/ipfix-rita/converter/converter.yaml`.
 
-By default, **IPFIX-RITA will run at start up unless it is stopped**. For more 
+By default, **IPFIX-RITA will run at start up unless it is stopped**. For more
 information see [Additional Info](docs/Additional%20Info.md). Full
 documentation for IPFIX-RITA can be found in the [docs](docs/) folder.
 
@@ -85,7 +85,7 @@ setting up your router for use with IPFIX-RITA.
 |   Cisco ASA  |       |     ✔      |            |                  |
 | Cisco ASR 9k |       |     ✔      |            |                  |
 |   SonicWall  |       |     ✔      |            |                  |
-|   MikroTik   |       |     ✔      |     ✔      |                  | 
+|   MikroTik   |   ✔   |     ✔      |     ✔      |                  |
 |     YAF      |   ✔   |            |            | Use `--uniflow`  |
 
 ## What Do I Do If My Router Isn't On the List?
@@ -164,7 +164,7 @@ Output:
         CAFile: null
 ...
 ```
-try connecting to mongo using 
+try connecting to mongo using
 ```
 mongo [ipaddress]:[port]
 mongo 10.0.0.5:27017
@@ -199,15 +199,20 @@ IPFix).
 The following bugs have been documented by Active Countermeasures and solutions
 are in development
 
-### IPFix Time Error
-A common error that might occur is something like
+### IPFix-RITA Fails on Reboot
+If IPFix-RITA is configured to write to a MongoDB database running on the
+Docker host (as in the default configuration), IPFix-RITA may encounter
+an error after a system reboot.
+The following error may arise:
 ```
-converter_1_a86985062afe | ERRO[1082] input map must contain key 'netflow.flowStartMilliseconds'  ...
+converter_1_eeb48d380f26 | ERRO[0038] could not connect to RITA MongoDB: could not connect to MongoDB (no TLS): could not connect to MongoDB: no reachable servers  stacktrace="[rita_dates.go:60 convert.go:194 convert.go:40 app.go:490 command.go:210 app.go:255 boot.go:18 proc.go:198 asm_amd64.s:2361]"
+```
+This is due to an error in which the MongoDB server starts before the Docker engine.
+Until a fix is implemented, Active Countermeasures recommends running the following command
+to resolve the issue
+```
+sudo systemctl restart mongod.service
 ```
-If you are using IPFix this is likely caused by the router using an IPFix version
-that uses timestamps relative to the system initilization time and not Unix timestamps.
-This is a known issue and we are working on a solution to support more IPFix versions.
-If this error persists, Active Countermeasures recommends switching to Netflow v9 or v5.
 
 ### Log Rotation Error
 It has been discovered that some flow logs will report flow start and end

converter/commands/check_config.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/activecm/ipfix-rita/converter/config"
 	"github.com/activecm/ipfix-rita/converter/config/yaml"
-	"github.com/activecm/ipfix-rita/converter/input/mgologstash"
+	"github.com/activecm/ipfix-rita/converter/input/logstash/mongodb"
 	"github.com/activecm/ipfix-rita/converter/output/rita"
 	"github.com/urfave/cli"
 )
@@ -35,7 +35,7 @@ func init() {
 			}
 			fmt.Printf("Loaded Configuration:\n%s\n", confStr)
 
-			db, err := mgologstash.NewLogstashMongoInputDB(conf.GetInputConfig().GetLogstashMongoDBConfig())
+			db, err := mongodb.NewLogstashMongoInputDB(conf.GetInputConfig().GetLogstashMongoDBConfig())
 			if err != nil {
 				return cli.NewExitError(fmt.Sprintf("%+v\n", err), 1)
 			}

converter/commands/convert.go

@@ -2,14 +2,16 @@ package commands
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
 	"github.com/activecm/ipfix-rita/converter/environment"
-	input "github.com/activecm/ipfix-rita/converter/input/mgologstash"
+	"github.com/activecm/ipfix-rita/converter/filter"
+	input "github.com/activecm/ipfix-rita/converter/input/logstash/mongodb"
 	"github.com/activecm/ipfix-rita/converter/logging"
 	"github.com/activecm/ipfix-rita/converter/output"
 	batchRITAOutput "github.com/activecm/ipfix-rita/converter/output/rita/batch/dates"
@@ -54,6 +56,8 @@ func convert(env environment.Environment, noRotate bool) error {
 	//TODO: Decide on how to scale these options depending on the specs
 	//of the computer
 
+	//-------------------------------Input setup-------------------------------
+
 	//pollWait is how long to wait before checking if the input buffer has
 	//more data
 	pollWait := 30 * time.Second
@@ -81,6 +85,40 @@ func convert(env environment.Environment, noRotate bool) error {
 		env.Logger,
 	)
 
+	//-------------------------------Filter setup-------------------------------
+
+	//Create the filter which will filter out flows as specified by the
+	//Filter config section
+	internalNets, errs := env.GetFilteringConfig().GetInternalSubnets()
+	if len(errs) != 0 {
+		for _, err := range errs {
+			env.Logger.Error(err, nil)
+		}
+		return errors.New("unable to parse filtering config")
+	}
+	neverIncludeNets, errs := env.GetFilteringConfig().GetNeverIncludeSubnets()
+	if len(errs) != 0 {
+		for _, err := range errs {
+			env.Logger.Error(err, nil)
+		}
+		return errors.New("unable to parse filtering config")
+	}
+	alwaysIncludeNets, errs := env.GetFilteringConfig().GetAlwaysIncludeSubnets()
+	if len(errs) != 0 {
+		for _, err := range errs {
+			env.Logger.Error(err, nil)
+		}
+		return errors.New("unable to parse filtering config")
+	}
+
+	flowFilter := filter.NewFlowBlacklist(
+		internalNets,
+		neverIncludeNets,
+		alwaysIncludeNets,
+	)
+
+	//------------------------------Stitching setup------------------------------
+
 	//sameSessionThreshold determines is used in the process of determining
 	//whether two flows should be stitched together or not.
 	//If the time between one flow ending and the other flow starting
@@ -132,9 +170,12 @@ func convert(env environment.Environment, noRotate bool) error {
 		outputBufferSize,
 		matcherSize,
 		matcherFlushToPercent,
+		flowFilter,
 		env.Logger,
 	)
 
+	//-------------------------------Output setup-------------------------------
+
 	//flushDeadline determines how long data may sit in a buffer
 	//before it is exported to MongoDB
 	flushDeadline := 1 * time.Minute
@@ -152,7 +193,7 @@ func convert(env environment.Environment, noRotate bool) error {
 		//which splits output records up based on the time the connection finished
 		writer, err = streamingRITAOutput.NewStreamingRITATimeIntervalWriter(
 			env.GetOutputConfig().GetRITAConfig(),
-			env.GetIPFIXConfig(),
+			internalNets,
 			bulkBatchSize, flushDeadline,
 			dayRotationPeriodMillis, gracePeriodMillis,
 			clock.New(), time.Local, dateFormatString,
@@ -164,7 +205,7 @@ func convert(env environment.Environment, noRotate bool) error {
 	} else {
 		writer, err = batchRITAOutput.NewBatchRITAConnDateWriter(
 			env.GetOutputConfig().GetRITAConfig(),
-			env.GetIPFIXConfig(),
+			internalNets,
 			bulkBatchSize, flushDeadline,
 			env.Logger,
 		)
@@ -174,6 +215,8 @@ func convert(env environment.Environment, noRotate bool) error {
 		env.Info("Database rotation has been disabled", nil)
 	}
 
+	//-------------------------------Execution-------------------------------
+
 	//input channels
 	inputData, inputErrors := reader.Drain(ctx)
 	//run the stitching manager and get the output channels

converter/config/config.go

@@ -26,8 +26,8 @@ import (
 //Config holds IPFIX-RITA (converter) configuration details
 type Config interface {
 	GetInputConfig() Input
+	GetFilteringConfig() Filtering
 	GetOutputConfig() Output
-	GetIPFIXConfig() IPFIX
 }
 
 //Serializable represents application configuration data
@@ -71,16 +71,18 @@ type Output interface {
 	GetRITAConfig() RITA
 }
 
-//RITA2 contains configuration for writing out the
+//RITA contains configuration for writing out the
 //stitched IPFIX/ Netflow records RITA compatible MongoDB databases
 type RITA interface {
 	GetConnectionConfig() MongoDBConnection
 	GetDBRoot() string
 	GetMetaDB() string
 }
 
-//IPFIX provides information for accessing IPFIX data
-//and information regarding the individual records
-type IPFIX interface {
-	GetLocalNetworks() ([]net.IPNet, []error)
+//Filtering contains information on local subnets and other networks/hosts
+//that should be filtered out of the result set
+type Filtering interface {
+	GetAlwaysIncludeSubnets() ([]net.IPNet, []error)
+	GetNeverIncludeSubnets() ([]net.IPNet, []error)
+	GetInternalSubnets() ([]net.IPNet, []error)
 }

converter/config/yaml/filtering.go

@@ -0,0 +1,59 @@
+package yaml
+
+import "net"
+import "github.com/pkg/errors"
+
+//filtering implements config.Filtering
+type filtering struct {
+	AlwaysInclude   []string `yaml:"AlwaysInclude"`
+	NeverInclude    []string `yaml:"NeverInclude"`
+	InternalSubnets []string `yaml:"InternalSubnets"`
+}
+
+func (f *filtering) GetAlwaysIncludeSubnets() ([]net.IPNet, []error) {
+	return f.parseSubnetList(f.AlwaysInclude)
+}
+
+func (f *filtering) GetNeverIncludeSubnets() ([]net.IPNet, []error) {
+	return f.parseSubnetList(f.NeverInclude)
+}
+
+func (f *filtering) GetInternalSubnets() ([]net.IPNet, []error) {
+	return f.parseSubnetList(f.InternalSubnets)
+}
+
+func (f *filtering) parseSubnetList(netList []string) ([]net.IPNet, []error) {
+	var errorList []error
+	var nets []net.IPNet
+	for j := range netList {
+		//parse as network
+		_, network, err := net.ParseCIDR(netList[j])
+		if err != nil {
+			//parse as IP
+			ipAddr := net.ParseIP(netList[j])
+
+			if ipAddr == nil {
+				errorList = append(errorList, errors.WithStack(err))
+				continue
+			}
+
+			network = f.ipToIPNet(ipAddr)
+		}
+
+		nets = append(nets, *network)
+	}
+	return nets, errorList
+}
+
+func (f *filtering) ipToIPNet(ipAddr net.IP) *net.IPNet {
+	var netmask net.IPMask
+	if ipAddr.To4() == nil {
+		netmask = net.CIDRMask(32, 32)
+	} else {
+		netmask = net.CIDRMask(128, 128)
+	}
+	return &net.IPNet{
+		IP:   ipAddr,
+		Mask: netmask,
+	}
+}

converter/config/yaml/yaml.go

@@ -32,9 +32,9 @@ func ReadConfigFile() ([]byte, error) {
 //yamlConfig contains the applications settings
 //as represented by a YAML string. Implements config.Config
 type yamlConfig struct {
-	Input  input  `yaml:"Input"`
-	Output output `yaml:"Output"`
-	IPFIX  ipfix  `yaml:"IPFIX"`
+	Input     input     `yaml:"Input"`
+	Output    output    `yaml:"Output"`
+	Filtering filtering `yaml:"Filtering"`
 }
 
 func (y *yamlConfig) GetInputConfig() config.Input {
@@ -45,8 +45,8 @@ func (y *yamlConfig) GetOutputConfig() config.Output {
 	return &y.Output
 }
 
-func (y *yamlConfig) GetIPFIXConfig() config.IPFIX {
-	return &y.IPFIX
+func (y *yamlConfig) GetFilteringConfig() config.Filtering {
+	return &y.Filtering
 }
 
 //NewYAMLConfig creates a new yamlConfig from