TUN-6856: Refactor to lay foundation for tracing ICMP

Remove send and return methods from Funnel interface. Users of Funnel can provide their own send and return methods without wrapper to comply with the interface.
Move packet router to ingress package to avoid circular dependency

Author: chungthuang

cmd/cloudflared/tunnel/configuration.go

@@ -22,7 +22,6 @@ import (
 
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
 	"github.com/cloudflare/cloudflared/edgediscovery/allregions"
-	"github.com/cloudflare/cloudflared/packet"
 
 	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/connection"
@@ -463,7 +462,7 @@ func parseConfigIPVersion(version string) (v allregions.ConfigIPVersion, err err
 	return
 }
 
-func newPacketConfig(c *cli.Context, logger *zerolog.Logger) (*packet.GlobalRouterConfig, error) {
+func newPacketConfig(c *cli.Context, logger *zerolog.Logger) (*ingress.GlobalRouterConfig, error) {
 	ipv4Src, err := determineICMPv4Src(c.String("icmpv4-src"), logger)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to determine IPv4 source address for ICMP proxy")
@@ -484,7 +483,7 @@ func newPacketConfig(c *cli.Context, logger *zerolog.Logger) (*packet.GlobalRout
 	if err != nil {
 		return nil, err
 	}
-	return &packet.GlobalRouterConfig{
+	return &ingress.GlobalRouterConfig{
 		ICMPRouter: icmpRouter,
 		IPv4Src:    ipv4Src,
 		IPv6Src:    ipv6Src,

connection/quic.go

@@ -57,7 +57,7 @@ type QUICConnection struct {
 	sessionManager datagramsession.Manager
 	// datagramMuxer mux/demux datagrams from quic connection
 	datagramMuxer        *quicpogs.DatagramMuxerV2
-	packetRouter         *packet.Router
+	packetRouter         *ingress.PacketRouter
 	controlStreamHandler ControlStreamHandler
 	connOptions          *tunnelpogs.ConnectionOptions
 }
@@ -72,7 +72,7 @@ func NewQUICConnection(
 	connOptions *tunnelpogs.ConnectionOptions,
 	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
-	packetRouterConfig *packet.GlobalRouterConfig,
+	packetRouterConfig *ingress.GlobalRouterConfig,
 ) (*QUICConnection, error) {
 	udpConn, err := createUDPConnForConnIndex(connIndex, logger)
 	if err != nil {
@@ -93,8 +93,7 @@ func NewQUICConnection(
 	sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
 	datagramMuxer := quicpogs.NewDatagramMuxerV2(session, logger, sessionDemuxChan)
 	sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
-	muxer := muxerWrapper{muxer: datagramMuxer}
-	packetRouter := packet.NewRouter(packetRouterConfig, &muxer, &muxer, logger, orchestrator.WarpRoutingEnabled)
+	packetRouter := ingress.NewPacketRouter(packetRouterConfig, datagramMuxer, logger, orchestrator.WarpRoutingEnabled)
 
 	return &QUICConnection{
 		session:              session,

ingress/icmp_darwin.go

@@ -11,7 +11,6 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"net"
 	"net/netip"
 	"strconv"
 	"sync"
@@ -129,10 +128,7 @@ func newICMPProxy(listenIP netip.Addr, zone string, logger *zerolog.Logger, idle
 	}, nil
 }
 
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
-	if pk == nil {
-		return errPacketNil
-	}
+func (ip *icmpProxy) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
 	originalEcho, err := getICMPEcho(pk.Message)
 	if err != nil {
 		return err
@@ -152,13 +148,11 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 		if err != nil {
 			return nil, err
 		}
-		originSender := originSender{
-			conn:             ip.conn,
-			echoIDTracker:    ip.echoIDTracker,
-			echoIDTrackerKey: echoIDTrackerKey,
-			assignedEchoID:   assignedEchoID,
+		closeCallback := func() error {
+			ip.echoIDTracker.release(echoIDTrackerKey, assignedEchoID)
+			return nil
 		}
-		icmpFlow := newICMPEchoFlow(pk.Src, &originSender, responder, int(assignedEchoID), originalEcho.ID, ip.encoder)
+		icmpFlow := newICMPEchoFlow(pk.Src, closeCallback, ip.conn, responder, int(assignedEchoID), originalEcho.ID, ip.encoder)
 		return icmpFlow, nil
 	}
 	funnelID := echoFunnelID(assignedEchoID)
@@ -250,23 +244,3 @@ func (ip *icmpProxy) sendReply(reply *echoReply) error {
 	}
 	return icmpFlow.returnToSrc(reply)
 }
-
-// originSender wraps icmp.PacketConn to implement packet.FunnelUniPipe interface
-type originSender struct {
-	conn             *icmp.PacketConn
-	echoIDTracker    *echoIDTracker
-	echoIDTrackerKey flow3Tuple
-	assignedEchoID   uint16
-}
-
-func (os *originSender) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	_, err := os.conn.WriteTo(pk.Data, &net.UDPAddr{
-		IP: dst.AsSlice(),
-	})
-	return err
-}
-
-func (os *originSender) Close() error {
-	os.echoIDTracker.release(os.echoIDTrackerKey, os.assignedEchoID)
-	return nil
-}

ingress/icmp_generic.go

@@ -18,7 +18,7 @@ var errICMPProxyNotImplemented = fmt.Errorf("ICMP proxy is not implemented on %s
 
 type icmpProxy struct{}
 
-func (ip icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
+func (ip icmpProxy) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
 	return errICMPProxyNotImplemented
 }
 

ingress/icmp_linux.go

@@ -97,10 +97,7 @@ func checkInPingGroup() error {
 	return fmt.Errorf("did not find group range in %s", pingGroupPath)
 }
 
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
-	if pk == nil {
-		return errPacketNil
-	}
+func (ip *icmpProxy) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
 	originalEcho, err := getICMPEcho(pk.Message)
 	if err != nil {
 		return err
@@ -113,13 +110,15 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 		}
 		ip.logger.Debug().Msgf("Opened ICMP socket listen on %s", conn.LocalAddr())
 		newConnChan <- conn
+		closeCallback := func() error {
+			return conn.Close()
+		}
 		localUDPAddr, ok := conn.LocalAddr().(*net.UDPAddr)
 		if !ok {
 			return nil, fmt.Errorf("ICMP listener address %s is not net.UDPAddr", conn.LocalAddr())
 		}
-		originSender := originSender{conn: conn}
 		echoID := localUDPAddr.Port
-		icmpFlow := newICMPEchoFlow(pk.Src, &originSender, responder, echoID, originalEcho.ID, packet.NewEncoder())
+		icmpFlow := newICMPEchoFlow(pk.Src, closeCallback, conn, responder, echoID, originalEcho.ID, packet.NewEncoder())
 		return icmpFlow, nil
 	}
 	funnelID := flow3Tuple{
@@ -187,22 +186,6 @@ func (ip *icmpProxy) listenResponse(flow *icmpEchoFlow, conn *icmp.PacketConn) e
 	}
 }
 
-// originSender wraps icmp.PacketConn to implement packet.FunnelUniPipe interface
-type originSender struct {
-	conn *icmp.PacketConn
-}
-
-func (os *originSender) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	_, err := os.conn.WriteTo(pk.Data, &net.UDPAddr{
-		IP: dst.AsSlice(),
-	})
-	return err
-}
-
-func (os *originSender) Close() error {
-	return os.conn.Close()
-}
-
 // Only linux uses flow3Tuple as FunnelID
 func (ft flow3Tuple) Type() string {
 	return "srcIP_dstIP_echoID"

ingress/icmp_posix.go

@@ -43,24 +43,54 @@ type flow3Tuple struct {
 
 // icmpEchoFlow implements the packet.Funnel interface.
 type icmpEchoFlow struct {
-	*packet.RawPacketFunnel
+	*packet.ActivityTracker
+	closeCallback  func() error
+	src            netip.Addr
+	originConn     *icmp.PacketConn
+	responder      *packetResponder
 	assignedEchoID int
 	originalEchoID int
 	// it's up to the user to ensure respEncoder is not used concurrently
 	respEncoder *packet.Encoder
 }
 
-func newICMPEchoFlow(src netip.Addr, sendPipe, returnPipe packet.FunnelUniPipe, assignedEchoID, originalEchoID int, respEncoder *packet.Encoder) *icmpEchoFlow {
+func newICMPEchoFlow(src netip.Addr, closeCallback func() error, originConn *icmp.PacketConn, responder *packetResponder, assignedEchoID, originalEchoID int, respEncoder *packet.Encoder) *icmpEchoFlow {
 	return &icmpEchoFlow{
-		RawPacketFunnel: packet.NewRawPacketFunnel(src, sendPipe, returnPipe),
+		ActivityTracker: packet.NewActivityTracker(),
+		closeCallback:   closeCallback,
+		src:             src,
+		originConn:      originConn,
+		responder:       responder,
 		assignedEchoID:  assignedEchoID,
 		originalEchoID:  originalEchoID,
 		respEncoder:     respEncoder,
 	}
 }
 
+func (ief *icmpEchoFlow) Equal(other packet.Funnel) bool {
+	otherICMPFlow, ok := other.(*icmpEchoFlow)
+	if !ok {
+		return false
+	}
+	if otherICMPFlow.src != ief.src {
+		return false
+	}
+	if otherICMPFlow.originalEchoID != ief.originalEchoID {
+		return false
+	}
+	if otherICMPFlow.assignedEchoID != ief.assignedEchoID {
+		return false
+	}
+	return true
+}
+
+func (ief *icmpEchoFlow) Close() error {
+	return ief.closeCallback()
+}
+
 // sendToDst rewrites the echo ID to the one assigned to this flow
 func (ief *icmpEchoFlow) sendToDst(dst netip.Addr, msg *icmp.Message) error {
+	ief.UpdateLastActive()
 	originalEcho, err := getICMPEcho(msg)
 	if err != nil {
 		return err
@@ -80,17 +110,21 @@ func (ief *icmpEchoFlow) sendToDst(dst netip.Addr, msg *icmp.Message) error {
 	if err != nil {
 		return err
 	}
-	return ief.SendToDst(dst, packet.RawPacket{Data: serializedPacket})
+	_, err = ief.originConn.WriteTo(serializedPacket, &net.UDPAddr{
+		IP: dst.AsSlice(),
+	})
+	return err
 }
 
 // returnToSrc rewrites the echo ID to the original echo ID from the eyeball
 func (ief *icmpEchoFlow) returnToSrc(reply *echoReply) error {
+	ief.UpdateLastActive()
 	reply.echo.ID = ief.originalEchoID
 	reply.msg.Body = reply.echo
 	pk := packet.ICMP{
 		IP: &packet.IP{
 			Src:      reply.from,
-			Dst:      ief.Src,
+			Dst:      ief.src,
 			Protocol: layers.IPProtocol(reply.msg.Type.Protocol()),
 			TTL:      packet.DefaultTTL,
 		},
@@ -100,7 +134,7 @@ func (ief *icmpEchoFlow) returnToSrc(reply *echoReply) error {
 	if err != nil {
 		return err
 	}
-	return ief.ReturnToSrc(serializedPacket)
+	return ief.responder.returnPacket(serializedPacket)
 }
 
 type echoReply struct {

ingress/icmp_posix_test.go

@@ -52,24 +52,26 @@ func TestFunnelIdleTimeout(t *testing.T) {
 			},
 		},
 	}
-	responder := echoFlowResponder{
-		decoder:  packet.NewICMPDecoder(),
-		respChan: make(chan []byte),
+	muxer := newMockMuxer(0)
+	responder := packetResponder{
+		datagramMuxer: muxer,
 	}
-	require.NoError(t, proxy.Request(&pk, &responder))
-	responder.validate(t, &pk)
+	require.NoError(t, proxy.Request(ctx, &pk, &responder))
+	validateEchoFlow(t, muxer, &pk)
 
 	// Send second request, should reuse the funnel
-	require.NoError(t, proxy.Request(&pk, nil))
-	responder.validate(t, &pk)
+	require.NoError(t, proxy.Request(ctx, &pk, &packetResponder{
+		datagramMuxer: nil,
+	}))
+	validateEchoFlow(t, muxer, &pk)
 
 	time.Sleep(idleTimeout * 2)
-	newResponder := echoFlowResponder{
-		decoder:  packet.NewICMPDecoder(),
-		respChan: make(chan []byte),
+	newMuxer := newMockMuxer(0)
+	newResponder := packetResponder{
+		datagramMuxer: newMuxer,
 	}
-	require.NoError(t, proxy.Request(&pk, &newResponder))
-	newResponder.validate(t, &pk)
+	require.NoError(t, proxy.Request(ctx, &pk, &newResponder))
+	validateEchoFlow(t, newMuxer, &pk)
 
 	cancel()
 	<-proxyDone

ingress/icmp_windows.go

@@ -265,7 +265,7 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 // Request sends an ICMP echo request and wait for a reply or timeout.
 // The async version of Win32 APIs take a callback whose memory is not garbage collected, so we use the synchronous version.
 // It's possible that a slow request will block other requests, so we set the timeout to only 1s.
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
+func (ip *icmpProxy) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
 	if pk == nil {
 		return errPacketNil
 	}
@@ -292,7 +292,7 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 	return nil
 }
 
-func (ip *icmpProxy) handleEchoReply(request *packet.ICMP, echoReq *icmp.Echo, data []byte, responder packet.FunnelUniPipe) error {
+func (ip *icmpProxy) handleEchoReply(request *packet.ICMP, echoReq *icmp.Echo, data []byte, responder *packetResponder) error {
 	var replyType icmp.Type
 	if request.Dst.Is4() {
 		replyType = ipv4.ICMPTypeEchoReply
@@ -331,7 +331,7 @@ func (ip *icmpProxy) handleEchoReply(request *packet.ICMP, echoReq *icmp.Echo, d
 	if err != nil {
 		return err
 	}
-	return responder.SendPacket(request.Src, serializedPacket)
+	return responder.returnPacket(serializedPacket)
 }
 
 func (ip *icmpProxy) icmpEchoRoundtrip(dst netip.Addr, echo *icmp.Echo) ([]byte, error) {

ingress/origin_icmp_proxy.go

@@ -37,7 +37,9 @@ func NewICMPRouter(ipv4Addr, ipv6Addr netip.Addr, ipv6Zone string, logger *zerol
 	ipv4Proxy, ipv4Err := newICMPProxy(ipv4Addr, "", logger, funnelIdleTimeout)
 	ipv6Proxy, ipv6Err := newICMPProxy(ipv6Addr, ipv6Zone, logger, funnelIdleTimeout)
 	if ipv4Err != nil && ipv6Err != nil {
-		return nil, fmt.Errorf("cannot create ICMPv4 proxy: %v nor ICMPv6 proxy: %v", ipv4Err, ipv6Err)
+		err := fmt.Errorf("cannot create ICMPv4 proxy: %v nor ICMPv6 proxy: %v", ipv4Err, ipv6Err)
+		logger.Debug().Err(err).Msg("ICMP proxy feature is disabled")
+		return nil, err
 	}
 	if ipv4Err != nil {
 		logger.Debug().Err(ipv4Err).Msg("failed to create ICMPv4 proxy, only ICMPv6 proxy is created")
@@ -73,15 +75,18 @@ func (ir *icmpRouter) Serve(ctx context.Context) error {
 	return fmt.Errorf("ICMPv4 proxy and ICMPv6 proxy are both nil")
 }
 
-func (ir *icmpRouter) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
+func (ir *icmpRouter) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
+	if pk == nil {
+		return errPacketNil
+	}
 	if pk.Dst.Is4() {
 		if ir.ipv4Proxy != nil {
-			return ir.ipv4Proxy.Request(pk, responder)
+			return ir.ipv4Proxy.Request(ctx, pk, responder)
 		}
 		return fmt.Errorf("ICMPv4 proxy was not instantiated")
 	}
 	if ir.ipv6Proxy != nil {
-		return ir.ipv6Proxy.Request(pk, responder)
+		return ir.ipv6Proxy.Request(ctx, pk, responder)
 	}
 	return fmt.Errorf("ICMPv6 proxy was not instantiated")
 }