event.go

package beater

import (
	"net"
	"strconv"
	"time"

	"github.com/elastic/beats/libbeat/beat"
	"github.com/elastic/beats/libbeat/common"
)

type Locality uint8

const (
	LocalityPrivate Locality = iota + 1
	LocalityPublic
)

var localityNames = map[Locality]string{
	LocalityPrivate: "private",
	LocalityPublic:  "public",
}

func (l Locality) String() string {
	name, found := localityNames[l]
	if found {
		return name
	}
	return "unknown (" + strconv.Itoa(int(l)) + ")"
}

type FlowEvent struct {
	TimeReceived time.Time // Time packet was read off socket.
	TimeCreated  time.Time // Time record was generated by "router".
	SequenceNum  uint32    // Message sequence number.
	Type         FlowType  // Flow type (e.g. Netflow v9, IPFIX)

	// Flow exporter's address.
	DeviceAddr net.IP

	// Flow times.
	StartTime time.Time
	LastTime  time.Time
	Duration  time.Duration

	// Flow volume.
	Bytes   uint64
	Packets uint64

	Direction *FlowDirection

	// Layer 3 info.
	SrcMAC net.HardwareAddr
	DstMAC net.HardwareAddr

	// Layer 4 Info.

	// IP version. https://www.iana.org/assignments/version-numbers/version-numbers.xhtml
	IPVersion        uint8 // 4 = ipv4, 6=ipv6
	IPClassOfTraffic uint8 // TOS for IPv4, Traffic Class for IPv6.
	SrcIP            net.IP
	DstIP            net.IP
	SrcLocality      Locality
	DstLocality      Locality
	SrcPort          uint16
	DstPort          uint16
	Locality         Locality
	TCPFlags         TCPFlag

	// Layer 4 protocol (transport). Comes from the IPv4 and IPv6 headers.
	// Netflow fields: protocolIdentifier (IPv4) and nextHeaderIPv6.
	TransportProtocol IPProtocol

	// Stable 5-tuple that's the same for both flow dirs.
	FiveTuple     string
	FiveTupleHash string

	// VLANs
	IngressVLAN uint16
	EgressVLAN  uint16

	// Raw Netflow fields.
	Netflow map[string]interface{}
}

func (f *FlowEvent) toBeatEvent() beat.Event {
	b := beat.Event{
		Timestamp: f.TimeReceived.UTC(),
		Fields: common.MapStr{
			"event": common.MapStr{
				"type":     f.Type.String(),
				"created":  f.TimeCreated.UTC(),
				"start":    f.StartTime.UTC(),
				"end":      f.LastTime.UTC(),
				"duration": f.Duration,
			},
			"device": common.MapStr{
				"ip": f.DeviceAddr,
			},
			"flow": common.MapStr{
				"sequence_num": f.SequenceNum,
				"five_tuple":   f.FiveTuple,
				"id":           f.FiveTupleHash,
				"locality":     f.Locality.String(),
			},
			"source": common.MapStr{
				"ip":       f.SrcIP.String(),
				"locality": f.SrcLocality.String(),
			},
			"destination": common.MapStr{
				"ip":       f.DstIP.String(),
				"locality": f.DstLocality.String(),
			},
			"network": common.MapStr{
				"protocol": f.TransportProtocol.String(),
				"total": common.MapStr{
					"bytes":   f.Bytes,
					"packets": f.Packets,
				},
			},
			"netflow": f.Netflow,
		},
	}

	if len(f.SrcMAC) > 0 {
		b.PutValue("source.mac", f.SrcMAC.String())
	}
	if len(f.DstMAC) > 0 {
		b.PutValue("destination.mac", f.DstMAC.String())
	}

	if f.SrcPort > 0 {
		b.PutValue("source.port", f.SrcPort)
	}
	if f.DstPort > 0 {
		b.PutValue("destination.port", f.DstPort)
	}

	if f.Direction != nil {
		switch *f.Direction {
		case Ingress:
			b.PutValue("network.direction", "inbound")
		case Egress:
			b.PutValue("network.direction", "outbound")
		}
	}

	if f.TCPFlags > 0 {
		b.PutValue("network.tcp.flags", f.TCPFlags.Strings())
	}

	if f.IngressVLAN > 0 {
		b.PutValue("source.vlan", f.IngressVLAN)
	}
	if f.EgressVLAN > 0 {
		b.PutValue("destination.vlan", f.EgressVLAN)
	}
	return b
}