[Auditbeat][Metricbeat] Cherry-pick elastic#12248 to 7.2: Fix directi…

…on of incoming IPv6 sockets (elastic#12249)

To determine the direction of a socket, we save the list of listening sockets and match non-listening sockets to them. If we find a match, the non-listening socket is `Incoming`, otherwise `Outgoing`.

A problem occurs when matching an IPv6 socket listening on all interfaces (`::`) with an IPv6 socket that has an IPv4-mapped IPv6 address (e.g. `::ffff:127.0.0.1`). Golang's `To4()` will determine it is an IPv4 address and miss the listening IPv6 socket.

With this PR, we specify the IP family explicitly instead of trying to determine it from the IP address.

Fixes elastic#3306.

(cherry picked from commit 01ca04b)

CHANGELOG.next.asciidoc

@@ -85,6 +85,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Login dataset: Fix re-read of utmp files. {pull}12028[12028]
 - Package dataset: Fixed a crash inside librpm after Auditbeat has been running for a while. {issue}12147[12147] {pull}12168[12168]
 - Fix formatting of config files on macOS and Windows. {pull}12148[12148]
+- Fix direction of incoming IPv6 sockets. {pull}12248[12248]
 
 *Filebeat*
 
@@ -123,6 +124,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fixed a socket leak in the postgresql module under Windows when SSL is disabled on the server. {pull}11393[11393]
 - Change some field type from scaled_float to long in aws module. {pull}11982[11982]
 - Fixed RabbitMQ `queue` metricset gathering when `consumer_utilisation` is set empty at the metrics source {pull}12089[12089] 
+- Fix direction of incoming IPv6 sockets. {pull}12248[12248]
 
 *Packetbeat*
 

metricbeat/helper/socket/listeners.go

@@ -19,6 +19,7 @@ package socket
 
 import (
 	"net"
+	"syscall"
 )
 
 // Direction indicates how a socket was initiated.
@@ -113,7 +114,7 @@ func (t *ListenerTable) Put(proto uint8, ip net.IP, port int) {
 // listeners in the table for the protocol and returns Inbound if there is a
 // match. If remotePort is 0 then Listening is returned.
 func (t *ListenerTable) Direction(
-	proto uint8,
+	family uint8, proto uint8,
 	localIP net.IP, localPort int,
 	remoteIP net.IP, remotePort int,
 ) Direction {
@@ -135,14 +136,13 @@ func (t *ListenerTable) Direction(
 
 	// Is there a listener that specific interface? OR
 	// Is there a listener on the "any" address (0.0.0.0 or ::)?
-	isIPv4 := localIP.To4() != nil
 	for _, ip := range interfaces.ips {
 		switch {
 		case ip.Equal(localIP):
 			return Inbound
-		case ip.Equal(net.IPv4zero) && isIPv4:
+		case family == syscall.AF_INET && ip.Equal(net.IPv4zero):
 			return Inbound
-		case ip.Equal(net.IPv6zero) && !isIPv4:
+		case family == syscall.AF_INET6 && ip.Equal(net.IPv6zero):
 			return Inbound
 		}
 	}

metricbeat/helper/socket/listeners_test.go

@@ -19,6 +19,7 @@ package socket
 
 import (
 	"net"
+	"syscall"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -33,28 +34,30 @@ func TestListenerTable(t *testing.T) {
 	rAddr := net.ParseIP("198.18.0.1")
 	ephemeralPort := 48199
 	ipv6Addr := net.ParseIP("2001:db8:fe80::217:f2ff:fe07:ed62")
+	ipv4InIpv6 := net.ParseIP("::ffff:127.0.0.1")
 
 	// Any socket with remote port of 0 is listening.
-	assert.Equal(t, Listening, l.Direction(proto, lAddr, httpPort, net.IPv4zero, 0))
+	assert.Equal(t, Listening, l.Direction(syscall.AF_INET, proto, lAddr, httpPort, net.IPv4zero, 0))
 
 	// Listener on 192.0.2.1:80
 	l.Put(proto, lAddr, httpPort)
 
-	assert.Equal(t, Inbound, l.Direction(proto, lAddr, httpPort, rAddr, ephemeralPort))
-	assert.Equal(t, Outbound, l.Direction(0, lAddr, httpPort, rAddr, ephemeralPort))
-	assert.Equal(t, Outbound, l.Direction(proto, lAddr, ephemeralPort, rAddr, ephemeralPort))
+	assert.Equal(t, Inbound, l.Direction(syscall.AF_INET, proto, lAddr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Outbound, l.Direction(syscall.AF_INET, 0, lAddr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Outbound, l.Direction(syscall.AF_INET, proto, lAddr, ephemeralPort, rAddr, ephemeralPort))
 
 	// Listener on 0.0.0.0:80
 	l.Reset()
 	l.Put(proto, net.IPv4zero, httpPort)
 
-	assert.Equal(t, Inbound, l.Direction(proto, lAddr, httpPort, rAddr, ephemeralPort))
-	assert.Equal(t, Outbound, l.Direction(proto, ipv6Addr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Inbound, l.Direction(syscall.AF_INET, proto, lAddr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Outbound, l.Direction(syscall.AF_INET6, proto, ipv6Addr, httpPort, rAddr, ephemeralPort))
 
 	// Listener on :::80
 	l.Reset()
 	l.Put(proto, net.IPv6zero, httpPort)
 
-	assert.Equal(t, Inbound, l.Direction(proto, ipv6Addr, httpPort, rAddr, ephemeralPort))
-	assert.Equal(t, Outbound, l.Direction(proto, lAddr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Inbound, l.Direction(syscall.AF_INET6, proto, ipv6Addr, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Inbound, l.Direction(syscall.AF_INET6, proto, ipv4InIpv6, httpPort, rAddr, ephemeralPort))
+	assert.Equal(t, Outbound, l.Direction(syscall.AF_INET, proto, lAddr, httpPort, rAddr, ephemeralPort))
 }

metricbeat/module/system/socket/socket.go

@@ -191,7 +191,7 @@ func (m *MetricSet) enrichConnectionData(c *connection) {
 	c.User = m.users.LookupUID(int(c.UID))
 
 	// Determine direction (incoming, outgoing, or listening).
-	c.Direction = m.listeners.Direction(uint8(syscall.IPPROTO_TCP),
+	c.Direction = m.listeners.Direction(uint8(c.Family), uint8(syscall.IPPROTO_TCP),
 		c.LocalIP, c.LocalPort, c.RemoteIP, c.RemotePort)
 
 	// Reverse DNS lookup on the remote IP.

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

@@ -364,7 +364,7 @@ func (ms *MetricSet) enrichSocket(socket *Socket) error {
 
 	socket.Username = userAccount.Username
 
-	socket.Direction = ms.listeners.Direction(uint8(syscall.IPPROTO_TCP),
+	socket.Direction = ms.listeners.Direction(uint8(socket.Family), uint8(syscall.IPPROTO_TCP),
 		socket.LocalIP, socket.LocalPort, socket.RemoteIP, socket.RemotePort)
 
 	if ms.ptable != nil {