decoders/netflow: add variable-length fields parsing support

Signed-off-by: b.koshenov <[email protected]>

Author: b.koshenov

decoders/netflow/netflow.go

@@ -2,6 +2,7 @@ package netflow
 
 import (
 	"bytes"
+	"encoding/binary"
 	"fmt"
 	"io"
 	"sync"
@@ -164,18 +165,28 @@ func GetTemplateSize(template []Field) int {
 
 // DecodeDataRecordFields decodes the fields(type and value) of DataRecord.
 func DecodeDataRecordFields(payload *bytes.Buffer, listFields []Field, record *[]DataField) error {
-	size := GetTemplateSize(listFields)
-	if payload.Len() < size {
-		return fmt.Errorf("decode dataset: there are fewer than %d bytes in the buffer", size)
-	}
-
 	*record = (*record)[:cap(*record)]
 	if len(*record) < len(listFields) {
 		*record = append(*record, make([]DataField, len(listFields)-len(*record))...)
 	}
 
+	var ok bool
 	for i, templateField := range listFields {
-		value := payload.Next(int(templateField.Length))
+
+		l := int(templateField.Length)
+
+		if templateField.IsVariableLength() {
+			if l, ok = getVariableLength(payload); !ok {
+				return fmt.Errorf("decode DataRecordFields: invalid variable-length: %d", l)
+			}
+		}
+
+		// XXX: Retaining a slice returned by Next() may be unsafe according to
+		// method's documentation as it may be invalidated by future Read call.
+		value := payload.Next(l)
+		if len(value) < l {
+			return fmt.Errorf("decode dataset: there are fewer than %d bytes in the buffer", l)
+		}
 
 		(*record)[i].Type = templateField.Type
 		(*record)[i].Value = value
@@ -184,6 +195,31 @@ func DecodeDataRecordFields(payload *bytes.Buffer, listFields []Field, record *[
 	return nil
 }
 
+func getVariableLength(payload *bytes.Buffer) (int, bool) {
+	b, err := payload.ReadByte()
+	if err != nil {
+		return 0, false
+	}
+
+	// RFC 7011 Sec. 7.Variable-Length Information Element.
+	if b != 0xff {
+		return int(b), true
+	}
+
+	// RFC 7011 Sec. 7.Variable-Length Information Element:
+	// The length may also be encoded into 3 octets before the Information
+	// Element, allowing the length of the Information Element to be greater
+	// than or equal to 255 octets. In this case, the first octet of the Length
+	// field MUST be 255, and the length is carried in the second and third
+	// octets.
+	length := payload.Next(2)
+	if len(length) < 2 {
+		return 0, false
+	}
+
+	return int(binary.BigEndian.Uint16(length)), true
+}
+
 type ErrorTemplateNotFound struct {
 	version      uint16
 	obsDomainId  uint32
@@ -281,24 +317,19 @@ func DecodeOptionsDataSet(payload *bytes.Buffer, fs *OptionsDataFlowSet, listFie
 func DecodeDataSet(payload *bytes.Buffer, listFields []Field, flowSet *DataFlowSet) error {
 	flowSet.Records = flowSet.Records[:cap(flowSet.Records)]
 
-	listFieldsSize := GetTemplateSize(listFields)
-
 	i := 0
-	for payload.Len() >= listFieldsSize {
-		payloadLim := bytes.NewBuffer(payload.Next(listFieldsSize))
-
+	for payload.Len() > 0 {
 		if i >= len(flowSet.Records) {
 			flowSet.Records = append(flowSet.Records, DataRecord{})
 		}
 
 		datafields := &flowSet.Records[i].Values
-		if err := DecodeDataRecordFields(payloadLim, listFields, datafields); err != nil {
+		if err := DecodeDataRecordFields(payload, listFields, datafields); err != nil {
 			return err
 		}
 
 		i++
 	}
-
 	flowSet.Records = flowSet.Records[:i]
 	return nil
 }
@@ -533,7 +564,9 @@ func (f *FlowMessage) DecodeIPFIXPacket(payload *bytes.Buffer, templates NetFlow
 	)
 
 	for i := 0; i < int(f.PacketIPFIX.Length) && payload.Len() > 0; i++ {
-		f.fsheader.ReadFrom(payload)
+		if ok := f.fsheader.ReadFrom(payload); !ok {
+			return NewErrorDecodingNetFlow("error decoding packet: invalid FlowSet header.")
+		}
 
 		nextrelpos := int(f.fsheader.Length) - flowSetHeaderSize
 		if nextrelpos < 0 {

decoders/netflow/netflow_test.go

@@ -222,10 +222,10 @@ func TestDecodeDataSet(t *testing.T) {
 	fs := DataFlowSet{}
 
 	// type and length of a single value in a netflow Data Record
-	nfListFields := []Field{{8, 4}, {225, 4}, {12, 4}, {226, 4}, {7, 2}, {227, 2}, {11, 2}, {228, 2}, {234, 4}, {4, 1}, {230, 1}, {323, 8}}
+	nfListFields := []Field{{8, 4, 0}, {225, 4, 0}, {12, 4, 0}, {226, 4, 0}, {7, 2, 0}, {227, 2, 0}, {11, 2, 0}, {228, 2, 0}, {234, 4, 0}, {4, 1, 0}, {230, 1, 0}, {323, 8, 0}}
 
 	// type and length of a single value in ipfix Flow Data Record
-	ipxListFields := []Field{{8, 4}, {12, 4}, {5, 1}, {4, 1}, {7, 2}, {11, 2}, {32, 2}, {10, 4}, {16, 4}, {17, 4}, {18, 4}, {14, 4}, {1, 4}, {2, 4}, {22, 4}, {21, 4}, {15, 4}, {9, 1}, {13, 1}, {6, 1}, {60, 1}, {152, 8}, {153, 8}}
+	ipxListFields := []Field{{8, 4, 0}, {12, 4, 0}, {5, 1, 0}, {4, 1, 0}, {7, 2, 0}, {11, 2, 0}, {32, 2, 0}, {10, 4, 0}, {16, 4, 0}, {17, 4, 0}, {18, 4, 0}, {14, 4, 0}, {1, 4, 0}, {2, 4, 0}, {22, 4, 0}, {21, 4, 0}, {15, 4, 0}, {9, 1, 0}, {13, 1, 0}, {6, 1, 0}, {60, 1, 0}, {152, 8, 0}, {153, 8, 0}}
 
 	// netflow data flowset
 	nfData := bytes.NewBuffer(netflowTestPackets[1].Data[210:])
@@ -245,10 +245,10 @@ func TestDecodeDataSet(t *testing.T) {
 
 func TestDecodeDataSetUsingFields(t *testing.T) {
 	// type and length of a single value in a netflow Data Record
-	nfListFields := []Field{{8, 4}, {225, 4}, {12, 4}, {226, 4}, {7, 2}, {227, 2}, {11, 2}, {228, 2}, {234, 4}, {4, 1}, {230, 1}, {323, 8}}
+	nfListFields := []Field{{8, 4, 0}, {225, 4, 0}, {12, 4, 0}, {226, 4, 0}, {7, 2, 0}, {227, 2, 0}, {11, 2, 0}, {228, 2, 0}, {234, 4, 0}, {4, 1, 0}, {230, 1, 0}, {323, 8, 0}}
 
 	// type and length of a single value in ipfix Flow Data Record
-	ipxListFields := []Field{{8, 4}, {12, 4}, {5, 1}, {4, 1}, {7, 2}, {11, 2}, {32, 2}, {10, 4}, {16, 4}, {17, 4}, {18, 4}, {14, 4}, {1, 4}, {2, 4}, {22, 4}, {21, 4}, {15, 4}, {9, 1}, {13, 1}, {6, 1}, {60, 1}, {152, 8}, {153, 8}}
+	ipxListFields := []Field{{8, 4, 0}, {12, 4, 0}, {5, 1, 0}, {4, 1, 0}, {7, 2, 0}, {11, 2, 0}, {32, 2, 0}, {10, 4, 0}, {16, 4, 0}, {17, 4, 0}, {18, 4, 0}, {14, 4, 0}, {1, 4, 0}, {2, 4, 0}, {22, 4, 0}, {21, 4, 0}, {15, 4, 0}, {9, 1, 0}, {13, 1, 0}, {6, 1, 0}, {60, 1, 0}, {152, 8, 0}, {153, 8, 0}}
 
 	dr := DataRecord{}
 

decoders/netflow/packet.go

@@ -114,6 +114,24 @@ type Field struct {
 
 	// The length (in bytes) of the field.
 	Length uint16
+
+	// IANA private enterprise number (PEN) of the authority defining the
+	// Information Element identifier in this Template Record.
+	//
+	// Note: it's valid only for v10(IPFIX).
+	EnterpriseNumber uint32
+}
+
+// IsVariableLength reports whether the Field is variable-length Information
+// Element (RFC 7011 sec. 3.2. Field Specifier Format).
+func (f *Field) IsVariableLength() bool {
+	return f.Length == 0xffff
+}
+
+// ContainsEnterpriseNumber reports whether the Field is an enterprise-specific
+// Information Element (RFC 7011 sec. 3.2. Field Specifier Format).
+func (f *Field) ContainsEnterpriseNumber() bool {
+	return f.Type&0x8000 != 0
 }
 
 // DataField contains the type and record value itself.
@@ -161,9 +179,17 @@ func (f *Field) ReadFrom(b *bytes.Buffer) bool {
 	if ok := utils.ReadUint16FromBuffer(b, &f.Type); !ok {
 		return false
 	}
+
 	if ok := utils.ReadUint16FromBuffer(b, &f.Length); !ok {
 		return false
 	}
+
+	if f.ContainsEnterpriseNumber() {
+		if ok := utils.ReadUint32FromBuffer(b, &f.EnterpriseNumber); !ok {
+			return false
+		}
+	}
+
 	return true
 }