port quic-go v0.52.0

Signed-off-by: roc <roc@imroc.cc>

Author: imroc

internal/http3/body.go

@@ -9,7 +9,7 @@ import (
 	"github.com/quic-go/quic-go"
 )
 
-// A Hijacker allows hijacking of the stream creating part of a quic.Session from a http.Response.Body.
+// A Hijacker allows hijacking of the stream creating part of a quic.Connection from a http.ResponseWriter.
 // It is used by WebTransport to create WebTransport streams after a session has been established.
 type Hijacker interface {
 	Connection() Connection

internal/http3/client.go

@@ -32,6 +32,13 @@ const (
 	defaultMaxResponseHeaderBytes = 10 * 1 << 20 // 10 MB
 )
 
+type errConnUnusable struct{ e error }
+
+func (e *errConnUnusable) Unwrap() error { return e.e }
+func (e *errConnUnusable) Error() string { return fmt.Sprintf("http3: conn unusable: %s", e.e.Error()) }
+
+const max1xxResponses = 5 // arbitrary bound on number of informational responses
+
 var defaultQuicConfig = &quic.Config{
 	MaxIncomingStreams: -1, // don't allow the server to create bidirectional streams
 	KeepAlivePeriod:    10 * time.Second,
@@ -233,7 +240,7 @@ func (c *ClientConn) roundTrip(req *http.Request) (*http.Response, error) {
 		c.maxResponseHeaderBytes,
 	)
 	if err != nil {
-		return nil, err
+		return nil, &errConnUnusable{e: err}
 	}
 
 	// Request Cancellation:
@@ -346,8 +353,7 @@ func (c *ClientConn) doRequest(req *http.Request, str *requestStream) (*http.Res
 	}
 
 	// copy from net/http: support 1xx responses
-	num1xx := 0               // number of informational 1xx headers received
-	const max1xxResponses = 5 // arbitrary bound on number of informational responses
+	var num1xx int // number of informational 1xx headers received
 
 	var res *http.Response
 	for {
@@ -363,10 +369,12 @@ func (c *ClientConn) doRequest(req *http.Request, str *requestStream) (*http.Res
 		if is1xxNonTerminal {
 			num1xx++
 			if num1xx > max1xxResponses {
-				return nil, errors.New("http: too many 1xx informational responses")
+				str.CancelRead(quic.StreamErrorCode(ErrCodeExcessiveLoad))
+				str.CancelWrite(quic.StreamErrorCode(ErrCodeExcessiveLoad))
+				return nil, errors.New("http3: too many 1xx informational responses")
 			}
 			traceGot1xxResponse(trace, resCode, textproto.MIMEHeader(res.Header))
-			if resCode == 100 {
+			if resCode == http.StatusContinue {
 				traceGot100Continue(trace)
 			}
 			continue

internal/http3/conn.go

@@ -2,6 +2,7 @@ package http3
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
@@ -19,6 +20,8 @@ import (
 	"github.com/quic-go/qpack"
 )
 
+var errGoAway = errors.New("connection in graceful shutdown")
+
 // Connection is an HTTP/3 connection.
 // It has all methods from the quic.Connection expect for AcceptStream, AcceptUniStream,
 // SendDatagram and ReceiveDatagram.
@@ -51,8 +54,10 @@ type connection struct {
 
 	decoder *qpack.Decoder
 
-	streamMx sync.Mutex
-	streams  map[quic.StreamID]*datagrammer
+	streamMx     sync.Mutex
+	streams      map[quic.StreamID]*datagrammer
+	lastStreamID quic.StreamID
+	maxStreamID  quic.StreamID
 
 	settings         *Settings
 	receivedSettings chan struct{}
@@ -80,6 +85,8 @@ func newConnection(
 		decoder:          qpack.NewDecoder(func(hf qpack.HeaderField) {}),
 		receivedSettings: make(chan struct{}),
 		streams:          make(map[quic.StreamID]*datagrammer),
+		maxStreamID:      InvalidStreamID,
+		lastStreamID:     InvalidStreamID,
 		Options:          options,
 	}
 	if idleTimeout > 0 {
@@ -100,6 +107,13 @@ func (c *connection) clearStream(id quic.StreamID) {
 	if c.idleTimeout > 0 && len(c.streams) == 0 {
 		c.idleTimer.Reset(c.idleTimeout)
 	}
+	// The server is performing a graceful shutdown.
+	// If no more streams are remaining, close the connection.
+	if c.maxStreamID != InvalidStreamID {
+		if len(c.streams) == 0 {
+			c.CloseWithError(quic.ApplicationErrorCode(ErrCodeNoError), "")
+		}
+	}
 }
 
 func (c *connection) openRequestStream(
@@ -109,13 +123,30 @@ func (c *connection) openRequestStream(
 	disableCompression bool,
 	maxHeaderBytes uint64,
 ) (*requestStream, error) {
+	if c.perspective == PerspectiveClient {
+		c.streamMx.Lock()
+		maxStreamID := c.maxStreamID
+		var nextStreamID quic.StreamID
+		if c.lastStreamID == InvalidStreamID {
+			nextStreamID = 0
+		} else {
+			nextStreamID = c.lastStreamID + 4
+		}
+		c.streamMx.Unlock()
+		// Streams with stream ID equal to or greater than the stream ID carried in the GOAWAY frame
+		// will be rejected, see section 5.2 of RFC 9114.
+		if maxStreamID != InvalidStreamID && nextStreamID >= maxStreamID {
+			return nil, errGoAway
+		}
+	}
 	str, err := c.OpenStreamSync(ctx)
 	if err != nil {
 		return nil, err
 	}
 	datagrams := newDatagrammer(func(b []byte) error { return c.sendDatagram(str.StreamID(), b) })
 	c.streamMx.Lock()
 	c.streams[str.StreamID()] = datagrams
+	c.lastStreamID = str.StreamID()
 	c.streamMx.Unlock()
 	qstr := newStateTrackingStream(str, c, datagrams)
 	rsp := &http.Response{}
@@ -247,41 +278,94 @@ func (c *connection) handleUnidirectionalStreams(hijack func(StreamType, quic.Co
 				c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeStreamCreationError), "duplicate control stream")
 				return
 			}
-			fp := &frameParser{conn: c.Connection, r: str}
-			f, err := fp.ParseNext()
-			if err != nil {
-				c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameError), "")
-				return
-			}
-			sf, ok := f.(*settingsFrame)
-			if !ok {
-				c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeMissingSettings), "")
-				return
-			}
-			c.settings = &Settings{
-				EnableDatagrams:       sf.Datagram,
-				EnableExtendedConnect: sf.ExtendedConnect,
-				Other:                 sf.Other,
-			}
-			close(c.receivedSettings)
-			if !sf.Datagram {
-				return
+			c.handleControlStream(str)
+		}(str)
+	}
+}
+
+func (c *connection) handleControlStream(str quic.ReceiveStream) {
+	fp := &frameParser{conn: c.Connection, r: str}
+	f, err := fp.ParseNext()
+	if err != nil {
+		var serr *quic.StreamError
+		if err == io.EOF || errors.As(err, &serr) {
+			c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeClosedCriticalStream), "")
+			return
+		}
+		c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameError), "")
+		return
+	}
+	sf, ok := f.(*settingsFrame)
+	if !ok {
+		c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeMissingSettings), "")
+		return
+	}
+	c.settings = &Settings{
+		EnableDatagrams:       sf.Datagram,
+		EnableExtendedConnect: sf.ExtendedConnect,
+		Other:                 sf.Other,
+	}
+	close(c.receivedSettings)
+	if sf.Datagram {
+		// If datagram support was enabled on our side as well as on the server side,
+		// we can expect it to have been negotiated both on the transport and on the HTTP/3 layer.
+		// Note: ConnectionState() will block until the handshake is complete (relevant when using 0-RTT).
+		if c.enableDatagrams && !c.ConnectionState().SupportsDatagrams {
+			c.CloseWithError(quic.ApplicationErrorCode(ErrCodeSettingsError), "missing QUIC Datagram support")
+			return
+		}
+		go func() {
+			if err := c.receiveDatagrams(); err != nil {
+				if c.logger != nil {
+					c.logger.Debug("receiving datagrams failed", "error", err)
+				}
 			}
-			// If datagram support was enabled on our side as well as on the server side,
-			// we can expect it to have been negotiated both on the transport and on the HTTP/3 layer.
-			// Note: ConnectionState() will block until the handshake is complete (relevant when using 0-RTT).
-			if c.enableDatagrams && !c.ConnectionState().SupportsDatagrams {
-				c.CloseWithError(quic.ApplicationErrorCode(ErrCodeSettingsError), "missing QUIC Datagram support")
+		}()
+	}
+
+	// we don't support server push, hence we don't expect any GOAWAY frames from the client
+	if c.perspective == PerspectiveServer {
+		return
+	}
+
+	for {
+		f, err := fp.ParseNext()
+		if err != nil {
+			var serr *quic.StreamError
+			if err == io.EOF || errors.As(err, &serr) {
+				c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeClosedCriticalStream), "")
 				return
 			}
-			go func() {
-				if err := c.receiveDatagrams(); err != nil {
-					if c.logger != nil {
-						c.logger.Debug("receiving datagrams failed", "error", err)
-					}
-				}
-			}()
-		}(str)
+			c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameError), "")
+			return
+		}
+		// GOAWAY is the only frame allowed at this point:
+		// * unexpected frames are ignored by the frame parser
+		// * we don't support any extension that might add support for more frames
+		goaway, ok := f.(*goAwayFrame)
+		if !ok {
+			c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameUnexpected), "")
+			return
+		}
+		if goaway.StreamID%4 != 0 { // client-initiated, bidirectional streams
+			c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeIDError), "")
+			return
+		}
+		c.streamMx.Lock()
+		if c.maxStreamID != InvalidStreamID && goaway.StreamID > c.maxStreamID {
+			c.streamMx.Unlock()
+			c.Connection.CloseWithError(quic.ApplicationErrorCode(ErrCodeIDError), "")
+			return
+		}
+		c.maxStreamID = goaway.StreamID
+		hasActiveStreams := len(c.streams) > 0
+		c.streamMx.Unlock()
+
+		// immediately close the connection if there are currently no active requests
+		if !hasActiveStreams {
+			c.CloseWithError(quic.ApplicationErrorCode(ErrCodeNoError), "")
+			return
+		}
 	}
 }
 
@@ -311,11 +395,10 @@ func (c *connection) receiveDatagrams() error {
 		streamID := quic.StreamID(4 * quarterStreamID)
 		c.streamMx.Lock()
 		dg, ok := c.streams[streamID]
+		c.streamMx.Unlock()
 		if !ok {
-			c.streamMx.Unlock()
-			return nil
+			continue
 		}
-		c.streamMx.Unlock()
 		dg.enqueue(b[n:])
 	}
 }

internal/http3/headers.go

@@ -211,13 +211,6 @@ func requestFromHeaders(headerFields []qpack.HeaderField) (*http.Request, error)
 	}, nil
 }
 
-func hostnameFromURL(url *url.URL) string {
-	if url != nil {
-		return url.Host
-	}
-	return ""
-}
-
 // updateResponseFromHeaders sets up http.Response as an HTTP/3 response,
 // using the decoded qpack header filed.
 // It is only called for the HTTP header (and not the HTTP trailer).
@@ -228,7 +221,7 @@ func updateResponseFromHeaders(rsp *http.Response, headerFields []qpack.HeaderFi
 		return err
 	}
 	if hdr.Status == "" {
-		return errors.New("missing status field")
+		return errors.New("missing :status field")
 	}
 	rsp.Proto = "HTTP/3.0"
 	rsp.ProtoMajor = 3

internal/http3/protocol.go

@@ -53,34 +53,9 @@ const (
 	StreamTypeBidi
 )
 
-// A StreamID in QUIC
-type StreamID int64
-
-// InitiatedBy says if the stream was initiated by the client or by the server
-func (s StreamID) InitiatedBy() Perspective {
-	if s%2 == 0 {
-		return PerspectiveClient
-	}
-	return PerspectiveServer
-}
-
-// Type says if this is a unidirectional or bidirectional stream
-func (s StreamID) Type() StreamType {
-	if s%4 >= 2 {
-		return StreamTypeUni
-	}
-	return StreamTypeBidi
-}
-
-// StreamNum returns how many streams in total are below this
-// Example: for stream 9 it returns 3 (i.e. streams 1, 5 and 9)
-func (s StreamID) StreamNum() StreamNum {
-	return StreamNum(s/4) + 1
-}
-
 // InvalidPacketNumber is a stream ID that is invalid.
 // The first valid stream ID in QUIC is 0.
-const InvalidStreamID StreamID = -1
+const InvalidStreamID quic.StreamID = -1
 
 // StreamNum is the stream number
 type StreamNum int64
@@ -94,11 +69,11 @@ const (
 )
 
 // StreamID calculates the stream ID.
-func (s StreamNum) StreamID(stype StreamType, pers Perspective) StreamID {
+func (s StreamNum) StreamID(stype StreamType, pers Perspective) quic.StreamID {
 	if s == 0 {
 		return InvalidStreamID
 	}
-	var first StreamID
+	var first quic.StreamID
 	switch stype {
 	case StreamTypeBidi:
 		switch pers {
@@ -115,5 +90,5 @@ func (s StreamNum) StreamID(stype StreamType, pers Perspective) StreamID {
 			first = 3
 		}
 	}
-	return first + 4*StreamID(s-1)
+	return first + 4*quic.StreamID(s-1)
 }

internal/http3/request_writer.go

@@ -16,7 +16,6 @@ import (
 	reqheader "github.com/imroc/req/v3/internal/header"
 	"github.com/quic-go/qpack"
 
-	"github.com/quic-go/quic-go"
 	"golang.org/x/net/http/httpguts"
 	"golang.org/x/net/http2/hpack"
 	"golang.org/x/net/idna"
@@ -39,13 +38,13 @@ func newRequestWriter() *requestWriter {
 	}
 }
 
-func (w *requestWriter) WriteRequestHeader(str quic.Stream, req *http.Request, gzip bool, dumps []*dump.Dumper) error {
+func (w *requestWriter) WriteRequestHeader(wr io.Writer, req *http.Request, gzip bool, dumps []*dump.Dumper) error {
 	// TODO: figure out how to add support for trailers
 	buf := &bytes.Buffer{}
 	if err := w.writeHeaders(buf, req, gzip, dumps); err != nil {
 		return err
 	}
-	if _, err := str.Write(buf.Bytes()); err != nil {
+	if _, err := wr.Write(buf.Bytes()); err != nil {
 		return err
 	}
 	trace := httptrace.ContextClientTrace(req.Context())