[Packetbeat] ECS 1.5 update

packetbeat/pb/ecs.go

@@ -0,0 +1,58 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package pb
+
+import "time"
+
+type ecsEvent struct {
+	ID   string `ecs:"id"`
+	Code string `ecs:"code"`
+	Kind string `ecs:"kind"`
+	// overridden because this needs to be an array
+	Category []string `ecs:"category"`
+	Action   string   `ecs:"action"`
+	Outcome  string   `ecs:"outcome"`
+	// overridden because this needs to be an array
+	Type          []string      `ecs:"type"`
+	Module        string        `ecs:"module"`
+	Dataset       string        `ecs:"dataset"`
+	Provider      string        `ecs:"provider"`
+	Severity      int64         `ecs:"severity"`
+	Original      string        `ecs:"original"`
+	Hash          string        `ecs:"hash"`
+	Duration      time.Duration `ecs:"duration"`
+	Sequence      int64         `ecs:"sequence"`
+	Timezone      string        `ecs:"timezone"`
+	Created       time.Time     `ecs:"created"`
+	Start         time.Time     `ecs:"start"`
+	End           time.Time     `ecs:"end"`
+	RiskScore     float64       `ecs:"risk_score"`
+	RiskScoreNorm float64       `ecs:"risk_score_norm"`
+	Ingested      time.Time     `ecs:"ingested"`
+	Reference     string        `ecs:"reference"`
+	Url           string        `ecs:"url"`
+}
+
+type ecsRelated struct {
+	IP   []string `ecs:"ip"`
+	User []string `ecs:"user"`
+	Hash []string `ecs:"hash"`
+
+	// for de-dup
+	ipSet map[string]struct{}
+}

packetbeat/pb/event.go

@@ -54,8 +54,9 @@ type Fields struct {
 	Destination *ecs.Destination `ecs:"destination"`
 	Client      *ecs.Client      `ecs:"client"`
 	Server      *ecs.Server      `ecs:"server"`
+	Related     *ecsRelated      `ecs:"related"`
 	Network     ecs.Network      `ecs:"network"`
-	Event       ecs.Event        `ecs:"event"`
+	Event       ecsEvent         `ecs:"event"`
 
 	SourceProcess      *ecs.Process `ecs:"source.process"`
 	DestinationProcess *ecs.Process `ecs:"destination.process"`
@@ -72,10 +73,11 @@ type Fields struct {
 // NewFields returns a new Fields value.
 func NewFields() *Fields {
 	return &Fields{
-		Event: ecs.Event{
+		Event: ecsEvent{
 			Duration: -1,
 			Kind:     "event",
-			Category: "network_traffic",
+			Type:     []string{"connection", "protocol"},
+			Category: []string{"network_traffic", "network"},
 		},
 	}
 }
@@ -112,6 +114,7 @@ func (f *Fields) SetSource(endpoint *common.Endpoint) {
 	if f.Source == nil {
 		f.Source = &ecs.Source{}
 	}
+	f.AddIP(endpoint.IP)
 	f.Source.IP = endpoint.IP
 	f.Source.Port = int64(endpoint.Port)
 	f.Source.Domain = endpoint.Domain
@@ -126,6 +129,7 @@ func (f *Fields) SetDestination(endpoint *common.Endpoint) {
 	if f.Destination == nil {
 		f.Destination = &ecs.Destination{}
 	}
+	f.AddIP(endpoint.IP)
 	f.Destination.IP = endpoint.IP
 	f.Destination.Port = int64(endpoint.Port)
 	f.Destination.Domain = endpoint.Domain
@@ -135,6 +139,21 @@ func (f *Fields) SetDestination(endpoint *common.Endpoint) {
 	}
 }
 
+// AddIP adds the given ip addresses to the related ECS IP field
+func (f *Fields) AddIP(ip ...string) {
+	if f.Related == nil {
+		f.Related = &ecsRelated{
+			ipSet: make(map[string]struct{}),
+		}
+	}
+	for _, ipAddress := range ip {
+		if _, ok := f.Related.ipSet[ipAddress]; !ok {
+			f.Related.ipSet[ipAddress] = struct{}{}
+			f.Related.IP = append(f.Related.IP, ipAddress)
+		}
+	}
+}
+
 func makeProcess(p *common.Process) *ecs.Process {
 	return &ecs.Process{
 		Name:             p.Name,

packetbeat/pb/event_test.go

@@ -41,7 +41,8 @@ func TestMarshalMapStr(t *testing.T) {
 	assert.Equal(t, common.MapStr{
 		"event": common.MapStr{
 			"kind":     "event",
-			"category": "network_traffic",
+			"category": []string{"network_traffic", "network"},
+			"type":     []string{"connection", "protocol"},
 		},
 		"source": common.MapStr{"ip": "127.0.0.1"},
 	}, m)

packetbeat/protos/amqp/amqp.go

@@ -435,6 +435,7 @@ func (amqp *amqpPlugin) publishTransaction(t *amqpTransaction) {
 	pbf.Event.Start = t.ts
 	pbf.Event.End = t.endTime
 	pbf.Event.Dataset = "amqp"
+	pbf.Event.Action = "amqp." + t.method
 	pbf.Network.Protocol = pbf.Event.Dataset
 	pbf.Network.Transport = "tcp"
 	pbf.Error.Message = t.notes

packetbeat/protos/dhcpv4/dhcpv4.go

@@ -133,12 +133,15 @@ func (p *dhcpv4Plugin) parseDHCPv4(pkt *protos.Packet) *beat.Event {
 
 	if !v4.ClientIPAddr().IsUnspecified() {
 		dhcpData.Put("client_ip", v4.ClientIPAddr().String())
+		pbf.AddIP(v4.ClientIPAddr().String())
 	}
 	if !v4.YourIPAddr().IsUnspecified() {
 		dhcpData.Put("assigned_ip", v4.YourIPAddr().String())
+		pbf.AddIP(v4.YourIPAddr().String())
 	}
 	if !v4.GatewayIPAddr().IsUnspecified() {
 		dhcpData.Put("relay_ip", v4.GatewayIPAddr().String())
+		pbf.AddIP(v4.GatewayIPAddr().String())
 	}
 	if serverName := v4.ServerHostNameToString(); serverName != "" {
 		dhcpData.Put("server_name", serverName)

packetbeat/protos/dhcpv4/dhcpv4_test.go

@@ -117,7 +117,8 @@ func TestParseDHCPRequest(t *testing.T) {
 				"port": 67,
 			},
 			"event": common.MapStr{
-				"category": "network_traffic",
+				"category": []string{"network_traffic", "network"},
+				"type":     []string{"connection", "protocol"},
 				"dataset":  "dhcpv4",
 				"kind":     "event",
 				"start":    pkt.Ts,
@@ -129,6 +130,9 @@ func TestParseDHCPRequest(t *testing.T) {
 				"bytes":        272,
 				"community_id": "1:t9O1j0qj71O4wJM7gnaHtgmfev8=",
 			},
+			"related": common.MapStr{
+				"ip": []string{"0.0.0.0", "255.255.255.255"},
+			},
 			"dhcpv4": common.MapStr{
 				"client_mac":     "00:0b:82:01:fc:42",
 				"flags":          "unicast",
@@ -197,7 +201,8 @@ func TestParseDHCPACK(t *testing.T) {
 				"bytes": 300,
 			},
 			"event": common.MapStr{
-				"category": "network_traffic",
+				"category": []string{"network_traffic", "network"},
+				"type":     []string{"connection", "protocol"},
 				"dataset":  "dhcpv4",
 				"kind":     "event",
 				"start":    pkt.Ts,
@@ -209,7 +214,9 @@ func TestParseDHCPACK(t *testing.T) {
 				"bytes":        300,
 				"community_id": "1:VbRSZnvQqvLiQRhYHLrdVI17sLQ=",
 			},
-
+			"related": common.MapStr{
+				"ip": []string{"192.168.0.1", "192.168.0.10"},
+			},
 			"dhcpv4": common.MapStr{
 				"assigned_ip":    "192.168.0.10",
 				"client_mac":     "00:0b:82:01:fc:42",

packetbeat/protos/dns/dns.go

@@ -391,7 +391,7 @@ func (dns *dnsPlugin) publishTransaction(t *dnsTransaction) {
 			fields["query"] = dnsQuestionToString(t.request.data.Question[0])
 			fields["resource"] = t.request.data.Question[0].Name
 		}
-		addDNSToMapStr(dnsEvent, t.response.data, dns.includeAuthorities,
+		addDNSToMapStr(dnsEvent, pbf, t.response.data, dns.includeAuthorities,
 			dns.includeAdditionals)
 
 		if t.response.data.Rcode == 0 {
@@ -414,7 +414,7 @@ func (dns *dnsPlugin) publishTransaction(t *dnsTransaction) {
 			fields["query"] = dnsQuestionToString(t.request.data.Question[0])
 			fields["resource"] = t.request.data.Question[0].Name
 		}
-		addDNSToMapStr(dnsEvent, t.request.data, dns.includeAuthorities,
+		addDNSToMapStr(dnsEvent, pbf, t.request.data, dns.includeAuthorities,
 			dns.includeAdditionals)
 
 		if dns.sendRequest {
@@ -430,7 +430,7 @@ func (dns *dnsPlugin) publishTransaction(t *dnsTransaction) {
 			fields["query"] = dnsQuestionToString(t.response.data.Question[0])
 			fields["resource"] = t.response.data.Question[0].Name
 		}
-		addDNSToMapStr(dnsEvent, t.response.data, dns.includeAuthorities,
+		addDNSToMapStr(dnsEvent, pbf, t.response.data, dns.includeAuthorities,
 			dns.includeAdditionals)
 		if dns.sendResponse {
 			fields["response"] = dnsToString(t.response.data)
@@ -448,7 +448,7 @@ func (dns *dnsPlugin) expireTransaction(t *dnsTransaction) {
 }
 
 // Adds the DNS message data to the supplied MapStr.
-func addDNSToMapStr(m common.MapStr, dns *mkdns.Msg, authority bool, additional bool) {
+func addDNSToMapStr(m common.MapStr, pbf *pb.Fields, dns *mkdns.Msg, authority bool, additional bool) {
 	m["id"] = dns.Id
 	m["op_code"] = dnsOpCodeToString(dns.Opcode)
 
@@ -533,6 +533,7 @@ func addDNSToMapStr(m common.MapStr, dns *mkdns.Msg, authority bool, additional
 		m["answers"], resolvedIPs = rrsToMapStrs(dns.Answer, true)
 		if len(resolvedIPs) > 0 {
 			m["resolved_ip"] = resolvedIPs
+			pbf.AddIP(resolvedIPs...)
 		}
 	}
 

packetbeat/protos/dns/names_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/packetbeat/pb"
 )
 
 type dnsTestMsg struct {
@@ -110,7 +111,7 @@ func assertDNSMessage(t testing.TB, q dnsTestMsg) {
 	}
 
 	mapStr := common.MapStr{}
-	addDNSToMapStr(mapStr, dns, true, true)
+	addDNSToMapStr(mapStr, pb.NewFields(), dns, true, true)
 	if q.question != nil {
 		for k, v := range q.question {
 			assert.NotNil(t, mapStr["question"].(common.MapStr)[k])

packetbeat/protos/icmp/icmp.go

@@ -287,7 +287,9 @@ func (icmp *icmpPlugin) publishTransaction(trans *icmpTransaction) {
 	evt, pbf := pb.NewBeatEvent(trans.ts)
 	pbf.Source = &ecs.Source{IP: trans.tuple.srcIP.String()}
 	pbf.Destination = &ecs.Destination{IP: trans.tuple.dstIP.String()}
+	pbf.AddIP(trans.tuple.srcIP.String(), trans.tuple.dstIP.String())
 	pbf.Event.Dataset = "icmp"
+	pbf.Event.Type = []string{"connection"}
 	pbf.Error.Message = trans.notes
 
 	// common fields - group "event"

packetbeat/protos/memcache/memcache.go

@@ -22,6 +22,7 @@ package memcache
 import (
 	"encoding/json"
 	"math"
+	"strings"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -388,25 +389,31 @@ func (t *transaction) Event(event *beat.Event) error {
 	mc := common.MapStr{}
 	event.Fields["memcache"] = mc
 
+	msg := t.request
+	if msg == nil {
+		msg = t.response
+	}
+
 	if t.request != nil {
 		_, err := t.request.SubEvent("request", mc)
 		if err != nil {
 			logp.Warn("error filling transaction request: %v", err)
 			return err
 		}
+		event.Fields["event.action"] = "memcache." + strings.ToLower(t.request.command.typ.String())
 	}
 	if t.response != nil {
 		_, err := t.response.SubEvent("response", mc)
 		if err != nil {
 			logp.Warn("error filling transaction response: %v", err)
 			return err
 		}
+		normalized := normalizeEventOutcome(memcacheStatusCode(t.response.status).String())
+		if normalized != "" {
+			event.Fields["event.outcome"] = normalized
+		}
 	}
 
-	msg := t.request
-	if msg == nil {
-		msg = t.response
-	}
 	if msg == nil {
 		mc["protocol_type"] = "unknown"
 	} else {
@@ -420,6 +427,19 @@ func (t *transaction) Event(event *beat.Event) error {
 	return nil
 }
 
+func normalizeEventOutcome(outcome string) string {
+	switch outcome {
+	case "Fail":
+		return "failure"
+	case "UNKNOWN":
+		return "unknown"
+	case "Success":
+		return "success"
+	default:
+		return ""
+	}
+}
+
 func computeTransactionStatus(requ, resp *message) string {
 	switch {
 	case requ == nil && resp != nil:

packetbeat/protos/nfs/request_handler.go

@@ -143,7 +143,12 @@ func (r *rpc) handleCall(xid string, xdr *xdr, ts time.Time, tcptuple *common.TC
 		pbf:   pbf,
 		event: evt,
 	}
-	fields["nfs"] = nfs.getRequestInfo(xdr)
+	info := nfs.getRequestInfo(xdr)
+	fields["nfs"] = info
+
+	if opcode, ok := info["opcode"].(string); ok && opcode != "" {
+		pbf.Event.Action = "nfs." + opcode
+	}
 
 	// use xid+src ip to uniquely identify request
 	reqID := xid + tcptuple.SrcIP.String()
@@ -190,7 +195,8 @@ func (r *rpc) handleReply(xid string, xdr *xdr, ts time.Time, tcptuple *common.T
 		if status == 0 {
 			nfsInfo := fields["nfs"].(common.MapStr)
 			nfsInfo["status"] = nfs.getNFSReplyStatus(xdr)
+		} else {
+			nfs.pbf.Event.Outcome = "failure"
 		}
-		r.results(nfs.event)
 	}
 }

packetbeat/protos/redis/redis.go

@@ -19,6 +19,7 @@ package redis
 
 import (
 	"bytes"
+	"strings"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -326,6 +327,11 @@ func (redis *redisPlugin) newTransaction(requ, resp *redisMessage) beat.Event {
 		fields["response"] = resp.message
 	}
 
+	pbf.Event.Action = "redis." + strings.ToLower(fields["method"].(string))
+	if resp.isError {
+		pbf.Event.Outcome = "failure"
+	}
+
 	return evt
 }