Release 2022.10.2

cloudflared's Makefile uses `shell go env GOOS` to determine the
LOCAL_OS regardless of it being provided. We therefore need pinned_go as
a dependency to run docker-generate-versions.

.docker-images

@@ -2,8 +2,7 @@ images:
   - name: cloudflared
     dockerfile: Dockerfile.$ARCH
     context: .
-    versions: 
-      - latest
+    version_file: versions
     registries:
     - name: docker.io/cloudflare
       user: env:DOCKER_USER

Dockerfile

@@ -7,8 +7,6 @@ ENV GO111MODULE=on \
     TARGET_GOOS=${TARGET_GOOS} \
     TARGET_GOARCH=${TARGET_GOARCH}
     
-LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
-
 WORKDIR /go/src/github.com/cloudflare/cloudflared/
 
 # copy our sources into the builder image
@@ -20,6 +18,8 @@ RUN make cloudflared
 # use a distroless base image with glibc
 FROM gcr.io/distroless/base-debian11:nonroot
 
+LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
+
 # copy our compiled binary
 COPY --from=builder --chown=nonroot /go/src/github.com/cloudflare/cloudflared/cloudflared /usr/local/bin/
 

Dockerfile.amd64

@@ -3,8 +3,6 @@ FROM golang:1.19 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0 
     
-LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
-
 WORKDIR /go/src/github.com/cloudflare/cloudflared/
 
 # copy our sources into the builder image
@@ -16,6 +14,8 @@ RUN GOOS=linux GOARCH=amd64 make cloudflared
 # use a distroless base image with glibc
 FROM gcr.io/distroless/base-debian11:nonroot
 
+LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
+
 # copy our compiled binary
 COPY --from=builder --chown=nonroot /go/src/github.com/cloudflare/cloudflared/cloudflared /usr/local/bin/
 

Dockerfile.arm64

@@ -3,8 +3,6 @@ FROM golang:1.19 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0 
     
-LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
-
 WORKDIR /go/src/github.com/cloudflare/cloudflared/
 
 # copy our sources into the builder image
@@ -16,6 +14,8 @@ RUN GOOS=linux GOARCH=arm64 make cloudflared
 # use a distroless base image with glibc
 FROM gcr.io/distroless/base-debian11:nonroot-arm64
 
+LABEL org.opencontainers.image.source="https://github.com/cloudflare/cloudflared"
+
 # copy our compiled binary
 COPY --from=builder --chown=nonroot /go/src/github.com/cloudflare/cloudflared/cloudflared /usr/local/bin/
 

Makefile

@@ -131,6 +131,10 @@ endif
 container:
 	docker build --build-arg=TARGET_ARCH=$(TARGET_ARCH) --build-arg=TARGET_OS=$(TARGET_OS) -t cloudflare/cloudflared-$(TARGET_OS)-$(TARGET_ARCH):"$(VERSION)" .
 
+.PHONY: generate-docker-version
+generate-docker-version:
+	echo latest $(VERSION) > versions
+
 .PHONY: test
 test: vet
 ifndef CI
@@ -146,7 +150,7 @@ test-ssh-server:
 	docker-compose -f ssh_server_tests/docker-compose.yml up
 
 cloudflared.1: cloudflared_man_template
-	cat cloudflared_man_template | sed -e 's/\$${VERSION}/$(VERSION)/; s/\$${DATE}/$(DATE)/' > cloudflared.1
+	sed -e 's/\$${VERSION}/$(VERSION)/; s/\$${DATE}/$(DATE)/' cloudflared_man_template > cloudflared.1
 
 install: cloudflared cloudflared.1
 	mkdir -p $(DESTDIR)$(INSTALL_BINDIR) $(DESTDIR)$(INSTALL_MANDIR)

README.md

@@ -25,12 +25,13 @@ routing), but for legacy reasons this requirement is still necessary:
 
 ## Installing `cloudflared`
 
-Downloads are available as standalone binaries, a Docker image, and Debian, RPM, and Homebrew packages. You can also find releases here on the `cloudflared` GitHub repository.
+Downloads are available as standalone binaries, a Docker image, and Debian, RPM, and Homebrew packages. You can also find releases [here](https://github.com/cloudflare/cloudflared/releases) on the `cloudflared` GitHub repository.
 
 * You can [install on macOS](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/installation#macos) via Homebrew or by downloading the [latest Darwin amd64 release](https://github.com/cloudflare/cloudflared/releases)
 * Binaries, Debian, and RPM packages for Linux [can be found here](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/installation#linux)
 * A Docker image of `cloudflared` is [available on DockerHub](https://hub.docker.com/r/cloudflare/cloudflared)
 * You can install on Windows machines with the [steps here](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/installation#windows)
+* Build from source with the [instructions here](https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/installation#build-from-source)
 
 User documentation for Cloudflare Tunnel can be found at https://developers.cloudflare.com/cloudflare-one/connections/connect-apps
 

RELEASE_NOTES

@@ -1,3 +1,29 @@
+2022.10.2
+- 2022-10-18 TUN-6869: Fix Makefile complaining about missing GO packages
+- 2022-10-18 TUN-6864: Don't reuse port in quic unit tests
+- 2022-10-18 TUN-6868: Return left padded tracing ID when tracing identity is converted to string
+
+2022.10.1
+- 2022-10-16 TUN-6861: Trace ICMP on Windows
+- 2022-10-15 TUN-6860: Send access configuration keys to the edge
+- 2022-10-14 TUN-6858: Trace ICMP reply
+- 2022-10-13 TUN-6855: Add DatagramV2Type for IP packet with trace and tracing spans
+- 2022-10-13 TUN-6856: Refactor to lay foundation for tracing ICMP
+- 2022-10-13 TUN-6604: Trace icmp echo request on Linux and Darwin
+- 2022-10-12 Fix log message (#591)
+- 2022-10-12 TUN-6853: Reuse source port when connecting to the edge for quic connections
+- 2022-10-11 TUN-6829: Allow user of datagramsession to control logging level of errors
+- 2022-10-10 RTG-2276 Update qtls and go mod tidy
+- 2022-10-05 Add post-quantum flag to quick tunnel
+- 2022-10-05 TUN-6823: Update github release message to pull from KV
+- 2022-10-04 TUN-6825: Fix cloudflared:version images require arch hyphens
+- 2022-10-03 TUN-6806: Add ingress rule number to log when filtering due to middlware handler
+- 2022-08-17 Label correct container
+- 2022-08-16 Fix typo in help text for `cloudflared tunnel route lb`
+- 2022-07-18 drop usage of cat when sed is invoked to generate the manpage
+- 2021-03-15 update-build-readme
+- 2021-03-15 fix link
+
 2022.10.0
 - 2022-09-30 TUN-6755: Remove unused publish functions
 - 2022-09-30 TUN-6813: Only proxy ICMP packets when warp-routing is enabled

cfsetup.yaml

@@ -78,6 +78,13 @@ stretch: &stretch
       # same logic as above, but for FIPS packages only
       - ./build-packages-fips.sh
       - make github-release-built-pkgs
+  generate-versions-file:
+    build_dir: *build_dir
+    builddeps: 
+      - *pinned_go
+      - build-essential
+    post-cache:
+      - make generate-docker-version
   build-deb:
     build_dir: *build_dir
     builddeps: &build_deb_deps

cmd/cloudflared/tunnel/cmd.go

@@ -665,6 +665,13 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			EnvVars: []string{"TUNNEL_MAX_FETCH_SIZE"},
 			Hidden:  true,
 		}),
+		altsrc.NewBoolFlag(&cli.BoolFlag{
+			Name:    "post-quantum",
+			Usage:   "When given creates an experimental post-quantum secure tunnel",
+			Aliases: []string{"pq"},
+			EnvVars: []string{"TUNNEL_POST_QUANTUM"},
+			Hidden:  FipsEnabled,
+		}),
 		selectProtocolFlag,
 		overwriteDNSFlag,
 	}...)

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,

cmd/cloudflared/tunnel/subcommands.go

@@ -821,7 +821,7 @@ Further information about managing Cloudflare WARP traffic to your tunnel is ava
 				Name:        "lb",
 				Action:      cliutil.ConfiguredAction(routeLbCommand),
 				Usage:       "Use this tunnel as a load balancer origin, creating pool and load balancer if necessary",
-				UsageText:   "cloudflared tunnel route dns [TUNNEL] [HOSTNAME] [LB-POOL]",
+				UsageText:   "cloudflared tunnel route lb [TUNNEL] [HOSTNAME] [LB-POOL]",
 				Description: `Creates Load Balancer with an origin pool that points to the tunnel.`,
 			},
 			buildRouteIPSubcommand(),

connection/quic.go

@@ -10,6 +10,7 @@ import (
 	"net/netip"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -42,6 +43,11 @@ const (
 	demuxChanCapacity = 16
 )
 
+var (
+	portForConnIndex = make(map[uint8]int, 0)
+	portMapMutex     sync.Mutex
+)
+
 // QUICConnection represents the type that facilitates Proxying via QUIC streams.
 type QUICConnection struct {
 	session      quic.Connection
@@ -51,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
 }
@@ -60,22 +66,34 @@ type QUICConnection struct {
 func NewQUICConnection(
 	quicConfig *quic.Config,
 	edgeAddr net.Addr,
+	connIndex uint8,
 	tlsConfig *tls.Config,
 	orchestrator Orchestrator,
 	connOptions *tunnelpogs.ConnectionOptions,
 	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
-	packetRouterConfig *packet.GlobalRouterConfig,
+	packetRouterConfig *ingress.GlobalRouterConfig,
 ) (*QUICConnection, error) {
-	session, err := quic.DialAddr(edgeAddr.String(), tlsConfig, quicConfig)
+	udpConn, err := createUDPConnForConnIndex(connIndex, logger)
+	if err != nil {
+		return nil, err
+	}
+
+	session, err := quic.Dial(udpConn, edgeAddr, edgeAddr.String(), tlsConfig, quicConfig)
 	if err != nil {
 		return nil, &EdgeQuicDialError{Cause: err}
 	}
 
+	// wrap the session, so that the UDPConn is closed after session is closed.
+	session = &wrapCloseableConnQuicConnection{
+		session,
+		udpConn,
+	}
+
 	sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
 	datagramMuxer := quicpogs.NewDatagramMuxerV2(session, logger, sessionDemuxChan)
 	sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
-	packetRouter := packet.NewRouter(packetRouterConfig, datagramMuxer, &returnPipe{muxer: datagramMuxer}, logger, orchestrator.WarpRoutingEnabled)
+	packetRouter := ingress.NewPacketRouter(packetRouterConfig, datagramMuxer, logger, orchestrator.WarpRoutingEnabled)
 
 	return &QUICConnection{
 		session:              session,
@@ -480,15 +498,69 @@ func (np *nopCloserReadWriter) Close() error {
 	return nil
 }
 
-// returnPipe wraps DatagramMuxerV2 to satisfy the packet.FunnelUniPipe interface
-type returnPipe struct {
+// muxerWrapper wraps DatagramMuxerV2 to satisfy the packet.FunnelUniPipe interface
+type muxerWrapper struct {
 	muxer *quicpogs.DatagramMuxerV2
 }
 
-func (rp *returnPipe) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	return rp.muxer.SendPacket(pk)
+func (rp *muxerWrapper) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
+	return rp.muxer.SendPacket(quicpogs.RawPacket(pk))
 }
 
-func (rp *returnPipe) Close() error {
+func (rp *muxerWrapper) ReceivePacket(ctx context.Context) (packet.RawPacket, error) {
+	pk, err := rp.muxer.ReceivePacket(ctx)
+	if err != nil {
+		return packet.RawPacket{}, err
+	}
+	rawPacket, ok := pk.(quicpogs.RawPacket)
+	if ok {
+		return packet.RawPacket(rawPacket), nil
+	}
+	return packet.RawPacket{}, fmt.Errorf("unexpected packet type %+v", pk)
+}
+
+func (rp *muxerWrapper) Close() error {
 	return nil
 }
+
+func createUDPConnForConnIndex(connIndex uint8, logger *zerolog.Logger) (*net.UDPConn, error) {
+	portMapMutex.Lock()
+	defer portMapMutex.Unlock()
+
+	// if port was not set yet, it will be zero, so bind will randomly allocate one.
+	if port, ok := portForConnIndex[connIndex]; ok {
+		udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: port})
+		// if there wasn't an error, or if port was 0 (independently of error or not, just return)
+		if err == nil {
+			return udpConn, nil
+		} else {
+			logger.Debug().Err(err).Msgf("Unable to reuse port %d for connIndex %d. Falling back to random allocation.", port, connIndex)
+		}
+	}
+
+	// if we reached here, then there was an error or port as not been allocated it.
+	udpConn, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4zero, Port: 0})
+	if err == nil {
+		udpAddr, ok := (udpConn.LocalAddr()).(*net.UDPAddr)
+		if !ok {
+			return nil, fmt.Errorf("unable to cast to udpConn")
+		}
+		portForConnIndex[connIndex] = udpAddr.Port
+	} else {
+		delete(portForConnIndex, connIndex)
+	}
+
+	return udpConn, err
+}
+
+type wrapCloseableConnQuicConnection struct {
+	quic.Connection
+	udpConn *net.UDPConn
+}
+
+func (w *wrapCloseableConnQuicConnection) CloseWithError(errorCode quic.ApplicationErrorCode, reason string) error {
+	err := w.Connection.CloseWithError(errorCode, reason)
+	w.udpConn.Close()
+
+	return err
+}

connection/quic_test.go

@@ -11,7 +11,6 @@ import (
 	"net/url"
 	"os"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -142,24 +141,33 @@ func TestQUICServer(t *testing.T) {
 		},
 	}
 
-	for _, test := range tests {
+	for i, test := range tests {
+		test := test // capture range variable
 		t.Run(test.desc, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
-			var wg sync.WaitGroup
-			wg.Add(1)
+
+			quicListener, err := quic.Listen(udpListener, testTLSServerConfig, testQUICConfig)
+			require.NoError(t, err)
+
+			serverDone := make(chan struct{})
 			go func() {
-				defer wg.Done()
 				quicServer(
-					t, udpListener, testTLSServerConfig, testQUICConfig,
-					test.dest, test.connectionType, test.metadata, test.message, test.expectedResponse,
+					ctx, t, quicListener, test.dest, test.connectionType, test.metadata, test.message, test.expectedResponse,
 				)
+				close(serverDone)
 			}()
 
-			qc := testQUICConnection(udpListener.LocalAddr(), t)
-			go qc.Serve(ctx)
+			qc := testQUICConnection(udpListener.LocalAddr(), t, uint8(i))
 
-			wg.Wait()
+			connDone := make(chan struct{})
+			go func() {
+				qc.Serve(ctx)
+				close(connDone)
+			}()
+
+			<-serverDone
 			cancel()
+			<-connDone
 		})
 	}
 }
@@ -177,23 +185,16 @@ func (fakeControlStream) IsStopped() bool {
 }
 
 func quicServer(
+	ctx context.Context,
 	t *testing.T,
-	conn net.PacketConn,
-	tlsConf *tls.Config,
-	config *quic.Config,
+	listener quic.Listener,
 	dest string,
 	connectionType quicpogs.ConnectionType,
 	metadata []quicpogs.Metadata,
 	message []byte,
 	expectedResponse []byte,
 ) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	earlyListener, err := quic.Listen(conn, tlsConf, config)
-	require.NoError(t, err)
-
-	session, err := earlyListener.Accept(ctx)
+	session, err := listener.Accept(ctx)
 	require.NoError(t, err)
 
 	quicStream, err := session.OpenStreamSync(context.Background())
@@ -482,6 +483,7 @@ func TestBuildHTTPRequest(t *testing.T) {
 
 	log := zerolog.Nop()
 	for _, test := range tests {
+		test := test // capture range variable
 		t.Run(test.name, func(t *testing.T) {
 			req, err := buildHTTPRequest(context.Background(), test.connectRequest, test.body, &log)
 			assert.NoError(t, err)
@@ -519,7 +521,8 @@ func TestServeUDPSession(t *testing.T) {
 		edgeQUICSessionChan <- edgeQUICSession
 	}()
 
-	qc := testQUICConnection(val, t)
+	// Random index to avoid reusing port
+	qc := testQUICConnection(val, t, 28)
 	go qc.Serve(ctx)
 
 	edgeQUICSession := <-edgeQUICSessionChan
@@ -567,6 +570,37 @@ func TestNopCloserReadWriterCloseAfterEOF(t *testing.T) {
 	require.Equal(t, err, io.EOF)
 }
 
+func TestCreateUDPConnReuseSourcePort(t *testing.T) {
+	logger := zerolog.Nop()
+	conn, err := createUDPConnForConnIndex(0, &logger)
+	require.NoError(t, err)
+
+	getPortFunc := func(conn *net.UDPConn) int {
+		addr := conn.LocalAddr().(*net.UDPAddr)
+		return addr.Port
+	}
+
+	initialPort := getPortFunc(conn)
+
+	// close conn
+	conn.Close()
+
+	// should get the same port as before.
+	conn, err = createUDPConnForConnIndex(0, &logger)
+	require.NoError(t, err)
+	require.Equal(t, initialPort, getPortFunc(conn))
+
+	// new index, should get a different port
+	conn1, err := createUDPConnForConnIndex(1, &logger)
+	require.NoError(t, err)
+	require.NotEqual(t, initialPort, getPortFunc(conn1))
+
+	// not closing the conn and trying to obtain a new conn for same index should give a different random port
+	conn, err = createUDPConnForConnIndex(0, &logger)
+	require.NoError(t, err)
+	require.NotEqual(t, initialPort, getPortFunc(conn))
+}
+
 func serveSession(ctx context.Context, qc *QUICConnection, edgeQUICSession quic.Connection, closeType closeReason, expectedReason string, t *testing.T) {
 	var (
 		payload = []byte(t.Name())
@@ -672,7 +706,7 @@ func (s mockSessionRPCServer) UnregisterUdpSession(ctx context.Context, sessionI
 	return nil
 }
 
-func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection {
+func testQUICConnection(udpListenerAddr net.Addr, t *testing.T, index uint8) *QUICConnection {
 	tlsClientConfig := &tls.Config{
 		InsecureSkipVerify: true,
 		NextProtos:         []string{"argotunnel"},
@@ -682,6 +716,7 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection
 	qc, err := NewQUICConnection(
 		testQUICConfig,
 		udpListenerAddr,
+		index,
 		tlsClientConfig,
 		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
 		&tunnelpogs.ConnectionOptions{},

datagramsession/session.go

@@ -24,6 +24,13 @@ func SessionIdleErr(timeout time.Duration) error {
 
 type transportSender func(session *packet.Session) error
 
+// ErrVithVariableSeverity are errors that have variable severity
+type ErrVithVariableSeverity interface {
+	error
+	// LogLevel return the severity of this error
+	LogLevel() zerolog.Level
+}
+
 // Session is a bidirectional pipe of datagrams between transport and dstConn
 // Destination can be a connection with origin or with eyeball
 // When the destination is origin:
@@ -53,7 +60,11 @@ func (s *Session) Serve(ctx context.Context, closeAfterIdle time.Duration) (clos
 				if errors.Is(err, net.ErrClosed) {
 					s.log.Debug().Msg("Destination connection closed")
 				} else {
-					s.log.Error().Err(err).Msg("Failed to send session payload from destination to transport")
+					level := zerolog.ErrorLevel
+					if variableErr, ok := err.(ErrVithVariableSeverity); ok {
+						level = variableErr.LogLevel()
+					}
+					s.log.WithLevel(level).Err(err).Msg("Failed to send session payload from destination to transport")
 				}
 				if closeSession {
 					s.closeChan <- err

github_message.py

@@ -1,29 +1,56 @@
 #!/usr/bin/python3
 """
-Creates Github Releases Notes with content hashes
+Create Github Releases Notes with binary checksums from Workers KV
 """
 
 import argparse
 import logging
 import os
-import hashlib
-import glob
+import requests
 
-from github import Github, GithubException, UnknownObjectException
+from github import Github, UnknownObjectException
 
 FORMAT = "%(levelname)s - %(asctime)s: %(message)s"
-logging.basicConfig(format=FORMAT)
+logging.basicConfig(format=FORMAT, level=logging.INFO)
 
 CLOUDFLARED_REPO = os.environ.get("GITHUB_REPO", "cloudflare/cloudflared")
 GITHUB_CONFLICT_CODE = "already_exists"
+BASE_KV_URL = 'https://api.cloudflare.com/client/v4/accounts/'
 
-def get_sha256(filename):
-    """ get the sha256 of a file """
-    sha256_hash = hashlib.sha256()
-    with open(filename,"rb") as f:
-        for byte_block in iter(lambda: f.read(4096),b""):
-            sha256_hash.update(byte_block)
-        return sha256_hash.hexdigest()
+
+def kv_get_keys(prefix, account, namespace, api_token):
+    """ get the KV keys for a given prefix """
+    response = requests.get(
+        BASE_KV_URL + account + "/storage/kv/namespaces/" +
+        namespace + "/keys" + "?prefix=" + prefix,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": "Bearer " + api_token,
+        },
+    )
+    if response.status_code != 200:
+        jsonResponse = response.json()
+        errors = jsonResponse["errors"]
+        if len(errors) > 0:
+            raise Exception("failed to get checksums: {0}", errors[0])
+    return response.json()["result"]
+
+
+def kv_get_value(key, account, namespace, api_token):
+    """ get the KV value for a provided key """
+    response = requests.get(
+        BASE_KV_URL + account + "/storage/kv/namespaces/" + namespace + "/values/" + key,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": "Bearer " + api_token,
+        },
+    )
+    if response.status_code != 200:
+        jsonResponse = response.json()
+        errors = jsonResponse["errors"]
+        if len(errors) > 0:
+            raise Exception("failed to get checksums: {0}", errors[0])
+    return response.text
 
 
 def update_or_add_message(msg, name, sha):
@@ -42,6 +69,7 @@ def update_or_add_message(msg, name, sha):
         return '{0}{1}```'.format(msg[:back], new_text)
     return '{0} \n### SHA256 Checksums:\n```\n{1}```'.format(msg, new_text)
 
+
 def get_release(repo, version):
     """ Get a Github Release matching the version tag. """
     try:
@@ -55,11 +83,20 @@ def get_release(repo, version):
 def parse_args():
     """ Parse and validate args """
     parser = argparse.ArgumentParser(
-        description="Creates Github Releases and uploads assets."
+        description="Updates a Github Release with checksums from KV"
     )
     parser.add_argument(
         "--api-key", default=os.environ.get("API_KEY"), help="Github API key"
     )
+    parser.add_argument(
+        "--kv-namespace-id", default=os.environ.get("KV_NAMESPACE"), help="workers KV namespace id"
+    )
+    parser.add_argument(
+        "--kv-account-id", default=os.environ.get("KV_ACCOUNT"), help="workers KV account id"
+    )
+    parser.add_argument(
+        "--kv-api-token", default=os.environ.get("KV_API_TOKEN"), help="workers KV API Token"
+    )
     parser.add_argument(
         "--release-version",
         metavar="version",
@@ -80,6 +117,18 @@ def parse_args():
         logging.error("Missing API key")
         is_valid = False
 
+    if not args.kv_namespace_id:
+        logging.error("Missing KV namespace id")
+        is_valid = False
+
+    if not args.kv_account_id:
+        logging.error("Missing KV account id")
+        is_valid = False
+
+    if not args.kv_api_token:
+        logging.error("Missing KV API token")
+        is_valid = False
+
     if is_valid:
         return args
 
@@ -95,16 +144,20 @@ def main():
         repo = client.get_repo(CLOUDFLARED_REPO)
         release = get_release(repo, args.release_version)
 
-        msg = release.body
+        msg = ""
 
-        for filepath in glob.glob("artifacts/*"):
-            pkg_hash = get_sha256(filepath)
-            # add the sha256 of the new artifact to the release message body
-            name = os.path.basename(filepath)
-            msg = update_or_add_message(msg, name, pkg_hash)
+        prefix = f"update_{args.release_version}_"
+        keys = kv_get_keys(prefix, args.kv_account_id,
+                           args.kv_namespace_id, args.kv_api_token)
+        for key in [k["name"] for k in keys]:
+            checksum = kv_get_value(
+                key, args.kv_account_id, args.kv_namespace_id, args.kv_api_token)
+            binary_name = key[len(prefix):]
+            msg = update_or_add_message(msg, binary_name, checksum)
 
         if args.dry_run:
-            logging.info("Skipping asset upload because of dry-run")
+            logging.info("Skipping release message update because of dry-run")
+            logging.info(f"Github message:\n{msg}")
             return
 
         # update the release body text
@@ -115,4 +168,4 @@ def main():
         exit(1)
 
 
-main()
+main()

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/cloudflare/brotli-go v0.0.0-20191101163834-d34379f7ff93
 	github.com/cloudflare/golibs v0.0.0-20170913112048-333127dbecfc
 	github.com/coredns/coredns v1.8.7
+	github.com/coreos/go-oidc/v3 v3.4.0
 	github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
 	github.com/facebookgo/grace v0.0.0-20180706040059-75cf19382434
 	github.com/fsnotify/fsnotify v1.4.9
@@ -56,7 +57,6 @@ require (
 	github.com/cheekybits/genny v1.0.0 // indirect
 	github.com/cloudflare/circl v1.2.1-0.20220809205628-0a9554f37a47 // indirect
 	github.com/coredns/caddy v1.1.1 // indirect
-	github.com/coreos/go-oidc/v3 v3.4.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/facebookgo/ensure v0.0.0-20160127193407-b4ab57deab51 // indirect
@@ -113,8 +113,8 @@ replace gopkg.in/yaml.v3 => gopkg.in/yaml.v3 v3.0.1
 // Post-quantum tunnel RTG-1339
 replace (
 	// branch go1.18
-	github.com/marten-seemann/qtls-go1-18 => github.com/cloudflare/qtls-pq v0.0.0-20220824105406-fb955667e0af
+	github.com/marten-seemann/qtls-go1-18 => github.com/cloudflare/qtls-pq v0.0.0-20221010110824-0053225e48b2
 
 	// branch go1.19
-	github.com/marten-seemann/qtls-go1-19 => github.com/cloudflare/qtls-pq v0.0.0-20220824104809-96561a41e0af
+	github.com/marten-seemann/qtls-go1-19 => github.com/cloudflare/qtls-pq v0.0.0-20221010110800-4f3769902fe0
 )

go.sum

@@ -132,10 +132,10 @@ github.com/cloudflare/circl v1.2.1-0.20220809205628-0a9554f37a47 h1:YzpECHxZ9TzO
 github.com/cloudflare/circl v1.2.1-0.20220809205628-0a9554f37a47/go.mod h1:qhx8gBILsYlbam7h09SvHDSkjpe3TfLA7b/z4rxJvkE=
 github.com/cloudflare/golibs v0.0.0-20170913112048-333127dbecfc h1:Dvk3ySBsOm5EviLx6VCyILnafPcQinXGP5jbTdHUJgE=
 github.com/cloudflare/golibs v0.0.0-20170913112048-333127dbecfc/go.mod h1:HlgKKR8V5a1wroIDDIz3/A+T+9Janfq+7n1P5sEFdi0=
-github.com/cloudflare/qtls-pq v0.0.0-20220824104809-96561a41e0af h1:JMpOQAaXjRRBkUX73fTNe9mConJLFl6FsIp9fHdLm7Y=
-github.com/cloudflare/qtls-pq v0.0.0-20220824104809-96561a41e0af/go.mod h1:aIsWqC0WXyUiUxBl/RfxAjDyWE9CCLqvSMnCMTd/+bc=
-github.com/cloudflare/qtls-pq v0.0.0-20220824105406-fb955667e0af h1:bhCmedjwrOSyzLtHVeQ+KhimcNTSfs0P5T7kbRQS+gA=
-github.com/cloudflare/qtls-pq v0.0.0-20220824105406-fb955667e0af/go.mod h1:mW0BgKFFDAiSmOdUwoORtjo0V2vqw5QzVYRtKQqw/Jg=
+github.com/cloudflare/qtls-pq v0.0.0-20221010110800-4f3769902fe0 h1:LEsjEHfKnIJEJU9QEIPVRuslxpBu+2kG2DXhxpkGT+o=
+github.com/cloudflare/qtls-pq v0.0.0-20221010110800-4f3769902fe0/go.mod h1:aIsWqC0WXyUiUxBl/RfxAjDyWE9CCLqvSMnCMTd/+bc=
+github.com/cloudflare/qtls-pq v0.0.0-20221010110824-0053225e48b2 h1:ErNoeVNqFXV+emlf4gY7Ms7/0DbQ8PT2UFxNyWBc51Q=
+github.com/cloudflare/qtls-pq v0.0.0-20221010110824-0053225e48b2/go.mod h1:mW0BgKFFDAiSmOdUwoORtjo0V2vqw5QzVYRtKQqw/Jg=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
@@ -741,6 +741,7 @@ golang.org/x/net v0.0.0-20220412020605-290c469a71a5/go.mod h1:CfG3xpIq0wQ8r1q4Su
 golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
 golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
+golang.org/x/net v0.0.0-20220826154423-83b083e8dc8b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
 golang.org/x/net v0.0.0-20220909164309-bea034e7d591 h1:D0B/7al0LLrVC8aWF4+oxpv/m8bc7ViFfVS8/gXGdqI=
 golang.org/x/net v0.0.0-20220909164309-bea034e7d591/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=

ingress/config.go

@@ -427,6 +427,12 @@ func (defaults *OriginRequestConfig) setHttp2Origin(overrides config.OriginReque
 	}
 }
 
+func (defaults *OriginRequestConfig) setAccess(overrides config.OriginRequestConfig) {
+	if val := overrides.Access; val != nil {
+		defaults.Access = *val
+	}
+}
+
 // SetConfig gets config for the requests that cloudflared sends to origins.
 // Each field has a setter method which sets a value for the field by trying to find:
 //   1. The user config for this rule
@@ -453,6 +459,8 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi
 	cfg.setProxyType(overrides)
 	cfg.setIPRules(overrides)
 	cfg.setHttp2Origin(overrides)
+	cfg.setAccess(overrides)
+
 	return cfg
 }
 
@@ -463,6 +471,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig
 	var keepAliveConnections *int
 	var keepAliveTimeout *config.CustomDuration
 	var proxyAddress *string
+	var access *config.AccessConfig
 
 	if c.ConnectTimeout != defaultHTTPConnectTimeout {
 		connectTimeout = &c.ConnectTimeout
@@ -482,6 +491,9 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig
 	if c.ProxyAddress != defaultProxyAddress {
 		proxyAddress = &c.ProxyAddress
 	}
+	if c.Access.Required {
+		access = &c.Access
+	}
 
 	return config.OriginRequestConfig{
 		ConnectTimeout:         connectTimeout,
@@ -501,6 +513,7 @@ func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig
 		ProxyType:              emptyStringToNil(c.ProxyType),
 		IPRules:                convertToRawIPRules(c.IPRules),
 		Http2Origin:            defaultBoolToNil(c.Http2Origin),
+		Access:                 access,
 	}
 }
 

ingress/icmp_darwin.go

@@ -11,16 +11,17 @@ import (
 	"context"
 	"fmt"
 	"math"
-	"net"
 	"net/netip"
 	"strconv"
 	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/net/icmp"
 
 	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 type icmpProxy struct {
@@ -129,14 +130,19 @@ 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 {
+	ctx, span := responder.requestSpan(ctx, pk)
+	defer responder.exportSpan()
+
 	originalEcho, err := getICMPEcho(pk.Message)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
+	span.SetAttributes(
+		attribute.Int("originalEchoID", originalEcho.ID),
+		attribute.Int("seq", originalEcho.Seq),
+	)
 	echoIDTrackerKey := flow3Tuple{
 		srcIP:          pk.Src,
 		dstIP:          pk.Dst,
@@ -145,28 +151,32 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 	// TODO: TUN-6744 assign unique flow per (src, echo ID)
 	assignedEchoID, success := ip.echoIDTracker.getOrAssign(echoIDTrackerKey)
 	if !success {
-		return fmt.Errorf("failed to assign unique echo ID")
+		err := fmt.Errorf("failed to assign unique echo ID")
+		tracing.EndWithErrorStatus(span, err)
+		return err
 	}
+	span.SetAttributes(attribute.Int("assignedEchoID", int(assignedEchoID)))
+
 	newFunnelFunc := func() (packet.Funnel, error) {
 		originalEcho, err := getICMPEcho(pk.Message)
 		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)
 	funnel, isNew, err := ip.srcFunnelTracker.GetOrRegister(funnelID, newFunnelFunc)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
 	if isNew {
+		span.SetAttributes(attribute.Bool("newFlow", true))
 		ip.logger.Debug().
 			Str("src", pk.Src.String()).
 			Str("dst", pk.Dst.String()).
@@ -176,9 +186,16 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 	}
 	icmpFlow, err := toICMPEchoFlow(funnel)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
-	return icmpFlow.sendToDst(pk.Dst, pk.Message)
+	err = icmpFlow.sendToDst(pk.Dst, pk.Message)
+	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
+		return err
+	}
+	tracing.End(span)
+	return nil
 }
 
 // Serve listens for responses to the requests until context is done
@@ -202,7 +219,7 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 			ip.logger.Debug().Err(err).Str("dst", from.String()).Msg("Failed to parse ICMP reply, continue to parse as full packet")
 			// In unit test, we found out when the listener listens on 0.0.0.0, the socket reads the full packet after
 			// the second reply
-			if err := ip.handleFullPacket(icmpDecoder, buf[:n]); err != nil {
+			if err := ip.handleFullPacket(ctx, icmpDecoder, buf[:n]); err != nil {
 				ip.logger.Debug().Err(err).Str("dst", from.String()).Msg("Failed to parse ICMP reply as full packet")
 			}
 			continue
@@ -211,14 +228,14 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 			ip.logger.Debug().Str("dst", from.String()).Msgf("Drop ICMP %s from reply", reply.msg.Type)
 			continue
 		}
-		if err := ip.sendReply(reply); err != nil {
+		if err := ip.sendReply(ctx, reply); err != nil {
 			ip.logger.Error().Err(err).Str("dst", from.String()).Msg("Failed to send ICMP reply")
 			continue
 		}
 	}
 }
 
-func (ip *icmpProxy) handleFullPacket(decoder *packet.ICMPDecoder, rawPacket []byte) error {
+func (ip *icmpProxy) handleFullPacket(ctx context.Context, decoder *packet.ICMPDecoder, rawPacket []byte) error {
 	icmpPacket, err := decoder.Decode(packet.RawPacket{Data: rawPacket})
 	if err != nil {
 		return err
@@ -232,13 +249,13 @@ func (ip *icmpProxy) handleFullPacket(decoder *packet.ICMPDecoder, rawPacket []b
 		msg:  icmpPacket.Message,
 		echo: echo,
 	}
-	if ip.sendReply(&reply); err != nil {
+	if ip.sendReply(ctx, &reply); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (ip *icmpProxy) sendReply(reply *echoReply) error {
+func (ip *icmpProxy) sendReply(ctx context.Context, reply *echoReply) error {
 	funnelID := echoFunnelID(reply.echo.ID)
 	funnel, ok := ip.srcFunnelTracker.Get(funnelID)
 	if !ok {
@@ -248,25 +265,19 @@ func (ip *icmpProxy) sendReply(reply *echoReply) error {
 	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
-	echoIDTrackerKey flow3Tuple
-	assignedEchoID   uint16
-}
+	_, span := icmpFlow.responder.replySpan(ctx, ip.logger)
+	defer icmpFlow.responder.exportSpan()
 
-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)
+	span.SetAttributes(
+		attribute.String("dst", reply.from.String()),
+		attribute.Int("echoID", reply.echo.ID),
+		attribute.Int("seq", reply.echo.Seq),
+		attribute.Int("originalEchoID", icmpFlow.originalEchoID),
+	)
+	if err := icmpFlow.returnToSrc(reply); err != nil {
+		tracing.EndWithErrorStatus(span, err)
+	}
+	tracing.End(span)
 	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

@@ -19,9 +19,10 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
-	"golang.org/x/net/icmp"
+	"go.opentelemetry.io/otel/attribute"
 
 	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 const (
@@ -97,29 +98,38 @@ 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 {
+	ctx, span := responder.requestSpan(ctx, pk)
+	defer responder.exportSpan()
+
 	originalEcho, err := getICMPEcho(pk.Message)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
-	newConnChan := make(chan *icmp.PacketConn, 1)
+	span.SetAttributes(
+		attribute.Int("originalEchoID", originalEcho.ID),
+		attribute.Int("seq", originalEcho.Seq),
+	)
+
 	newFunnelFunc := func() (packet.Funnel, error) {
 		conn, err := newICMPConn(ip.listenIP, ip.ipv6Zone)
 		if err != nil {
+			tracing.EndWithErrorStatus(span, err)
 			return nil, errors.Wrap(err, "failed to open ICMP socket")
 		}
 		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}
+		span.SetAttributes(attribute.Int("port", localUDPAddr.Port))
+
 		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{
@@ -129,22 +139,24 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 	}
 	funnel, isNew, err := ip.srcFunnelTracker.GetOrRegister(funnelID, newFunnelFunc)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
 	icmpFlow, err := toICMPEchoFlow(funnel)
 	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return err
 	}
 	if isNew {
+		span.SetAttributes(attribute.Bool("newFlow", true))
 		ip.logger.Debug().
 			Str("src", pk.Src.String()).
 			Str("dst", pk.Dst.String()).
 			Int("originalEchoID", originalEcho.ID).
 			Msg("New flow")
-		conn := <-newConnChan
 		go func() {
 			defer ip.srcFunnelTracker.Unregister(funnelID, icmpFlow)
-			if err := ip.listenResponse(icmpFlow, conn); err != nil {
+			if err := ip.listenResponse(ctx, icmpFlow); err != nil {
 				ip.logger.Debug().Err(err).
 					Str("src", pk.Src.String()).
 					Str("dst", pk.Dst.String()).
@@ -154,8 +166,10 @@ func (ip *icmpProxy) Request(pk *packet.ICMP, responder packet.FunnelUniPipe) er
 		}()
 	}
 	if err := icmpFlow.sendToDst(pk.Dst, pk.Message); err != nil {
+		tracing.EndWithErrorStatus(span, err)
 		return errors.Wrap(err, "failed to send ICMP echo request")
 	}
+	tracing.End(span)
 	return nil
 }
 
@@ -164,43 +178,53 @@ func (ip *icmpProxy) Serve(ctx context.Context) error {
 	return ctx.Err()
 }
 
-func (ip *icmpProxy) listenResponse(flow *icmpEchoFlow, conn *icmp.PacketConn) error {
+func (ip *icmpProxy) listenResponse(ctx context.Context, flow *icmpEchoFlow) error {
 	buf := make([]byte, mtu)
 	for {
-		n, from, err := conn.ReadFrom(buf)
-		if err != nil {
+		retryable, err := ip.handleResponse(ctx, flow, buf)
+		if err != nil && !retryable {
 			return err
 		}
-		reply, err := parseReply(from, buf[:n])
-		if err != nil {
-			ip.logger.Error().Err(err).Str("dst", from.String()).Msg("Failed to parse ICMP reply")
-			continue
-		}
-		if !isEchoReply(reply.msg) {
-			ip.logger.Debug().Str("dst", from.String()).Msgf("Drop ICMP %s from reply", reply.msg.Type)
-			continue
-		}
-		if err := flow.returnToSrc(reply); err != nil {
-			ip.logger.Err(err).Str("dst", from.String()).Msg("Failed to send ICMP reply")
-			continue
-		}
 	}
 }
 
-// originSender wraps icmp.PacketConn to implement packet.FunnelUniPipe interface
-type originSender struct {
-	conn *icmp.PacketConn
-}
+func (ip *icmpProxy) handleResponse(ctx context.Context, flow *icmpEchoFlow, buf []byte) (retryableErr bool, err error) {
+	_, span := flow.responder.replySpan(ctx, ip.logger)
+	defer flow.responder.exportSpan()
 
-func (os *originSender) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	_, err := os.conn.WriteTo(pk.Data, &net.UDPAddr{
-		IP: dst.AsSlice(),
-	})
-	return err
-}
+	span.SetAttributes(
+		attribute.Int("originalEchoID", flow.originalEchoID),
+	)
 
-func (os *originSender) Close() error {
-	return os.conn.Close()
+	n, from, err := flow.originConn.ReadFrom(buf)
+	if err != nil {
+		tracing.EndWithErrorStatus(span, err)
+		return false, err
+	}
+	reply, err := parseReply(from, buf[:n])
+	if err != nil {
+		ip.logger.Error().Err(err).Str("dst", from.String()).Msg("Failed to parse ICMP reply")
+		tracing.EndWithErrorStatus(span, err)
+		return true, err
+	}
+	if !isEchoReply(reply.msg) {
+		err := fmt.Errorf("Expect ICMP echo reply, got %s", reply.msg.Type)
+		ip.logger.Debug().Str("dst", from.String()).Msgf("Drop ICMP %s from reply", reply.msg.Type)
+		tracing.EndWithErrorStatus(span, err)
+		return true, err
+	}
+	span.SetAttributes(
+		attribute.String("dst", reply.from.String()),
+		attribute.Int("echoID", reply.echo.ID),
+		attribute.Int("seq", reply.echo.Seq),
+	)
+	if err := flow.returnToSrc(reply); err != nil {
+		ip.logger.Err(err).Str("dst", from.String()).Msg("Failed to send ICMP reply")
+		tracing.EndWithErrorStatus(span, err)
+		return true, err
+	}
+	tracing.End(span)
+	return true, nil
 }
 
 // Only linux uses flow3Tuple as FunnelID

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

@@ -21,11 +21,13 @@ import (
 	"github.com/google/gopacket/layers"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/net/icmp"
 	"golang.org/x/net/ipv4"
 	"golang.org/x/net/ipv6"
 
 	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 const (
@@ -265,34 +267,51 @@ 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 {
-	if pk == nil {
-		return errPacketNil
-	}
+func (ip *icmpProxy) Request(ctx context.Context, pk *packet.ICMP, responder *packetResponder) error {
 	defer func() {
 		if r := recover(); r != nil {
 			ip.logger.Error().Interface("error", r).Msgf("Recover panic from sending icmp request/response, error %s", debug.Stack())
 		}
 	}()
 
+	_, requestSpan := responder.requestSpan(ctx, pk)
+	defer responder.exportSpan()
+
 	echo, err := getICMPEcho(pk.Message)
 	if err != nil {
 		return err
 	}
-	respData, err := ip.icmpEchoRoundtrip(pk.Dst, echo)
+	requestSpan.SetAttributes(
+		attribute.Int("originalEchoID", echo.ID),
+		attribute.Int("seq", echo.Seq),
+	)
+
+	resp, err := ip.icmpEchoRoundtrip(pk.Dst, echo)
 	if err != nil {
 		ip.logger.Err(err).Msg("ICMP echo roundtrip failed")
+		tracing.EndWithErrorStatus(requestSpan, err)
 		return err
 	}
+	tracing.End(requestSpan)
+	responder.exportSpan()
 
-	err = ip.handleEchoReply(pk, echo, respData, responder)
+	_, replySpan := responder.replySpan(ctx, ip.logger)
+	replySpan.SetAttributes(
+		attribute.Int("originalEchoID", echo.ID),
+		attribute.Int("seq", echo.Seq),
+		attribute.Int64("rtt", int64(resp.rtt())),
+		attribute.String("status", resp.status().String()),
+	)
+	err = ip.handleEchoReply(pk, echo, resp, responder)
 	if err != nil {
+		tracing.EndWithErrorStatus(replySpan, err)
 		return errors.Wrap(err, "failed to handle ICMP echo reply")
 	}
+	tracing.End(replySpan)
 	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, resp echoResp, responder *packetResponder) error {
 	var replyType icmp.Type
 	if request.Dst.Is4() {
 		replyType = ipv4.ICMPTypeEchoReply
@@ -313,7 +332,7 @@ func (ip *icmpProxy) handleEchoReply(request *packet.ICMP, echoReq *icmp.Echo, d
 			Body: &icmp.Echo{
 				ID:   echoReq.ID,
 				Seq:  echoReq.Seq,
-				Data: data,
+				Data: resp.payload(),
 			},
 		},
 	}
@@ -331,19 +350,20 @@ 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) {
+func (ip *icmpProxy) icmpEchoRoundtrip(dst netip.Addr, echo *icmp.Echo) (echoResp, error) {
 	if dst.Is6() {
 		if ip.srcSocketAddr == nil {
 			return nil, fmt.Errorf("cannot send ICMPv6 using ICMPv4 proxy")
 		}
 		resp, err := ip.icmp6SendEcho(dst, echo)
 		if err != nil {
+
 			return nil, errors.Wrap(err, "failed to send/receive ICMPv6 echo")
 		}
-		return resp.data, nil
+		return resp, nil
 	}
 	if ip.srcSocketAddr != nil {
 		return nil, fmt.Errorf("cannot send ICMPv4 using ICMPv6 proxy")
@@ -352,7 +372,7 @@ func (ip *icmpProxy) icmpEchoRoundtrip(dst netip.Addr, echo *icmp.Echo) ([]byte,
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to send/receive ICMPv4 echo")
 	}
-	return resp.data, nil
+	return resp, nil
 }
 
 /*
@@ -371,7 +391,7 @@ func (ip *icmpProxy) icmpEchoRoundtrip(dst netip.Addr, echo *icmp.Echo) ([]byte,
 	To retain the reference allocated objects, conversion from pointer to uintptr must happen as arguments to the
 	syscall function
 */
-func (ip *icmpProxy) icmpSendEcho(dst netip.Addr, echo *icmp.Echo) (*echoResp, error) {
+func (ip *icmpProxy) icmpSendEcho(dst netip.Addr, echo *icmp.Echo) (*echoV4Resp, error) {
 	dataSize := len(echo.Data)
 	replySize := echoReplySize + uintptr(dataSize)
 	replyBuf := make([]byte, replySize)
@@ -399,7 +419,7 @@ func (ip *icmpProxy) icmpSendEcho(dst netip.Addr, echo *icmp.Echo) (*echoResp, e
 	} else if replyCount > 1 {
 		ip.logger.Warn().Msgf("Received %d ICMP echo replies, only sending 1 back", replyCount)
 	}
-	return newEchoResp(replyBuf)
+	return newEchoV4Resp(replyBuf)
 }
 
 // Third definition of https://docs.microsoft.com/en-us/windows/win32/api/inaddr/ns-inaddr-in_addr#syntax is address in uint32
@@ -411,12 +431,30 @@ func inAddrV4(ip netip.Addr) (uint32, error) {
 	return endian.Uint32(v4[:]), nil
 }
 
-type echoResp struct {
+type echoResp interface {
+	status() ipStatus
+	rtt() uint32
+	payload() []byte
+}
+
+type echoV4Resp struct {
 	reply *echoReply
 	data  []byte
 }
 
-func newEchoResp(replyBuf []byte) (*echoResp, error) {
+func (r *echoV4Resp) status() ipStatus {
+	return r.reply.Status
+}
+
+func (r *echoV4Resp) rtt() uint32 {
+	return r.reply.RoundTripTime
+}
+
+func (r *echoV4Resp) payload() []byte {
+	return r.data
+}
+
+func newEchoV4Resp(replyBuf []byte) (*echoV4Resp, error) {
 	if len(replyBuf) == 0 {
 		return nil, fmt.Errorf("reply buffer is empty")
 	}
@@ -430,7 +468,7 @@ func newEchoResp(replyBuf []byte) (*echoResp, error) {
 	if dataBufStart < int(echoReplySize) {
 		return nil, fmt.Errorf("reply buffer size %d is too small to hold data of size %d", len(replyBuf), int(reply.DataSize))
 	}
-	return &echoResp{
+	return &echoV4Resp{
 		reply: &reply,
 		data:  replyBuf[dataBufStart:],
 	}, nil
@@ -502,6 +540,18 @@ type echoV6Resp struct {
 	data  []byte
 }
 
+func (r *echoV6Resp) status() ipStatus {
+	return r.reply.Status
+}
+
+func (r *echoV6Resp) rtt() uint32 {
+	return r.reply.RoundTripTime
+}
+
+func (r *echoV6Resp) payload() []byte {
+	return r.data
+}
+
 func newEchoV6Resp(replyBuf []byte, dataSize int) (*echoV6Resp, error) {
 	if len(replyBuf) == 0 {
 		return nil, fmt.Errorf("reply buffer is empty")

ingress/icmp_windows_test.go

@@ -59,7 +59,7 @@ func TestParseEchoReply(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		resp, err := newEchoResp(test.replyBuf)
+		resp, err := newEchoV4Resp(test.replyBuf)
 		if test.expectedReply == nil {
 			require.Error(t, err)
 			require.Nil(t, resp)

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")
 }

ingress/origin_icmp_proxy_test.go

@@ -1,10 +1,12 @@
 package ingress
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"net"
 	"net/netip"
+	"runtime"
 	"strings"
 	"sync"
 	"testing"
@@ -17,6 +19,8 @@ import (
 	"golang.org/x/net/ipv6"
 
 	"github.com/cloudflare/cloudflared/packet"
+	quicpogs "github.com/cloudflare/cloudflared/quic"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 var (
@@ -52,9 +56,9 @@ func testICMPRouterEcho(t *testing.T, sendIPv4 bool) {
 		close(proxyDone)
 	}()
 
-	responder := echoFlowResponder{
-		decoder:  packet.NewICMPDecoder(),
-		respChan: make(chan []byte, 1),
+	muxer := newMockMuxer(1)
+	responder := packetResponder{
+		datagramMuxer: muxer,
 	}
 
 	protocol := layers.IPProtocolICMPv6
@@ -90,14 +94,103 @@ func testICMPRouterEcho(t *testing.T, sendIPv4 bool) {
 					},
 				},
 			}
-			require.NoError(t, router.Request(&pk, &responder))
-			responder.validate(t, &pk)
+			require.NoError(t, router.Request(ctx, &pk, &responder))
+			validateEchoFlow(t, muxer, &pk)
 		}
 	}
 	cancel()
 	<-proxyDone
 }
 
+func TestTraceICMPRouterEcho(t *testing.T) {
+	tracingCtx := "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1"
+
+	router, err := NewICMPRouter(localhostIP, localhostIPv6, "", &noopLogger)
+	require.NoError(t, err)
+
+	proxyDone := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		router.Serve(ctx)
+		close(proxyDone)
+	}()
+
+	// Buffer 3 packets, request span, reply span and reply
+	muxer := newMockMuxer(3)
+	tracingIdentity, err := tracing.NewIdentity(tracingCtx)
+	require.NoError(t, err)
+	serializedIdentity, err := tracingIdentity.MarshalBinary()
+	require.NoError(t, err)
+
+	responder := packetResponder{
+		datagramMuxer:      muxer,
+		tracedCtx:          tracing.NewTracedContext(ctx, tracingIdentity.String(), &noopLogger),
+		serializedIdentity: serializedIdentity,
+	}
+
+	echo := &icmp.Echo{
+		ID:   12910,
+		Seq:  182,
+		Data: []byte(t.Name()),
+	}
+	pk := packet.ICMP{
+		IP: &packet.IP{
+			Src:      localhostIP,
+			Dst:      localhostIP,
+			Protocol: layers.IPProtocolICMPv4,
+			TTL:      packet.DefaultTTL,
+		},
+		Message: &icmp.Message{
+			Type: ipv4.ICMPTypeEcho,
+			Code: 0,
+			Body: echo,
+		},
+	}
+
+	require.NoError(t, router.Request(ctx, &pk, &responder))
+	// On Windows, request span is returned before reply
+	if runtime.GOOS != "windows" {
+		validateEchoFlow(t, muxer, &pk)
+	}
+
+	// Request span
+	receivedPacket := <-muxer.cfdToEdge
+	requestSpan, ok := receivedPacket.(*quicpogs.TracingSpanPacket)
+	require.True(t, ok)
+	require.NotEmpty(t, requestSpan.Spans)
+	require.True(t, bytes.Equal(serializedIdentity, requestSpan.TracingIdentity))
+
+	if runtime.GOOS == "windows" {
+		validateEchoFlow(t, muxer, &pk)
+	}
+
+	// Reply span
+	receivedPacket = <-muxer.cfdToEdge
+	replySpan, ok := receivedPacket.(*quicpogs.TracingSpanPacket)
+	require.True(t, ok)
+	require.NotEmpty(t, replySpan.Spans)
+	require.True(t, bytes.Equal(serializedIdentity, replySpan.TracingIdentity))
+	require.False(t, bytes.Equal(requestSpan.Spans, replySpan.Spans))
+
+	echo.Seq++
+	pk.Body = echo
+	// Only first request for a flow is traced. The edge will not send tracing context for the second request
+	newResponder := packetResponder{
+		datagramMuxer: muxer,
+	}
+	require.NoError(t, router.Request(ctx, &pk, &newResponder))
+	validateEchoFlow(t, muxer, &pk)
+
+	select {
+	case receivedPacket = <-muxer.cfdToEdge:
+		panic(fmt.Sprintf("Receive unexpected packet %+v", receivedPacket))
+	default:
+	}
+
+	cancel()
+	<-proxyDone
+}
+
 // TestConcurrentRequests makes sure icmpRouter can send concurrent requests to the same destination with different
 // echo ID. This simulates concurrent ping to the same destination.
 func TestConcurrentRequestsToSameDst(t *testing.T) {
@@ -123,9 +216,10 @@ func TestConcurrentRequestsToSameDst(t *testing.T) {
 		echoID := 38451 + i
 		go func() {
 			defer wg.Done()
-			responder := echoFlowResponder{
-				decoder:  packet.NewICMPDecoder(),
-				respChan: make(chan []byte, 1),
+
+			muxer := newMockMuxer(1)
+			responder := packetResponder{
+				datagramMuxer: muxer,
 			}
 			for seq := 0; seq < endSeq; seq++ {
 				pk := &packet.ICMP{
@@ -145,15 +239,15 @@ func TestConcurrentRequestsToSameDst(t *testing.T) {
 						},
 					},
 				}
-				require.NoError(t, router.Request(pk, &responder))
-				responder.validate(t, pk)
+				require.NoError(t, router.Request(ctx, pk, &responder))
+				validateEchoFlow(t, muxer, pk)
 			}
 		}()
 		go func() {
 			defer wg.Done()
-			responder := echoFlowResponder{
-				decoder:  packet.NewICMPDecoder(),
-				respChan: make(chan []byte, 1),
+			muxer := newMockMuxer(1)
+			responder := packetResponder{
+				datagramMuxer: muxer,
 			}
 			for seq := 0; seq < endSeq; seq++ {
 				pk := &packet.ICMP{
@@ -173,8 +267,8 @@ func TestConcurrentRequestsToSameDst(t *testing.T) {
 						},
 					},
 				}
-				require.NoError(t, router.Request(pk, &responder))
-				responder.validate(t, pk)
+				require.NoError(t, router.Request(ctx, pk, &responder))
+				validateEchoFlow(t, muxer, pk)
 			}
 		}()
 	}
@@ -241,9 +335,9 @@ func testICMPRouterRejectNotEcho(t *testing.T, srcDstIP netip.Addr, msgs []icmp.
 	router, err := NewICMPRouter(localhostIP, localhostIPv6, "", &noopLogger)
 	require.NoError(t, err)
 
-	responder := echoFlowResponder{
-		decoder:  packet.NewICMPDecoder(),
-		respChan: make(chan []byte),
+	muxer := newMockMuxer(1)
+	responder := packetResponder{
+		datagramMuxer: muxer,
 	}
 	protocol := layers.IPProtocolICMPv4
 	if srcDstIP.Is6() {
@@ -259,7 +353,7 @@ func testICMPRouterRejectNotEcho(t *testing.T, srcDstIP netip.Addr, msgs []icmp.
 			},
 			Message: &m,
 		}
-		require.Error(t, router.Request(&pk, &responder))
+		require.Error(t, router.Request(context.Background(), &pk, &responder))
 	}
 }
 
@@ -285,9 +379,10 @@ func (efr *echoFlowResponder) Close() error {
 	return nil
 }
 
-func (efr *echoFlowResponder) validate(t *testing.T, echoReq *packet.ICMP) {
-	pk := <-efr.respChan
-	decoded, err := efr.decoder.Decode(packet.RawPacket{Data: pk})
+func validateEchoFlow(t *testing.T, muxer *mockMuxer, echoReq *packet.ICMP) {
+	pk := <-muxer.cfdToEdge
+	decoder := packet.NewICMPDecoder()
+	decoded, err := decoder.Decode(packet.RawPacket{Data: pk.Payload()})
 	require.NoError(t, err)
 	require.Equal(t, decoded.Src, echoReq.Dst)
 	require.Equal(t, decoded.Dst, echoReq.Src)

ingress/packet_router.go

@@ -0,0 +1,186 @@
+package ingress
+
+import (
+	"context"
+	"fmt"
+	"net/netip"
+
+	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+
+	"github.com/cloudflare/cloudflared/packet"
+	quicpogs "github.com/cloudflare/cloudflared/quic"
+	"github.com/cloudflare/cloudflared/tracing"
+)
+
+// Upstream of raw packets
+type muxer interface {
+	SendPacket(pk quicpogs.Packet) error
+	// ReceivePacket waits for the next raw packet from upstream
+	ReceivePacket(ctx context.Context) (quicpogs.Packet, error)
+}
+
+// PacketRouter routes packets between Upstream and ICMPRouter. Currently it rejects all other type of ICMP packets
+type PacketRouter struct {
+	globalConfig           *GlobalRouterConfig
+	muxer                  muxer
+	logger                 *zerolog.Logger
+	checkRouterEnabledFunc func() bool
+	icmpDecoder            *packet.ICMPDecoder
+	encoder                *packet.Encoder
+}
+
+// GlobalRouterConfig is the configuration shared by all instance of Router.
+type GlobalRouterConfig struct {
+	ICMPRouter *icmpRouter
+	IPv4Src    netip.Addr
+	IPv6Src    netip.Addr
+	Zone       string
+}
+
+// NewPacketRouter creates a PacketRouter that handles ICMP packets. Packets are read from muxer but dropped if globalConfig is nil.
+func NewPacketRouter(globalConfig *GlobalRouterConfig, muxer muxer, logger *zerolog.Logger, checkRouterEnabledFunc func() bool) *PacketRouter {
+	return &PacketRouter{
+		globalConfig:           globalConfig,
+		muxer:                  muxer,
+		logger:                 logger,
+		checkRouterEnabledFunc: checkRouterEnabledFunc,
+		icmpDecoder:            packet.NewICMPDecoder(),
+		encoder:                packet.NewEncoder(),
+	}
+}
+
+func (r *PacketRouter) Serve(ctx context.Context) error {
+	for {
+		rawPacket, responder, err := r.nextPacket(ctx)
+		if err != nil {
+			return err
+		}
+		r.handlePacket(ctx, rawPacket, responder)
+	}
+}
+
+func (r *PacketRouter) nextPacket(ctx context.Context) (packet.RawPacket, *packetResponder, error) {
+	pk, err := r.muxer.ReceivePacket(ctx)
+	if err != nil {
+		return packet.RawPacket{}, nil, err
+	}
+	responder := &packetResponder{
+		datagramMuxer: r.muxer,
+	}
+	switch pk.Type() {
+	case quicpogs.DatagramTypeIP:
+		return packet.RawPacket{Data: pk.Payload()}, responder, nil
+	case quicpogs.DatagramTypeIPWithTrace:
+		var identity tracing.Identity
+		if err := identity.UnmarshalBinary(pk.Metadata()); err != nil {
+			r.logger.Err(err).Bytes("tracingIdentity", pk.Metadata()).Msg("Failed to unmarshal tracing identity")
+		} else {
+			responder.tracedCtx = tracing.NewTracedContext(ctx, identity.String(), r.logger)
+			responder.serializedIdentity = pk.Metadata()
+		}
+		return packet.RawPacket{Data: pk.Payload()}, responder, nil
+	default:
+		return packet.RawPacket{}, nil, fmt.Errorf("unexpected datagram type %d", pk.Type())
+	}
+}
+
+func (r *PacketRouter) handlePacket(ctx context.Context, rawPacket packet.RawPacket, responder *packetResponder) {
+	// ICMP Proxy feature is disabled, drop packets
+	if r.globalConfig == nil {
+		return
+	}
+
+	if enabled := r.checkRouterEnabledFunc(); !enabled {
+		return
+	}
+
+	icmpPacket, err := r.icmpDecoder.Decode(rawPacket)
+	if err != nil {
+		r.logger.Err(err).Msg("Failed to decode ICMP packet from quic datagram")
+		return
+	}
+
+	if icmpPacket.TTL <= 1 {
+		if err := r.sendTTLExceedMsg(ctx, icmpPacket, rawPacket, r.encoder); err != nil {
+			r.logger.Err(err).Msg("Failed to return ICMP TTL exceed error")
+		}
+		return
+	}
+	icmpPacket.TTL--
+
+	if err := r.globalConfig.ICMPRouter.Request(ctx, icmpPacket, responder); err != nil {
+		r.logger.Err(err).
+			Str("src", icmpPacket.Src.String()).
+			Str("dst", icmpPacket.Dst.String()).
+			Interface("type", icmpPacket.Type).
+			Msg("Failed to send ICMP packet")
+	}
+}
+
+func (r *PacketRouter) sendTTLExceedMsg(ctx context.Context, pk *packet.ICMP, rawPacket packet.RawPacket, encoder *packet.Encoder) error {
+	var srcIP netip.Addr
+	if pk.Dst.Is4() {
+		srcIP = r.globalConfig.IPv4Src
+	} else {
+		srcIP = r.globalConfig.IPv6Src
+	}
+	ttlExceedPacket := packet.NewICMPTTLExceedPacket(pk.IP, rawPacket, srcIP)
+
+	encodedTTLExceed, err := encoder.Encode(ttlExceedPacket)
+	if err != nil {
+		return err
+	}
+	return r.muxer.SendPacket(quicpogs.RawPacket(encodedTTLExceed))
+}
+
+// packetResponder should not be used concurrently. This assumption is upheld because reply packets are ready one-by-one
+type packetResponder struct {
+	datagramMuxer      muxer
+	tracedCtx          *tracing.TracedContext
+	serializedIdentity []byte
+	// hadReply tracks if there has been any reply for this flow
+	hadReply bool
+}
+
+func (pr *packetResponder) tracingEnabled() bool {
+	return pr.tracedCtx != nil
+}
+
+func (pr *packetResponder) returnPacket(rawPacket packet.RawPacket) error {
+	pr.hadReply = true
+	return pr.datagramMuxer.SendPacket(quicpogs.RawPacket(rawPacket))
+}
+
+func (pr *packetResponder) requestSpan(ctx context.Context, pk *packet.ICMP) (context.Context, trace.Span) {
+	if !pr.tracingEnabled() {
+		return ctx, tracing.NewNoopSpan()
+	}
+	return pr.tracedCtx.Tracer().Start(pr.tracedCtx, "icmp-echo-request", trace.WithAttributes(
+		attribute.String("src", pk.Src.String()),
+		attribute.String("dst", pk.Dst.String()),
+	))
+}
+
+func (pr *packetResponder) replySpan(ctx context.Context, logger *zerolog.Logger) (context.Context, trace.Span) {
+	if !pr.tracingEnabled() || pr.hadReply {
+		return ctx, tracing.NewNoopSpan()
+	}
+	return pr.tracedCtx.Tracer().Start(pr.tracedCtx, "icmp-echo-reply")
+}
+
+func (pr *packetResponder) exportSpan() {
+	if !pr.tracingEnabled() {
+		return
+	}
+	spans := pr.tracedCtx.GetProtoSpans()
+	if len(spans) > 0 {
+		// Make sure spans are cleared after they are sent
+		defer pr.tracedCtx.ClearSpans()
+		pr.datagramMuxer.SendPacket(&quicpogs.TracingSpanPacket{
+			Spans:           spans,
+			TracingIdentity: pr.serializedIdentity,
+		})
+	}
+}

ingress/packet_router_test.go

@@ -1,4 +1,4 @@
-package packet
+package ingress
 
 import (
 	"bytes"
@@ -10,32 +10,28 @@ import (
 	"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"
 	"golang.org/x/net/ipv6"
+
+	"github.com/cloudflare/cloudflared/packet"
+	quicpogs "github.com/cloudflare/cloudflared/quic"
 )
 
 var (
-	noopLogger   = zerolog.Nop()
 	packetConfig = &GlobalRouterConfig{
-		ICMPRouter: &mockICMPRouter{},
+		ICMPRouter: nil,
 		IPv4Src:    netip.MustParseAddr("172.16.0.1"),
 		IPv6Src:    netip.MustParseAddr("fd51:2391:523:f4ee::1"),
 	}
 )
 
 func TestRouterReturnTTLExceed(t *testing.T) {
-	upstream := &mockUpstream{
-		source: make(chan RawPacket),
-	}
-	returnPipe := &mockFunnelUniPipe{
-		uniPipe: make(chan RawPacket),
-	}
+	muxer := newMockMuxer(0)
 	routerEnabled := &routerEnabledChecker{}
 	routerEnabled.set(true)
-	router := NewRouter(packetConfig, upstream, returnPipe, &noopLogger, routerEnabled.isEnabled)
+	router := NewPacketRouter(packetConfig, muxer, &noopLogger, routerEnabled.isEnabled)
 	ctx, cancel := context.WithCancel(context.Background())
 	routerStopped := make(chan struct{})
 	go func() {
@@ -43,8 +39,8 @@ func TestRouterReturnTTLExceed(t *testing.T) {
 		close(routerStopped)
 	}()
 
-	pk := ICMP{
-		IP: &IP{
+	pk := packet.ICMP{
+		IP: &packet.IP{
 			Src:      netip.MustParseAddr("192.168.1.1"),
 			Dst:      netip.MustParseAddr("10.0.0.1"),
 			Protocol: layers.IPProtocolICMPv4,
@@ -60,9 +56,9 @@ func TestRouterReturnTTLExceed(t *testing.T) {
 			},
 		},
 	}
-	assertTTLExceed(t, &pk, router.globalConfig.IPv4Src, upstream, returnPipe)
-	pk = ICMP{
-		IP: &IP{
+	assertTTLExceed(t, &pk, router.globalConfig.IPv4Src, muxer)
+	pk = packet.ICMP{
+		IP: &packet.IP{
 			Src:      netip.MustParseAddr("fd51:2391:523:f4ee::1"),
 			Dst:      netip.MustParseAddr("fd51:2391:697:f4ee::2"),
 			Protocol: layers.IPProtocolICMPv6,
@@ -78,21 +74,16 @@ func TestRouterReturnTTLExceed(t *testing.T) {
 			},
 		},
 	}
-	assertTTLExceed(t, &pk, router.globalConfig.IPv6Src, upstream, returnPipe)
+	assertTTLExceed(t, &pk, router.globalConfig.IPv6Src, muxer)
 
 	cancel()
 	<-routerStopped
 }
 
 func TestRouterCheckEnabled(t *testing.T) {
-	upstream := &mockUpstream{
-		source: make(chan RawPacket),
-	}
-	returnPipe := &mockFunnelUniPipe{
-		uniPipe: make(chan RawPacket),
-	}
+	muxer := newMockMuxer(0)
 	routerEnabled := &routerEnabledChecker{}
-	router := NewRouter(packetConfig, upstream, returnPipe, &noopLogger, routerEnabled.isEnabled)
+	router := NewPacketRouter(packetConfig, muxer, &noopLogger, routerEnabled.isEnabled)
 	ctx, cancel := context.WithCancel(context.Background())
 	routerStopped := make(chan struct{})
 	go func() {
@@ -100,8 +91,8 @@ func TestRouterCheckEnabled(t *testing.T) {
 		close(routerStopped)
 	}()
 
-	pk := ICMP{
-		IP: &IP{
+	pk := packet.ICMP{
+		IP: &packet.IP{
 			Src:      netip.MustParseAddr("192.168.1.1"),
 			Dst:      netip.MustParseAddr("10.0.0.1"),
 			Protocol: layers.IPProtocolICMPv4,
@@ -119,23 +110,28 @@ func TestRouterCheckEnabled(t *testing.T) {
 	}
 
 	// router is disabled
-	require.NoError(t, upstream.send(&pk))
+	encoder := packet.NewEncoder()
+	encodedPacket, err := encoder.Encode(&pk)
+	require.NoError(t, err)
+	sendPacket := quicpogs.RawPacket(encodedPacket)
+
+	muxer.edgeToCfd <- sendPacket
 	select {
 	case <-time.After(time.Millisecond * 10):
-	case <-returnPipe.uniPipe:
+	case <-muxer.cfdToEdge:
 		t.Error("Unexpected reply when router is disabled")
 	}
 	routerEnabled.set(true)
 	// router is enabled, expects reply
-	require.NoError(t, upstream.send(&pk))
-	<-returnPipe.uniPipe
+	muxer.edgeToCfd <- sendPacket
+	<-muxer.cfdToEdge
 
 	routerEnabled.set(false)
 	// router is disabled
-	require.NoError(t, upstream.send(&pk))
+	muxer.edgeToCfd <- sendPacket
 	select {
 	case <-time.After(time.Millisecond * 10):
-	case <-returnPipe.uniPipe:
+	case <-muxer.cfdToEdge:
 		t.Error("Unexpected reply when router is disabled")
 	}
 
@@ -143,66 +139,88 @@ func TestRouterCheckEnabled(t *testing.T) {
 	<-routerStopped
 }
 
-func assertTTLExceed(t *testing.T, originalPacket *ICMP, expectedSrc netip.Addr, upstream *mockUpstream, returnPipe *mockFunnelUniPipe) {
-	encoder := NewEncoder()
+func assertTTLExceed(t *testing.T, originalPacket *packet.ICMP, expectedSrc netip.Addr, muxer *mockMuxer) {
+	encoder := packet.NewEncoder()
 	rawPacket, err := encoder.Encode(originalPacket)
 	require.NoError(t, err)
-	upstream.source <- rawPacket
+	muxer.edgeToCfd <- quicpogs.RawPacket(rawPacket)
 
-	resp := <-returnPipe.uniPipe
-	decoder := NewICMPDecoder()
-	decoded, err := decoder.Decode(resp)
+	resp := <-muxer.cfdToEdge
+	decoder := packet.NewICMPDecoder()
+	decoded, err := decoder.Decode(packet.RawPacket(resp.(quicpogs.RawPacket)))
 	require.NoError(t, err)
 
 	require.Equal(t, expectedSrc, decoded.Src)
 	require.Equal(t, originalPacket.Src, decoded.Dst)
 	require.Equal(t, originalPacket.Protocol, decoded.Protocol)
-	require.Equal(t, DefaultTTL, decoded.TTL)
+	require.Equal(t, packet.DefaultTTL, decoded.TTL)
 	if originalPacket.Dst.Is4() {
 		require.Equal(t, ipv4.ICMPTypeTimeExceeded, decoded.Type)
 	} else {
 		require.Equal(t, ipv6.ICMPTypeTimeExceeded, decoded.Type)
 	}
 	require.Equal(t, 0, decoded.Code)
-	assertICMPChecksum(t, decoded)
 	timeExceed, ok := decoded.Body.(*icmp.TimeExceeded)
 	require.True(t, ok)
 	require.True(t, bytes.Equal(rawPacket.Data, timeExceed.Data))
 }
 
-type mockUpstream struct {
-	source chan RawPacket
+type mockMuxer struct {
+	cfdToEdge chan quicpogs.Packet
+	edgeToCfd chan quicpogs.Packet
 }
 
-func (ms *mockUpstream) send(pk Packet) error {
-	encoder := NewEncoder()
-	rawPacket, err := encoder.Encode(pk)
-	if err != nil {
-		return err
+func newMockMuxer(capacity int) *mockMuxer {
+	return &mockMuxer{
+		cfdToEdge: make(chan quicpogs.Packet, capacity),
+		edgeToCfd: make(chan quicpogs.Packet, capacity),
 	}
-	ms.source <- rawPacket
+}
+
+// Copy packet, because icmpProxy expects the encoder buffer to be reusable after the packet is sent
+func (mm *mockMuxer) SendPacket(pk quicpogs.Packet) error {
+	payload := pk.Payload()
+	copiedPayload := make([]byte, len(payload))
+	copy(copiedPayload, payload)
+
+	metadata := pk.Metadata()
+	copiedMetadata := make([]byte, len(metadata))
+	copy(copiedMetadata, metadata)
+
+	var copiedPacket quicpogs.Packet
+	switch pk.Type() {
+	case quicpogs.DatagramTypeIP:
+		copiedPacket = quicpogs.RawPacket(packet.RawPacket{
+			Data: copiedPayload,
+		})
+	case quicpogs.DatagramTypeIPWithTrace:
+		copiedPacket = &quicpogs.TracedPacket{
+			Packet: packet.RawPacket{
+				Data: copiedPayload,
+			},
+			TracingIdentity: copiedMetadata,
+		}
+	case quicpogs.DatagramTypeTracingSpan:
+		copiedPacket = &quicpogs.TracingSpanPacket{
+			Spans:           copiedPayload,
+			TracingIdentity: copiedMetadata,
+		}
+	default:
+		return fmt.Errorf("unexpected metadata type %d", pk.Type())
+	}
+	mm.cfdToEdge <- copiedPacket
 	return nil
 }
 
-func (ms *mockUpstream) ReceivePacket(ctx context.Context) (RawPacket, error) {
+func (mm *mockMuxer) ReceivePacket(ctx context.Context) (quicpogs.Packet, error) {
 	select {
 	case <-ctx.Done():
-		return RawPacket{}, ctx.Err()
-	case pk := <-ms.source:
+		return nil, ctx.Err()
+	case pk := <-mm.edgeToCfd:
 		return pk, nil
 	}
 }
 
-type mockICMPRouter struct{}
-
-func (mir mockICMPRouter) Serve(ctx context.Context) error {
-	return fmt.Errorf("Serve not implemented by mockICMPRouter")
-}
-
-func (mir mockICMPRouter) Request(pk *ICMP, responder FunnelUniPipe) error {
-	return fmt.Errorf("Request not implemented by mockICMPRouter")
-}
-
 type routerEnabledChecker struct {
 	enabled uint32
 }

packet/funnel.go

@@ -16,11 +16,9 @@ var (
 
 // 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
+	// Updates the last time traffic went through this funnel
+	UpdateLastActive()
+	// LastActive returns the last time there is traffic through this funnel
 	LastActive() time.Time
 	// Close closes the funnel. Further call to SendToDst or ReturnToSrc should return an error
 	Close() error
@@ -36,73 +34,26 @@ type FunnelUniPipe interface {
 	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
+type ActivityTracker struct {
 	// 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,
+func NewActivityTracker() *ActivityTracker {
+	return &ActivityTracker{
 		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 (at *ActivityTracker) UpdateLastActive() {
+	atomic.StoreInt64(&at.lastActive, time.Now().Unix())
 }
 
-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)
+func (at *ActivityTracker) LastActive() time.Time {
+	lastActive := atomic.LoadInt64(&at.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

packet/router.go

@@ -1,108 +0,0 @@
-package packet
-
-import (
-	"context"
-	"net/netip"
-
-	"github.com/rs/zerolog"
-)
-
-// ICMPRouter sends ICMP messages and listens for their responses
-type ICMPRouter interface {
-	// Serve starts listening for responses to the requests until context is done
-	Serve(ctx context.Context) error
-	// Request sends an ICMP message. Implementations should not modify pk after the function returns.
-	Request(pk *ICMP, responder FunnelUniPipe) error
-}
-
-// Upstream of raw packets
-type Upstream interface {
-	// ReceivePacket waits for the next raw packet from upstream
-	ReceivePacket(ctx context.Context) (RawPacket, error)
-}
-
-// Router routes packets between Upstream and ICMPRouter. Currently it rejects all other type of ICMP packets
-type Router struct {
-	upstream               Upstream
-	returnPipe             FunnelUniPipe
-	globalConfig           *GlobalRouterConfig
-	logger                 *zerolog.Logger
-	checkRouterEnabledFunc func() bool
-}
-
-// GlobalRouterConfig is the configuration shared by all instance of Router.
-type GlobalRouterConfig struct {
-	ICMPRouter ICMPRouter
-	IPv4Src    netip.Addr
-	IPv6Src    netip.Addr
-	Zone       string
-}
-
-func NewRouter(globalConfig *GlobalRouterConfig, upstream Upstream, returnPipe FunnelUniPipe, logger *zerolog.Logger, checkRouterEnabledFunc func() bool) *Router {
-	return &Router{
-		upstream:               upstream,
-		returnPipe:             returnPipe,
-		globalConfig:           globalConfig,
-		logger:                 logger,
-		checkRouterEnabledFunc: checkRouterEnabledFunc,
-	}
-}
-
-func (r *Router) Serve(ctx context.Context) error {
-	icmpDecoder := NewICMPDecoder()
-	encoder := NewEncoder()
-	for {
-		rawPacket, err := r.upstream.ReceivePacket(ctx)
-		if err != nil {
-			return err
-		}
-
-		// Drop packets if ICMPRouter wasn't created
-		if r.globalConfig == nil {
-			continue
-		}
-
-		if enabled := r.checkRouterEnabledFunc(); !enabled {
-			continue
-		}
-
-		icmpPacket, err := icmpDecoder.Decode(rawPacket)
-		if err != nil {
-			r.logger.Err(err).Msg("Failed to decode ICMP packet from quic datagram")
-			continue
-		}
-
-		if icmpPacket.TTL <= 1 {
-			if err := r.sendTTLExceedMsg(icmpPacket, rawPacket, encoder); err != nil {
-				r.logger.Err(err).Msg("Failed to return ICMP TTL exceed error")
-			}
-			continue
-		}
-		icmpPacket.TTL--
-
-		if err := r.globalConfig.ICMPRouter.Request(icmpPacket, r.returnPipe); err != nil {
-			r.logger.Err(err).
-				Str("src", icmpPacket.Src.String()).
-				Str("dst", icmpPacket.Dst.String()).
-				Interface("type", icmpPacket.Type).
-				Msg("Failed to send ICMP packet")
-			continue
-		}
-	}
-}
-
-func (r *Router) sendTTLExceedMsg(pk *ICMP, rawPacket RawPacket, encoder *Encoder) error {
-	var srcIP netip.Addr
-	if pk.Dst.Is4() {
-		srcIP = r.globalConfig.IPv4Src
-	} else {
-		srcIP = r.globalConfig.IPv6Src
-	}
-	ttlExceedPacket := NewICMPTTLExceedPacket(pk.IP, rawPacket, srcIP)
-
-	encodedTTLExceed, err := encoder.Encode(ttlExceedPacket)
-	if err != nil {
-		return err
-	}
-	return r.returnPipe.SendPacket(pk.Src, encodedTTLExceed)
-}

proxy/proxy.go

@@ -103,7 +103,8 @@ func (p *Proxy) ProxyHTTP(
 	ruleSpan.End()
 	if err, applied := p.applyIngressMiddleware(rule, req, w); err != nil {
 		if applied {
-			p.log.Error().Msg(err.Error())
+			rule, srv := ruleField(p.ingressRules, ruleNum)
+			p.logRequestError(err, cfRay, "", rule, srv)
 			return nil
 		}
 		return err

quic/datagram.go

@@ -113,9 +113,12 @@ func extractSessionID(b []byte) (uuid.UUID, []byte, error) {
 // SuffixSessionID appends the session ID at the end of the payload. Suffix is more performant than prefix because
 // the payload slice might already have enough capacity to append the session ID at the end
 func SuffixSessionID(sessionID uuid.UUID, b []byte) ([]byte, error) {
-	if len(b)+len(sessionID) > MaxDatagramFrameSize {
+	return suffixMetadata(b, sessionID[:])
+}
+
+func suffixMetadata(payload, metadata []byte) ([]byte, error) {
+	if len(payload)+len(metadata) > MaxDatagramFrameSize {
 		return nil, fmt.Errorf("datagram size exceed %d", MaxDatagramFrameSize)
 	}
-	b = append(b, sessionID[:]...)
-	return b, nil
+	return append(payload, metadata...), nil
 }

quic/datagram_test.go

@@ -1,6 +1,7 @@
 package quic
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"crypto/rsa"
@@ -23,6 +24,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 var (
@@ -121,6 +123,15 @@ func testDatagram(t *testing.T, version uint8, sessionToPayloads []*packet.Sessi
 
 	logger := zerolog.Nop()
 
+	tracingIdentity, err := tracing.NewIdentity("ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1")
+	require.NoError(t, err)
+	serializedTracingID, err := tracingIdentity.MarshalBinary()
+	require.NoError(t, err)
+	tracingSpan := &TracingSpanPacket{
+		Spans:           []byte("tracing"),
+		TracingIdentity: serializedTracingID,
+	}
+
 	errGroup, ctx := errgroup.WithContext(context.Background())
 	// Run edge side of datagram muxer
 	errGroup.Go(func() error {
@@ -140,18 +151,17 @@ func testDatagram(t *testing.T, version uint8, sessionToPayloads []*packet.Sessi
 			muxer := NewDatagramMuxerV2(quicSession, &logger, sessionDemuxChan)
 			muxer.ServeReceive(ctx)
 
-			icmpDecoder := packet.NewICMPDecoder()
 			for _, pk := range packets {
 				received, err := muxer.ReceivePacket(ctx)
 				require.NoError(t, err)
-
-				receivedICMP, err := icmpDecoder.Decode(received)
+				validateIPPacket(t, received, &pk)
+				received, err = muxer.ReceivePacket(ctx)
 				require.NoError(t, err)
-				require.Equal(t, pk.IP, receivedICMP.IP)
-				require.Equal(t, pk.Type, receivedICMP.Type)
-				require.Equal(t, pk.Code, receivedICMP.Code)
-				require.Equal(t, pk.Body, receivedICMP.Body)
+				validateIPPacketWithTracing(t, received, &pk, serializedTracingID)
 			}
+			received, err := muxer.ReceivePacket(ctx)
+			require.NoError(t, err)
+			validateTracingSpans(t, received, tracingSpan)
 		default:
 			return fmt.Errorf("unknown datagram version %d", version)
 		}
@@ -188,10 +198,15 @@ func testDatagram(t *testing.T, version uint8, sessionToPayloads []*packet.Sessi
 			for _, pk := range packets {
 				encodedPacket, err := encoder.Encode(&pk)
 				require.NoError(t, err)
-				require.NoError(t, muxerV2.SendPacket(encodedPacket))
+				require.NoError(t, muxerV2.SendPacket(RawPacket(encodedPacket)))
+				require.NoError(t, muxerV2.SendPacket(&TracedPacket{
+					Packet:          encodedPacket,
+					TracingIdentity: serializedTracingID,
+				}))
 			}
+			require.NoError(t, muxerV2.SendPacket(tracingSpan))
 			// Payload larger than transport MTU, should not be sent
-			require.Error(t, muxerV2.SendPacket(packet.RawPacket{
+			require.Error(t, muxerV2.SendPacket(RawPacket{
 				Data: largePayload,
 			}))
 			muxer = muxerV2
@@ -217,6 +232,38 @@ func testDatagram(t *testing.T, version uint8, sessionToPayloads []*packet.Sessi
 	require.NoError(t, errGroup.Wait())
 }
 
+func validateIPPacket(t *testing.T, receivedPacket Packet, expectedICMP *packet.ICMP) {
+	require.Equal(t, DatagramTypeIP, receivedPacket.Type())
+	rawPacket := receivedPacket.(RawPacket)
+	decoder := packet.NewICMPDecoder()
+	receivedICMP, err := decoder.Decode(packet.RawPacket(rawPacket))
+	require.NoError(t, err)
+	validateICMP(t, expectedICMP, receivedICMP)
+}
+
+func validateIPPacketWithTracing(t *testing.T, receivedPacket Packet, expectedICMP *packet.ICMP, serializedTracingID []byte) {
+	require.Equal(t, DatagramTypeIPWithTrace, receivedPacket.Type())
+	tracedPacket := receivedPacket.(*TracedPacket)
+	decoder := packet.NewICMPDecoder()
+	receivedICMP, err := decoder.Decode(tracedPacket.Packet)
+	require.NoError(t, err)
+	validateICMP(t, expectedICMP, receivedICMP)
+	require.True(t, bytes.Equal(tracedPacket.TracingIdentity, serializedTracingID))
+}
+
+func validateICMP(t *testing.T, expected, actual *packet.ICMP) {
+	require.Equal(t, expected.IP, actual.IP)
+	require.Equal(t, expected.Type, actual.Type)
+	require.Equal(t, expected.Code, actual.Code)
+	require.Equal(t, expected.Body, actual.Body)
+}
+
+func validateTracingSpans(t *testing.T, receivedPacket Packet, expectedSpan *TracingSpanPacket) {
+	require.Equal(t, DatagramTypeTracingSpan, receivedPacket.Type())
+	tracingSpans := receivedPacket.(*TracingSpanPacket)
+	require.Equal(t, tracingSpans, expectedSpan)
+}
+
 func newQUICListener(t *testing.T, config *quic.Config) quic.Listener {
 	// Create a simple tls config.
 	tlsConfig := generateTLSConfig()

quic/datagramv2.go

@@ -9,15 +9,28 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/cloudflare/cloudflared/packet"
+	"github.com/cloudflare/cloudflared/tracing"
 )
 
 type DatagramV2Type byte
 
 const (
+	// UDP payload
 	DatagramTypeUDP DatagramV2Type = iota
+	// Full IP packet
 	DatagramTypeIP
+	// DatagramTypeIP + tracing ID
+	DatagramTypeIPWithTrace
+	// Tracing spans in protobuf format
+	DatagramTypeTracingSpan
 )
 
+type Packet interface {
+	Type() DatagramV2Type
+	Payload() []byte
+	Metadata() []byte
+}
+
 const (
 	typeIDLen = 1
 	// Same as sessionDemuxChan capacity
@@ -41,7 +54,7 @@ type DatagramMuxerV2 struct {
 	session          quic.Connection
 	logger           *zerolog.Logger
 	sessionDemuxChan chan<- *packet.Session
-	packetDemuxChan  chan packet.RawPacket
+	packetDemuxChan  chan Packet
 }
 
 func NewDatagramMuxerV2(
@@ -54,7 +67,7 @@ func NewDatagramMuxerV2(
 		session:          quicSession,
 		logger:           &logger,
 		sessionDemuxChan: sessionDemuxChan,
-		packetDemuxChan:  make(chan packet.RawPacket, packetChanCapacity),
+		packetDemuxChan:  make(chan Packet, packetChanCapacity),
 	}
 }
 
@@ -79,14 +92,19 @@ func (dm *DatagramMuxerV2) SendToSession(session *packet.Session) error {
 	return nil
 }
 
-// SendPacket suffix the datagram type to the packet. The other end of the QUIC connection can demultiplex by parsing
-// the payload as IP and look at the source and destination.
-func (dm *DatagramMuxerV2) SendPacket(pk packet.RawPacket) error {
-	payloadWithVersion, err := SuffixType(pk.Data, DatagramTypeIP)
+// SendPacket sends a packet with datagram version in the suffix. If ctx is a TracedContext, it adds the tracing
+// context between payload and datagram version.
+// The other end of the QUIC connection can demultiplex by parsing the payload as IP and look at the source and destination.
+func (dm *DatagramMuxerV2) SendPacket(pk Packet) error {
+	payloadWithMetadata, err := suffixMetadata(pk.Payload(), pk.Metadata())
+	if err != nil {
+		return err
+	}
+	payloadWithMetadataAndType, err := SuffixType(payloadWithMetadata, pk.Type())
 	if err != nil {
 		return errors.Wrap(err, "Failed to suffix datagram type, it will be dropped")
 	}
-	if err := dm.session.SendMessage(payloadWithVersion); err != nil {
+	if err := dm.session.SendMessage(payloadWithMetadataAndType); err != nil {
 		return errors.Wrap(err, "Failed to send datagram back to edge")
 	}
 	return nil
@@ -108,10 +126,10 @@ func (dm *DatagramMuxerV2) ServeReceive(ctx context.Context) error {
 	}
 }
 
-func (dm *DatagramMuxerV2) ReceivePacket(ctx context.Context) (packet.RawPacket, error) {
+func (dm *DatagramMuxerV2) ReceivePacket(ctx context.Context) (pk Packet, err error) {
 	select {
 	case <-ctx.Done():
-		return packet.RawPacket{}, ctx.Err()
+		return nil, ctx.Err()
 	case pk := <-dm.packetDemuxChan:
 		return pk, nil
 	}
@@ -126,10 +144,8 @@ func (dm *DatagramMuxerV2) demux(ctx context.Context, msgWithType []byte) error
 	switch msgType {
 	case DatagramTypeUDP:
 		return dm.handleSession(ctx, msg)
-	case DatagramTypeIP:
-		return dm.handlePacket(ctx, msg)
 	default:
-		return fmt.Errorf("Unexpected datagram type %d", msgType)
+		return dm.handlePacket(ctx, msg, msgType)
 	}
 }
 
@@ -150,13 +166,93 @@ func (dm *DatagramMuxerV2) handleSession(ctx context.Context, session []byte) er
 	}
 }
 
-func (dm *DatagramMuxerV2) handlePacket(ctx context.Context, pk []byte) error {
+func (dm *DatagramMuxerV2) handlePacket(ctx context.Context, pk []byte, msgType DatagramV2Type) error {
+	var demuxedPacket Packet
+	switch msgType {
+	case DatagramTypeIP:
+		demuxedPacket = RawPacket(packet.RawPacket{Data: pk})
+	case DatagramTypeIPWithTrace:
+		tracingIdentity, payload, err := extractTracingIdentity(pk)
+		if err != nil {
+			return err
+		}
+		demuxedPacket = &TracedPacket{
+			Packet:          packet.RawPacket{Data: payload},
+			TracingIdentity: tracingIdentity,
+		}
+	case DatagramTypeTracingSpan:
+		tracingIdentity, spans, err := extractTracingIdentity(pk)
+		if err != nil {
+			return err
+		}
+		demuxedPacket = &TracingSpanPacket{
+			Spans:           spans,
+			TracingIdentity: tracingIdentity,
+		}
+	default:
+		return fmt.Errorf("Unexpected datagram type %d", msgType)
+	}
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case dm.packetDemuxChan <- packet.RawPacket{
-		Data: pk,
-	}:
+	case dm.packetDemuxChan <- demuxedPacket:
 		return nil
 	}
 }
+
+func extractTracingIdentity(pk []byte) (tracingIdentity []byte, payload []byte, err error) {
+	if len(pk) < tracing.IdentityLength {
+		return nil, nil, fmt.Errorf("packet with tracing context should have at least %d bytes, got %v", tracing.IdentityLength, pk)
+	}
+	tracingIdentity = pk[len(pk)-tracing.IdentityLength:]
+	payload = pk[:len(pk)-tracing.IdentityLength]
+	return tracingIdentity, payload, nil
+}
+
+type RawPacket packet.RawPacket
+
+func (rw RawPacket) Type() DatagramV2Type {
+	return DatagramTypeIP
+}
+
+func (rw RawPacket) Payload() []byte {
+	return rw.Data
+}
+
+func (rw RawPacket) Metadata() []byte {
+	return []byte{}
+}
+
+type TracedPacket struct {
+	Packet          packet.RawPacket
+	TracingIdentity []byte
+}
+
+func (tp *TracedPacket) Type() DatagramV2Type {
+	return DatagramTypeIPWithTrace
+}
+
+func (tp *TracedPacket) Payload() []byte {
+	return tp.Packet.Data
+}
+
+func (tp *TracedPacket) Metadata() []byte {
+	return tp.TracingIdentity
+}
+
+type TracingSpanPacket struct {
+	Spans           []byte
+	TracingIdentity []byte
+}
+
+func (tsp *TracingSpanPacket) Type() DatagramV2Type {
+	return DatagramTypeTracingSpan
+}
+
+func (tsp *TracingSpanPacket) Payload() []byte {
+	return tsp.Spans
+}
+
+func (tsp *TracingSpanPacket) Metadata() []byte {
+	return tsp.TracingIdentity
+}

supervisor/tunnel.go

@@ -20,8 +20,8 @@ import (
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/edgediscovery/allregions"
 	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/orchestration"
-	"github.com/cloudflare/cloudflared/packet"
 	quicpogs "github.com/cloudflare/cloudflared/quic"
 	"github.com/cloudflare/cloudflared/retry"
 	"github.com/cloudflare/cloudflared/signal"
@@ -70,7 +70,7 @@ type TunnelConfig struct {
 	MuxerConfig      *connection.MuxerConfig
 	ProtocolSelector connection.ProtocolSelector
 	EdgeTLSConfigs   map[connection.Protocol]*tls.Config
-	PacketConfig     *packet.GlobalRouterConfig
+	PacketConfig     *ingress.GlobalRouterConfig
 }
 
 func (c *TunnelConfig) registrationOptions(connectionID uint8, OriginLocalIP string, uuid uuid.UUID) *tunnelpogs.RegistrationOptions {
@@ -254,8 +254,9 @@ func (e *EdgeTunnelServer) Serve(ctx context.Context, connIndex uint8, protocolF
 		if !ok {
 			return err
 		}
+
 		e.config.Observer.SendReconnect(connIndex)
-		connLog.Logger().Info().Msgf("Retrying connection in up to %s seconds", duration)
+		connLog.Logger().Info().Msgf("Retrying connection in up to %s", duration)
 	}
 
 	// Check if the connection error was from an IP issue with the host or
@@ -656,6 +657,7 @@ func (e *EdgeTunnelServer) serveQUIC(
 	quicConn, err := connection.NewQUICConnection(
 		quicConfig,
 		edgeAddr,
+		connIndex,
 		tlsConfig,
 		e.orchestrator,
 		connOptions,

tracing/client.go

@@ -27,6 +27,8 @@ type InMemoryClient interface {
 	// ProtoSpans returns a copy of the list of in-memory stored spans as otlp
 	// protobuf byte array.
 	ProtoSpans() ([]byte, error)
+	// Clear spans removes all in-memory spans
+	ClearSpans()
 }
 
 // InMemoryOtlpClient is a client implementation for otlptrace.Client
@@ -78,6 +80,12 @@ func (mc *InMemoryOtlpClient) ProtoSpans() ([]byte, error) {
 	return proto.Marshal(pbRequest)
 }
 
+func (mc *InMemoryOtlpClient) ClearSpans() {
+	mc.mu.Lock()
+	defer mc.mu.Unlock()
+	mc.spans = make([]*tracepb.ResourceSpans, 0)
+}
+
 // NoopOtlpClient is a client implementation for otlptrace.Client that does nothing
 type NoopOtlpClient struct{}
 
@@ -102,3 +110,5 @@ func (mc *NoopOtlpClient) Spans() (string, error) {
 func (mc *NoopOtlpClient) ProtoSpans() ([]byte, error) {
 	return nil, errNoopTracer
 }
+
+func (mc *NoopOtlpClient) ClearSpans() {}

tracing/identity.go

@@ -0,0 +1,110 @@
+package tracing
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+const (
+	// 16 bytes for tracing ID, 8 bytes for span ID and 1 byte for flags
+	IdentityLength = 16 + 8 + 1
+)
+
+type Identity struct {
+	// Based on https://www.jaegertracing.io/docs/1.36/client-libraries/#value
+	// parent span ID is always 0 for our case
+	traceIDUpper uint64
+	traceIDLower uint64
+	spanID       uint64
+	flags        uint8
+}
+
+// TODO: TUN-6604 Remove this. To reconstruct into Jaeger propagation format, convert tracingContext to tracing.Identity
+func (tc *Identity) String() string {
+	return fmt.Sprintf("%016x%016x:%x:0:%x", tc.traceIDUpper, tc.traceIDLower, tc.spanID, tc.flags)
+}
+
+func (tc *Identity) MarshalBinary() ([]byte, error) {
+	buf := bytes.NewBuffer(make([]byte, 0, IdentityLength))
+	for _, field := range []interface{}{
+		tc.traceIDUpper,
+		tc.traceIDLower,
+		tc.spanID,
+		tc.flags,
+	} {
+		if err := binary.Write(buf, binary.BigEndian, field); err != nil {
+			return nil, err
+		}
+	}
+	return buf.Bytes(), nil
+}
+
+func (tc *Identity) UnmarshalBinary(data []byte) error {
+	if len(data) < IdentityLength {
+		return fmt.Errorf("expect tracingContext to have at least %d bytes, got %d", IdentityLength, len(data))
+	}
+
+	buf := bytes.NewBuffer(data)
+	for _, field := range []interface{}{
+		&tc.traceIDUpper,
+		&tc.traceIDLower,
+		&tc.spanID,
+		&tc.flags,
+	} {
+		if err := binary.Read(buf, binary.BigEndian, field); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func NewIdentity(trace string) (*Identity, error) {
+	parts := strings.Split(trace, separator)
+	if len(parts) != 4 {
+		return nil, fmt.Errorf("trace '%s' doesn't have exactly 4 parts separated by %s", trace, separator)
+	}
+	const base = 16
+	tracingID, err := padTracingID(parts[0])
+	if err != nil {
+		return nil, err
+	}
+	traceIDUpper, err := strconv.ParseUint(tracingID[:16], base, 64)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse first 16 bytes of tracing ID as uint64, err: %w", err)
+	}
+	traceIDLower, err := strconv.ParseUint(tracingID[16:], base, 64)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse last 16 bytes of tracing ID as uint64, err: %w", err)
+	}
+	spanID, err := strconv.ParseUint(parts[1], base, 64)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse span ID as uint64, err: %w", err)
+	}
+	flags, err := strconv.ParseUint(parts[3], base, 8)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse flag as uint8, err: %w", err)
+	}
+	return &Identity{
+		traceIDUpper: traceIDUpper,
+		traceIDLower: traceIDLower,
+		spanID:       spanID,
+		flags:        uint8(flags),
+	}, nil
+}
+
+func padTracingID(tracingID string) (string, error) {
+	if len(tracingID) == 0 {
+		return "", fmt.Errorf("missing tracing ID")
+	}
+	if len(tracingID) == traceID128bitsWidth {
+		return tracingID, nil
+	}
+	// Correctly left pad the trace to a length of 32
+	left := traceID128bitsWidth - len(tracingID)
+	paddedTracingID := strings.Repeat("0", left) + tracingID
+	return paddedTracingID, nil
+}

tracing/identity_test.go

@@ -0,0 +1,67 @@
+package tracing
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewIdentity(t *testing.T) {
+	testCases := []struct {
+		testCase string
+		trace    string
+		expected string
+	}{
+		{
+			testCase: "full length trace",
+			trace:    "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+			expected: "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "short trace ID",
+			trace:    "ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+			expected: "0000ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "short trace ID with 0s in the middle",
+			trace:    "ad8a01fde11f000002efdce36873:52726f6cabc144f5:0:1",
+			expected: "0000ad8a01fde11f000002efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "short trace ID with 0s in the beginning and middle",
+			trace:    "001ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+			expected: "0001ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0:1",
+		},
+		{
+			testCase: "no trace",
+			trace:    "",
+		},
+		{
+			testCase: "missing flags",
+			trace:    "ec31ad8a01fde11fdcabe2efdce36873:52726f6cabc144f5:0",
+		},
+		{
+			testCase: "missing separator",
+			trace:    "ec31ad8a01fde11fdcabe2efdce3687352726f6cabc144f501",
+		},
+	}
+
+	for _, testCase := range testCases {
+		identity, err := NewIdentity(testCase.trace)
+		if testCase.expected != "" {
+			require.NoError(t, err)
+			require.Equal(t, testCase.expected, identity.String())
+
+			serializedIdentity, err := identity.MarshalBinary()
+			require.NoError(t, err)
+			deserializedIdentity := new(Identity)
+			err = deserializedIdentity.UnmarshalBinary(serializedIdentity)
+			require.NoError(t, err)
+			require.Equal(t, identity, deserializedIdentity)
+
+		} else {
+			require.Error(t, err)
+			require.Nil(t, identity)
+		}
+	}
+}

tracing/tracing.go

@@ -189,6 +189,10 @@ func (cft *cfdTracer) AddSpans(headers http.Header) {
 	headers[CanonicalCloudflaredTracingHeader] = []string{enc}
 }
 
+func (cft *cfdTracer) ClearSpans() {
+	cft.exporter.ClearSpans()
+}
+
 // End will set the OK status for the span and then end it.
 func End(span trace.Span) {
 	endSpan(span, -1, codes.Ok, nil)
@@ -246,7 +250,6 @@ func extractTraceFromString(ctx context.Context, trace string) (context.Context,
 		parts[0] = strings.Repeat("0", left) + parts[0]
 		trace = strings.Join(parts, separator)
 	}
-
 	// Override the 'cf-trace-id' as 'uber-trace-id' so the jaeger propagator can extract it.
 	traceHeader := map[string]string{TracerContextNameOverride: trace}
 	remoteCtx := otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(traceHeader))
@@ -277,6 +280,11 @@ func extractTrace(req *http.Request) (context.Context, bool) {
 	if traceHeader[TracerContextNameOverride] == "" {
 		return nil, false
 	}
+
 	remoteCtx := otel.GetTextMapPropagator().Extract(req.Context(), propagation.MapCarrier(traceHeader))
 	return remoteCtx, true
 }
+
+func NewNoopSpan() trace.Span {
+	return trace.SpanFromContext(nil)
+}

vendor/github.com/marten-seemann/qtls-go1-18/handshake_server.go

@@ -270,7 +270,7 @@ func (hs *serverHandshakeState) processClientHello() error {
 
 	hs.ecdheOk = supportsECDHE(c.config, hs.clientHello.supportedCurves, hs.clientHello.supportedPoints)
 
-	if hs.ecdheOk {
+	if hs.ecdheOk && len(hs.clientHello.supportedPoints) > 0 {
 		// Although omitting the ec_point_formats extension is permitted, some
 		// old OpenSSL version will refuse to handshake if not present.
 		//
@@ -351,6 +351,13 @@ func supportsECDHE(c *config, supportedCurves []CurveID, supportedPoints []uint8
 			break
 		}
 	}
+	// Per RFC 8422, Section 5.1.2, if the Supported Point Formats extension is
+	// missing, uncompressed points are supported. If supportedPoints is empty,
+	// the extension must be missing, as an empty extension body is rejected by
+	// the parser. See https://go.dev/issue/49126.
+	if len(supportedPoints) == 0 {
+		supportsPointFormat = true
+	}
 
 	return supportsCurve && supportsPointFormat
 }

vendor/github.com/marten-seemann/qtls-go1-19/handshake_server.go

@@ -270,7 +270,7 @@ func (hs *serverHandshakeState) processClientHello() error {
 
 	hs.ecdheOk = supportsECDHE(c.config, hs.clientHello.supportedCurves, hs.clientHello.supportedPoints)
 
-	if hs.ecdheOk {
+	if hs.ecdheOk && len(hs.clientHello.supportedPoints) > 0 {
 		// Although omitting the ec_point_formats extension is permitted, some
 		// old OpenSSL version will refuse to handshake if not present.
 		//
@@ -351,6 +351,13 @@ func supportsECDHE(c *config, supportedCurves []CurveID, supportedPoints []uint8
 			break
 		}
 	}
+	// Per RFC 8422, Section 5.1.2, if the Supported Point Formats extension is
+	// missing, uncompressed points are supported. If supportedPoints is empty,
+	// the extension must be missing, as an empty extension body is rejected by
+	// the parser. See https://go.dev/issue/49126.
+	if len(supportedPoints) == 0 {
+		supportsPointFormat = true
+	}
 
 	return supportsCurve && supportsPointFormat
 }

vendor/modules.txt

@@ -197,10 +197,10 @@ github.com/marten-seemann/qtls-go1-16
 # github.com/marten-seemann/qtls-go1-17 v0.1.2
 ## explicit; go 1.17
 github.com/marten-seemann/qtls-go1-17
-# github.com/marten-seemann/qtls-go1-18 v0.1.2 => github.com/cloudflare/qtls-pq v0.0.0-20220824105406-fb955667e0af
+# github.com/marten-seemann/qtls-go1-18 v0.1.2 => github.com/cloudflare/qtls-pq v0.0.0-20221010110824-0053225e48b2
 ## explicit; go 1.18
 github.com/marten-seemann/qtls-go1-18
-# github.com/marten-seemann/qtls-go1-19 v0.1.0-beta.1 => github.com/cloudflare/qtls-pq v0.0.0-20220824104809-96561a41e0af
+# github.com/marten-seemann/qtls-go1-19 v0.1.0-beta.1 => github.com/cloudflare/qtls-pq v0.0.0-20221010110800-4f3769902fe0
 ## explicit; go 1.19
 github.com/marten-seemann/qtls-go1-19
 # github.com/mattn/go-colorable v0.1.8
@@ -583,5 +583,5 @@ zombiezen.com/go/capnproto2/std/capnp/rpc
 # github.com/lucas-clemente/quic-go => github.com/chungthuang/quic-go v0.27.1-0.20220809135021-ca330f1dec9f
 # github.com/prometheus/golang_client => github.com/prometheus/golang_client v1.12.1
 # gopkg.in/yaml.v3 => gopkg.in/yaml.v3 v3.0.1
-# github.com/marten-seemann/qtls-go1-18 => github.com/cloudflare/qtls-pq v0.0.0-20220824105406-fb955667e0af
-# github.com/marten-seemann/qtls-go1-19 => github.com/cloudflare/qtls-pq v0.0.0-20220824104809-96561a41e0af
+# github.com/marten-seemann/qtls-go1-18 => github.com/cloudflare/qtls-pq v0.0.0-20221010110824-0053225e48b2
+# github.com/marten-seemann/qtls-go1-19 => github.com/cloudflare/qtls-pq v0.0.0-20221010110800-4f3769902fe0