Skip to content

feat(netflow): align IPFIX decoder with RFC 7011 #481

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lspgn wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(netflow): align IPFIX decoder with RFC 7011 #481

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
109 changes: 95 additions & 14 deletions decoders/netflow/netflow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,13 +98,42 @@ func DecodeField(payload *bytes.Buffer, field *Field, pen bool) error {
}
if pen && field.Type&0x8000 != 0 {
field.PenProvided = true
field.Type = field.Type ^ 0x8000
return utils.BinaryDecoder(payload,
&field.Pen,
)
}
return nil
}

func isIPFIXOptionsTemplateWithdrawal(raw []byte) ([]uint16, bool) {
if len(raw) == 0 || len(raw)%4 != 0 {
return nil, false
}
templateIds := make([]uint16, 0, len(raw)/4)
for offset := 0; offset < len(raw); offset += 4 {
templateId := binary.BigEndian.Uint16(raw[offset : offset+2])
fieldCount := binary.BigEndian.Uint16(raw[offset+2 : offset+4])
if fieldCount != 0 {
return nil, false
}
templateIds = append(templateIds, templateId)
}
return templateIds, true
}

func templateIdFromKey(key uint64) uint16 {
return uint16(key)
}

func obsDomainFromKey(key uint64) uint32 {
return uint32((key >> 16) & 0xffffffff)
}

func versionFromKey(key uint64) uint16 {
return uint16(key >> 48)
}

// DecodeIPFIXOptionsTemplateSet decodes an IPFIX options template flow set.
func DecodeIPFIXOptionsTemplateSet(payload *bytes.Buffer) ([]IPFIXOptionsTemplateRecord, error) {
var records []IPFIXOptionsTemplateRecord
Expand All @@ -118,6 +147,12 @@ func DecodeIPFIXOptionsTemplateSet(payload *bytes.Buffer) ([]IPFIXOptionsTemplat
if err != nil {
return records, fmt.Errorf("IPFIXOptionsTemplateSet: header [%w]", err)
}
if optsTemplateRecord.TemplateId < 256 {
return records, fmt.Errorf("IPFIXOptionsTemplateSet: template ID below 256")
}
if optsTemplateRecord.ScopeFieldCount == 0 {
return records, fmt.Errorf("IPFIXOptionsTemplateSet: scope field count is zero")
}

fields := make([]Field, int(optsTemplateRecord.ScopeFieldCount)) // max 65532 which would be 589KB
for i := 0; i < int(optsTemplateRecord.ScopeFieldCount); i++ {
Expand Down Expand Up @@ -166,6 +201,9 @@ func DecodeTemplateSet(version uint16, payload *bytes.Buffer) ([]TemplateRecord,
if int(templateRecord.FieldCount) < 0 {
return records, fmt.Errorf("TemplateSet: zero count")
}
if version == 10 && templateRecord.FieldCount > 0 && templateRecord.TemplateId < 256 {
return records, fmt.Errorf("TemplateSet: template ID below 256")
}

fields := make([]Field, int(templateRecord.FieldCount)) // max 65532 which would be 589KB
for i := 0; i < int(templateRecord.FieldCount); i++ {
Expand Down Expand Up @@ -377,28 +415,65 @@ func DecodeMessageCommonFlowSet(payload *bytes.Buffer, templates NetFlowTemplate

if templates != nil {
for _, record := range records {
if record.FieldCount == 0 {
if record.TemplateId == 2 {
for key, tmpl := range templates.GetTemplates() {
if _, ok := tmpl.(TemplateRecord); ok && obsDomainFromKey(key) == obsDomainId && versionFromKey(key) == version {
_, _ = templates.RemoveTemplate(version, obsDomainId, templateIdFromKey(key))
}
}
continue
}
if _, err := templates.RemoveTemplate(version, obsDomainId, record.TemplateId); err != nil {
return flowSet, &FlowError{version, "IPFIX TemplateSet Withdrawal", obsDomainId, fsheader.Id, err}
}
continue
}
if err := templates.AddTemplate(version, obsDomainId, record.TemplateId, record); err != nil {
return flowSet, &FlowError{version, "IPFIX TemplateSet", obsDomainId, fsheader.Id, err}
}
}
}

} else if fsheader.Id == 3 && version == 10 {
templateReader := bytes.NewBuffer(payload.Next(nextrelpos))
records, err := DecodeIPFIXOptionsTemplateSet(templateReader)
if err != nil {
return flowSet, &FlowError{version, "IPFIX OptionsTemplateSet", obsDomainId, fsheader.Id, err}
}
optsTemplatefs := IPFIXOptionsTemplateFlowSet{
FlowSetHeader: fsheader,
Records: records,
}
flowSet = optsTemplatefs
raw := payload.Next(nextrelpos)
if templateIds, ok := isIPFIXOptionsTemplateWithdrawal(raw); ok {
if templates != nil {
for _, templateId := range templateIds {
if templateId == 3 {
for key, tmpl := range templates.GetTemplates() {
if _, ok := tmpl.(IPFIXOptionsTemplateRecord); ok && obsDomainFromKey(key) == obsDomainId && versionFromKey(key) == version {
_, _ = templates.RemoveTemplate(version, obsDomainId, templateIdFromKey(key))
}
}
continue
}
if _, err := templates.RemoveTemplate(version, obsDomainId, templateId); err != nil {
return flowSet, &FlowError{version, "IPFIX OptionsTemplateSet Withdrawal", obsDomainId, fsheader.Id, err}
}
}
}
flowSet = RawFlowSet{
FlowSetHeader: fsheader,
Records: raw,
}
} else {
templateReader := bytes.NewBuffer(raw)
records, err := DecodeIPFIXOptionsTemplateSet(templateReader)
if err != nil {
return flowSet, &FlowError{version, "IPFIX OptionsTemplateSet", obsDomainId, fsheader.Id, err}
}
optsTemplatefs := IPFIXOptionsTemplateFlowSet{
FlowSetHeader: fsheader,
Records: records,
}
flowSet = optsTemplatefs

if templates != nil {
for _, record := range records {
if err := templates.AddTemplate(version, obsDomainId, record.TemplateId, record); err != nil {
return flowSet, &FlowError{version, "IPFIX OptionsTemplateSet", obsDomainId, fsheader.Id, err}
if templates != nil {
for _, record := range records {
if err := templates.AddTemplate(version, obsDomainId, record.TemplateId, record); err != nil {
return flowSet, &FlowError{version, "IPFIX OptionsTemplateSet", obsDomainId, fsheader.Id, err}
}
}
}
}
Expand Down Expand Up @@ -493,6 +568,12 @@ func DecodeMessageIPFIX(payload *bytes.Buffer, templates NetFlowTemplateSystem,
); err != nil {
return &DecoderError{"IPFIX header", err}
}
if packetIPFIX.Length < 16 {
return &DecoderError{"IPFIX header", fmt.Errorf("message length too small")}
}
if payload.Len() < int(packetIPFIX.Length-16) {
return &DecoderError{"IPFIX header", fmt.Errorf("truncated message")}
}
/*size = packetIPFIX.Length
packetIPFIX.Version = version
obsDomainId = packetIPFIX.ObservationDomainId*/
Expand Down
120 changes: 120 additions & 0 deletions decoders/netflow/netflow_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,131 @@ package netflow

import (
"bytes"
"encoding/binary"
"testing"

"github.com/stretchr/testify/assert"
)

func buildIPFIXMessage(obsDomainId uint32, sets []byte) []byte {
var buf bytes.Buffer
_ = binary.Write(&buf, binary.BigEndian, uint16(10))
_ = binary.Write(&buf, binary.BigEndian, uint16(16+len(sets)))
_ = binary.Write(&buf, binary.BigEndian, uint32(0))
_ = binary.Write(&buf, binary.BigEndian, uint32(0))
_ = binary.Write(&buf, binary.BigEndian, obsDomainId)
_, _ = buf.Write(sets)
return buf.Bytes()
}

func buildSet(setId uint16, records []byte) []byte {
var buf bytes.Buffer
_ = binary.Write(&buf, binary.BigEndian, setId)
_ = binary.Write(&buf, binary.BigEndian, uint16(4+len(records)))
_, _ = buf.Write(records)
return buf.Bytes()
}

func buildOptionsTemplateRecord(templateId, fieldCount, scopeFieldCount uint16, fields []byte) []byte {
var buf bytes.Buffer
_ = binary.Write(&buf, binary.BigEndian, templateId)
_ = binary.Write(&buf, binary.BigEndian, fieldCount)
_ = binary.Write(&buf, binary.BigEndian, scopeFieldCount)
_, _ = buf.Write(fields)
return buf.Bytes()
}

func buildEnterpriseFieldSpecifier(ie uint16, length uint16, pen uint32) []byte {
var buf bytes.Buffer
_ = binary.Write(&buf, binary.BigEndian, ie|0x8000)
_ = binary.Write(&buf, binary.BigEndian, length)
_ = binary.Write(&buf, binary.BigEndian, pen)
return buf.Bytes()
}

func buildTemplateWithdrawalRecord(templateId uint16) []byte {
var buf bytes.Buffer
_ = binary.Write(&buf, binary.BigEndian, templateId)
_ = binary.Write(&buf, binary.BigEndian, uint16(0))
return buf.Bytes()
}

func TestDecodeIPFIXOptionsTemplateEnterpriseBitCleared(t *testing.T) {
scopeField := buildEnterpriseFieldSpecifier(1, 4, 9)
optionField := buildEnterpriseFieldSpecifier(2, 4, 9)
record := buildOptionsTemplateRecord(256, 2, 1, append(scopeField, optionField...))
sets := buildSet(3, record)
message := buildIPFIXMessage(123, sets)

templates := CreateTemplateSystem()
var packet IPFIXPacket
err := DecodeMessageVersion(bytes.NewBuffer(message), templates, nil, &packet)
assert.NoError(t, err)
assert.Len(t, packet.FlowSets, 1)

optionsSet, ok := packet.FlowSets[0].(IPFIXOptionsTemplateFlowSet)
assert.True(t, ok)
if !ok || len(optionsSet.Records) == 0 {
t.Fatalf("expected options template records")
}

recordDecoded := optionsSet.Records[0]
if assert.Len(t, recordDecoded.Scopes, 1) {
assert.True(t, recordDecoded.Scopes[0].PenProvided)
assert.Equal(t, uint16(1), recordDecoded.Scopes[0].Type)
assert.Equal(t, uint32(9), recordDecoded.Scopes[0].Pen)
}
if assert.Len(t, recordDecoded.Options, 1) {
assert.True(t, recordDecoded.Options[0].PenProvided)
assert.Equal(t, uint16(2), recordDecoded.Options[0].Type)
assert.Equal(t, uint32(9), recordDecoded.Options[0].Pen)
}
}

func TestDecodeIPFIXOptionsTemplateWithdrawalAll(t *testing.T) {
templates := CreateTemplateSystem()
optionsTemplate := IPFIXOptionsTemplateRecord{
TemplateId: 256,
FieldCount: 2,
ScopeFieldCount: 1,
Scopes: []Field{{Type: 1, Length: 4}},
Options: []Field{{Type: 2, Length: 4}},
}
assert.NoError(t, templates.AddTemplate(10, 42, 256, optionsTemplate))
optionsTemplate.TemplateId = 257
assert.NoError(t, templates.AddTemplate(10, 42, 257, optionsTemplate))
assert.NoError(t, templates.AddTemplate(10, 42, 300, TemplateRecord{TemplateId: 300, FieldCount: 1}))

withdrawalRecord := buildTemplateWithdrawalRecord(3)
message := buildIPFIXMessage(42, buildSet(3, withdrawalRecord))
var packet IPFIXPacket
err := DecodeMessageVersion(bytes.NewBuffer(message), templates, nil, &packet)
assert.NoError(t, err)

_, err = templates.GetTemplate(10, 42, 256)
assert.ErrorIs(t, err, ErrorTemplateNotFound)
_, err = templates.GetTemplate(10, 42, 257)
assert.ErrorIs(t, err, ErrorTemplateNotFound)
_, err = templates.GetTemplate(10, 42, 300)
assert.NoError(t, err)
}

func TestDecodeIPFIXTemplateWithdrawalSingle(t *testing.T) {
templates := CreateTemplateSystem()
assert.NoError(t, templates.AddTemplate(10, 7, 256, TemplateRecord{TemplateId: 256, FieldCount: 1}))

var record bytes.Buffer
_ = binary.Write(&record, binary.BigEndian, uint16(256))
_ = binary.Write(&record, binary.BigEndian, uint16(0))
message := buildIPFIXMessage(7, buildSet(2, record.Bytes()))

var packet IPFIXPacket
err := DecodeMessageVersion(bytes.NewBuffer(message), templates, nil, &packet)
assert.NoError(t, err)
_, err = templates.GetTemplate(10, 7, 256)
assert.ErrorIs(t, err, ErrorTemplateNotFound)
}

func TestDecodeNetFlowV9(t *testing.T) {
templates := CreateTemplateSystem()

Expand Down
Loading