Release 2022.9.0

I can only reproduce the flakiness, which is the hello world still
responding when it should be shut down already, in Windows (both in
TeamCity as well as my local VM). Locally, it only happens when the
machine is under high load.

Anyway, it's valid that the proxies take some time to shut down since
they handle that via channels asynchronously with regards to the event
that updates the configuration.
Hence, nothing is wrong, as long as they eventually shut down, which the
test still verifies.

.github/workflows/check.yaml

@@ -4,7 +4,7 @@ jobs:
   check:
     strategy:
       matrix:
-        go-version: [1.17.x]
+        go-version: [1.19.x]
         os: [ubuntu-latest, macos-latest, windows-latest]
     runs-on: ${{ matrix.os }}
     steps:

CHANGES.md

@@ -1,3 +1,7 @@
+## 2022.9.0
+### New Features
+- cloudflared now rejects ingress rules with invalid http status codes for http_status.
+
 ## 2022.8.1
 ### New Features
 - cloudflared now remembers if it connected to a certain protocol successfully. If it did, it does not fall back to a lower

RELEASE_NOTES

@@ -1,3 +1,31 @@
+2022.9.0
+- 2022-09-05 TUN-6737: Fix datagramV2Type should be declared in its own block so it starts at 0
+- 2022-09-01 TUN-6725: Fix testProxySSEAllData
+- 2022-09-01 TUN-6726: Fix maxDatagramPayloadSize for Windows QUIC datagrams
+- 2022-09-01 TUN-6729: Fix flaky TestClosePreviousProxies
+- 2022-09-01 TUN-6728: Verify http status code ingress rule
+- 2022-08-25 TUN-6695: Implement ICMP proxy for linux
+
+2022.8.4
+- 2022-08-31 TUN-6717: Update Github action to run with Go 1.19
+- 2022-08-31 TUN-6720: Remove forcibly closing connection during reconnect signal
+- 2022-08-29 Release 2022.8.3
+
+2022.8.3
+- 2022-08-26 TUN-6708: Fix replace flow logic
+- 2022-08-25 TUN-6705: Tunnel should retry connections forever
+- 2022-08-25 TUN-6704: Honor protocol flag when edge discovery is unreachable
+- 2022-08-25 TUN-6699: Add metric for packet too big dropped
+- 2022-08-24 TUN-6691: Properly error check for net.ErrClosed
+- 2022-08-22 TUN-6679: Allow client side of quic request to close body
+- 2022-08-22 TUN-6586: Change ICMP proxy to only build for Darwin and use echo ID to track flows
+- 2022-08-18 TUN-6530: Implement ICMPv4 proxy
+- 2022-08-17 TUN-6666: Define packet package
+- 2022-08-17 TUN-6667: DatagramMuxerV2 provides a method to receive RawPacket
+- 2022-08-16 TUN-6657: Ask for Tunnel ID and Configuration on Bug Report
+- 2022-08-16 TUN-6676: Add suport for trailers in http2 connections
+- 2022-08-11 TUN-6575: Consume cf-trace-id from incoming http2 TCP requests
+
 2022.8.2
 - 2022-08-16 TUN-6656: Docker for arm64 should not be deployed in an amd64 container
 

cmd/cloudflared/tunnel/cmd.go

@@ -375,7 +375,7 @@ func StartServer(
 		errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, quickTunnelURL, orchestrator, log)
 	}()
 
-	reconnectCh := make(chan supervisor.ReconnectSignal, 1)
+	reconnectCh := make(chan supervisor.ReconnectSignal, c.Int("ha-connections"))
 	if c.IsSet("stdin-control") {
 		log.Info().Msg("Enabling control through stdin")
 		go stdinControl(reconnectCh, log)

component-tests/test_reconnect.py

@@ -47,17 +47,12 @@ class TestReconnect:
         cloudflared.stdin.flush()
 
     def assert_reconnect(self, config, cloudflared, repeat):
-        wait_tunnel_ready(tunnel_url=config.get_url(), require_min_connections=self.default_ha_conns)
+        wait_tunnel_ready(tunnel_url=config.get_url(),
+                          require_min_connections=self.default_ha_conns)
         for _ in range(repeat):
-            for i in range(self.default_ha_conns):
+            for _ in range(self.default_ha_conns):
                 self.send_reconnect(cloudflared, self.default_reconnect_secs)
-                expect_connections = self.default_ha_conns-i-1
-                if expect_connections > 0:
-                    # Don't check if tunnel returns 200 here because there is a race condition between wait_tunnel_ready
-                    # retrying to get 200 response and reconnecting
-                    wait_tunnel_ready(require_min_connections=expect_connections)
-                else:
-                    check_tunnel_not_connected()
-
+            check_tunnel_not_connected()
             sleep(self.default_reconnect_secs * 2)
-            wait_tunnel_ready(tunnel_url=config.get_url(), require_min_connections=self.default_ha_conns)
+            wait_tunnel_ready(tunnel_url=config.get_url(),
+                              require_min_connections=self.default_ha_conns)

component-tests/util.py

@@ -15,6 +15,7 @@ from constants import METRICS_PORT, MAX_RETRIES, BACKOFF_SECS
 
 LOGGER = logging.getLogger(__name__)
 
+
 def select_platform(plat):
     return pytest.mark.skipif(
         platform.system() != plat, reason=f"Only runs on {plat}")
@@ -108,13 +109,15 @@ def _log_cloudflared_logs(cfd_logs):
             LOGGER.warning(line)
 
 
-@retry(stop_max_attempt_number=MAX_RETRIES * BACKOFF_SECS, wait_fixed=1000)
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
 def check_tunnel_not_connected():
     url = f'http://localhost:{METRICS_PORT}/ready'
 
     try:
-        resp = requests.get(url, timeout=1)
+        resp = requests.get(url, timeout=BACKOFF_SECS)
         assert resp.status_code == 503, f"Expect {url} returns 503, got {resp.status_code}"
+        assert resp.json()[
+            "readyConnections"] == 0, "Expected all connections to be terminated (pending reconnect)"
     # cloudflared might already terminate
     except requests.exceptions.ConnectionError as e:
         LOGGER.warning(f"Failed to connect to {url}, error: {e}")

ingress/icmp_darwin.go

@@ -27,7 +27,6 @@ type icmpProxy struct {
 	echoIDTracker  *echoIDTracker
 	conn           *icmp.PacketConn
 	logger         *zerolog.Logger
-	encoder        *packet.Encoder
 }
 
 // echoIDTracker tracks which ID has been assigned. It first loops through assignment from lastAssignment to then end,
@@ -112,13 +111,8 @@ func (snf echoFlowID) String() string {
 	return strconv.FormatUint(uint64(snf), 10)
 }
 
-func newICMPProxy(listenIP net.IP, logger *zerolog.Logger) (ICMPProxy, error) {
-	network := "udp6"
-	if listenIP.To4() != nil {
-		network = "udp4"
-	}
-	// Opens a non-privileged ICMP socket
-	conn, err := icmp.ListenPacket(network, listenIP.String())
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+	conn, err := newICMPConn(listenIP)
 	if err != nil {
 		return nil, err
 	}
@@ -127,11 +121,13 @@ func newICMPProxy(listenIP net.IP, logger *zerolog.Logger) (ICMPProxy, error) {
 		echoIDTracker:  newEchoIDTracker(),
 		conn:           conn,
 		logger:         logger,
-		encoder:        packet.NewEncoder(),
 	}, nil
 }
 
 func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+	if pk == nil {
+		return errPacketNil
+	}
 	switch body := pk.Message.Body.(type) {
 	case *icmp.Echo:
 		return ip.sendICMPEchoRequest(pk, body, responder)
@@ -140,12 +136,14 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) er
 	}
 }
 
-func (ip *icmpProxy) ListenResponse(ctx context.Context) error {
+// Serve listens for responses to the requests until context is done
+func (ip *icmpProxy) Serve(ctx context.Context) error {
 	go func() {
 		<-ctx.Done()
 		ip.conn.Close()
 	}()
-	buf := make([]byte, 1500)
+	buf := make([]byte, mtu)
+	encoder := packet.NewEncoder()
 	for {
 		n, src, err := ip.conn.ReadFrom(buf)
 		if err != nil {
@@ -159,7 +157,7 @@ func (ip *icmpProxy) ListenResponse(ctx context.Context) error {
 		}
 		switch body := msg.Body.(type) {
 		case *icmp.Echo:
-			if err := ip.handleEchoResponse(msg, body); err != nil {
+			if err := ip.handleEchoResponse(encoder, msg, body); err != nil {
 				ip.logger.Error().Err(err).
 					Str("src", src.String()).
 					Str("flowID", echoFlowID(body.ID).String()).
@@ -206,7 +204,7 @@ func (ip *icmpProxy) sendICMPEchoRequest(pk *packet.ICMP, echo *icmp.Echo, respo
 	return err
 }
 
-func (ip *icmpProxy) handleEchoResponse(msg *icmp.Message, echo *icmp.Echo) error {
+func (ip *icmpProxy) handleEchoResponse(encoder *packet.Encoder, msg *icmp.Message, echo *icmp.Echo) error {
 	flowID := echoFlowID(echo.ID)
 	flow, ok := ip.srcFlowTracker.Get(flowID)
 	if !ok {
@@ -220,7 +218,7 @@ func (ip *icmpProxy) handleEchoResponse(msg *icmp.Message, echo *icmp.Echo) erro
 		},
 		Message: msg,
 	}
-	serializedPacket, err := ip.encoder.Encode(&icmpPacket)
+	serializedPacket, err := encoder.Encode(&icmpPacket)
 	if err != nil {
 		return errors.Wrap(err, "Failed to encode ICMP message")
 	}

ingress/icmp_generic.go

@@ -1,15 +1,15 @@
-//go:build !darwin
+//go:build !darwin && !linux
 
 package ingress
 
 import (
 	"fmt"
-	"net"
+	"net/netip"
 	"runtime"
 
 	"github.com/rs/zerolog"
 )
 
-func newICMPProxy(listenIP net.IP, logger *zerolog.Logger) (ICMPProxy, error) {
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
 	return nil, fmt.Errorf("ICMP proxy is not implemented on %s", runtime.GOOS)
 }

ingress/icmp_linux.go

@@ -0,0 +1,267 @@
+//go:build linux
+
+package ingress
+
+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"
+
+	"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
+	listenIP         netip.Addr
+	logger           *zerolog.Logger
+	shutdownC        chan struct{}
+}
+
+func newICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
+	if err := testPermission(listenIP); err != nil {
+		return nil, err
+	}
+	return &icmpProxy{
+		srcToFlowTracker: newSrcToConnTracker(),
+		listenIP:         listenIP,
+		logger:           logger,
+		shutdownC:        make(chan struct{}),
+	}, 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
+		// change the group ID
+		return err
+	}
+	// This conn is only to test if cloudflared has permission to open this type of socket
+	conn.Close()
+	return nil
+}
+
+func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FlowResponder) error {
+	if pk == nil {
+		return errPacketNil
+	}
+	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) 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()
+
+		if err := f.handleResponse(encoder, src, buf[:n]); err != nil {
+			f.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 {
+	// TODO: TUN-6654 Check for IPv6
+	msg, err := icmp.ParseMessage(int(layers.IPProtocolICMPv4), rawPacket)
+	if err != nil {
+		return err
+	}
+
+	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() {
+	atomic.StoreInt64(&f.lastActive, time.Now().Unix())
+}

ingress/icmp_linux_test.go

@@ -0,0 +1,52 @@
+//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/ingress.go

@@ -182,11 +182,14 @@ func validateIngress(ingress []config.UnvalidatedIngressRule, defaults OriginReq
 			path := strings.TrimPrefix(r.Service, prefix)
 			service = &unixSocketPath{path: path, scheme: "https"}
 		} else if prefix := "http_status:"; strings.HasPrefix(r.Service, prefix) {
-			status, err := strconv.Atoi(strings.TrimPrefix(r.Service, prefix))
+			statusCode, err := strconv.Atoi(strings.TrimPrefix(r.Service, prefix))
 			if err != nil {
-				return Ingress{}, errors.Wrap(err, "invalid HTTP status")
+				return Ingress{}, errors.Wrap(err, "invalid HTTP status code")
 			}
-			srv := newStatusCode(status)
+			if statusCode < 100 || statusCode > 999 {
+				return Ingress{}, fmt.Errorf("invalid HTTP status code: %d", statusCode)
+			}
+			srv := newStatusCode(statusCode)
 			service = &srv
 		} else if r.Service == HelloWorldService || r.Service == "hello-world" || r.Service == "helloworld" {
 			service = new(helloWorld)

ingress/ingress_test.go

@@ -208,6 +208,14 @@ ingress:
 			args: args{rawYAML: `
 ingress:
  - service: http_status:asdf
+`},
+			wantErr: true,
+		},
+		{
+			name: "Invalid HTTP status code",
+			args: args{rawYAML: `
+ingress:
+ - service: http_status:8080
 `},
 			wantErr: true,
 		},

ingress/origin_icmp_proxy.go

@@ -2,21 +2,43 @@ package ingress
 
 import (
 	"context"
-	"net"
+	"fmt"
+	"net/netip"
+	"time"
 
 	"github.com/rs/zerolog"
+	"golang.org/x/net/icmp"
 
 	"github.com/cloudflare/cloudflared/packet"
 )
 
+const (
+	defaultCloseAfterIdle = time.Second * 15
+	mtu                   = 1500
+)
+
+var (
+	errFlowInactive = fmt.Errorf("flow is inactive")
+	errPacketNil    = fmt.Errorf("packet is nil")
+)
+
 // ICMPProxy sends ICMP messages and listens for their responses
 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
-	// ListenResponse listens for responses to the requests until context is done
-	ListenResponse(ctx context.Context) error
 }
 
-func NewICMPProxy(listenIP net.IP, logger *zerolog.Logger) (ICMPProxy, error) {
+func NewICMPProxy(listenIP netip.Addr, logger *zerolog.Logger) (ICMPProxy, error) {
 	return newICMPProxy(listenIP, logger)
 }
+
+// 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())
+}

ingress/origin_icmp_proxy_test.go

@@ -24,19 +24,19 @@ var (
 // TestICMPProxyEcho makes sure we can send ICMP echo via the Request method and receives response via the
 // ListenResponse method
 func TestICMPProxyEcho(t *testing.T) {
-	skipNonDarwin(t)
+	onlyDarwinOrLinux(t)
 	const (
 		echoID = 36571
 		endSeq = 100
 	)
 
-	proxy, err := NewICMPProxy(localhostIP.AsSlice(), &noopLogger)
+	proxy, err := NewICMPProxy(localhostIP, &noopLogger)
 	require.NoError(t, err)
 
 	proxyDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() {
-		proxy.ListenResponse(ctx)
+		proxy.Serve(ctx)
 		close(proxyDone)
 	}()
 
@@ -72,7 +72,7 @@ func TestICMPProxyEcho(t *testing.T) {
 
 // TestICMPProxyRejectNotEcho makes sure it rejects messages other than echo
 func TestICMPProxyRejectNotEcho(t *testing.T) {
-	skipNonDarwin(t)
+	onlyDarwinOrLinux(t)
 	msgs := []icmp.Message{
 		{
 			Type: ipv4.ICMPTypeDestinationUnreachable,
@@ -97,7 +97,7 @@ func TestICMPProxyRejectNotEcho(t *testing.T) {
 			},
 		},
 	}
-	proxy, err := NewICMPProxy(localhostIP.AsSlice(), &noopLogger)
+	proxy, err := NewICMPProxy(localhostIP, &noopLogger)
 	require.NoError(t, err)
 
 	responder := echoFlowResponder{
@@ -117,8 +117,8 @@ func TestICMPProxyRejectNotEcho(t *testing.T) {
 	}
 }
 
-func skipNonDarwin(t *testing.T) {
-	if runtime.GOOS != "darwin" {
+func onlyDarwinOrLinux(t *testing.T) {
+	if runtime.GOOS != "darwin" && runtime.GOOS != "linux" {
 		t.Skip("Cannot create non-privileged datagram-oriented ICMP endpoint on Windows")
 	}
 }
@@ -146,6 +146,5 @@ func (efr *echoFlowResponder) validate(t *testing.T, echoReq *packet.ICMP) {
 	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)
 }

orchestration/orchestrator_test.go

@@ -500,7 +500,8 @@ func TestClosePreviousProxies(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, http.StatusTeapot, resp.StatusCode)
 
-	// The hello-world server in config v1 should have been stopped
+	// The hello-world server in config v1 should have been stopped. We wait a bit since it's closed asynchronously.
+	time.Sleep(time.Millisecond * 10)
 	resp, err = proxyHTTP(originProxyV1, hostname)
 	require.Error(t, err)
 	require.Nil(t, resp)

proxy/proxy_test.go

@@ -130,6 +130,10 @@ func (w *mockSSERespWriter) Write(data []byte) (int, error) {
 	return len(data), nil
 }
 
+func (w *mockSSERespWriter) WriteString(str string) (int, error) {
+	return w.Write([]byte(str))
+}
+
 func (w *mockSSERespWriter) ReadBytes() []byte {
 	return <-w.writeNotification
 }
@@ -156,7 +160,6 @@ func TestProxySingleOrigin(t *testing.T) {
 	t.Run("testProxyHTTP", testProxyHTTP(proxy))
 	t.Run("testProxyWebsocket", testProxyWebsocket(proxy))
 	t.Run("testProxySSE", testProxySSE(proxy))
-	t.Run("testProxySSEAllData", testProxySSEAllData(proxy))
 	cancel()
 }
 
@@ -276,17 +279,15 @@ func testProxySSE(proxy connection.OriginProxy) func(t *testing.T) {
 
 // Regression test to guarantee that we always write the contents downstream even if EOF is reached without
 // hitting the delimiter
-func testProxySSEAllData(proxy *Proxy) func(t *testing.T) {
-	return func(t *testing.T) {
-		eyeballReader := io.NopCloser(strings.NewReader("data\r\r"))
-		responseWriter := newMockSSERespWriter()
+func TestProxySSEAllData(t *testing.T) {
+	eyeballReader := io.NopCloser(strings.NewReader("data\r\r"))
+	responseWriter := newMockSSERespWriter()
 
-		// responseWriter uses an unbuffered channel, so we call in a different go-routine
-		go cfio.Copy(responseWriter, eyeballReader)
+	// responseWriter uses an unbuffered channel, so we call in a different go-routine
+	go cfio.Copy(responseWriter, eyeballReader)
 
-		result := string(<-responseWriter.writeNotification)
-		require.Equal(t, "data\r\r", result)
-	}
+	result := string(<-responseWriter.writeNotification)
+	require.Equal(t, "data\r\r", result)
 }
 
 func TestProxyMultipleOrigins(t *testing.T) {

quic/datagramv2.go

@@ -16,12 +16,16 @@ type datagramV2Type byte
 const (
 	udp datagramV2Type = iota
 	ip
+)
+
+const (
+	typeIDLen = 1
 	// Same as sessionDemuxChan capacity
 	packetChanCapacity = 16
 )
 
 func suffixType(b []byte, datagramType datagramV2Type) ([]byte, error) {
-	if len(b)+1 > MaxDatagramFrameSize {
+	if len(b)+typeIDLen > MaxDatagramFrameSize {
 		return nil, fmt.Errorf("datagram size %d exceeds max frame size %d", len(b), MaxDatagramFrameSize)
 	}
 	b = append(b, byte(datagramType))
@@ -114,11 +118,11 @@ func (dm *DatagramMuxerV2) ReceivePacket(ctx context.Context) (packet.RawPacket,
 }
 
 func (dm *DatagramMuxerV2) demux(ctx context.Context, msgWithType []byte) error {
-	if len(msgWithType) < 1 {
-		return fmt.Errorf("QUIC datagram should have at least 1 byte")
+	if len(msgWithType) < typeIDLen {
+		return fmt.Errorf("QUIC datagram should have at least %d byte", typeIDLen)
 	}
-	msgType := datagramV2Type(msgWithType[len(msgWithType)-1])
-	msg := msgWithType[0 : len(msgWithType)-1]
+	msgType := datagramV2Type(msgWithType[len(msgWithType)-typeIDLen])
+	msg := msgWithType[0 : len(msgWithType)-typeIDLen]
 	switch msgType {
 	case udp:
 		return dm.handleSession(ctx, msg)

quic/param_windows.go

@@ -7,5 +7,5 @@ const (
 	// 1220 is the default value https://github.com/lucas-clemente/quic-go/blob/84e03e59760ceee37359688871bb0688fcc4e98f/internal/protocol/params.go#L138
 	MaxDatagramFrameSize = 1220
 	//  3 more bytes are reserved at https://github.com/lucas-clemente/quic-go/blob/v0.24.0/internal/wire/datagram_frame.go#L61
-	maxDatagramPayloadSize = MaxDatagramFrameSize - 3 - sessionIDLen
+	maxDatagramPayloadSize = MaxDatagramFrameSize - 3 - sessionIDLen - typeIDLen
 )

supervisor/supervisor.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net"
+	"net/netip"
 	"strings"
 	"time"
 
@@ -117,9 +117,12 @@ func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrato
 		connAwareLogger:   log,
 	}
 	if useDatagramV2(config) {
-		// For non-privileged datagram-oriented ICMP endpoints, network must be "udp4" or "udp6"
 		// TODO: TUN-6654 listen for IPv6 and decide if it should listen on specific IP
-		icmpProxy, err := ingress.NewICMPProxy(net.IPv4zero, config.Log)
+		listenIP, err := netip.ParseAddr("0.0.0.0")
+		if err != nil {
+			return nil, err
+		}
+		icmpProxy, err := ingress.NewICMPProxy(listenIP, config.Log)
 		if err != nil {
 			log.Logger().Warn().Err(err).Msg("Failed to create icmp proxy, will continue to use datagram v1")
 		} else {
@@ -156,7 +159,7 @@ func (s *Supervisor) Run(
 ) error {
 	if s.edgeTunnelServer.icmpProxy != nil {
 		go func() {
-			if err := s.edgeTunnelServer.icmpProxy.ListenResponse(ctx); err != nil {
+			if err := s.edgeTunnelServer.icmpProxy.Serve(ctx); err != nil {
 				s.log.Logger().Err(err).Msg("icmp proxy terminated")
 			}
 		}()
@@ -295,8 +298,7 @@ func (s *Supervisor) initialize(
 			s.config.ProtocolSelector.Current(),
 			false,
 		}
-		ch := signal.New(make(chan struct{}))
-		go s.startTunnel(ctx, i, ch)
+		go s.startTunnel(ctx, i, s.newConnectedTunnelSignal(i))
 		time.Sleep(registrationInterval)
 	}
 	return nil

supervisor/tunnel.go

@@ -546,7 +546,13 @@ func (e *EdgeTunnelServer) serveH2mux(
 	})
 
 	errGroup.Go(func() error {
-		return listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
+		err := listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
+		if err != nil {
+			// forcefully break the connection (this is only used for testing)
+			// errgroup will return context canceled for the handler.ServeClassicTunnel
+			connLog.Logger().Debug().Msg("Forcefully breaking h2mux connection")
+		}
+		return err
 	})
 
 	return errGroup.Wait()
@@ -580,8 +586,8 @@ func (e *EdgeTunnelServer) serveHTTP2(
 		err := listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
 		if err != nil {
 			// forcefully break the connection (this is only used for testing)
+			// errgroup will return context canceled for the h2conn.Serve
 			connLog.Logger().Debug().Msg("Forcefully breaking http2 connection")
-			_ = tlsServerConn.Close()
 		}
 		return err
 	})
@@ -636,8 +642,8 @@ func (e *EdgeTunnelServer) serveQUIC(
 		err := listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
 		if err != nil {
 			// forcefully break the connection (this is only used for testing)
+			// errgroup will return context canceled for the quicConn.Serve
 			connLogger.Logger().Debug().Msg("Forcefully breaking quic connection")
-			quicConn.Close()
 		}
 		return err
 	})