TUN-6530: Implement ICMPv4 proxy

This proxy uses unprivileged datagram-oriented endpoint and is shared by all quic connections

Author: chungthuang

connection/quic.go

@@ -48,6 +48,7 @@ type QUICConnection struct {
 	sessionManager datagramsession.Manager
 	// datagramMuxer mux/demux datagrams from quic connection
 	datagramMuxer        quicpogs.BaseDatagramMuxer
+	packetRouter         *packetRouter
 	controlStreamHandler ControlStreamHandler
 	connOptions          *tunnelpogs.ConnectionOptions
 }
@@ -61,14 +62,28 @@ func NewQUICConnection(
 	connOptions *tunnelpogs.ConnectionOptions,
 	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
+	icmpProxy ingress.ICMPProxy,
 ) (*QUICConnection, error) {
 	session, err := quic.DialAddr(edgeAddr.String(), tlsConfig, quicConfig)
 	if err != nil {
 		return nil, &EdgeQuicDialError{Cause: err}
 	}
 
 	sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
-	datagramMuxer := quicpogs.NewDatagramMuxer(session, logger, sessionDemuxChan)
+	var (
+		datagramMuxer quicpogs.BaseDatagramMuxer
+		pr            *packetRouter
+	)
+	if icmpProxy != nil {
+		pr = &packetRouter{
+			muxer:     quicpogs.NewDatagramMuxerV2(session, logger, sessionDemuxChan),
+			icmpProxy: icmpProxy,
+			logger:    logger,
+		}
+		datagramMuxer = pr.muxer
+	} else {
+		datagramMuxer = quicpogs.NewDatagramMuxer(session, logger, sessionDemuxChan)
+	}
 	sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
 
 	return &QUICConnection{
@@ -77,6 +92,7 @@ func NewQUICConnection(
 		logger:               logger,
 		sessionManager:       sessionManager,
 		datagramMuxer:        datagramMuxer,
+		packetRouter:         pr,
 		controlStreamHandler: controlStreamHandler,
 		connOptions:          connOptions,
 	}, nil
@@ -117,6 +133,12 @@ func (q *QUICConnection) Serve(ctx context.Context) error {
 		defer cancel()
 		return q.datagramMuxer.ServeReceive(ctx)
 	})
+	if q.packetRouter != nil {
+		errGroup.Go(func() error {
+			defer cancel()
+			return q.packetRouter.serve(ctx)
+		})
+	}
 
 	return errGroup.Wait()
 }
@@ -305,6 +327,32 @@ func (q *QUICConnection) UpdateConfiguration(ctx context.Context, version int32,
 	return q.orchestrator.UpdateConfig(version, config)
 }
 
+type packetRouter struct {
+	muxer     *quicpogs.DatagramMuxerV2
+	icmpProxy ingress.ICMPProxy
+	logger    *zerolog.Logger
+}
+
+func (pr *packetRouter) serve(ctx context.Context) error {
+	icmpDecoder := packet.NewICMPDecoder()
+	for {
+		pk, err := pr.muxer.ReceivePacket(ctx)
+		if err != nil {
+			return err
+		}
+		icmpPacket, err := icmpDecoder.Decode(pk)
+		if err != nil {
+			pr.logger.Err(err).Msg("Failed to decode ICMP packet from quic datagram")
+			continue
+		}
+
+		if err := pr.icmpProxy.Request(icmpPacket, pr.muxer); err != nil {
+			pr.logger.Err(err).Str("src", icmpPacket.Src.String()).Str("dst", icmpPacket.Dst.String()).Msg("Failed to send ICMP packet")
+			continue
+		}
+	}
+}
+
 // streamReadWriteAcker is a light wrapper over QUIC streams with a callback to send response back to
 // the client.
 type streamReadWriteAcker struct {

connection/quic_test.go

@@ -682,6 +682,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection
 		&tunnelpogs.ConnectionOptions{},
 		fakeControlStream{},
 		&log,
+		nil,
 	)
 	require.NoError(t, err)
 	return qc

ingress/origin_icmp_proxy.go

@@ -0,0 +1,139 @@
+package ingress
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+
+	"github.com/google/gopacket/layers"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+	"golang.org/x/net/icmp"
+
+	"github.com/cloudflare/cloudflared/packet"
+)
+
+// ICMPProxy sends ICMP messages and listens for their responses
+type ICMPProxy interface {
+	// Request sends an ICMP message
+	Request(pk *packet.ICMP, responder packet.FlowResponder) error
+	// ListenResponse listens for responses to the requests until context is done
+	ListenResponse(ctx context.Context) error
+}
+
+// TODO: TUN-6654 Extend support to IPv6
+type icmpProxy struct {
+	srcFlowTracker *packet.FlowTracker
+	conn           *icmp.PacketConn
+	logger         *zerolog.Logger
+	encoder        *packet.Encoder
+}
+
+// TODO: TUN-6586: Use echo ID as FlowID
+type seqNumFlowID int
+
+func (snf seqNumFlowID) ID() string {
+	return strconv.FormatInt(int64(snf), 10)
+}
+
+func NewICMPProxy(network string, listenIP net.IP, logger *zerolog.Logger) (*icmpProxy, error) {
+	conn, err := icmp.ListenPacket(network, listenIP.String())
+	if err != nil {
+		return nil, err
+	}
+	return &icmpProxy{
+		srcFlowTracker: packet.NewFlowTracker(),
+		conn:           conn,
+		logger:         logger,
+		encoder:        packet.NewEncoder(),
+	}, nil
+}
+
+func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+	switch body := pk.Message.Body.(type) {
+	case *icmp.Echo:
+		return ip.sendICMPEchoRequest(pk, body, responder)
+	default:
+		return fmt.Errorf("sending ICMP %s is not implemented", pk.Type)
+	}
+}
+
+func (ip *icmpProxy) ListenResponse(ctx context.Context) error {
+	go func() {
+		<-ctx.Done()
+		ip.conn.Close()
+	}()
+	buf := make([]byte, 1500)
+	for {
+		n, src, err := ip.conn.ReadFrom(buf)
+		if err != nil {
+			return err
+		}
+		// TODO: TUN-6654 Check for IPv6
+		msg, err := icmp.ParseMessage(int(layers.IPProtocolICMPv4), buf[:n])
+		if err != nil {
+			ip.logger.Error().Err(err).Str("src", src.String()).Msg("Failed to parse ICMP message")
+			continue
+		}
+		switch body := msg.Body.(type) {
+		case *icmp.Echo:
+			if err := ip.handleEchoResponse(msg, body); err != nil {
+				ip.logger.Error().Err(err).Str("src", src.String()).Msg("Failed to handle ICMP response")
+				continue
+			}
+		default:
+			ip.logger.Warn().
+				Str("icmpType", fmt.Sprintf("%s", msg.Type)).
+				Msgf("Responding to this type of ICMP is not implemented")
+			continue
+		}
+	}
+}
+
+func (ip *icmpProxy) sendICMPEchoRequest(pk *packet.ICMP, echo *icmp.Echo, responder packet.FlowResponder) error {
+	flow := packet.Flow{
+		Src:       pk.Src,
+		Dst:       pk.Dst,
+		Responder: responder,
+	}
+	// TODO: TUN-6586 rewrite ICMP echo request identifier and use it to track flows
+	flowID := seqNumFlowID(echo.Seq)
+	// TODO: TUN-6588 clean up flows
+	if replaced := ip.srcFlowTracker.Register(flowID, &flow, true); replaced {
+		ip.logger.Info().Str("src", flow.Src.String()).Str("dst", flow.Dst.String()).Msg("Replaced flow")
+	}
+	var pseudoHeader []byte = nil
+	serializedMsg, err := pk.Marshal(pseudoHeader)
+	if err != nil {
+		return errors.Wrap(err, "Failed to encode ICMP message")
+	}
+	// The address needs to be of type UDPAddr when conn is created without priviledge
+	_, err = ip.conn.WriteTo(serializedMsg, &net.UDPAddr{
+		IP: pk.Dst.AsSlice(),
+	})
+	return err
+}
+
+func (ip *icmpProxy) handleEchoResponse(msg *icmp.Message, echo *icmp.Echo) error {
+	flow, ok := ip.srcFlowTracker.Get(seqNumFlowID(echo.Seq))
+	if !ok {
+		return fmt.Errorf("flow not found")
+	}
+	icmpPacket := packet.ICMP{
+		IP: &packet.IP{
+			Src:      flow.Dst,
+			Dst:      flow.Src,
+			Protocol: layers.IPProtocol(msg.Type.Protocol()),
+		},
+		Message: msg,
+	}
+	serializedPacket, err := ip.encoder.Encode(&icmpPacket)
+	if err != nil {
+		return errors.Wrap(err, "Failed to encode ICMP message")
+	}
+	if err := flow.Responder.SendPacket(serializedPacket); err != nil {
+		return errors.Wrap(err, "Failed to send packet to the edge")
+	}
+	return nil
+}

ingress/origin_icmp_proxy_test.go

@@ -0,0 +1,150 @@
+package ingress
+
+import (
+	"context"
+	"fmt"
+	"net/netip"
+	"runtime"
+	"testing"
+
+	"github.com/google/gopacket/layers"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/icmp"
+	"golang.org/x/net/ipv4"
+
+	"github.com/cloudflare/cloudflared/packet"
+)
+
+var (
+	noopLogger  = zerolog.Nop()
+	localhostIP = netip.MustParseAddr("127.0.0.1")
+)
+
+// TestICMPProxyEcho makes sure we can send ICMP echo via the Request method and receives response via the
+// ListenResponse method
+func TestICMPProxyEcho(t *testing.T) {
+	skipWindows(t)
+	const (
+		echoID = 36571
+		endSeq = 100
+	)
+	proxy, err := NewICMPProxy("udp4", localhostIP.AsSlice(), &noopLogger)
+	require.NoError(t, err)
+
+	proxyDone := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		proxy.ListenResponse(ctx)
+		close(proxyDone)
+	}()
+
+	responder := echoFlowResponder{
+		decoder:  packet.NewICMPDecoder(),
+		respChan: make(chan []byte),
+	}
+
+	ip := packet.IP{
+		Src:      localhostIP,
+		Dst:      localhostIP,
+		Protocol: layers.IPProtocolICMPv4,
+	}
+	for i := 0; i < endSeq; i++ {
+		pk := packet.ICMP{
+			IP: &ip,
+			Message: &icmp.Message{
+				Type: ipv4.ICMPTypeEcho,
+				Code: 0,
+				Body: &icmp.Echo{
+					ID:   echoID,
+					Seq:  i,
+					Data: []byte(fmt.Sprintf("icmp echo seq %d", i)),
+				},
+			},
+		}
+		require.NoError(t, proxy.Request(&pk, &responder))
+		responder.validate(t, &pk)
+	}
+	cancel()
+	<-proxyDone
+}
+
+// TestICMPProxyRejectNotEcho makes sure it rejects messages other than echo
+func TestICMPProxyRejectNotEcho(t *testing.T) {
+	skipWindows(t)
+	msgs := []icmp.Message{
+		{
+			Type: ipv4.ICMPTypeDestinationUnreachable,
+			Code: 1,
+			Body: &icmp.DstUnreach{
+				Data: []byte("original packet"),
+			},
+		},
+		{
+			Type: ipv4.ICMPTypeTimeExceeded,
+			Code: 1,
+			Body: &icmp.TimeExceeded{
+				Data: []byte("original packet"),
+			},
+		},
+		{
+			Type: ipv4.ICMPType(2),
+			Code: 0,
+			Body: &icmp.PacketTooBig{
+				MTU:  1280,
+				Data: []byte("original packet"),
+			},
+		},
+	}
+	proxy, err := NewICMPProxy("udp4", localhostIP.AsSlice(), &noopLogger)
+	require.NoError(t, err)
+
+	responder := echoFlowResponder{
+		decoder:  packet.NewICMPDecoder(),
+		respChan: make(chan []byte),
+	}
+	for _, m := range msgs {
+		pk := packet.ICMP{
+			IP: &packet.IP{
+				Src:      localhostIP,
+				Dst:      localhostIP,
+				Protocol: layers.IPProtocolICMPv4,
+			},
+			Message: &m,
+		}
+		require.Error(t, proxy.Request(&pk, &responder))
+	}
+}
+
+func skipWindows(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("Cannot create non-privileged datagram-oriented ICMP endpoint on Windows")
+	}
+}
+
+type echoFlowResponder struct {
+	decoder  *packet.ICMPDecoder
+	respChan chan []byte
+}
+
+func (efr *echoFlowResponder) SendPacket(pk packet.RawPacket) error {
+	copiedPacket := make([]byte, len(pk.Data))
+	copy(copiedPacket, pk.Data)
+	efr.respChan <- copiedPacket
+	return nil
+}
+
+func (efr *echoFlowResponder) validate(t *testing.T, echoReq *packet.ICMP) {
+	pk := <-efr.respChan
+	decoded, err := efr.decoder.Decode(packet.RawPacket{Data: pk})
+	require.NoError(t, err)
+	require.Equal(t, decoded.Src, echoReq.Dst)
+	require.Equal(t, decoded.Dst, echoReq.Src)
+	require.Equal(t, echoReq.Protocol, decoded.Protocol)
+
+	require.Equal(t, ipv4.ICMPTypeEchoReply, decoded.Type)
+	require.Equal(t, 0, decoded.Code)
+	require.NotZero(t, decoded.Checksum)
+	// TODO: TUN-6586: Enable this validation when ICMP echo ID matches on Linux
+	//require.Equal(t, echoReq.Body, decoded.Body)
+}