http3, integrationtests/self, interop/http09, .: WebTransport, HTTP/3…

… trailers

This is a squashed commit of quic-go#3256. Subsequent commits will remove:

- WebTransport
- Trailers
- Preemptive MASQUE/DATAGRAM features

To slim down this PR.

closed_session.go

@@ -89,7 +89,7 @@ func (s *closedLocalSession) destroy(error) {
 	})
 }
 
-func (s *closedLocalSession) getPerspective() protocol.Perspective {
+func (s *closedLocalSession) Perspective() Perspective {
 	return s.perspective
 }
 
@@ -106,7 +106,7 @@ func newClosedRemoteSession(pers protocol.Perspective) packetHandler {
 	return &closedRemoteSession{perspective: pers}
 }
 
-func (s *closedRemoteSession) handlePacket(*receivedPacket)         {}
-func (s *closedRemoteSession) shutdown()                            {}
-func (s *closedRemoteSession) destroy(error)                        {}
-func (s *closedRemoteSession) getPerspective() protocol.Perspective { return s.perspective }
+func (s *closedRemoteSession) handlePacket(*receivedPacket) {}
+func (s *closedRemoteSession) shutdown()                    {}
+func (s *closedRemoteSession) destroy(error)                {}
+func (s *closedRemoteSession) Perspective() Perspective     { return s.perspective }

closed_session_test.go

@@ -28,7 +28,7 @@ var _ = Describe("Closed local session", func() {
 	})
 
 	It("tells its perspective", func() {
-		Expect(sess.getPerspective()).To(Equal(protocol.PerspectiveClient))
+		Expect(sess.Perspective()).To(Equal(protocol.PerspectiveClient))
 		// stop the session
 		sess.shutdown()
 	})

http3/body.go

@@ -1,84 +1,56 @@
 package http3
 
 import (
-	"fmt"
 	"io"
 
 	"github.com/lucas-clemente/quic-go"
+	"github.com/marten-seemann/qpack"
 )
 
+type trailerFunc func([]qpack.HeaderField, error)
+
 // The body of a http.Request or http.Response.
 type body struct {
-	str quic.Stream
+	str RequestStream
 
 	// only set for the http.Response
 	// The channel is closed when the user is done with this response:
 	// either when Read() errors, or when Close() is called.
 	reqDone       chan<- struct{}
 	reqDoneClosed bool
 
-	onFrameError func()
-
-	bytesRemainingInFrame uint64
+	onTrailers trailerFunc
 }
 
-var _ io.ReadCloser = &body{}
+var (
+	_ io.ReadCloser  = &body{}
+	_ WebTransporter = &body{}
+)
 
-func newRequestBody(str quic.Stream, onFrameError func()) *body {
+func newRequestBody(str RequestStream, onTrailers trailerFunc) *body {
 	return &body{
-		str:          str,
-		onFrameError: onFrameError,
+		str:        str,
+		onTrailers: onTrailers,
 	}
 }
 
-func newResponseBody(str quic.Stream, done chan<- struct{}, onFrameError func()) *body {
+func newResponseBody(str RequestStream, onTrailers trailerFunc, done chan<- struct{}) *body {
 	return &body{
-		str:          str,
-		onFrameError: onFrameError,
-		reqDone:      done,
+		str:        str,
+		onTrailers: onTrailers,
+		reqDone:    done,
 	}
 }
 
-func (r *body) Read(b []byte) (int, error) {
-	n, err := r.readImpl(b)
+func (r *body) Read(p []byte) (n int, err error) {
+	n, err = r.str.DataReader().Read(p)
 	if err != nil {
-		r.requestDone()
-	}
-	return n, err
-}
-
-func (r *body) readImpl(b []byte) (int, error) {
-	if r.bytesRemainingInFrame == 0 {
-	parseLoop:
-		for {
-			frame, err := parseNextFrame(r.str)
-			if err != nil {
-				return 0, err
-			}
-			switch f := frame.(type) {
-			case *headersFrame:
-				// skip HEADERS frames
-				continue
-			case *dataFrame:
-				r.bytesRemainingInFrame = f.Length
-				break parseLoop
-			default:
-				r.onFrameError()
-				// parseNextFrame skips over unknown frame types
-				// Therefore, this condition is only entered when we parsed another known frame type.
-				return 0, fmt.Errorf("peer sent an unexpected frame: %T", f)
-			}
+		// Read trailers if present
+		if err == io.EOF && r.onTrailers != nil {
+			r.onTrailers(r.str.ReadHeaders())
 		}
+		r.requestDone()
 	}
-
-	var n int
-	var err error
-	if r.bytesRemainingInFrame < uint64(len(b)) {
-		n, err = r.str.Read(b[:r.bytesRemainingInFrame])
-	} else {
-		n, err = r.str.Read(b)
-	}
-	r.bytesRemainingInFrame -= uint64(n)
 	return n, err
 }
 
@@ -96,3 +68,7 @@ func (r *body) Close() error {
 	r.str.CancelRead(quic.StreamErrorCode(errorRequestCanceled))
 	return nil
 }
+
+func (r *body) WebTransport() (WebTransport, error) {
+	return r.str.WebTransport()
+}

http3/body_test.go

@@ -4,10 +4,13 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"net/http"
 
 	"github.com/golang/mock/gomock"
 	"github.com/lucas-clemente/quic-go"
 	mockquic "github.com/lucas-clemente/quic-go/internal/mocks/quic"
+	"github.com/lucas-clemente/quic-go/quicvarint"
+	"github.com/marten-seemann/qpack"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -27,27 +30,34 @@ func (t bodyType) String() string {
 	return "response"
 }
 
-var _ = Describe("Body", func() {
+var _ = Describe("body", func() {
 	var (
-		rb            *body
-		str           *mockquic.MockStream
-		buf           *bytes.Buffer
-		reqDone       chan struct{}
-		errorCbCalled bool
+		rb          *body
+		sess        *mockquic.MockEarlySession
+		conn        *connection
+		str         *mockquic.MockStream
+		rstr        RequestStream
+		buf         *bytes.Buffer
+		trailers    []qpack.HeaderField
+		trailersErr error
+		reqDone     chan struct{}
 	)
 
-	errorCb := func() { errorCbCalled = true }
+	onTrailers := func(fields []qpack.HeaderField, err error) {
+		trailers = fields[:]
+		trailersErr = err
+	}
 
 	getDataFrame := func(data []byte) []byte {
-		b := &bytes.Buffer{}
-		(&dataFrame{Length: uint64(len(data))}).Write(b)
-		b.Write(data)
-		return b.Bytes()
+		buf := &bytes.Buffer{}
+		quicvarint.Write(buf, uint64(FrameTypeData))
+		quicvarint.Write(buf, uint64(len(data)))
+		buf.Write(data)
+		return buf.Bytes()
 	}
 
 	BeforeEach(func() {
 		buf = &bytes.Buffer{}
-		errorCbCalled = false
 	})
 
 	for _, bt := range []bodyType{bodyTypeRequest, bodyTypeResponse} {
@@ -62,13 +72,18 @@ var _ = Describe("Body", func() {
 				str.EXPECT().Read(gomock.Any()).DoAndReturn(func(b []byte) (int, error) {
 					return buf.Read(b)
 				}).AnyTimes()
+				str.EXPECT().StreamID().AnyTimes()
+
+				sess = mockquic.NewMockEarlySession(mockCtrl)
+				conn = newMockConn(sess, Settings{}, Settings{})
+				rstr = newRequestStream(conn, str, 0, 0)
 
 				switch bodyType {
 				case bodyTypeRequest:
-					rb = newRequestBody(str, errorCb)
+					rb = newRequestBody(rstr, onTrailers)
 				case bodyTypeResponse:
 					reqDone = make(chan struct{})
-					rb = newResponseBody(str, reqDone, errorCb)
+					rb = newResponseBody(rstr, onTrailers, reqDone)
 				}
 			})
 
@@ -119,40 +134,54 @@ var _ = Describe("Body", func() {
 			It("reads multiple DATA frames", func() {
 				buf.Write(getDataFrame([]byte("foo")))
 				buf.Write(getDataFrame([]byte("bar")))
-				b := make([]byte, 6)
-				n, err := rb.Read(b)
+				b, err := io.ReadAll(rb)
 				Expect(err).ToNot(HaveOccurred())
-				Expect(n).To(Equal(3))
-				Expect(b[:n]).To(Equal([]byte("foo")))
-				n, err = rb.Read(b)
+				Expect(len(b)).To(Equal(6))
+				Expect(b).To(Equal([]byte("foobar")))
+			})
+
+			It("reads trailers", func() {
+				buf.Write(getDataFrame([]byte("foo")))
+				buf.Write(getDataFrame([]byte("bar")))
+				fields := []qpack.HeaderField{
+					{Name: "foo", Value: "1"},
+					{Name: "bar", Value: "2"},
+				}
+				err := writeHeadersFrame(buf, fields, http.DefaultMaxHeaderBytes)
 				Expect(err).ToNot(HaveOccurred())
-				Expect(n).To(Equal(3))
-				Expect(b[:n]).To(Equal([]byte("bar")))
+				body, err := io.ReadAll(rb)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(len(body)).To(Equal(6))
+				Expect(body).To(Equal([]byte("foobar")))
+				Expect(trailers).To(Equal(fields))
+				Expect(trailersErr).ToNot(HaveOccurred())
 			})
 
-			It("skips HEADERS frames", func() {
+			It("receives an error on malformed trailers", func() {
 				buf.Write(getDataFrame([]byte("foo")))
-				(&headersFrame{Length: 10}).Write(buf)
-				buf.Write(make([]byte, 10))
 				buf.Write(getDataFrame([]byte("bar")))
-				b := make([]byte, 6)
-				n, err := io.ReadFull(rb, b)
+				quicvarint.Write(buf, uint64(FrameTypeHeaders))
+				quicvarint.Write(buf, 0x10)
+				buf.Write(make([]byte, 0x10))
+				sess.EXPECT().CloseWithError(quic.ApplicationErrorCode(errorGeneralProtocolError), gomock.Any())
+				body, err := io.ReadAll(rb)
 				Expect(err).ToNot(HaveOccurred())
-				Expect(n).To(Equal(6))
-				Expect(b).To(Equal([]byte("foobar")))
+				Expect(len(body)).To(Equal(6))
+				Expect(body).To(Equal([]byte("foobar")))
+				Expect(trailersErr).To(HaveOccurred())
 			})
 
 			It("errors when it can't parse the frame", func() {
 				buf.Write([]byte("invalid"))
 				_, err := rb.Read([]byte{0})
-				Expect(err).To(HaveOccurred())
+				Expect(err).To(Equal(io.EOF))
 			})
 
-			It("errors on unexpected frames, and calls the error callback", func() {
-				(&settingsFrame{}).Write(buf)
+			It("errors on unexpected frames, and closes the QUIC session", func() {
+				sess.EXPECT().CloseWithError(quic.ApplicationErrorCode(errorFrameUnexpected), gomock.Any())
+				Settings{}.writeFrame(buf)
 				_, err := rb.Read([]byte{0})
-				Expect(err).To(MatchError("peer sent an unexpected frame: *http3.settingsFrame"))
-				Expect(errorCbCalled).To(BeTrue())
+				Expect(err).To(MatchError(&FrameTypeError{Want: FrameTypeData, Type: FrameTypeSettings}))
 			})
 
 			if bodyType == bodyTypeResponse {

http3/capsule_type.go

@@ -0,0 +1,30 @@
+package http3
+
+import "fmt"
+
+const (
+	CapsuleTypeRegisterDatagramContext   CapsuleType = 0x00
+	CapsuleTypeCloseDatagramContext      CapsuleType = 0x01
+	CapsuleTypeDatagram                  CapsuleType = 0x02
+	CapsuleTypeRegisterDatagramNoContext CapsuleType = 0x03
+)
+
+// A CapsuleType represents an HTTP capsule type.
+// https://www.ietf.org/archive/id/draft-ietf-masque-h3-datagram-03.html#name-capsule-types
+type CapsuleType uint64
+
+// String returns the IETF registered name for t if available.
+func (t CapsuleType) String() string {
+	switch t {
+	case CapsuleTypeRegisterDatagramContext:
+		return "REGISTER_DATAGRAM_CONTEXT"
+	case CapsuleTypeCloseDatagramContext:
+		return "CLOSE_DATAGRAM_CONTEXT"
+	case CapsuleTypeDatagram:
+		return "DATAGRAM"
+	case CapsuleTypeRegisterDatagramNoContext:
+		return "REGISTER_DATAGRAM_NO_CONTEXT"
+	default:
+		return fmt.Sprintf("%#x", uint64(t))
+	}
+}