diff --git a/h2_bundle.go b/h2_bundle.go index 8f2ee116..86bbc0cc 100644 --- a/h2_bundle.go +++ b/h2_bundle.go @@ -4564,6 +4564,7 @@ type http2stream struct { state http2streamState trailer Header // accumulated trailers + readDeadline *time.Timer // nil if unused writeDeadline *time.Timer // nil if unused wroteHeaders bool // whether we wrote headers (not status 100) @@ -4882,6 +4883,8 @@ func (sc *http2serverConn) serve() { } case *http2startPushRequest: sc.startPush(v) + case func(*http2serverConn): + v(sc) default: panic(fmt.Sprintf("unexpected type %T", v)) } @@ -5507,6 +5510,9 @@ func (sc *http2serverConn) closeStream(st *http2stream, err error) { panic(fmt.Sprintf("invariant; can't close stream in state %v", st.state)) } st.state = http2stateClosed + if st.readDeadline != nil { + st.readDeadline.Stop() + } if st.writeDeadline != nil { st.writeDeadline.Stop() } @@ -5532,6 +5538,7 @@ func (sc *http2serverConn) closeStream(st *http2stream, err error) { p.CloseWithError(err) } + st.cancelCtx() st.cw.Close() // signals Handler's CloseNotifier, unblocks writes, etc sc.writeSched.CloseStream(st.id) } @@ -5769,6 +5776,16 @@ func (st *http2stream) copyTrailersToHandlerRequest() { } } +// onReadTimeout is run on its own goroutine (from time.AfterFunc) +// when the stream's ReadTimeout has fired. +func (st *http2stream) onReadTimeout() { + if st.body != nil { + // Wrap the ErrDeadlineExceeded to avoid callers depending on us + // returning the bare error. + st.body.CloseWithError(fmt.Errorf("%w", os.ErrDeadlineExceeded)) + } +} + // onWriteTimeout is run on its own goroutine (from time.AfterFunc) // when the stream's WriteTimeout has fired. func (st *http2stream) onWriteTimeout() { @@ -6586,6 +6603,62 @@ func (rws *http2responseWriterState) promoteUndeclaredTrailers() { } } +func (w *http2responseWriter) SetReadDeadline(deadline time.Time) error { + st := w.rws.stream + if !deadline.IsZero() && deadline.Before(time.Now()) { + // If we're setting a deadline in the past, reset the stream immediately + // so writes after SetWriteDeadline returns will fail. + st.onReadTimeout() + + return nil + } + w.rws.conn.sendServeMsg(func(sc *http2serverConn) { + if st.readDeadline != nil { + if !st.readDeadline.Stop() { + // Deadline already exceeded, or stream has been closed. + return + } + } + if deadline.IsZero() { + st.readDeadline = nil + } else if st.readDeadline == nil { + st.readDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onReadTimeout) + } else { + st.readDeadline.Reset(deadline.Sub(time.Now())) + } + }) + + return nil +} + +func (w *http2responseWriter) SetWriteDeadline(deadline time.Time) error { + st := w.rws.stream + if !deadline.IsZero() && deadline.Before(time.Now()) { + // If we're setting a deadline in the past, reset the stream immediately + // so writes after SetWriteDeadline returns will fail. + st.onWriteTimeout() + + return nil + } + w.rws.conn.sendServeMsg(func(sc *http2serverConn) { + if st.writeDeadline != nil { + if !st.writeDeadline.Stop() { + // Deadline already exceeded, or stream has been closed. + return + } + } + if deadline.IsZero() { + st.writeDeadline = nil + } else if st.writeDeadline == nil { + st.writeDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onWriteTimeout) + } else { + st.writeDeadline.Reset(deadline.Sub(time.Now())) + } + }) + + return nil +} + func (w *http2responseWriter) Flush() { rws := w.rws if rws == nil { diff --git a/h2_server_frames_test.go b/h2_server_frames_test.go new file mode 100644 index 00000000..2b6eca9f --- /dev/null +++ b/h2_server_frames_test.go @@ -0,0 +1,193 @@ +//go:build !nethttpomithttp2 + +package http_test + +import ( + "bytes" + "crypto/tls" + "errors" + "fmt" + "io" + "os" + "testing" + "time" + + . "github.com/bogdanfinn/fhttp" + "github.com/bogdanfinn/fhttp/http2" + "github.com/bogdanfinn/fhttp/http2/hpack" + "github.com/bogdanfinn/fhttp/httptest" +) + +// h2Servers set up a Server with each HTTP/2 server implementation: the one +// bundled in h2_bundle.go and the http2 package's. +var h2Servers = []struct { + name string + configure func(*Server) +}{ + {"bundled", func(s *Server) { ExportHttp2ConfigureServer(s, nil) }}, + {"http2", func(s *Server) { http2.ConfigureServer(s, nil) }}, +} + +// h2Client is a raw HTTP/2 client connection, for frames a real client would +// not send. +type h2Client struct { + t *testing.T + fr *http2.Framer + buf bytes.Buffer + enc *hpack.Encoder +} + +func newH2Client(t *testing.T, configure func(*Server), h HandlerFunc, settings ...http2.Setting) *h2Client { + t.Helper() + ts := httptest.NewUnstartedServer(h) + configure(ts.Config) + ts.TLS = ts.Config.TLSConfig + ts.StartTLS() + t.Cleanup(ts.Close) + + conn, err := tls.Dial("tcp", ts.Listener.Addr().String(), &tls.Config{InsecureSkipVerify: true, NextProtos: []string{"h2"}}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + conn.SetDeadline(time.Now().Add(10 * time.Second)) + if _, err := io.WriteString(conn, http2.ClientPreface); err != nil { + t.Fatal(err) + } + c := &h2Client{t: t, fr: http2.NewFramer(conn, conn)} + c.enc = hpack.NewEncoder(&c.buf) + if err := c.fr.WriteSettings(settings...); err != nil { + t.Fatal(err) + } + for { + f, err := c.fr.ReadFrame() + if err != nil { + t.Fatal(err) + } + if sf, ok := f.(*http2.SettingsFrame); ok && !sf.IsAck() { + break + } + } + if err := c.fr.WriteSettingsAck(); err != nil { + t.Fatal(err) + } + return c +} + +// writeRequest sends a GET request's headers on the stream, ending the +// stream unless the request is to have a body. +func (c *h2Client) writeRequest(streamID uint32, endStream bool) { + c.t.Helper() + c.buf.Reset() + for _, f := range [][2]string{{":method", "GET"}, {":scheme", "https"}, {":authority", "example.com"}, {":path", "/"}} { + c.enc.WriteField(hpack.HeaderField{Name: f[0], Value: f[1]}) + } + if err := c.fr.WriteHeaders(http2.HeadersFrameParam{StreamID: streamID, BlockFragment: c.buf.Bytes(), EndStream: endStream, EndHeaders: true}); err != nil { + c.t.Fatal(err) + } +} + +func waitFor(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(5 * time.Second): + t.Fatal(what) + } +} + +// A stream error must cancel the handler's request context (golang.org/issue/67036). +func TestH2ServerCancelsHandlerOnStreamError(t *testing.T) { + for _, srv := range h2Servers { + t.Run(srv.name, func(t *testing.T) { + started, done := make(chan struct{}), make(chan struct{}) + c := newH2Client(t, srv.configure, func(w ResponseWriter, r *Request) { + close(started) + <-r.Context().Done() + close(done) + }) + c.writeRequest(1, true) + waitFor(t, started, "handler did not start") + // A second set of request headers on the stream is a stream error. + c.writeRequest(1, true) + waitFor(t, done, "handler context was not canceled after the stream error") + }) + } +} + +// A write deadline must release a handler whose write is blocked on flow +// control, and cancel its request context. +func TestH2ServerWriteDeadlineUnblocksFlowControlledWrite(t *testing.T) { + for _, srv := range h2Servers { + t.Run(srv.name, func(t *testing.T) { + errc, done := make(chan error, 1), make(chan struct{}) + // The client grants no flow-control window, so the handler's write blocks. + c := newH2Client(t, srv.configure, func(w ResponseWriter, r *Request) { + defer close(done) + if err := NewResponseController(w).SetWriteDeadline(time.Now().Add(50 * time.Millisecond)); err != nil { + errc <- err + return + } + _, err := w.Write(make([]byte, 32<<10)) + errc <- err + <-r.Context().Done() + }, http2.Setting{ID: http2.SettingInitialWindowSize, Val: 0}) + c.writeRequest(1, true) + select { + case err := <-errc: + if err == nil || errors.Is(err, ErrNotSupported) { + t.Fatalf("blocked Write returned %v after the write deadline, want a write error", err) + } + case <-time.After(5 * time.Second): + t.Fatal("blocked Write did not return after the write deadline") + } + waitFor(t, done, "handler context was not canceled after the write deadline") + }) + } +} + +// A read deadline, in the future or already past, must release a handler +// waiting for a request body that never comes. +func TestH2ServerReadDeadline(t *testing.T) { + for _, srv := range h2Servers { + for _, d := range []time.Duration{50 * time.Millisecond, -time.Second} { + t.Run(fmt.Sprintf("%s/%v", srv.name, d), func(t *testing.T) { + errc := make(chan error, 1) + c := newH2Client(t, srv.configure, func(w ResponseWriter, r *Request) { + if err := NewResponseController(w).SetReadDeadline(time.Now().Add(d)); err != nil { + errc <- err + return + } + _, err := io.ReadAll(r.Body) + errc <- err + }) + c.writeRequest(1, false) // the body never comes + select { + case err := <-errc: + if !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("reading the body returned %v, want os.ErrDeadlineExceeded", err) + } + case <-time.After(5 * time.Second): + t.Fatal("reading the body did not return after the read deadline") + } + }) + } + } +} + +// A read deadline on a request without a body must not panic when it fires +// (golang.org/issue/58237). +func TestH2ServerReadDeadlineWithoutBody(t *testing.T) { + for _, srv := range h2Servers { + t.Run(srv.name, func(t *testing.T) { + done := make(chan struct{}) + c := newH2Client(t, srv.configure, func(w ResponseWriter, r *Request) { + defer close(done) + NewResponseController(w).SetReadDeadline(time.Now().Add(time.Millisecond)) + time.Sleep(10 * time.Millisecond) + }) + c.writeRequest(1, true) + waitFor(t, done, "handler did not finish") + }) + } +} diff --git a/http2/http2.go b/http2/http2.go index b77d8917..d30bb70c 100644 --- a/http2/http2.go +++ b/http2/http2.go @@ -33,11 +33,19 @@ import ( ) var ( - VerboseLogs bool - logFrameWrites bool - logFrameReads bool - inTests bool - disableExtendedConnectProtocol bool + VerboseLogs bool + logFrameWrites bool + logFrameReads bool + inTests bool + + // Enabling extended CONNECT by causes browsers to attempt to use + // WebSockets-over-HTTP/2. This results in problems when the server's websocket + // package doesn't support extended CONNECT. + // + // Disable extended CONNECT by default for now. + // + // Issue #71128. + disableExtendedConnectProtocol = true ) func init() { @@ -50,8 +58,8 @@ func init() { logFrameWrites = true logFrameReads = true } - if strings.Contains(e, "http2xconnect=0") { - disableExtendedConnectProtocol = true + if strings.Contains(e, "http2xconnect=1") { + disableExtendedConnectProtocol = false } } diff --git a/http2/server.go b/http2/server.go index 9173f1c4..d77d9f59 100644 --- a/http2/server.go +++ b/http2/server.go @@ -591,6 +591,7 @@ type stream struct { resetQueued bool // RST_STREAM queued for write; set by sc.resetStream gotTrailerHeader bool // HEADER frame for trailers was seen wroteHeaders bool // whether we wrote headers (not status 100) + readDeadline *time.Timer // nil if unused writeDeadline *time.Timer // nil if unused trailer http.Header // accumulated trailers @@ -905,6 +906,8 @@ func (sc *serverConn) serve() { } case *startPushRequest: sc.startPush(v) + case func(*serverConn): + v(sc) default: panic(fmt.Sprintf("unexpected type %T", v)) } @@ -1516,6 +1519,9 @@ func (sc *serverConn) closeStream(st *stream, err error) { panic(fmt.Sprintf("invariant; can't close stream in state %v", st.state)) } st.state = stateClosed + if st.readDeadline != nil { + st.readDeadline.Stop() + } if st.writeDeadline != nil { st.writeDeadline.Stop() } @@ -1541,6 +1547,7 @@ func (sc *serverConn) closeStream(st *stream, err error) { p.CloseWithError(err) } + st.cancelCtx() st.cw.Close() // signals Handler's CloseNotifier, unblocks writes, etc sc.writeSched.CloseStream(st.id) } @@ -1772,6 +1779,16 @@ func (st *stream) copyTrailersToHandlerRequest() { } } +// onReadTimeout is run on its own goroutine (from time.AfterFunc) +// when the stream's ReadTimeout has fired. +func (st *stream) onReadTimeout() { + if st.body != nil { + // Wrap the ErrDeadlineExceeded to avoid callers depending on us + // returning the bare error. + st.body.CloseWithError(fmt.Errorf("%w", os.ErrDeadlineExceeded)) + } +} + // onWriteTimeout is run on its own goroutine (from time.AfterFunc) // when the stream's WriteTimeout has fired. func (st *stream) onWriteTimeout() { @@ -1993,6 +2010,11 @@ func (sc *serverConn) newWriterAndRequest(st *stream, f *MetaHeadersFrame) (*res protocol: f.PseudoValue("protocol"), } + // extended connect is disabled, so we should not see :protocol + if disableExtendedConnectProtocol && rp.protocol != "" { + return nil, nil, streamError(f.StreamID, ErrCodeProtocol) + } + isConnect := rp.method == "CONNECT" if isConnect { if rp.protocol == "" && (rp.path != "" || rp.scheme != "" || rp.authority == "") { @@ -2572,6 +2594,58 @@ func (rws *responseWriterState) promoteUndeclaredTrailers() { } } +func (w *responseWriter) SetReadDeadline(deadline time.Time) error { + st := w.rws.stream + if !deadline.IsZero() && deadline.Before(time.Now()) { + // If we're setting a deadline in the past, reset the stream immediately + // so writes after SetWriteDeadline returns will fail. + st.onReadTimeout() + return nil + } + w.rws.conn.sendServeMsg(func(sc *serverConn) { + if st.readDeadline != nil { + if !st.readDeadline.Stop() { + // Deadline already exceeded, or stream has been closed. + return + } + } + if deadline.IsZero() { + st.readDeadline = nil + } else if st.readDeadline == nil { + st.readDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onReadTimeout) + } else { + st.readDeadline.Reset(deadline.Sub(time.Now())) + } + }) + return nil +} + +func (w *responseWriter) SetWriteDeadline(deadline time.Time) error { + st := w.rws.stream + if !deadline.IsZero() && deadline.Before(time.Now()) { + // If we're setting a deadline in the past, reset the stream immediately + // so writes after SetWriteDeadline returns will fail. + st.onWriteTimeout() + return nil + } + w.rws.conn.sendServeMsg(func(sc *serverConn) { + if st.writeDeadline != nil { + if !st.writeDeadline.Stop() { + // Deadline already exceeded, or stream has been closed. + return + } + } + if deadline.IsZero() { + st.writeDeadline = nil + } else if st.writeDeadline == nil { + st.writeDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onWriteTimeout) + } else { + st.writeDeadline.Reset(deadline.Sub(time.Now())) + } + }) + return nil +} + func (w *responseWriter) Flush() { rws := w.rws if rws == nil { diff --git a/http2/server_extended_connect_test.go b/http2/server_extended_connect_test.go new file mode 100644 index 00000000..04f0bc6c --- /dev/null +++ b/http2/server_extended_connect_test.go @@ -0,0 +1,64 @@ +package http2 + +import ( + "os" + "os/exec" + "testing" + "time" + + http "github.com/bogdanfinn/fhttp" +) + +// Extended CONNECT is off unless GODEBUG=http2xconnect=1 at startup +// (golang/go#71128). GODEBUG is read at init, so each setting runs in a new +// process. +func TestServerExtendedConnect(t *testing.T) { + if want := os.Getenv("FHTTP_TEST_XCONNECT"); want != "" { + testServerExtendedConnect(t, want == "on") + return + } + for _, tt := range []struct{ godebug, want string }{ + {"", "off"}, + {"http2xconnect=0", "off"}, + {"http2xconnect=1", "on"}, + } { + cmd := exec.Command(os.Args[0], "-test.run=^TestServerExtendedConnect$") + cmd.Env = append(os.Environ(), "FHTTP_TEST_XCONNECT="+tt.want, "GODEBUG="+tt.godebug) + if out, err := cmd.CombinedOutput(); err != nil { + t.Errorf("GODEBUG=%q: %v\n%s", tt.godebug, err, out) + } + } +} + +// testServerExtendedConnect checks that the server advertises extended CONNECT +// and accepts requests with :protocol exactly when it is enabled. +func testServerExtendedConnect(t *testing.T, enabled bool) { + gotProtocol := make(chan string, 1) + st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) { + gotProtocol <- r.Header.Get(":protocol") + }) + defer st.Close() + + advertised := false + st.greetAndCheckSettings(func(s Setting) error { + advertised = advertised || s.ID == SettingEnableConnectProtocol && s.Val == 1 + return nil + }) + if advertised != enabled { + t.Fatalf("SETTINGS_ENABLE_CONNECT_PROTOCOL advertised = %v, want %v", advertised, enabled) + } + + st.bodylessReq1(":method", "CONNECT", ":protocol", "websocket") + if !enabled { + st.wantRSTStream(1, ErrCodeProtocol) + return + } + select { + case p := <-gotProtocol: + if p != "websocket" { + t.Fatalf(":protocol = %q, want %q", p, "websocket") + } + case <-time.After(5 * time.Second): + t.Fatal("handler was not called for an extended CONNECT request") + } +} diff --git a/requestwrite_test.go b/requestwrite_test.go index 53dee7bd..96d6bcdc 100644 --- a/requestwrite_test.go +++ b/requestwrite_test.go @@ -12,6 +12,7 @@ import ( "io" "net" "net/url" + "slices" "strings" "testing" "testing/iotest" @@ -978,3 +979,49 @@ func (c *dumpConn) RemoteAddr() net.Addr { return nil } func (c *dumpConn) SetDeadline(t time.Time) error { return nil } func (c *dumpConn) SetReadDeadline(t time.Time) error { return nil } func (c *dumpConn) SetWriteDeadline(t time.Time) error { return nil } + +// The transfer headers take their values from the request but keep the name +// the caller wrote. +func TestRequestWriteKeepsTransferHeaderNames(t *testing.T) { + tests := []struct { + name string + method string + body io.Reader + length int64 + header Header + trailer Header + want []string // transfer header lines, in order + }{ + {"lowercase", "POST", strings.NewReader("abc"), 3, Header{"content-length": {"3"}}, nil, []string{"content-length: 3"}}, + {"uppercase with a wrong value", "POST", strings.NewReader("abc"), 3, Header{"CONTENT-LENGTH": {"99"}}, nil, []string{"CONTENT-LENGTH: 3"}}, + {"two spellings", "POST", strings.NewReader("abc"), 3, Header{"content-length": {"3"}, "Content-length": {"3"}}, nil, []string{"Content-Length: 3"}}, + {"GET drops it", "GET", nil, 0, Header{"content-length": {"3"}}, nil, nil}, + {"lowercase transfer-encoding", "POST", io.MultiReader(strings.NewReader("abc")), -1, Header{"transfer-encoding": {"chunked"}}, nil, []string{"transfer-encoding: chunked"}}, + {"lowercase trailer", "POST", io.MultiReader(strings.NewReader("abc")), -1, Header{"trailer": {"x"}}, Header{"X-Foo": nil}, []string{"Transfer-Encoding: chunked", "trailer: X-Foo"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := dummyRequest(tt.method) + req.Header, req.ContentLength, req.Trailer = tt.header, tt.length, tt.trailer + if tt.body != nil { + req.Body = io.NopCloser(tt.body) + } + var buf bytes.Buffer + if err := req.Write(&buf); err != nil { + t.Fatal(err) + } + head, _, _ := strings.Cut(buf.String(), "\r\n\r\n") + var got []string + for _, line := range strings.Split(head, "\r\n")[1:] { + name, _, _ := strings.Cut(line, ":") + switch strings.ToLower(name) { + case "content-length", "transfer-encoding", "trailer": + got = append(got, line) + } + } + if !slices.Equal(got, tt.want) { + t.Errorf("transfer headers %q, want %q\n%s", got, tt.want, head) + } + }) + } +} diff --git a/transfer.go b/transfer.go index bdb30abe..9e24cdcb 100644 --- a/transfer.go +++ b/transfer.go @@ -275,6 +275,8 @@ func (t *transferWriter) shouldSendContentLength() bool { // addHeaders adds transfer headers to an existing header object, replacing any the caller set with the same name. // Content-Length always comes from the request, as in net/http. func (t *transferWriter) addHeaders(hdrs *Header) error { + // Not setTransferHeader: the caller's Content-Length goes even when none is sent. + contentLength := headerNameFold(*hdrs, "Content-Length") deleteHeaderFold(*hdrs, "Content-Length") if t.Close && !hasToken(t.Header.get("Connection"), "close") { hdrs.Add("Connection", "close") @@ -284,7 +286,7 @@ func (t *transferWriter) addHeaders(hdrs *Header) error { // function of the sanitized field triple (Body, ContentLength, // TransferEncoding) if t.shouldSendContentLength() { - setTransferHeader(*hdrs, "Content-Length", strconv.FormatInt(t.ContentLength, 10)) + (*hdrs)[contentLength] = []string{strconv.FormatInt(t.ContentLength, 10)} } else if chunked(t.TransferEncoding) { setTransferHeader(*hdrs, "Transfer-Encoding", "chunked") } @@ -311,10 +313,30 @@ func (t *transferWriter) addHeaders(hdrs *Header) error { return nil } -// setTransferHeader sets key to value in h, removing the caller's value under any case of key. +// setTransferHeader sets key to value in h, replacing the caller's value under +// any case of key but keeping the name the caller wrote. func setTransferHeader(h Header, key, value string) { + name := headerNameFold(h, key) deleteHeaderFold(h, key) - h.Set(key, value) + h[name] = []string{value} +} + +// headerNameFold returns the name under which h has key, if h has it under +// exactly one case, and otherwise the canonical form of key. +func headerNameFold(h Header, key string) string { + name := "" + for k := range h { + if strings.EqualFold(k, key) { + if name != "" { + return CanonicalHeaderKey(key) + } + name = k + } + } + if name == "" { + return CanonicalHeaderKey(key) + } + return name } // deleteHeaderFold deletes key from h under any case.