Support receiving events from audit multicast group (elastic#5081)

Update to go-libaudit v0.0.6

Closes elastic#4850
(cherry picked from commit c76f14f)

CHANGELOG.asciidoc

@@ -78,6 +78,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 
 *Auditbeat*
 
+- Add support for receiving audit events using a multicast socket. [issue]4850[4850]
+
 ==== Deprecated
 
 *Affecting all Beats*

NOTICE

@@ -314,8 +314,8 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
 --------------------------------------------------------------------
 Dependency: github.com/elastic/go-libaudit
-Version: v0.0.5
-Revision: 3a005d3d0bbcee26d60e3ab2f1890699746f4da6
+Version: v0.0.6
+Revision: df0d4981f3fce65ffd3d7411dfec3e03231b491c
 License type (autodetected): Apache License 2.0
 ./vendor/github.com/elastic/go-libaudit/LICENSE:
 --------------------------------------------------------------------

auditbeat/module/audit/kernel/_meta/docs.asciidoc

@@ -12,13 +12,6 @@ This metricset establishes a subscription to the kernel to receive the events
 as they occur. So unlike most other metricsets, the `period` configuration
 option is unused because it is not implemented using polling.
 
-The Linux kernel only supports a single subscriber to the audit events so this
-metricset cannot be used simultaneously with a service like `auditd`. `auditd`
-should be disabled if this metricset is being used. If you wish to continue to
-use `auditd` instead of this metricset to receive audit messages from the kernel
-then consider using {filebeat}/filebeat-module-auditd.html[Filebeat] to collect
-the daemon's log files.
-
 The Linux Audit Framework can send multiple messages for a single auditable
 event. For example, a `rename` syscall causes the kernel to send eight separate
 messages. Each message describes a different aspect of the activity that is
@@ -48,6 +41,24 @@ following example shows all configuration options with their default values.
   kernel.include_warnings: false
 ----
 
+*`kernel.socket_type`*:: This optional setting controls the type of
+socket that {beatname_uc} uses to receive events from the kernel. The two
+options are `unicast` and `multicast`.
++
+`unicast` should be used when {beatname_uc} is the primary userspace daemon for
+receiving audit events and managing the rules. Only a single process can receive
+audit events through the "unicast" connection so any other daemons should be
+stopped (e.g. stop `auditd`).
++
+`multicast` can be used in kernel versions 3.16 and newer. By using `multicast`
+{beatname_uc} will receive an audit event broadcast that is not exclusive to a
+a single process. This is ideal for situations where `auditd` is running and
+managing the rules. If `multicast` is specified, but the kernel version is less
+than 3.16 {beatname_uc} will automatically revert to `unicast`.
++
+By default {beatname_uc} will use `multicast` if the kernel version is 3.16 or
+newer and no rules have been defined. Otherwise `unicast` will be used.
+
 *`kernel.resolve_ids`*:: This boolean setting enables the resolution of UIDs and
 GIDs to their associated names. The default value is true.
 

auditbeat/module/audit/kernel/audit_linux.go

@@ -2,7 +2,9 @@ package kernel
 
 import (
 	"os"
+	"strconv"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/pkg/errors"
@@ -48,16 +50,17 @@ type MetricSet struct {
 
 // New constructs a new MetricSet.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
-	cfgwarn.Experimental("The %v metricset is a beta feature", metricsetName)
+	cfgwarn.Beta("The %v metricset is a beta feature", metricsetName)
 
 	config := defaultConfig
 	if err := base.Module().UnpackConfig(&config); err != nil {
 		return nil, errors.Wrap(err, "failed to unpack the audit.kernel config")
 	}
 
-	debugf("%v the metricset is running as euid=%v", logPrefix, os.Geteuid())
+	_, _, kernel, _ := kernelVersion()
+	debugf("%v the metricset is running as euid=%v on kernel=%v", logPrefix, os.Geteuid(), kernel)
 
-	client, err := libaudit.NewAuditClient(nil)
+	client, err := newAuditClient(&config)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create audit.kernel client")
 	}
@@ -71,6 +74,37 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}, nil
 }
 
+func newAuditClient(c *Config) (*libaudit.AuditClient, error) {
+	hasMulticast := hasMulticastSupport()
+
+	switch c.SocketType {
+	// Attempt to determine the optimal socket_type.
+	case "":
+		// Use multicast only when no rules are present. Specifying rules
+		// implies you want control over the audit framework so you should be
+		// using unicast.
+		if rules, _ := c.rules(); len(rules) == 0 && hasMulticast {
+			c.SocketType = "multicast"
+			logp.Info("%v kernel.socket_type=multicast will be used.", logPrefix)
+		}
+	case "multicast":
+		if !hasMulticast {
+			logp.Warn("%v kernel.socket_type is set to multicast "+
+				"but based on the kernel version multicast audit subscriptions "+
+				"are not supported. unicast will be used instead.", logPrefix)
+			c.SocketType = "unicast"
+		}
+	}
+
+	switch c.SocketType {
+	case "multicast":
+		return libaudit.NewMulticastAuditClient(nil)
+	default:
+		c.SocketType = "unicast"
+		return libaudit.NewAuditClient(nil)
+	}
+}
+
 // Run initializes the audit client and receives audit messages from the
 // kernel until the reporter's done channel is closed.
 func (ms *MetricSet) Run(reporter mb.PushReporter) {
@@ -115,8 +149,14 @@ func (ms *MetricSet) addRules(reporter mb.PushReporter) error {
 		return nil
 	}
 
+	client, err := libaudit.NewAuditClient(nil)
+	if err != nil {
+		return errors.Wrap(err, "failed to create audit client for adding rules")
+	}
+	defer client.Close()
+
 	// Delete existing rules.
-	n, err := ms.client.DeleteRules()
+	n, err := client.DeleteRules()
 	if err != nil {
 		return errors.Wrap(err, "failed to delete existing rules")
 	}
@@ -125,7 +165,7 @@ func (ms *MetricSet) addRules(reporter mb.PushReporter) error {
 	// Add rules from config.
 	var failCount int
 	for _, rule := range rules {
-		if err = ms.client.AddRule(rule.data); err != nil {
+		if err = client.AddRule(rule.data); err != nil {
 			// Treat rule add errors as warnings and continue.
 			err = errors.Wrapf(err, "failed to add kernel rule '%v'", rule.flags)
 			reporter.Error(err)
@@ -139,6 +179,17 @@ func (ms *MetricSet) addRules(reporter mb.PushReporter) error {
 }
 
 func (ms *MetricSet) initClient() error {
+	if ms.config.SocketType == "multicast" {
+		// This request will fail with EPERM if this process does not have
+		// CAP_AUDIT_CONTROL, but we will ignore the response. The user will be
+		// required to ensure that auditing is enabled if the process is only
+		// given CAP_AUDIT_READ.
+		err := ms.client.SetEnabled(true, libaudit.NoWait)
+		return errors.Wrap(err, "failed to enable auditing in the kernel")
+	}
+
+	// Unicast client initialization (requires CAP_AUDIT_CONTROL and that the
+	// process be in initial PID namespace).
 	status, err := ms.client.GetStatus()
 	if err != nil {
 		return errors.Wrap(err, "failed to get audit status")
@@ -348,3 +399,54 @@ func (s *stream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
 func (s *stream) EventsLost(count int) {
 	lostMetric.Inc()
 }
+
+func hasMulticastSupport() bool {
+	// Check the kernel version because 3.16+ should have multicast
+	// support.
+	major, minor, _, err := kernelVersion()
+	if err != nil {
+		// Assume not supported.
+		return false
+	}
+
+	switch {
+	case major > 3,
+		major == 3 && minor >= 16:
+		return true
+	}
+
+	return false
+}
+
+func kernelVersion() (major, minor int, full string, err error) {
+	var uname syscall.Utsname
+	if err := syscall.Uname(&uname); err != nil {
+		return 0, 0, "", err
+	}
+
+	data := make([]byte, len(uname.Release))
+	for i, v := range uname.Release {
+		if v == 0 {
+			break
+		}
+		data[i] = byte(v)
+	}
+
+	release := string(data)
+	parts := strings.SplitN(release, ".", 3)
+	if len(parts) < 2 {
+		return 0, 0, release, errors.Errorf("failed to parse uname release '%v'", release)
+	}
+
+	major, err = strconv.Atoi(parts[0])
+	if err != nil {
+		return 0, 0, release, errors.Wrapf(err, "failed to parse major version from '%v'", release)
+	}
+
+	minor, err = strconv.Atoi(parts[1])
+	if err != nil {
+		return 0, 0, release, errors.Wrapf(err, "failed to parse minor version from '%v'", release)
+	}
+
+	return major, minor, release, nil
+}

auditbeat/module/audit/kernel/audit_linux_test.go

@@ -1,24 +1,30 @@
 package kernel
 
 import (
+	"flag"
+	"fmt"
+	"os"
+	"os/exec"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 	"github.com/elastic/go-libaudit"
 )
 
+// Specify the -audit flag when running these tests to interact with the real
+// kernel instead of mocks. If running in Docker this requires being in the
+// host PID namespace (--pid=host) and having CAP_AUDIT_CONTROL and
+// CAP_AUDIT_WRITE (so use --privileged).
+var audit = flag.Bool("audit", false, "interact with the real audit framework")
+
 var userLoginMsg = `type=USER_LOGIN msg=audit(1492896301.818:19955): pid=12635 uid=0 auid=4294967295 ses=4294967295 msg='op=login acct=28696E76616C6964207573657229 exe="/usr/sbin/sshd" hostname=? addr=179.38.151.221 terminal=sshd res=failed'`
 
 func TestData(t *testing.T) {
 	// Create a mock netlink client that provides the expected responses.
 	mock := NewMock().
-		// GetRules response with zero rules. Used by DeleteAll rules.
-		returnACK().returnDone().
-		// AddRule response.
-		returnACK().
-		// AddRule response.
-		returnACK().
 		// Get Status response for initClient
 		returnACK().returnStatus().
 		// Send a single audit message from the kernel.
@@ -47,9 +53,74 @@ func getConfig() map[string]interface{} {
 		"module":              "audit",
 		"metricsets":          []string{"kernel"},
 		"kernel.failure_mode": "log",
-		"kernel.audit_rules": `
-		   -w /etc/passwd -p wa -k auth
-		   -a always,exit -F arch=b64 -S execve -k exec
-		`,
+		"kernel.socket_type":  "unicast",
+	}
+}
+
+func TestMulticastClient(t *testing.T) {
+	if !*audit {
+		t.Skip("-audit was not specified")
+	}
+
+	if !hasMulticastSupport() {
+		t.Skip("no multicast support")
+	}
+
+	c := map[string]interface{}{
+		"module":             "audit",
+		"metricsets":         []string{"kernel"},
+		"kernel.socket_type": "multicast",
+		"kernel.audit_rules": fmt.Sprintf(`
+		   -a always,exit -F arch=b64 -F ppid=%d -S execve -k exec
+		`, os.Getpid()),
+	}
+
+	// Any commands executed by this process will generate events due to the
+	// PPID filter we applied to the rule.
+	time.AfterFunc(time.Second, func() { exec.Command("cat", "/proc/self/status").Output() })
+
+	ms := mbtest.NewPushMetricSet(t, c)
+	events, errs := mbtest.RunPushMetricSet(5*time.Second, ms)
+	if len(errs) > 0 {
+		t.Fatalf("received errors: %+v", errs)
+	}
+
+	// The number of events is non-deterministic so there is no validation.
+	t.Logf("received %d messages via multicast", len(events))
+}
+
+func TestUnicastClient(t *testing.T) {
+	if !*audit {
+		t.Skip("-audit was not specified")
+	}
+
+	c := map[string]interface{}{
+		"module":             "audit",
+		"metricsets":         []string{"kernel"},
+		"kernel.socket_type": "unicast",
+		"kernel.audit_rules": fmt.Sprintf(`
+		   -a always,exit -F arch=b64 -F ppid=%d -S execve -k exec
+		`, os.Getpid()),
+	}
+
+	// Any commands executed by this process will generate events due to the
+	// PPID filter we applied to the rule.
+	time.AfterFunc(time.Second, func() { exec.Command("cat", "/proc/self/status").Output() })
+
+	ms := mbtest.NewPushMetricSet(t, c)
+	events, errs := mbtest.RunPushMetricSet(5*time.Second, ms)
+	if len(errs) > 0 {
+		t.Fatalf("received errors: %+v", errs)
+	}
+
+	t.Log(events)
+	assert.Len(t, events, 1)
+}
+
+func TestKernelVersion(t *testing.T) {
+	major, minor, full, err := kernelVersion()
+	if err != nil {
+		t.Fatal(err)
 	}
+	t.Logf("major=%v, minor=%v, full=%v", major, minor, full)
 }

auditbeat/module/audit/kernel/config.go

@@ -22,6 +22,7 @@ type Config struct {
 	RawMessage   bool   `config:"kernel.include_raw_message"` // Include the list of raw audit messages in the event.
 	Warnings     bool   `config:"kernel.include_warnings"`    // Include warnings in the event (for dev/debug purposes only).
 	RulesBlob    string `config:"kernel.audit_rules"`         // Audit rules. One rule per line.
+	SocketType   string `config:"kernel.socket_type"`         // Socket type to use with the kernel (unicast or multicast).
 
 	// Tuning options (advanced, use with care)
 	ReassemblerMaxInFlight uint32        `config:"kernel.reassembler.max_in_flight"`
@@ -35,7 +36,7 @@ type auditRule struct {
 }
 
 // Validate validates the rules specified in the config.
-func (c Config) Validate() error {
+func (c *Config) Validate() error {
 	var errs multierror.Errors
 	_, err := c.rules()
 	if err != nil {
@@ -45,6 +46,15 @@ func (c Config) Validate() error {
 	if err != nil {
 		errs = append(errs, err)
 	}
+
+	c.SocketType = strings.ToLower(c.SocketType)
+	switch c.SocketType {
+	case "", "unicast", "multicast":
+	default:
+		errs = append(errs, errors.Errorf("invalid kernel.socket_type "+
+			"'%v' (use unicast, multicast, or don't set a value)", c.SocketType))
+	}
+
 	return errs.Err()
 }
 

auditbeat/module/audit/kernel/config_linux_test.go

@@ -65,6 +65,14 @@ func TestConfigValidateFailureMode(t *testing.T) {
 	t.Log(err)
 }
 
+func TestConfigValidateConnectionType(t *testing.T) {
+	config := defaultConfig
+	config.SocketType = "Satellite"
+	err := config.Validate()
+	assert.Error(t, err)
+	t.Log(err)
+}
+
 func parseConfig(t testing.TB, yaml string) (Config, error) {
 	c, err := common.NewConfigWithYAML([]byte(yaml), "")
 	if err != nil {

metricbeat/mb/testing/modules.go

@@ -74,7 +74,7 @@ func newMetricSet(t testing.TB, config interface{}) mb.MetricSet {
 	}
 	m, metricsets, err := mb.NewModule(c, mb.Registry)
 	if err != nil {
-		t.Fatal(err)
+		t.Fatal("failed to create new MetricSet", err)
 	}
 	if m == nil {
 		t.Fatal("no module instantiated")