[Filebeat] Fixes for NetFlow v9 devices from various vendors (#15449) (…

…#15554)

- Allow for zero scope fields in options template

NetFlow v9 spec allows for options templates that contain no scope
fields. The netflow input was treating this case as an error and
discarding the template, but that is only applicable to IPFIX.

- Use additional fields to populate bytes/pkt counters

Some devices out there (Cisco NSEL) use fields 231/232 as bytes
counters, when those are supposed to be layer 4 payload counters.

This updates the ECS fields populator to use those fields when the
expected ones are not found.

- Support a classId of 32 bits

While the spec mandates a classId of 8 bits, some Cisco ASA devices
actually use a 32 bit version of this field.

This patches the field to allow up to 32-bit integers and updates the
index pattern to use `long` for the `netflow.class_id` field.

- Add more fields from v9 Cisco devices

Fixes #14212

(cherry picked from commit c3a3604)

CHANGELOG.next.asciidoc

@@ -71,6 +71,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*
 

filebeat/docs/fields.asciidoc

@@ -16972,7 +16972,7 @@ type: long
 *`netflow.class_id`*::
 +
 --
-type: short
+type: long
 
 --
 

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