fim: implement ebpf backend

auditbeat/.gitignore

@@ -6,4 +6,3 @@ module/*/_meta/config.yml
 /auditbeat
 /auditbeat.test
 /docs/html_docs
-

auditbeat/auditbeat.reference.yml

@@ -116,6 +116,11 @@ auditbeat.modules:
   # Set to true to publish fields with null values in events.
   #keep_null: false
 
+  # Select the backend which will be used to source events.
+  # Valid values: ebpf, fsnotify.
+  # Default: fsnotify.
+  force_backend: fsnotify
+
   # Parse detailed information for the listed fields. Field paths in the list below
   # that are a prefix of other field paths imply the longer field path. A set of
   # fields may be specified using an RE2 regular expression quoted in //. For example
@@ -1785,4 +1790,3 @@ logging.files:
 #features:
 #  fqdn:
 #    enabled: true
-

auditbeat/docker-compose.yml

@@ -14,6 +14,7 @@ services:
       - KIBANA_PORT=5601
     volumes:
       - ${PWD}/..:/go/src/github.com/elastic/beats/
+      - /sys/kernel/tracing/:/sys/kernel/tracing/
     command: make
     privileged: true
     pid: host

auditbeat/internal/ebpf/seccomp_linux.go

@@ -0,0 +1,40 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+	"runtime"
+
+	"github.com/elastic/beats/v7/libbeat/common/seccomp"
+)
+
+func init() {
+	switch runtime.GOARCH {
+	case "amd64", "arm64":
+		syscalls := []string{
+			"bpf",
+			"eventfd2",        // needed by ringbuf
+			"perf_event_open", // needed by tracepoints
+		}
+		if err := seccomp.ModifyDefaultPolicy(seccomp.AddSyscall, syscalls...); err != nil {
+			panic(err)
+		}
+	}
+}

auditbeat/internal/ebpf/watcher_linux.go

@@ -0,0 +1,175 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/elastic/ebpfevents"
+)
+
+func init() {
+	gWatcher = &watcher{}
+}
+
+type EventMask uint64
+
+type Watcher interface {
+	Subscribe(string, EventMask) (<-chan ebpfevents.Event, <-chan error)
+	Unsubscribe(string)
+}
+
+var gWatcher *watcher
+
+type client struct {
+	name   string
+	mask   EventMask
+	events chan ebpfevents.Event
+	errors chan error
+}
+
+type watcher struct {
+	sync.Mutex
+	ctx     context.Context
+	cancel  context.CancelFunc
+	loader  *ebpfevents.Loader
+	clients map[string]client
+	status  status
+	err     error
+}
+
+type status int
+
+const (
+	stopped status = iota
+	started
+)
+
+func GetWatcher() (Watcher, error) {
+	gWatcher.Lock()
+	defer gWatcher.Unlock()
+
+	if gWatcher.err != nil {
+		return nil, gWatcher.err
+	}
+
+	if gWatcher.status == stopped {
+		startLocked()
+	}
+
+	return gWatcher, gWatcher.err
+}
+
+func (w *watcher) Subscribe(name string, events EventMask) (<-chan ebpfevents.Event, <-chan error) {
+	w.Lock()
+	defer w.Unlock()
+
+	if w.status == stopped {
+		startLocked()
+	}
+
+	w.clients[name] = client{
+		name:   name,
+		mask:   events,
+		events: make(chan ebpfevents.Event),
+		errors: make(chan error),
+	}
+
+	return w.clients[name].events, w.clients[name].errors
+}
+
+func (w *watcher) Unsubscribe(name string) {
+	w.Lock()
+	defer w.Unlock()
+
+	delete(w.clients, name)
+
+	if w.nclients() == 0 {
+		stopLocked()
+	}
+}
+
+func startLocked() {
+	loader, err := ebpfevents.NewLoader()
+	if err != nil {
+		gWatcher.err = fmt.Errorf("new ebpf loader: %w", err)
+		return
+	}
+
+	gWatcher.loader = loader
+	gWatcher.clients = make(map[string]client)
+
+	events := make(chan ebpfevents.Event)
+	errors := make(chan error)
+	gWatcher.ctx, gWatcher.cancel = context.WithCancel(context.Background())
+
+	go gWatcher.loader.EventLoop(gWatcher.ctx, events, errors)
+	go func() {
+		for {
+			select {
+			case err := <-errors:
+				for _, client := range gWatcher.clients {
+					client.errors <- err
+				}
+				continue
+			case ev := <-events:
+				for _, client := range gWatcher.clients {
+					if client.mask&EventMask(ev.Type) != 0 {
+						client.events <- ev
+					}
+				}
+				continue
+			case <-gWatcher.ctx.Done():
+				return
+			}
+		}
+	}()
+
+	gWatcher.status = started
+}
+
+func stopLocked() {
+	_ = gWatcher.close()
+	gWatcher.status = stopped
+	gWatcher.err = nil
+}
+
+func (w *watcher) nclients() int {
+	return len(w.clients)
+}
+
+func (w *watcher) close() error {
+	if w.cancel != nil {
+		w.cancel()
+	}
+
+	if w.loader != nil {
+		_ = w.loader.Close()
+	}
+
+	for _, cl := range w.clients {
+		close(cl.events)
+		close(cl.errors)
+	}
+
+	return nil
+}

auditbeat/internal/ebpf/watcher_other.go

@@ -0,0 +1,28 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build !linux
+
+package ebpf
+
+import "errors"
+
+var ErrNotSupported = errors.New("not supported")
+
+func NewWatcher() (Watcher, error) {
+	return nil, ErrNotSupported
+}

auditbeat/internal/ebpf/watcher_test.go

@@ -0,0 +1,63 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:build linux
+
+package ebpf
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const allEvents = EventMask(math.MaxUint64)
+
+func TestWatcherStartStop(t *testing.T) {
+	assert.Equal(t, gWatcher.status, stopped)
+	assert.Nil(t, gWatcher.err)
+
+	w, err := GetWatcher()
+	if err != nil {
+		t.Skipf("skipping ebpf watcher test: %v", err)
+	}
+	assert.Equal(t, gWatcher.status, started)
+	assert.Equal(t, 0, gWatcher.nclients())
+
+	_, _ = w.Subscribe("test-1", allEvents)
+	assert.Equal(t, 1, gWatcher.nclients())
+
+	_, _ = w.Subscribe("test-2", allEvents)
+	assert.Equal(t, 2, gWatcher.nclients())
+
+	w.Unsubscribe("test-2")
+	assert.Equal(t, 1, gWatcher.nclients())
+
+	w.Unsubscribe("dummy")
+	assert.Equal(t, 1, gWatcher.nclients())
+
+	assert.Equal(t, gWatcher.status, started)
+	w.Unsubscribe("test-1")
+	assert.Equal(t, 0, gWatcher.nclients())
+	assert.Equal(t, gWatcher.status, stopped)
+
+	_, _ = w.Subscribe("new", allEvents)
+	assert.Equal(t, 1, gWatcher.nclients())
+	assert.Equal(t, gWatcher.status, started)
+	w.Unsubscribe("new")
+}

auditbeat/module/file_integrity/config.go

@@ -18,10 +18,12 @@
 package file_integrity
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"path/filepath"
 	"regexp"
+	"runtime"
 	"sort"
 	"strings"
 
@@ -72,6 +74,18 @@ const (
 	XXH64       HashType = "xxh64"
 )
 
+type Backend string
+
+const (
+	BackendFSNotify Backend = "fsnotify"
+	BackendEBPF     Backend = "ebpf"
+)
+
+func (b *Backend) Unpack(v string) error {
+	*b = Backend(v)
+	return nil
+}
+
 // Config contains the configuration parameters for the file integrity
 // metricset.
 type Config struct {
@@ -86,6 +100,7 @@ type Config struct {
 	Recursive           bool            `config:"recursive"` // Recursive enables recursive monitoring of directories.
 	ExcludeFiles        []match.Matcher `config:"exclude_files"`
 	IncludeFiles        []match.Matcher `config:"include_files"`
+	ForceBackend        Backend         `config:"force_backend"`
 }
 
 // Validate validates the config data and return an error explaining all the
@@ -160,6 +175,11 @@ nextHash:
 	if err != nil {
 		errs = append(errs, fmt.Errorf("invalid scan_rate_per_sec value: %w", err))
 	}
+
+	if c.ForceBackend == BackendEBPF && runtime.GOOS != "linux" {
+		errs = append(errs, errors.New("force_backend can only be specified on linux"))
+	}
+
 	return errs.Err()
 }