TUN-6696: Refactor flow into funnel and close idle funnels

A funnel is an abstraction for 1 source to many destinations.
As part of this refactoring, shared logic between Darwin and Linux are moved into icmp_posix

connection/quic.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/netip"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -346,13 +347,31 @@ func (pr *packetRouter) serve(ctx context.Context) error {
 			continue
 		}
 
-		if err := pr.icmpProxy.Request(icmpPacket, pr.muxer); err != nil {
+		flowPipe := muxerResponder{muxer: pr.muxer}
-			pr.logger.Err(err).Str("src", icmpPacket.Src.String()).Str("dst", icmpPacket.Dst.String()).Msg("Failed to send ICMP packet")
+		if err := pr.icmpProxy.Request(icmpPacket, &flowPipe); err != nil {
+			pr.logger.Err(err).
+				Str("src", icmpPacket.Src.String()).
+				Str("dst", icmpPacket.Dst.String()).
+				Interface("type", icmpPacket.Type).
+				Msg("Failed to send ICMP packet")
 			continue
 		}
 	}
 }
 
+// muxerResponder wraps DatagramMuxerV2 to satisfy the packet.FunnelUniPipe interface
+type muxerResponder struct {
+	muxer *quicpogs.DatagramMuxerV2
+}
+
+func (mr *muxerResponder) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
+	return mr.muxer.SendPacket(pk)
+}
+
+func (mr *muxerResponder) Close() error {
+	return nil
+}
+
 // streamReadWriteAcker is a light wrapper over QUIC streams with a callback to send response back to
 // the client.
 type streamReadWriteAcker struct {

ingress/icmp_darwin.go

@@ -2,6 +2,11 @@
 
 package ingress
 
+// This file implements ICMPProxy for Darwin. It uses a non-privileged ICMP socket to send echo requests and listen for
+// echo replies. The source IP of the requests are rewritten to the bind IP of the socket and the socket reads all
+// messages, so we use echo ID to distinguish the replies. Each (source IP, destination IP, echo ID) is assigned a
+// unique echo ID.
+
 import (
 	"context"
 	"fmt"
@@ -10,9 +15,8 @@ import (
 	"net/netip"
 	"strconv"
 	"sync"
+	"time"
 
-	"github.com/google/gopacket/layers"
-	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"golang.org/x/net/icmp"
 
@@ -20,13 +24,14 @@ import (
 )
 
 // TODO: TUN-6654 Extend support to IPv6
-// On Darwin, a non-privileged ICMP socket can read messages from all echo IDs, so we use it for all sources.
 type icmpProxy struct {
-	// TODO: TUN-6588 clean up flows
+	srcFunnelTracker *packet.FunnelTracker
-	srcFlowTracker *packet.FlowTracker
+	echoIDTracker    *echoIDTracker
-	echoIDTracker  *echoIDTracker
+	conn             *icmp.PacketConn
-	conn           *icmp.PacketConn
+	// Response is handled in one-by-one, so encoder can be shared between funnels
-	logger         *zerolog.Logger
+	encoder     *packet.Encoder
+	logger      *zerolog.Logger
+	idleTimeout time.Duration
 }
 
 // echoIDTracker tracks which ID has been assigned. It first loops through assignment from lastAssignment to then end,
@@ -92,8 +97,8 @@ func (eit *echoIDTracker) release(srcIP netip.Addr, id uint16) bool {
 	eit.lock.Lock()
 	defer eit.lock.Unlock()
 
-	currentID, ok := eit.srcIPMapping[srcIP]
+	currentID, exists := eit.srcIPMapping[srcIP]
-	if ok && id == currentID {
+	if exists && id == currentID {
 		delete(eit.srcIPMapping, srcIP)
 		eit.assignment[id] = false
 		return true
@@ -101,39 +106,68 @@ func (eit *echoIDTracker) release(srcIP netip.Addr, id uint16) bool {
 	return false
 }
 
-type echoFlowID uint16
+type echoFunnelID uint16
 
-func (snf echoFlowID) Type() string {
+func (snf echoFunnelID) Type() string {
 	return "echoID"
 }
 
-func (snf echoFlowID) String() string {
+func (snf echoFunnelID) String() string {
 	return strconv.FormatUint(uint64(snf), 10)
 }
 
-func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger, idleTimeout time.Duration) (ICMPProxy, error) {
 	conn, err := newICMPConn(listenIP)
 	if err != nil {
 		return nil, err
 	}
 	return &icmpProxy{
-		srcFlowTracker: packet.NewFlowTracker(),
+		srcFunnelTracker: packet.NewFunnelTracker(),
-		echoIDTracker:  newEchoIDTracker(),
+		echoIDTracker:    newEchoIDTracker(),
-		conn:           conn,
+		encoder:          packet.NewEncoder(),
-		logger:         logger,
+		conn:             conn,
+		logger:           logger,
+		idleTimeout:      idleTimeout,
 	}, nil
 }
 
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
 	if pk == nil {
 		return errPacketNil
 	}
-	switch body := pk.Message.Body.(type) {
+	// TODO: TUN-6744 assign unique flow per (src, echo ID)
-	case *icmp.Echo:
+	echoID, exists := ip.echoIDTracker.get(pk.Src)
-		return ip.sendICMPEchoRequest(pk, body, responder)
+	if !exists {
-	default:
+		originalEcho, err := getICMPEcho(pk.Message)
-		return fmt.Errorf("sending ICMP %s is not implemented", pk.Type)
+		if err != nil {
+			return err
+		}
+		echoID, exists = ip.echoIDTracker.assign(pk.Src)
+		if !exists {
+			return fmt.Errorf("failed to assign unique echo ID")
+		}
+		funnelID := echoFunnelID(echoID)
+		originSender := originSender{
+			conn:          ip.conn,
+			echoIDTracker: ip.echoIDTracker,
+			srcIP:         pk.Src,
+			echoID:        echoID,
+		}
+		icmpFlow := newICMPEchoFlow(pk.Src, &originSender, responder, int(echoID), originalEcho.ID, ip.encoder)
+		if replaced := ip.srcFunnelTracker.Register(funnelID, icmpFlow); replaced {
+			ip.logger.Info().Str("src", pk.Src.String()).Msg("Replaced funnel")
+		}
+		return icmpFlow.sendToDst(pk.Dst, pk.Message)
 	}
+	funnel, exists := ip.srcFunnelTracker.Get(echoFunnelID(echoID))
+	if !exists {
+		return packet.ErrFunnelNotFound
+	}
+	icmpFlow, err := toICMPEchoFlow(funnel)
+	if err != nil {
+		return err
+	}
+	return icmpFlow.sendToDst(pk.Dst, pk.Message)
 }
 
 // Serve listens for responses to the requests until context is done
@@ -142,88 +176,54 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 		<-ctx.Done()
 		ip.conn.Close()
 	}()
+	go func() {
+		ip.srcFunnelTracker.ScheduleCleanup(ctx, ip.idleTimeout)
+	}()
 	buf := make([]byte, mtu)
-	encoder := packet.NewEncoder()
 	for {
 		n, src, err := ip.conn.ReadFrom(buf)
 		if err != nil {
 			return err
 		}
-		// TODO: TUN-6654 Check for IPv6
+		if err := ip.handleResponse(src, buf[:n]); err != nil {
-		msg, err := icmp.ParseMessage(int(layers.IPProtocolICMPv4), buf[:n])
+			ip.logger.Err(err).Str("src", src.String()).Msg("Failed to handle ICMP response")
-		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(encoder, msg, body); err != nil {
-				ip.logger.Error().Err(err).
-					Str("src", src.String()).
-					Str("flowID", echoFlowID(body.ID).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 {
+func (ip *icmpProxy) handleResponse(from net.Addr, rawMsg []byte) error {
-	echoID, ok := ip.echoIDTracker.get(pk.Src)
+	reply, err := parseReply(from, rawMsg)
-	if !ok {
-		echoID, ok = ip.echoIDTracker.assign(pk.Src)
-		if !ok {
-			return fmt.Errorf("failed to assign unique echo ID")
-		}
-		flowID := echoFlowID(echoID)
-		flow := packet.Flow{
-			Src:       pk.Src,
-			Dst:       pk.Dst,
-			Responder: responder,
-		}
-		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")
-		}
-	}
-
-	echo.ID = int(echoID)
-	var pseudoHeader []byte = nil
-	serializedMsg, err := pk.Marshal(pseudoHeader)
 	if err != nil {
-		return errors.Wrap(err, "Failed to encode ICMP message")
+		return err
 	}
-	// The address needs to be of type UDPAddr when conn is created without priviledge
+	funnel, exists := ip.srcFunnelTracker.Get(echoFunnelID(reply.echo.ID))
-	_, err = ip.conn.WriteTo(serializedMsg, &net.UDPAddr{
+	if !exists {
-		IP: pk.Dst.AsSlice(),
+		return packet.ErrFunnelNotFound
+	}
+	icmpFlow, err := toICMPEchoFlow(funnel)
+	if err != nil {
+		return err
+	}
+	return icmpFlow.returnToSrc(reply)
+}
+
+// originSender wraps icmp.PacketConn to implement packet.FunnelUniPipe interface
+type originSender struct {
+	conn          *icmp.PacketConn
+	echoIDTracker *echoIDTracker
+	srcIP         netip.Addr
+	echoID        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 (ip *icmpProxy) handleEchoResponse(encoder *packet.Encoder, msg *icmp.Message, echo *icmp.Echo) error {
+func (os *originSender) Close() error {
-	flowID := echoFlowID(echo.ID)
+	os.echoIDTracker.release(os.srcIP, os.echoID)
-	flow, ok := ip.srcFlowTracker.Get(flowID)
-	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 := 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/icmp_generic.go

@@ -10,6 +10,6 @@ import (
 	"github.com/rs/zerolog"
 )
 
-func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger, idleTimeout time.Duration) (ICMPProxy, error) {
 	return nil, fmt.Errorf("ICMP proxy is not implemented on %s", runtime.GOOS)
 }

ingress/icmp_linux.go

@@ -2,16 +2,18 @@
 
 package ingress
 
+// This file implements ICMPProxy for Linux. Each (source IP, destination IP, echo ID) opens a non-privileged ICMP socket.
+// The source IP of the requests are rewritten to the bind IP of the socket and echo ID rewritten to the port number of
+// the socket. The kernel ensures the socket only reads replies whose echo ID matches the port number.
+// For more information about the socket, see https://man7.org/linux/man-pages/man7/icmp.7.html and https://lwn.net/Articles/422330/
+
 import (
 	"context"
 	"fmt"
 	"net"
 	"net/netip"
-	"sync"
-	"sync/atomic"
 	"time"
 
-	"github.com/google/gopacket/layers"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"golang.org/x/net/icmp"
@@ -19,30 +21,27 @@ import (
 	"github.com/cloudflare/cloudflared/packet"
 )
 
-// The request echo ID is rewritten to the port of the socket. The kernel uses the reply echo ID to demultiplex
-// We can open a socket for each source so multiple sources requesting the same destination doesn't collide
 type icmpProxy struct {
-	srcToFlowTracker *srcToFlowTracker
+	srcFunnelTracker *packet.FunnelTracker
 	listenIP         netip.Addr
 	logger           *zerolog.Logger
-	shutdownC        chan struct{}
+	idleTimeout      time.Duration
 }
 
-func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger, idleTimeout time.Duration) (ICMPProxy, error) {
 	if err := testPermission(listenIP); err != nil {
 		return nil, err
 	}
 	return &icmpProxy{
-		srcToFlowTracker: newSrcToConnTracker(),
+		srcFunnelTracker: packet.NewFunnelTracker(),
 		listenIP:         listenIP,
 		logger:           logger,
-		shutdownC:        make(chan struct{}),
+		idleTimeout:      idleTimeout,
 	}, nil
 }
 
 func testPermission(listenIP netip.Addr) error {
 	// Opens a non-privileged ICMP socket. On Linux the group ID of the process needs to be in ping_group_range
-	// For more information, see https://man7.org/linux/man-pages/man7/icmp.7.html and https://lwn.net/Articles/422330/
 	conn, err := newICMPConn(listenIP)
 	if err != nil {
 		// TODO: TUN-6715 check if cloudflared is in ping_group_range if the check failed. If not log instruction to
@@ -54,213 +53,105 @@ func testPermission(listenIP netip.Addr) error {
 	return nil
 }
 
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
 	if pk == nil {
 		return errPacketNil
 	}
-	echo, err := getICMPEcho(pk)
+	funnelID := srcIPFunnelID(pk.Src)
-	if err != nil {
+	funnel, exists := ip.srcFunnelTracker.Get(funnelID)
-		return err
+	if !exists {
-	}
+		originalEcho, err := getICMPEcho(pk.Message)
-	return ip.sendICMPEchoRequest(pk, echo, responder)
-}
-
-func (ip *icmpProxy) Serve(ctx context.Context) error {
-	<-ctx.Done()
-	close(ip.shutdownC)
-	return ctx.Err()
-}
-
-func (ip *icmpProxy) sendICMPEchoRequest(pk *packet.ICMP, echo *icmp.Echo, responder packet.FlowResponder) error {
-	icmpFlow, ok := ip.srcToFlowTracker.get(pk.Src)
-	if ok {
-		return icmpFlow.send(pk)
-	}
-
-	conn, err := newICMPConn(ip.listenIP)
-	if err != nil {
-		return err
-	}
-	flow := packet.Flow{
-		Src:       pk.Src,
-		Dst:       pk.Dst,
-		Responder: responder,
-	}
-	icmpFlow = newICMPFlow(conn, &flow, uint16(echo.ID), ip.logger)
-	go func() {
-		defer ip.srcToFlowTracker.delete(pk.Src)
-
-		if err := icmpFlow.serve(ip.shutdownC, defaultCloseAfterIdle); err != nil {
-			ip.logger.Debug().Err(err).Uint16("flowID", icmpFlow.echoID).Msg("flow terminated")
-		}
-	}()
-	ip.srcToFlowTracker.set(pk.Src, icmpFlow)
-	return icmpFlow.send(pk)
-}
-
-type srcIPFlowID netip.Addr
-
-func (sifd srcIPFlowID) Type() string {
-	return "srcIP"
-}
-
-func (sifd srcIPFlowID) String() string {
-	return netip.Addr(sifd).String()
-}
-
-type srcToFlowTracker struct {
-	lock sync.RWMutex
-	// srcIPToConn tracks source IP to ICMP connection
-	srcToFlow map[netip.Addr]*icmpFlow
-}
-
-func newSrcToConnTracker() *srcToFlowTracker {
-	return &srcToFlowTracker{
-		srcToFlow: make(map[netip.Addr]*icmpFlow),
-	}
-}
-
-func (sft *srcToFlowTracker) get(srcIP netip.Addr) (*icmpFlow, bool) {
-	sft.lock.RLock()
-	defer sft.lock.RUnlock()
-
-	flow, ok := sft.srcToFlow[srcIP]
-	return flow, ok
-}
-
-func (sft *srcToFlowTracker) set(srcIP netip.Addr, flow *icmpFlow) {
-	sft.lock.Lock()
-	defer sft.lock.Unlock()
-
-	sft.srcToFlow[srcIP] = flow
-}
-
-func (sft *srcToFlowTracker) delete(srcIP netip.Addr) {
-	sft.lock.Lock()
-	defer sft.lock.Unlock()
-
-	delete(sft.srcToFlow, srcIP)
-}
-
-type icmpFlow struct {
-	conn   *icmp.PacketConn
-	flow   *packet.Flow
-	echoID uint16
-	// last active unix time. Unit is seconds
-	lastActive int64
-	logger     *zerolog.Logger
-}
-
-func newICMPFlow(conn *icmp.PacketConn, flow *packet.Flow, echoID uint16, logger *zerolog.Logger) *icmpFlow {
-	return &icmpFlow{
-		conn:       conn,
-		flow:       flow,
-		echoID:     echoID,
-		lastActive: time.Now().Unix(),
-		logger:     logger,
-	}
-}
-
-func (f *icmpFlow) serve(shutdownC chan struct{}, closeAfterIdle time.Duration) error {
-	errC := make(chan error)
-	go func() {
-		errC <- f.listenResponse()
-	}()
-
-	checkIdleTicker := time.NewTicker(closeAfterIdle)
-	defer f.conn.Close()
-	defer checkIdleTicker.Stop()
-	for {
-		select {
-		case err := <-errC:
-			return err
-		case <-shutdownC:
-			return nil
-		case <-checkIdleTicker.C:
-			now := time.Now().Unix()
-			lastActive := atomic.LoadInt64(&f.lastActive)
-			if now > lastActive+int64(closeAfterIdle.Seconds()) {
-				return errFlowInactive
-			}
-		}
-	}
-}
-
-func (f *icmpFlow) send(pk *packet.ICMP) error {
-	f.updateLastActive()
-
-	// For IPv4, the pseudoHeader is not used because the checksum is always calculated
-	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 = f.conn.WriteTo(serializedMsg, &net.UDPAddr{
-		IP: pk.Dst.AsSlice(),
-	})
-	return err
-}
-
-func (f *icmpFlow) listenResponse() error {
-	buf := make([]byte, mtu)
-	encoder := packet.NewEncoder()
-	for {
-		n, src, err := f.conn.ReadFrom(buf)
 		if err != nil {
 			return err
 		}
-		f.updateLastActive()
+		conn, err := newICMPConn(ip.listenIP)
+		if err != nil {
+			return errors.Wrap(err, "failed to open ICMP socket")
+		}
+		localUDPAddr, ok := conn.LocalAddr().(*net.UDPAddr)
+		if !ok {
+			return 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())
+		if replaced := ip.srcFunnelTracker.Register(funnelID, icmpFlow); replaced {
+			ip.logger.Info().Str("src", pk.Src.String()).Msg("Replaced funnel")
+		}
+		if err := icmpFlow.sendToDst(pk.Dst, pk.Message); err != nil {
+			return errors.Wrap(err, "failed to send ICMP echo request")
+		}
+		go func() {
+			defer ip.srcFunnelTracker.Unregister(funnelID, icmpFlow)
+			if err := ip.listenResponse(icmpFlow, conn); err != nil {
+				ip.logger.Err(err).
+					Str("funnelID", funnelID.String()).
+					Int("echoID", echoID).
+					Msg("Failed to listen for ICMP echo response")
+			}
+		}()
+		return nil
+	}
+	icmpFlow, err := toICMPEchoFlow(funnel)
+	if err != nil {
+		return err
+	}
+	if err := icmpFlow.sendToDst(pk.Dst, pk.Message); err != nil {
+		return errors.Wrap(err, "failed to send ICMP echo request")
+	}
+	return nil
+}
 
-		if err := f.handleResponse(encoder, src, buf[:n]); err != nil {
+func (ip *icmpProxy) Serve(ctx context.Context) error {
-			f.logger.Err(err).Str("dst", src.String()).Msg("Failed to handle ICMP response")
+	ip.srcFunnelTracker.ScheduleCleanup(ctx, ip.idleTimeout)
+	return ctx.Err()
+}
+
+func (ip *icmpProxy) listenResponse(flow *icmpEchoFlow, conn *icmp.PacketConn) error {
+	buf := make([]byte, mtu)
+	for {
+		n, src, err := conn.ReadFrom(buf)
+		if err != nil {
+			return err
+		}
+
+		if err := ip.handleResponse(flow, src, buf[:n]); err != nil {
+			ip.logger.Err(err).Str("dst", src.String()).Msg("Failed to handle ICMP response")
 			continue
 		}
 	}
 }
 
-func (f *icmpFlow) handleResponse(encoder *packet.Encoder, from net.Addr, rawPacket []byte) error {
+func (ip *icmpProxy) handleResponse(flow *icmpEchoFlow, from net.Addr, rawMsg []byte) error {
-	// TODO: TUN-6654 Check for IPv6
+	reply, err := parseReply(from, rawMsg)
-	msg, err := icmp.ParseMessage(int(layers.IPProtocolICMPv4), rawPacket)
 	if err != nil {
 		return err
 	}
-
+	return flow.returnToSrc(reply)
-	echo, ok := msg.Body.(*icmp.Echo)
-	if !ok {
-		return fmt.Errorf("received unexpected icmp type %s from non-privileged ICMP socket", msg.Type)
-	}
-
-	addrPort, err := netip.ParseAddrPort(from.String())
-	if err != nil {
-		return err
-	}
-	icmpPacket := packet.ICMP{
-		IP: &packet.IP{
-			Src:      addrPort.Addr(),
-			Dst:      f.flow.Src,
-			Protocol: layers.IPProtocol(msg.Type.Protocol()),
-		},
-		Message: &icmp.Message{
-			Type: msg.Type,
-			Code: msg.Code,
-			Body: &icmp.Echo{
-				ID:   int(f.echoID),
-				Seq:  echo.Seq,
-				Data: echo.Data,
-			},
-		},
-	}
-	serializedPacket, err := encoder.Encode(&icmpPacket)
-	if err != nil {
-		return errors.Wrap(err, "Failed to encode ICMP message")
-	}
-	if err := f.flow.Responder.SendPacket(serializedPacket); err != nil {
-		return errors.Wrap(err, "Failed to send packet to the edge")
-	}
-	return nil
 }
 
-func (f *icmpFlow) updateLastActive() {
+// originSender wraps icmp.PacketConn to implement packet.FunnelUniPipe interface
-	atomic.StoreInt64(&f.lastActive, time.Now().Unix())
+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()
+}
+
+type srcIPFunnelID netip.Addr
+
+func (sifd srcIPFunnelID) Type() string {
+	return "srcIP"
+}
+
+func (sifd srcIPFunnelID) String() string {
+	return netip.Addr(sifd).String()
 }

ingress/icmp_linux_test.go

@@ -1,52 +0,0 @@
-//go:build linux
-
-package ingress
-
-import (
-	"errors"
-	"net"
-	"net/netip"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-
-	"github.com/cloudflare/cloudflared/packet"
-)
-
-func TestCloseIdleFlow(t *testing.T) {
-	const (
-		echoID      = 19234
-		idleTimeout = time.Millisecond * 100
-	)
-	conn, err := newICMPConn(localhostIP)
-	require.NoError(t, err)
-	flow := packet.Flow{
-		Src: netip.MustParseAddr("172.16.0.1"),
-	}
-	icmpFlow := newICMPFlow(conn, &flow, echoID, &noopLogger)
-	shutdownC := make(chan struct{})
-	flowErr := make(chan error)
-	go func() {
-		flowErr <- icmpFlow.serve(shutdownC, idleTimeout)
-	}()
-
-	require.Equal(t, errFlowInactive, <-flowErr)
-}
-
-func TestCloseConnStopFlow(t *testing.T) {
-	const (
-		echoID = 19234
-	)
-	conn, err := newICMPConn(localhostIP)
-	require.NoError(t, err)
-	flow := packet.Flow{
-		Src: netip.MustParseAddr("172.16.0.1"),
-	}
-	icmpFlow := newICMPFlow(conn, &flow, echoID, &noopLogger)
-	shutdownC := make(chan struct{})
-	conn.Close()
-
-	err = icmpFlow.serve(shutdownC, defaultCloseAfterIdle)
-	require.True(t, errors.Is(err, net.ErrClosed))
-}

ingress/icmp_posix.go

@@ -0,0 +1,142 @@
+//go:build darwin || linux
+
+package ingress
+
+// This file extracts logic shared by Linux and Darwin implementation if ICMPProxy.
+
+import (
+	"fmt"
+	"net"
+	"net/netip"
+
+	"github.com/google/gopacket/layers"
+	"golang.org/x/net/icmp"
+
+	"github.com/cloudflare/cloudflared/packet"
+)
+
+// Opens a non-privileged ICMP socket on Linux and Darwin
+func newICMPConn(listenIP netip.Addr) (*icmp.PacketConn, error) {
+	network := "udp6"
+	if listenIP.Is4() {
+		network = "udp4"
+	}
+	return icmp.ListenPacket(network, listenIP.String())
+}
+
+func netipAddr(addr net.Addr) (netip.Addr, bool) {
+	udpAddr, ok := addr.(*net.UDPAddr)
+	if !ok {
+		return netip.Addr{}, false
+	}
+	return netip.AddrFromSlice(udpAddr.IP)
+}
+
+type flowID struct {
+	srcIP  netip.Addr
+	echoID int
+}
+
+func (fi *flowID) Type() string {
+	return "srcIP_echoID"
+}
+
+func (fi *flowID) String() string {
+	return fmt.Sprintf("%s:%d", fi.srcIP, fi.echoID)
+}
+
+// icmpEchoFlow implements the packet.Funnel interface.
+type icmpEchoFlow struct {
+	*packet.RawPacketFunnel
+	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 {
+	return &icmpEchoFlow{
+		RawPacketFunnel: packet.NewRawPacketFunnel(src, sendPipe, returnPipe),
+		assignedEchoID:  assignedEchoID,
+		originalEchoID:  originalEchoID,
+		respEncoder:     respEncoder,
+	}
+}
+
+// sendToDst rewrites the echo ID to the one assigned to this flow
+func (ief *icmpEchoFlow) sendToDst(dst netip.Addr, msg *icmp.Message) error {
+	originalEcho, err := getICMPEcho(msg)
+	if err != nil {
+		return err
+	}
+	sendMsg := icmp.Message{
+		Type: msg.Type,
+		Code: msg.Code,
+		Body: &icmp.Echo{
+			ID:   ief.assignedEchoID,
+			Seq:  originalEcho.Seq,
+			Data: originalEcho.Data,
+		},
+	}
+	// For IPv4, the pseudoHeader is not used because the checksum is always calculated
+	var pseudoHeader []byte = nil
+	serializedPacket, err := sendMsg.Marshal(pseudoHeader)
+	if err != nil {
+		return err
+	}
+	return ief.SendToDst(dst, packet.RawPacket{Data: serializedPacket})
+}
+
+// returnToSrc rewrites the echo ID to the original echo ID from the eyeball
+func (ief *icmpEchoFlow) returnToSrc(reply *echoReply) error {
+	reply.echo.ID = ief.originalEchoID
+	reply.msg.Body = reply.echo
+	pk := packet.ICMP{
+		IP: &packet.IP{
+			Src:      reply.from,
+			Dst:      ief.Src,
+			Protocol: layers.IPProtocol(reply.msg.Type.Protocol()),
+		},
+		Message: reply.msg,
+	}
+	serializedPacket, err := ief.respEncoder.Encode(&pk)
+	if err != nil {
+		return err
+	}
+	return ief.ReturnToSrc(serializedPacket)
+}
+
+type echoReply struct {
+	from netip.Addr
+	msg  *icmp.Message
+	echo *icmp.Echo
+}
+
+func parseReply(from net.Addr, rawMsg []byte) (*echoReply, error) {
+	// TODO: TUN-6654 Check for IPv6
+	msg, err := icmp.ParseMessage(int(layers.IPProtocolICMPv4), rawMsg)
+	if err != nil {
+		return nil, err
+	}
+	echo, err := getICMPEcho(msg)
+	if err != nil {
+		return nil, err
+	}
+	fromAddr, ok := netipAddr(from)
+	if !ok {
+		return nil, fmt.Errorf("cannot convert %s to netip.Addr", from)
+	}
+	return &echoReply{
+		from: fromAddr,
+		msg:  msg,
+		echo: echo,
+	}, nil
+}
+
+func toICMPEchoFlow(funnel packet.Funnel) (*icmpEchoFlow, error) {
+	icmpFlow, ok := funnel.(*icmpEchoFlow)
+	if !ok {
+		return nil, fmt.Errorf("%v is not *ICMPEchoFunnel", funnel)
+	}
+	return icmpFlow, nil
+}

ingress/icmp_posix_test.go

@@ -0,0 +1,76 @@
+//go:build darwin || linux
+
+package ingress
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"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"
+)
+
+func TestFunnelIdleTimeout(t *testing.T) {
+	const (
+		idleTimeout = time.Second
+		echoID      = 42573
+		startSeq    = 8129
+	)
+	logger := zerolog.New(os.Stderr)
+	proxy, err := newICMPProxy(localhostIP, &logger, idleTimeout)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	proxyDone := make(chan struct{})
+	go func() {
+		proxy.Serve(ctx)
+		close(proxyDone)
+	}()
+
+	// Send a packet to register the flow
+	pk := packet.ICMP{
+		IP: &packet.IP{
+			Src:      localhostIP,
+			Dst:      localhostIP,
+			Protocol: layers.IPProtocolICMPv4,
+		},
+		Message: &icmp.Message{
+			Type: ipv4.ICMPTypeEcho,
+			Code: 0,
+			Body: &icmp.Echo{
+				ID:   echoID,
+				Seq:  startSeq,
+				Data: []byte(t.Name()),
+			},
+		},
+	}
+	responder := echoFlowResponder{
+		decoder:  packet.NewICMPDecoder(),
+		respChan: make(chan []byte),
+	}
+	require.NoError(t, proxy.Request(&pk, &responder))
+	responder.validate(t, &pk)
+
+	// Send second request, should reuse the funnel
+	require.NoError(t, proxy.Request(&pk, nil))
+	responder.validate(t, &pk)
+
+	time.Sleep(idleTimeout * 2)
+	newResponder := echoFlowResponder{
+		decoder:  packet.NewICMPDecoder(),
+		respChan: make(chan []byte),
+	}
+	require.NoError(t, proxy.Request(&pk, &newResponder))
+	newResponder.validate(t, &pk)
+
+	cancel()
+	<-proxyDone
+}

ingress/icmp_windows.go

@@ -15,6 +15,7 @@ import (
 	"runtime/debug"
 	"sync"
 	"syscall"
+	"time"
 	"unsafe"
 
 	"github.com/google/gopacket/layers"
@@ -143,7 +144,7 @@ type icmpProxy struct {
 	encoderPool sync.Pool
 }
 
-func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger, idleTimeout time.Duration) (ICMPProxy, error) {
 	handle, _, err := IcmpCreateFile_proc.Call()
 	// Windows procedure calls always return non-nil error constructed from the result of GetLastError.
 	// Caller need to inspect the primary returned value
@@ -167,7 +168,7 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 	return ctx.Err()
 }
 
-func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error {
 	if pk == nil {
 		return errPacketNil
 	}
@@ -176,7 +177,7 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) er
 			ip.logger.Error().Interface("error", r).Msgf("Recover panic from sending icmp request/response, error %s", debug.Stack())
 		}
 	}()
-	echo, err := getICMPEcho(pk)
+	echo, err := getICMPEcho(pk.Message)
 	if err != nil {
 		return err
 	}
@@ -193,7 +194,7 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) er
 	return nil
 }
 
-func (ip *icmpProxy) handleEchoResponse(request *packet.ICMP, echoReq *icmp.Echo, resp *echoResp, responder packet.FlowResponder) error {
+func (ip *icmpProxy) handleEchoResponse(request *packet.ICMP, echoReq *icmp.Echo, resp *echoResp, responder packet.FunnelUniPipe) error {
 	var replyType icmp.Type
 	if request.Dst.Is4() {
 		replyType = ipv4.ICMPTypeEchoReply
@@ -222,7 +223,7 @@ func (ip *icmpProxy) handleEchoResponse(request *packet.ICMP, echoReq *icmp.Echo
 	if err != nil {
 		return err
 	}
-	return responder.SendPacket(serializedPacket)
+	return responder.SendPacket(request.Src, serializedPacket)
 }
 
 func (ip *icmpProxy) encodeICMPReply(pk *packet.ICMP) (packet.RawPacket, error) {
@@ -262,7 +263,7 @@ func (ip *icmpProxy) icmpSendEcho(dst netip.Addr, echo *icmp.Echo) (*echoResp, e
 	}
 	replyCount, _, err := IcmpSendEcho_proc.Call(ip.handle, uintptr(inAddr), uintptr(unsafe.Pointer(&echo.Data[0])),
 		uintptr(dataSize), noIPHeaderOption, uintptr(unsafe.Pointer(&replyBuf[0])),
-		replySize, icmpTimeoutMs)
+		replySize, icmpRequestTimeoutMs)
 	if replyCount == 0 {
 		// status is returned in 5th to 8th byte of reply buffer
 		if status, err := unmarshalIPStatus(replyBuf[4:8]); err == nil {

ingress/icmp_windows_test.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"net/netip"
 	"testing"
+	"time"
 	"unsafe"
 
 	"golang.org/x/net/icmp"
@@ -72,7 +73,7 @@ func TestParseEchoReply(t *testing.T) {
 
 //  TestSendEchoErrors makes sure icmpSendEcho handles error cases
 func TestSendEchoErrors(t *testing.T) {
-	proxy, err := newICMPProxy(localhostIP, &noopLogger)
+	proxy, err := newICMPProxy(localhostIP, &noopLogger, time.Second)
 	require.NoError(t, err)
 	winProxy := proxy.(*icmpProxy)
 

ingress/origin_icmp_proxy.go

@@ -13,14 +13,15 @@ import (
 )
 
 const (
-	defaultCloseAfterIdle = time.Second * 15
+	// funnelIdleTimeout controls how long to wait to close a funnel without send/return
-	mtu                   = 1500
+	funnelIdleTimeout = time.Second * 10
-	icmpTimeoutMs         = 1000
+	mtu               = 1500
+	// icmpRequestTimeoutMs controls how long to wait for a reply
+	icmpRequestTimeoutMs = 1000
 )
 
 var (
-	errFlowInactive = fmt.Errorf("flow is inactive")
+	errPacketNil = fmt.Errorf("packet is nil")
-	errPacketNil    = fmt.Errorf("packet is nil")
 )
 
 // ICMPProxy sends ICMP messages and listens for their responses
@@ -28,26 +29,17 @@ type ICMPProxy interface {
 	// Serve starts listening for responses to the requests until context is done
 	Serve(ctx context.Context) error
 	// Request sends an ICMP message
-	Request(pk *packet.ICMP, responder packet.FlowResponder) error
+	Request(pk *packet.ICMP, responder packet.FunnelUniPipe) error
 }
 
 func NewICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
-	return newICMPProxy(listenIP, logger)
+	return newICMPProxy(listenIP, logger, funnelIdleTimeout)
 }
 
-// Opens a non-privileged ICMP socket on Linux and Darwin
+func getICMPEcho(msg *icmp.Message) (*icmp.Echo, error) {
-func newICMPConn(listenIP netip.Addr) (*icmp.PacketConn, error) {
+	echo, ok := msg.Body.(*icmp.Echo)
-	network := "udp6"
-	if listenIP.Is4() {
-		network = "udp4"
-	}
-	return icmp.ListenPacket(network, listenIP.String())
-}
-
-func getICMPEcho(pk *packet.ICMP) (*icmp.Echo, error) {
-	echo, ok := pk.Message.Body.(*icmp.Echo)
 	if !ok {
-		return nil, fmt.Errorf("expect ICMP echo, got %s", pk.Type)
+		return nil, fmt.Errorf("expect ICMP echo, got %s", msg.Type)
 	}
 	return echo, nil
 }

ingress/origin_icmp_proxy_test.go

@@ -3,6 +3,7 @@ package ingress
 import (
 	"context"
 	"fmt"
+	"net"
 	"net/netip"
 	"testing"
 
@@ -47,26 +48,48 @@ func TestICMPProxyEcho(t *testing.T) {
 		respChan: make(chan []byte, 1),
 	}
 
-	ip := packet.IP{
+	ips := []packet.IP{
-		Src:      localhostIP,
+		{
-		Dst:      localhostIP,
+			Src:      localhostIP,
-		Protocol: layers.IPProtocolICMPv4,
+			Dst:      localhostIP,
+			Protocol: layers.IPProtocolICMPv4,
+		},
 	}
-	for i := 0; i < endSeq; i++ {
+
-		pk := packet.ICMP{
+	addrs, err := net.InterfaceAddrs()
-			IP: &ip,
+	require.NoError(t, err)
-			Message: &icmp.Message{
+	for _, addr := range addrs {
-				Type: ipv4.ICMPTypeEcho,
+		if ipnet, ok := addr.(*net.IPNet); ok {
-				Code: 0,
+			ip := ipnet.IP
-				Body: &icmp.Echo{
+			if !ipnet.IP.IsLoopback() && ip.IsPrivate() && ip.To4() != nil {
-					ID:   echoID,
+				localIP := netip.MustParseAddr(ipnet.IP.String())
-					Seq:  i,
+				ips = append(ips, packet.IP{
-					Data: []byte(fmt.Sprintf("icmp echo seq %d", i)),
+					Src:      localIP,
-				},
+					Dst:      localIP,
-			},
+					Protocol: layers.IPProtocolICMPv4,
+				})
+				break
+			}
+		}
+	}
+
+	for seq := 0; seq < endSeq; seq++ {
+		for i, ip := range ips {
+			pk := packet.ICMP{
+				IP: &ip,
+				Message: &icmp.Message{
+					Type: ipv4.ICMPTypeEcho,
+					Code: 0,
+					Body: &icmp.Echo{
+						ID:   echoID + i,
+						Seq:  seq,
+						Data: []byte(fmt.Sprintf("icmp echo seq %d", seq)),
+					},
+				},
+			}
+			require.NoError(t, proxy.Request(&pk, &responder))
+			responder.validate(t, &pk)
 		}
-		require.NoError(t, proxy.Request(&pk, &responder))
-		responder.validate(t, &pk)
 	}
 	cancel()
 	<-proxyDone
@@ -123,13 +146,18 @@ type echoFlowResponder struct {
 	respChan chan []byte
 }
 
-func (efr *echoFlowResponder) SendPacket(pk packet.RawPacket) error {
+func (efr *echoFlowResponder) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
 	copiedPacket := make([]byte, len(pk.Data))
 	copy(copiedPacket, pk.Data)
 	efr.respChan <- copiedPacket
 	return nil
 }
 
+func (efr *echoFlowResponder) Close() error {
+	close(efr.respChan)
+	return nil
+}
+
 func (efr *echoFlowResponder) validate(t *testing.T, echoReq *packet.ICMP) {
 	pk := <-efr.respChan
 	decoded, err := efr.decoder.Decode(packet.RawPacket{Data: pk})

packet/flow.go

@@ -1,94 +0,0 @@
-package packet
-
-import (
-	"errors"
-	"fmt"
-	"net/netip"
-	"sync"
-)
-
-var (
-	ErrFlowNotFound = errors.New("flow not found")
-)
-
-// FlowID represents a key type that can be used by FlowTracker
-type FlowID interface {
-	// Type returns the name of the type that implements the FlowID
-	Type() string
-	fmt.Stringer
-}
-
-type Flow struct {
-	Src       netip.Addr
-	Dst       netip.Addr
-	Responder FlowResponder
-}
-
-func isSameFlow(f1, f2 *Flow) bool {
-	if f1 == nil || f2 == nil {
-		return false
-	}
-	return *f1 == *f2
-}
-
-// FlowResponder sends response packets to the flow
-type FlowResponder interface {
-	// SendPacket returns a packet to the flow. It must not modify the packet,
-	// and after return it must not read the packet
-	SendPacket(pk RawPacket) error
-}
-
-// FlowTracker tracks flow from the perspective of eyeball to origin
-type FlowTracker struct {
-	lock  sync.RWMutex
-	flows map[FlowID]*Flow
-}
-
-func NewFlowTracker() *FlowTracker {
-	return &FlowTracker{
-		flows: make(map[FlowID]*Flow),
-	}
-}
-
-func (sft *FlowTracker) Get(id FlowID) (*Flow, bool) {
-	sft.lock.RLock()
-	defer sft.lock.RUnlock()
-	flow, ok := sft.flows[id]
-	return flow, ok
-}
-
-// Registers a flow. If shouldReplace = true, replace the current flow
-func (sft *FlowTracker) Register(id FlowID, flow *Flow, shouldReplace bool) (replaced bool) {
-	sft.lock.Lock()
-	defer sft.lock.Unlock()
-	currentFlow, ok := sft.flows[id]
-	if !ok {
-		sft.flows[id] = flow
-		return false
-	}
-
-	if shouldReplace && !isSameFlow(currentFlow, flow) {
-		sft.flows[id] = flow
-		return true
-	}
-	return false
-}
-
-// Unregisters a flow. If force = true, delete it even if it maps to a different flow
-func (sft *FlowTracker) Unregister(id FlowID, flow *Flow, force bool) (forceDeleted bool) {
-	sft.lock.Lock()
-	defer sft.lock.Unlock()
-	currentFlow, ok := sft.flows[id]
-	if !ok {
-		return false
-	}
-	if isSameFlow(currentFlow, flow) {
-		delete(sft.flows, id)
-		return false
-	}
-	if force {
-		delete(sft.flows, id)
-		return true
-	}
-	return false
-}

packet/funnel.go

@@ -0,0 +1,191 @@
+package packet
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+var (
+	ErrFunnelNotFound = errors.New("funnel not found")
+)
+
+// Funnel is an abstraction to pipe from 1 src to 1 or more destinations
+type Funnel interface {
+	// SendToDst sends a raw packet to a destination
+	SendToDst(dst netip.Addr, pk RawPacket) error
+	// ReturnToSrc returns a raw packet to the source
+	ReturnToSrc(pk RawPacket) error
+	// LastActive returns the last time SendToDst or ReturnToSrc is called
+	LastActive() time.Time
+	// Close closes the funnel. Further call to SendToDst or ReturnToSrc should return an error
+	Close() error
+	// Equal compares if 2 funnels are equivalent
+	Equal(other Funnel) bool
+}
+
+// FunnelUniPipe is a unidirectional pipe for sending raw packets
+type FunnelUniPipe interface {
+	// SendPacket sends a packet to/from the funnel. It must not modify the packet,
+	// and after return it must not read the packet
+	SendPacket(dst netip.Addr, pk RawPacket) error
+	Close() error
+}
+
+// RawPacketFunnel is an implementation of Funnel that sends raw packets. It can be embedded in other structs to
+// satisfy the Funnel interface.
+type RawPacketFunnel struct {
+	Src netip.Addr
+	// last active unix time. Unit is seconds
+	lastActive int64
+	sendPipe   FunnelUniPipe
+	returnPipe FunnelUniPipe
+}
+
+func NewRawPacketFunnel(src netip.Addr, sendPipe, returnPipe FunnelUniPipe) *RawPacketFunnel {
+	return &RawPacketFunnel{
+		Src:        src,
+		lastActive: time.Now().Unix(),
+		sendPipe:   sendPipe,
+		returnPipe: returnPipe,
+	}
+}
+
+func (rpf *RawPacketFunnel) SendToDst(dst netip.Addr, pk RawPacket) error {
+	rpf.updateLastActive()
+	return rpf.sendPipe.SendPacket(dst, pk)
+}
+
+func (rpf *RawPacketFunnel) ReturnToSrc(pk RawPacket) error {
+	rpf.updateLastActive()
+	return rpf.returnPipe.SendPacket(rpf.Src, pk)
+}
+
+func (rpf *RawPacketFunnel) updateLastActive() {
+	atomic.StoreInt64(&rpf.lastActive, time.Now().Unix())
+}
+
+func (rpf *RawPacketFunnel) LastActive() time.Time {
+	lastActive := atomic.LoadInt64(&rpf.lastActive)
+	return time.Unix(lastActive, 0)
+}
+
+func (rpf *RawPacketFunnel) Close() error {
+	sendPipeErr := rpf.sendPipe.Close()
+	returnPipeErr := rpf.returnPipe.Close()
+	if sendPipeErr != nil {
+		return sendPipeErr
+	}
+	if returnPipeErr != nil {
+		return returnPipeErr
+	}
+	return nil
+}
+
+func (rpf *RawPacketFunnel) Equal(other Funnel) bool {
+	otherRawFunnel, ok := other.(*RawPacketFunnel)
+	if !ok {
+		return false
+	}
+	if rpf.Src != otherRawFunnel.Src {
+		return false
+	}
+	if rpf.sendPipe != otherRawFunnel.sendPipe {
+		return false
+	}
+	if rpf.returnPipe != otherRawFunnel.returnPipe {
+		return false
+	}
+	return true
+}
+
+// FunnelID represents a key type that can be used by FunnelTracker
+type FunnelID interface {
+	// Type returns the name of the type that implements the FunnelID
+	Type() string
+	fmt.Stringer
+}
+
+// FunnelTracker tracks funnel from the perspective of eyeball to origin
+type FunnelTracker struct {
+	lock    sync.RWMutex
+	funnels map[FunnelID]Funnel
+}
+
+func NewFunnelTracker() *FunnelTracker {
+	return &FunnelTracker{
+		funnels: make(map[FunnelID]Funnel),
+	}
+}
+
+func (ft *FunnelTracker) ScheduleCleanup(ctx context.Context, idleTimeout time.Duration) {
+	checkIdleTicker := time.NewTicker(idleTimeout)
+	defer checkIdleTicker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-checkIdleTicker.C:
+			ft.cleanup(idleTimeout)
+		}
+	}
+}
+
+func (ft *FunnelTracker) cleanup(idleTimeout time.Duration) {
+	ft.lock.Lock()
+	defer ft.lock.Unlock()
+
+	now := time.Now()
+	for id, funnel := range ft.funnels {
+		lastActive := funnel.LastActive()
+		if now.After(lastActive.Add(idleTimeout)) {
+			funnel.Close()
+			delete(ft.funnels, id)
+		}
+	}
+}
+
+func (ft *FunnelTracker) Get(id FunnelID) (Funnel, bool) {
+	ft.lock.RLock()
+	defer ft.lock.RUnlock()
+	funnel, ok := ft.funnels[id]
+	return funnel, ok
+}
+
+// Registers a funnel. It replaces the current funnel.
+func (ft *FunnelTracker) Register(id FunnelID, funnel Funnel) (replaced bool) {
+	ft.lock.Lock()
+	defer ft.lock.Unlock()
+	currentFunnel, exists := ft.funnels[id]
+	if !exists {
+		ft.funnels[id] = funnel
+		return false
+	}
+	replaced = !currentFunnel.Equal(funnel)
+	if replaced {
+		currentFunnel.Close()
+	}
+	ft.funnels[id] = funnel
+	return replaced
+}
+
+// Unregisters a funnel if the funnel equals to the current funnel
+func (ft *FunnelTracker) Unregister(id FunnelID, funnel Funnel) (deleted bool) {
+	ft.lock.Lock()
+	defer ft.lock.Unlock()
+	currentFunnel, exists := ft.funnels[id]
+	if !exists {
+		return true
+	}
+	if currentFunnel.Equal(funnel) {
+		currentFunnel.Close()
+		delete(ft.funnels, id)
+		return true
+	}
+	return false
+}