Release 2024.11.0

## Summary
The initial purpose of this PR was to bump the base image from buster to bookworm however these tests are no longer exercised hence the removal

Closes VULN-66059

.gitignore

@@ -17,3 +17,4 @@ cscope.*
 ssh_server_tests/.env
 /.cover
 built_artifacts/
+component-tests/.venv

.teamcity/install-cloudflare-go.sh

@@ -3,6 +3,6 @@
 cd /tmp
 git clone -q https://github.com/cloudflare/go
 cd go/src
-# https://github.com/cloudflare/go/tree/ec0a014545f180b0c74dfd687698657a9e86e310 is version go1.22.2-devel-cf
-git checkout -q ec0a014545f180b0c74dfd687698657a9e86e310
-./make.bash
+# https://github.com/cloudflare/go/tree/f4334cdc0c3f22a3bfdd7e66f387e3ffc65a5c38 is version go1.22.5-devel-cf
+git checkout -q f4334cdc0c3f22a3bfdd7e66f387e3ffc65a5c38
+./make.bash

.teamcity/windows/component-test.ps1

@@ -37,7 +37,7 @@ if ($LASTEXITCODE -ne 0) { throw "Failed unit tests" }
 
 Write-Output "Running component tests"
 
-python -m pip --disable-pip-version-check install --upgrade -r component-tests/requirements.txt
+python -m pip --disable-pip-version-check install --upgrade -r component-tests/requirements.txt --use-pep517
 python component-tests/setup.py --type create
 python -m pytest component-tests -o log_cli=true --log-cli-level=INFO
 if ($LASTEXITCODE -ne 0) {

.teamcity/windows/install-cloudflare-go.ps1

@@ -9,8 +9,8 @@ Set-Location "$Env:Temp"
 git clone -q https://github.com/cloudflare/go
 Write-Output "Building go..."
 cd go/src
-# https://github.com/cloudflare/go/tree/ec0a014545f180b0c74dfd687698657a9e86e310 is version go1.22.2-devel-cf
-git checkout -q ec0a014545f180b0c74dfd687698657a9e86e310
+# https://github.com/cloudflare/go/tree/f4334cdc0c3f22a3bfdd7e66f387e3ffc65a5c38 is version go1.22.5-devel-cf
+git checkout -q f4334cdc0c3f22a3bfdd7e66f387e3ffc65a5c38
 & ./make.bat
 
-Write-Output "Installed"
+Write-Output "Installed"

.teamcity/windows/install-go-msi.ps1

@@ -1,6 +1,6 @@
 $ErrorActionPreference = "Stop"
 $ProgressPreference = "SilentlyContinue"
-$GoMsiVersion = "go1.22.2.windows-amd64.msi"
+$GoMsiVersion = "go1.22.5.windows-amd64.msi"
 
 Write-Output "Downloading go installer..."
 
@@ -17,4 +17,4 @@ Install-Package "$Env:Temp\$GoMsiVersion" -Force
 # Go installer updates global $PATH
 go env
 
-Write-Output "Installed"
+Write-Output "Installed"

Dockerfile

@@ -1,7 +1,7 @@
 # use a builder image for building cloudflare
 ARG TARGET_GOOS
 ARG TARGET_GOARCH
-FROM golang:1.22.2 as builder
+FROM golang:1.22.5 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0 \
     TARGET_GOOS=${TARGET_GOOS} \

Dockerfile.amd64

@@ -1,5 +1,5 @@
 # use a builder image for building cloudflare
-FROM golang:1.22.2 as builder
+FROM golang:1.22.5 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0
 

Dockerfile.arm64

@@ -1,5 +1,5 @@
 # use a builder image for building cloudflare
-FROM golang:1.22.2 as builder
+FROM golang:1.22.5 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0
 

Makefile

@@ -165,10 +165,6 @@ cover:
 	# Generate the HTML report that can be viewed from the browser in CI.
 	$Q go tool cover -html ".cover/c.out" -o .cover/all.html
 
-.PHONY: test-ssh-server
-test-ssh-server:
-	docker-compose -f ssh_server_tests/docker-compose.yml up
-
 .PHONY: install-go
 install-go:
 	rm -rf ${CF_GO_PATH}

RELEASE_NOTES

@@ -1,3 +1,15 @@
+2024.11.0
+- 2024-11-05 VULN-66059: remove ssh server tests
+- 2024-11-04 TUN-8700: Add datagram v3 muxer
+- 2024-11-04 TUN-8646: Allow experimental feature support for datagram v3
+- 2024-11-04 TUN-8641: Expose methods to simplify V3 Datagram parsing on the edge
+- 2024-10-31 TUN-8708: Bump python min version to 3.10
+- 2024-10-31 TUN-8667: Add datagram v3 session manager
+- 2024-10-25 TUN-8692: remove dashes from session id
+- 2024-10-24 TUN-8694: Rework release script
+- 2024-10-24 TUN-8661: Refactor connection methods to support future different datagram muxing methods
+- 2024-07-22 TUN-8553: Bump go to 1.22.5 and go-boring 1.22.5-1
+
 2024.10.1
 - 2024-10-23 TUN-8694: Fix github release script
 - 2024-10-21 Revert "TUN-8592: Use metadata from the edge to determine if request body is empty for QUIC transport"

cfsetup.yaml

@@ -1,4 +1,4 @@
-pinned_go: &pinned_go go-boring=1.22.2-1
+pinned_go: &pinned_go go-boring=1.22.5-1
 
 build_dir: &build_dir /cfsetup_build
 default-flavor: bullseye

component-tests/README.md

@@ -1,9 +1,9 @@
 # Requirements
-1. Python 3.7 or later with packages in the given `requirements.txt`
-   - E.g. with conda:
-   - `conda create -n component-tests python=3.7`  
-   - `conda activate component-tests`
-   - `pip3 install -r requirements.txt`
+1. Python 3.10 or later with packages in the given `requirements.txt`
+   - E.g. with venv:
+   - `python3 -m venv ./.venv`  
+   - `source ./.venv/bin/activate`
+   - `python3 -m pip install -r requirements.txt`
 
 2. Create a config yaml file, for example:
 ```

connection/connection.go

@@ -36,6 +36,13 @@ var (
 	flushableContentTypes = []string{sseContentType, grpcContentType}
 )
 
+// TunnelConnection represents the connection to the edge.
+// The Serve method is provided to allow clients to handle any errors from the connection encountered during
+// processing of the connection. Cancelling of the context provided to Serve will close the connection.
+type TunnelConnection interface {
+	Serve(ctx context.Context) error
+}
+
 type Orchestrator interface {
 	UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
 	GetConfigJSON() ([]byte, error)

connection/quic.go

@@ -1,51 +1,16 @@
 package connection
 
 import (
-	"bufio"
 	"context"
 	"crypto/tls"
 	"fmt"
-	"io"
 	"net"
-	"net/http"
 	"net/netip"
 	"runtime"
-	"strconv"
-	"strings"
 	"sync"
-	"sync/atomic"
-	"time"
 
-	"github.com/google/uuid"
-	"github.com/pkg/errors"
 	"github.com/quic-go/quic-go"
 	"github.com/rs/zerolog"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/trace"
-	"golang.org/x/sync/errgroup"
-
-	"github.com/cloudflare/cloudflared/datagramsession"
-	"github.com/cloudflare/cloudflared/ingress"
-	"github.com/cloudflare/cloudflared/management"
-	"github.com/cloudflare/cloudflared/packet"
-	cfdquic "github.com/cloudflare/cloudflared/quic"
-	"github.com/cloudflare/cloudflared/tracing"
-	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	rpcquic "github.com/cloudflare/cloudflared/tunnelrpc/quic"
-)
-
-const (
-	// HTTPHeaderKey is used to get or set http headers in QUIC ALPN if the underlying proxy connection type is HTTP.
-	HTTPHeaderKey = "HttpHeader"
-	// HTTPMethodKey is used to get or set http method in QUIC ALPN if the underlying proxy connection type is HTTP.
-	HTTPMethodKey = "HttpMethod"
-	// HTTPHostKey is used to get or set http Method in QUIC ALPN if the underlying proxy connection type is HTTP.
-	HTTPHostKey = "HttpHost"
-
-	QUICMetadataFlowID = "FlowID"
-	// emperically this capacity has been working well
-	demuxChanCapacity = 16
 )
 
 var (
@@ -53,48 +18,21 @@ var (
 	portMapMutex     sync.Mutex
 )
 
-// QUICConnection represents the type that facilitates Proxying via QUIC streams.
-type QUICConnection struct {
-	session      quic.Connection
-	logger       *zerolog.Logger
-	orchestrator Orchestrator
-	// sessionManager tracks active sessions. It receives datagrams from quic connection via datagramMuxer
-	sessionManager datagramsession.Manager
-	// datagramMuxer mux/demux datagrams from quic connection
-	datagramMuxer        *cfdquic.DatagramMuxerV2
-	packetRouter         *ingress.PacketRouter
-	controlStreamHandler ControlStreamHandler
-	connOptions          *tunnelpogs.ConnectionOptions
-	connIndex            uint8
-
-	rpcTimeout         time.Duration
-	streamWriteTimeout time.Duration
-	gracePeriod        time.Duration
-}
-
-// NewQUICConnection returns a new instance of QUICConnection.
-func NewQUICConnection(
+func DialQuic(
 	ctx context.Context,
 	quicConfig *quic.Config,
+	tlsConfig *tls.Config,
 	edgeAddr netip.AddrPort,
 	localAddr net.IP,
 	connIndex uint8,
-	tlsConfig *tls.Config,
-	orchestrator Orchestrator,
-	connOptions *tunnelpogs.ConnectionOptions,
-	controlStreamHandler ControlStreamHandler,
 	logger *zerolog.Logger,
-	packetRouterConfig *ingress.GlobalRouterConfig,
-	rpcTimeout time.Duration,
-	streamWriteTimeout time.Duration,
-	gracePeriod time.Duration,
-) (*QUICConnection, error) {
+) (quic.Connection, error) {
 	udpConn, err := createUDPConnForConnIndex(connIndex, localAddr, edgeAddr, logger)
 	if err != nil {
 		return nil, err
 	}
 
-	session, err := quic.Dial(ctx, udpConn, net.UDPAddrFromAddrPort(edgeAddr), tlsConfig, quicConfig)
+	conn, err := quic.Dial(ctx, udpConn, net.UDPAddrFromAddrPort(edgeAddr), tlsConfig, quicConfig)
 	if err != nil {
 		// close the udp server socket in case of error connecting to the edge
 		udpConn.Close()
@@ -102,506 +40,11 @@ func NewQUICConnection(
 	}
 
 	// wrap the session, so that the UDPConn is closed after session is closed.
-	session = &wrapCloseableConnQuicConnection{
-		session,
+	conn = &wrapCloseableConnQuicConnection{
+		conn,
 		udpConn,
 	}
-
-	sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
-	datagramMuxer := cfdquic.NewDatagramMuxerV2(session, logger, sessionDemuxChan)
-	sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
-	packetRouter := ingress.NewPacketRouter(packetRouterConfig, datagramMuxer, logger)
-
-	return &QUICConnection{
-		session:              session,
-		orchestrator:         orchestrator,
-		logger:               logger,
-		sessionManager:       sessionManager,
-		datagramMuxer:        datagramMuxer,
-		packetRouter:         packetRouter,
-		controlStreamHandler: controlStreamHandler,
-		connOptions:          connOptions,
-		connIndex:            connIndex,
-		rpcTimeout:           rpcTimeout,
-		streamWriteTimeout:   streamWriteTimeout,
-		gracePeriod:          gracePeriod,
-	}, nil
-}
-
-// Serve starts a QUIC session that begins accepting streams.
-func (q *QUICConnection) Serve(ctx context.Context) error {
-	// origintunneld assumes the first stream is used for the control plane
-	controlStream, err := q.session.OpenStream()
-	if err != nil {
-		return fmt.Errorf("failed to open a registration control stream: %w", err)
-	}
-
-	// If either goroutine returns nil error, we rely on this cancellation to make sure the other goroutine exits
-	// as fast as possible as well. Nil error means we want to exit for good (caller code won't retry serving this
-	// connection).
-	// If either goroutine returns a non nil error, then the error group cancels the context, thus also canceling the
-	// other goroutine as fast as possible.
-	ctx, cancel := context.WithCancel(ctx)
-	errGroup, ctx := errgroup.WithContext(ctx)
-
-	// In the future, if cloudflared can autonomously push traffic to the edge, we have to make sure the control
-	// stream is already fully registered before the other goroutines can proceed.
-	errGroup.Go(func() error {
-		// err is equal to nil if we exit due to unregistration. If that happens we want to wait the full
-		// amount of the grace period, allowing requests to finish before we cancel the context, which will
-		// make cloudflared exit.
-		if err := q.serveControlStream(ctx, controlStream); err == nil {
-			select {
-			case <-ctx.Done():
-			case <-time.Tick(q.gracePeriod):
-			}
-		}
-		cancel()
-		return err
-	})
-	errGroup.Go(func() error {
-		defer cancel()
-		return q.acceptStream(ctx)
-	})
-	errGroup.Go(func() error {
-		defer cancel()
-		return q.sessionManager.Serve(ctx)
-	})
-	errGroup.Go(func() error {
-		defer cancel()
-		return q.datagramMuxer.ServeReceive(ctx)
-	})
-	errGroup.Go(func() error {
-		defer cancel()
-		return q.packetRouter.Serve(ctx)
-	})
-
-	return errGroup.Wait()
-}
-
-func (q *QUICConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
-	// This blocks until the control plane is done.
-	err := q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
-	if err != nil {
-		// Not wrapping error here to be consistent with the http2 message.
-		return err
-	}
-
-	return nil
-}
-
-// Close closes the session with no errors specified.
-func (q *QUICConnection) Close() {
-	q.session.CloseWithError(0, "")
-}
-
-func (q *QUICConnection) acceptStream(ctx context.Context) error {
-	defer q.Close()
-	for {
-		quicStream, err := q.session.AcceptStream(ctx)
-		if err != nil {
-			// context.Canceled is usually a user ctrl+c. We don't want to log an error here as it's intentional.
-			if errors.Is(err, context.Canceled) || q.controlStreamHandler.IsStopped() {
-				return nil
-			}
-			return fmt.Errorf("failed to accept QUIC stream: %w", err)
-		}
-		go q.runStream(quicStream)
-	}
-}
-
-func (q *QUICConnection) runStream(quicStream quic.Stream) {
-	ctx := quicStream.Context()
-	stream := cfdquic.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
-	defer stream.Close()
-
-	// we are going to fuse readers/writers from stream <- cloudflared -> origin, and we want to guarantee that
-	// code executed in the code path of handleStream don't trigger an earlier close to the downstream write stream.
-	// So, we wrap the stream with a no-op write closer and only this method can actually close write side of the stream.
-	// A call to close will simulate a close to the read-side, which will fail subsequent reads.
-	noCloseStream := &nopCloserReadWriter{ReadWriteCloser: stream}
-	ss := rpcquic.NewCloudflaredServer(q.handleDataStream, q, q, q.rpcTimeout)
-	if err := ss.Serve(ctx, noCloseStream); err != nil {
-		q.logger.Debug().Err(err).Msg("Failed to handle QUIC stream")
-
-		// if we received an error at this level, then close write side of stream with an error, which will result in
-		// RST_STREAM frame.
-		quicStream.CancelWrite(0)
-	}
-}
-
-func (q *QUICConnection) handleDataStream(ctx context.Context, stream *rpcquic.RequestServerStream) error {
-	request, err := stream.ReadConnectRequestData()
-	if err != nil {
-		return err
-	}
-
-	if err, connectResponseSent := q.dispatchRequest(ctx, stream, err, request); err != nil {
-		q.logger.Err(err).Str("type", request.Type.String()).Str("dest", request.Dest).Msg("Request failed")
-
-		// if the connectResponse was already sent and we had an error, we need to propagate it up, so that the stream is
-		// closed with an RST_STREAM frame
-		if connectResponseSent {
-			return err
-		}
-
-		if writeRespErr := stream.WriteConnectResponseData(err); writeRespErr != nil {
-			return writeRespErr
-		}
-	}
-
-	return nil
-}
-
-// dispatchRequest will dispatch the request depending on the type and returns an error if it occurs.
-// More importantly, it also tells if the during processing of the request the ConnectResponse metadata was sent downstream.
-// This is important since it informs
-func (q *QUICConnection) dispatchRequest(ctx context.Context, stream *rpcquic.RequestServerStream, err error, request *pogs.ConnectRequest) (error, bool) {
-	originProxy, err := q.orchestrator.GetOriginProxy()
-	if err != nil {
-		return err, false
-	}
-
-	switch request.Type {
-	case pogs.ConnectionTypeHTTP, pogs.ConnectionTypeWebsocket:
-		tracedReq, err := buildHTTPRequest(ctx, request, stream, q.connIndex, q.logger)
-		if err != nil {
-			return err, false
-		}
-		w := newHTTPResponseAdapter(stream)
-		return originProxy.ProxyHTTP(&w, tracedReq, request.Type == pogs.ConnectionTypeWebsocket), w.connectResponseSent
-
-	case pogs.ConnectionTypeTCP:
-		rwa := &streamReadWriteAcker{RequestServerStream: stream}
-		metadata := request.MetadataMap()
-		return originProxy.ProxyTCP(ctx, rwa, &TCPRequest{
-			Dest:      request.Dest,
-			FlowID:    metadata[QUICMetadataFlowID],
-			CfTraceID: metadata[tracing.TracerContextName],
-			ConnIndex: q.connIndex,
-		}), rwa.connectResponseSent
-	default:
-		return errors.Errorf("unsupported error type: %s", request.Type), false
-	}
-}
-
-// RegisterUdpSession is the RPC method invoked by edge to register and run a session
-func (q *QUICConnection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16, closeAfterIdleHint time.Duration, traceContext string) (*tunnelpogs.RegisterUdpSessionResponse, error) {
-	traceCtx := tracing.NewTracedContext(ctx, traceContext, q.logger)
-	ctx, registerSpan := traceCtx.Tracer().Start(traceCtx, "register-session", trace.WithAttributes(
-		attribute.String("session-id", sessionID.String()),
-		attribute.String("dst", fmt.Sprintf("%s:%d", dstIP, dstPort)),
-	))
-	log := q.logger.With().Int(management.EventTypeKey, int(management.UDP)).Logger()
-	// Each session is a series of datagram from an eyeball to a dstIP:dstPort.
-	// (src port, dst IP, dst port) uniquely identifies a session, so it needs a dedicated connected socket.
-	originProxy, err := ingress.DialUDP(dstIP, dstPort)
-	if err != nil {
-		log.Err(err).Msgf("Failed to create udp proxy to %s:%d", dstIP, dstPort)
-		tracing.EndWithErrorStatus(registerSpan, err)
-		return nil, err
-	}
-	registerSpan.SetAttributes(
-		attribute.Bool("socket-bind-success", true),
-		attribute.String("src", originProxy.LocalAddr().String()),
-	)
-
-	session, err := q.sessionManager.RegisterSession(ctx, sessionID, originProxy)
-	if err != nil {
-		originProxy.Close()
-		log.Err(err).Str("sessionID", sessionID.String()).Msgf("Failed to register udp session")
-		tracing.EndWithErrorStatus(registerSpan, err)
-		return nil, err
-	}
-
-	go q.serveUDPSession(session, closeAfterIdleHint)
-
-	log.Debug().
-		Str("sessionID", sessionID.String()).
-		Str("src", originProxy.LocalAddr().String()).
-		Str("dst", fmt.Sprintf("%s:%d", dstIP, dstPort)).
-		Msgf("Registered session")
-	tracing.End(registerSpan)
-
-	resp := tunnelpogs.RegisterUdpSessionResponse{
-		Spans: traceCtx.GetProtoSpans(),
-	}
-
-	return &resp, nil
-}
-
-func (q *QUICConnection) serveUDPSession(session *datagramsession.Session, closeAfterIdleHint time.Duration) {
-	ctx := q.session.Context()
-	closedByRemote, err := session.Serve(ctx, closeAfterIdleHint)
-	// If session is terminated by remote, then we know it has been unregistered from session manager and edge
-	if !closedByRemote {
-		if err != nil {
-			q.closeUDPSession(ctx, session.ID, err.Error())
-		} else {
-			q.closeUDPSession(ctx, session.ID, "terminated without error")
-		}
-	}
-	q.logger.Debug().Err(err).
-		Int(management.EventTypeKey, int(management.UDP)).
-		Str("sessionID", session.ID.String()).
-		Msg("Session terminated")
-}
-
-// closeUDPSession first unregisters the session from session manager, then it tries to unregister from edge
-func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUID, message string) {
-	q.sessionManager.UnregisterSession(ctx, sessionID, message, false)
-	quicStream, err := q.session.OpenStream()
-	if err != nil {
-		// Log this at debug because this is not an error if session was closed due to lost connection
-		// with edge
-		q.logger.Debug().Err(err).
-			Int(management.EventTypeKey, int(management.UDP)).
-			Str("sessionID", sessionID.String()).
-			Msgf("Failed to open quic stream to unregister udp session with edge")
-		return
-	}
-
-	stream := cfdquic.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
-	defer stream.Close()
-	rpcClientStream, err := rpcquic.NewSessionClient(ctx, stream, q.rpcTimeout)
-	if err != nil {
-		// Log this at debug because this is not an error if session was closed due to lost connection
-		// with edge
-		q.logger.Err(err).Str("sessionID", sessionID.String()).
-			Msgf("Failed to open rpc stream to unregister udp session with edge")
-		return
-	}
-	defer rpcClientStream.Close()
-
-	if err := rpcClientStream.UnregisterUdpSession(ctx, sessionID, message); err != nil {
-		q.logger.Err(err).Str("sessionID", sessionID.String()).
-			Msgf("Failed to unregister udp session with edge")
-	}
-}
-
-// UnregisterUdpSession is the RPC method invoked by edge to unregister and terminate a sesssion
-func (q *QUICConnection) UnregisterUdpSession(ctx context.Context, sessionID uuid.UUID, message string) error {
-	return q.sessionManager.UnregisterSession(ctx, sessionID, message, true)
-}
-
-// UpdateConfiguration is the RPC method invoked by edge when there is a new configuration
-func (q *QUICConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
-	return q.orchestrator.UpdateConfig(version, config)
-}
-
-// streamReadWriteAcker is a light wrapper over QUIC streams with a callback to send response back to
-// the client.
-type streamReadWriteAcker struct {
-	*rpcquic.RequestServerStream
-	connectResponseSent bool
-}
-
-// AckConnection acks response back to the proxy.
-func (s *streamReadWriteAcker) AckConnection(tracePropagation string) error {
-	metadata := []pogs.Metadata{}
-	// Only add tracing if provided by origintunneld
-	if tracePropagation != "" {
-		metadata = append(metadata, pogs.Metadata{
-			Key: tracing.CanonicalCloudflaredTracingHeader,
-			Val: tracePropagation,
-		})
-	}
-	s.connectResponseSent = true
-	return s.WriteConnectResponseData(nil, metadata...)
-}
-
-// httpResponseAdapter translates responses written by the HTTP Proxy into ones that can be used in QUIC.
-type httpResponseAdapter struct {
-	*rpcquic.RequestServerStream
-	headers             http.Header
-	connectResponseSent bool
-}
-
-func newHTTPResponseAdapter(s *rpcquic.RequestServerStream) httpResponseAdapter {
-	return httpResponseAdapter{RequestServerStream: s, headers: make(http.Header)}
-}
-
-func (hrw *httpResponseAdapter) AddTrailer(trailerName, trailerValue string) {
-	// we do not support trailers over QUIC
-}
-
-func (hrw *httpResponseAdapter) WriteRespHeaders(status int, header http.Header) error {
-	metadata := make([]pogs.Metadata, 0)
-	metadata = append(metadata, pogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(status)})
-	for k, vv := range header {
-		for _, v := range vv {
-			httpHeaderKey := fmt.Sprintf("%s:%s", HTTPHeaderKey, k)
-			metadata = append(metadata, pogs.Metadata{Key: httpHeaderKey, Val: v})
-		}
-	}
-
-	return hrw.WriteConnectResponseData(nil, metadata...)
-}
-
-func (hrw *httpResponseAdapter) Write(p []byte) (int, error) {
-	// Make sure to send WriteHeader response if not called yet
-	if !hrw.connectResponseSent {
-		hrw.WriteRespHeaders(http.StatusOK, hrw.headers)
-	}
-	return hrw.RequestServerStream.Write(p)
-}
-
-func (hrw *httpResponseAdapter) Header() http.Header {
-	return hrw.headers
-}
-
-// This is a no-op Flush because this adapter is over a quic.Stream and we don't need Flush here.
-func (hrw *httpResponseAdapter) Flush() {}
-
-func (hrw *httpResponseAdapter) WriteHeader(status int) {
-	hrw.WriteRespHeaders(status, hrw.headers)
-}
-
-func (hrw *httpResponseAdapter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
-	conn := &localProxyConnection{hrw.ReadWriteCloser}
-	readWriter := bufio.NewReadWriter(
-		bufio.NewReader(hrw.ReadWriteCloser),
-		bufio.NewWriter(hrw.ReadWriteCloser),
-	)
-	return conn, readWriter, nil
-}
-
-func (hrw *httpResponseAdapter) WriteErrorResponse(err error) {
-	hrw.WriteConnectResponseData(err, pogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(http.StatusBadGateway)})
-}
-
-func (hrw *httpResponseAdapter) WriteConnectResponseData(respErr error, metadata ...pogs.Metadata) error {
-	hrw.connectResponseSent = true
-	return hrw.RequestServerStream.WriteConnectResponseData(respErr, metadata...)
-}
-
-func buildHTTPRequest(
-	ctx context.Context,
-	connectRequest *pogs.ConnectRequest,
-	body io.ReadCloser,
-	connIndex uint8,
-	log *zerolog.Logger,
-) (*tracing.TracedHTTPRequest, error) {
-	metadata := connectRequest.MetadataMap()
-	dest := connectRequest.Dest
-	method := metadata[HTTPMethodKey]
-	host := metadata[HTTPHostKey]
-	isWebsocket := connectRequest.Type == pogs.ConnectionTypeWebsocket
-
-	req, err := http.NewRequestWithContext(ctx, method, dest, body)
-	if err != nil {
-		return nil, err
-	}
-
-	req.Host = host
-	for _, metadata := range connectRequest.Metadata {
-		if strings.Contains(metadata.Key, HTTPHeaderKey) {
-			// metadata.Key is off the format httpHeaderKey:<HTTPHeader>
-			httpHeaderKey := strings.Split(metadata.Key, ":")
-			if len(httpHeaderKey) != 2 {
-				return nil, fmt.Errorf("header Key: %s malformed", metadata.Key)
-			}
-			req.Header.Add(httpHeaderKey[1], metadata.Val)
-		}
-	}
-	// Go's http.Client automatically sends chunked request body if this value is not set on the
-	// *http.Request struct regardless of header:
-	// https://go.googlesource.com/go/+/go1.8rc2/src/net/http/transfer.go#154.
-	if err := setContentLength(req); err != nil {
-		return nil, fmt.Errorf("Error setting content-length: %w", err)
-	}
-
-	// Go's client defaults to chunked encoding after a 200ms delay if the following cases are true:
-	//   * the request body blocks
-	//   * the content length is not set (or set to -1)
-	//   * the method doesn't usually have a body (GET, HEAD, DELETE, ...)
-	//   * there is no transfer-encoding=chunked already set.
-	// So, if transfer cannot be chunked and content length is 0, we dont set a request body.
-	if !isWebsocket && !isTransferEncodingChunked(req) && req.ContentLength == 0 {
-		req.Body = http.NoBody
-	}
-	stripWebsocketUpgradeHeader(req)
-
-	// Check for tracing on request
-	tracedReq := tracing.NewTracedHTTPRequest(req, connIndex, log)
-	return tracedReq, err
-}
-
-func setContentLength(req *http.Request) error {
-	var err error
-	if contentLengthStr := req.Header.Get("Content-Length"); contentLengthStr != "" {
-		req.ContentLength, err = strconv.ParseInt(contentLengthStr, 10, 64)
-	}
-	return err
-}
-
-func isTransferEncodingChunked(req *http.Request) bool {
-	transferEncodingVal := req.Header.Get("Transfer-Encoding")
-	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding suggests that this can be a comma
-	// separated value as well.
-	return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
-}
-
-// A helper struct that guarantees a call to close only affects read side, but not write side.
-type nopCloserReadWriter struct {
-	io.ReadWriteCloser
-
-	// for use by Read only
-	// we don't need a memory barrier here because there is an implicit assumption that
-	// Read calls can't happen concurrently by different go-routines.
-	sawEOF bool
-	// should be updated and read using atomic primitives.
-	// value is read in Read method and written in Close method, which could be done by different
-	// go-routines.
-	closed uint32
-}
-
-func (np *nopCloserReadWriter) Read(p []byte) (n int, err error) {
-	if np.sawEOF {
-		return 0, io.EOF
-	}
-
-	if atomic.LoadUint32(&np.closed) > 0 {
-		return 0, fmt.Errorf("closed by handler")
-	}
-
-	n, err = np.ReadWriteCloser.Read(p)
-	if err == io.EOF {
-		np.sawEOF = true
-	}
-
-	return
-}
-
-func (np *nopCloserReadWriter) Close() error {
-	atomic.StoreUint32(&np.closed, 1)
-
-	return nil
-}
-
-// muxerWrapper wraps DatagramMuxerV2 to satisfy the packet.FunnelUniPipe interface
-type muxerWrapper struct {
-	muxer *cfdquic.DatagramMuxerV2
-}
-
-func (rp *muxerWrapper) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
-	return rp.muxer.SendPacket(cfdquic.RawPacket(pk))
-}
-
-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.(cfdquic.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
+	return conn, nil
 }
 
 func createUDPConnForConnIndex(connIndex uint8, localIP net.IP, edgeIP netip.AddrPort, logger *zerolog.Logger) (*net.UDPConn, error) {

connection/quic_connection.go

@@ -0,0 +1,444 @@
+package connection
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/netip"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/quic-go/quic-go"
+	"github.com/rs/zerolog"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/cloudflare/cloudflared/packet"
+	cfdquic "github.com/cloudflare/cloudflared/quic"
+	"github.com/cloudflare/cloudflared/tracing"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	rpcquic "github.com/cloudflare/cloudflared/tunnelrpc/quic"
+)
+
+const (
+	// HTTPHeaderKey is used to get or set http headers in QUIC ALPN if the underlying proxy connection type is HTTP.
+	HTTPHeaderKey = "HttpHeader"
+	// HTTPMethodKey is used to get or set http method in QUIC ALPN if the underlying proxy connection type is HTTP.
+	HTTPMethodKey = "HttpMethod"
+	// HTTPHostKey is used to get or set http host in QUIC ALPN if the underlying proxy connection type is HTTP.
+	HTTPHostKey = "HttpHost"
+
+	QUICMetadataFlowID = "FlowID"
+)
+
+// quicConnection represents the type that facilitates Proxying via QUIC streams.
+type quicConnection struct {
+	conn                 quic.Connection
+	logger               *zerolog.Logger
+	orchestrator         Orchestrator
+	datagramHandler      DatagramSessionHandler
+	controlStreamHandler ControlStreamHandler
+	connOptions          *tunnelpogs.ConnectionOptions
+	connIndex            uint8
+
+	rpcTimeout         time.Duration
+	streamWriteTimeout time.Duration
+	gracePeriod        time.Duration
+}
+
+// NewTunnelConnection takes a [quic.Connection] to wrap it for use with cloudflared application logic.
+func NewTunnelConnection(
+	ctx context.Context,
+	conn quic.Connection,
+	connIndex uint8,
+	orchestrator Orchestrator,
+	datagramSessionHandler DatagramSessionHandler,
+	controlStreamHandler ControlStreamHandler,
+	connOptions *pogs.ConnectionOptions,
+	rpcTimeout time.Duration,
+	streamWriteTimeout time.Duration,
+	gracePeriod time.Duration,
+	logger *zerolog.Logger,
+) (TunnelConnection, error) {
+	return &quicConnection{
+		conn:                 conn,
+		logger:               logger,
+		orchestrator:         orchestrator,
+		datagramHandler:      datagramSessionHandler,
+		controlStreamHandler: controlStreamHandler,
+		connOptions:          connOptions,
+		connIndex:            connIndex,
+		rpcTimeout:           rpcTimeout,
+		streamWriteTimeout:   streamWriteTimeout,
+		gracePeriod:          gracePeriod,
+	}, nil
+}
+
+// Serve starts a QUIC connection that begins accepting streams.
+func (q *quicConnection) Serve(ctx context.Context) error {
+	// The edge assumes the first stream is used for the control plane
+	controlStream, err := q.conn.OpenStream()
+	if err != nil {
+		return fmt.Errorf("failed to open a registration control stream: %w", err)
+	}
+
+	// If either goroutine returns nil error, we rely on this cancellation to make sure the other goroutine exits
+	// as fast as possible as well. Nil error means we want to exit for good (caller code won't retry serving this
+	// connection).
+	// If either goroutine returns a non nil error, then the error group cancels the context, thus also canceling the
+	// other goroutine as fast as possible.
+	ctx, cancel := context.WithCancel(ctx)
+	errGroup, ctx := errgroup.WithContext(ctx)
+
+	// In the future, if cloudflared can autonomously push traffic to the edge, we have to make sure the control
+	// stream is already fully registered before the other goroutines can proceed.
+	errGroup.Go(func() error {
+		// err is equal to nil if we exit due to unregistration. If that happens we want to wait the full
+		// amount of the grace period, allowing requests to finish before we cancel the context, which will
+		// make cloudflared exit.
+		if err := q.serveControlStream(ctx, controlStream); err == nil {
+			select {
+			case <-ctx.Done():
+			case <-time.Tick(q.gracePeriod):
+			}
+		}
+		cancel()
+		return err
+
+	})
+	errGroup.Go(func() error {
+		defer cancel()
+		return q.acceptStream(ctx)
+	})
+	errGroup.Go(func() error {
+		defer cancel()
+		return q.datagramHandler.Serve(ctx)
+	})
+
+	return errGroup.Wait()
+}
+
+// serveControlStream will serve the RPC; blocking until the control plane is done.
+func (q *quicConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
+	return q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
+}
+
+// Close the connection with no errors specified.
+func (q *quicConnection) Close() {
+	q.conn.CloseWithError(0, "")
+}
+
+func (q *quicConnection) acceptStream(ctx context.Context) error {
+	defer q.Close()
+	for {
+		quicStream, err := q.conn.AcceptStream(ctx)
+		if err != nil {
+			// context.Canceled is usually a user ctrl+c. We don't want to log an error here as it's intentional.
+			if errors.Is(err, context.Canceled) || q.controlStreamHandler.IsStopped() {
+				return nil
+			}
+			return fmt.Errorf("failed to accept QUIC stream: %w", err)
+		}
+		go q.runStream(quicStream)
+	}
+}
+
+func (q *quicConnection) runStream(quicStream quic.Stream) {
+	ctx := quicStream.Context()
+	stream := cfdquic.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
+	defer stream.Close()
+
+	// we are going to fuse readers/writers from stream <- cloudflared -> origin, and we want to guarantee that
+	// code executed in the code path of handleStream don't trigger an earlier close to the downstream write stream.
+	// So, we wrap the stream with a no-op write closer and only this method can actually close write side of the stream.
+	// A call to close will simulate a close to the read-side, which will fail subsequent reads.
+	noCloseStream := &nopCloserReadWriter{ReadWriteCloser: stream}
+	ss := rpcquic.NewCloudflaredServer(q.handleDataStream, q.datagramHandler, q, q.rpcTimeout)
+	if err := ss.Serve(ctx, noCloseStream); err != nil {
+		q.logger.Debug().Err(err).Msg("Failed to handle QUIC stream")
+
+		// if we received an error at this level, then close write side of stream with an error, which will result in
+		// RST_STREAM frame.
+		quicStream.CancelWrite(0)
+	}
+}
+
+func (q *quicConnection) handleDataStream(ctx context.Context, stream *rpcquic.RequestServerStream) error {
+	request, err := stream.ReadConnectRequestData()
+	if err != nil {
+		return err
+	}
+
+	if err, connectResponseSent := q.dispatchRequest(ctx, stream, request); err != nil {
+		q.logger.Err(err).Str("type", request.Type.String()).Str("dest", request.Dest).Msg("Request failed")
+
+		// if the connectResponse was already sent and we had an error, we need to propagate it up, so that the stream is
+		// closed with an RST_STREAM frame
+		if connectResponseSent {
+			return err
+		}
+
+		if writeRespErr := stream.WriteConnectResponseData(err); writeRespErr != nil {
+			return writeRespErr
+		}
+	}
+
+	return nil
+}
+
+// dispatchRequest will dispatch the request to the origin depending on the type and returns an error if it occurs.
+// Also returns if the connect response was sent to the downstream during processing of the origin request.
+func (q *quicConnection) dispatchRequest(ctx context.Context, stream *rpcquic.RequestServerStream, request *pogs.ConnectRequest) (err error, connectResponseSent bool) {
+	originProxy, err := q.orchestrator.GetOriginProxy()
+	if err != nil {
+		return err, false
+	}
+
+	switch request.Type {
+	case pogs.ConnectionTypeHTTP, pogs.ConnectionTypeWebsocket:
+		tracedReq, err := buildHTTPRequest(ctx, request, stream, q.connIndex, q.logger)
+		if err != nil {
+			return err, false
+		}
+		w := newHTTPResponseAdapter(stream)
+		return originProxy.ProxyHTTP(&w, tracedReq, request.Type == pogs.ConnectionTypeWebsocket), w.connectResponseSent
+
+	case pogs.ConnectionTypeTCP:
+		rwa := &streamReadWriteAcker{RequestServerStream: stream}
+		metadata := request.MetadataMap()
+		return originProxy.ProxyTCP(ctx, rwa, &TCPRequest{
+			Dest:      request.Dest,
+			FlowID:    metadata[QUICMetadataFlowID],
+			CfTraceID: metadata[tracing.TracerContextName],
+			ConnIndex: q.connIndex,
+		}), rwa.connectResponseSent
+	default:
+		return errors.Errorf("unsupported error type: %s", request.Type), false
+	}
+}
+
+// UpdateConfiguration is the RPC method invoked by edge when there is a new configuration
+func (q *quicConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
+	return q.orchestrator.UpdateConfig(version, config)
+}
+
+// streamReadWriteAcker is a light wrapper over QUIC streams with a callback to send response back to
+// the client.
+type streamReadWriteAcker struct {
+	*rpcquic.RequestServerStream
+	connectResponseSent bool
+}
+
+// AckConnection acks response back to the proxy.
+func (s *streamReadWriteAcker) AckConnection(tracePropagation string) error {
+	metadata := []pogs.Metadata{}
+	// Only add tracing if provided by the edge request
+	if tracePropagation != "" {
+		metadata = append(metadata, pogs.Metadata{
+			Key: tracing.CanonicalCloudflaredTracingHeader,
+			Val: tracePropagation,
+		})
+	}
+	s.connectResponseSent = true
+	return s.WriteConnectResponseData(nil, metadata...)
+}
+
+// httpResponseAdapter translates responses written by the HTTP Proxy into ones that can be used in QUIC.
+type httpResponseAdapter struct {
+	*rpcquic.RequestServerStream
+	headers             http.Header
+	connectResponseSent bool
+}
+
+func newHTTPResponseAdapter(s *rpcquic.RequestServerStream) httpResponseAdapter {
+	return httpResponseAdapter{RequestServerStream: s, headers: make(http.Header)}
+}
+
+func (hrw *httpResponseAdapter) AddTrailer(trailerName, trailerValue string) {
+	// we do not support trailers over QUIC
+}
+
+func (hrw *httpResponseAdapter) WriteRespHeaders(status int, header http.Header) error {
+	metadata := make([]pogs.Metadata, 0)
+	metadata = append(metadata, pogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(status)})
+	for k, vv := range header {
+		for _, v := range vv {
+			httpHeaderKey := fmt.Sprintf("%s:%s", HTTPHeaderKey, k)
+			metadata = append(metadata, pogs.Metadata{Key: httpHeaderKey, Val: v})
+		}
+	}
+
+	return hrw.WriteConnectResponseData(nil, metadata...)
+}
+
+func (hrw *httpResponseAdapter) Write(p []byte) (int, error) {
+	// Make sure to send WriteHeader response if not called yet
+	if !hrw.connectResponseSent {
+		hrw.WriteRespHeaders(http.StatusOK, hrw.headers)
+	}
+	return hrw.RequestServerStream.Write(p)
+}
+
+func (hrw *httpResponseAdapter) Header() http.Header {
+	return hrw.headers
+}
+
+// This is a no-op Flush because this adapter is over a quic.Stream and we don't need Flush here.
+func (hrw *httpResponseAdapter) Flush() {}
+
+func (hrw *httpResponseAdapter) WriteHeader(status int) {
+	hrw.WriteRespHeaders(status, hrw.headers)
+}
+
+func (hrw *httpResponseAdapter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	conn := &localProxyConnection{hrw.ReadWriteCloser}
+	readWriter := bufio.NewReadWriter(
+		bufio.NewReader(hrw.ReadWriteCloser),
+		bufio.NewWriter(hrw.ReadWriteCloser),
+	)
+	return conn, readWriter, nil
+}
+
+func (hrw *httpResponseAdapter) WriteErrorResponse(err error) {
+	hrw.WriteConnectResponseData(err, pogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(http.StatusBadGateway)})
+}
+
+func (hrw *httpResponseAdapter) WriteConnectResponseData(respErr error, metadata ...pogs.Metadata) error {
+	hrw.connectResponseSent = true
+	return hrw.RequestServerStream.WriteConnectResponseData(respErr, metadata...)
+}
+
+func buildHTTPRequest(
+	ctx context.Context,
+	connectRequest *pogs.ConnectRequest,
+	body io.ReadCloser,
+	connIndex uint8,
+	log *zerolog.Logger,
+) (*tracing.TracedHTTPRequest, error) {
+	metadata := connectRequest.MetadataMap()
+	dest := connectRequest.Dest
+	method := metadata[HTTPMethodKey]
+	host := metadata[HTTPHostKey]
+	isWebsocket := connectRequest.Type == pogs.ConnectionTypeWebsocket
+
+	req, err := http.NewRequestWithContext(ctx, method, dest, body)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Host = host
+	for _, metadata := range connectRequest.Metadata {
+		if strings.Contains(metadata.Key, HTTPHeaderKey) {
+			// metadata.Key is off the format httpHeaderKey:<HTTPHeader>
+			httpHeaderKey := strings.Split(metadata.Key, ":")
+			if len(httpHeaderKey) != 2 {
+				return nil, fmt.Errorf("header Key: %s malformed", metadata.Key)
+			}
+			req.Header.Add(httpHeaderKey[1], metadata.Val)
+		}
+	}
+	// Go's http.Client automatically sends chunked request body if this value is not set on the
+	// *http.Request struct regardless of header:
+	// https://go.googlesource.com/go/+/go1.8rc2/src/net/http/transfer.go#154.
+	if err := setContentLength(req); err != nil {
+		return nil, fmt.Errorf("Error setting content-length: %w", err)
+	}
+
+	// Go's client defaults to chunked encoding after a 200ms delay if the following cases are true:
+	//   * the request body blocks
+	//   * the content length is not set (or set to -1)
+	//   * the method doesn't usually have a body (GET, HEAD, DELETE, ...)
+	//   * there is no transfer-encoding=chunked already set.
+	// So, if transfer cannot be chunked and content length is 0, we dont set a request body.
+	if !isWebsocket && !isTransferEncodingChunked(req) && req.ContentLength == 0 {
+		req.Body = http.NoBody
+	}
+	stripWebsocketUpgradeHeader(req)
+
+	// Check for tracing on request
+	tracedReq := tracing.NewTracedHTTPRequest(req, connIndex, log)
+	return tracedReq, err
+}
+
+func setContentLength(req *http.Request) error {
+	var err error
+	if contentLengthStr := req.Header.Get("Content-Length"); contentLengthStr != "" {
+		req.ContentLength, err = strconv.ParseInt(contentLengthStr, 10, 64)
+	}
+	return err
+}
+
+func isTransferEncodingChunked(req *http.Request) bool {
+	transferEncodingVal := req.Header.Get("Transfer-Encoding")
+	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding suggests that this can be a comma
+	// separated value as well.
+	return strings.Contains(strings.ToLower(transferEncodingVal), "chunked")
+}
+
+// A helper struct that guarantees a call to close only affects read side, but not write side.
+type nopCloserReadWriter struct {
+	io.ReadWriteCloser
+
+	// for use by Read only
+	// we don't need a memory barrier here because there is an implicit assumption that
+	// Read calls can't happen concurrently by different go-routines.
+	sawEOF bool
+	// should be updated and read using atomic primitives.
+	// value is read in Read method and written in Close method, which could be done by different
+	// go-routines.
+	closed uint32
+}
+
+func (np *nopCloserReadWriter) Read(p []byte) (n int, err error) {
+	if np.sawEOF {
+		return 0, io.EOF
+	}
+
+	if atomic.LoadUint32(&np.closed) > 0 {
+		return 0, fmt.Errorf("closed by handler")
+	}
+
+	n, err = np.ReadWriteCloser.Read(p)
+	if err == io.EOF {
+		np.sawEOF = true
+	}
+
+	return
+}
+
+func (np *nopCloserReadWriter) Close() error {
+	atomic.StoreUint32(&np.closed, 1)
+
+	return nil
+}
+
+// muxerWrapper wraps DatagramMuxerV2 to satisfy the packet.FunnelUniPipe interface
+type muxerWrapper struct {
+	muxer *cfdquic.DatagramMuxerV2
+}
+
+func (rp *muxerWrapper) SendPacket(dst netip.Addr, pk packet.RawPacket) error {
+	return rp.muxer.SendPacket(cfdquic.RawPacket(pk))
+}
+
+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.(cfdquic.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
+}

connection/quic_connection_test.go

@@ -15,7 +15,6 @@ import (
 	"net/http"
 	"net/netip"
 	"net/url"
-	"os"
 	"strings"
 	"testing"
 	"time"
@@ -30,10 +29,11 @@ import (
 	"golang.org/x/net/nettest"
 
 	"github.com/cloudflare/cloudflared/datagramsession"
+	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/packet"
 	cfdquic "github.com/cloudflare/cloudflared/quic"
 	"github.com/cloudflare/cloudflared/tracing"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 	rpcquic "github.com/cloudflare/cloudflared/tunnelrpc/quic"
 )
 
@@ -164,11 +164,11 @@ func TestQUICServer(t *testing.T) {
 				close(serverDone)
 			}()
 
-			qc := testQUICConnection(netip.MustParseAddrPort(udpListener.LocalAddr().String()), t, uint8(i))
+			tunnelConn, _ := testTunnelConnection(t, netip.MustParseAddrPort(udpListener.LocalAddr().String()), uint8(i))
 
 			connDone := make(chan struct{})
 			go func() {
-				qc.Serve(ctx)
+				tunnelConn.Serve(ctx)
 				close(connDone)
 			}()
 
@@ -528,13 +528,14 @@ func TestServeUDPSession(t *testing.T) {
 	}()
 
 	// Random index to avoid reusing port
-	qc := testQUICConnection(netip.MustParseAddrPort(udpListener.LocalAddr().String()), t, 28)
-	go qc.Serve(ctx)
+	tunnelConn, datagramConn := testTunnelConnection(t, netip.MustParseAddrPort(udpListener.LocalAddr().String()), 28)
+	go tunnelConn.Serve(ctx)
 
 	edgeQUICSession := <-edgeQUICSessionChan
-	serveSession(ctx, qc, edgeQUICSession, closedByOrigin, io.EOF.Error(), t)
-	serveSession(ctx, qc, edgeQUICSession, closedByTimeout, datagramsession.SessionIdleErr(time.Millisecond*50).Error(), t)
-	serveSession(ctx, qc, edgeQUICSession, closedByRemote, "eyeball closed connection", t)
+
+	serveSession(ctx, datagramConn, edgeQUICSession, closedByOrigin, io.EOF.Error(), t)
+	serveSession(ctx, datagramConn, edgeQUICSession, closedByTimeout, datagramsession.SessionIdleErr(time.Millisecond*50).Error(), t)
+	serveSession(ctx, datagramConn, edgeQUICSession, closedByRemote, "eyeball closed connection", t)
 	cancel()
 }
 
@@ -619,19 +620,19 @@ func testCreateUDPConnReuseSourcePortForEdgeIP(t *testing.T, edgeIP netip.AddrPo
 	require.NotEqual(t, initialPort, getPortFunc(conn))
 }
 
-func serveSession(ctx context.Context, qc *QUICConnection, edgeQUICSession quic.Connection, closeType closeReason, expectedReason string, t *testing.T) {
+func serveSession(ctx context.Context, datagramConn *datagramV2Connection, edgeQUICSession quic.Connection, closeType closeReason, expectedReason string, t *testing.T) {
 	var (
 		payload = []byte(t.Name())
 	)
 	sessionID := uuid.New()
 	cfdConn, originConn := net.Pipe()
 	// Registers and run a new session
-	session, err := qc.sessionManager.RegisterSession(ctx, sessionID, cfdConn)
+	session, err := datagramConn.sessionManager.RegisterSession(ctx, sessionID, cfdConn)
 	require.NoError(t, err)
 
 	sessionDone := make(chan struct{})
 	go func() {
-		qc.serveUDPSession(session, time.Millisecond*50)
+		datagramConn.serveUDPSession(session, time.Millisecond*50)
 		close(sessionDone)
 	}()
 
@@ -655,7 +656,7 @@ func serveSession(ctx context.Context, qc *QUICConnection, edgeQUICSession quic.
 	case closedByOrigin:
 		originConn.Close()
 	case closedByRemote:
-		err = qc.UnregisterUdpSession(ctx, sessionID, expectedReason)
+		err = datagramConn.UnregisterUdpSession(ctx, sessionID, expectedReason)
 		require.NoError(t, err)
 	case closedByTimeout:
 	}
@@ -726,33 +727,58 @@ func (s mockSessionRPCServer) UnregisterUdpSession(ctx context.Context, sessionI
 	return nil
 }
 
-func testQUICConnection(udpListenerAddr netip.AddrPort, t *testing.T, index uint8) *QUICConnection {
+func testTunnelConnection(t *testing.T, serverAddr netip.AddrPort, index uint8) (TunnelConnection, *datagramV2Connection) {
 	tlsClientConfig := &tls.Config{
 		InsecureSkipVerify: true,
 		NextProtos:         []string{"argotunnel"},
 	}
 	// Start a mock httpProxy
-	log := zerolog.New(os.Stdout)
+	log := zerolog.New(io.Discard)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	qc, err := NewQUICConnection(
+
+	// Dial the QUIC connection to the edge
+	conn, err := DialQuic(
 		ctx,
 		testQUICConfig,
-		udpListenerAddr,
-		nil,
-		index,
 		tlsClientConfig,
-		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
-		&tunnelpogs.ConnectionOptions{},
-		fakeControlStream{},
+		serverAddr,
+		nil, // connect on a random port
+		index,
 		&log,
-		nil,
+	)
+
+	// Start a session manager for the connection
+	sessionDemuxChan := make(chan *packet.Session, 4)
+	datagramMuxer := cfdquic.NewDatagramMuxerV2(conn, &log, sessionDemuxChan)
+	sessionManager := datagramsession.NewManager(&log, datagramMuxer.SendToSession, sessionDemuxChan)
+	packetRouter := ingress.NewPacketRouter(nil, datagramMuxer, &log)
+
+	datagramConn := &datagramV2Connection{
+		conn,
+		sessionManager,
+		datagramMuxer,
+		packetRouter,
+		15 * time.Second,
+		0 * time.Second,
+		&log,
+	}
+
+	tunnelConn, err := NewTunnelConnection(
+		ctx,
+		conn,
+		index,
+		&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
+		datagramConn,
+		fakeControlStream{},
+		&pogs.ConnectionOptions{},
 		15*time.Second,
 		0*time.Second,
 		0*time.Second,
+		&log,
 	)
 	require.NoError(t, err)
-	return qc
+	return tunnelConn, datagramConn
 }
 
 type mockReaderNoopWriter struct {

connection/quic_datagram_v2.go

@@ -0,0 +1,200 @@
+package connection
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/quic-go/quic-go"
+	"github.com/rs/zerolog"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/cloudflare/cloudflared/datagramsession"
+	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/management"
+	"github.com/cloudflare/cloudflared/packet"
+	cfdquic "github.com/cloudflare/cloudflared/quic"
+	"github.com/cloudflare/cloudflared/tracing"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	rpcquic "github.com/cloudflare/cloudflared/tunnelrpc/quic"
+)
+
+const (
+	// emperically this capacity has been working well
+	demuxChanCapacity = 16
+)
+
+// DatagramSessionHandler is a service that can serve datagrams for a connection and handle sessions from incoming
+// connection streams.
+type DatagramSessionHandler interface {
+	Serve(context.Context) error
+
+	pogs.SessionManager
+}
+
+type datagramV2Connection struct {
+	conn quic.Connection
+
+	// sessionManager tracks active sessions. It receives datagrams from quic connection via datagramMuxer
+	sessionManager datagramsession.Manager
+	// datagramMuxer mux/demux datagrams from quic connection
+	datagramMuxer *cfdquic.DatagramMuxerV2
+	packetRouter  *ingress.PacketRouter
+
+	rpcTimeout         time.Duration
+	streamWriteTimeout time.Duration
+
+	logger *zerolog.Logger
+}
+
+func NewDatagramV2Connection(ctx context.Context,
+	conn quic.Connection,
+	packetConfig *ingress.GlobalRouterConfig,
+	rpcTimeout time.Duration,
+	streamWriteTimeout time.Duration,
+	logger *zerolog.Logger,
+) DatagramSessionHandler {
+	sessionDemuxChan := make(chan *packet.Session, demuxChanCapacity)
+	datagramMuxer := cfdquic.NewDatagramMuxerV2(conn, logger, sessionDemuxChan)
+	sessionManager := datagramsession.NewManager(logger, datagramMuxer.SendToSession, sessionDemuxChan)
+	packetRouter := ingress.NewPacketRouter(packetConfig, datagramMuxer, logger)
+
+	return &datagramV2Connection{
+		conn,
+		sessionManager,
+		datagramMuxer,
+		packetRouter,
+		rpcTimeout,
+		streamWriteTimeout,
+		logger,
+	}
+}
+
+func (d *datagramV2Connection) Serve(ctx context.Context) error {
+	// If either goroutine returns nil error, we rely on this cancellation to make sure the other goroutine exits
+	// as fast as possible as well. Nil error means we want to exit for good (caller code won't retry serving this
+	// connection).
+	// If either goroutine returns a non nil error, then the error group cancels the context, thus also canceling the
+	// other goroutine as fast as possible.
+	ctx, cancel := context.WithCancel(ctx)
+	errGroup, ctx := errgroup.WithContext(ctx)
+
+	errGroup.Go(func() error {
+		defer cancel()
+		return d.sessionManager.Serve(ctx)
+	})
+	errGroup.Go(func() error {
+		defer cancel()
+		return d.datagramMuxer.ServeReceive(ctx)
+	})
+	errGroup.Go(func() error {
+		defer cancel()
+		return d.packetRouter.Serve(ctx)
+	})
+
+	return errGroup.Wait()
+}
+
+// RegisterUdpSession is the RPC method invoked by edge to register and run a session
+func (q *datagramV2Connection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16, closeAfterIdleHint time.Duration, traceContext string) (*tunnelpogs.RegisterUdpSessionResponse, error) {
+	traceCtx := tracing.NewTracedContext(ctx, traceContext, q.logger)
+	ctx, registerSpan := traceCtx.Tracer().Start(traceCtx, "register-session", trace.WithAttributes(
+		attribute.String("session-id", sessionID.String()),
+		attribute.String("dst", fmt.Sprintf("%s:%d", dstIP, dstPort)),
+	))
+	log := q.logger.With().Int(management.EventTypeKey, int(management.UDP)).Logger()
+	// Each session is a series of datagram from an eyeball to a dstIP:dstPort.
+	// (src port, dst IP, dst port) uniquely identifies a session, so it needs a dedicated connected socket.
+	originProxy, err := ingress.DialUDP(dstIP, dstPort)
+	if err != nil {
+		log.Err(err).Msgf("Failed to create udp proxy to %s:%d", dstIP, dstPort)
+		tracing.EndWithErrorStatus(registerSpan, err)
+		return nil, err
+	}
+	registerSpan.SetAttributes(
+		attribute.Bool("socket-bind-success", true),
+		attribute.String("src", originProxy.LocalAddr().String()),
+	)
+
+	session, err := q.sessionManager.RegisterSession(ctx, sessionID, originProxy)
+	if err != nil {
+		originProxy.Close()
+		log.Err(err).Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(sessionID)).Msgf("Failed to register udp session")
+		tracing.EndWithErrorStatus(registerSpan, err)
+		return nil, err
+	}
+
+	go q.serveUDPSession(session, closeAfterIdleHint)
+
+	log.Debug().
+		Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(sessionID)).
+		Str("src", originProxy.LocalAddr().String()).
+		Str("dst", fmt.Sprintf("%s:%d", dstIP, dstPort)).
+		Msgf("Registered session")
+	tracing.End(registerSpan)
+
+	resp := tunnelpogs.RegisterUdpSessionResponse{
+		Spans: traceCtx.GetProtoSpans(),
+	}
+
+	return &resp, nil
+}
+
+// UnregisterUdpSession is the RPC method invoked by edge to unregister and terminate a sesssion
+func (q *datagramV2Connection) UnregisterUdpSession(ctx context.Context, sessionID uuid.UUID, message string) error {
+	return q.sessionManager.UnregisterSession(ctx, sessionID, message, true)
+}
+
+func (q *datagramV2Connection) serveUDPSession(session *datagramsession.Session, closeAfterIdleHint time.Duration) {
+	ctx := q.conn.Context()
+	closedByRemote, err := session.Serve(ctx, closeAfterIdleHint)
+	// If session is terminated by remote, then we know it has been unregistered from session manager and edge
+	if !closedByRemote {
+		if err != nil {
+			q.closeUDPSession(ctx, session.ID, err.Error())
+		} else {
+			q.closeUDPSession(ctx, session.ID, "terminated without error")
+		}
+	}
+	q.logger.Debug().Err(err).
+		Int(management.EventTypeKey, int(management.UDP)).
+		Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(session.ID)).
+		Msg("Session terminated")
+}
+
+// closeUDPSession first unregisters the session from session manager, then it tries to unregister from edge
+func (q *datagramV2Connection) closeUDPSession(ctx context.Context, sessionID uuid.UUID, message string) {
+	q.sessionManager.UnregisterSession(ctx, sessionID, message, false)
+	quicStream, err := q.conn.OpenStream()
+	if err != nil {
+		// Log this at debug because this is not an error if session was closed due to lost connection
+		// with edge
+		q.logger.Debug().Err(err).
+			Int(management.EventTypeKey, int(management.UDP)).
+			Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(sessionID)).
+			Msgf("Failed to open quic stream to unregister udp session with edge")
+		return
+	}
+
+	stream := cfdquic.NewSafeStreamCloser(quicStream, q.streamWriteTimeout, q.logger)
+	defer stream.Close()
+	rpcClientStream, err := rpcquic.NewSessionClient(ctx, stream, q.rpcTimeout)
+	if err != nil {
+		// Log this at debug because this is not an error if session was closed due to lost connection
+		// with edge
+		q.logger.Err(err).Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(sessionID)).
+			Msgf("Failed to open rpc stream to unregister udp session with edge")
+		return
+	}
+	defer rpcClientStream.Close()
+
+	if err := rpcClientStream.UnregisterUdpSession(ctx, sessionID, message); err != nil {
+		q.logger.Err(err).Str(datagramsession.LogFieldSessionID, datagramsession.FormatSessionID(sessionID)).
+			Msgf("Failed to unregister udp session with edge")
+	}
+}

connection/quic_datagram_v3.go

@@ -0,0 +1,48 @@
+package connection
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/quic-go/quic-go"
+	"github.com/rs/zerolog"
+
+	cfdquic "github.com/cloudflare/cloudflared/quic/v3"
+	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+)
+
+type datagramV3Connection struct {
+	conn quic.Connection
+	// datagramMuxer mux/demux datagrams from quic connection
+	datagramMuxer cfdquic.DatagramConn
+	logger        *zerolog.Logger
+}
+
+func NewDatagramV3Connection(ctx context.Context,
+	conn quic.Connection,
+	sessionManager cfdquic.SessionManager,
+	logger *zerolog.Logger,
+) DatagramSessionHandler {
+	datagramMuxer := cfdquic.NewDatagramConn(conn, sessionManager, logger)
+
+	return &datagramV3Connection{
+		conn,
+		datagramMuxer,
+		logger,
+	}
+}
+
+func (d *datagramV3Connection) Serve(ctx context.Context) error {
+	return d.datagramMuxer.Serve(ctx)
+}
+
+func (d *datagramV3Connection) RegisterUdpSession(ctx context.Context, sessionID uuid.UUID, dstIP net.IP, dstPort uint16, closeAfterIdleHint time.Duration, traceContext string) (*pogs.RegisterUdpSessionResponse, error) {
+	return nil, fmt.Errorf("datagram v3 does not support RegisterUdpSession RPC")
+}
+
+func (d *datagramV3Connection) UnregisterUdpSession(ctx context.Context, sessionID uuid.UUID, message string) error {
+	return fmt.Errorf("datagram v3 does not support UnregisterUdpSession RPC")
+}

datagramsession/manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -20,8 +21,15 @@ const (
 
 var (
 	errSessionManagerClosed = fmt.Errorf("session manager closed")
+	LogFieldSessionID       = "sessionID"
 )
 
+func FormatSessionID(sessionID uuid.UUID) string {
+	sessionIDStr := sessionID.String()
+	sessionIDStr = strings.ReplaceAll(sessionIDStr, "-", "")
+	return sessionIDStr
+}
+
 // Manager defines the APIs to manage sessions from the same transport.
 type Manager interface {
 	// Serve starts the event loop
@@ -127,7 +135,7 @@ func (m *manager) registerSession(ctx context.Context, registration *registerSes
 func (m *manager) newSession(id uuid.UUID, dstConn io.ReadWriteCloser) *Session {
 	logger := m.log.With().
 		Int(management.EventTypeKey, int(management.UDP)).
-		Str("sessionID", id.String()).Logger()
+		Str(LogFieldSessionID, FormatSessionID(id)).Logger()
 	return &Session{
 		ID:       id,
 		sendFunc: m.sendFunc,
@@ -174,7 +182,7 @@ func (m *manager) unregisterSession(unregistration *unregisterSessionEvent) {
 func (m *manager) sendToSession(datagram *packet.Session) {
 	session, ok := m.sessions[datagram.ID]
 	if !ok {
-		m.log.Error().Str("sessionID", datagram.ID.String()).Msg("session not found")
+		m.log.Error().Str(LogFieldSessionID, FormatSessionID(datagram.ID)).Msg("session not found")
 		return
 	}
 	// session writes to destination over a connected UDP socket, which should not be blocking, so this call doesn't

dev.Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.22.2 as builder
+FROM golang:1.22.5 as builder
 ENV GO111MODULE=on \
     CGO_ENABLED=0
 WORKDIR /go/src/github.com/cloudflare/cloudflared/

github_release.py

@@ -286,21 +286,20 @@ def main():
         else:
             client = Github(args.api_key)
             repo = client.get_repo(CLOUDFLARED_REPO)
-            release = get_or_create_release(repo, args.release_version, args.dry_run)
 
             if os.path.isdir(args.path):
                 onlyfiles = [f for f in listdir(args.path) if isfile(join(args.path, f))]
                 for filename in onlyfiles:
                     binary_path = os.path.join(args.path, filename)
                     assert_asset_version(binary_path, args.release_version)
+                release = get_or_create_release(repo, args.release_version, args.dry_run)
                 for filename in onlyfiles:
                     binary_path = os.path.join(args.path, filename)
                     upload_asset(release, binary_path, filename, args.release_version, args.kv_account_id, args.namespace_id,
                     args.kv_api_token)
                     move_asset(binary_path, filename)
             else:
-                upload_asset(release, args.path, args.name, args.release_version, args.kv_account_id, args.namespace_id,
-                    args.kv_api_token)
+                raise Exception("the argument path must be a directory")
 
     except Exception as e:
         logging.exception(e)

ingress/origin_udp_proxy.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"net/netip"
 )
 
 type UDPProxy interface {
@@ -30,3 +31,16 @@ func DialUDP(dstIP net.IP, dstPort uint16) (UDPProxy, error) {
 
 	return &udpProxy{udpConn}, nil
 }
+
+func DialUDPAddrPort(dest netip.AddrPort) (*net.UDPConn, error) {
+	addr := net.UDPAddrFromAddrPort(dest)
+
+	// We use nil as local addr to force runtime to find the best suitable local address IP given the destination
+	// address as context.
+	udpConn, err := net.DialUDP("udp", nil, addr)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create UDP proxy to origin (%v:%v): %w", dest.Addr(), dest.Port(), err)
+	}
+
+	return udpConn, nil
+}

proxy/logger.go

@@ -16,11 +16,14 @@ const (
 	logFieldLBProbe       = "lbProbe"
 	logFieldRule          = "ingressRule"
 	logFieldOriginService = "originService"
-	logFieldFlowID        = "flowID"
 	logFieldConnIndex     = "connIndex"
 	logFieldDestAddr      = "destAddr"
 )
 
+var (
+	LogFieldFlowID = "flowID"
+)
+
 // newHTTPLogger creates a child zerolog.Logger from the provided with added context from the HTTP request, ingress
 // services, and connection index.
 func newHTTPLogger(logger *zerolog.Logger, connIndex uint8, req *http.Request, rule int, serviceName string) zerolog.Logger {
@@ -47,7 +50,7 @@ func newTCPLogger(logger *zerolog.Logger, req *connection.TCPRequest) zerolog.Lo
 		Int(management.EventTypeKey, int(management.TCP)).
 		Uint8(logFieldConnIndex, req.ConnIndex).
 		Str(logFieldOriginService, ingress.ServiceWarpRouting).
-		Str(logFieldFlowID, req.FlowID).
+		Str(LogFieldFlowID, req.FlowID).
 		Str(logFieldDestAddr, req.Dest).
 		Uint8(logFieldConnIndex, req.ConnIndex).
 		Logger()

quic/v3/datagram.go

@@ -24,10 +24,10 @@ const (
 	datagramTypeLen = 1
 
 	// 1280 is the default datagram packet length used before MTU discovery: https://github.com/quic-go/quic-go/blob/v0.45.0/internal/protocol/params.go#L12
-	maxDatagramLen = 1280
+	maxDatagramPayloadLen = 1280
 )
 
-func parseDatagramType(data []byte) (DatagramType, error) {
+func ParseDatagramType(data []byte) (DatagramType, error) {
 	if len(data) < datagramTypeLen {
 		return 0, ErrDatagramHeaderTooSmall
 	}
@@ -100,10 +100,10 @@ func (s *UDPSessionRegistrationDatagram) MarshalBinary() (data []byte, err error
 	}
 	var maxPayloadLen int
 	if ipv6 {
-		maxPayloadLen = maxDatagramLen - sessionRegistrationIPv6DatagramHeaderLen
+		maxPayloadLen = maxDatagramPayloadLen + sessionRegistrationIPv6DatagramHeaderLen
 		flags |= sessionRegistrationFlagsIPMask
 	} else {
-		maxPayloadLen = maxDatagramLen - sessionRegistrationIPv4DatagramHeaderLen
+		maxPayloadLen = maxDatagramPayloadLen + sessionRegistrationIPv4DatagramHeaderLen
 	}
 	// Make sure that the payload being bundled can actually fit in the payload destination
 	if len(s.Payload) > maxPayloadLen {
@@ -140,7 +140,7 @@ func (s *UDPSessionRegistrationDatagram) MarshalBinary() (data []byte, err error
 }
 
 func (s *UDPSessionRegistrationDatagram) UnmarshalBinary(data []byte) error {
-	datagramType, err := parseDatagramType(data)
+	datagramType, err := ParseDatagramType(data)
 	if err != nil {
 		return err
 	}
@@ -192,10 +192,10 @@ type UDPSessionPayloadDatagram struct {
 }
 
 const (
-	datagramPayloadHeaderLen = datagramTypeLen + datagramRequestIdLen
+	DatagramPayloadHeaderLen = datagramTypeLen + datagramRequestIdLen
 
 	// The maximum size that a proxied UDP payload can be in a [UDPSessionPayloadDatagram]
-	maxPayloadPlusHeaderLen = maxDatagramLen - datagramPayloadHeaderLen
+	maxPayloadPlusHeaderLen = maxDatagramPayloadLen + DatagramPayloadHeaderLen
 )
 
 // The datagram structure for UDPSessionPayloadDatagram is:
@@ -230,7 +230,7 @@ func MarshalPayloadHeaderTo(requestID RequestID, payload []byte) error {
 }
 
 func (s *UDPSessionPayloadDatagram) UnmarshalBinary(data []byte) error {
-	datagramType, err := parseDatagramType(data)
+	datagramType, err := ParseDatagramType(data)
 	if err != nil {
 		return err
 	}
@@ -270,7 +270,7 @@ const (
 	datagramSessionRegistrationResponseLen = datagramTypeLen + datagramRespTypeLen + datagramRequestIdLen + datagramRespErrMsgLen
 
 	// The maximum size that an error message can be in a [UDPSessionRegistrationResponseDatagram].
-	maxResponseErrorMessageLen = maxDatagramLen - datagramSessionRegistrationResponseLen
+	maxResponseErrorMessageLen = maxDatagramPayloadLen - datagramSessionRegistrationResponseLen
 )
 
 // SessionRegistrationResp represents all of the responses that a UDP session registration response
@@ -284,6 +284,8 @@ const (
 	ResponseDestinationUnreachable SessionRegistrationResp = 0x01
 	// Session registration was unable to bind to a local UDP socket.
 	ResponseUnableToBindSocket SessionRegistrationResp = 0x02
+	// Session registration is already bound to another connection.
+	ResponseSessionAlreadyConnected SessionRegistrationResp = 0x03
 	// Session registration failed with an unexpected error but provided a message.
 	ResponseErrorWithMsg SessionRegistrationResp = 0xff
 )
@@ -330,7 +332,7 @@ func (s *UDPSessionRegistrationResponseDatagram) MarshalBinary() (data []byte, e
 }
 
 func (s *UDPSessionRegistrationResponseDatagram) UnmarshalBinary(data []byte) error {
-	datagramType, err := parseDatagramType(data)
+	datagramType, err := ParseDatagramType(data)
 	if err != nil {
 		return wrapUnmarshalErr(err)
 	}

quic/v3/datagram_errors.go

@@ -7,7 +7,7 @@ import (
 
 var (
 	ErrInvalidDatagramType                 error = errors.New("invalid datagram type expected")
-	ErrDatagramHeaderTooSmall              error = fmt.Errorf("datagram should have at least %d bytes", datagramTypeLen)
+	ErrDatagramHeaderTooSmall              error = fmt.Errorf("datagram should have at least %d byte", datagramTypeLen)
 	ErrDatagramPayloadTooLarge             error = errors.New("payload length is too large to be bundled in datagram")
 	ErrDatagramPayloadHeaderTooSmall       error = errors.New("payload length is too small to fit the datagram header")
 	ErrDatagramPayloadInvalidSize          error = errors.New("datagram provided is an invalid size")

quic/v3/datagram_test.go

@@ -21,7 +21,7 @@ func makePayload(size int) []byte {
 }
 
 func TestSessionRegistration_MarshalUnmarshal(t *testing.T) {
-	payload := makePayload(1254)
+	payload := makePayload(1280)
 	tests := []*v3.UDPSessionRegistrationDatagram{
 		// Default (IPv4)
 		{
@@ -236,7 +236,7 @@ func TestSessionPayload(t *testing.T) {
 	})
 
 	t.Run("payload size too large", func(t *testing.T) {
-		datagram := makePayload(17 + 1264) // 1263 is the largest payload size allowed
+		datagram := makePayload(17 + 1281) // 1280 is the largest payload size allowed
 		err := v3.MarshalPayloadHeaderTo(testRequestID, datagram)
 		if err != nil {
 			t.Error(err)

quic/v3/manager.go

@@ -0,0 +1,87 @@
+package v3
+
+import (
+	"errors"
+	"net"
+	"net/netip"
+	"sync"
+
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/ingress"
+)
+
+var (
+	ErrSessionNotFound         = errors.New("session not found")
+	ErrSessionBoundToOtherConn = errors.New("session is in use by another connection")
+)
+
+type SessionManager interface {
+	// RegisterSession will register a new session if it does not already exist for the request ID.
+	// During new session creation, the session will also bind the UDP socket for the origin.
+	// If the session exists for a different connection, it will return [ErrSessionBoundToOtherConn].
+	RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramWriter) (Session, error)
+	// GetSession returns an active session if available for the provided connection.
+	// If the session does not exist, it will return [ErrSessionNotFound]. If the session exists for a different
+	// connection, it will return [ErrSessionBoundToOtherConn].
+	GetSession(requestID RequestID) (Session, error)
+	// UnregisterSession will remove a session from the current session manager. It will attempt to close the session
+	// before removal.
+	UnregisterSession(requestID RequestID)
+}
+
+type DialUDP func(dest netip.AddrPort) (*net.UDPConn, error)
+
+type sessionManager struct {
+	sessions map[RequestID]Session
+	mutex    sync.RWMutex
+	log      *zerolog.Logger
+}
+
+func NewSessionManager(log *zerolog.Logger, originDialer DialUDP) SessionManager {
+	return &sessionManager{
+		sessions: make(map[RequestID]Session),
+		log:      log,
+	}
+}
+
+func (s *sessionManager) RegisterSession(request *UDPSessionRegistrationDatagram, conn DatagramWriter) (Session, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	// Check to make sure session doesn't already exist for requestID
+	_, exists := s.sessions[request.RequestID]
+	if exists {
+		return nil, ErrSessionBoundToOtherConn
+	}
+	// Attempt to bind the UDP socket for the new session
+	origin, err := ingress.DialUDPAddrPort(request.Dest)
+	if err != nil {
+		return nil, err
+	}
+	// Create and insert the new session in the map
+	session := NewSession(request.RequestID, request.IdleDurationHint, origin, conn, s.log)
+	s.sessions[request.RequestID] = session
+	return session, nil
+}
+
+func (s *sessionManager) GetSession(requestID RequestID) (Session, error) {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+	session, exists := s.sessions[requestID]
+	if exists {
+		return session, nil
+	}
+	return nil, ErrSessionNotFound
+}
+
+func (s *sessionManager) UnregisterSession(requestID RequestID) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	// Get the session and make sure to close it if it isn't already closed
+	session, exists := s.sessions[requestID]
+	if exists {
+		// We ignore any errors when attempting to close the session
+		_ = session.Close()
+	}
+	delete(s.sessions, requestID)
+}

quic/v3/manager_test.go

@@ -0,0 +1,74 @@
+package v3_test
+
+import (
+	"errors"
+	"net/netip"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/ingress"
+	v3 "github.com/cloudflare/cloudflared/quic/v3"
+)
+
+func TestRegisterSession(t *testing.T) {
+	log := zerolog.Nop()
+	manager := v3.NewSessionManager(&log, ingress.DialUDPAddrPort)
+
+	request := v3.UDPSessionRegistrationDatagram{
+		RequestID:        testRequestID,
+		Dest:             netip.MustParseAddrPort("127.0.0.1:5000"),
+		Traced:           false,
+		IdleDurationHint: 5 * time.Second,
+		Payload:          nil,
+	}
+	session, err := manager.RegisterSession(&request, &noopEyeball{})
+	if err != nil {
+		t.Fatalf("register session should've succeeded: %v", err)
+	}
+	if request.RequestID != session.ID() {
+		t.Fatalf("session id doesn't match: %v != %v", request.RequestID, session.ID())
+	}
+
+	// We shouldn't be able to register another session with the same request id
+	_, err = manager.RegisterSession(&request, &noopEyeball{})
+	if !errors.Is(err, v3.ErrSessionBoundToOtherConn) {
+		t.Fatalf("session should not be able to be registered again: %v", err)
+	}
+
+	// Get session
+	sessionGet, err := manager.GetSession(request.RequestID)
+	if err != nil {
+		t.Fatalf("get session failed: %v", err)
+	}
+	if session.ID() != sessionGet.ID() {
+		t.Fatalf("session's do not match: %v != %v", session.ID(), sessionGet.ID())
+	}
+
+	// Remove the session
+	manager.UnregisterSession(request.RequestID)
+
+	// Get session should fail
+	_, err = manager.GetSession(request.RequestID)
+	if !errors.Is(err, v3.ErrSessionNotFound) {
+		t.Fatalf("get session failed: %v", err)
+	}
+
+	// Closing the original session should return that the socket is already closed (by the session unregistration)
+	err = session.Close()
+	if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
+		t.Fatalf("session should've closed without issue: %v", err)
+	}
+}
+
+func TestGetSession_Empty(t *testing.T) {
+	log := zerolog.Nop()
+	manager := v3.NewSessionManager(&log, ingress.DialUDPAddrPort)
+
+	_, err := manager.GetSession(testRequestID)
+	if !errors.Is(err, v3.ErrSessionNotFound) {
+		t.Fatalf("get session find no session: %v", err)
+	}
+}

quic/v3/muxer.go

@@ -0,0 +1,227 @@
+package v3
+
+import (
+	"context"
+	"errors"
+
+	"github.com/rs/zerolog"
+)
+
+const (
+	// Allocating a 16 channel buffer here allows for the writer to be slightly faster than the reader.
+	// This has worked previously well for datagramv2, so we will start with this as well
+	demuxChanCapacity = 16
+)
+
+// DatagramConn is the bridge that multiplexes writes and reads of datagrams for UDP sessions and ICMP packets to
+// a connection.
+type DatagramConn interface {
+	DatagramWriter
+	// Serve provides a server interface to process and handle incoming QUIC datagrams and demux their datagram v3 payloads.
+	Serve(context.Context) error
+}
+
+// DatagramWriter provides the Muxer interface to create proper Datagrams when sending over a connection.
+type DatagramWriter interface {
+	SendUDPSessionDatagram(datagram []byte) error
+	SendUDPSessionResponse(id RequestID, resp SessionRegistrationResp) error
+	//SendICMPPacket(packet packet.IP) error
+}
+
+// QuicConnection provides an interface that matches [quic.Connection] for only the datagram operations.
+//
+// We currently rely on the mutex for the [quic.Connection.SendDatagram] and [quic.Connection.ReceiveDatagram] and
+// do not have any locking for them. If the implementation in quic-go were to ever change, we would need to make
+// sure that we lock properly on these operations.
+type QuicConnection interface {
+	Context() context.Context
+	SendDatagram(payload []byte) error
+	ReceiveDatagram(context.Context) ([]byte, error)
+}
+
+type datagramConn struct {
+	conn           QuicConnection
+	sessionManager SessionManager
+	logger         *zerolog.Logger
+
+	datagrams  chan []byte
+	readErrors chan error
+}
+
+func NewDatagramConn(conn QuicConnection, sessionManager SessionManager, logger *zerolog.Logger) DatagramConn {
+	log := logger.With().Uint8("datagramVersion", 3).Logger()
+	return &datagramConn{
+		conn:           conn,
+		sessionManager: sessionManager,
+		logger:         &log,
+		datagrams:      make(chan []byte, demuxChanCapacity),
+		readErrors:     make(chan error, 2),
+	}
+}
+
+func (c *datagramConn) SendUDPSessionDatagram(datagram []byte) error {
+	return c.conn.SendDatagram(datagram)
+}
+
+func (c *datagramConn) SendUDPSessionResponse(id RequestID, resp SessionRegistrationResp) error {
+	datagram := UDPSessionRegistrationResponseDatagram{
+		RequestID:    id,
+		ResponseType: resp,
+	}
+	data, err := datagram.MarshalBinary()
+	if err != nil {
+		return err
+	}
+	return c.conn.SendDatagram(data)
+}
+
+var errReadTimeout error = errors.New("receive datagram timeout")
+
+// pollDatagrams will read datagrams from the underlying connection until the provided context is done.
+func (c *datagramConn) pollDatagrams(ctx context.Context) {
+	for ctx.Err() == nil {
+		datagram, err := c.conn.ReceiveDatagram(ctx)
+		// If the read returns an error, we want to return the failure to the channel.
+		if err != nil {
+			c.readErrors <- err
+			return
+		}
+		c.datagrams <- datagram
+	}
+	if ctx.Err() != nil {
+		c.readErrors <- ctx.Err()
+	}
+}
+
+// Serve will begin the process of receiving datagrams from the [quic.Connection] and demuxing them to their destination.
+// The [DatagramConn] when serving, will be responsible for the sessions it accepts.
+func (c *datagramConn) Serve(ctx context.Context) error {
+	connCtx := c.conn.Context()
+	// We want to make sure that we cancel the reader context if the Serve method returns. This could also mean that the
+	// underlying connection is also closing, but that is handled outside of the context of the datagram muxer.
+	readCtx, cancel := context.WithCancel(connCtx)
+	defer cancel()
+	go c.pollDatagrams(readCtx)
+	for {
+		// We make sure to monitor the context of cloudflared and the underlying connection to return if any errors occur.
+		var datagram []byte
+		select {
+		// Monitor the context of cloudflared
+		case <-ctx.Done():
+			return ctx.Err()
+		// Monitor the context of the underlying connection
+		case <-connCtx.Done():
+			return connCtx.Err()
+		// Monitor for any hard errors from reading the connection
+		case err := <-c.readErrors:
+			return err
+		// Otherwise, wait and dequeue datagrams as they come in
+		case d := <-c.datagrams:
+			datagram = d
+		}
+
+		// Each incoming datagram will be processed in a new go routine to handle the demuxing and action associated.
+		go func() {
+			typ, err := ParseDatagramType(datagram)
+			if err != nil {
+				c.logger.Err(err).Msgf("unable to parse datagram type: %d", typ)
+				return
+			}
+			switch typ {
+			case UDPSessionRegistrationType:
+				reg := &UDPSessionRegistrationDatagram{}
+				err := reg.UnmarshalBinary(datagram)
+				if err != nil {
+					c.logger.Err(err).Msgf("unable to unmarshal session registration datagram")
+					return
+				}
+				// We bind the new session to the quic connection context instead of cloudflared context to allow for the
+				// quic connection to close and close only the sessions bound to it. Closing of cloudflared will also
+				// initiate the close of the quic connection, so we don't have to worry about the application context
+				// in the scope of a session.
+				c.handleSessionRegistrationDatagram(connCtx, reg)
+			case UDPSessionPayloadType:
+				payload := &UDPSessionPayloadDatagram{}
+				err := payload.UnmarshalBinary(datagram)
+				if err != nil {
+					c.logger.Err(err).Msgf("unable to unmarshal session payload datagram")
+					return
+				}
+				c.handleSessionPayloadDatagram(payload)
+			case UDPSessionRegistrationResponseType:
+				// cloudflared should never expect to receive UDP session responses as it will not initiate new
+				// sessions towards the edge.
+				c.logger.Error().Msgf("unexpected datagram type received: %d", UDPSessionRegistrationResponseType)
+				return
+			default:
+				c.logger.Error().Msgf("unknown datagram type received: %d", typ)
+			}
+		}()
+	}
+}
+
+// This method handles new registrations of a session and the serve loop for the session.
+func (c *datagramConn) handleSessionRegistrationDatagram(ctx context.Context, datagram *UDPSessionRegistrationDatagram) {
+	session, err := c.sessionManager.RegisterSession(datagram, c)
+	if err != nil {
+		c.logger.Err(err).Msgf("session registration failure")
+		c.handleSessionRegistrationFailure(datagram.RequestID, err)
+		return
+	}
+	// Make sure to eventually remove the session from the session manager when the session is closed
+	defer c.sessionManager.UnregisterSession(session.ID())
+
+	// Respond that we are able to process the new session
+	err = c.SendUDPSessionResponse(datagram.RequestID, ResponseOk)
+	if err != nil {
+		c.logger.Err(err).Msgf("session registration failure: unable to send session registration response")
+		return
+	}
+
+	// We bind the context of the session to the [quic.Connection] that initiated the session.
+	// [Session.Serve] is blocking and will continue this go routine till the end of the session lifetime.
+	err = session.Serve(ctx)
+	if err == nil {
+		// We typically don't expect a session to close without some error response. [SessionIdleErr] is the typical
+		// expected error response.
+		c.logger.Warn().Msg("session was closed without explicit close or timeout")
+		return
+	}
+	// SessionIdleErr and SessionCloseErr are valid and successful error responses to end a session.
+	if errors.Is(err, SessionIdleErr{}) || errors.Is(err, SessionCloseErr) {
+		c.logger.Debug().Msg(err.Error())
+		return
+	}
+
+	// All other errors should be reported as errors
+	c.logger.Err(err).Msgf("session was closed with an error")
+}
+
+func (c *datagramConn) handleSessionRegistrationFailure(requestID RequestID, regErr error) {
+	var errResp SessionRegistrationResp
+	switch regErr {
+	case ErrSessionBoundToOtherConn:
+		errResp = ResponseSessionAlreadyConnected
+	default:
+		errResp = ResponseUnableToBindSocket
+	}
+	err := c.SendUDPSessionResponse(requestID, errResp)
+	if err != nil {
+		c.logger.Err(err).Msgf("unable to send session registration error response (%d)", errResp)
+	}
+}
+
+// Handles incoming datagrams that need to be sent to a registered session.
+func (c *datagramConn) handleSessionPayloadDatagram(datagram *UDPSessionPayloadDatagram) {
+	s, err := c.sessionManager.GetSession(datagram.RequestID)
+	if err != nil {
+		c.logger.Err(err).Msgf("unable to find session")
+		return
+	}
+	// We ignore the bytes written to the socket because any partial write must return an error.
+	_, err = s.Write(datagram.Payload)
+	if err != nil {
+		c.logger.Err(err).Msgf("unable to write payload for unavailable session")
+		return
+	}
+}

quic/v3/muxer_test.go

@@ -0,0 +1,497 @@
+package v3_test
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"net"
+	"net/netip"
+	"slices"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/ingress"
+	v3 "github.com/cloudflare/cloudflared/quic/v3"
+)
+
+type noopEyeball struct{}
+
+func (noopEyeball) SendUDPSessionDatagram(datagram []byte) error {
+	return nil
+}
+
+func (noopEyeball) SendUDPSessionResponse(id v3.RequestID, resp v3.SessionRegistrationResp) error {
+	return nil
+}
+
+type mockEyeball struct {
+	// datagram sent via SendUDPSessionDatagram
+	recvData chan []byte
+	// responses sent via SendUDPSessionResponse
+	recvResp chan struct {
+		id   v3.RequestID
+		resp v3.SessionRegistrationResp
+	}
+}
+
+func newMockEyeball() mockEyeball {
+	return mockEyeball{
+		recvData: make(chan []byte, 1),
+		recvResp: make(chan struct {
+			id   v3.RequestID
+			resp v3.SessionRegistrationResp
+		}, 1),
+	}
+}
+
+func (m *mockEyeball) SendUDPSessionDatagram(datagram []byte) error {
+	b := make([]byte, len(datagram))
+	copy(b, datagram)
+	m.recvData <- b
+	return nil
+}
+
+func (m *mockEyeball) SendUDPSessionResponse(id v3.RequestID, resp v3.SessionRegistrationResp) error {
+	m.recvResp <- struct {
+		id   v3.RequestID
+		resp v3.SessionRegistrationResp
+	}{
+		id, resp,
+	}
+	return nil
+}
+
+func TestDatagramConn_New(t *testing.T) {
+	log := zerolog.Nop()
+	conn := v3.NewDatagramConn(newMockQuicConn(), v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+	if conn == nil {
+		t.Fatal("expected valid connection")
+	}
+}
+
+func TestDatagramConn_SendUDPSessionDatagram(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+
+	payload := []byte{0xef, 0xef}
+	conn.SendUDPSessionDatagram(payload)
+	p := <-quic.recv
+	if !slices.Equal(p, payload) {
+		t.Fatal("datagram sent does not match datagram received on quic side")
+	}
+}
+
+func TestDatagramConn_SendUDPSessionResponse(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+
+	conn.SendUDPSessionResponse(testRequestID, v3.ResponseDestinationUnreachable)
+	resp := <-quic.recv
+	var response v3.UDPSessionRegistrationResponseDatagram
+	err := response.UnmarshalBinary(resp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	expected := v3.UDPSessionRegistrationResponseDatagram{
+		RequestID:    testRequestID,
+		ResponseType: v3.ResponseDestinationUnreachable,
+	}
+	if response != expected {
+		t.Fatal("datagram response sent does not match expected datagram response received")
+	}
+}
+
+func TestDatagramConnServe_ApplicationClosed(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	err := conn.Serve(ctx)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe_ConnectionClosed(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+	quic.ctx = ctx
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+
+	err := conn.Serve(context.Background())
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe_ReceiveDatagramError(t *testing.T) {
+	log := zerolog.Nop()
+	quic := &mockQuicConnReadError{err: net.ErrClosed}
+	conn := v3.NewDatagramConn(quic, v3.NewSessionManager(&log, ingress.DialUDPAddrPort), &log)
+
+	err := conn.Serve(context.Background())
+	if !errors.Is(err, net.ErrClosed) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe_ErrorDatagramTypes(t *testing.T) {
+	for _, test := range []struct {
+		name     string
+		input    []byte
+		expected string
+	}{
+		{
+			"empty",
+			[]byte{},
+			"{\"level\":\"error\",\"datagramVersion\":3,\"error\":\"datagram should have at least 1 byte\",\"message\":\"unable to parse datagram type: 0\"}\n",
+		},
+		{
+			"unexpected",
+			[]byte{byte(v3.UDPSessionRegistrationResponseType)},
+			"{\"level\":\"error\",\"datagramVersion\":3,\"message\":\"unexpected datagram type received: 3\"}\n",
+		},
+		{
+			"unknown",
+			[]byte{99},
+			"{\"level\":\"error\",\"datagramVersion\":3,\"message\":\"unknown datagram type received: 99\"}\n",
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			logOutput := new(LockedBuffer)
+			log := zerolog.New(logOutput)
+			quic := newMockQuicConn()
+			quic.send <- test.input
+			conn := v3.NewDatagramConn(quic, &mockSessionManager{}, &log)
+
+			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+			defer cancel()
+			err := conn.Serve(ctx)
+			// we cancel the Serve method to check to see if the log output was written since the unsupported datagram
+			// is dropped with only a log message as a side-effect.
+			if !errors.Is(err, context.DeadlineExceeded) {
+				t.Fatal(err)
+			}
+
+			out := logOutput.String()
+			if out != test.expected {
+				t.Fatalf("incorrect log output expected: %s", out)
+			}
+		})
+	}
+}
+
+type LockedBuffer struct {
+	bytes.Buffer
+	l sync.Mutex
+}
+
+func (b *LockedBuffer) Write(p []byte) (n int, err error) {
+	b.l.Lock()
+	defer b.l.Unlock()
+	return b.Buffer.Write(p)
+}
+
+func (b *LockedBuffer) String() string {
+	b.l.Lock()
+	defer b.l.Unlock()
+	return b.Buffer.String()
+}
+
+func TestDatagramConnServe_RegisterSession_SessionManagerError(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	expectedErr := errors.New("unable to register session")
+	sessionManager := mockSessionManager{expectedRegErr: expectedErr}
+	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	// Send new session registration
+	datagram := newRegisterSessionDatagram(testRequestID)
+	quic.send <- datagram
+
+	// Wait for session registration response with failure
+	datagram = <-quic.recv
+	var resp v3.UDPSessionRegistrationResponseDatagram
+	err := resp.UnmarshalBinary(datagram)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp.RequestID != testRequestID && resp.ResponseType != v3.ResponseUnableToBindSocket {
+		t.Fatalf("expected registration response failure")
+	}
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	cancel(expectedContextCanceled)
+	err = <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	session := newMockSession()
+	sessionManager := mockSessionManager{session: &session}
+	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	// Send new session registration
+	datagram := newRegisterSessionDatagram(testRequestID)
+	quic.send <- datagram
+
+	// Wait for session registration response with success
+	datagram = <-quic.recv
+	var resp v3.UDPSessionRegistrationResponseDatagram
+	err := resp.UnmarshalBinary(datagram)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if resp.RequestID != testRequestID && resp.ResponseType != v3.ResponseOk {
+		t.Fatalf("expected registration response ok")
+	}
+
+	// We expect the session to be served
+	timer := time.NewTimer(15 * time.Second)
+	defer timer.Stop()
+	select {
+	case <-session.served:
+		break
+	case <-timer.C:
+		t.Fatalf("expected session serve to be called")
+	}
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	cancel(expectedContextCanceled)
+	err = <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe_Payload_GetSessionError(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	sessionManager := mockSessionManager{session: nil, expectedGetErr: v3.ErrSessionNotFound}
+	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	// Send new session registration
+	datagram := newSessionPayloadDatagram(testRequestID, []byte{0xef, 0xef})
+	quic.send <- datagram
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	cancel(expectedContextCanceled)
+	err := <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
+func TestDatagramConnServe_Payload(t *testing.T) {
+	log := zerolog.Nop()
+	quic := newMockQuicConn()
+	session := newMockSession()
+	sessionManager := mockSessionManager{session: &session}
+	conn := v3.NewDatagramConn(quic, &sessionManager, &log)
+
+	// Setup the muxer
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(errors.New("other error"))
+	done := make(chan error, 1)
+	go func() {
+		done <- conn.Serve(ctx)
+	}()
+
+	// Send new session registration
+	expectedPayload := []byte{0xef, 0xef}
+	datagram := newSessionPayloadDatagram(testRequestID, expectedPayload)
+	quic.send <- datagram
+
+	// Session should receive the payload
+	payload := <-session.recv
+	if !slices.Equal(expectedPayload, payload) {
+		t.Fatalf("expected session receieve the payload sent via the muxer")
+	}
+
+	// Cancel the muxer Serve context and make sure it closes with the expected error
+	cancel(expectedContextCanceled)
+	err := <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
+func newRegisterSessionDatagram(id v3.RequestID) []byte {
+	datagram := v3.UDPSessionRegistrationDatagram{
+		RequestID:        id,
+		Dest:             netip.MustParseAddrPort("127.0.0.1:8080"),
+		IdleDurationHint: 5 * time.Second,
+	}
+	payload, err := datagram.MarshalBinary()
+	if err != nil {
+		panic(err)
+	}
+	return payload
+}
+
+func newRegisterResponseSessionDatagram(id v3.RequestID, resp v3.SessionRegistrationResp) []byte {
+	datagram := v3.UDPSessionRegistrationResponseDatagram{
+		RequestID:    id,
+		ResponseType: resp,
+	}
+	payload, err := datagram.MarshalBinary()
+	if err != nil {
+		panic(err)
+	}
+	return payload
+}
+
+func newSessionPayloadDatagram(id v3.RequestID, payload []byte) []byte {
+	datagram := make([]byte, len(payload)+17)
+	err := v3.MarshalPayloadHeaderTo(id, datagram[:])
+	if err != nil {
+		panic(err)
+	}
+	copy(datagram[17:], payload)
+	return datagram
+}
+
+type mockQuicConn struct {
+	ctx  context.Context
+	send chan []byte
+	recv chan []byte
+}
+
+func newMockQuicConn() *mockQuicConn {
+	return &mockQuicConn{
+		ctx:  context.Background(),
+		send: make(chan []byte, 1),
+		recv: make(chan []byte, 1),
+	}
+}
+
+func (m *mockQuicConn) Context() context.Context {
+	return m.ctx
+}
+
+func (m *mockQuicConn) SendDatagram(payload []byte) error {
+	b := make([]byte, len(payload))
+	copy(b, payload)
+	m.recv <- b
+	return nil
+}
+
+func (m *mockQuicConn) ReceiveDatagram(_ context.Context) ([]byte, error) {
+	return <-m.send, nil
+}
+
+type mockQuicConnReadError struct {
+	err error
+}
+
+func (m *mockQuicConnReadError) Context() context.Context {
+	return context.Background()
+}
+
+func (m *mockQuicConnReadError) SendDatagram(payload []byte) error {
+	return nil
+}
+
+func (m *mockQuicConnReadError) ReceiveDatagram(_ context.Context) ([]byte, error) {
+	return nil, m.err
+}
+
+type mockSessionManager struct {
+	session v3.Session
+
+	expectedRegErr error
+	expectedGetErr error
+}
+
+func (m *mockSessionManager) RegisterSession(request *v3.UDPSessionRegistrationDatagram, conn v3.DatagramWriter) (v3.Session, error) {
+	return m.session, m.expectedRegErr
+}
+
+func (m *mockSessionManager) GetSession(requestID v3.RequestID) (v3.Session, error) {
+	return m.session, m.expectedGetErr
+}
+
+func (m *mockSessionManager) UnregisterSession(requestID v3.RequestID) {}
+
+type mockSession struct {
+	served chan struct{}
+	recv   chan []byte
+}
+
+func newMockSession() mockSession {
+	return mockSession{
+		served: make(chan struct{}),
+		recv:   make(chan []byte, 1),
+	}
+}
+
+func (m *mockSession) ID() v3.RequestID {
+	return testRequestID
+}
+
+func (m *mockSession) Serve(ctx context.Context) error {
+	close(m.served)
+	return v3.SessionCloseErr
+}
+
+func (m *mockSession) Write(payload []byte) (n int, err error) {
+	b := make([]byte, len(payload))
+	copy(b, payload)
+	m.recv <- b
+	return len(b), nil
+}
+
+func (m *mockSession) Close() error {
+	return nil
+}

quic/v3/request.go

@@ -3,6 +3,7 @@ package v3
 import (
 	"encoding/binary"
 	"errors"
+	"fmt"
 )
 
 const (
@@ -37,6 +38,10 @@ func RequestIDFromSlice(data []byte) (RequestID, error) {
 	}, nil
 }
 
+func (id RequestID) String() string {
+	return fmt.Sprintf("%016x%016x", id.hi, id.lo)
+}
+
 // Compare returns an integer comparing two IPs.
 // The result will be 0 if id == id2, -1 if id < id2, and +1 if id > id2.
 // The definition of "less than" is the same as the [RequestID.Less] method.
@@ -70,3 +75,15 @@ func (id RequestID) MarshalBinaryTo(data []byte) error {
 	binary.BigEndian.PutUint64(data[8:], id.lo)
 	return nil
 }
+
+func (id *RequestID) UnmarshalBinary(data []byte) error {
+	if len(data) != 16 {
+		return fmt.Errorf("invalid length slice provided to unmarshal: %d (expected 16)", len(data))
+	}
+
+	*id = RequestID{
+		binary.BigEndian.Uint64(data[:8]),
+		binary.BigEndian.Uint64(data[8:]),
+	}
+	return nil
+}

quic/v3/request_test.go

@@ -5,6 +5,8 @@ import (
 	"slices"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	v3 "github.com/cloudflare/cloudflared/quic/v3"
 )
 
@@ -48,3 +50,15 @@ func TestRequestIDParsing(t *testing.T) {
 		t.Fatalf("buf1 != buf2: %+v %+v", buf1, buf2)
 	}
 }
+
+func TestRequestID_MarshalBinary(t *testing.T) {
+	buf := make([]byte, 16)
+	err := testRequestID.MarshalBinaryTo(buf)
+	require.NoError(t, err)
+	require.Len(t, buf, 16)
+
+	parsed := v3.RequestID{}
+	err = parsed.UnmarshalBinary(buf)
+	require.NoError(t, err)
+	require.Equal(t, testRequestID, parsed)
+}

quic/v3/session.go

@@ -0,0 +1,192 @@
+package v3
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/rs/zerolog"
+)
+
+const (
+	// A default is provided in the case that the client does not provide a close idle timeout.
+	defaultCloseIdleAfter = 210 * time.Second
+
+	// The maximum payload from the origin that we will be able to read. However, even though we will
+	// read 1500 bytes from the origin, we limit the amount of bytes to be proxied to less than
+	// this value (maxDatagramPayloadLen).
+	maxOriginUDPPacketSize = 1500
+)
+
+// SessionCloseErr indicates that the session's Close method was called.
+var SessionCloseErr error = errors.New("session was closed")
+
+// SessionIdleErr is returned when the session was closed because there was no communication
+// in either direction over the session for the timeout period.
+type SessionIdleErr struct {
+	timeout time.Duration
+}
+
+func (e SessionIdleErr) Error() string {
+	return fmt.Sprintf("session idle for %v", e.timeout)
+}
+
+func (e SessionIdleErr) Is(target error) bool {
+	_, ok := target.(SessionIdleErr)
+	return ok
+}
+
+func newSessionIdleErr(timeout time.Duration) error {
+	return SessionIdleErr{timeout}
+}
+
+type Session interface {
+	io.WriteCloser
+	ID() RequestID
+	// Serve starts the event loop for processing UDP packets
+	Serve(ctx context.Context) error
+}
+
+type session struct {
+	id             RequestID
+	closeAfterIdle time.Duration
+	origin         io.ReadWriteCloser
+	eyeball        DatagramWriter
+	// activeAtChan is used to communicate the last read/write time
+	activeAtChan chan time.Time
+	closeChan    chan error
+	log          *zerolog.Logger
+}
+
+func NewSession(id RequestID, closeAfterIdle time.Duration, origin io.ReadWriteCloser, eyeball DatagramWriter, log *zerolog.Logger) Session {
+	return &session{
+		id:             id,
+		closeAfterIdle: closeAfterIdle,
+		origin:         origin,
+		eyeball:        eyeball,
+		// activeAtChan has low capacity. It can be full when there are many concurrent read/write. markActive() will
+		// drop instead of blocking because last active time only needs to be an approximation
+		activeAtChan: make(chan time.Time, 1),
+		closeChan:    make(chan error, 1),
+		log:          log,
+	}
+}
+
+func (s *session) ID() RequestID {
+	return s.id
+}
+
+func (s *session) Serve(ctx context.Context) error {
+	go func() {
+		// QUIC implementation copies data to another buffer before returning https://github.com/quic-go/quic-go/blob/v0.24.0/session.go#L1967-L1975
+		// This makes it safe to share readBuffer between iterations
+		readBuffer := [maxOriginUDPPacketSize + DatagramPayloadHeaderLen]byte{}
+		// To perform a zero copy write when passing the datagram to the connection, we prepare the buffer with
+		// the required datagram header information. We can reuse this buffer for this session since the header is the
+		// same for the each read.
+		MarshalPayloadHeaderTo(s.id, readBuffer[:DatagramPayloadHeaderLen])
+		for {
+			// Read from the origin UDP socket
+			n, err := s.origin.Read(readBuffer[DatagramPayloadHeaderLen:])
+			if errors.Is(err, net.ErrClosed) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
+				s.log.Debug().Msg("Session (origin) connection closed")
+			}
+			if err != nil {
+				s.closeChan <- err
+				return
+			}
+			if n < 0 {
+				s.log.Warn().Int("packetSize", n).Msg("Session (origin) packet read was negative and was dropped")
+				continue
+			}
+			if n > maxDatagramPayloadLen {
+				s.log.Error().Int("packetSize", n).Msg("Session (origin) packet read was too large and was dropped")
+				continue
+			}
+			// Sending a packet to the session does block on the [quic.Connection], however, this is okay because it
+			// will cause back-pressure to the kernel buffer if the writes are not fast enough to the edge.
+			err = s.eyeball.SendUDPSessionDatagram(readBuffer[:DatagramPayloadHeaderLen+n])
+			if err != nil {
+				s.closeChan <- err
+				return
+			}
+			// Mark the session as active since we proxied a valid packet from the origin.
+			s.markActive()
+		}
+	}()
+	return s.waitForCloseCondition(ctx, s.closeAfterIdle)
+}
+
+func (s *session) Write(payload []byte) (n int, err error) {
+	n, err = s.origin.Write(payload)
+	if err != nil {
+		s.log.Err(err).Msg("Failed to write payload to session (remote)")
+		return n, err
+	}
+	// Write must return a non-nil error if it returns n < len(p). https://pkg.go.dev/io#Writer
+	if n < len(payload) {
+		s.log.Err(io.ErrShortWrite).Msg("Failed to write the full payload to session (remote)")
+		return n, io.ErrShortWrite
+	}
+	// Mark the session as active since we proxied a packet to the origin.
+	s.markActive()
+	return n, err
+}
+
+// Sends the last active time to the idle checker loop without blocking. activeAtChan will only be full when there
+// are many concurrent read/write. It is fine to lose some precision
+func (s *session) markActive() {
+	select {
+	case s.activeAtChan <- time.Now():
+	default:
+	}
+}
+
+func (s *session) Close() error {
+	// Make sure that we only close the origin connection once
+	return sync.OnceValue(func() error {
+		// We don't want to block on sending to the close channel if it is already full
+		select {
+		case s.closeChan <- SessionCloseErr:
+		default:
+		}
+		return s.origin.Close()
+	})()
+}
+
+func (s *session) waitForCloseCondition(ctx context.Context, closeAfterIdle time.Duration) error {
+	// Closing the session at the end cancels read so Serve() can return
+	defer s.Close()
+	if closeAfterIdle == 0 {
+		// provide deafult is caller doesn't specify one
+		closeAfterIdle = defaultCloseIdleAfter
+	}
+
+	checkIdleTimer := time.NewTimer(closeAfterIdle)
+	defer checkIdleTimer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case reason := <-s.closeChan:
+			return reason
+		case <-checkIdleTimer.C:
+			// The check idle timer will only return after an idle period since the last active
+			// operation (read or write).
+			return newSessionIdleErr(closeAfterIdle)
+		case <-s.activeAtChan:
+			// The session is still active, we want to reset the timer. First we have to stop the timer, drain the
+			// current value and then reset. It's okay if we lose some time on this operation as we don't need to
+			// close an idle session directly on-time.
+			if !checkIdleTimer.Stop() {
+				<-checkIdleTimer.C
+			}
+			checkIdleTimer.Reset(closeAfterIdle)
+		}
+	}
+}

quic/v3/session_fuzz_test.go

@@ -0,0 +1,23 @@
+package v3_test
+
+import (
+	"testing"
+)
+
+// FuzzSessionWrite verifies that we don't run into any panics when writing variable sized payloads to the origin.
+func FuzzSessionWrite(f *testing.F) {
+	f.Fuzz(func(t *testing.T, b []byte) {
+		testSessionWrite(t, b)
+	})
+}
+
+// FuzzSessionServe verifies that we don't run into any panics when reading variable sized payloads from the origin.
+func FuzzSessionServe(f *testing.F) {
+	f.Fuzz(func(t *testing.T, b []byte) {
+		// The origin transport read is bound to 1280 bytes
+		if len(b) > 1280 {
+			b = b[:1280]
+		}
+		testSessionServe_Origin(t, b)
+	})
+}

quic/v3/session_test.go

@@ -0,0 +1,283 @@
+package v3_test
+
+import (
+	"context"
+	"errors"
+	"net"
+	"slices"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	v3 "github.com/cloudflare/cloudflared/quic/v3"
+)
+
+var expectedContextCanceled = errors.New("expected context canceled")
+
+func TestSessionNew(t *testing.T) {
+	log := zerolog.Nop()
+	session := v3.NewSession(testRequestID, 5*time.Second, nil, &noopEyeball{}, &log)
+	if testRequestID != session.ID() {
+		t.Fatalf("session id doesn't match: %s != %s", testRequestID, session.ID())
+	}
+}
+
+func testSessionWrite(t *testing.T, payload []byte) {
+	log := zerolog.Nop()
+	origin := newTestOrigin(makePayload(1280))
+	session := v3.NewSession(testRequestID, 5*time.Second, &origin, &noopEyeball{}, &log)
+	n, err := session.Write(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != len(payload) {
+		t.Fatal("unable to write the whole payload")
+	}
+	if !slices.Equal(payload, origin.write[:len(payload)]) {
+		t.Fatal("payload provided from origin and read value are not the same")
+	}
+}
+
+func TestSessionWrite_Max(t *testing.T) {
+	payload := makePayload(1280)
+	testSessionWrite(t, payload)
+}
+
+func TestSessionWrite_Min(t *testing.T) {
+	payload := makePayload(0)
+	testSessionWrite(t, payload)
+}
+
+func TestSessionServe_OriginMax(t *testing.T) {
+	payload := makePayload(1280)
+	testSessionServe_Origin(t, payload)
+}
+
+func TestSessionServe_OriginMin(t *testing.T) {
+	payload := makePayload(0)
+	testSessionServe_Origin(t, payload)
+}
+
+func testSessionServe_Origin(t *testing.T, payload []byte) {
+	log := zerolog.Nop()
+	eyeball := newMockEyeball()
+	origin := newTestOrigin(payload)
+	session := v3.NewSession(testRequestID, 3*time.Second, &origin, &eyeball, &log)
+	defer session.Close()
+
+	ctx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(context.Canceled)
+	done := make(chan error)
+	go func() {
+		done <- session.Serve(ctx)
+	}()
+
+	select {
+	case data := <-eyeball.recvData:
+		// check received data matches provided from origin
+		expectedData := makePayload(1500)
+		v3.MarshalPayloadHeaderTo(testRequestID, expectedData[:])
+		copy(expectedData[17:], payload)
+		if !slices.Equal(expectedData[:17+len(payload)], data) {
+			t.Fatal("expected datagram did not equal expected")
+		}
+		cancel(expectedContextCanceled)
+	case err := <-ctx.Done():
+		// we expect the payload to return before the context to cancel on the session
+		t.Fatal(err)
+	}
+
+	err := <-done
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+	if !errors.Is(context.Cause(ctx), expectedContextCanceled) {
+		t.Fatal(err)
+	}
+}
+
+func TestSessionServe_OriginTooLarge(t *testing.T) {
+	log := zerolog.Nop()
+	eyeball := newMockEyeball()
+	payload := makePayload(1281)
+	origin := newTestOrigin(payload)
+	session := v3.NewSession(testRequestID, 2*time.Second, &origin, &eyeball, &log)
+	defer session.Close()
+
+	done := make(chan error)
+	go func() {
+		done <- session.Serve(context.Background())
+	}()
+
+	select {
+	case data := <-eyeball.recvData:
+		// we never expect a read to make it here because the origin provided a payload that is too large
+		// for cloudflared to proxy and it will drop it.
+		t.Fatalf("we should never proxy a payload of this size: %d", len(data))
+	case err := <-done:
+		if !errors.Is(err, v3.SessionIdleErr{}) {
+			t.Error(err)
+		}
+	}
+}
+
+func TestSessionClose_Multiple(t *testing.T) {
+	log := zerolog.Nop()
+	origin := newTestOrigin(makePayload(128))
+	session := v3.NewSession(testRequestID, 5*time.Second, &origin, &noopEyeball{}, &log)
+	err := session.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !origin.closed.Load() {
+		t.Fatal("origin wasn't closed")
+	}
+	// subsequent closes shouldn't call close again or cause any errors
+	err = session.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSessionServe_IdleTimeout(t *testing.T) {
+	log := zerolog.Nop()
+	origin := newTestIdleOrigin(10 * time.Second) // Make idle time longer than closeAfterIdle
+	closeAfterIdle := 2 * time.Second
+	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, &noopEyeball{}, &log)
+	err := session.Serve(context.Background())
+	if !errors.Is(err, v3.SessionIdleErr{}) {
+		t.Fatal(err)
+	}
+	// session should be closed
+	if !origin.closed {
+		t.Fatalf("session should be closed after Serve returns")
+	}
+	// closing a session again should not return an error
+	err = session.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSessionServe_ParentContextCanceled(t *testing.T) {
+	log := zerolog.Nop()
+	// Make idle time and idle timeout longer than closeAfterIdle
+	origin := newTestIdleOrigin(10 * time.Second)
+	closeAfterIdle := 10 * time.Second
+
+	session := v3.NewSession(testRequestID, closeAfterIdle, &origin, &noopEyeball{}, &log)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	err := session.Serve(ctx)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatal(err)
+	}
+	// session should be closed
+	if !origin.closed {
+		t.Fatalf("session should be closed after Serve returns")
+	}
+	// closing a session again should not return an error
+	err = session.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestSessionServe_ReadErrors(t *testing.T) {
+	log := zerolog.Nop()
+	origin := newTestErrOrigin(net.ErrClosed, nil)
+	session := v3.NewSession(testRequestID, 30*time.Second, &origin, &noopEyeball{}, &log)
+	err := session.Serve(context.Background())
+	if !errors.Is(err, net.ErrClosed) {
+		t.Fatal(err)
+	}
+}
+
+type testOrigin struct {
+	// bytes from Write
+	write []byte
+	// bytes provided to Read
+	read     []byte
+	readOnce atomic.Bool
+	closed   atomic.Bool
+}
+
+func newTestOrigin(payload []byte) testOrigin {
+	return testOrigin{
+		read: payload,
+	}
+}
+
+func (o *testOrigin) Read(p []byte) (n int, err error) {
+	if o.closed.Load() {
+		return -1, net.ErrClosed
+	}
+	if o.readOnce.Load() {
+		// We only want to provide one read so all other reads will be blocked
+		time.Sleep(10 * time.Second)
+	}
+	o.readOnce.Store(true)
+	return copy(p, o.read), nil
+}
+
+func (o *testOrigin) Write(p []byte) (n int, err error) {
+	if o.closed.Load() {
+		return -1, net.ErrClosed
+	}
+	o.write = make([]byte, len(p))
+	copy(o.write, p)
+	return len(p), nil
+}
+
+func (o *testOrigin) Close() error {
+	o.closed.Store(true)
+	return nil
+}
+
+type testIdleOrigin struct {
+	duration time.Duration
+	closed   bool
+}
+
+func newTestIdleOrigin(d time.Duration) testIdleOrigin {
+	return testIdleOrigin{
+		duration: d,
+	}
+}
+
+func (o *testIdleOrigin) Read(p []byte) (n int, err error) {
+	time.Sleep(o.duration)
+	return 0, nil
+}
+
+func (o *testIdleOrigin) Write(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (o *testIdleOrigin) Close() error {
+	o.closed = true
+	return nil
+}
+
+type testErrOrigin struct {
+	readErr  error
+	writeErr error
+}
+
+func newTestErrOrigin(readErr error, writeErr error) testErrOrigin {
+	return testErrOrigin{readErr, writeErr}
+}
+
+func (o *testErrOrigin) Read(p []byte) (n int, err error) {
+	return 0, o.readErr
+}
+
+func (o *testErrOrigin) Write(p []byte) (n int, err error) {
+	return len(p), o.writeErr
+}
+
+func (o *testErrOrigin) Close() error {
+	return nil
+}

ssh_server_tests/Dockerfile

@@ -1,14 +0,0 @@
-FROM python:3-buster
-
-RUN wget https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64.deb \
-  && dpkg -i cloudflared-linux-amd64.deb
-
-RUN pip install pexpect
-
-COPY tests.py .
-COPY ssh /root/.ssh
-RUN chmod 600 /root/.ssh/id_rsa
-
-ARG SSH_HOSTNAME
-RUN bash -c 'sed -i "s/{{hostname}}/${SSH_HOSTNAME}/g" /root/.ssh/authorized_keys_config'
-RUN bash -c 'sed -i "s/{{hostname}}/${SSH_HOSTNAME}/g" /root/.ssh/short_lived_cert_config'

ssh_server_tests/README.md

@@ -1,23 +0,0 @@
-# Cloudflared SSH server smoke tests
-
-Runs several tests in a docker container against a server that is started out of band of these tests.
-Cloudflared token also needs to be retrieved out of band.
-SSH server hostname and user need to be configured in a docker environment file
-
-
-## Running tests
-
-* Build cloudflared:
-make cloudflared
-
-* Start server:
-sudo ./cloudflared tunnel --hostname HOSTNAME --ssh-server
-
-* Fetch token: 
-./cloudflared access login HOSTNAME
-
-* Create docker env file:
-echo "SSH_HOSTNAME=HOSTNAME\nSSH_USER=USERNAME\n" > ssh_server_tests/.env
-
-* Run tests:
-make test-ssh-server

ssh_server_tests/docker-compose.yml

@@ -1,18 +0,0 @@
-version: "3.1"
-
-services:
-  ssh_test:
-    build:
-      context: .
-      args:
-        - SSH_HOSTNAME=${SSH_HOSTNAME}
-    volumes:
-      - "~/.cloudflared/:/root/.cloudflared"
-    env_file:
-     - .env
-    environment:
-      - AUTHORIZED_KEYS_SSH_CONFIG=/root/.ssh/authorized_keys_config
-      - SHORT_LIVED_CERT_SSH_CONFIG=/root/.ssh/short_lived_cert_config
-      - REMOTE_SCP_FILENAME=scp_test.txt
-      - ROOT_ONLY_TEST_FILE_PATH=~/permission_test.txt
-    entrypoint: "python tests.py"

ssh_server_tests/ssh/authorized_keys_config

@@ -1,5 +0,0 @@
-Host *
-    AddressFamily inet
-
-Host {{hostname}}
-  ProxyCommand /usr/local/bin/cloudflared access ssh --hostname %h

ssh_server_tests/ssh/id_rsa

@@ -1,49 +0,0 @@
------BEGIN OPENSSH PRIVATE KEY-----
-b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAACFwAAAAdzc2gtcn
-NhAAAAAwEAAQAAAgEAvi26NDQ8cYTTztqPe9ZgF5HR/rIo5FoDgL5NbbZKW6h0txP9Fd8s
-id9Bgmo+aGkeM327tPVVMQ6UFmdRksOCIDWQDjNLF8b6S+Fu95tvMKSbGreRoR32OvgZKV
-I6KmOsF4z4GIv9naPplZswtKEUhSSI+/gPdAs9wfwalqZ77e82QJ727bYMeC3lzuoT+KBI
-dYufJ4OQhLtpHrqhB5sn7s6+oCv/u85GSln5SIC18Hi2t9lW4tgb5tH8P0kEDDWGfPS5ok
-qGi4kFTvwBXOCS2r4dhi5hRkpP7PqG4np0OCfvK5IRRJ27fCnj0loc+puZJAxnPMbuJr64
-vwxRx78PM/V0PDUsl0P6aR/vbe0XmF9FGqbWf2Tar1p4r6C9/bMzcDz8seYT8hzLIHP3+R
-l1hdlsTLm+1EzhaExKId+tjXegKGG4nU24h6qHEnRxLQDMwEsdkfj4E1pVypZJXVyNj99D
-o84vi0EUnu7R4HmQb/C+Pu7qMDtLT3Zk7O5Mg4LQ+cTz9V0noYEAyG46nAB4U/nJzBnV1J
-+aAdpioHmUAYhLYlQ9Kiy7LCJi92g9Wqa4wxMKxBbO5ZeH++p2p2lUi/oQNqx/2dLYFmy0
-wxvJHbZIhAaFbOeCvHg1ucIAQznli2jOr2qoB+yKRRPAp/3NXnZg1v7ce2CkwiAD52wjtC
-kAAAdILMJUeyzCVHsAAAAHc3NoLXJzYQAAAgEAvi26NDQ8cYTTztqPe9ZgF5HR/rIo5FoD
-gL5NbbZKW6h0txP9Fd8sid9Bgmo+aGkeM327tPVVMQ6UFmdRksOCIDWQDjNLF8b6S+Fu95
-tvMKSbGreRoR32OvgZKVI6KmOsF4z4GIv9naPplZswtKEUhSSI+/gPdAs9wfwalqZ77e82
-QJ727bYMeC3lzuoT+KBIdYufJ4OQhLtpHrqhB5sn7s6+oCv/u85GSln5SIC18Hi2t9lW4t
-gb5tH8P0kEDDWGfPS5okqGi4kFTvwBXOCS2r4dhi5hRkpP7PqG4np0OCfvK5IRRJ27fCnj
-0loc+puZJAxnPMbuJr64vwxRx78PM/V0PDUsl0P6aR/vbe0XmF9FGqbWf2Tar1p4r6C9/b
-MzcDz8seYT8hzLIHP3+Rl1hdlsTLm+1EzhaExKId+tjXegKGG4nU24h6qHEnRxLQDMwEsd
-kfj4E1pVypZJXVyNj99Do84vi0EUnu7R4HmQb/C+Pu7qMDtLT3Zk7O5Mg4LQ+cTz9V0noY
-EAyG46nAB4U/nJzBnV1J+aAdpioHmUAYhLYlQ9Kiy7LCJi92g9Wqa4wxMKxBbO5ZeH++p2
-p2lUi/oQNqx/2dLYFmy0wxvJHbZIhAaFbOeCvHg1ucIAQznli2jOr2qoB+yKRRPAp/3NXn
-Zg1v7ce2CkwiAD52wjtCkAAAADAQABAAACAQCbnVsyAFQ9J00Rg/HIiUATyTQlzq57O9SF
-8jH1RiZOHedzLx32WaleH5rBFiJ+2RTnWUjQ57aP77fpJR2wk93UcT+w/vPBPwXsNUjRvx
-Qan3ZzRCYbyiKDWiNslmYV7X0RwD36CAK8jTVDP7t48h2SXLTiSLaMY+5i3uD6yLu7k/O2
-qNyw4jgN1rCmwQ8acD0aQec3NAZ7NcbsaBX/3Uutsup0scwOZtlJWZoLY5Z8cKpCgcsAz4
-j1NHnNZvey7dFgSffj/ktdvf7kBH0w/GnuJ4aNF0Jte70u0kiw5TZYBQVFh74tgUu6a6SJ
-qUbxIYUL5EJNjxGsDn+phHEemw3aMv0CwZG6Tqaionlna7bLsl9Bg1HTGclczVWx8uqC+M
-6agLmkhYCHG0rVj8h5smjXAQXtmvIDVYDOlJZZoF9VAOCj6QfmJUH1NAGpCs1HDHbeOxGA
-OLCh4d3F4rScPqhGdtSt4W13VFIvXn2Qqoz9ufepZsee1SZqpcerxywx2wN9ZAzu+X8lTN
-i+TA2B3vWpqqucOEsp4JwDN+VMKZqKUGUDWcm/eHSaG6wq0q734LUlgM85TjaIg8QsNtWV
-giB1nWwsYIuH4rsFNFGEwURYdGBcw6idH0GZ7I4RaIB5F9oOza1d601E0APHYrtnx9yOiK
-nOtJ+5ZmVZovaDRfu1aQAAAQBU/EFaNUzoVhO04pS2L6BlByt963bOIsSJhdlEzek5AAli
-eaf1S/PD6xWCc0IGY+GZE0HPbhsKYanjqOpWldcA2T7fzf4oz4vFBfUkPYo/MLSlLCYsDd
-IH3wBkCssnfR5EkzNgxnOvq646Nl64BMvxwSIXGPktdq9ZALxViwricSRzCFURnh5vLHWU
-wBzSgAA0UlZ9E64GtAv066+AoZCp83GhTLRC4o0naE2e/K4op4BCFHLrZ8eXmDRK3NJj80
-Vkn+uhrk+SHmbjIhmS57Pv9p8TWyRvemph/nMUuZGKBUu2X+JQxggck0KigIrXjsmciCsM
-BIM3mYDDfjYbyVhTAAABAQDkV8O1bWUsAIqk7RU+iDZojN5kaO+zUvj1TafX8QX1sY6pu4
-Z2cfSEka1532BaehM95bQm7BCPw4cYg56XidmCQTZ9WaWqxVrOo48EKXUtZMZx6nKFOKlq
-MT2XTMnGT9n7kFCfEjSVkAjuJ9ZTFLOaoXAaVRnxeHQwOKaup5KKP9GSzNIw328U+96s3V
-WKHeT4pMjHBccgW/qX/tRRidZw5in5uBC9Ew5y3UACFTkNOnhUwVfyUNbBZJ2W36msQ3KD
-AN7nOrQHqhd3NFyCEy2ovIAKVBacr/VEX6EsRUshIehJzz8EY9f3kXL7WT2QDoz2giPeBJ
-HJdEpXt43UpszjAAABAQDVNpqNdHUlCs9XnbIvc6ZRrNh79wt65YFfvh/QEuA33KnA6Ri6
-EgnV5IdUWXS/UFaYcm2udydrBpVIVifSYl3sioHBylpri23BEy38PKwVXvghUtfpN6dWGn
-NZUG25fQPtIzqi+lo953ZjIj+Adi17AeVv4P4NiLrZeM9lXfWf2pEPOecxXs1IwAf9IiDQ
-WepAwRLsu42eEnHA+DSJPZUkSbISfM5X345k0g6EVATX/yLL3CsqClPzPtsqjh6rbEfFg3
-2OfIMcWV77gOlGWGQ+bUHc8kV6xJqV9QVacLWzfLvIqHF0wQMf8WLOVHEzkfiq4VjwhVqr
-/+FFvljm5nSDAAAAEW1pa2VAQzAyWTUwVEdKR0g4AQ==
------END OPENSSH PRIVATE KEY-----

ssh_server_tests/ssh/id_rsa.pub

@@ -1 +0,0 @@
-ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAACAQC+Lbo0NDxxhNPO2o971mAXkdH+sijkWgOAvk1ttkpbqHS3E/0V3yyJ30GCaj5oaR4zfbu09VUxDpQWZ1GSw4IgNZAOM0sXxvpL4W73m28wpJsat5GhHfY6+BkpUjoqY6wXjPgYi/2do+mVmzC0oRSFJIj7+A90Cz3B/BqWpnvt7zZAnvbttgx4LeXO6hP4oEh1i58ng5CEu2keuqEHmyfuzr6gK/+7zkZKWflIgLXweLa32Vbi2Bvm0fw/SQQMNYZ89LmiSoaLiQVO/AFc4JLavh2GLmFGSk/s+obienQ4J+8rkhFEnbt8KePSWhz6m5kkDGc8xu4mvri/DFHHvw8z9XQ8NSyXQ/ppH+9t7ReYX0UaptZ/ZNqvWnivoL39szNwPPyx5hPyHMsgc/f5GXWF2WxMub7UTOFoTEoh362Nd6AoYbidTbiHqocSdHEtAMzASx2R+PgTWlXKlkldXI2P30Ojzi+LQRSe7tHgeZBv8L4+7uowO0tPdmTs7kyDgtD5xPP1XSehgQDIbjqcAHhT+cnMGdXUn5oB2mKgeZQBiEtiVD0qLLssImL3aD1aprjDEwrEFs7ll4f76nanaVSL+hA2rH/Z0tgWbLTDG8kdtkiEBoVs54K8eDW5wgBDOeWLaM6vaqgH7IpFE8Cn/c1edmDW/tx7YKTCIAPnbCO0KQ== mike@C02Y50TGJGH8

ssh_server_tests/ssh/short_lived_cert_config

@@ -1,11 +0,0 @@
-Host *
-    AddressFamily inet
-
-Host {{hostname}}
-  ProxyCommand bash -c '/usr/local/bin/cloudflared access ssh-gen --hostname %h; ssh -F /root/.ssh/short_lived_cert_config -tt %r@cfpipe-{{hostname}} >&2 <&1'
-
-Host cfpipe-{{hostname}}
-  HostName {{hostname}}
-  ProxyCommand /usr/local/bin/cloudflared access ssh --hostname %h
-  IdentityFile ~/.cloudflared/{{hostname}}-cf_key
-  CertificateFile ~/.cloudflared/{{hostname}}-cf_key-cert.pub

ssh_server_tests/tests.py

@@ -1,195 +0,0 @@
-"""
-Cloudflared Integration tests
-"""
-
-import unittest
-import subprocess
-import os
-import tempfile
-from contextlib import contextmanager
-
-from pexpect import pxssh
-
-
-class TestSSHBase(unittest.TestCase):
-    """
-    SSH test base class containing constants and helper funcs
-    """
-
-    HOSTNAME = os.environ["SSH_HOSTNAME"]
-    SSH_USER = os.environ["SSH_USER"]
-    SSH_TARGET = f"{SSH_USER}@{HOSTNAME}"
-    AUTHORIZED_KEYS_SSH_CONFIG = os.environ["AUTHORIZED_KEYS_SSH_CONFIG"]
-    SHORT_LIVED_CERT_SSH_CONFIG = os.environ["SHORT_LIVED_CERT_SSH_CONFIG"]
-    SSH_OPTIONS = {"StrictHostKeyChecking": "no"}
-
-    @classmethod
-    def get_ssh_command(cls, pty=True):
-        """
-        Return ssh command arg list. If pty is true, a PTY is forced for the session.
-        """
-        cmd = [
-            "ssh",
-            "-o",
-            "StrictHostKeyChecking=no",
-            "-F",
-            cls.AUTHORIZED_KEYS_SSH_CONFIG,
-            cls.SSH_TARGET,
-        ]
-        if not pty:
-            cmd += ["-T"]
-        else:
-            cmd += ["-tt"]
-
-        return cmd
-
-    @classmethod
-    @contextmanager
-    def ssh_session_manager(cls, *args, **kwargs):
-        """
-        Context manager for interacting with a pxssh session.
-        Disables pty echo on the remote server and ensures session is terminated afterward.
-        """
-        session = pxssh.pxssh(options=cls.SSH_OPTIONS)
-
-        session.login(
-            cls.HOSTNAME,
-            username=cls.SSH_USER,
-            original_prompt=r"[#@$]",
-            ssh_config=kwargs.get("ssh_config", cls.AUTHORIZED_KEYS_SSH_CONFIG),
-            ssh_tunnels=kwargs.get("ssh_tunnels", {}),
-        )
-        try:
-            session.sendline("stty -echo")
-            session.prompt()
-            yield session
-        finally:
-            session.logout()
-
-    @staticmethod
-    def get_command_output(session, cmd):
-        """
-        Executes command on remote ssh server and waits for prompt.
-        Returns command output
-        """
-        session.sendline(cmd)
-        session.prompt()
-        return session.before.decode().strip()
-
-    def exec_command(self, cmd, shell=False):
-        """
-        Executes command locally. Raises Assertion error for non-zero return code.
-        Returns stdout and stderr
-        """
-        proc = subprocess.Popen(
-            cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=shell
-        )
-        raw_out, raw_err = proc.communicate()
-
-        out = raw_out.decode()
-        err = raw_err.decode()
-        self.assertEqual(proc.returncode, 0, msg=f"stdout: {out} stderr: {err}")
-        return out.strip(), err.strip()
-
-
-class TestSSHCommandExec(TestSSHBase):
-    """
-    Tests inline ssh command exec
-    """
-
-    # Name of file to be downloaded over SCP on remote server.
-    REMOTE_SCP_FILENAME = os.environ["REMOTE_SCP_FILENAME"]
-
-    @classmethod
-    def get_scp_base_command(cls):
-        return [
-            "scp",
-            "-o",
-            "StrictHostKeyChecking=no",
-            "-v",
-            "-F",
-            cls.AUTHORIZED_KEYS_SSH_CONFIG,
-        ]
-
-    @unittest.skip(
-        "This creates files on the remote. Should be skipped until server is dockerized."
-    )
-    def test_verbose_scp_sink_mode(self):
-        with tempfile.NamedTemporaryFile() as fl:
-            self.exec_command(
-                self.get_scp_base_command() + [fl.name, f"{self.SSH_TARGET}:"]
-            )
-
-    def test_verbose_scp_source_mode(self):
-        with tempfile.TemporaryDirectory() as tmpdirname:
-            self.exec_command(
-                self.get_scp_base_command()
-                + [f"{self.SSH_TARGET}:{self.REMOTE_SCP_FILENAME}", tmpdirname]
-            )
-            local_filename = os.path.join(tmpdirname, self.REMOTE_SCP_FILENAME)
-
-            self.assertTrue(os.path.exists(local_filename))
-            self.assertTrue(os.path.getsize(local_filename) > 0)
-
-    def test_pty_command(self):
-        base_cmd = self.get_ssh_command()
-
-        out, _ = self.exec_command(base_cmd + ["whoami"])
-        self.assertEqual(out.strip().lower(), self.SSH_USER.lower())
-
-        out, _ = self.exec_command(base_cmd + ["tty"])
-        self.assertNotEqual(out, "not a tty")
-
-    def test_non_pty_command(self):
-        base_cmd = self.get_ssh_command(pty=False)
-
-        out, _ = self.exec_command(base_cmd + ["whoami"])
-        self.assertEqual(out.strip().lower(), self.SSH_USER.lower())
-
-        out, _ = self.exec_command(base_cmd + ["tty"])
-        self.assertEqual(out, "not a tty")
-
-
-class TestSSHShell(TestSSHBase):
-    """
-    Tests interactive SSH shell
-    """
-
-    # File path to a file on the remote server with root only read privileges.
-    ROOT_ONLY_TEST_FILE_PATH = os.environ["ROOT_ONLY_TEST_FILE_PATH"]
-
-    def test_ssh_pty(self):
-        with self.ssh_session_manager() as session:
-
-            # Test shell launched as correct user
-            username = self.get_command_output(session, "whoami")
-            self.assertEqual(username.lower(), self.SSH_USER.lower())
-
-            # Test USER env variable set
-            user_var = self.get_command_output(session, "echo $USER")
-            self.assertEqual(user_var.lower(), self.SSH_USER.lower())
-
-            # Test HOME env variable set to true user home.
-            home_env = self.get_command_output(session, "echo $HOME")
-            pwd = self.get_command_output(session, "pwd")
-            self.assertEqual(pwd, home_env)
-
-            # Test shell launched in correct user home dir.
-            self.assertIn(username, pwd)
-
-            # Ensure shell launched with correct user's permissions and privs.
-            # Can't read root owned 0700 files.
-            output = self.get_command_output(
-                session, f"cat {self.ROOT_ONLY_TEST_FILE_PATH}"
-            )
-            self.assertIn("Permission denied", output)
-
-    def test_short_lived_cert_auth(self):
-        with self.ssh_session_manager(
-            ssh_config=self.SHORT_LIVED_CERT_SSH_CONFIG
-        ) as session:
-            username = self.get_command_output(session, "whoami")
-            self.assertEqual(username.lower(), self.SSH_USER.lower())
-
-
-unittest.main()

supervisor/supervisor.go

@@ -12,7 +12,9 @@ import (
 
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/edgediscovery"
+	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/orchestration"
+	v3 "github.com/cloudflare/cloudflared/quic/v3"
 	"github.com/cloudflare/cloudflared/retry"
 	"github.com/cloudflare/cloudflared/signal"
 	"github.com/cloudflare/cloudflared/tunnelstate"
@@ -80,9 +82,12 @@ func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrato
 	edgeAddrHandler := NewIPAddrFallback(config.MaxEdgeAddrRetries)
 	edgeBindAddr := config.EdgeBindAddr
 
+	sessionManager := v3.NewSessionManager(config.Log, ingress.DialUDPAddrPort)
+
 	edgeTunnelServer := EdgeTunnelServer{
 		config:            config,
 		orchestrator:      orchestrator,
+		sessionManager:    sessionManager,
 		edgeAddrs:         edgeIPs,
 		edgeAddrHandler:   edgeAddrHandler,
 		edgeBindAddr:      edgeBindAddr,

supervisor/tunnel.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/netip"
 	"runtime/debug"
+	"slices"
 	"strings"
 	"sync"
 	"time"
@@ -24,6 +25,7 @@ import (
 	"github.com/cloudflare/cloudflared/management"
 	"github.com/cloudflare/cloudflared/orchestration"
 	quicpogs "github.com/cloudflare/cloudflared/quic"
+	v3 "github.com/cloudflare/cloudflared/quic/v3"
 	"github.com/cloudflare/cloudflared/retry"
 	"github.com/cloudflare/cloudflared/signal"
 	"github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -87,14 +89,6 @@ func (c *TunnelConfig) connectionOptions(originLocalAddr string, numPreviousAtte
 	}
 }
 
-func (c *TunnelConfig) SupportedFeatures() []string {
-	supported := []string{features.FeatureSerializedHeaders}
-	if c.NamedTunnel == nil {
-		supported = append(supported, features.FeatureQuickReconnects)
-	}
-	return supported
-}
-
 func StartTunnelDaemon(
 	ctx context.Context,
 	config *TunnelConfig,
@@ -181,6 +175,7 @@ func (f *ipAddrFallback) ShouldGetNewAddress(connIndex uint8, err error) (needsN
 type EdgeTunnelServer struct {
 	config            *TunnelConfig
 	orchestrator      *orchestration.Orchestrator
+	sessionManager    v3.SessionManager
 	edgeAddrHandler   EdgeAddrHandler
 	edgeAddrs         *edgediscovery.Edge
 	edgeBindAddr      net.IP
@@ -590,32 +585,65 @@ func (e *EdgeTunnelServer) serveQUIC(
 		InitialPacketSize:          initialPacketSize,
 	}
 
-	quicConn, err := connection.NewQUICConnection(
+	// Dial the QUIC connection to the edge
+	conn, err := connection.DialQuic(
 		ctx,
 		quicConfig,
+		tlsConfig,
 		edgeAddr,
 		e.edgeBindAddr,
 		connIndex,
-		tlsConfig,
-		e.orchestrator,
-		connOptions,
-		controlStreamHandler,
 		connLogger.Logger(),
-		e.config.PacketConfig,
-		e.config.RPCTimeout,
-		e.config.WriteStreamTimeout,
-		e.config.GracePeriod,
 	)
 	if err != nil {
-		connLogger.ConnAwareLogger().Err(err).Msgf("Failed to create new quic connection")
+		connLogger.ConnAwareLogger().Err(err).Msgf("Failed to dial a quic connection")
 		return err, true
 	}
 
+	var datagramSessionManager connection.DatagramSessionHandler
+	if slices.Contains(connOptions.Client.Features, features.FeatureDatagramV3) {
+		datagramSessionManager = connection.NewDatagramV3Connection(
+			ctx,
+			conn,
+			e.sessionManager,
+			connLogger.Logger(),
+		)
+	} else {
+		datagramSessionManager = connection.NewDatagramV2Connection(
+			ctx,
+			conn,
+			e.config.PacketConfig,
+			e.config.RPCTimeout,
+			e.config.WriteStreamTimeout,
+			connLogger.Logger(),
+		)
+	}
+
+	// Wrap the [quic.Connection] as a TunnelConnection
+	tunnelConn, err := connection.NewTunnelConnection(
+		ctx,
+		conn,
+		connIndex,
+		e.orchestrator,
+		datagramSessionManager,
+		controlStreamHandler,
+		connOptions,
+		e.config.RPCTimeout,
+		e.config.WriteStreamTimeout,
+		e.config.GracePeriod,
+		connLogger.Logger(),
+	)
+	if err != nil {
+		connLogger.ConnAwareLogger().Err(err).Msgf("Failed to create new tunnel connection")
+		return err, true
+	}
+
+	// Serve the TunnelConnection
 	errGroup, serveCtx := errgroup.WithContext(ctx)
 	errGroup.Go(func() error {
-		err := quicConn.Serve(serveCtx)
+		err := tunnelConn.Serve(serveCtx)
 		if err != nil {
-			connLogger.ConnAwareLogger().Err(err).Msg("Failed to serve quic connection")
+			connLogger.ConnAwareLogger().Err(err).Msg("Failed to serve tunnel connection")
 		}
 		return err
 	})
@@ -624,8 +652,8 @@ func (e *EdgeTunnelServer) serveQUIC(
 		err := listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
 		if err != nil {
 			// forcefully break the connection (this is only used for testing)
-			// errgroup will return context canceled for the quicConn.Serve
-			connLogger.Logger().Debug().Msg("Forcefully breaking quic connection")
+			// errgroup will return context canceled for the tunnelConn.Serve
+			connLogger.Logger().Debug().Msg("Forcefully breaking tunnel connection")
 		}
 		return err
 	})