[Auditbeat] fim(kprobes): enrich file events by coupling add_process_metadata processor

CHANGELOG.next.asciidoc

@@ -151,6 +151,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Add opt-in eBPF backend for file_integrity module. {pull}37223[37223]
 - Add process data to file events (Linux only, eBPF backend). {pull}38199[38199]
 - Add container id to file events (Linux only, eBPF backend). {pull}38328[38328]
+- Add process.entity_id, process.group.name and process.group.id in add_process_metadata processor. Make fim module with kprobes backend to always add an appropriately configured add_process_metadata processor to enrich file events {pull}38776[38776]
 
 *Filebeat*
 

auditbeat/module/file_integrity/eventreader_kprobes.go

@@ -23,11 +23,16 @@ import (
 	"errors"
 	"fmt"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"github.com/elastic/beats/v7/auditbeat/module/file_integrity/kprobes"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/processors/add_process_metadata"
 
+	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/mapstr"
 
 	"golang.org/x/sys/unix"
 )
@@ -39,6 +44,41 @@ type kProbesReader struct {
 	log     *logp.Logger
 
 	parsers []FileParser
+
+	processor beat.Processor
+}
+
+var processMetadataProcessorOnce = sync.OnceValues(func() (beat.Processor, error) {
+	config, err := conf.NewConfigFrom(mapstr.M{
+		"match_pids":     []string{"process.pid"},
+		"overwrite_keys": true,
+	})
+
+	if err != nil {
+		return nil, err
+	}
+
+	return add_process_metadata.NewWithCache(config)
+})
+
+func newKProbesReader(config Config, l *logp.Logger, parsers []FileParser) (*kProbesReader, error) {
+
+	processor, err := processMetadataProcessorOnce()
+	if err != nil {
+		return nil, err
+	}
+
+	return &kProbesReader{
+		config:    config,
+		eventC:    make(chan Event),
+		log:       l,
+		parsers:   parsers,
+		processor: processor,
+	}, nil
+}
+
+func (r kProbesReader) Processor() beat.Processor {
+	return r.processor
 }
 
 func (r kProbesReader) Start(done <-chan struct{}) (<-chan Event, error) {
@@ -152,6 +192,13 @@ func (r kProbesReader) nextEvent(done <-chan struct{}) *Event {
 			start := time.Now()
 			e := NewEvent(event.Path, kProbeTypeToAction(event.Op), SourceKProbes,
 				r.config.MaxFileSizeBytes, r.config.HashTypes, r.parsers)
+
+			if e.Process == nil {
+				e.Process = &Process{}
+			}
+
+			e.Process.PID = event.PID
+
 			e.rtt = time.Since(start)
 
 			return &e

auditbeat/module/file_integrity/eventreader_linux.go

@@ -58,11 +58,7 @@ func NewEventReader(c Config, logger *logp.Logger) (EventProducer, error) {
 	if c.Backend == BackendKprobes {
 		l := logger.Named("kprobes")
 		l.Info("selected backend: kprobes")
-		return &kProbesReader{
-			config:  c,
-			log:     l,
-			parsers: FileParsers(c),
-		}, nil
+		return newKProbesReader(c, l, FileParsers(c))
 	}
 
 	// unimplemented

auditbeat/module/file_integrity/metricset.go

@@ -27,6 +27,7 @@ import (
 	bolt "go.etcd.io/bbolt"
 
 	"github.com/elastic/beats/v7/auditbeat/datastore"
+	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/mb/parse"
 	"github.com/elastic/elastic-agent-libs/logp"
@@ -62,6 +63,11 @@ type EventProducer interface {
 	Start(done <-chan struct{}) (<-chan Event, error)
 }
 
+// eventProducerWithProcessor is an EventProducer that requires a Processor
+type eventProducerWithProcessor interface {
+	Processor() beat.Processor
+}
+
 // MetricSet for monitoring file integrity.
 type MetricSet struct {
 	mb.BaseMetricSet
@@ -78,6 +84,9 @@ type MetricSet struct {
 
 	// Used when a hash can't be calculated
 	nullHashes map[HashType]Digest
+
+	// Processors
+	processors []beat.Processor
 }
 
 // New returns a new file.MetricSet.
@@ -105,6 +114,13 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		log:           logger,
 	}
 
+	// reader supports  a processor
+	if rWithProcessor, ok := r.(eventProducerWithProcessor); ok {
+		if proc := rWithProcessor.Processor(); proc != nil {
+			ms.processors = append(ms.processors, proc)
+		}
+	}
+
 	ms.nullHashes = make(map[HashType]Digest, len(config.HashTypes))
 	for _, hashType := range ms.config.HashTypes {
 		// One byte is enough so that the hashes are persisted to the datastore.
@@ -117,6 +133,10 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return ms, nil
 }
 
+func (ms *MetricSet) Processors() []beat.Processor {
+	return ms.processors
+}
+
 // Run runs the MetricSet. The method will not return control to the caller
 // until it is finished (to stop it close the reporter.Done() channel).
 func (ms *MetricSet) Run(reporter mb.PushReporterV2) {

filebeat/docs/fields.asciidoc

@@ -111136,6 +111136,13 @@ type: keyword
 
 --
 
+*`o365.audit.Activity`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.Actor`*::
 +
 --
@@ -111360,6 +111367,13 @@ type: object
 
 --
 
+*`o365.audit.Experience`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.ExtendedProperties.*`*::
 +
 --
@@ -111605,13 +111619,41 @@ type: keyword
 
 --
 
+*`o365.audit.ObjectDisplayName`*::
++
+--
+type: keyword
+
+--
+
+*`o365.audit.ObjectType`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.Operation`*::
 +
 --
 type: keyword
 
 --
 
+*`o365.audit.OperationId`*::
++
+--
+type: keyword
+
+--
+
+*`o365.audit.OperationProperties`*::
++
+--
+type: object
+
+--
+
 *`o365.audit.OrganizationId`*::
 +
 --
@@ -111661,6 +111703,13 @@ type: keyword
 
 --
 
+*`o365.audit.RequestId`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.ResultStatus`*::
 +
 --
@@ -111801,6 +111850,13 @@ type: keyword
 
 --
 
+*`o365.audit.Timestamp`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.UniqueSharingId`*::
 +
 --
@@ -111857,6 +111913,20 @@ type: keyword
 
 --
 
+*`o365.audit.WorkspaceId`*::
++
+--
+type: keyword
+
+--
+
+*`o365.audit.WorkspaceName`*::
++
+--
+type: keyword
+
+--
+
 *`o365.audit.YammerNetworkId`*::
 +
 --

libbeat/processors/add_process_metadata/add_process_metadata.go

@@ -34,6 +34,7 @@ import (
 	"github.com/elastic/elastic-agent-libs/mapstr"
 	"github.com/elastic/elastic-agent-system-metrics/metric/system/cgroup"
 	"github.com/elastic/elastic-agent-system-metrics/metric/system/resolve"
+	"github.com/elastic/go-sysinfo"
 )
 
 const (
@@ -65,16 +66,19 @@ type addProcessMetadata struct {
 	cidProvider  cidProvider
 	log          *logp.Logger
 	mappings     mapstr.M
+	uniqueID     []byte
 }
 
 type processMetadata struct {
-	name, title, exe, username, userid string
-	args                               []string
-	env                                map[string]string
-	startTime                          time.Time
-	pid, ppid                          int
-	//
-	fields mapstr.M
+	entityID           string
+	name, title, exe   string
+	username, userid   string
+	groupname, groupid string
+	args               []string
+	env                map[string]string
+	startTime          time.Time
+	pid, ppid          int
+	fields             mapstr.M
 }
 
 type processMetadataProvider interface {
@@ -134,6 +138,13 @@ func newProcessMetadataProcessorWithProvider(cfg *conf.C, provider processMetada
 		log:      log,
 		mappings: mappings,
 	}
+
+	if host, _ := sysinfo.Host(); host != nil {
+		if uniqueID := host.Info().UniqueID; uniqueID != "" {
+			p.uniqueID = []byte(uniqueID)
+		}
+	}
+
 	// don't use cgroup.ProcessCgroupPaths to save it from doing the work when container id disabled
 	if ok := containsValue(mappings, "container.id"); ok {
 		if withCache && config.CgroupCacheExpireTime != 0 {
@@ -311,6 +322,7 @@ func (p *addProcessMetadata) String() string {
 
 func (p *processMetadata) toMap() mapstr.M {
 	process := mapstr.M{
+		"entity_id":  p.entityID,
 		"name":       p.name,
 		"title":      p.title,
 		"executable": p.exe,
@@ -332,6 +344,16 @@ func (p *processMetadata) toMap() mapstr.M {
 		}
 		process["owner"] = user
 	}
+	if p.groupname != "" || p.groupid != "" {
+		group := mapstr.M{}
+		if p.groupname != "" {
+			group["name"] = p.groupname
+		}
+		if p.groupid != "" {
+			group["id"] = p.groupid
+		}
+		process["group"] = group
+	}
 
 	return mapstr.M{
 		"process": process,