[Auditbeat] Cherry-pick #12028 to 6.8: Login: Fix re-read of utmp files (#12108)

The `login` dataset is not using the previous file offset when reading new entries in a utmp file. As a result, whenever a new login event occurs, all records are re-read.

Also expands the documentation, moves test files to testdata/, and adds a test case that adds a utmp record to the test file and re-reads it to make sure this bug does not happen again.

(cherry picked from commit 683f4f7)

Author: Christoph Wurm

CHANGELOG.next.asciidoc

@@ -39,6 +39,7 @@ https://github.com/elastic/beats/compare/v6.7.2...6.8[Check the HEAD diff]
 
 - Package dataset: Log error when Homebrew is not installed. {pull}11667[11667]
 - Process dataset: Fixed a memory leak under Windows. {pull}12100[12100]
+- 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 @@ experimental[]
 
 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,17 +8,19 @@ package login
 
 import (
 	"encoding/binary"
+	"io"
 	"io/ioutil"
 	"net"
 	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/auditbeat/core"
+	abtest "github.com/elastic/beats/auditbeat/testing"
 	"github.com/elastic/beats/libbeat/common"
-	"github.com/elastic/beats/libbeat/paths"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 )
 
@@ -27,12 +29,13 @@ func TestData(t *testing.T) {
 		t.Skip("Test only works on little-endian systems - skipping.")
 	}
 
-	defer setup(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()
 
 	events, errs := mbtest.ReportingFetchV2(f)
 	if len(errs) > 0 {
@@ -45,21 +48,98 @@ func TestData(t *testing.T) {
 		t.Fatalf("only one event expected, got %d", len(events))
 	}
 
+	events[0].RootFields.Put("event.origin", "/var/log/wtmp")
 	fullEvent := mbtest.StandardizeEvent(f, events[0], core.AddDatasetToEvent)
 	mbtest.WriteEventToDataJSON(t, fullEvent, "")
 }
 
-func TestFailedLogins(t *testing.T) {
+func TestWtmp(t *testing.T) {
 	if byteOrder != binary.LittleEndian {
 		t.Skip("Test only works on little-endian systems - skipping.")
 	}
 
-	defer setup(t)()
+	defer abtest.SetupDataDir(t)()
+
+	dir := setupTestDir(t)
+	defer os.RemoveAll(dir)
+
+	wtmpFilepath := filepath.Join(dir, "wtmp")
+
+	config := getBaseConfig()
+	config["login.wtmp_file_pattern"] = wtmpFilepath
+	config["login.btmp_file_pattern"] = ""
+	f := mbtest.NewReportingMetricSetV2(t, config)
+	defer f.(*MetricSet).utmpReader.bucket.DeleteBucket()
+
+	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", len(events))
+	}
+
+	// utmpdump: [7] [14962] [ts/2] [vagrant ] [pts/2       ] [10.0.2.2            ] [10.0.2.2       ] [2019-01-24T09:51:51,367964+00:00]
+	checkFieldValue(t, events[0].RootFields, "event.kind", "event")
+	checkFieldValue(t, events[0].RootFields, "event.action", "user_login")
+	checkFieldValue(t, events[0].RootFields, "event.outcome", "success")
+	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")
+	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) {
+	if byteOrder != binary.LittleEndian {
+		t.Skip("Test only works on little-endian systems - skipping.")
+	}
+
+	defer abtest.SetupDataDir(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()
 
 	events, errs := mbtest.ReportingFetchV2(f)
 	if len(errs) > 0 {
@@ -144,14 +224,47 @@ func getBaseConfig() map[string]interface{} {
 	}
 }
 
-// setup is copied from file_integrity/metricset_test.go.
-// TODO: Move to shared location and use in all unit tests.
-func setup(t testing.TB) func() {
-	// path.data should be set so that the DB is written to a predictable location.
-	var err error
-	paths.Paths.Data, err = ioutil.TempDir("", "beat-data-dir")
+// 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.Fatal()
+		t.Fatalf("failed to copy %v to %v", src, dst)
 	}
-	return func() { os.RemoveAll(paths.Paths.Data) }
 }

x-pack/auditbeat/tests/files/btmp_ubuntu1804 renamed to x-pack/auditbeat/module/system/login/testdata/btmp_ubuntu1804

@@ -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/files/wtmp renamed to x-pack/auditbeat/module/system/login/testdata/wtmp

@@ -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 experimental and that generates a warning, TODO: remove later