Auditbeat: stop monitoring excluded paths (#21282) (#22486)

If the files match the excluded pattern, recursiveWatcher does not emit events for them, as well as errors if any.

(cherry picked from commit af8eedd)

Co-authored-by: Peter Deng <[email protected]>

CHANGELOG.next.asciidoc

@@ -170,6 +170,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - system/socket: Fixed tracking of long-running connections. {pull}19033[19033]
 - auditd: Fix an error condition causing a lot of `audit_send_reply` kernel threads being created. {pull}22673[22673]
 - system/socket: Fixed start failure when run under config reloader. {issue}20851[20851] {pull}21693[21693]
+- file_integrity: stop monitoring excluded paths {issue}21278[21278] {pull}21282[21282]
 
 *Filebeat*
 

auditbeat/module/file_integrity/eventreader_fsnotify.go

@@ -39,20 +39,22 @@ type reader struct {
 
 // NewEventReader creates a new EventProducer backed by fsnotify.
 func NewEventReader(c Config) (EventProducer, error) {
-	watcher, err := monitor.New(c.Recursive)
-	if err != nil {
-		return nil, err
-	}
-
 	return &reader{
-		watcher: watcher,
-		config:  c,
-		log:     logp.NewLogger(moduleName),
+		config: c,
+		log:    logp.NewLogger(moduleName),
 	}, nil
 }
 
 func (r *reader) Start(done <-chan struct{}) (<-chan Event, error) {
+	watcher, err := monitor.New(r.config.Recursive, r.config.IsExcludedPath)
+	if err != nil {
+		return nil, err
+	}
+
+	r.watcher = watcher
 	if err := r.watcher.Start(); err != nil {
+		// Ensure that watcher is closed so that we don't leak watchers
+		r.watcher.Close()
 		return nil, errors.Wrap(err, "unable to start watcher")
 	}
 

auditbeat/module/file_integrity/monitor/monitor.go

@@ -37,15 +37,15 @@ type Watcher interface {
 
 // New creates a new Watcher backed by fsnotify with optional recursive
 // logic.
-func New(recursive bool) (Watcher, error) {
-	fsnotify, err := fsnotify.NewWatcher()
+func New(recursive bool, IsExcludedPath func(path string) bool) (Watcher, error) {
+	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
 		return nil, err
 	}
 	// Use our simulated recursive watches unless the fsnotify implementation
 	// supports OS-provided recursive watches
-	if recursive && fsnotify.SetRecursive() != nil {
-		return newRecursiveWatcher(fsnotify), nil
+	if recursive && watcher.SetRecursive() != nil {
+		return newRecursiveWatcher(watcher, IsExcludedPath), nil
 	}
-	return (*nonRecursiveWatcher)(fsnotify), nil
+	return (*nonRecursiveWatcher)(watcher), nil
 }

auditbeat/module/file_integrity/monitor/monitor_test.go

@@ -32,6 +32,10 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func alwaysInclude(path string) bool {
+	return false
+}
+
 func TestNonRecursive(t *testing.T) {
 	dir, err := ioutil.TempDir("", "monitor")
 	assertNoError(t, err)
@@ -44,7 +48,7 @@ func TestNonRecursive(t *testing.T) {
 	}
 	defer os.RemoveAll(dir)
 
-	watcher, err := New(false)
+	watcher, err := New(false, alwaysInclude)
 	assertNoError(t, err)
 
 	assertNoError(t, watcher.Add(dir))
@@ -92,7 +96,7 @@ func TestRecursive(t *testing.T) {
 	}
 	defer os.RemoveAll(dir)
 
-	watcher, err := New(true)
+	watcher, err := New(true, alwaysInclude)
 	assertNoError(t, err)
 
 	assertNoError(t, watcher.Add(dir))
@@ -147,7 +151,7 @@ func TestRecursiveNoFollowSymlink(t *testing.T) {
 
 	// Start the watcher
 
-	watcher, err := New(true)
+	watcher, err := New(true, alwaysInclude)
 	assertNoError(t, err)
 
 	assertNoError(t, watcher.Add(dir))
@@ -208,7 +212,7 @@ func TestRecursiveSubdirPermissions(t *testing.T) {
 
 	// Setup watches on watched dir
 
-	watcher, err := New(true)
+	watcher, err := New(true, alwaysInclude)
 	assertNoError(t, err)
 
 	assertNoError(t, watcher.Start())
@@ -263,6 +267,104 @@ func TestRecursiveSubdirPermissions(t *testing.T) {
 	}
 }
 
+func TestRecursiveExcludedPaths(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("Skipping permissions test on Windows")
+	}
+
+	// Create dir to be watched
+
+	dir, err := ioutil.TempDir("", "monitor")
+	assertNoError(t, err)
+	if runtime.GOOS == "darwin" {
+		if dirAlt, err := filepath.EvalSymlinks(dir); err == nil {
+			dir = dirAlt
+		}
+	}
+	defer os.RemoveAll(dir)
+
+	// Create not watched dir
+
+	outDir, err := ioutil.TempDir("", "non-watched")
+	assertNoError(t, err)
+	if runtime.GOOS == "darwin" {
+		if dirAlt, err := filepath.EvalSymlinks(outDir); err == nil {
+			outDir = dirAlt
+		}
+	}
+	defer os.RemoveAll(outDir)
+
+	// Populate not watched subdir
+
+	for _, name := range []string{"a", "b", "c"} {
+		path := filepath.Join(outDir, name)
+		assertNoError(t, os.Mkdir(path, 0755))
+		assertNoError(t, ioutil.WriteFile(filepath.Join(path, name), []byte("Hello"), 0644))
+	}
+
+	// excludes file/dir named "b"
+	selectiveExclude := func(path string) bool {
+		r := filepath.Base(path) == "b"
+		t.Logf("path: %v, excluded: %v\n", path, r)
+		return r
+	}
+
+	// Setup watches on watched dir
+
+	watcher, err := New(true, selectiveExclude)
+	assertNoError(t, err)
+
+	assertNoError(t, watcher.Start())
+	assertNoError(t, watcher.Add(dir))
+
+	defer func() {
+		assertNoError(t, watcher.Close())
+	}()
+
+	// No event is received
+
+	ev, err := readTimeout(t, watcher)
+	assert.Equal(t, errReadTimeout, err)
+	if err != errReadTimeout {
+		t.Fatalf("Expected timeout, got event %+v", ev)
+	}
+
+	// Move the outside directory into the watched
+
+	dest := filepath.Join(dir, "subdir")
+	assertNoError(t, os.Rename(outDir, dest))
+
+	// Receive all events
+
+	var evs []fsnotify.Event
+	for {
+		// No event is received
+		ev, err := readTimeout(t, watcher)
+		if err == errReadTimeout {
+			break
+		}
+		assertNoError(t, err)
+		evs = append(evs, ev)
+	}
+
+	// Verify that events for all accessible files are received
+	// "b" and "b/b" are missing as they are excluded
+
+	expected := map[string]fsnotify.Op{
+		dest:                       fsnotify.Create,
+		filepath.Join(dest, "a"):   fsnotify.Create,
+		filepath.Join(dest, "a/a"): fsnotify.Create,
+		filepath.Join(dest, "c"):   fsnotify.Create,
+		filepath.Join(dest, "c/c"): fsnotify.Create,
+	}
+	assert.Len(t, evs, len(expected))
+	for _, ev := range evs {
+		op, found := expected[ev.Name]
+		assert.True(t, found, ev.Name)
+		assert.Equal(t, op, ev.Op)
+	}
+}
+
 func testDirOps(t *testing.T, dir string, watcher Watcher) {
 	fpath := filepath.Join(dir, "file.txt")
 	fpath2 := filepath.Join(dir, "file2.txt")

auditbeat/module/file_integrity/monitor/recursive.go

@@ -36,16 +36,19 @@ type recursiveWatcher struct {
 	addC    chan string
 	addErrC chan error
 	log     *logp.Logger
+
+	isExcludedPath func(path string) bool
 }
 
-func newRecursiveWatcher(inner *fsnotify.Watcher) *recursiveWatcher {
+func newRecursiveWatcher(inner *fsnotify.Watcher, IsExcludedPath func(path string) bool) *recursiveWatcher {
 	return &recursiveWatcher{
-		inner:   inner,
-		tree:    FileTree{},
-		eventC:  make(chan fsnotify.Event, 1),
-		addC:    make(chan string),
-		addErrC: make(chan error),
-		log:     logp.NewLogger(moduleName),
+		inner:          inner,
+		tree:           FileTree{},
+		eventC:         make(chan fsnotify.Event, 1),
+		addC:           make(chan string),
+		addErrC:        make(chan error),
+		log:            logp.NewLogger(moduleName),
+		isExcludedPath: IsExcludedPath,
 	}
 }
 
@@ -82,8 +85,16 @@ func (watcher *recursiveWatcher) ErrorChannel() <-chan error {
 }
 
 func (watcher *recursiveWatcher) addRecursive(path string) error {
+	if watcher.isExcludedPath(path) {
+		return nil
+	}
+
 	var errs multierror.Errors
 	err := filepath.Walk(path, func(path string, info os.FileInfo, fnErr error) error {
+		if watcher.isExcludedPath(path) {
+			return nil
+		}
+
 		if fnErr != nil {
 			errs = append(errs, errors.Wrapf(fnErr, "error walking path '%s'", path))
 			// If FileInfo is not nil, the directory entry can be processed