feat(network): add traceroute probes

traceroute/fuzz_linux_test.go

@@ -0,0 +1,33 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+//go:build linux
+
+package traceroute
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"golang.org/x/net/ipv4"
+)
+
+func FuzzNetworkReplyParsers(f *testing.F) {
+	f.Add([]byte{})
+	f.Add(make([]byte, ipv4.HeaderLen))
+	f.Add(make([]byte, 128))
+	target := net.ParseIP("203.0.113.10").To4()
+	local := net.ParseIP("192.0.2.20").To4()
+	peer := &net.IPAddr{IP: net.ParseIP("192.0.2.1")}
+
+	f.Fuzz(func(_ *testing.T, packet []byte) {
+		parseUDPICMPReply(packet, peer, target, 31000, 33434, 9, time.Millisecond)
+		parseTCPICMPReply(packet, peer, target, 31000, 443, 0x12345678, time.Millisecond)
+		matchesICMPProbe(packet, peer.IP, target, 17, 23)
+		matchesTCPResponse(&ipv4.Header{Src: target, Dst: local}, packet,
+			local, target, 31000, 443, 0x12345678)
+	})
+}

traceroute/icmp_unix.go

@@ -0,0 +1,180 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+//go:build !windows
+
+package traceroute
+
+import (
+	"context"
+	"errors"
+	"net"
+	"sync/atomic"
+	"time"
+
+	"golang.org/x/net/icmp"
+	"golang.org/x/net/ipv4"
+)
+
+var nextICMPID = uint32(time.Now().UnixNano()) //nolint:gochecknoglobals,gosec
+
+func traceICMP(ctx context.Context, target net.IP, cfg options) (Result, error) {
+	conn, err := icmp.ListenPacket("ip4:icmp", "0.0.0.0")
+	if err != nil {
+		return Result{}, err
+	}
+	defer conn.Close() //nolint:errcheck
+
+	result := Result{Routes: make([]*Route, 0, cfg.maxTTL)}
+	buffer := make([]byte, 2048)
+	for ttl := 1; ttl <= cfg.maxTTL; ttl++ {
+		replies := make([]probeReply, 0, cfg.attempts)
+		hopReached := false
+		for attempt := 0; attempt < cfg.attempts; attempt++ {
+			if err := ctx.Err(); err != nil {
+				return result, err
+			}
+			token := atomic.AddUint32(&nextICMPID, 1)
+			id := int(uint16(token >> 16))
+			sequence := int(uint16(token))
+			reply, err := sendICMPProbe(ctx, conn, buffer, target, ttl, id, sequence, cfg.timeout)
+			if err != nil {
+				if len(replies) > 0 {
+					result.Routes = append(result.Routes, repliesToRoute(replies))
+				}
+				return result, err
+			}
+			replies = append(replies, reply)
+			hopReached = hopReached || reply.reached
+		}
+		result.Routes = append(result.Routes, repliesToRoute(replies))
+		if hopReached {
+			result.Reached = true
+			return result, nil
+		}
+	}
+	return result, nil
+}
+
+func sendICMPProbe(ctx context.Context, conn *icmp.PacketConn, buffer []byte, target net.IP,
+	ttl, id, sequence int, timeout time.Duration,
+) (probeReply, error) {
+	if err := conn.IPv4PacketConn().SetTTL(ttl); err != nil {
+		return probeReply{}, err
+	}
+	message, err := (&icmp.Message{
+		Type: ipv4.ICMPTypeEcho,
+		Body: &icmp.Echo{ID: id, Seq: sequence},
+	}).Marshal(nil)
+	if err != nil {
+		return probeReply{}, err
+	}
+	startedAt := time.Now()
+	if _, err := conn.WriteTo(message, &net.IPAddr{IP: target}); err != nil {
+		return probeReply{}, err
+	}
+	deadline := probeDeadline(ctx, startedAt, timeout)
+	for {
+		if err := conn.SetReadDeadline(probeReadDeadline(ctx, deadline)); err != nil {
+			return probeReply{}, err
+		}
+		n, peer, err := conn.ReadFrom(buffer)
+		if err != nil {
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				return probeReply{}, ctxErr
+			}
+			if errors.Is(err, net.ErrClosed) {
+				return probeReply{}, err
+			}
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+				if time.Now().Before(deadline) {
+					continue
+				}
+				return probeReply{timedOut: true}, nil
+			}
+			return probeReply{}, err
+		}
+		from, ok := ipFromAddr(peer)
+		if !ok || !matchesICMPProbe(buffer[:n], from, target, id, sequence) {
+			continue
+		}
+		return probeReply{
+			ip:      from,
+			rtt:     time.Since(startedAt),
+			reached: from.Equal(target),
+		}, nil
+	}
+}
+
+func matchesICMPProbe(packet []byte, from, target net.IP, id, sequence int) bool {
+	message, err := icmp.ParseMessage(1, packet)
+	if err != nil {
+		return false
+	}
+	if message.Type == ipv4.ICMPTypeEchoReply {
+		echo, ok := message.Body.(*icmp.Echo)
+		return ok && from.Equal(target) && echo.ID == id && echo.Seq == sequence
+	}
+	quoted := quotedDatagram(message)
+	if len(quoted) < ipv4.HeaderLen || quoted[0]>>4 != ipv4.Version {
+		return false
+	}
+	header, err := ipv4.ParseHeader(quoted)
+	if err != nil || header.Protocol != 1 || len(quoted) < header.Len+8 || !header.Dst.Equal(target) {
+		return false
+	}
+	inner, err := icmp.ParseMessage(1, quoted[header.Len:])
+	if err != nil {
+		return false
+	}
+	echo, ok := inner.Body.(*icmp.Echo)
+	return ok && echo.ID == id && echo.Seq == sequence
+}
+
+func quotedDatagram(message *icmp.Message) []byte {
+	switch body := message.Body.(type) {
+	case *icmp.TimeExceeded:
+		return body.Data
+	case *icmp.DstUnreach:
+		return body.Data
+	case *icmp.ParamProb:
+		return body.Data
+	default:
+		return nil
+	}
+}
+
+func ipFromAddr(addr net.Addr) (net.IP, bool) {
+	switch value := addr.(type) {
+	case *net.IPAddr:
+		if value != nil && value.IP != nil {
+			return value.IP, true
+		}
+	case *net.UDPAddr:
+		if value != nil && value.IP != nil {
+			return value.IP, true
+		}
+	}
+	return nil, false
+}
+
+func probeDeadline(ctx context.Context, startedAt time.Time, timeout time.Duration) time.Time {
+	deadline := startedAt.Add(timeout)
+	if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(deadline) {
+		return ctxDeadline
+	}
+	return deadline
+}
+
+func probeReadDeadline(ctx context.Context, deadline time.Time) time.Time {
+	if ctx.Done() == nil {
+		return deadline
+	}
+	pollDeadline := time.Now().Add(100 * time.Millisecond)
+	if pollDeadline.Before(deadline) {
+		return pollDeadline
+	}
+	return deadline
+}

traceroute/integration_linux_test.go

@@ -0,0 +1,175 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+//go:build linux
+
+package traceroute
+
+import (
+	"context"
+	"errors"
+	"net"
+	"os"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTCPTraceIntegration(t *testing.T) {
+	if os.Getenv("CLIUTILS_TRACEROUTE_INTEGRATION") == "" {
+		t.Skip("set CLIUTILS_TRACEROUTE_INTEGRATION=1 to run raw socket integration tests")
+	}
+	target := net.ParseIP("127.0.0.1").To4()
+	listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: target})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = listener.Close() })
+	port := listener.Addr().(*net.TCPAddr).Port
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	result, err := Trace(ctx, target, Options{
+		Protocol: ProtocolTCP,
+		Port:     uint16(port),
+		MaxTTL:   1,
+		Attempts: 1,
+		Timeout:  time.Second,
+	})
+	require.NoError(t, err)
+	require.Len(t, result.Routes, 1)
+	require.Len(t, result.Routes[0].Items, 1)
+	assert.True(t, result.Reached)
+	assert.Equal(t, target.String(), result.Routes[0].Items[0].IP)
+}
+
+func TestTCPTraceRemoteHopIntegration(t *testing.T) {
+	address := os.Getenv("CLIUTILS_TRACEROUTE_REMOTE")
+	if address == "" {
+		t.Skip("set CLIUTILS_TRACEROUTE_REMOTE=IPv4:port to test a live intermediate hop")
+	}
+	host, portText, err := net.SplitHostPort(address)
+	require.NoError(t, err)
+	target := net.ParseIP(host).To4()
+	require.NotNil(t, target)
+	port, err := strconv.ParseUint(portText, 10, 16)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	result, err := Trace(ctx, target, Options{
+		Protocol: ProtocolTCP,
+		Port:     uint16(port),
+		MaxTTL:   2,
+		Attempts: 1,
+		Timeout:  2 * time.Second,
+	})
+	require.NoError(t, err)
+	require.NotEmpty(t, result.Routes)
+	require.NotEmpty(t, result.Routes[0].Items)
+	assert.NotEqual(t, "*", result.Routes[0].Items[0].IP)
+}
+
+func TestTCPTraceConcurrentIntegration(t *testing.T) {
+	if os.Getenv("CLIUTILS_TRACEROUTE_INTEGRATION") == "" {
+		t.Skip("set CLIUTILS_TRACEROUTE_INTEGRATION=1 to run raw socket integration tests")
+	}
+
+	target := net.ParseIP("127.0.0.1").To4()
+	listener, err := net.ListenTCP("tcp4", &net.TCPAddr{IP: target})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = listener.Close() })
+	port := uint16(listener.Addr().(*net.TCPAddr).Port)
+
+	const workers = 64
+	start := make(chan struct{})
+	errorsCh := make(chan error, workers)
+	var waitGroup sync.WaitGroup
+	for range workers {
+		waitGroup.Add(1)
+		go func() {
+			defer waitGroup.Done()
+			<-start
+			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+			defer cancel()
+			result, err := Trace(ctx, target, Options{
+				Protocol: ProtocolTCP,
+				Port:     port,
+				MaxTTL:   1,
+				Attempts: 1,
+				Timeout:  time.Second,
+			})
+			if err != nil {
+				errorsCh <- err
+				return
+			}
+			if !result.Reached {
+				errorsCh <- errors.New("TCP traceroute did not reach local listener")
+			}
+		}()
+	}
+	close(start)
+	waitGroup.Wait()
+	close(errorsCh)
+	for err := range errorsCh {
+		t.Error(err)
+	}
+}
+
+func TestUDPTraceLateApplicationResponseIntegration(t *testing.T) {
+	if os.Getenv("CLIUTILS_TRACEROUTE_INTEGRATION") == "" {
+		t.Skip("set CLIUTILS_TRACEROUTE_INTEGRATION=1 to run raw socket integration tests")
+	}
+
+	target := net.ParseIP("127.0.0.1").To4()
+	server, err := net.ListenUDP("udp4", &net.UDPAddr{IP: target})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = server.Close() })
+	require.NoError(t, server.SetReadDeadline(time.Now().Add(2*time.Second)))
+	serverPort := uint16(server.LocalAddr().(*net.UDPAddr).Port)
+
+	peers := make(chan *net.UDPAddr, 2)
+	serverErrors := make(chan error, 2)
+	go func() {
+		var buffer [MaxUDPProbePayload]byte
+		for probe := 0; probe < 2; probe++ {
+			_, peer, err := server.ReadFromUDP(buffer[:])
+			if err != nil {
+				serverErrors <- err
+				return
+			}
+			peer = &net.UDPAddr{IP: append(net.IP(nil), peer.IP...), Port: peer.Port, Zone: peer.Zone}
+			peers <- peer
+			if probe == 0 {
+				go func() {
+					time.Sleep(220 * time.Millisecond)
+					_, err := server.WriteToUDP([]byte("late"), peer)
+					serverErrors <- err
+				}()
+			}
+		}
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	result, err := Trace(ctx, target, Options{
+		Protocol: ProtocolUDP,
+		Port:     serverPort,
+		MaxTTL:   1,
+		Attempts: 2,
+		Timeout:  150 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	require.Len(t, result.Routes, 1)
+	assert.False(t, result.Reached)
+	assert.Equal(t, 2, result.Routes[0].Failed)
+
+	firstPeer := <-peers
+	secondPeer := <-peers
+	assert.NotEqual(t, firstPeer.Port, secondPeer.Port)
+	require.NoError(t, <-serverErrors)
+}