Adds windows collection of failed conns. (#26617)

Author: derekwbrown

pkg/network/driver/ddnpmapi.h

@@ -241,7 +241,9 @@ typedef enum _ConnectionStatus {
     CONN_STAT_ATTEMPTED,
     CONN_STAT_ESTABLISHED,
     CONN_STAT_ACKRST,
-    CONN_STAT_TIMEOUT
+    CONN_STAT_TIMEOUT,
+    CONN_STAT_EST_SENT_RST,
+    CONN_STAT_EST_RECV_RST
 } CONNECTION_STATUS;
 
 typedef struct _tcpFlowData {

pkg/network/driver/types.go

@@ -68,6 +68,18 @@ const (
 	TCPFlowEstablishedMask = C.TCP_FLOW_ESTABLISHED_MASK
 )
 
+type ConnectionStatus C.enum__ConnectionStatus
+
+const (
+	ConnectionStatusUnknown     ConnectionStatus = C.CONN_STAT_UNKNOWN
+	ConnectionStatusAttempted   ConnectionStatus = C.CONN_STAT_ATTEMPTED
+	ConnectionStatusEstablished ConnectionStatus = C.CONN_STAT_ESTABLISHED
+	ConnectionStatusACKRST      ConnectionStatus = C.CONN_STAT_ACKRST
+	ConnectionStatusTimeout     ConnectionStatus = C.CONN_STAT_TIMEOUT
+	ConnectionStatusSentRst     ConnectionStatus = C.CONN_STAT_EST_SENT_RST
+	ConnectionStatusRecvRst     ConnectionStatus = C.CONN_STAT_EST_RECV_RST
+)
+
 const (
 	DirectionInbound  = C.DIRECTION_INBOUND
 	DirectionOutbound = C.DIRECTION_OUTBOUND

pkg/network/driver/types_windows.go

@@ -119,9 +119,22 @@ func FlowToConnStat(cs *ConnectionStats, flow *driver.PerFlowData, enableMonoton
 
 		tf := flow.TCPFlow()
 		if tf != nil {
+			cs.TCPFailures = make(map[uint32]uint32)
 			cs.Monotonic.Retransmits = uint32(tf.RetransmitCount)
 			cs.RTT = uint32(tf.SRTT)
 			cs.RTTVar = uint32(tf.RttVariance)
+
+			switch driver.ConnectionStatus(tf.ConnectionStatus) {
+			case driver.ConnectionStatusACKRST:
+				cs.TCPFailures[111] = 1 // ECONNREFUSED in posix is 111
+			case driver.ConnectionStatusTimeout:
+				cs.TCPFailures[110] = 1 // ETIMEDOUT in posix is 110
+			case driver.ConnectionStatusSentRst:
+				fallthrough
+			case driver.ConnectionStatusRecvRst:
+				cs.TCPFailures[104] = 1 // ECONNRESET in posix is 104
+			}
+
 		}
 
 		if isTCPFlowEstablished(flow) {

pkg/network/event_windows.go

@@ -2368,15 +2368,21 @@ var failedConnectionsBuildModes = map[ebpftest.BuildMode]struct{}{
 	ebpftest.RuntimeCompiled: {},
 }
 
-func (s *TracerSuite) TestTCPFailureConnectionTimeout() {
-	t := s.T()
+func checkSkipFailureConnectionsTests(t *testing.T) {
 	if _, ok := failedConnectionsBuildModes[ebpftest.GetBuildMode()]; !ok {
 		t.Skip("Skipping test on unsupported build mode: ", ebpftest.GetBuildMode())
 	}
+
+}
+func (s *TracerSuite) TestTCPFailureConnectionTimeout() {
+	t := s.T()
+
+	checkSkipFailureConnectionsTests(t)
 	// TODO: remove this check when we fix this test on kernels < 4.19
 	if kv < kernel.VersionCode(4, 19, 0) {
 		t.Skip("Skipping test on kernels < 4.19")
 	}
+
 	setupDropTrafficRule(t)
 	cfg := testConfig()
 	cfg.TCPFailedConnectionsEnabled = true
@@ -2427,109 +2433,6 @@ func (s *TracerSuite) TestTCPFailureConnectionTimeout() {
 	}, 3*time.Second, 1000*time.Millisecond, "Failed connection not recorded properly")
 }
 
-func (s *TracerSuite) TestTCPFailureConnectionRefused() {
-	t := s.T()
-	if _, ok := failedConnectionsBuildModes[ebpftest.GetBuildMode()]; !ok {
-		t.Skip("Skipping test on unsupported build mode: ", ebpftest.GetBuildMode())
-	}
-	cfg := testConfig()
-	cfg.TCPFailedConnectionsEnabled = true
-	tr := setupTracer(t, cfg)
-
-	// try to connect to a port where no server is accepting connections
-	srvAddr := "127.0.0.1:9998"
-	conn, err := net.Dial("tcp", srvAddr)
-	if err == nil {
-		conn.Close() // If the connection unexpectedly succeeds, close it immediately.
-		require.Fail(t, "expected connection to be refused, but it succeeded")
-	}
-	require.Error(t, err, "expected connection refused error but got none")
-
-	// Check if the connection was recorded as refused
-	require.Eventually(t, func() bool {
-		conns := getConnections(t, tr)
-		// Check for the refusal record
-		return findFailedConnectionByRemoteAddr(srvAddr, conns, 111)
-	}, 3*time.Second, 100*time.Millisecond, "Failed connection not recorded properly")
-}
-
-func (s *TracerSuite) TestTCPFailureConnectionReset() {
-	t := s.T()
-	if _, ok := failedConnectionsBuildModes[ebpftest.GetBuildMode()]; !ok {
-		t.Skip("Skipping test on unsupported build mode: ", ebpftest.GetBuildMode())
-	}
-	cfg := testConfig()
-	cfg.TCPFailedConnectionsEnabled = true
-	tr := setupTracer(t, cfg)
-
-	srv := NewTCPServer(func(c net.Conn) {
-		if tcpConn, ok := c.(*net.TCPConn); ok {
-			tcpConn.SetLinger(0)
-			buf := make([]byte, 10)
-			_, _ = c.Read(buf)
-			time.Sleep(10 * time.Millisecond)
-		}
-		c.Close()
-	})
-
-	require.NoError(t, srv.Run(), "error running server")
-	t.Cleanup(srv.Shutdown)
-
-	serverAddr := srv.ln.Addr()
-	c, err := net.Dial("tcp", serverAddr.String())
-	require.NoError(t, err, "could not connect to server: ", err)
-
-	// Write to the server and expect a reset
-	_, writeErr := c.Write([]byte("ping"))
-	if writeErr != nil {
-		t.Log("Write error:", writeErr)
-	}
-
-	// Read from server to ensure that the server has a chance to reset the connection
-	_, readErr := c.Read(make([]byte, 4))
-	require.Error(t, readErr, "expected connection reset error but got none")
-
-	// Check if the connection was recorded as reset
-	require.Eventually(t, func() bool {
-		conns := getConnections(t, tr)
-		// 104 is the errno for ECONNRESET
-		return findFailedConnection(t, c.LocalAddr().String(), serverAddr.String(), conns, 104)
-	}, 3*time.Second, 100*time.Millisecond, "Failed connection not recorded properly")
-
-	require.NoError(t, c.Close(), "error closing client connection")
-}
-
-// findFailedConnection is a utility function to find a failed connection based on specific TCP error codes
-func findFailedConnection(t *testing.T, local, remote string, conns *network.Connections, errorCode uint32) bool {
-	// Extract the address and port from the net.Addr types
-	localAddrPort, err := netip.ParseAddrPort(local)
-	if err != nil {
-		t.Logf("Failed to parse local address: %v", err)
-		return false
-	}
-	remoteAddrPort, err := netip.ParseAddrPort(remote)
-	if err != nil {
-		t.Logf("Failed to parse remote address: %v", err)
-		return false
-	}
-
-	failureFilter := func(cs network.ConnectionStats) bool {
-		localMatch := netip.AddrPortFrom(cs.Source.Addr, cs.SPort) == localAddrPort
-		remoteMatch := netip.AddrPortFrom(cs.Dest.Addr, cs.DPort) == remoteAddrPort
-		return localMatch && remoteMatch && cs.TCPFailures[errorCode] > 0
-	}
-
-	return network.FirstConnection(conns, failureFilter) != nil
-}
-
-// for some failed connections we don't know the local addr/port so we need to search by remote addr only
-func findFailedConnectionByRemoteAddr(remoteAddr string, conns *network.Connections, errorCode uint32) bool {
-	failureFilter := func(cs network.ConnectionStats) bool {
-		return netip.MustParseAddrPort(remoteAddr) == netip.AddrPortFrom(cs.Dest.Addr, cs.DPort) && cs.TCPFailures[errorCode] > 0
-	}
-	return network.FirstConnection(conns, failureFilter) != nil
-}
-
 func setupDropTrafficRule(tb testing.TB) (ns string) {
 	state := testutil.IptablesSave(tb)
 	tb.Cleanup(func() {

pkg/network/tracer/tracer_linux_test.go

@@ -18,6 +18,7 @@ import (
 	"math/rand"
 	"net"
 	nethttp "net/http"
+	"net/netip"
 	"os"
 	"runtime"
 	"strconv"
@@ -1253,3 +1254,106 @@ func (s *TracerSuite) TestTCPDirection() {
 	conn = incomingConns[0]
 	assert.Equal(t, conn.Direction, network.INCOMING, "connection direction must be incoming: %s", conn)
 }
+
+func (s *TracerSuite) TestTCPFailureConnectionRefused() {
+	t := s.T()
+
+	checkSkipFailureConnectionsTests(t)
+
+	cfg := testConfig()
+	cfg.TCPFailedConnectionsEnabled = true
+	tr := setupTracer(t, cfg)
+
+	// try to connect to a port where no server is accepting connections
+	srvAddr := "127.0.0.1:9998"
+	conn, err := net.Dial("tcp", srvAddr)
+	if err == nil {
+		conn.Close() // If the connection unexpectedly succeeds, close it immediately.
+		require.Fail(t, "expected connection to be refused, but it succeeded")
+	}
+	require.Error(t, err, "expected connection refused error but got none")
+
+	// Check if the connection was recorded as refused
+	require.Eventually(t, func() bool {
+		conns := getConnections(t, tr)
+		// Check for the refusal record
+		return findFailedConnectionByRemoteAddr(srvAddr, conns, 111)
+	}, 3*time.Second, 100*time.Millisecond, "Failed connection not recorded properly")
+}
+
+func (s *TracerSuite) TestTCPFailureConnectionReset() {
+	t := s.T()
+
+	checkSkipFailureConnectionsTests(t)
+
+	cfg := testConfig()
+	cfg.TCPFailedConnectionsEnabled = true
+	tr := setupTracer(t, cfg)
+
+	srv := NewTCPServer(func(c net.Conn) {
+		if tcpConn, ok := c.(*net.TCPConn); ok {
+			tcpConn.SetLinger(0)
+			buf := make([]byte, 10)
+			_, _ = c.Read(buf)
+			time.Sleep(10 * time.Millisecond)
+		}
+		c.Close()
+	})
+
+	require.NoError(t, srv.Run(), "error running server")
+	t.Cleanup(srv.Shutdown)
+
+	serverAddr := srv.ln.Addr()
+	c, err := net.Dial("tcp", serverAddr.String())
+	require.NoError(t, err, "could not connect to server: ", err)
+
+	// Write to the server and expect a reset
+	_, writeErr := c.Write([]byte("ping"))
+	if writeErr != nil {
+		t.Log("Write error:", writeErr)
+	}
+
+	// Read from server to ensure that the server has a chance to reset the connection
+	_, readErr := c.Read(make([]byte, 4))
+	require.Error(t, readErr, "expected connection reset error but got none")
+
+	// Check if the connection was recorded as reset
+	require.Eventually(t, func() bool {
+		conns := getConnections(t, tr)
+		// 104 is the errno for ECONNRESET
+		return findFailedConnection(t, c.LocalAddr().String(), serverAddr.String(), conns, 104)
+	}, 3*time.Second, 100*time.Millisecond, "Failed connection not recorded properly")
+
+	require.NoError(t, c.Close(), "error closing client connection")
+}
+
+// findFailedConnection is a utility function to find a failed connection based on specific TCP error codes
+func findFailedConnection(t *testing.T, local, remote string, conns *network.Connections, errorCode uint32) bool { // nolint:unused
+	// Extract the address and port from the net.Addr types
+	localAddrPort, err := netip.ParseAddrPort(local)
+	if err != nil {
+		t.Logf("Failed to parse local address: %v", err)
+		return false
+	}
+	remoteAddrPort, err := netip.ParseAddrPort(remote)
+	if err != nil {
+		t.Logf("Failed to parse remote address: %v", err)
+		return false
+	}
+
+	failureFilter := func(cs network.ConnectionStats) bool {
+		localMatch := netip.AddrPortFrom(cs.Source.Addr, cs.SPort) == localAddrPort
+		remoteMatch := netip.AddrPortFrom(cs.Dest.Addr, cs.DPort) == remoteAddrPort
+		return localMatch && remoteMatch && cs.TCPFailures[errorCode] > 0
+	}
+
+	return network.FirstConnection(conns, failureFilter) != nil
+}
+
+// for some failed connections we don't know the local addr/port so we need to search by remote addr only
+func findFailedConnectionByRemoteAddr(remoteAddr string, conns *network.Connections, errorCode uint32) bool {
+	failureFilter := func(cs network.ConnectionStats) bool {
+		return netip.MustParseAddrPort(remoteAddr) == netip.AddrPortFrom(cs.Dest.Addr, cs.DPort) && cs.TCPFailures[errorCode] > 0
+	}
+	return network.FirstConnection(conns, failureFilter) != nil
+}

pkg/network/tracer/tracer_test.go

@@ -8,6 +8,9 @@
 package tracer
 
 import (
+	"github.com/DataDog/datadog-agent/pkg/network/testutil"
+	"testing"
+
 	sysconfigtypes "github.com/DataDog/datadog-agent/cmd/system-probe/config/types"
 	"github.com/DataDog/datadog-agent/pkg/network/config"
 	"github.com/DataDog/datadog-agent/pkg/network/driver"
@@ -30,3 +33,24 @@ func testConfig() *config.Config {
 	cfg := config.New()
 	return cfg
 }
+
+// nolint:unused   // this function currently unused but will be.
+func setupDropTrafficRule(tb testing.TB) (ns string) {
+	//
+	// note.  This does not seem to function as advertised; localhost traffic is not being
+	// blocked.  More testing is necessary.
+	tb.Cleanup(func() {
+		cmds := []string{
+			"powershell -c \"Remove-NetFirewallRule -DisplayName 'Datadog Test Rule'\"",
+		}
+		testutil.RunCommands(tb, cmds, false)
+	})
+	cmds := []string{
+		"powershell -c \"New-NetFirewallRule -DisplayName 'Datadog Test Rule' -Direction Outbound -Action Block -Profile Any -RemotePort 10000 -Protocol TCP\"",
+	}
+	testutil.RunCommands(tb, cmds, false)
+	return
+}
+
+func checkSkipFailureConnectionsTests(_ *testing.T) {
+}

pkg/network/tracer/tracer_windows_test.go

@@ -13,8 +13,8 @@
         "JMXFETCH_HASH": "6c5d4ae2f858aa57fac7f92d6615931bf95f9a99c92979090f2de9d4b3c1f0d3",
         "MACOS_BUILD_VERSION": "master",
         "WINDOWS_DDNPM_DRIVER": "release-signed",
-        "WINDOWS_DDNPM_VERSION": "2.6.0",
-        "WINDOWS_DDNPM_SHASUM": "b1611ad4ceb8366c88767aeb638abefb226081efbf546b8b886952dd1b18ec05",
+        "WINDOWS_DDNPM_VERSION": "2.7.0",
+        "WINDOWS_DDNPM_SHASUM": "de6a2f437b906d1d0f3cfc9222c7f686b3d69726355c940476448a34535064c8",
         "SECURITY_AGENT_POLICIES_VERSION": "master",
         "WINDOWS_DDPROCMON_DRIVER": "release-signed",
         "WINDOWS_DDPROCMON_VERSION": "1.0.1",
@@ -32,8 +32,8 @@
         "JMXFETCH_HASH": "6c5d4ae2f858aa57fac7f92d6615931bf95f9a99c92979090f2de9d4b3c1f0d3",
         "MACOS_BUILD_VERSION": "master",
         "WINDOWS_DDNPM_DRIVER": "release-signed",
-        "WINDOWS_DDNPM_VERSION": "2.6.0",
-        "WINDOWS_DDNPM_SHASUM": "b1611ad4ceb8366c88767aeb638abefb226081efbf546b8b886952dd1b18ec05",
+        "WINDOWS_DDNPM_VERSION": "2.7.0",
+        "WINDOWS_DDNPM_SHASUM": "de6a2f437b906d1d0f3cfc9222c7f686b3d69726355c940476448a34535064c8",
         "SECURITY_AGENT_POLICIES_VERSION": "master",
         "WINDOWS_DDPROCMON_DRIVER": "release-signed",
         "WINDOWS_DDPROCMON_VERSION": "1.0.1",