[7.17](backport #39133) [Auditbeat/FIM/fsnotify]: remove time window where a child file operation of a directory can be lost

CHANGELOG.next.asciidoc

@@ -35,6 +35,9 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 *Auditbeat*
 
 
+- Prevent scenario of losing children-related file events in a directory for recursive fsnotify backend of auditbeat file integrity module {pull}39133[39133]
+
+
 *Filebeat*
 
 - Fix handling of un-parsed JSON in O365 module. {issue}37800[37800] {pull}38709[38709]

auditbeat/module/file_integrity/monitor/monitor_test.go

@@ -22,7 +22,7 @@
 
 import (
 	"errors"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -59,7 +59,7 @@
 	testDirOps(t, dir, watcher)
 
 	subdir := filepath.Join(dir, "subdir")
 	os.Mkdir(subdir, 0750)
 
 	ev, err := readTimeout(t, watcher)
 	assertNoError(t, err)
@@ -107,7 +107,7 @@
 	testDirOps(t, dir, watcher)
 
 	subdir := filepath.Join(dir, "subdir")
 	os.Mkdir(subdir, 0750)
 
 	ev, err := readTimeout(t, watcher)
 	assertNoError(t, err)
@@ -177,6 +177,11 @@
 		t.Skip("Skipping permissions test on Windows")
 	}
 
+	if os.Getuid() == 0 {
+		t.Skip("skipping as root can access every file and thus this unittest will fail")
+		return
+	}
+
 	// Create dir to be watched
 
 	dir, err := ioutil.TempDir("", "monitor")
@@ -227,7 +232,7 @@
 
 	ev, err := readTimeout(t, watcher)
 	assert.Equal(t, errReadTimeout, err)
 	if err != errReadTimeout {
 		t.Fatalf("Expected timeout, got event %+v", ev)
 	}
 
@@ -242,7 +247,7 @@
 	for {
 		// No event is received
 		ev, err := readTimeout(t, watcher)
-		if err == errReadTimeout {
+		if errors.Is(err, errReadTimeout) {
 			break
 		}
 		assertNoError(t, err)
@@ -326,7 +331,7 @@
 
 	ev, err := readTimeout(t, watcher)
 	assert.Equal(t, errReadTimeout, err)
 	if err != errReadTimeout {
 		t.Fatalf("Expected timeout, got event %+v", ev)
 	}
 
@@ -341,7 +346,7 @@
 	for {
 		// No event is received
 		ev, err := readTimeout(t, watcher)
 		if err == errReadTimeout {
 			break
 		}
 		assertNoError(t, err)
@@ -384,12 +389,12 @@
 	for i := 0; i < 3; i++ {
 		f, err := os.OpenFile(fpath, os.O_RDWR|os.O_APPEND, 0640)
 		assertNoError(t, err)
 		f.WriteString(" world\n")
 		f.Sync()
 		f.Close()
 
 		ev, err = readTimeout(t, watcher)
 		if err == nil || err != errReadTimeout {
 			break
 		}
 	}

auditbeat/module/file_integrity/monitor/recursive.go

@@ -18,12 +18,13 @@
 package monitor
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/joeshaw/multierror"
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
@@ -84,37 +85,63 @@
 	return watcher.inner.Errors
 }
 
+func (watcher *recursiveWatcher) watchFile(path string, info os.FileInfo) error {
+	var err error
+	if info == nil {
+		info, err = os.Lstat(path)
+		if err != nil {
+			return err
+		}
+	}
+
+	if info.IsDir() {
+		if err = watcher.tree.AddDir(path); err != nil {
+			return err
+		}
+
+		if err = watcher.inner.Add(path); err != nil {
+			return err
+		}
+
+		return nil
+	}
+
+	return watcher.tree.AddFile(path)
+}
+
 func (watcher *recursiveWatcher) addRecursive(path string) error {
 	if watcher.isExcludedPath(path) {
 		return nil
 	}
 
+	if err := watcher.watchFile(path, nil); err != nil {
+		return fmt.Errorf("failed adding watcher to '%s': %w", path, err)
+	}
+
 	var errs multierror.Errors
-	err := filepath.Walk(path, func(path string, info os.FileInfo, fnErr error) error {
-		if watcher.isExcludedPath(path) {
+	err := filepath.Walk(path, func(walkPath string, info os.FileInfo, fnErr error) error {
+		if walkPath == path {
+			return nil
+		}
+
+		if watcher.isExcludedPath(walkPath) {
 			return nil
 		}
 
 		if fnErr != nil {
-			errs = append(errs, errors.Wrapf(fnErr, "error walking path '%s'", path))
+			errs = append(errs, fmt.Errorf("error walking path '%s': %w", walkPath, fnErr))
 			// If FileInfo is not nil, the directory entry can be processed
 			// even if there was some error
 			if info == nil {
 				return nil
 			}
 		}
-		var err error
-		if info.IsDir() {
-			if err = watcher.tree.AddDir(path); err == nil {
-				if err = watcher.inner.Add(path); err != nil {
-					errs = append(errs, errors.Wrapf(err, "failed adding watcher to '%s'", path))
-					return nil
-				}
-			}
-		} else {
-			err = watcher.tree.AddFile(path)
+
+		if err := watcher.watchFile(walkPath, info); err != nil {
+			errs = append(errs, fmt.Errorf("failed adding watcher to '%s': %w", walkPath, err))
 		}
-		return err
+
+		return nil
 	})
 	watcher.log.Debugw("Added recursive watch", "path", path)