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

CHANGELOG.next.asciidoc

@@ -156,6 +156,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

@@ -26,9 +26,10 @@ import (
 	"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"
 
 	"github.com/elastic/elastic-agent-libs/logp"
-
 	"golang.org/x/sys/unix"
 )
 
@@ -39,6 +40,30 @@ type kProbesReader struct {
 	log     *logp.Logger
 
 	parsers []FileParser
+
+	processor beat.Processor
+}
+
+func newKProbesReader(config Config, l *logp.Logger, parsers []FileParser) (*kProbesReader, error) {
+	processor, err := add_process_metadata.NewWithConfig(
+		add_process_metadata.ConfigOverwriteKeys(true),
+		add_process_metadata.ConfigMatchPIDs([]string{"process.pid"}),
+	)
+	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 +177,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

@@ -28,6 +28,7 @@ import (
 
 	"github.com/elastic/beats/v7/auditbeat/ab"
 	"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"
@@ -63,6 +64,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
@@ -79,6 +85,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.
@@ -106,6 +115,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.
@@ -118,6 +134,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) {

libbeat/processors/add_process_metadata/add_process_metadata.go

@@ -34,6 +34,7 @@
 	"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,18 @@
 	cidProvider  cidProvider
 	log          *logp.Logger
 	mappings     mapstr.M
+	uniqueID     []byte
 }
 
 type processMetadata struct {
+	entityID           string
 	name, title, exe, username, userid string
 	args                               []string
 	env                                map[string]string
 	startTime                          time.Time
 	pid, ppid                          int
+	groupname, groupid string
 	capEffective, capPermitted         []string
-	//
 	fields mapstr.M
 }
 
@@ -93,33 +96,48 @@
 
 // New constructs a new add_process_metadata processor.
 func New(cfg *conf.C) (beat.Processor, error) {
-	return newProcessMetadataProcessorWithProvider(cfg, &procCache, false)
+	config := defaultConfig()
+	if err := cfg.Unpack(&config); err != nil {
+		return nil, fmt.Errorf("fail to unpack the %v configuration: %w", processorName, err)
+	}
+
+	return newProcessMetadataProcessorWithProvider(config, &procCache, false)
 }
 
 // NewWithCache construct a new add_process_metadata processor with cache for container IDs.
 // Resulting processor implements `Close()` to release the cache resources.
 func NewWithCache(cfg *conf.C) (beat.Processor, error) {
+	config := defaultConfig()
+	if err := cfg.Unpack(&config); err != nil {
+		return nil, fmt.Errorf("fail to unpack the %v configuration: %w", processorName, err)
+	}
+
+	return newProcessMetadataProcessorWithProvider(config, &procCache, true)
+}
+
+func NewWithConfig(opts ...ConfigOption) (beat.Processor, error) {
+	cfg := defaultConfig()
+
+	for _, o := range opts {
+		o(&cfg)
+	}
+
 	return newProcessMetadataProcessorWithProvider(cfg, &procCache, true)
 }
 
-func newProcessMetadataProcessorWithProvider(cfg *conf.C, provider processMetadataProvider, withCache bool) (proc beat.Processor, err error) {
+func newProcessMetadataProcessorWithProvider(config config, provider processMetadataProvider, withCache bool) (proc beat.Processor, err error) {
 	// Logging (each processor instance has a unique ID).
 	var (
 		id  = int(instanceID.Inc())
 		log = logp.NewLogger(processorName).With("instance_id", id)
 	)
 
-	config := defaultConfig()
-	if err = cfg.Unpack(&config); err != nil {
-		return nil, fmt.Errorf("fail to unpack the %v configuration: %w", processorName, err)
-	}
-
 	// If neither option is configured, then add a default. A default cgroup_regex
 	// cannot be added to the struct returned by defaultConfig() because if
 	// config_regex is set, it would take precedence over any user-configured
 	// cgroup_prefixes.
-	hasCgroupPrefixes, _ := cfg.Has("cgroup_prefixes", -1)
-	hasCgroupRegex, _ := cfg.Has("cgroup_regex", -1)
+	hasCgroupPrefixes := len(config.CgroupPrefixes) > 0
+	hasCgroupRegex := config.CgroupRegex != nil
 	if !hasCgroupPrefixes && !hasCgroupRegex {
 		config.CgroupRegex = defaultCgroupRegex
 	}
@@ -135,6 +153,13 @@
 		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 {
@@ -312,6 +337,7 @@
 
 func (p *processMetadata) toMap() mapstr.M {
 	process := mapstr.M{
+		"entity_id":  p.entityID,
 		"name":       p.name,
 		"title":      p.title,
 		"executable": p.exe,
@@ -339,6 +365,16 @@
 	if len(p.capPermitted) > 0 {
 		process.Put("thread.capabilities.permitted", p.capPermitted)
 	}
+	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,

libbeat/processors/add_process_metadata/add_process_metadata_test.go

@@ -48,10 +48,11 @@ func TestAddProcessMetadata(t *testing.T) {
 	startTime := time.Now()
 	testProcs := testProvider{
 		1: {
-			name:  "systemd",
-			title: "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
-			exe:   "/usr/lib/systemd/systemd",
-			args:  []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
+			name:     "systemd",
+			entityID: "XCOVE56SVVEOKBNX",
+			title:    "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
+			exe:      "/usr/lib/systemd/systemd",
+			args:     []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
 			env: map[string]string{
 				"HOME":       "/",
 				"TERM":       "linux",
@@ -67,10 +68,11 @@ func TestAddProcessMetadata(t *testing.T) {
 			capPermitted: capMock,
 		},
 		3: {
-			name:  "systemd",
-			title: "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
-			exe:   "/usr/lib/systemd/systemd",
-			args:  []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
+			name:     "systemd",
+			entityID: "XCOVE56SVVEOKBNX",
+			title:    "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
+			exe:      "/usr/lib/systemd/systemd",
+			args:     []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
 			env: map[string]string{
 				"HOME":       "/",
 				"TERM":       "linux",
@@ -159,6 +161,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				},
 				"process": mapstr.M{
 					"name":       "systemd",
+					"entity_id":  "XCOVE56SVVEOKBNX",
 					"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 					"executable": "/usr/lib/systemd/systemd",
 					"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -250,6 +253,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				"parent": mapstr.M{
 					"process": mapstr.M{
 						"name":       "systemd",
+						"entity_id":  "XCOVE56SVVEOKBNX",
 						"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 						"executable": "/usr/lib/systemd/systemd",
 						"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -290,6 +294,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				"parent": mapstr.M{
 					"process": mapstr.M{
 						"name":       "systemd",
+						"entity_id":  "XCOVE56SVVEOKBNX",
 						"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 						"executable": "/usr/lib/systemd/systemd",
 						"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -337,6 +342,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				"parent": mapstr.M{
 					"process": mapstr.M{
 						"name":       "systemd",
+						"entity_id":  "XCOVE56SVVEOKBNX",
 						"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 						"executable": "/usr/lib/systemd/systemd",
 						"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -541,6 +547,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				},
 				"process": mapstr.M{
 					"name":       "systemd",
+					"entity_id":  "XCOVE56SVVEOKBNX",
 					"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 					"executable": "/usr/lib/systemd/systemd",
 					"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -672,6 +679,7 @@ func TestAddProcessMetadata(t *testing.T) {
 				},
 				"process": mapstr.M{
 					"name":       "systemd",
+					"entity_id":  "XCOVE56SVVEOKBNX",
 					"title":      "/usr/lib/systemd/systemd --switched-root --system --deserialize 22",
 					"executable": "/usr/lib/systemd/systemd",
 					"args":       []string{"/usr/lib/systemd/systemd", "--switched-root", "--system", "--deserialize", "22"},
@@ -768,7 +776,7 @@ func TestAddProcessMetadata(t *testing.T) {
 			config: mapstr.M{
 				"cgroup_regex": "",
 			},
-			initErr: errors.New("fail to unpack the add_process_metadata configuration: cgroup_regexp must contain exactly one capturing group for the container ID accessing config"),
+			initErr: errors.New("cgroup_regexp must contain exactly one capturing group for the container ID accessing config"),
 		},
 		{
 			description: "cgroup_prefixes configured",
@@ -789,17 +797,23 @@ func TestAddProcessMetadata(t *testing.T) {
 		},
 	} {
 		t.Run(test.description, func(t *testing.T) {
-			config, err := conf.NewConfigFrom(test.config)
-			if err != nil {
-				t.Fatal(err)
+			configC, err := conf.NewConfigFrom(test.config)
+			assert.NoError(t, err)
+
+			config := defaultConfig()
+			if err := configC.Unpack(&config); err != nil {
+				if test.initErr == nil {
+					t.Fatal(err)
+				}
+				assert.EqualError(t, err, test.initErr.Error())
+				return
 			}
 
 			proc, err := newProcessMetadataProcessorWithProvider(config, testProcs, true)
-			if test.initErr == nil {
-				if err != nil {
+			if err != nil {
+				if test.initErr == nil {
 					t.Fatal(err)
 				}
-			} else {
 				assert.EqualError(t, err, test.initErr.Error())
 				return
 			}
@@ -830,7 +844,11 @@ func TestAddProcessMetadata(t *testing.T) {
 			"include_fields": []string{"process.name"},
 		}
 
-		config, err := conf.NewConfigFrom(c)
+		configC, err := conf.NewConfigFrom(c)
+		assert.NoError(t, err)
+
+		config := defaultConfig()
+		err = configC.Unpack(&config)
 		assert.NoError(t, err)
 
 		proc, err := newProcessMetadataProcessorWithProvider(config, testProcs, true)

libbeat/processors/add_process_metadata/cache.go

@@ -53,7 +53,7 @@ func (pc *processCache) getEntryUnlocked(pid int) (entry processCacheEntry, vali
 	if entry, valid = pc.cache[pid]; valid {
 		valid = entry.expiration.After(time.Now())
 	}
-	return
+	return entry, valid
 }
 
 func (pc *processCache) GetProcessMetadata(pid int) (*processMetadata, error) {