Extended IPFIX decoder: support decoding messages with multiple sets and sets with multiple records. More robust decoding: if decoding a set fails, still decode subsequent sets. Added ReadCount() and Peek() functionality to reader.Reader and allSetIds() to MemCache.

Author: antongulenko

ipfix/decoder.go

@@ -23,6 +23,8 @@
 package ipfix
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -96,75 +98,115 @@ func NewDecoder(raddr net.IP, b []byte) *Decoder {
 
 // Decode decodes the IPFIX raw data
 func (d *Decoder) Decode(mem MemCache) (*Message, error) {
-	var (
-		msg = new(Message)
-		err error
-	)
+	var msg = new(Message)
 
 	// IPFIX Message Header decoding
-	if err = msg.Header.unmarshal(d.reader); err != nil {
+	if err := msg.Header.unmarshal(d.reader); err != nil {
 		return nil, err
 	}
 	// IPFIX Message Header validation
-	if err = msg.Header.validate(); err != nil {
+	if err := msg.Header.validate(); err != nil {
 		return nil, err
 	}
 
 	// Add source IP address as Agent ID
 	msg.AgentID = d.raddr.String()
 
+	// In case there are multiple non-fatal errors, collect them and report all of them.
+	// The rest of the received sets will still be interpreted, until a fatal error is encountered.
+	// A non-fatal error is for example an illegal data record or unknown template id.
+	var decodeErrors []error
 	for d.reader.Len() > 4 {
+		if err := d.decodeSet(mem, msg); err != nil {
+			switch err.(type) {
+			case nonfatalError:
+				decodeErrors = append(decodeErrors, err)
+			default:
+				return nil, err
+			}
+		}
+	}
 
-		setHeader := new(SetHeader)
-		setHeader.unmarshal(d.reader)
+	return msg, combineErrors(decodeErrors...)
+}
 
-		if setHeader.Length < 4 {
-			return nil, io.ErrUnexpectedEOF
-		}
+type nonfatalError error
 
-		switch {
-		case setHeader.SetID == 2:
-			// Template set
-			tr := TemplateRecord{}
-			tr.unmarshal(d.reader)
-			mem.insert(tr.TemplateID, d.raddr, tr)
-		case setHeader.SetID == 3:
-			// Option set
-			tr := TemplateRecord{}
-			tr.unmarshalOpts(d.reader)
-			mem.insert(tr.TemplateID, d.raddr, tr)
-		case setHeader.SetID >= 4 && setHeader.SetID <= 255:
-			// Reserved
-		default:
-			// data
-			tr, ok := mem.retrieve(setHeader.SetID, d.raddr)
-			if !ok {
-				select {
-				case rpcChan <- RPCRequest{
-					ID: setHeader.SetID,
-					IP: d.raddr,
-				}:
-				default:
-				}
-				return msg, fmt.Errorf("%s unknown ipfix template id# %d",
-					d.raddr.String(),
-					setHeader.SetID,
-				)
+func (d *Decoder) decodeSet(mem MemCache, msg *Message) error {
+	startCount := d.reader.ReadCount()
+
+	setHeader := new(SetHeader)
+	if err := setHeader.unmarshal(d.reader); err != nil {
+		return err
+	}
+	if setHeader.Length < 4 {
+		return io.ErrUnexpectedEOF
+	}
+
+	var tr TemplateRecord
+	var err error
+	// This check is somewhat redundant with the switch-clause below, but the retrieve() operation should not be executed inside the loop.
+	if setHeader.SetID > 255 {
+		var ok bool
+		tr, ok = mem.retrieve(setHeader.SetID, d.raddr)
+		if !ok {
+			select {
+			case rpcChan <- RPCRequest{
+				ID: setHeader.SetID,
+				IP: d.raddr,
+			}:
+			default:
 			}
+			err = nonfatalError(fmt.Errorf("%s unknown ipfix template id# %d. Known ids: %v",
+				d.raddr.String(),
+				setHeader.SetID,
+				mem.allSetIds(),
+			))
+		}
+	}
 
-			// data records
-			for d.reader.Len() > 2 {
-				data, err := decodeData(d.reader, tr)
-				if err != nil {
-					return msg, err
-				}
+	for err == nil && setHeader.Length > uint16(d.reader.ReadCount()-startCount) {
+		if setId := setHeader.SetID; setId == 2 || setId == 3 {
+			// Template record or template option record
 
+			// Check if only padding is left in this set. A template id of zero indicates padding bytes, which MUST be zero.
+			templateId, err := d.reader.PeekUint16()
+			if err == nil && templateId == 0 {
+				break
+			}
+
+			tr := TemplateRecord{}
+			if setId == 2 {
+				err = tr.unmarshal(d.reader)
+			} else {
+				err = tr.unmarshalOpts(d.reader)
+			}
+			if err == nil {
+				mem.insert(tr.TemplateID, d.raddr, tr)
+			}
+		} else if setId >= 4 && setId <= 255 {
+			// Reserved set, do not read any records
+			break
+		} else {
+			// Data set
+			var data []DecodedField
+			data, err = d.decodeData(tr)
+			if err == nil {
 				msg.DataSets = append(msg.DataSets, data)
 			}
 		}
 	}
 
-	return msg, nil
+	// Skip the rest of the set in order to properly continue with the next set
+	// This is necessary if the set is padded, has a reserved set ID, or a nonfatal error occurred
+	leftoverBytes := setHeader.Length - uint16(d.reader.ReadCount()-startCount)
+	if leftoverBytes > 0 {
+		_, skipErr := d.reader.Read(int(leftoverBytes))
+		if skipErr != nil {
+			err = skipErr
+		}
+	}
+	return err
 }
 
 // RFC 7011 - part 3.1. Message Header Format
@@ -336,20 +378,25 @@ func (f *TemplateFieldSpecifier) unmarshal(r *reader.Reader) error {
 // |             ...               |              ...              |
 // +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 
-func (tr *TemplateRecord) unmarshal(r *reader.Reader) {
+func (tr *TemplateRecord) unmarshal(r *reader.Reader) error {
 	var (
 		th = TemplateHeader{}
 		tf = TemplateFieldSpecifier{}
 	)
 
-	th.unmarshal(r)
+	if err := th.unmarshal(r); err != nil {
+		return err
+	}
 	tr.TemplateID = th.TemplateID
 	tr.FieldCount = th.FieldCount
 
 	for i := th.FieldCount; i > 0; i-- {
-		tf.unmarshal(r)
+		if err := tf.unmarshal(r); err != nil {
+			return err
+		}
 		tr.FieldSpecifiers = append(tr.FieldSpecifiers, tf)
 	}
+	return nil
 }
 
 //  0                   1                   2                   3
@@ -380,34 +427,42 @@ func (tr *TemplateRecord) unmarshal(r *reader.Reader) {
 //  |     Option M Field Length     |      Padding (optional)       |
 //  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 
-func (tr *TemplateRecord) unmarshalOpts(r *reader.Reader) {
+func (tr *TemplateRecord) unmarshalOpts(r *reader.Reader) error {
 	var (
 		th = TemplateHeader{}
 		tf = TemplateFieldSpecifier{}
 	)
 
-	th.unmarshalOpts(r)
+	if err := th.unmarshalOpts(r); err != nil {
+		return err
+	}
 	tr.TemplateID = th.TemplateID
 	tr.FieldCount = th.FieldCount
 	tr.ScopeFieldCount = th.ScopeFieldCount
 
 	for i := th.ScopeFieldCount; i > 0; i-- {
-		tf.unmarshal(r)
+		if err := tf.unmarshal(r); err != nil {
+			return err
+		}
 		tr.ScopeFieldSpecifiers = append(tr.FieldSpecifiers, tf)
 	}
 
 	for i := th.FieldCount - th.ScopeFieldCount; i > 0; i-- {
-		tf.unmarshal(r)
+		if err := tf.unmarshal(r); err != nil {
+			return err
+		}
 		tr.FieldSpecifiers = append(tr.FieldSpecifiers, tf)
 	}
+	return nil
 }
 
-func decodeData(r *reader.Reader, tr TemplateRecord) ([]DecodedField, error) {
+func (d *Decoder) decodeData(tr TemplateRecord) ([]DecodedField, error) {
 	var (
 		fields []DecodedField
 		err    error
 		b      []byte
 	)
+	r := d.reader
 
 	for i := 0; i < len(tr.FieldSpecifiers); i++ {
 		b, err = r.Read(int(tr.FieldSpecifiers[i].Length))
@@ -421,8 +476,8 @@ func decodeData(r *reader.Reader, tr TemplateRecord) ([]DecodedField, error) {
 		}]
 
 		if !ok {
-			return nil, fmt.Errorf("IPFIX element key (%d) not exist",
-				tr.FieldSpecifiers[i].ElementID)
+			return nil, nonfatalError(fmt.Errorf("IPFIX element key (%d) not exist",
+				tr.FieldSpecifiers[i].ElementID))
 		}
 
 		fields = append(fields, DecodedField{
@@ -443,8 +498,8 @@ func decodeData(r *reader.Reader, tr TemplateRecord) ([]DecodedField, error) {
 		}]
 
 		if !ok {
-			return nil, fmt.Errorf("IPFIX element key (%d) not exist (scope)",
-				tr.ScopeFieldSpecifiers[i].ElementID)
+			return nil, nonfatalError(fmt.Errorf("IPFIX element key (%d) not exist (scope)",
+				tr.ScopeFieldSpecifiers[i].ElementID))
 		}
 
 		fields = append(fields, DecodedField{
@@ -455,3 +510,19 @@ func decodeData(r *reader.Reader, tr TemplateRecord) ([]DecodedField, error) {
 
 	return fields, nil
 }
+
+func combineErrors(errorSlice ...error) (err error) {
+	switch len(errorSlice) {
+	case 0:
+	case 1:
+		err = errorSlice[0]
+	default:
+		var errMsg bytes.Buffer
+		errMsg.WriteString("Multiple errors:")
+		for _, subError := range errorSlice {
+			errMsg.WriteString("\n- " + subError.Error())
+		}
+		err = errors.New(errMsg.String())
+	}
+	return
+}