[Auditbeat] Login: Fix re-read of utmp files

CHANGELOG.next.asciidoc

@@ -76,6 +76,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Package dataset: dlopen versioned librpm shared objects. {pull}11565[11565]
 - Package dataset: Nullify Librpm's rpmsqEnable. {pull}11628[11628]
 - Package dataset: Log error when Homebrew is not installed. {pull}11667[11667]
+- Login dataset: Fix re-read of utmp files. {pull}12028[12028]
 
 *Filebeat*
 

x-pack/auditbeat/module/system/login/_meta/docs.asciidoc

@@ -4,7 +4,22 @@ beta[]
 
 This is the `login` dataset of the system module.
 
-It is implemented for Linux only.
+[float]
+=== Implementation
+
+The `login` dataset is implemented for Linux only.
+
+On Linux, the dataset reads the https://en.wikipedia.org/wiki/Utmp[utmp] files
+that keep track of logins and logouts to the system. They are usually located
+at `/var/log/wtmp` (successful logins) and `/var/log/btmp` (failed logins).
+
+The file patterns used to locate the files can be configured using
+`login.wtmp_file_pattern` and `login.btmp_file_pattern`. By default,
+both the current files and any rotated files (e.g. `wtmp.1`, `wtmp.2`)
+are read.
+
+utmp files are binary, but you can display their contents using the
+`utmpdump` utility.
 
 [float]
 ==== Example dashboard

x-pack/auditbeat/module/system/login/login_test.go

@@ -8,7 +8,11 @@ package login
 
 import (
 	"encoding/binary"
+	"io"
+	"io/ioutil"
 	"net"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
@@ -28,7 +32,7 @@ func TestData(t *testing.T) {
 	defer abtest.SetupDataDir(t)()
 
 	config := getBaseConfig()
-	config["login.wtmp_file_pattern"] = "../../../tests/files/wtmp"
+	config["login.wtmp_file_pattern"] = "./testdata/wtmp"
 	config["login.btmp_file_pattern"] = ""
 	f := mbtest.NewReportingMetricSetV2(t, config)
 	defer f.(*MetricSet).utmpReader.bucket.DeleteBucket()
@@ -56,8 +60,13 @@ func TestWtmp(t *testing.T) {
 
 	defer abtest.SetupDataDir(t)()
 
+	dir := setupTestDir(t)
+	defer os.RemoveAll(dir)
+
+	wtmpFilepath := filepath.Join(dir, "wtmp")
+
 	config := getBaseConfig()
-	config["login.wtmp_file_pattern"] = "../../../tests/files/wtmp"
+	config["login.wtmp_file_pattern"] = wtmpFilepath
 	config["login.btmp_file_pattern"] = ""
 	f := mbtest.NewReportingMetricSetV2(t, config)
 	defer f.(*MetricSet).utmpReader.bucket.DeleteBucket()
@@ -85,6 +94,40 @@ func TestWtmp(t *testing.T) {
 	checkFieldValue(t, events[0].RootFields, "user.terminal", "pts/2")
 	assert.True(t, events[0].Timestamp.Equal(time.Date(2019, 1, 24, 9, 51, 51, 367964000, time.UTC)),
 		"Timestamp is not equal: %+v", events[0].Timestamp)
+
+	// Append logout event to wtmp file and check that it's read
+	wtmpFile, err := os.OpenFile(wtmpFilepath, os.O_APPEND|os.O_WRONLY, 0644)
+	if err != nil {
+		t.Fatalf("error opening %v: %v", wtmpFilepath, err)
+	}
+
+	loginUtmp := utmpC{
+		Type: DEAD_PROCESS,
+	}
+	copy(loginUtmp.Device[:], "pts/2")
+
+	err = binary.Write(wtmpFile, byteOrder, loginUtmp)
+	if err != nil {
+		t.Fatalf("error writing to %v: %v", wtmpFilepath, err)
+	}
+
+	events, errs = mbtest.ReportingFetchV2(f)
+	if len(errs) > 0 {
+		t.Fatalf("received error: %+v", errs[0])
+	}
+
+	if len(events) == 0 {
+		t.Fatal("no events were generated")
+	} else if len(events) != 1 {
+		t.Fatalf("only one event expected, got %d: %v", len(events), events)
+	}
+
+	checkFieldValue(t, events[0].RootFields, "event.kind", "event")
+	checkFieldValue(t, events[0].RootFields, "event.action", "user_logout")
+	checkFieldValue(t, events[0].RootFields, "process.pid", 14962)
+	checkFieldValue(t, events[0].RootFields, "source.ip", "10.0.2.2")
+	checkFieldValue(t, events[0].RootFields, "user.name", "vagrant")
+	checkFieldValue(t, events[0].RootFields, "user.terminal", "pts/2")
 }
 
 func TestBtmp(t *testing.T) {
@@ -96,7 +139,7 @@ func TestBtmp(t *testing.T) {
 
 	config := getBaseConfig()
 	config["login.wtmp_file_pattern"] = ""
-	config["login.btmp_file_pattern"] = "../../../tests/files/btmp_ubuntu1804"
+	config["login.btmp_file_pattern"] = "./testdata/btmp*"
 	f := mbtest.NewReportingMetricSetV2(t, config)
 	defer f.(*MetricSet).utmpReader.bucket.DeleteBucket()
 
@@ -190,3 +233,48 @@ func getBaseConfig() map[string]interface{} {
 		"datasets": []string{"login"},
 	}
 }
+
+// setupTestDir creates a temporary directory, copies the test files into it,
+// and returns the path.
+func setupTestDir(t *testing.T) string {
+	tmp, err := ioutil.TempDir("", "auditbeat-login-test-dir")
+	if err != nil {
+		t.Fatal("failed to create temp dir")
+	}
+
+	copyDir(t, "./testdata", tmp)
+
+	return tmp
+}
+
+func copyDir(t *testing.T, src, dst string) {
+	files, err := ioutil.ReadDir(src)
+	if err != nil {
+		t.Fatalf("failed to read %v", src)
+	}
+
+	for _, file := range files {
+		srcFile := filepath.Join(src, file.Name())
+		dstFile := filepath.Join(dst, file.Name())
+		copyFile(t, srcFile, dstFile)
+	}
+}
+
+func copyFile(t *testing.T, src, dst string) {
+	in, err := os.Open(src)
+	if err != nil {
+		t.Fatalf("failed to open %v", src)
+	}
+	defer in.Close()
+
+	out, err := os.Create(dst)
+	if err != nil {
+		t.Fatalf("failed to open %v", dst)
+	}
+	defer out.Close()
+
+	_, err = io.Copy(out, in)
+	if err != nil {
+		t.Fatalf("failed to copy %v to %v", src, dst)
+	}
+}

x-pack/auditbeat/module/system/login/utmp.go

@@ -146,11 +146,10 @@ func (r *UtmpFileReader) findFiles(filePattern string, utmpType UtmpType) ([]Utm
 		}
 
 		utmpFiles = append(utmpFiles, UtmpFile{
-			Inode:  Inode(fileInfo.Sys().(*syscall.Stat_t).Ino),
-			Path:   path,
-			Size:   fileInfo.Size(),
-			Offset: 0,
-			Type:   utmpType,
+			Inode: Inode(fileInfo.Sys().(*syscall.Stat_t).Ino),
+			Path:  path,
+			Size:  fileInfo.Size(),
+			Type:  utmpType,
 		})
 	}
 
@@ -178,6 +177,7 @@ func (r *UtmpFileReader) readNewInFile(loginRecordC chan<- LoginRecord, errorC c
 	if !isKnownFile {
 		r.log.Debugf("Found new file: %v (utmpFile=%+v)", utmpFile.Path, utmpFile)
 	}
+	utmpFile.Offset = savedUtmpFile.Offset
 
 	size := utmpFile.Size
 	oldSize := savedUtmpFile.Size
@@ -211,7 +211,7 @@ func (r *UtmpFileReader) readNewInFile(loginRecordC chan<- LoginRecord, errorC c
 		defer func() {
 			// Once we start reading a file, we update the file record even if something fails -
 			// otherwise we will just keep trying to re-read very frequently forever.
-			r.updateSavedUtmpFile(utmpFile, f)
+			err := r.updateSavedUtmpFile(utmpFile, f)
 			if err != nil {
 				errorC <- errors.Wrapf(err, "error updating file record for file %v", utmpFile.Path)
 			}

x-pack/auditbeat/tests/system/test_metricsets.py

@@ -34,8 +34,8 @@ def test_metricset_login(self):
                   "user.name", "user.terminal"]
 
         config = {
-            "login.wtmp_file_pattern": os.path.abspath(os.path.join(self.beat_path, "tests/files/wtmp")),
-            "login.btmp_file_pattern": "-1"
+            "login.wtmp_file_pattern": os.path.abspath(os.path.join(self.beat_path, "module/system/login/testdata/wtmp*")),
+            "login.btmp_file_pattern": os.path.abspath(os.path.join(self.beat_path, "module/system/login/testdata/btmp*")),
         }
 
         # Metricset is beta and that generates a warning, TODO: remove later