/
client.go
333 lines (298 loc) · 10 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
package http3
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptrace"
"net/textproto"
"sync"
"time"
"github.com/quic-go/quic-go"
"github.com/quic-go/quic-go/internal/protocol"
"github.com/quic-go/quic-go/quicvarint"
"github.com/quic-go/qpack"
)
const (
// MethodGet0RTT allows a GET request to be sent using 0-RTT.
// Note that 0-RTT doesn't provide replay protection and should only be used for idempotent requests.
MethodGet0RTT = "GET_0RTT"
// MethodHead0RTT allows a HEAD request to be sent using 0-RTT.
// Note that 0-RTT doesn't provide replay protection and should only be used for idempotent requests.
MethodHead0RTT = "HEAD_0RTT"
)
const (
defaultUserAgent = "quic-go HTTP/3"
defaultMaxResponseHeaderBytes = 10 * 1 << 20 // 10 MB
)
var defaultQuicConfig = &quic.Config{
MaxIncomingStreams: -1, // don't allow the server to create bidirectional streams
KeepAlivePeriod: 10 * time.Second,
}
// SingleDestinationRoundTripper is an HTTP/3 client doing requests to a single remote server.
type SingleDestinationRoundTripper struct {
Connection quic.Connection
// Enable support for HTTP/3 datagrams (RFC 9297).
// If a QUICConfig is set, datagram support also needs to be enabled on the QUIC layer by setting EnableDatagrams.
EnableDatagrams bool
// Additional HTTP/3 settings.
// It is invalid to specify any settings defined by RFC 9114 (HTTP/3) and RFC 9297 (HTTP Datagrams).
AdditionalSettings map[uint64]uint64
StreamHijacker func(FrameType, quic.ConnectionTracingID, quic.Stream, error) (hijacked bool, err error)
UniStreamHijacker func(StreamType, quic.ConnectionTracingID, quic.ReceiveStream, error) (hijacked bool)
// MaxResponseHeaderBytes specifies a limit on how many response bytes are
// allowed in the server's response header.
// Zero means to use a default limit.
MaxResponseHeaderBytes int64
// DisableCompression, if true, prevents the Transport from requesting compression with an
// "Accept-Encoding: gzip" request header when the Request contains no existing Accept-Encoding value.
// If the Transport requests gzip on its own and gets a gzipped response, it's transparently
// decoded in the Response.Body.
// However, if the user explicitly requested gzip it is not automatically uncompressed.
DisableCompression bool
Logger *slog.Logger
initOnce sync.Once
hconn *connection
requestWriter *requestWriter
decoder *qpack.Decoder
}
var _ http.RoundTripper = &SingleDestinationRoundTripper{}
func (c *SingleDestinationRoundTripper) Start() Connection {
c.initOnce.Do(func() { c.init() })
return c.hconn
}
func (c *SingleDestinationRoundTripper) init() {
c.decoder = qpack.NewDecoder(func(hf qpack.HeaderField) {})
c.requestWriter = newRequestWriter()
c.hconn = newConnection(c.Connection, c.EnableDatagrams, protocol.PerspectiveClient, c.Logger)
// send the SETTINGs frame, using 0-RTT data, if possible
go func() {
if err := c.setupConn(c.hconn); err != nil {
if c.Logger != nil {
c.Logger.Debug("Setting up connection failed", "error", err)
}
c.hconn.CloseWithError(quic.ApplicationErrorCode(ErrCodeInternalError), "")
}
}()
if c.StreamHijacker != nil {
go c.handleBidirectionalStreams()
}
go c.hconn.HandleUnidirectionalStreams(c.UniStreamHijacker)
}
func (c *SingleDestinationRoundTripper) setupConn(conn *connection) error {
// open the control stream
str, err := conn.OpenUniStream()
if err != nil {
return err
}
b := make([]byte, 0, 64)
b = quicvarint.Append(b, streamTypeControlStream)
// send the SETTINGS frame
b = (&settingsFrame{Datagram: c.EnableDatagrams, Other: c.AdditionalSettings}).Append(b)
_, err = str.Write(b)
return err
}
func (c *SingleDestinationRoundTripper) handleBidirectionalStreams() {
for {
str, err := c.hconn.AcceptStream(context.Background())
if err != nil {
if c.Logger != nil {
c.Logger.Debug("accepting bidirectional stream failed", "error", err)
}
return
}
go func(str quic.Stream) {
_, err := parseNextFrame(str, func(ft FrameType, e error) (processed bool, err error) {
id := c.hconn.Context().Value(quic.ConnectionTracingKey).(quic.ConnectionTracingID)
return c.StreamHijacker(ft, id, str, e)
})
if err == errHijacked {
return
}
if err != nil {
if c.Logger != nil {
c.Logger.Debug("error handling stream", "error", err)
}
}
c.hconn.CloseWithError(quic.ApplicationErrorCode(ErrCodeFrameUnexpected), "received HTTP/3 frame on bidirectional stream")
}(str)
}
}
func (c *SingleDestinationRoundTripper) maxHeaderBytes() uint64 {
if c.MaxResponseHeaderBytes <= 0 {
return defaultMaxResponseHeaderBytes
}
return uint64(c.MaxResponseHeaderBytes)
}
// RoundTrip executes a request and returns a response
func (c *SingleDestinationRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
c.initOnce.Do(func() { c.init() })
rsp, err := c.roundTrip(req)
if err != nil && req.Context().Err() != nil {
// if the context was canceled, return the context cancellation error
err = req.Context().Err()
}
return rsp, err
}
func (c *SingleDestinationRoundTripper) roundTrip(req *http.Request) (*http.Response, error) {
// Immediately send out this request, if this is a 0-RTT request.
switch req.Method {
case MethodGet0RTT:
// don't modify the original request
reqCopy := *req
req = &reqCopy
req.Method = http.MethodGet
case MethodHead0RTT:
// don't modify the original request
reqCopy := *req
req = &reqCopy
req.Method = http.MethodHead
default:
// wait for the handshake to complete
earlyConn, ok := c.Connection.(quic.EarlyConnection)
if ok {
select {
case <-earlyConn.HandshakeComplete():
case <-req.Context().Done():
return nil, req.Context().Err()
}
}
}
// It is only possible to send an Extended CONNECT request once the SETTINGS were received.
// See section 3 of RFC 8441.
if isExtendedConnectRequest(req) {
connCtx := c.Connection.Context()
// wait for the server's SETTINGS frame to arrive
select {
case <-c.hconn.ReceivedSettings():
case <-connCtx.Done():
return nil, context.Cause(connCtx)
}
if !c.hconn.Settings().EnableExtendedConnect {
return nil, errors.New("http3: server didn't enable Extended CONNECT")
}
}
reqDone := make(chan struct{})
str, err := c.hconn.openRequestStream(req.Context(), c.requestWriter, reqDone, c.DisableCompression, c.maxHeaderBytes())
if err != nil {
return nil, err
}
// Request Cancellation:
// This go routine keeps running even after RoundTripOpt() returns.
// It is shut down when the application is done processing the body.
done := make(chan struct{})
go func() {
defer close(done)
select {
case <-req.Context().Done():
str.CancelWrite(quic.StreamErrorCode(ErrCodeRequestCanceled))
str.CancelRead(quic.StreamErrorCode(ErrCodeRequestCanceled))
case <-reqDone:
}
}()
rsp, err := c.doRequest(req, str)
if err != nil { // if any error occurred
close(reqDone)
<-done
return nil, maybeReplaceError(err)
}
return rsp, maybeReplaceError(err)
}
func (c *SingleDestinationRoundTripper) OpenRequestStream(ctx context.Context) (RequestStream, error) {
c.initOnce.Do(func() { c.init() })
return c.hconn.openRequestStream(ctx, c.requestWriter, nil, c.DisableCompression, c.maxHeaderBytes())
}
// cancelingReader reads from the io.Reader.
// It cancels writing on the stream if any error other than io.EOF occurs.
type cancelingReader struct {
r io.Reader
str Stream
}
func (r *cancelingReader) Read(b []byte) (int, error) {
n, err := r.r.Read(b)
if err != nil && err != io.EOF {
r.str.CancelWrite(quic.StreamErrorCode(ErrCodeRequestCanceled))
}
return n, err
}
func (c *SingleDestinationRoundTripper) sendRequestBody(str Stream, body io.ReadCloser, contentLength int64) error {
defer body.Close()
buf := make([]byte, bodyCopyBufferSize)
sr := &cancelingReader{str: str, r: body}
if contentLength == -1 {
_, err := io.CopyBuffer(str, sr, buf)
return err
}
// make sure we don't send more bytes than the content length
n, err := io.CopyBuffer(str, io.LimitReader(sr, contentLength), buf)
if err != nil {
return err
}
var extra int64
extra, err = io.CopyBuffer(io.Discard, sr, buf)
n += extra
if n > contentLength {
str.CancelWrite(quic.StreamErrorCode(ErrCodeRequestCanceled))
return fmt.Errorf("http: ContentLength=%d with Body length %d", contentLength, n)
}
return err
}
func (c *SingleDestinationRoundTripper) doRequest(req *http.Request, str *requestStream) (*http.Response, error) {
if err := str.SendRequestHeader(req); err != nil {
return nil, err
}
if req.Body == nil {
str.Close()
} else {
// send the request body asynchronously
go func() {
contentLength := int64(-1)
// According to the documentation for http.Request.ContentLength,
// a value of 0 with a non-nil Body is also treated as unknown content length.
if req.ContentLength > 0 {
contentLength = req.ContentLength
}
if err := c.sendRequestBody(str, req.Body, contentLength); err != nil {
if c.Logger != nil {
c.Logger.Debug("error writing request", "error", err)
}
}
str.Close()
}()
}
// copy from net/http: support 1xx responses
trace := httptrace.ContextClientTrace(req.Context())
num1xx := 0 // number of informational 1xx headers received
const max1xxResponses = 5 // arbitrary bound on number of informational responses
var res *http.Response
for {
var err error
res, err = str.ReadResponse()
if err != nil {
return nil, err
}
resCode := res.StatusCode
is1xx := 100 <= resCode && resCode <= 199
// treat 101 as a terminal status, see https://github.com/golang/go/issues/26161
is1xxNonTerminal := is1xx && resCode != http.StatusSwitchingProtocols
if is1xxNonTerminal {
num1xx++
if num1xx > max1xxResponses {
return nil, errors.New("http: too many 1xx informational responses")
}
if trace != nil && trace.Got1xxResponse != nil {
if err := trace.Got1xxResponse(resCode, textproto.MIMEHeader(res.Header)); err != nil {
return nil, err
}
}
continue
}
break
}
connState := c.hconn.ConnectionState().TLS
res.TLS = &connState
res.Request = req
return res, nil
}