TUN-8640: Add ICMP support for datagram V3

Closes TUN-8640

Author: DevinCarr

connection/quic_connection_test.go

@@ -752,7 +752,8 @@ func testTunnelConnection(t *testing.T, serverAddr netip.AddrPort, index uint8)
 	sessionDemuxChan := make(chan *packet.Session, 4)
 	datagramMuxer := cfdquic.NewDatagramMuxerV2(conn, &log, sessionDemuxChan)
 	sessionManager := datagramsession.NewManager(&log, datagramMuxer.SendToSession, sessionDemuxChan)
-	packetRouter := ingress.NewPacketRouter(nil, datagramMuxer, 0, &log)
+	var connIndex uint8 = 0
+	packetRouter := ingress.NewPacketRouter(nil, datagramMuxer, connIndex, &log)
 
 	datagramConn := &datagramV2Connection{
 		conn,

connection/quic_datagram_v3.go

@@ -10,6 +10,7 @@ import (
 	"github.com/quic-go/quic-go"
 	"github.com/rs/zerolog"
 
+	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/management"
 	cfdquic "github.com/cloudflare/cloudflared/quic/v3"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -25,6 +26,7 @@ type datagramV3Connection struct {
 func NewDatagramV3Connection(ctx context.Context,
 	conn quic.Connection,
 	sessionManager cfdquic.SessionManager,
+	icmpRouter ingress.ICMPRouter,
 	index uint8,
 	metrics cfdquic.Metrics,
 	logger *zerolog.Logger,
@@ -34,7 +36,7 @@ func NewDatagramV3Connection(ctx context.Context,
 		Int(management.EventTypeKey, int(management.UDP)).
 		Uint8(LogFieldConnIndex, index).
 		Logger()
-	datagramMuxer := cfdquic.NewDatagramConn(conn, sessionManager, index, metrics, &log)
+	datagramMuxer := cfdquic.NewDatagramConn(conn, sessionManager, icmpRouter, index, metrics, &log)
 
 	return &datagramV3Connection{
 		conn,

quic/v3/datagram.go

@@ -222,11 +222,11 @@ const (
 //
 // This method should be used in-place of MarshalBinary which will allocate in-place the required byte array to return.
 func MarshalPayloadHeaderTo(requestID RequestID, payload []byte) error {
-	if len(payload) < 17 {
+	if len(payload) < DatagramPayloadHeaderLen {
 		return wrapMarshalErr(ErrDatagramPayloadHeaderTooSmall)
 	}
 	payload[0] = byte(UDPSessionPayloadType)
-	return requestID.MarshalBinaryTo(payload[1:17])
+	return requestID.MarshalBinaryTo(payload[1:DatagramPayloadHeaderLen])
 }
 
 func (s *UDPSessionPayloadDatagram) UnmarshalBinary(data []byte) error {
@@ -239,18 +239,18 @@ func (s *UDPSessionPayloadDatagram) UnmarshalBinary(data []byte) error {
 	}
 
 	// Make sure that the slice provided is the right size to be parsed.
-	if len(data) < 17 || len(data) > maxPayloadPlusHeaderLen {
+	if len(data) < DatagramPayloadHeaderLen || len(data) > maxPayloadPlusHeaderLen {
 		return wrapUnmarshalErr(ErrDatagramPayloadInvalidSize)
 	}
 
-	requestID, err := RequestIDFromSlice(data[1:17])
+	requestID, err := RequestIDFromSlice(data[1:DatagramPayloadHeaderLen])
 	if err != nil {
 		return wrapUnmarshalErr(err)
 	}
 
 	*s = UDPSessionPayloadDatagram{
 		RequestID: requestID,
-		Payload:   data[17:],
+		Payload:   data[DatagramPayloadHeaderLen:],
 	}
 	return nil
 }
@@ -370,3 +370,61 @@ func (s *UDPSessionRegistrationResponseDatagram) UnmarshalBinary(data []byte) er
 	}
 	return nil
 }
+
+// ICMPDatagram is used to propagate ICMPv4 and ICMPv6 payloads.
+type ICMPDatagram struct {
+	Payload []byte
+}
+
+// The maximum size that an ICMP packet can be.
+const maxICMPPayloadLen = maxDatagramPayloadLen
+
+// The datagram structure for ICMPDatagram is:
+//
+//   0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7
+//   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+//  0|      Type     |                                               |
+//   +-+-+-+-+-+-+-+-+                                               +
+//   .                          Payload                              .
+//   .                                                               .
+//   .                                                               .
+//   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+
+func (d *ICMPDatagram) MarshalBinary() (data []byte, err error) {
+	if len(d.Payload) > maxICMPPayloadLen {
+		return nil, wrapMarshalErr(ErrDatagramICMPPayloadTooLarge)
+	}
+	// We shouldn't attempt to marshal an ICMP datagram with no ICMP payload provided
+	if len(d.Payload) == 0 {
+		return nil, wrapMarshalErr(ErrDatagramICMPPayloadMissing)
+	}
+	// Make room for the 1 byte ICMPType header
+	datagram := make([]byte, len(d.Payload)+datagramTypeLen)
+	datagram[0] = byte(ICMPType)
+	copy(datagram[1:], d.Payload)
+	return datagram, nil
+}
+
+func (d *ICMPDatagram) UnmarshalBinary(data []byte) error {
+	datagramType, err := ParseDatagramType(data)
+	if err != nil {
+		return wrapUnmarshalErr(err)
+	}
+	if datagramType != ICMPType {
+		return wrapUnmarshalErr(ErrInvalidDatagramType)
+	}
+
+	if len(data[1:]) > maxDatagramPayloadLen {
+		return wrapUnmarshalErr(ErrDatagramICMPPayloadTooLarge)
+	}
+
+	// We shouldn't attempt to unmarshal an ICMP datagram with no ICMP payload provided
+	if len(data[1:]) == 0 {
+		return wrapUnmarshalErr(ErrDatagramICMPPayloadMissing)
+	}
+
+	payload := make([]byte, len(data[1:]))
+	copy(payload, data[1:])
+	d.Payload = payload
+	return nil
+}

quic/v3/datagram_errors.go

@@ -15,6 +15,8 @@ var (
 	ErrDatagramResponseInvalidSize         error = errors.New("datagram response is an invalid size")
 	ErrDatagramResponseMsgTooLargeMaximum  error = fmt.Errorf("datagram response error message length exceeds the length of the datagram maximum: %d", maxResponseErrorMessageLen)
 	ErrDatagramResponseMsgTooLargeDatagram error = fmt.Errorf("datagram response error message length exceeds the length of the provided datagram")
+	ErrDatagramICMPPayloadTooLarge         error = fmt.Errorf("datagram icmp payload exceeds %d bytes", maxICMPPayloadLen)
+	ErrDatagramICMPPayloadMissing          error = errors.New("datagram icmp payload is missing")
 )
 
 func wrapMarshalErr(err error) error {

quic/v3/datagram_test.go

@@ -160,6 +160,12 @@ func TestTypeUnmarshalErrors(t *testing.T) {
 		if !errors.Is(err, v3.ErrDatagramHeaderTooSmall) {
 			t.Errorf("expected invalid length to throw error")
 		}
+
+		d4 := v3.ICMPDatagram{}
+		err = d4.UnmarshalBinary([]byte{})
+		if !errors.Is(err, v3.ErrDatagramHeaderTooSmall) {
+			t.Errorf("expected invalid length to throw error")
+		}
 	})
 
 	t.Run("invalid types", func(t *testing.T) {
@@ -180,6 +186,12 @@ func TestTypeUnmarshalErrors(t *testing.T) {
 		if !errors.Is(err, v3.ErrInvalidDatagramType) {
 			t.Errorf("expected invalid type to throw error")
 		}
+
+		d4 := v3.ICMPDatagram{}
+		err = d4.UnmarshalBinary([]byte{byte(v3.UDPSessionPayloadType)})
+		if !errors.Is(err, v3.ErrInvalidDatagramType) {
+			t.Errorf("expected invalid type to throw error")
+		}
 	})
 }
 
@@ -343,6 +355,54 @@ func TestSessionRegistrationResponse(t *testing.T) {
 	})
 }
 
+func TestICMPDatagram(t *testing.T) {
+	t.Run("basic", func(t *testing.T) {
+		payload := makePayload(128)
+		datagram := v3.ICMPDatagram{Payload: payload}
+		marshaled, err := datagram.MarshalBinary()
+		if err != nil {
+			t.Error(err)
+		}
+		unmarshaled := &v3.ICMPDatagram{}
+		err = unmarshaled.UnmarshalBinary(marshaled)
+		if err != nil {
+			t.Error(err)
+		}
+		require.Equal(t, payload, unmarshaled.Payload)
+	})
+
+	t.Run("payload size empty", func(t *testing.T) {
+		payload := []byte{}
+		datagram := v3.ICMPDatagram{Payload: payload}
+		_, err := datagram.MarshalBinary()
+		if !errors.Is(err, v3.ErrDatagramICMPPayloadMissing) {
+			t.Errorf("expected an error: %s", err)
+		}
+		payload = []byte{byte(v3.ICMPType)}
+		unmarshaled := &v3.ICMPDatagram{}
+		err = unmarshaled.UnmarshalBinary(payload)
+		if !errors.Is(err, v3.ErrDatagramICMPPayloadMissing) {
+			t.Errorf("expected an error: %s", err)
+		}
+	})
+
+	t.Run("payload size too large", func(t *testing.T) {
+		payload := makePayload(1280 + 1) // larger than the datagram size could be
+		datagram := v3.ICMPDatagram{Payload: payload}
+		_, err := datagram.MarshalBinary()
+		if !errors.Is(err, v3.ErrDatagramICMPPayloadTooLarge) {
+			t.Errorf("expected an error: %s", err)
+		}
+		payload = makePayload(1280 + 2) // larger than the datagram size could be + header
+		payload[0] = byte(v3.ICMPType)
+		unmarshaled := &v3.ICMPDatagram{}
+		err = unmarshaled.UnmarshalBinary(payload)
+		if !errors.Is(err, v3.ErrDatagramICMPPayloadTooLarge) {
+			t.Errorf("expected an error: %s", err)
+		}
+	})
+}
+
 func compareRegistrationDatagrams(t *testing.T, l *v3.UDPSessionRegistrationDatagram, r *v3.UDPSessionRegistrationDatagram) bool {
 	require.Equal(t, l.Payload, r.Payload)
 	return l.RequestID == r.RequestID &&
@@ -377,3 +437,13 @@ func FuzzRegistrationResponseDatagram(f *testing.F) {
 		}
 	})
 }
+
+func FuzzICMPDatagram(f *testing.F) {
+	f.Fuzz(func(t *testing.T, data []byte) {
+		unmarshaled := v3.ICMPDatagram{}
+		err := unmarshaled.UnmarshalBinary(data)
+		if err == nil {
+			_, _ = unmarshaled.MarshalBinary()
+		}
+	})
+}

quic/v3/icmp.go

@@ -0,0 +1,52 @@
+package v3
+
+import (
+	"context"
+
+	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
+)
+
+// packetResponder is an implementation of the [ingress.ICMPResponder] which provides the ICMP Flow manager the
+// return path to return and ICMP Echo response back to the QUIC muxer.
+type packetResponder struct {
+	datagramMuxer DatagramICMPWriter
+	connID        uint8
+}
+
+func newPacketResponder(datagramMuxer DatagramICMPWriter, connID uint8) ingress.ICMPResponder {
+	return &packetResponder{
+		datagramMuxer,
+		connID,
+	}
+}
+
+func (pr *packetResponder) ConnectionIndex() uint8 {
+	return pr.connID
+}
+
+func (pr *packetResponder) ReturnPacket(pk *packet.ICMP) error {
+	return pr.datagramMuxer.SendICMPPacket(pk)
+}
+
+func (pr *packetResponder) AddTraceContext(tracedCtx *tracing.TracedContext, serializedIdentity []byte) {
+	// datagram v3 does not support tracing ICMP packets
+}
+
+func (pr *packetResponder) RequestSpan(ctx context.Context, pk *packet.ICMP) (context.Context, trace.Span) {
+	// datagram v3 does not support tracing ICMP packets
+	return ctx, tracing.NewNoopSpan()
+}
+
+func (pr *packetResponder) ReplySpan(ctx context.Context, logger *zerolog.Logger) (context.Context, trace.Span) {
+	// datagram v3 does not support tracing ICMP packets
+	return ctx, tracing.NewNoopSpan()
+}
+
+func (pr *packetResponder) ExportSpan() {
+	// datagram v3 does not support tracing ICMP packets
+}

quic/v3/icmp_test.go

@@ -0,0 +1,45 @@
+package v3_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/packet"
+)
+
+type noopICMPRouter struct{}
+
+func (noopICMPRouter) Request(ctx context.Context, pk *packet.ICMP, responder ingress.ICMPResponder) error {
+	return nil
+}
+func (noopICMPRouter) ConvertToTTLExceeded(pk *packet.ICMP, rawPacket packet.RawPacket) *packet.ICMP {
+	return nil
+}
+
+type mockICMPRouter struct {
+	recv chan *packet.ICMP
+}
+
+func newMockICMPRouter() *mockICMPRouter {
+	return &mockICMPRouter{
+		recv: make(chan *packet.ICMP, 1),
+	}
+}
+
+func (m *mockICMPRouter) Request(ctx context.Context, pk *packet.ICMP, responder ingress.ICMPResponder) error {
+	m.recv <- pk
+	return nil
+}
+func (mockICMPRouter) ConvertToTTLExceeded(pk *packet.ICMP, rawPacket packet.RawPacket) *packet.ICMP {
+	return packet.NewICMPTTLExceedPacket(pk.IP, rawPacket, testLocalAddr.AddrPort().Addr())
+}
+
+func assertICMPEqual(t *testing.T, expected *packet.ICMP, actual *packet.ICMP) {
+	if expected.Src != actual.Src {
+		t.Fatalf("Src address not equal: %+v\t%+v", expected, actual)
+	}
+	if expected.Dst != actual.Dst {
+		t.Fatalf("Dst address not equal: %+v\t%+v", expected, actual)
+	}
+}