[Auditbeat] System module: Add entity_id fields (#10500)

Implements `{entity}.entity_id` as a SHA-256 hash as proposed in #10463.

Closes #10463.

CHANGELOG.next.asciidoc

@@ -228,6 +228,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - System module `process` dataset: Add user information to processes. {pull}9963[9963]
 - Add system `package` dataset. {pull}10225[10225]
 - Add system module `login` dataset. {pull}9327[9327]
+- Add `entity_id` fields. {pull}10500[10500]
 
 *Filebeat*
 

auditbeat/docs/fields.asciidoc

@@ -6237,6 +6237,16 @@ Origin of the event. This can be a file path (e.g. `/var/log/log.1`), or the nam
 --
 
 
+*`user.entity_id`*::
++
+--
+type: keyword
+
+ID uniquely identifying the user on a host. It is computed as a SHA-256 hash of the host ID, user ID, and user name.
+
+
+--
+
 *`user.terminal`*::
 +
 --
@@ -6245,6 +6255,28 @@ type: keyword
 Terminal of the user.
 
 
+--
+
+
+*`process.entity_id`*::
++
+--
+type: keyword
+
+ID uniquely identifying the process. It is computed as a SHA-256 hash of the host ID, PID, and process start time.
+
+
+--
+
+
+*`socket.entity_id`*::
++
+--
+type: keyword
+
+ID uniquely identifying the socket. It is computed as a SHA-256 hash of the host ID, socket inode, local IP, local port, remote IP, and remote port.
+
+
 --
 
 [float]
@@ -6424,6 +6456,17 @@ The operating system's kernel version.
 
 
 
+*`system.audit.package.entity_id`*::
++
+--
+type: keyword
+
+ID uniquely identifying the package. It is computed as a SHA-256 hash of the
+  host ID, package name, and package version.
+
+
+--
+
 *`system.audit.package.name`*::
 +
 --

x-pack/auditbeat/module/system/_meta/fields.yml

@@ -17,11 +17,34 @@
   - name: user
     type: group
     fields:
+    - name: entity_id
+      type: keyword
+      description: >
+        ID uniquely identifying the user on a host. It is computed as a SHA-256 hash
+        of the host ID, user ID, and user name.
     - name: terminal
       type: keyword
       description: >
         Terminal of the user.
 
+  - name: process
+    type: group
+    fields:
+    - name: entity_id
+      type: keyword
+      description: >
+        ID uniquely identifying the process. It is computed as a SHA-256 hash of the
+        host ID, PID, and process start time.
+
+  - name: socket
+    type: group
+    fields:
+    - name: entity_id
+      type: keyword
+      description: >
+        ID uniquely identifying the socket. It is computed as a SHA-256 hash of the
+        host ID, socket inode, local IP, local port, remote IP, and remote port.
+
   - name: system.audit
     type: group
     description: >

x-pack/auditbeat/module/system/entity_hash.go

@@ -0,0 +1,26 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package system
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"hash"
+)
+
+// EntityHash calculates a standard entity hash.
+type EntityHash struct {
+	hash.Hash
+}
+
+// NewEntityHash creates a new EntityHash.
+func NewEntityHash() EntityHash {
+	return EntityHash{sha256.New()}
+}
+
+// Sum returns the hash as a string.
+func (h *EntityHash) Sum() string {
+	return hex.EncodeToString(h.Hash.Sum(nil))
+}

x-pack/auditbeat/module/system/package/_meta/data.json

@@ -7,7 +7,7 @@
     "event": {
         "action": "existing_package",
         "dataset": "package",
-        "id": "9ac4ea4c-5a0c-475f-b4c9-ec9d981ff11b",
+        "id": "ed069c3f-1d30-4e17-845b-cc915cf108b4",
         "kind": "state",
         "module": "system"
     },
@@ -18,6 +18,7 @@
     "system": {
         "audit": {
             "package": {
+                "entity_id": "c2c455d9f99375d9fefc61da52fa93778976d54b1964164778058980531d77dc",
                 "installtime": "2018-08-30T18:41:23.85657356+01:00",
                 "name": "zstd",
                 "summary": "Zstandard is a real-time compression algorithm",

x-pack/auditbeat/module/system/package/_meta/fields.yml

@@ -4,6 +4,11 @@
     `package` contains information about an installed or removed package.
   release: experimental
   fields:
+  - name: entity_id
+    type: keyword
+    description: >
+      ID uniquely identifying the package. It is computed as a SHA-256 hash of the
+        host ID, package name, and package version.
   - name: name
     type: keyword
     description: >

x-pack/auditbeat/module/system/package/package.go

@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/x-pack/auditbeat/cache"
+	"github.com/elastic/beats/x-pack/auditbeat/module/system"
 	"github.com/elastic/go-sysinfo"
 	"github.com/elastic/go-sysinfo/types"
 )
@@ -85,7 +86,7 @@ func init() {
 
 // MetricSet collects data about the system's packages.
 type MetricSet struct {
-	mb.BaseMetricSet
+	system.SystemMetricSet
 	config    config
 	log       *logp.Logger
 	cache     *cache.Cache
@@ -155,6 +156,15 @@ func (pkg Package) toMapStr() common.MapStr {
 	return mapstr
 }
 
+// entityID creates an ID that uniquely identifies this package across machines.
+func (pkg Package) entityID(hostID string) string {
+	h := system.NewEntityHash()
+	h.Write([]byte(hostID))
+	h.Write([]byte(pkg.Name))
+	h.Write([]byte(pkg.Version))
+	return h.Sum()
+}
+
 func getOS() (*types.OSInfo, error) {
 	host, err := sysinfo.Host()
 	if err != nil {
@@ -184,11 +194,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}
 
 	ms := &MetricSet{
-		BaseMetricSet: base,
-		config:        config,
-		log:           logp.NewLogger(metricsetName),
-		cache:         cache.New(),
-		bucket:        bucket,
+		SystemMetricSet: system.NewSystemMetricSet(base),
+		config:          config,
+		log:             logp.NewLogger(metricsetName),
+		cache:           cache.New(),
+		bucket:          bucket,
 	}
 
 	osInfo, err := getOS()
@@ -282,7 +292,7 @@ func (ms *MetricSet) reportState(report mb.ReporterV2) error {
 		return errors.Wrap(err, "error generating state ID")
 	}
 	for _, pkg := range packages {
-		event := packageEvent(pkg, eventTypeState, eventActionExistingPackage)
+		event := ms.packageEvent(pkg, eventTypeState, eventActionExistingPackage)
 		event.RootFields.Put("event.id", stateID.String())
 		report.Event(event)
 	}
@@ -327,19 +337,19 @@ func (ms *MetricSet) reportChanges(report mb.ReporterV2) error {
 			if missingPkg.Name == newPkg.Name {
 				found = true
 				updated[newPkg.Name] = struct{}{}
-				report.Event(packageEvent(newPkg, eventTypeEvent, eventActionPackageUpdated))
+				report.Event(ms.packageEvent(newPkg, eventTypeEvent, eventActionPackageUpdated))
 				break
 			}
 		}
 
 		if !found {
-			report.Event(packageEvent(missingPkg, eventTypeEvent, eventActionPackageRemoved))
+			report.Event(ms.packageEvent(missingPkg, eventTypeEvent, eventActionPackageRemoved))
 		}
 	}
 
 	for _, newPkg := range newPackages {
 		if _, contains := updated[newPkg.Name]; !contains {
-			report.Event(packageEvent(newPkg, eventTypeEvent, eventActionPackageInstalled))
+			report.Event(ms.packageEvent(newPkg, eventTypeEvent, eventActionPackageInstalled))
 		}
 	}
 
@@ -360,7 +370,7 @@ func convertToPackage(cacheValues []interface{}) []*Package {
 	return packages
 }
 
-func packageEvent(pkg *Package, eventType string, action eventAction) mb.Event {
+func (ms *MetricSet) packageEvent(pkg *Package, eventType string, action eventAction) mb.Event {
 	event := mb.Event{
 		RootFields: common.MapStr{
 			"event": common.MapStr{
@@ -372,6 +382,8 @@ func packageEvent(pkg *Package, eventType string, action eventAction) mb.Event {
 		MetricSetFields: pkg.toMapStr(),
 	}
 
+	event.MetricSetFields.Put("entity_id", pkg.entityID(ms.HostID()))
+
 	if pkg.Error != nil {
 		event.RootFields.Put("error.message", pkg.Error.Error())
 	}

x-pack/auditbeat/module/system/process/_meta/data.json

@@ -15,6 +15,7 @@
         "args": [
             "zsh"
         ],
+        "entity_id": "e2e0c5f51b093b71afed6af23debc906090d2f2f2afa9b930bf7e0803a6b53d5",
         "executable": "/bin/zsh",
         "name": "zsh",
         "pid": 12936,

x-pack/auditbeat/module/system/process/process.go

@@ -5,6 +5,7 @@
 package process
 
 import (
+	"encoding/binary"
 	"fmt"
 	"os"
 	"os/user"
@@ -21,6 +22,7 @@ import (
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/x-pack/auditbeat/cache"
+	"github.com/elastic/beats/x-pack/auditbeat/module/system"
 	"github.com/elastic/go-sysinfo"
 	"github.com/elastic/go-sysinfo/types"
 )
@@ -71,7 +73,7 @@ func init() {
 
 // MetricSet collects data about the host.
 type MetricSet struct {
-	mb.BaseMetricSet
+	system.SystemMetricSet
 	config    Config
 	cache     *cache.Cache
 	log       *logp.Logger
@@ -111,6 +113,15 @@ func (p Process) toMapStr() common.MapStr {
 	}
 }
 
+// entityID creates an ID that uniquely identifies this process across machines.
+func (p Process) entityID(hostID string) string {
+	h := system.NewEntityHash()
+	h.Write([]byte(hostID))
+	binary.Write(h, binary.LittleEndian, int64(p.Info.PID))
+	binary.Write(h, binary.LittleEndian, int64(p.Info.StartTime.Nanosecond()))
+	return h.Sum()
+}
+
 // New constructs a new MetricSet.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	cfgwarn.Experimental("The %v/%v dataset is experimental", moduleName, metricsetName)
@@ -126,11 +137,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}
 
 	ms := &MetricSet{
-		BaseMetricSet: base,
-		config:        config,
-		log:           logp.NewLogger(metricsetName),
-		cache:         cache.New(),
-		bucket:        bucket,
+		SystemMetricSet: system.NewSystemMetricSet(base),
+		config:          config,
+		log:             logp.NewLogger(metricsetName),
+		cache:           cache.New(),
+		bucket:          bucket,
 	}
 
 	// Load from disk: Time when state was last sent
@@ -204,12 +215,12 @@ func (ms *MetricSet) reportState(report mb.ReporterV2) error {
 	}
 	for _, p := range processes {
 		if p.Error == nil {
-			event := processEvent(p, eventTypeState, eventActionExistingProcess)
+			event := ms.processEvent(p, eventTypeState, eventActionExistingProcess)
 			event.RootFields.Put("event.id", stateID.String())
 			report.Event(event)
 		} else {
 			ms.log.Warn(p.Error)
-			report.Event(processEvent(p, eventTypeError, eventActionProcessError))
+			report.Event(ms.processEvent(p, eventTypeError, eventActionProcessError))
 		}
 	}
 
@@ -245,25 +256,25 @@ func (ms *MetricSet) reportChanges(report mb.ReporterV2) error {
 		p := cacheValue.(*Process)
 
 		if p.Error == nil {
-			report.Event(processEvent(p, eventTypeEvent, eventActionProcessStarted))
+			report.Event(ms.processEvent(p, eventTypeEvent, eventActionProcessStarted))
 		} else {
 			ms.log.Warn(p.Error)
-			report.Event(processEvent(p, eventTypeError, eventActionProcessError))
+			report.Event(ms.processEvent(p, eventTypeError, eventActionProcessError))
 		}
 	}
 
 	for _, cacheValue := range stopped {
 		p := cacheValue.(*Process)
 
 		if p.Error == nil {
-			report.Event(processEvent(p, eventTypeEvent, eventActionProcessStopped))
+			report.Event(ms.processEvent(p, eventTypeEvent, eventActionProcessStopped))
 		}
 	}
 
 	return nil
 }
 
-func processEvent(process *Process, eventType string, action eventAction) mb.Event {
+func (ms *MetricSet) processEvent(process *Process, eventType string, action eventAction) mb.Event {
 	event := mb.Event{
 		RootFields: common.MapStr{
 			"event": common.MapStr{
@@ -302,6 +313,8 @@ func processEvent(process *Process, eventType string, action eventAction) mb.Eve
 		event.RootFields.Put("error.message", process.Error.Error())
 	}
 
+	event.RootFields.Put("process.entity_id", process.entityID(ms.HostID()))
+
 	return event
 }