Update ICMP protocol to use ECS fields (#10062)

* Update ICMP protocol to use ECS fields

Only a few fields were changed. No dashboards were changed because there are no ICMP specific dashboards.

Here's a summary of what fields changed.

Part of #7968

Changed

- responsetime -> event.duration (unit are now nanoseconds)
- bytes_in -> source.bytes
- bytes_out -> destination.bytes

Added

- event.dataset = icmp
- event.end
- event.start
- network.community_id
- network.transport = icmp or ipv6-icmp
- network.type = ipv4/ipv6

Unchanged Packetbeat Fields

- status
- type = icmp (we might remove this since we have event.dataset)
- path = destination.ip (what is requested, not sure if this still makes sense)

* Change NNNN to 10062

CHANGELOG.next.asciidoc

@@ -58,6 +58,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Changed DNS protocol fields to align with ECS. {pull}9941[9941]
 - Removed trailing dot from domain names reported by the DNS protocol. {pull}9941[9941]
 - Changed TLS protocol fields to align with ECS. {pull}9980[9980]
+- Changed ICMP protocol fields to align with ECS. {pull}10062[10062]
 
 *Winlogbeat*
 

packetbeat/pb/event.go

@@ -53,6 +53,9 @@ type Fields struct {
 	SourceProcess      *ecs.Process `ecs:"source.process"`
 	DestinationProcess *ecs.Process `ecs:"destination.process"`
 	Process            *ecs.Process `ecs:"process"`
+
+	ICMPType uint8 // ICMP message type for use in computing network.community_id.
+	ICMPCode uint8 // ICMP message code for use in computing network.community_id.
 }
 
 // NewFields returns a new Fields value.
@@ -151,11 +154,11 @@ func (f *Fields) ComputeValues(localIPs []net.IP) error {
 		flow.Protocol = 6
 	case f.Network.Transport == "icmp":
 		flow.Protocol = 1
-		// TODO: Populate the ICMP type/code.
 	case f.Network.Transport == "ipv6-icmp":
 		flow.Protocol = 58
-		// TODO: Populate the ICMP type/code.
 	}
+	flow.ICMP.Type = f.ICMPType
+	flow.ICMP.Code = f.ICMPCode
 	if flow.Protocol > 0 && len(flow.SourceIP) > 0 && len(flow.DestinationIP) > 0 {
 		f.Network.CommunityID = flowhash.CommunityID.Hash(flow)
 	}
@@ -241,7 +244,7 @@ func (f *Fields) MarshalMapStr(m common.MapStr) error {
 		structField := typ.Field(i)
 		tag := structField.Tag.Get("ecs")
 		if tag == "" {
-			panic(errors.Errorf("missing tag on field %v", structField.Name))
+			continue
 		}
 
 		fieldValue := val.Field(i)

packetbeat/protos/icmp/icmp.go

@@ -21,12 +21,13 @@ import (
 	"net"
 	"time"
 
-	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/monitoring"
+	"github.com/elastic/ecs/code/go/ecs"
 
 	"github.com/elastic/beats/packetbeat/flows"
+	"github.com/elastic/beats/packetbeat/pb"
 	"github.com/elastic/beats/packetbeat/protos"
 
 	"github.com/tsg/gopacket/layers"
@@ -283,86 +284,68 @@ func (icmp *icmpPlugin) publishTransaction(trans *icmpTransaction) {
 
 	logp.Debug("icmp", "Publishing transaction. %s", &trans.tuple)
 
-	// common fields - group "env"
-	fields := common.MapStr{
-		"client": common.MapStr{
-			"ip": trans.tuple.srcIP,
-		},
-		"server": common.MapStr{
-			"ip": trans.tuple.dstIP,
-		},
-	}
+	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.Event.Dataset = "icmp"
 
 	// common fields - group "event"
-	fields["type"] = "icmp"            // protocol name
+	fields := evt.Fields
+	fields["type"] = pbf.Event.Dataset
 	fields["path"] = trans.tuple.dstIP // what is requested (dst ip)
 	if trans.HasError() {
 		fields["status"] = common.ERROR_STATUS
 	} else {
 		fields["status"] = common.OK_STATUS
 	}
-	if len(trans.notes) > 0 {
-		fields["notes"] = trans.notes
-	}
 
-	// common fields - group "measurements"
-	responsetime, hasResponseTime := trans.ResponseTimeMillis()
-	if hasResponseTime {
-		fields["responsetime"] = responsetime
-	}
-	switch icmp.direction(trans) {
-	case directionFromInside:
-		if trans.request != nil {
-			fields["bytes_out"] = trans.request.length
-		}
-		if trans.response != nil {
-			fields["bytes_in"] = trans.response.length
-		}
-	case directionFromOutside:
-		if trans.request != nil {
-			fields["bytes_in"] = trans.request.length
-		}
-		if trans.response != nil {
-			fields["bytes_out"] = trans.response.length
-		}
+	icmpEvent := common.MapStr{
+		"version": trans.tuple.icmpVersion,
 	}
-
-	// event fields - group "icmp"
-	icmpEvent := common.MapStr{}
 	fields["icmp"] = icmpEvent
 
-	icmpEvent["version"] = trans.tuple.icmpVersion
+	pbf.Network.Transport = pbf.Event.Dataset
+	if trans.tuple.icmpVersion == 6 {
+		pbf.Network.Transport = "ipv6-icmp"
+	}
 
 	if trans.request != nil {
-		request := common.MapStr{}
-		icmpEvent["request"] = request
+		pbf.Event.Start = trans.request.ts
+		pbf.Source.Bytes = int64(trans.request.length)
 
-		request["message"] = humanReadable(&trans.tuple, trans.request)
-		request["type"] = trans.request.Type
-		request["code"] = trans.request.code
+		request := common.MapStr{
+			"message": humanReadable(&trans.tuple, trans.request),
+			"type":    trans.request.Type,
+			"code":    trans.request.code,
+		}
+		icmpEvent["request"] = request
 
-		// TODO: Add more info. The IPv4/IPv6 payload could be interesting.
-		// if icmp.SendRequest {
-		//     request["payload"] = ""
-		// }
+		pbf.ICMPType = trans.request.Type
+		pbf.ICMPCode = trans.request.code
 	}
 
 	if trans.response != nil {
-		response := common.MapStr{}
+		pbf.Event.End = trans.response.ts
+		pbf.Destination.Bytes = int64(trans.response.length)
+
+		response := common.MapStr{
+			"message": humanReadable(&trans.tuple, trans.response),
+			"type":    trans.response.Type,
+			"code":    trans.response.code,
+		}
 		icmpEvent["response"] = response
 
-		response["message"] = humanReadable(&trans.tuple, trans.response)
-		response["type"] = trans.response.Type
-		response["code"] = trans.response.code
+		if trans.request == nil {
+			pbf.ICMPType = trans.response.Type
+			pbf.ICMPCode = trans.response.code
+		}
+	}
 
-		// TODO: Add more info. The IPv4/IPv6 payload could be interesting.
-		// if icmp.SendResponse {
-		//     response["payload"] = ""
-		// }
+	if len(trans.notes) == 1 {
+		evt.PutValue("error.message", trans.notes[0])
+	} else if len(trans.notes) > 1 {
+		evt.PutValue("error.message", trans.notes)
 	}
 
-	icmp.results(beat.Event{
-		Timestamp: trans.ts,
-		Fields:    fields,
-	})
+	icmp.results(evt)
 }

packetbeat/protos/icmp/transaction.go

@@ -34,10 +34,3 @@ func (t *icmpTransaction) HasError() bool {
 		(t.response != nil && isError(&t.tuple, t.response)) ||
 		(t.request != nil && t.response == nil && requiresCounterpart(&t.tuple, t.request))
 }
-
-func (t *icmpTransaction) ResponseTimeMillis() (int32, bool) {
-	if t.request != nil && t.response != nil {
-		return int32(t.response.ts.Sub(t.request.ts).Nanoseconds() / 1e6), true
-	}
-	return 0, false
-}

packetbeat/protos/icmp/transaction_test.go

@@ -21,7 +21,6 @@ package icmp
 
 import (
 	"testing"
-	"time"
 
 	"github.com/tsg/gopacket/layers"
 
@@ -65,23 +64,3 @@ func TestIcmpTransactionHasErrorICMPv6(t *testing.T) {
 	trans5 := icmpTransaction{tuple: tuple, request: &icmpMessage{Type: layers.ICMPv6TypeRedirect}, response: nil}
 	assert.False(t, trans5.HasError(), "non-transactional request without response")
 }
-
-func TestIcmpTransactionResponseTimeMillis(t *testing.T) {
-	reqTime := time.Now()
-	resTime := reqTime.Add(time.Duration(1) * time.Second)
-
-	trans1 := icmpTransaction{request: &icmpMessage{ts: reqTime}, response: &icmpMessage{ts: resTime}}
-	time1, hasTime1 := trans1.ResponseTimeMillis()
-	assert.Equal(t, int32(1000), time1, "request with response")
-	assert.True(t, hasTime1, "request with response")
-
-	trans2 := icmpTransaction{request: &icmpMessage{ts: reqTime}}
-	time2, hasTime2 := trans2.ResponseTimeMillis()
-	assert.Equal(t, int32(0), time2, "request without response")
-	assert.False(t, hasTime2, "request without response")
-
-	trans3 := icmpTransaction{response: &icmpMessage{ts: resTime}}
-	time3, hasTime3 := trans3.ResponseTimeMillis()
-	assert.Equal(t, int32(0), time3, "response without request")
-	assert.False(t, hasTime3, "response without request")
-}

packetbeat/tests/system/test_0050_icmp.py

@@ -11,9 +11,9 @@ def test_2_pings(self):
         assert len(objs) == 2
         assert all([o["icmp.version"] == 4 for o in objs])
         assert objs[0]["@timestamp"] == "2015-10-19T21:47:49.900Z"
-        assert objs[0]["responsetime"] == 12
+        assert objs[0]["event.duration"] == 12152000
         assert objs[1]["@timestamp"] == "2015-10-19T21:47:49.924Z"
-        assert objs[1]["responsetime"] == 11
+        assert objs[1]["event.duration"] == 11935000
         self.assert_common_fields(objs)
         self.assert_common_icmp4_fields(objs[0])
         self.assert_common_icmp4_fields(objs[1])
@@ -26,7 +26,7 @@ def test_icmp4_ping(self):
         assert len(objs) == 1
         assert objs[0]["icmp.version"] == 4
         assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.817Z"
-        assert objs[0]["responsetime"] == 20
+        assert objs[0]["event.duration"] == 20130000
         self.assert_common_fields(objs)
         self.assert_common_icmp4_fields(objs[0])
 
@@ -38,7 +38,7 @@ def test_icmp4_ping_over_vlan(self):
         assert len(objs) == 1
         assert objs[0]["icmp.version"] == 4
         assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.849Z"
-        assert objs[0]["responsetime"] == 12
+        assert objs[0]["event.duration"] == 12192000
         self.assert_common_fields(objs)
         self.assert_common_icmp4_fields(objs[0])
 
@@ -50,7 +50,7 @@ def test_icmp6_ping(self):
         assert len(objs) == 1
         assert objs[0]["icmp.version"] == 6
         assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.872Z"
-        assert objs[0]["responsetime"] == 16
+        assert objs[0]["event.duration"] == 16439000
         self.assert_common_fields(objs)
         self.assert_common_icmp6_fields(objs[0])
 
@@ -62,18 +62,20 @@ def test_icmp6_ping_over_vlan(self):
         assert len(objs) == 1
         assert objs[0]["icmp.version"] == 6
         assert objs[0]["@timestamp"] == "2015-10-19T20:49:23.901Z"
-        assert objs[0]["responsetime"] == 12
+        assert objs[0]["event.duration"] == 12333000
         self.assert_common_fields(objs)
         self.assert_common_icmp6_fields(objs[0])
 
     def assert_common_fields(self, objs):
         assert all([o["type"] == "icmp" for o in objs])
-        assert all([o["bytes_in"] == 4 for o in objs])
-        assert all([o["bytes_out"] == 4 for o in objs])
+        assert all([o["event.dataset"] == "icmp" for o in objs])
+        assert all([o["source.bytes"] == 4 for o in objs])
+        assert all([o["destination.bytes"] == 4 for o in objs])
         assert all([("server.port" in o) == False for o in objs])
         assert all([("transport" in o) == False for o in objs])
 
     def assert_common_icmp4_fields(self, obj):
+        assert obj["network.transport"] == "icmp"
         assert obj["server.ip"] == "10.0.0.2"
         assert obj["client.ip"] == "10.0.0.1"
         assert obj["path"] == "10.0.0.2"
@@ -86,6 +88,7 @@ def assert_common_icmp4_fields(self, obj):
         assert obj["icmp.response.code"] == 0
 
     def assert_common_icmp6_fields(self, obj):
+        assert obj["network.transport"] == "ipv6-icmp"
         assert obj["server.ip"] == "::2"
         assert obj["client.ip"] == "::1"
         assert obj["path"] == "::2"