[Auditbeat] Socket metricset

metricbeat/helper/socket/netlink.go

@@ -0,0 +1,54 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// +build linux
+
+package socket
+
+import (
+	"os"
+	"sync/atomic"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/gosigar/sys/linux"
+)
+
+// NetlinkSession communicates with the kernel's netlink subsystem.
+type NetlinkSession struct {
+	readBuffer []byte
+	seq        uint32
+}
+
+// NewNetlinkSession creates a new netlink session.
+func NewNetlinkSession() *NetlinkSession {
+	return &NetlinkSession{
+		readBuffer: make([]byte, os.Getpagesize()),
+	}
+}
+
+// GetSocketList retrieves the current list of sockets from the kernel.
+func (session *NetlinkSession) GetSocketList() ([]*linux.InetDiagMsg, error) {
+	// Send request over netlink and parse responses.
+	req := linux.NewInetDiagReq()
+	req.Header.Seq = atomic.AddUint32(&session.seq, 1)
+	sockets, err := linux.NetlinkInetDiagWithBuf(req, session.readBuffer, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed requesting socket dump")
+	}
+	return sockets, nil
+}

metricbeat/module/system/socket/socket.go

@@ -24,17 +24,17 @@ import (
 	"net"
 	"os"
 	"path/filepath"
-	"sync/atomic"
 	"syscall"
 
+	"github.com/pkg/errors"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
+	sock "github.com/elastic/beats/metricbeat/helper/socket"
 	"github.com/elastic/beats/metricbeat/mb"
 	"github.com/elastic/beats/metricbeat/mb/parse"
 	"github.com/elastic/beats/metricbeat/module/system"
 	"github.com/elastic/gosigar/sys/linux"
-
-	"github.com/pkg/errors"
 )
 
 var (
@@ -50,14 +50,13 @@ func init() {
 
 type MetricSet struct {
 	mb.BaseMetricSet
-	readBuffer    []byte
-	seq           uint32
-	ptable        *ProcTable
+	netlink       *sock.NetlinkSession
+	ptable        *sock.ProcTable
 	euid          int
 	previousConns hashSet
 	currentConns  hashSet
 	reverseLookup *ReverseLookupCache
-	listeners     *ListenerTable
+	listeners     *sock.ListenerTable
 	users         UserCache
 }
 
@@ -72,7 +71,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, errors.New("unexpected module type")
 	}
 
-	ptable, err := NewProcTable(filepath.Join(systemModule.HostFS, "/proc"))
+	ptable, err := sock.NewProcTable(filepath.Join(systemModule.HostFS, "/proc"))
 	if err != nil {
 		return nil, err
 	}
@@ -83,12 +82,12 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 	m := &MetricSet{
 		BaseMetricSet: base,
-		readBuffer:    make([]byte, os.Getpagesize()),
+		netlink:       sock.NewNetlinkSession(),
 		ptable:        ptable,
 		euid:          os.Geteuid(),
 		previousConns: hashSet{},
 		currentConns:  hashSet{},
-		listeners:     NewListenerTable(),
+		listeners:     sock.NewListenerTable(),
 		users:         NewUserCache(),
 	}
 
@@ -114,10 +113,7 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 		debugf("process table refresh had failures: %v", err)
 	}
 
-	// Send request over netlink and parse responses.
-	req := linux.NewInetDiagReq()
-	req.Header.Seq = atomic.AddUint32(&m.seq, 1)
-	sockets, err := linux.NetlinkInetDiagWithBuf(req, m.readBuffer, nil)
+	sockets, err := m.netlink.GetSocketList()
 	if err != nil {
 		return nil, errors.Wrap(err, "failed requesting socket dump")
 	}
@@ -192,7 +188,7 @@ func (m *MetricSet) enrichConnectionData(c *connection) {
 		c.LocalIP, c.LocalPort, c.RemoteIP, c.RemotePort)
 
 	// Reverse DNS lookup on the remote IP.
-	if m.reverseLookup != nil && c.Direction != Listening {
+	if m.reverseLookup != nil && c.Direction != sock.Listening {
 		hostname, err := m.reverseLookup.Lookup(c.RemoteIP)
 		if err != nil {
 			c.DestHostError = err
@@ -227,7 +223,7 @@ type connection struct {
 	RemotePort int
 
 	State     linux.TCPState
-	Direction Direction
+	Direction sock.Direction
 
 	DestHost            string // Reverse lookup of dest IP.
 	DestHostETLDPlusOne string

x-pack/auditbeat/include/list.go

@@ -10,5 +10,6 @@ import (
 	_ "github.com/elastic/beats/x-pack/auditbeat/module/system/host"
 	_ "github.com/elastic/beats/x-pack/auditbeat/module/system/packages"
 	_ "github.com/elastic/beats/x-pack/auditbeat/module/system/processes"
+	_ "github.com/elastic/beats/x-pack/auditbeat/module/system/socket"
 	_ "github.com/elastic/beats/x-pack/auditbeat/module/system/user"
 )

x-pack/auditbeat/module/system/_meta/config.yml.tmpl

@@ -1,22 +1,20 @@
 {{ if .Reference -}}
 {{ end -}}
+{{ if ne .GOOS "windows" -}}
 - module: system
 
   metricsets:
     - host
     - packages
     - processes
     {{ if eq .GOOS "linux" -}}
+    - socket
     - user
     {{- end }}
 
   state.period: 12h
 
   report_changes: true
-
-  {{ if eq .GOOS "darwin" -}}
-  {{ else if eq .GOOS "windows" -}}
-  {{ else -}}
-  {{- end }}
+{{- end }}
 {{ if .Reference }}
 {{- end }}

x-pack/auditbeat/module/system/socket/_meta/data.json

@@ -0,0 +1,33 @@
+{
+    "@timestamp": "2017-10-12T08:05:34.853Z",
+    "agent": {
+        "hostname": "host.example.com",
+        "name": "host.example.com"
+    },
+    "user": {
+        "name": "vagrant",
+        "id": 1000
+    },
+    "process": {
+        "pid": 4021,
+        "name": "lynx"
+    },
+    "source": {
+        "ip": "10.0.2.15",
+        "port": 33956
+    },
+    "destination": {
+        "port": 443,
+        "ip": "52.10.168.186"
+    },
+    "event": {
+        "type": "event",
+        "action": "socket_opened",
+        "module": "system",
+        "dataset": "socket"
+    },
+    "network": {
+        "type": "ipv4",
+        "direction": "outbound"
+    }
+}

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

@@ -0,0 +1,8 @@
+The System `socket` metricset provides ... TODO.
+
+The module is implemented for Linux only.
+
+[float]
+=== Configuration options
+
+TODO

x-pack/auditbeat/module/system/socket/config.go

@@ -0,0 +1,31 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package socket
+
+import (
+	"time"
+)
+
+// Config defines the socket metricset's configuration options.
+type Config struct {
+	StatePeriod       time.Duration `config:"state.period"`
+	SocketStatePeriod time.Duration `config:"socket.state.period"`
+}
+
+// Validate validates the host metricset config.
+func (c *Config) Validate() error {
+	return nil
+}
+
+func (c *Config) effectiveStatePeriod() time.Duration {
+	if c.SocketStatePeriod != 0 {
+		return c.SocketStatePeriod
+	}
+	return c.StatePeriod
+}
+
+var defaultConfig = Config{
+	StatePeriod: 1 * time.Hour,
+}