[ECS] Packetbeat ecs 1.8 (elastic#23783)

* Packetbeat changes for ECS 1.8

* Remove unused parameter

CHANGELOG.next.asciidoc

@@ -981,6 +981,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Change build process for x-pack distribution {pull}21979[21979]
 - Tuned the internal queue size to reduce the chances of events being dropped. {pull}22650[22650]
 - Add support for "http.request.mime_type" and "http.response.mime_type". {pull}22940[22940]
+- Upgrade to ECS 1.8.0. {pull}23783[23783]
 
 *Functionbeat*
 

packetbeat/cmd/root.go

@@ -37,7 +37,7 @@ const (
 	Name = "packetbeat"
 
 	// ecsVersion specifies the version of ECS that Packetbeat is implementing.
-	ecsVersion = "1.7.0"
+	ecsVersion = "1.8.0"
 )
 
 // withECSVersion is a modifier that adds ecs.version to events.

packetbeat/protos/amqp/amqp.go

@@ -455,6 +455,10 @@ func (amqp *amqpPlugin) publishTransaction(t *amqpTransaction) {
 	}
 	fields["amqp"] = t.amqp
 
+	if userID, found := t.amqp["user-id"]; found {
+		fields["user.id"] = userID
+	}
+
 	//let's try to convert request/response to a readable format
 	if amqp.sendRequest {
 		if t.method == "basic.publish" {

packetbeat/protos/cassandra/pub.go

@@ -69,7 +69,9 @@ func (pub *transPub) createEvent(requ, resp *message) beat.Event {
 
 	evt, pbf := pb.NewBeatEvent(ts)
 	pbf.SetSource(&src)
+	pbf.AddIP(src.IP)
 	pbf.SetDestination(&dst)
+	pbf.AddIP(dst.IP)
 	pbf.Event.Dataset = "cassandra"
 	pbf.Network.Transport = "tcp"
 	pbf.Network.Protocol = pbf.Event.Dataset

packetbeat/protos/http/http.go

@@ -19,6 +19,7 @@ package http
 
 import (
 	"bytes"
+	"encoding/base64"
 	"fmt"
 	"net"
 	"net/url"
@@ -439,6 +440,11 @@ func (http *httpPlugin) handleHTTP(
 	m.tcpTuple = *tcptuple
 	m.direction = dir
 	m.cmdlineTuple = http.watcher.FindProcessesTupleTCP(tcptuple.IPPort())
+
+	if !http.redactAuthorization {
+		m.username = extractBasicAuthUser(m.headers)
+	}
+
 	http.hideHeaders(m)
 
 	if m.isRequest {
@@ -533,6 +539,8 @@ func (http *httpPlugin) newTransaction(requ, resp *message) beat.Event {
 	evt, pbf := pb.NewBeatEvent(ts)
 	pbf.SetSource(src)
 	pbf.SetDestination(dst)
+	pbf.AddIP(src.IP)
+	pbf.AddIP(dst.IP)
 	pbf.Network.Transport = "tcp"
 	pbf.Network.Protocol = "http"
 
@@ -552,6 +560,9 @@ func (http *httpPlugin) newTransaction(requ, resp *message) beat.Event {
 		host, port := extractHostHeader(string(requ.host))
 		if net.ParseIP(host) == nil {
 			pbf.Destination.Domain = host
+			pbf.AddHost(host)
+		} else {
+			pbf.AddIP(host)
 		}
 		if port == 0 {
 			port = int(pbf.Destination.Port)
@@ -560,6 +571,7 @@ func (http *httpPlugin) newTransaction(requ, resp *message) beat.Event {
 		}
 		pbf.Event.Start = requ.ts
 		pbf.Network.ForwardedIP = string(requ.realIP)
+		pbf.AddIP(string(requ.realIP))
 		pbf.Error.Message = requ.notes
 
 		// http
@@ -568,6 +580,7 @@ func (http *httpPlugin) newTransaction(requ, resp *message) beat.Event {
 		httpFields.RequestBodyBytes = int64(requ.contentLength)
 		httpFields.RequestMethod = bytes.ToLower(requ.method)
 		httpFields.RequestReferrer = requ.referer
+		pbf.AddHost(string(requ.referer))
 		if requ.sendBody && len(requ.body) > 0 {
 			httpFields.RequestBodyBytes = int64(len(requ.body))
 			httpFields.RequestBodyContent = common.NetString(requ.body)
@@ -588,6 +601,11 @@ func (http *httpPlugin) newTransaction(requ, resp *message) beat.Event {
 		}
 		fields["method"] = httpFields.RequestMethod
 		fields["query"] = fmt.Sprintf("%s %s", requ.method, path)
+
+		if requ.username != "" {
+			fields["user.name"] = requ.username
+			pbf.AddUser(requ.username)
+		}
 	}
 
 	if resp != nil {
@@ -913,3 +931,28 @@ func (ml *messageList) pop() *message {
 func (ml *messageList) last() *message {
 	return ml.tail
 }
+
+func extractBasicAuthUser(headers map[string]common.NetString) string {
+	const prefix = "Basic "
+
+	auth := string(headers["authorization"])
+	if len(auth) < len(prefix) || !strings.EqualFold(auth[:len(prefix)], prefix) {
+		return ""
+	}
+
+	c, err := base64.StdEncoding.DecodeString(auth[len(prefix):])
+	if err != nil {
+		c, err = base64.RawStdEncoding.DecodeString(auth[len(prefix):])
+		if err != nil {
+			return ""
+		}
+	}
+
+	cs := string(c)
+	s := strings.IndexByte(cs, ':')
+	if s < 0 {
+		return ""
+	}
+
+	return cs[:s]
+}

packetbeat/protos/http/http_parser.go

@@ -62,6 +62,7 @@ type message struct {
 	isChunked     bool
 	headers       map[string]common.NetString
 	size          uint64
+	username      string
 
 	rawHeaders []byte
 

packetbeat/protos/http/http_test.go

@@ -921,6 +921,36 @@ func TestHttpParser_RedactAuthorization(t *testing.T) {
 	assert.True(t, proxyObscured)
 }
 
+func TestExtractBasicAuthUser(t *testing.T) {
+	logp.TestingSetup(logp.WithSelectors("http", "httpdetailed"))
+
+	http := httpModForTests(nil)
+	http.parserConfig.sendHeaders = true
+	http.parserConfig.sendAllHeaders = true
+
+	data := []byte("POST /services/ObjectControl?ID=client0 HTTP/1.1\r\n" +
+		"User-Agent: Mozilla/4.0 (compatible; MSIE 6.0; MS Web Services Client Protocol 2.0.50727.5472)\r\n" +
+		"Content-Type: text/xml; charset=utf-8\r\n" +
+		"SOAPAction: \"\"\r\n" +
+		"Authorization: Basic ZHVtbXk6NmQlc1AwOC1XemZ3Cg\r\n" +
+		"Proxy-Authorization: Basic cHJveHk6MWM3MGRjM2JhZDIwCg==\r\n" +
+		"Host: production.example.com\r\n" +
+		"Content-Length: 0\r\n" +
+		"Expect: 100-continue\r\n" +
+		"Accept-Encoding: gzip\r\n" +
+		"X-Forwarded-For: 10.216.89.132\r\n" +
+		"\r\n")
+
+	st := &stream{data: data, message: new(message)}
+
+	ok, _ := testParseStream(http, st, 0)
+
+	username := extractBasicAuthUser(st.message.headers)
+
+	assert.True(t, ok)
+	assert.Equal(t, "dummy", username)
+}
+
 func TestHttpParser_RedactAuthorization_raw(t *testing.T) {
 	http := httpModForTests(nil)
 	http.redactAuthorization = true

packetbeat/protos/mongodb/mongodb.go

@@ -382,7 +382,9 @@ func (mongodb *mongodbPlugin) publishTransaction(t *transaction) {
 
 	evt, pbf := pb.NewBeatEvent(t.ts)
 	pbf.SetSource(&t.src)
+	pbf.AddIP(t.src.IP)
 	pbf.SetDestination(&t.dst)
+	pbf.AddIP(t.dst.IP)
 	pbf.Source.Bytes = int64(t.bytesIn)
 	pbf.Destination.Bytes = int64(t.bytesOut)
 	pbf.Event.Dataset = "mongodb"

packetbeat/protos/mysql/mysql.go

@@ -1159,7 +1159,9 @@ func (mysql *mysqlPlugin) publishTransaction(t *mysqlTransaction) {
 
 	evt, pbf := pb.NewBeatEvent(t.ts)
 	pbf.SetSource(&t.src)
+	pbf.AddIP(t.src.IP)
 	pbf.SetDestination(&t.dst)
+	pbf.AddIP(t.dst.IP)
 	pbf.Source.Bytes = int64(t.bytesIn)
 	pbf.Destination.Bytes = int64(t.bytesOut)
 	pbf.Event.Dataset = "mysql"

packetbeat/protos/nfs/request_handler.go

@@ -79,7 +79,9 @@ func (r *rpc) handleCall(xid string, xdr *xdr, ts time.Time, tcptuple *common.TC
 
 	evt, pbf := pb.NewBeatEvent(ts)
 	pbf.SetSource(&src)
+	pbf.AddIP(src.IP)
 	pbf.SetDestination(&dst)
+	pbf.AddIP(dst.IP)
 	pbf.Source.Bytes = int64(xdr.size())
 	pbf.Event.Dataset = "nfs"
 	pbf.Event.Start = ts
@@ -102,6 +104,8 @@ func (r *rpc) handleCall(xid string, xdr *xdr, ts time.Time, tcptuple *common.TC
 		"xid": xid,
 	}
 
+	fields := evt.Fields
+
 	authFlavor := xdr.getUInt()
 	authOpaque := xdr.getDynamicOpaque()
 	switch authFlavor {
@@ -119,8 +123,14 @@ func (r *rpc) handleCall(xid string, xdr *xdr, ts time.Time, tcptuple *common.TC
 			pbf.Source.Domain = machine
 		}
 		cred["machinename"] = machine
+		fields["host.hostname"] = machine
+
 		cred["uid"] = credXdr.getUInt()
+		fields["user.id"] = cred["uid"]
+
 		cred["gid"] = credXdr.getUInt()
+		fields["group.id"] = cred["gid"]
+
 		cred["gids"] = credXdr.getUIntVector()
 		rpcInfo["cred"] = cred
 	case 6:
@@ -133,7 +143,6 @@ func (r *rpc) handleCall(xid string, xdr *xdr, ts time.Time, tcptuple *common.TC
 	xdr.getUInt()
 	xdr.getDynamicOpaque()
 
-	fields := evt.Fields
 	fields["status"] = common.OK_STATUS // all packages are OK for now
 	fields["type"] = pbf.Event.Dataset
 	fields["rpc"] = rpcInfo

packetbeat/protos/sip/plugin.go

@@ -178,7 +178,9 @@ func (p *plugin) buildEvent(m *message, pkt *protos.Packet) (*beat.Event, error)
 
 	src, dst := m.getEndpoints()
 	pbf.SetSource(src)
+	pbf.AddIP(src.IP)
 	pbf.SetDestination(dst)
+	pbf.AddIP(dst.IP)
 
 	p.populateEventFields(m, pbf, sipFields)
 

packetbeat/tests/system/config/golden-tests.yml

@@ -35,3 +35,8 @@ test_cases:
   - name: SIP Authenticated Register
     pcap: pcaps/sip_authenticated_register.pcap
     config: {}
+
+  - name: HTTP Basic Auth
+    pcap: pcaps/http_basicauth.pcap
+    config:
+      http_send_all_headers: true

packetbeat/tests/system/golden/established_tls-expected.json

@@ -250,4 +250,4 @@
         "tls.version_protocol": "tls",
         "type": "tls"
     }
-]
+]