484 lines
13 KiB
Go
484 lines
13 KiB
Go
package sflow
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
"io"
|
|
|
|
"github.com/influxdata/telegraf"
|
|
"github.com/influxdata/telegraf/plugins/inputs/sflow/binaryio"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
type PacketDecoder struct {
|
|
onPacket func(p *V5Format)
|
|
Log telegraf.Logger
|
|
}
|
|
|
|
func NewDecoder() *PacketDecoder {
|
|
return &PacketDecoder{}
|
|
}
|
|
|
|
func (d *PacketDecoder) debug(args ...interface{}) {
|
|
if d.Log != nil {
|
|
d.Log.Debug(args...)
|
|
}
|
|
}
|
|
|
|
func (d *PacketDecoder) OnPacket(f func(p *V5Format)) {
|
|
d.onPacket = f
|
|
}
|
|
|
|
func (d *PacketDecoder) Decode(r io.Reader) error {
|
|
var err error
|
|
var packet *V5Format
|
|
for err == nil {
|
|
packet, err = d.DecodeOnePacket(r)
|
|
if err != nil {
|
|
break
|
|
}
|
|
d.onPacket(packet)
|
|
}
|
|
if err != nil && errors.Cause(err) == io.EOF {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
type AddressType uint32 // must be uint32
|
|
|
|
const (
|
|
AddressTypeUnknown AddressType = 0
|
|
AddressTypeIPV4 AddressType = 1
|
|
AddressTypeIPV6 AddressType = 2
|
|
)
|
|
|
|
func (d *PacketDecoder) DecodeOnePacket(r io.Reader) (*V5Format, error) {
|
|
p := &V5Format{}
|
|
err := read(r, &p.Version, "version")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if p.Version != 5 {
|
|
return nil, fmt.Errorf("Version %d not supported, only version 5", p.Version)
|
|
}
|
|
var addressIPType AddressType
|
|
if err = read(r, &addressIPType, "address ip type"); err != nil {
|
|
return nil, err
|
|
}
|
|
switch addressIPType {
|
|
case AddressTypeUnknown:
|
|
p.AgentAddress.IP = make([]byte, 0)
|
|
case AddressTypeIPV4:
|
|
p.AgentAddress.IP = make([]byte, 4)
|
|
case AddressTypeIPV6:
|
|
p.AgentAddress.IP = make([]byte, 16)
|
|
default:
|
|
return nil, fmt.Errorf("Unknown address IP type %d", addressIPType)
|
|
}
|
|
if err = read(r, &p.AgentAddress.IP, "Agent Address IP"); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = read(r, &p.SubAgentID, "SubAgentID"); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = read(r, &p.SequenceNumber, "SequenceNumber"); err != nil {
|
|
return nil, err
|
|
}
|
|
if err = read(r, &p.Uptime, "Uptime"); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
p.Samples, err = d.decodeSamples(r)
|
|
return p, err
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeSamples(r io.Reader) ([]Sample, error) {
|
|
result := []Sample{}
|
|
// # of samples
|
|
var numOfSamples uint32
|
|
if err := read(r, &numOfSamples, "sample count"); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for i := 0; i < int(numOfSamples); i++ {
|
|
sam, err := d.decodeSample(r)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
result = append(result, sam)
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeSample(r io.Reader) (Sample, error) {
|
|
var err error
|
|
sam := Sample{}
|
|
if err := read(r, &sam.SampleType, "SampleType"); err != nil {
|
|
return sam, err
|
|
}
|
|
sampleDataLen := uint32(0)
|
|
if err := read(r, &sampleDataLen, "Sample data length"); err != nil {
|
|
return sam, err
|
|
}
|
|
mr := binaryio.MinReader(r, int64(sampleDataLen))
|
|
defer mr.Close()
|
|
|
|
switch sam.SampleType {
|
|
case SampleTypeFlowSample:
|
|
sam.SampleData, err = d.decodeFlowSample(mr)
|
|
case SampleTypeFlowSampleExpanded:
|
|
sam.SampleData, err = d.decodeFlowSampleExpanded(mr)
|
|
default:
|
|
d.debug("Unknown sample type: ", sam.SampleType)
|
|
}
|
|
return sam, err
|
|
}
|
|
|
|
type InterfaceFormatType uint8 // sflow_version_5.txt line 1497
|
|
const (
|
|
InterfaceFormatTypeSingleInterface InterfaceFormatType = 0
|
|
InterfaceFormatTypePacketDiscarded InterfaceFormatType = 1
|
|
)
|
|
|
|
func (d *PacketDecoder) decodeFlowSample(r io.Reader) (t SampleDataFlowSampleExpanded, err error) {
|
|
if err := read(r, &t.SequenceNumber, "SequenceNumber"); err != nil {
|
|
return t, err
|
|
}
|
|
var sourceID uint32
|
|
if err := read(r, &sourceID, "SourceID"); err != nil { // source_id sflow_version_5.txt line: 1622
|
|
return t, err
|
|
}
|
|
// split source id to source id type and source id index
|
|
t.SourceIDIndex = sourceID & 0x00ffffff // sflow_version_5.txt line: 1468
|
|
t.SourceIDType = sourceID >> 24 // source_id_type sflow_version_5.txt Line 1465
|
|
if err := read(r, &t.SamplingRate, "SamplingRate"); err != nil {
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.SamplePool, "SamplePool"); err != nil {
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.Drops, "Drops"); err != nil { // sflow_version_5.txt line 1636
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.InputIfIndex, "InputIfIndex"); err != nil {
|
|
return t, err
|
|
}
|
|
t.InputIfFormat = t.InputIfIndex >> 30
|
|
t.InputIfIndex = t.InputIfIndex & 0x3FFFFFFF
|
|
|
|
if err := read(r, &t.OutputIfIndex, "OutputIfIndex"); err != nil {
|
|
return t, err
|
|
}
|
|
t.OutputIfFormat = t.OutputIfIndex >> 30
|
|
t.OutputIfIndex = t.OutputIfIndex & 0x3FFFFFFF
|
|
|
|
switch t.SourceIDIndex {
|
|
case t.OutputIfIndex:
|
|
t.SampleDirection = "egress"
|
|
case t.InputIfIndex:
|
|
t.SampleDirection = "ingress"
|
|
}
|
|
|
|
t.FlowRecords, err = d.decodeFlowRecords(r, t.SamplingRate)
|
|
return t, err
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeFlowSampleExpanded(r io.Reader) (t SampleDataFlowSampleExpanded, err error) {
|
|
if err := read(r, &t.SequenceNumber, "SequenceNumber"); err != nil { // sflow_version_5.txt line 1701
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.SourceIDType, "SourceIDType"); err != nil { // sflow_version_5.txt line: 1706 + 16878
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.SourceIDIndex, "SourceIDIndex"); err != nil { // sflow_version_5.txt line: 1689
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.SamplingRate, "SamplingRate"); err != nil { // sflow_version_5.txt line: 1707
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.SamplePool, "SamplePool"); err != nil { // sflow_version_5.txt line: 1708
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.Drops, "Drops"); err != nil { // sflow_version_5.txt line: 1712
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.InputIfFormat, "InputIfFormat"); err != nil { // sflow_version_5.txt line: 1727
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.InputIfIndex, "InputIfIndex"); err != nil {
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.OutputIfFormat, "OutputIfFormat"); err != nil { // sflow_version_5.txt line: 1728
|
|
return t, err
|
|
}
|
|
if err := read(r, &t.OutputIfIndex, "OutputIfIndex"); err != nil {
|
|
return t, err
|
|
}
|
|
|
|
switch t.SourceIDIndex {
|
|
case t.OutputIfIndex:
|
|
t.SampleDirection = "egress"
|
|
case t.InputIfIndex:
|
|
t.SampleDirection = "ingress"
|
|
}
|
|
|
|
t.FlowRecords, err = d.decodeFlowRecords(r, t.SamplingRate)
|
|
return t, err
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeFlowRecords(r io.Reader, samplingRate uint32) (recs []FlowRecord, err error) {
|
|
var flowDataLen uint32
|
|
var count uint32
|
|
if err := read(r, &count, "FlowRecord count"); err != nil {
|
|
return recs, err
|
|
}
|
|
for i := uint32(0); i < count; i++ {
|
|
fr := FlowRecord{}
|
|
if err := read(r, &fr.FlowFormat, "FlowFormat"); err != nil { // sflow_version_5.txt line 1597
|
|
return recs, err
|
|
}
|
|
if err := read(r, &flowDataLen, "Flow data length"); err != nil {
|
|
return recs, err
|
|
}
|
|
|
|
mr := binaryio.MinReader(r, int64(flowDataLen))
|
|
|
|
switch fr.FlowFormat {
|
|
case FlowFormatTypeRawPacketHeader: // sflow_version_5.txt line 1938
|
|
fr.FlowData, err = d.decodeRawPacketHeaderFlowData(mr, samplingRate)
|
|
default:
|
|
d.debug("Unknown flow format: ", fr.FlowFormat)
|
|
}
|
|
if err != nil {
|
|
mr.Close()
|
|
return recs, err
|
|
}
|
|
|
|
recs = append(recs, fr)
|
|
mr.Close()
|
|
}
|
|
|
|
return recs, err
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeRawPacketHeaderFlowData(r io.Reader, samplingRate uint32) (h RawPacketHeaderFlowData, err error) {
|
|
if err := read(r, &h.HeaderProtocol, "HeaderProtocol"); err != nil { // sflow_version_5.txt line 1940
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.FrameLength, "FrameLength"); err != nil { // sflow_version_5.txt line 1942
|
|
return h, err
|
|
}
|
|
h.Bytes = h.FrameLength * samplingRate
|
|
|
|
if err := read(r, &h.StrippedOctets, "StrippedOctets"); err != nil { // sflow_version_5.txt line 1967
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.HeaderLength, "HeaderLength"); err != nil {
|
|
return h, err
|
|
}
|
|
|
|
mr := binaryio.MinReader(r, int64(h.HeaderLength))
|
|
defer mr.Close()
|
|
|
|
switch h.HeaderProtocol {
|
|
case HeaderProtocolTypeEthernetISO88023:
|
|
h.Header, err = d.decodeEthHeader(mr)
|
|
default:
|
|
d.debug("Unknown header protocol type: ", h.HeaderProtocol)
|
|
}
|
|
|
|
return h, err
|
|
}
|
|
|
|
// ethHeader answers a decode Directive that will decode an ethernet frame header
|
|
// according to https://en.wikipedia.org/wiki/Ethernet_frame
|
|
func (d *PacketDecoder) decodeEthHeader(r io.Reader) (h EthHeader, err error) {
|
|
// we may have to read out StrippedOctets bytes and throw them away first?
|
|
if err := read(r, &h.DestinationMAC, "DestinationMAC"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.SourceMAC, "SourceMAC"); err != nil {
|
|
return h, err
|
|
}
|
|
var tagOrEType uint16
|
|
if err := read(r, &tagOrEType, "tagOrEtype"); err != nil {
|
|
return h, err
|
|
}
|
|
switch tagOrEType {
|
|
case 0x8100: // could be?
|
|
var discard uint16
|
|
if err := read(r, &discard, "unknown"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.EtherTypeCode, "EtherTypeCode"); err != nil {
|
|
return h, err
|
|
}
|
|
default:
|
|
h.EtherTypeCode = tagOrEType
|
|
}
|
|
h.EtherType = ETypeMap[h.EtherTypeCode]
|
|
switch h.EtherType {
|
|
case "IPv4":
|
|
h.IPHeader, err = d.decodeIPv4Header(r)
|
|
case "IPv6":
|
|
h.IPHeader, err = d.decodeIPv6Header(r)
|
|
default:
|
|
}
|
|
if err != nil {
|
|
return h, err
|
|
}
|
|
return h, err
|
|
}
|
|
|
|
// https://en.wikipedia.org/wiki/IPv4#Header
|
|
func (d *PacketDecoder) decodeIPv4Header(r io.Reader) (h IPV4Header, err error) {
|
|
if err := read(r, &h.Version, "Version"); err != nil {
|
|
return h, err
|
|
}
|
|
h.InternetHeaderLength = h.Version & 0x0F
|
|
h.Version = h.Version & 0xF0
|
|
if err := read(r, &h.DSCP, "DSCP"); err != nil {
|
|
return h, err
|
|
}
|
|
h.ECN = h.DSCP & 0x03
|
|
h.DSCP = h.DSCP >> 2
|
|
if err := read(r, &h.TotalLength, "TotalLength"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.Identification, "Identification"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.FragmentOffset, "FragmentOffset"); err != nil {
|
|
return h, err
|
|
}
|
|
h.Flags = uint8(h.FragmentOffset >> 13)
|
|
h.FragmentOffset = h.FragmentOffset & 0x1FFF
|
|
if err := read(r, &h.TTL, "TTL"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.Protocol, "Protocol"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.HeaderChecksum, "HeaderChecksum"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.SourceIP, "SourceIP"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.DestIP, "DestIP"); err != nil {
|
|
return h, err
|
|
}
|
|
switch h.Protocol {
|
|
case IPProtocolTCP:
|
|
h.ProtocolHeader, err = d.decodeTCPHeader(r)
|
|
case IPProtocolUDP:
|
|
h.ProtocolHeader, err = d.decodeUDPHeader(r)
|
|
default:
|
|
d.debug("Unknown IP protocol: ", h.Protocol)
|
|
}
|
|
return h, err
|
|
}
|
|
|
|
// https://en.wikipedia.org/wiki/IPv6_packet
|
|
func (d *PacketDecoder) decodeIPv6Header(r io.Reader) (h IPV6Header, err error) {
|
|
var fourByteBlock uint32
|
|
if err := read(r, &fourByteBlock, "IPv6 header octet 0"); err != nil {
|
|
return h, err
|
|
}
|
|
version := fourByteBlock >> 28
|
|
if version != 0x6 {
|
|
return h, fmt.Errorf("Unexpected IPv6 header version 0x%x", version)
|
|
}
|
|
h.DSCP = uint8((fourByteBlock & 0xFC00000) >> 22)
|
|
h.ECN = uint8((fourByteBlock & 0x300000) >> 20)
|
|
|
|
// flowLabel := fourByteBlock & 0xFFFFF // not currently being used.
|
|
if err := read(r, &h.PayloadLength, "PayloadLength"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.NextHeaderProto, "NextHeaderProto"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.HopLimit, "HopLimit"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.SourceIP, "SourceIP"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.DestIP, "DestIP"); err != nil {
|
|
return h, err
|
|
}
|
|
switch h.NextHeaderProto {
|
|
case IPProtocolTCP:
|
|
h.ProtocolHeader, err = d.decodeTCPHeader(r)
|
|
case IPProtocolUDP:
|
|
h.ProtocolHeader, err = d.decodeUDPHeader(r)
|
|
default:
|
|
// not handled
|
|
d.debug("Unknown IP protocol: ", h.NextHeaderProto)
|
|
}
|
|
return h, err
|
|
}
|
|
|
|
// https://en.wikipedia.org/wiki/Transmission_Control_Protocol#TCP_segment_structure
|
|
func (d *PacketDecoder) decodeTCPHeader(r io.Reader) (h TCPHeader, err error) {
|
|
if err := read(r, &h.SourcePort, "SourcePort"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.DestinationPort, "DestinationPort"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.Sequence, "Sequence"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.AckNumber, "AckNumber"); err != nil {
|
|
return h, err
|
|
}
|
|
// Next up: bit reading!
|
|
// data offset 4 bits
|
|
// reserved 3 bits
|
|
// flags 9 bits
|
|
var dataOffsetAndReservedAndFlags uint16
|
|
if err := read(r, &dataOffsetAndReservedAndFlags, "TCP Header Octet offset 12"); err != nil {
|
|
return h, err
|
|
}
|
|
h.TCPHeaderLength = uint8((dataOffsetAndReservedAndFlags >> 12) * 4)
|
|
h.Flags = dataOffsetAndReservedAndFlags & 0x1FF
|
|
// done bit reading
|
|
|
|
if err := read(r, &h.TCPWindowSize, "TCPWindowSize"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.Checksum, "Checksum"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.TCPUrgentPointer, "TCPUrgentPointer"); err != nil {
|
|
return h, err
|
|
}
|
|
|
|
return h, err
|
|
}
|
|
|
|
func (d *PacketDecoder) decodeUDPHeader(r io.Reader) (h UDPHeader, err error) {
|
|
if err := read(r, &h.SourcePort, "SourcePort"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.DestinationPort, "DestinationPort"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.UDPLength, "UDPLength"); err != nil {
|
|
return h, err
|
|
}
|
|
if err := read(r, &h.Checksum, "Checksum"); err != nil {
|
|
return h, err
|
|
}
|
|
return h, err
|
|
}
|
|
|
|
func read(r io.Reader, data interface{}, name string) error {
|
|
err := binary.Read(r, binary.BigEndian, data)
|
|
return errors.Wrapf(err, "failed to read %s", name)
|
|
}
|