[Filebeat] Fixes for NetFlow v9 devices from various vendors

CHANGELOG.next.asciidoc

@@ -230,6 +230,9 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix SSL config in input.yml for Filebeat httpjson input in the MISP module. {pull}14767[14767]
 - Check content-type when creating new reader in s3 input. {pull}15252[15252] {issue}15225[15225]
 - Fix session reset detection and a crash in Netflow input. {pull}14904[14904]
+- netflow: Allow for options templates without scope fields. {pull}15449[15449]
+- netflow: Fix bytes/packets counters on some devices (NSEL and Netstream). {pull}15449[15449]
+- netflow: Fix compatibility with some Cisco devices by changing the field `class_id` from short to long. {pull}15449[15449]
 
 *Heartbeat*
 

x-pack/filebeat/input/netflow/_meta/fields.yml

@@ -194,7 +194,7 @@
           type: long
 
         - name: class_id
-          type: short
+          type: long
 
         - name: minimum_ttl
           type: short

x-pack/filebeat/input/netflow/convert.go

@@ -245,22 +245,10 @@ func flowToBeatEvent(flow record.Record) (event beat.Event) {
 		ecsNetwork["transport"] = IPProtocol(proto).String()
 		ecsNetwork["iana_number"] = proto
 	}
-	countBytes, hasBytes := getKeyUint64(flow.Fields, "octetDeltaCount")
-	if !hasBytes {
-		countBytes, hasBytes = getKeyUint64(flow.Fields, "octetTotalCount")
-	}
-	countPkts, hasPkts := getKeyUint64(flow.Fields, "packetDeltaCount")
-	if !hasPkts {
-		countPkts, hasPkts = getKeyUint64(flow.Fields, "packetTotalCount")
-	}
-	revBytes, hasRevBytes := getKeyUint64(flow.Fields, "reverseOctetDeltaCount")
-	if !hasRevBytes {
-		revBytes, hasRevBytes = getKeyUint64(flow.Fields, "reverseOctetTotalCount")
-	}
-	revPkts, hasRevPkts := getKeyUint64(flow.Fields, "reversePacketDeltaCount")
-	if !hasRevPkts {
-		revPkts, hasRevPkts = getKeyUint64(flow.Fields, "reversePacketTotalCount")
-	}
+	countBytes, hasBytes := getKeyUint64Alternatives(flow.Fields, "octetDeltaCount", "octetTotalCount", "initiatorOctets")
+	countPkts, hasPkts := getKeyUint64Alternatives(flow.Fields, "packetDeltaCount", "packetTotalCount", "initiatorPackets")
+	revBytes, hasRevBytes := getKeyUint64Alternatives(flow.Fields, "reverseOctetDeltaCount", "reverseOctetTotalCount", "responderOctets")
+	revPkts, hasRevPkts := getKeyUint64Alternatives(flow.Fields, "reversePacketDeltaCount", "reversePacketTotalCount", "responderPackets")
 
 	if hasRevBytes {
 		ecsDest["bytes"] = revBytes
@@ -337,6 +325,18 @@ func getKeyUint64(dict record.Map, key string) (value uint64, found bool) {
 	return
 }
 
+func getKeyUint64Alternatives(dict record.Map, keys ...string) (value uint64, found bool) {
+	var iface interface{}
+	for _, key := range keys {
+		if iface, found = dict[key]; found {
+			if value, found = iface.(uint64); found {
+				return
+			}
+		}
+	}
+	return
+}
+
 func getKeyString(dict record.Map, key string) (value string, found bool) {
 	iface, found := dict[key]
 	if !found {

x-pack/filebeat/input/netflow/decoder/fields/cisco.csv

@@ -270,6 +270,20 @@ netscalerUnknown465,5951,465,unsigned32
 ingressAclID,0,33000,aclid
 egressAclID,0,33001,aclid
 fwExtEvent,0,33002,unsigned16
+fwEventLevel,0,33003,unsigned32
+fwEventLevelID,0,33004,unsigned32
+fwConfiguredValue,0,33005,unsigned32
+fwCtsSrcSGT,0,34000,unsigned32
+fwExtEventAlt,0,35001,unsigned32
+fwBlackoutSecs,0,35004,unsigned32
+fwHalfOpenHigh,0,35005,unsigned32
+fwHalfOpenRate,0,35006,unsigned32
+fwZonePairID,0,35007,unsigned32
+fwMaxSessions,0,35008,unsigned32
+fwZonePairName,0,35009,unsigned32
+fwExtEventDesc,0,35010,string
+fwSummaryPktCount,0,35011,unsigned32
+fwHalfOpenCount,0,35012,unsigned32
 username,0,40000,string
 XlateSourceAddressIPV4,0,40001,ipv4Address
 XlateDestinationAddressIPV4,0,40002,ipv4Address

x-pack/filebeat/input/netflow/decoder/fields/doc.go

@@ -8,3 +8,4 @@ package fields
 //go:generate go run gen.go -output zfields_cert.go -export CertFields --column-pen=1 --column-id=2 --column-name=3 --column-type=4 cert_pen6871.csv
 //go:generate go run gen.go -output zfields_cisco.go -export CiscoFields --column-pen=2 --column-id=3 --column-name=1 --column-type=4 cisco.csv
 //go:generate go run gen.go -output zfields_assorted.go -export AssortedFields --column-pen=1 --column-id=2 --column-name=3 --column-type=4 assorted.csv
+//go:generate go fmt

x-pack/filebeat/input/netflow/decoder/fields/ipfix-information-elements.csv

@@ -1,3 +1,9 @@
+; WARNING: This is an edited version of the original IANA document!
+;
+; Changes
+; =======
+; 2020-01-14 - @adriansr: Change field 51 (classId) from unsigned8 to unsigned32
+;
 ;ElementID,Name,Abstract Data Type,Data Type Semantics,Status,Description,Units,Range,References,Requester,Revision,Date
 0,Reserved,,,,,,,,[RFC5102],,2013-02-18
 1,octetDeltaCount,unsigned64,deltaCounter,current,"The number of octets since the previous report (if any)
@@ -335,7 +341,7 @@ Sampling.  Use with samplerRandomInterval.",,,,[RFC7270],0,2014-04-04
 50,samplerRandomInterval,unsigned32,quantity,deprecated,"Deprecated in favor of 305 samplingPacketInterval.  Packet
 interval at which to sample -- in case of random sampling.  Used in
 connection with the samplerMode 0x02 (random sampling) value.",,,,[RFC7270],0,2014-04-04
-51,classId,unsigned8,identifier,deprecated,"Deprecated in favor of 302 selectorId.  Characterizes the traffic
+51,classId,unsigned32,identifier,deprecated,"Deprecated in favor of 302 selectorId.  Characterizes the traffic
 class, i.e., QoS treatment.",,,,[RFC7270],0,2014-04-04
 52,minimumTTL,unsigned8,,current,Minimum TTL value observed for any packet in this Flow.,hops,,"See [RFC791] for the definition of the IPv4
 Time to Live field.

x-pack/filebeat/input/netflow/decoder/fields/zfields_cisco.go

@@ -280,6 +280,20 @@ var CiscoFields = FieldDict{
 	Key{EnterpriseID: 0, FieldID: 33000}:  {Name: "ingressAclID", Decoder: ACLID},
 	Key{EnterpriseID: 0, FieldID: 33001}:  {Name: "egressAclID", Decoder: ACLID},
 	Key{EnterpriseID: 0, FieldID: 33002}:  {Name: "fwExtEvent", Decoder: Unsigned16},
+	Key{EnterpriseID: 0, FieldID: 33003}:  {Name: "fwEventLevel", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 33004}:  {Name: "fwEventLevelID", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 33005}:  {Name: "fwConfiguredValue", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 34000}:  {Name: "fwCtsSrcSGT", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35001}:  {Name: "fwExtEventAlt", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35004}:  {Name: "fwBlackoutSecs", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35005}:  {Name: "fwHalfOpenHigh", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35006}:  {Name: "fwHalfOpenRate", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35007}:  {Name: "fwZonePairID", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35008}:  {Name: "fwMaxSessions", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35009}:  {Name: "fwZonePairName", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35010}:  {Name: "fwExtEventDesc", Decoder: String},
+	Key{EnterpriseID: 0, FieldID: 35011}:  {Name: "fwSummaryPktCount", Decoder: Unsigned32},
+	Key{EnterpriseID: 0, FieldID: 35012}:  {Name: "fwHalfOpenCount", Decoder: Unsigned32},
 	Key{EnterpriseID: 0, FieldID: 40000}:  {Name: "username", Decoder: String},
 	Key{EnterpriseID: 0, FieldID: 40001}:  {Name: "XlateSourceAddressIPV4", Decoder: Ipv4Address},
 	Key{EnterpriseID: 0, FieldID: 40002}:  {Name: "XlateDestinationAddressIPV4", Decoder: Ipv4Address},

x-pack/filebeat/input/netflow/decoder/fields/zfields_ipfix.go

@@ -59,7 +59,7 @@ var IpfixFields = FieldDict{
 	Key{EnterpriseID: 0, FieldID: 48}: {Name: "samplerId", Decoder: Unsigned8},
 	Key{EnterpriseID: 0, FieldID: 49}: {Name: "samplerMode", Decoder: Unsigned8},
 	Key{EnterpriseID: 0, FieldID: 50}: {Name: "samplerRandomInterval", Decoder: Unsigned32},
-	Key{EnterpriseID: 0, FieldID: 51}: {Name: "classId", Decoder: Unsigned8},
+	Key{EnterpriseID: 0, FieldID: 51}: {Name: "classId", Decoder: Unsigned32},
 	Key{EnterpriseID: 0, FieldID: 52}: {Name: "minimumTTL", Decoder: Unsigned8},
 	Key{EnterpriseID: 0, FieldID: 53}: {Name: "maximumTTL", Decoder: Unsigned8},
 	Key{EnterpriseID: 0, FieldID: 54}: {Name: "fragmentIdentification", Decoder: Unsigned32},

x-pack/filebeat/input/netflow/decoder/ipfix/decoder.go

@@ -108,6 +108,7 @@ func (d DecoderIPFIX) ReadOptionsTemplateFlowSet(buf *bytes.Buffer) (templates [
 		}
 		template.ID = tID
 		template.ScopeFields = scopeCount
+		template.IsOptions = true
 		templates = append(templates, &template)
 	}
 	return templates, nil

x-pack/filebeat/input/netflow/decoder/template/template.go

@@ -29,6 +29,9 @@ type Template struct {
 	Length         int
 	VariableLength bool
 	ScopeFields    int
+	// IsOptions signals that this is an options template. Previously
+	// ScopeFields>0 was used for this, but that's unreliable under v9.
+	IsOptions bool
 }
 
 type FieldTemplate struct {
@@ -84,7 +87,7 @@ func (t *Template) Apply(data *bytes.Buffer, n int) ([]record.Record, error) {
 		}
 	}
 	makeFn := t.makeFlow
-	if t.ScopeFields > 0 {
+	if t.IsOptions {
 		makeFn = t.makeOptions
 	}
 	events := make([]record.Record, 0, alloc)

x-pack/filebeat/input/netflow/decoder/template/template_test.go

@@ -313,6 +313,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:      7,
 				ScopeFields: 1,
+				IsOptions:   true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: 2, Info: &fields.Field{Name: "destinationTransportPort", Decoder: fields.Unsigned16}},
@@ -343,6 +344,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:      7,
 				ScopeFields: 2,
+				IsOptions:   true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: 2, Info: &fields.Field{Name: "destinationTransportPort", Decoder: fields.Unsigned16}},
@@ -386,6 +388,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:      7,
 				ScopeFields: 3,
+				IsOptions:   true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: 2, Info: &fields.Field{Name: "destinationTransportPort", Decoder: fields.Unsigned16}},
@@ -415,6 +418,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:      7,
 				ScopeFields: 1,
+				IsOptions:   true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: 2, Info: &fields.Field{Name: "destinationTransportPort", Decoder: fields.Unsigned16}},
@@ -446,6 +450,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:      7,
 				ScopeFields: 2,
+				IsOptions:   true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: 2, Info: &fields.Field{Name: "destinationTransportPort", Decoder: fields.Unsigned16}},
@@ -489,6 +494,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:         6,
 				ScopeFields:    1,
+				IsOptions:      true,
 				VariableLength: true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
@@ -522,6 +528,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 			record: Template{
 				Length:         6,
 				ScopeFields:    1,
+				IsOptions:      true,
 				VariableLength: true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
@@ -571,6 +578,7 @@ func TestOptionsTemplate_Apply(t *testing.T) {
 				Length:         6,
 				VariableLength: true,
 				ScopeFields:    2,
+				IsOptions:      true,
 				Fields: []FieldTemplate{
 					{Length: 4, Info: &fields.Field{Name: "sourceIPv4Address", Decoder: fields.Ipv4Address}},
 					{Length: VariableLength, Info: &fields.Field{Name: "vpnIdentifier", Decoder: fields.OctetArray}},

x-pack/filebeat/input/netflow/decoder/v9/decoder.go

@@ -184,7 +184,7 @@ func (d DecoderV9) ReadOptionsTemplateFlowSet(buf *bytes.Buffer) (templates []*t
 		if buf.Len() < int(length) {
 			return nil, io.EOF
 		}
-		if scopeLen == 0 || scopeLen&3 != 0 || optsLen&3 != 0 {
+		if (scopeLen+optsLen) == 0 || scopeLen&3 != 0 || optsLen&3 != 0 {
 			return nil, fmt.Errorf("bad length for options template. scope=%d options=%d", scopeLen, optsLen)
 		}
 		template, err := ReadFields(d, buf, (scopeLen+optsLen)/4)
@@ -193,6 +193,7 @@ func (d DecoderV9) ReadOptionsTemplateFlowSet(buf *bytes.Buffer) (templates []*t
 		}
 		template.ID = tID
 		template.ScopeFields = scopeLen / 4
+		template.IsOptions = true
 		templates = append(templates, &template)
 	}
 	return templates, nil