1 // Copyright 2014 The Go Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style
3 // license that can be found in the LICENSE file.
4 5 // TODO: turn off the serve goroutine when idle, so
6 // an idle conn only has the readFrames goroutine active. (which could
7 // also be optimized probably to pin less memory in crypto/tls). This
8 // would involve tracking when the serve goroutine is active (atomic
9 // int32 read/CAS probably?) and starting it up when frames arrive,
10 // and shutting it down when all handlers exit. the occasional PING
11 // packets could use time.AfterFunc to call sc.wakeStartServeLoop()
12 // (which is a no-op if already running) and then queue the PING write
13 // as normal. The serve loop would then exit in most cases (if no
14 // Handlers running) and not be woken up again until the PING packet
15 // returns.
16 17 // TODO (maybe): add a mechanism for Handlers to going into
18 // half-closed-local mode (rw.(io.Closer) test?) but not exit their
19 // handler, and continue to be able to read from the
20 // Request.Body. This would be a somewhat semantic change from HTTP/1
21 // (or at least what we expose in net/http), so I'd probably want to
22 // add it there too. For now, this package says that returning from
23 // the Handler ServeHTTP function means you're both done reading and
24 // done writing, without a way to stop just one or the other.
25 26 package http2
27 28 import (
29 "bufio"
30 "bytes"
31 "context"
32 "crypto/rand"
33 "crypto/tls"
34 "errors"
35 "fmt"
36 "io"
37 "log"
38 "math"
39 "net"
40 "net/http"
41 "net/textproto"
42 "net/url"
43 "os"
44 "reflect"
45 "runtime"
46 "strconv"
47 "strings"
48 "sync"
49 "time"
50 51 "golang.org/x/net/http/httpguts"
52 "golang.org/x/net/http2/hpack"
53 "golang.org/x/net/internal/httpcommon"
54 )
55 56 const (
57 prefaceTimeout = 10 * time.Second
58 firstSettingsTimeout = 2 * time.Second // should be in-flight with preface anyway
59 handlerChunkWriteSize = 4 << 10
60 defaultMaxStreams = 250 // TODO: make this 100 as the GFE seems to?
61 62 // maxQueuedControlFrames is the maximum number of control frames like
63 // SETTINGS, PING and RST_STREAM that will be queued for writing before
64 // the connection is closed to prevent memory exhaustion attacks.
65 maxQueuedControlFrames = 10000
66 )
67 68 var (
69 errClientDisconnected = errors.New("client disconnected")
70 errClosedBody = errors.New("body closed by handler")
71 errHandlerComplete = errors.New("http2: request body closed due to handler exiting")
72 errStreamClosed = errors.New("http2: stream closed")
73 )
74 75 var responseWriterStatePool = sync.Pool{
76 New: func() interface{} {
77 rws := &responseWriterState{}
78 rws.bw = bufio.NewWriterSize(chunkWriter{rws}, handlerChunkWriteSize)
79 return rws
80 },
81 }
82 83 // Test hooks.
84 var (
85 testHookOnConn func()
86 testHookGetServerConn func(*serverConn)
87 testHookOnPanicMu *sync.Mutex // nil except in tests
88 testHookOnPanic func(sc *serverConn, panicVal interface{}) (rePanic bool)
89 )
90 91 // Server is an HTTP/2 server.
92 type Server struct {
93 // MaxHandlers limits the number of http.Handler ServeHTTP goroutines
94 // which may run at a time over all connections.
95 // Negative or zero no limit.
96 // TODO: implement
97 MaxHandlers int
98 99 // MaxConcurrentStreams optionally specifies the number of
100 // concurrent streams that each client may have open at a
101 // time. This is unrelated to the number of http.Handler goroutines
102 // which may be active globally, which is MaxHandlers.
103 // If zero, MaxConcurrentStreams defaults to at least 100, per
104 // the HTTP/2 spec's recommendations.
105 MaxConcurrentStreams uint32
106 107 // MaxDecoderHeaderTableSize optionally specifies the http2
108 // SETTINGS_HEADER_TABLE_SIZE to send in the initial settings frame. It
109 // informs the remote endpoint of the maximum size of the header compression
110 // table used to decode header blocks, in octets. If zero, the default value
111 // of 4096 is used.
112 MaxDecoderHeaderTableSize uint32
113 114 // MaxEncoderHeaderTableSize optionally specifies an upper limit for the
115 // header compression table used for encoding request headers. Received
116 // SETTINGS_HEADER_TABLE_SIZE settings are capped at this limit. If zero,
117 // the default value of 4096 is used.
118 MaxEncoderHeaderTableSize uint32
119 120 // MaxReadFrameSize optionally specifies the largest frame
121 // this server is willing to read. A valid value is between
122 // 16k and 16M, inclusive. If zero or otherwise invalid, a
123 // default value is used.
124 MaxReadFrameSize uint32
125 126 // PermitProhibitedCipherSuites, if true, permits the use of
127 // cipher suites prohibited by the HTTP/2 spec.
128 PermitProhibitedCipherSuites bool
129 130 // IdleTimeout specifies how long until idle clients should be
131 // closed with a GOAWAY frame. PING frames are not considered
132 // activity for the purposes of IdleTimeout.
133 // If zero or negative, there is no timeout.
134 IdleTimeout time.Duration
135 136 // ReadIdleTimeout is the timeout after which a health check using a ping
137 // frame will be carried out if no frame is received on the connection.
138 // If zero, no health check is performed.
139 ReadIdleTimeout time.Duration
140 141 // PingTimeout is the timeout after which the connection will be closed
142 // if a response to a ping is not received.
143 // If zero, a default of 15 seconds is used.
144 PingTimeout time.Duration
145 146 // WriteByteTimeout is the timeout after which a connection will be
147 // closed if no data can be written to it. The timeout begins when data is
148 // available to write, and is extended whenever any bytes are written.
149 // If zero or negative, there is no timeout.
150 WriteByteTimeout time.Duration
151 152 // MaxUploadBufferPerConnection is the size of the initial flow
153 // control window for each connections. The HTTP/2 spec does not
154 // allow this to be smaller than 65535 or larger than 2^32-1.
155 // If the value is outside this range, a default value will be
156 // used instead.
157 MaxUploadBufferPerConnection int32
158 159 // MaxUploadBufferPerStream is the size of the initial flow control
160 // window for each stream. The HTTP/2 spec does not allow this to
161 // be larger than 2^32-1. If the value is zero or larger than the
162 // maximum, a default value will be used instead.
163 MaxUploadBufferPerStream int32
164 165 // NewWriteScheduler constructs a write scheduler for a connection.
166 // If nil, a default scheduler is chosen.
167 NewWriteScheduler func() WriteScheduler
168 169 // CountError, if non-nil, is called on HTTP/2 server errors.
170 // It's intended to increment a metric for monitoring, such
171 // as an expvar or Prometheus metric.
172 // The errType consists of only ASCII word characters.
173 CountError func(errType string)
174 175 // Internal state. This is a pointer (rather than embedded directly)
176 // so that we don't embed a Mutex in this struct, which will make the
177 // struct non-copyable, which might break some callers.
178 state *serverInternalState
179 }
180 181 type serverInternalState struct {
182 mu sync.Mutex
183 activeConns map[*serverConn]struct{}
184 185 // Pool of error channels. This is per-Server rather than global
186 // because channels can't be reused across synctest bubbles.
187 errChanPool sync.Pool
188 }
189 190 func (s *serverInternalState) registerConn(sc *serverConn) {
191 if s == nil {
192 return // if the Server was used without calling ConfigureServer
193 }
194 s.mu.Lock()
195 s.activeConns[sc] = struct{}{}
196 s.mu.Unlock()
197 }
198 199 func (s *serverInternalState) unregisterConn(sc *serverConn) {
200 if s == nil {
201 return // if the Server was used without calling ConfigureServer
202 }
203 s.mu.Lock()
204 delete(s.activeConns, sc)
205 s.mu.Unlock()
206 }
207 208 func (s *serverInternalState) startGracefulShutdown() {
209 if s == nil {
210 return // if the Server was used without calling ConfigureServer
211 }
212 s.mu.Lock()
213 for sc := range s.activeConns {
214 sc.startGracefulShutdown()
215 }
216 s.mu.Unlock()
217 }
218 219 // Global error channel pool used for uninitialized Servers.
220 // We use a per-Server pool when possible to avoid using channels across synctest bubbles.
221 var errChanPool = sync.Pool{
222 New: func() any { return make(chan error, 1) },
223 }
224 225 func (s *serverInternalState) getErrChan() chan error {
226 if s == nil {
227 return errChanPool.Get().(chan error) // Server used without calling ConfigureServer
228 }
229 return s.errChanPool.Get().(chan error)
230 }
231 232 func (s *serverInternalState) putErrChan(ch chan error) {
233 if s == nil {
234 errChanPool.Put(ch) // Server used without calling ConfigureServer
235 return
236 }
237 s.errChanPool.Put(ch)
238 }
239 240 // ConfigureServer adds HTTP/2 support to a net/http Server.
241 //
242 // The configuration conf may be nil.
243 //
244 // ConfigureServer must be called before s begins serving.
245 func ConfigureServer(s *http.Server, conf *Server) error {
246 if s == nil {
247 panic("nil *http.Server")
248 }
249 if conf == nil {
250 conf = new(Server)
251 }
252 conf.state = &serverInternalState{
253 activeConns: make(map[*serverConn]struct{}),
254 errChanPool: sync.Pool{New: func() any { return make(chan error, 1) }},
255 }
256 if h1, h2 := s, conf; h2.IdleTimeout == 0 {
257 if h1.IdleTimeout != 0 {
258 h2.IdleTimeout = h1.IdleTimeout
259 } else {
260 h2.IdleTimeout = h1.ReadTimeout
261 }
262 }
263 s.RegisterOnShutdown(conf.state.startGracefulShutdown)
264 265 if s.TLSConfig == nil {
266 s.TLSConfig = new(tls.Config)
267 } else if s.TLSConfig.CipherSuites != nil && s.TLSConfig.MinVersion < tls.VersionTLS13 {
268 // If they already provided a TLS 1.0–1.2 CipherSuite list, return an
269 // error if it is missing ECDHE_RSA_WITH_AES_128_GCM_SHA256 or
270 // ECDHE_ECDSA_WITH_AES_128_GCM_SHA256.
271 haveRequired := false
272 for _, cs := range s.TLSConfig.CipherSuites {
273 switch cs {
274 case tls.TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
275 // Alternative MTI cipher to not discourage ECDSA-only servers.
276 // See http://golang.org/cl/30721 for further information.
277 tls.TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256:
278 haveRequired = true
279 }
280 }
281 if !haveRequired {
282 return fmt.Errorf("http2: TLSConfig.CipherSuites is missing an HTTP/2-required AES_128_GCM_SHA256 cipher (need at least one of TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 or TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256)")
283 }
284 }
285 286 // Note: not setting MinVersion to tls.VersionTLS12,
287 // as we don't want to interfere with HTTP/1.1 traffic
288 // on the user's server. We enforce TLS 1.2 later once
289 // we accept a connection. Ideally this should be done
290 // during next-proto selection, but using TLS <1.2 with
291 // HTTP/2 is still the client's bug.
292 293 s.TLSConfig.PreferServerCipherSuites = true
294 295 if !strSliceContains(s.TLSConfig.NextProtos, NextProtoTLS) {
296 s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, NextProtoTLS)
297 }
298 if !strSliceContains(s.TLSConfig.NextProtos, "http/1.1") {
299 s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, "http/1.1")
300 }
301 302 if s.TLSNextProto == nil {
303 s.TLSNextProto = map[string]func(*http.Server, *tls.Conn, http.Handler){}
304 }
305 protoHandler := func(hs *http.Server, c net.Conn, h http.Handler, sawClientPreface bool) {
306 if testHookOnConn != nil {
307 testHookOnConn()
308 }
309 // The TLSNextProto interface predates contexts, so
310 // the net/http package passes down its per-connection
311 // base context via an exported but unadvertised
312 // method on the Handler. This is for internal
313 // net/http<=>http2 use only.
314 var ctx context.Context
315 type baseContexter interface {
316 BaseContext() context.Context
317 }
318 if bc, ok := h.(baseContexter); ok {
319 ctx = bc.BaseContext()
320 }
321 conf.ServeConn(c, &ServeConnOpts{
322 Context: ctx,
323 Handler: h,
324 BaseConfig: hs,
325 SawClientPreface: sawClientPreface,
326 })
327 }
328 s.TLSNextProto[NextProtoTLS] = func(hs *http.Server, c *tls.Conn, h http.Handler) {
329 protoHandler(hs, c, h, false)
330 }
331 // The "unencrypted_http2" TLSNextProto key is used to pass off non-TLS HTTP/2 conns.
332 //
333 // A connection passed in this method has already had the HTTP/2 preface read from it.
334 s.TLSNextProto[nextProtoUnencryptedHTTP2] = func(hs *http.Server, c *tls.Conn, h http.Handler) {
335 nc, err := unencryptedNetConnFromTLSConn(c)
336 if err != nil {
337 if lg := hs.ErrorLog; lg != nil {
338 lg.Print(err)
339 } else {
340 log.Print(err)
341 }
342 go c.Close()
343 return
344 }
345 protoHandler(hs, nc, h, true)
346 }
347 return nil
348 }
349 350 // ServeConnOpts are options for the Server.ServeConn method.
351 type ServeConnOpts struct {
352 // Context is the base context to use.
353 // If nil, context.Background is used.
354 Context context.Context
355 356 // BaseConfig optionally sets the base configuration
357 // for values. If nil, defaults are used.
358 BaseConfig *http.Server
359 360 // Handler specifies which handler to use for processing
361 // requests. If nil, BaseConfig.Handler is used. If BaseConfig
362 // or BaseConfig.Handler is nil, http.DefaultServeMux is used.
363 Handler http.Handler
364 365 // UpgradeRequest is an initial request received on a connection
366 // undergoing an h2c upgrade. The request body must have been
367 // completely read from the connection before calling ServeConn,
368 // and the 101 Switching Protocols response written.
369 UpgradeRequest *http.Request
370 371 // Settings is the decoded contents of the HTTP2-Settings header
372 // in an h2c upgrade request.
373 Settings []byte
374 375 // SawClientPreface is set if the HTTP/2 connection preface
376 // has already been read from the connection.
377 SawClientPreface bool
378 }
379 380 func (o *ServeConnOpts) context() context.Context {
381 if o != nil && o.Context != nil {
382 return o.Context
383 }
384 return context.Background()
385 }
386 387 func (o *ServeConnOpts) baseConfig() *http.Server {
388 if o != nil && o.BaseConfig != nil {
389 return o.BaseConfig
390 }
391 return new(http.Server)
392 }
393 394 func (o *ServeConnOpts) handler() http.Handler {
395 if o != nil {
396 if o.Handler != nil {
397 return o.Handler
398 }
399 if o.BaseConfig != nil && o.BaseConfig.Handler != nil {
400 return o.BaseConfig.Handler
401 }
402 }
403 return http.DefaultServeMux
404 }
405 406 // ServeConn serves HTTP/2 requests on the provided connection and
407 // blocks until the connection is no longer readable.
408 //
409 // ServeConn starts speaking HTTP/2 assuming that c has not had any
410 // reads or writes. It writes its initial settings frame and expects
411 // to be able to read the preface and settings frame from the
412 // client. If c has a ConnectionState method like a *tls.Conn, the
413 // ConnectionState is used to verify the TLS ciphersuite and to set
414 // the Request.TLS field in Handlers.
415 //
416 // ServeConn does not support h2c by itself. Any h2c support must be
417 // implemented in terms of providing a suitably-behaving net.Conn.
418 //
419 // The opts parameter is optional. If nil, default values are used.
420 func (s *Server) ServeConn(c net.Conn, opts *ServeConnOpts) {
421 if opts == nil {
422 opts = &ServeConnOpts{}
423 }
424 s.serveConn(c, opts, nil)
425 }
426 427 func (s *Server) serveConn(c net.Conn, opts *ServeConnOpts, newf func(*serverConn)) {
428 baseCtx, cancel := serverConnBaseContext(c, opts)
429 defer cancel()
430 431 http1srv := opts.baseConfig()
432 conf := configFromServer(http1srv, s)
433 sc := &serverConn{
434 srv: s,
435 hs: http1srv,
436 conn: c,
437 baseCtx: baseCtx,
438 remoteAddrStr: c.RemoteAddr().String(),
439 bw: newBufferedWriter(c, conf.WriteByteTimeout),
440 handler: opts.handler(),
441 streams: make(map[uint32]*stream),
442 readFrameCh: make(chan readFrameResult),
443 wantWriteFrameCh: make(chan FrameWriteRequest, 8),
444 serveMsgCh: make(chan interface{}, 8),
445 wroteFrameCh: make(chan frameWriteResult, 1), // buffered; one send in writeFrameAsync
446 bodyReadCh: make(chan bodyReadMsg), // buffering doesn't matter either way
447 doneServing: make(chan struct{}),
448 clientMaxStreams: math.MaxUint32, // Section 6.5.2: "Initially, there is no limit to this value"
449 advMaxStreams: conf.MaxConcurrentStreams,
450 initialStreamSendWindowSize: initialWindowSize,
451 initialStreamRecvWindowSize: conf.MaxUploadBufferPerStream,
452 maxFrameSize: initialMaxFrameSize,
453 pingTimeout: conf.PingTimeout,
454 countErrorFunc: conf.CountError,
455 serveG: newGoroutineLock(),
456 pushEnabled: true,
457 sawClientPreface: opts.SawClientPreface,
458 }
459 if newf != nil {
460 newf(sc)
461 }
462 463 s.state.registerConn(sc)
464 defer s.state.unregisterConn(sc)
465 466 // The net/http package sets the write deadline from the
467 // http.Server.WriteTimeout during the TLS handshake, but then
468 // passes the connection off to us with the deadline already set.
469 // Write deadlines are set per stream in serverConn.newStream.
470 // Disarm the net.Conn write deadline here.
471 if sc.hs.WriteTimeout > 0 {
472 sc.conn.SetWriteDeadline(time.Time{})
473 }
474 475 if s.NewWriteScheduler != nil {
476 sc.writeSched = s.NewWriteScheduler()
477 } else {
478 sc.writeSched = newRoundRobinWriteScheduler()
479 }
480 481 // These start at the RFC-specified defaults. If there is a higher
482 // configured value for inflow, that will be updated when we send a
483 // WINDOW_UPDATE shortly after sending SETTINGS.
484 sc.flow.add(initialWindowSize)
485 sc.inflow.init(initialWindowSize)
486 sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf)
487 sc.hpackEncoder.SetMaxDynamicTableSizeLimit(conf.MaxEncoderHeaderTableSize)
488 489 fr := NewFramer(sc.bw, c)
490 if conf.CountError != nil {
491 fr.countError = conf.CountError
492 }
493 fr.ReadMetaHeaders = hpack.NewDecoder(conf.MaxDecoderHeaderTableSize, nil)
494 fr.MaxHeaderListSize = sc.maxHeaderListSize()
495 fr.SetMaxReadFrameSize(conf.MaxReadFrameSize)
496 sc.framer = fr
497 498 if tc, ok := c.(connectionStater); ok {
499 sc.tlsState = new(tls.ConnectionState)
500 *sc.tlsState = tc.ConnectionState()
501 // 9.2 Use of TLS Features
502 // An implementation of HTTP/2 over TLS MUST use TLS
503 // 1.2 or higher with the restrictions on feature set
504 // and cipher suite described in this section. Due to
505 // implementation limitations, it might not be
506 // possible to fail TLS negotiation. An endpoint MUST
507 // immediately terminate an HTTP/2 connection that
508 // does not meet the TLS requirements described in
509 // this section with a connection error (Section
510 // 5.4.1) of type INADEQUATE_SECURITY.
511 if sc.tlsState.Version < tls.VersionTLS12 {
512 sc.rejectConn(ErrCodeInadequateSecurity, "TLS version too low")
513 return
514 }
515 516 if sc.tlsState.ServerName == "" {
517 // Client must use SNI, but we don't enforce that anymore,
518 // since it was causing problems when connecting to bare IP
519 // addresses during development.
520 //
521 // TODO: optionally enforce? Or enforce at the time we receive
522 // a new request, and verify the ServerName matches the :authority?
523 // But that precludes proxy situations, perhaps.
524 //
525 // So for now, do nothing here again.
526 }
527 528 if !conf.PermitProhibitedCipherSuites && isBadCipher(sc.tlsState.CipherSuite) {
529 // "Endpoints MAY choose to generate a connection error
530 // (Section 5.4.1) of type INADEQUATE_SECURITY if one of
531 // the prohibited cipher suites are negotiated."
532 //
533 // We choose that. In my opinion, the spec is weak
534 // here. It also says both parties must support at least
535 // TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256 so there's no
536 // excuses here. If we really must, we could allow an
537 // "AllowInsecureWeakCiphers" option on the server later.
538 // Let's see how it plays out first.
539 sc.rejectConn(ErrCodeInadequateSecurity, fmt.Sprintf("Prohibited TLS 1.2 Cipher Suite: %x", sc.tlsState.CipherSuite))
540 return
541 }
542 }
543 544 if opts.Settings != nil {
545 fr := &SettingsFrame{
546 FrameHeader: FrameHeader{valid: true},
547 p: opts.Settings,
548 }
549 if err := fr.ForeachSetting(sc.processSetting); err != nil {
550 sc.rejectConn(ErrCodeProtocol, "invalid settings")
551 return
552 }
553 opts.Settings = nil
554 }
555 556 if hook := testHookGetServerConn; hook != nil {
557 hook(sc)
558 }
559 560 if opts.UpgradeRequest != nil {
561 sc.upgradeRequest(opts.UpgradeRequest)
562 opts.UpgradeRequest = nil
563 }
564 565 sc.serve(conf)
566 }
567 568 func serverConnBaseContext(c net.Conn, opts *ServeConnOpts) (ctx context.Context, cancel func()) {
569 ctx, cancel = context.WithCancel(opts.context())
570 ctx = context.WithValue(ctx, http.LocalAddrContextKey, c.LocalAddr())
571 if hs := opts.baseConfig(); hs != nil {
572 ctx = context.WithValue(ctx, http.ServerContextKey, hs)
573 }
574 return
575 }
576 577 func (sc *serverConn) rejectConn(err ErrCode, debug string) {
578 sc.vlogf("http2: server rejecting conn: %v, %s", err, debug)
579 // ignoring errors. hanging up anyway.
580 sc.framer.WriteGoAway(0, err, []byte(debug))
581 sc.bw.Flush()
582 sc.conn.Close()
583 }
584 585 type serverConn struct {
586 // Immutable:
587 srv *Server
588 hs *http.Server
589 conn net.Conn
590 bw *bufferedWriter // writing to conn
591 handler http.Handler
592 baseCtx context.Context
593 framer *Framer
594 doneServing chan struct{} // closed when serverConn.serve ends
595 readFrameCh chan readFrameResult // written by serverConn.readFrames
596 wantWriteFrameCh chan FrameWriteRequest // from handlers -> serve
597 wroteFrameCh chan frameWriteResult // from writeFrameAsync -> serve, tickles more frame writes
598 bodyReadCh chan bodyReadMsg // from handlers -> serve
599 serveMsgCh chan interface{} // misc messages & code to send to / run on the serve loop
600 flow outflow // conn-wide (not stream-specific) outbound flow control
601 inflow inflow // conn-wide inbound flow control
602 tlsState *tls.ConnectionState // shared by all handlers, like net/http
603 remoteAddrStr string
604 writeSched WriteScheduler
605 countErrorFunc func(errType string)
606 607 // Everything following is owned by the serve loop; use serveG.check():
608 serveG goroutineLock // used to verify funcs are on serve()
609 pushEnabled bool
610 sawClientPreface bool // preface has already been read, used in h2c upgrade
611 sawFirstSettings bool // got the initial SETTINGS frame after the preface
612 needToSendSettingsAck bool
613 unackedSettings int // how many SETTINGS have we sent without ACKs?
614 queuedControlFrames int // control frames in the writeSched queue
615 clientMaxStreams uint32 // SETTINGS_MAX_CONCURRENT_STREAMS from client (our PUSH_PROMISE limit)
616 advMaxStreams uint32 // our SETTINGS_MAX_CONCURRENT_STREAMS advertised the client
617 curClientStreams uint32 // number of open streams initiated by the client
618 curPushedStreams uint32 // number of open streams initiated by server push
619 curHandlers uint32 // number of running handler goroutines
620 maxClientStreamID uint32 // max ever seen from client (odd), or 0 if there have been no client requests
621 maxPushPromiseID uint32 // ID of the last push promise (even), or 0 if there have been no pushes
622 streams map[uint32]*stream
623 unstartedHandlers []unstartedHandler
624 initialStreamSendWindowSize int32
625 initialStreamRecvWindowSize int32
626 maxFrameSize int32
627 peerMaxHeaderListSize uint32 // zero means unknown (default)
628 canonHeader map[string]string // http2-lower-case -> Go-Canonical-Case
629 canonHeaderKeysSize int // canonHeader keys size in bytes
630 writingFrame bool // started writing a frame (on serve goroutine or separate)
631 writingFrameAsync bool // started a frame on its own goroutine but haven't heard back on wroteFrameCh
632 needsFrameFlush bool // last frame write wasn't a flush
633 inGoAway bool // we've started to or sent GOAWAY
634 inFrameScheduleLoop bool // whether we're in the scheduleFrameWrite loop
635 needToSendGoAway bool // we need to schedule a GOAWAY frame write
636 pingSent bool
637 sentPingData [8]byte
638 goAwayCode ErrCode
639 shutdownTimer *time.Timer // nil until used
640 idleTimer *time.Timer // nil if unused
641 readIdleTimeout time.Duration
642 pingTimeout time.Duration
643 readIdleTimer *time.Timer // nil if unused
644 645 // Owned by the writeFrameAsync goroutine:
646 headerWriteBuf bytes.Buffer
647 hpackEncoder *hpack.Encoder
648 649 // Used by startGracefulShutdown.
650 shutdownOnce sync.Once
651 }
652 653 func (sc *serverConn) maxHeaderListSize() uint32 {
654 n := sc.hs.MaxHeaderBytes
655 if n <= 0 {
656 n = http.DefaultMaxHeaderBytes
657 }
658 return uint32(adjustHTTP1MaxHeaderSize(int64(n)))
659 }
660 661 func (sc *serverConn) curOpenStreams() uint32 {
662 sc.serveG.check()
663 return sc.curClientStreams + sc.curPushedStreams
664 }
665 666 // stream represents a stream. This is the minimal metadata needed by
667 // the serve goroutine. Most of the actual stream state is owned by
668 // the http.Handler's goroutine in the responseWriter. Because the
669 // responseWriter's responseWriterState is recycled at the end of a
670 // handler, this struct intentionally has no pointer to the
671 // *responseWriter{,State} itself, as the Handler ending nils out the
672 // responseWriter's state field.
673 type stream struct {
674 // immutable:
675 sc *serverConn
676 id uint32
677 body *pipe // non-nil if expecting DATA frames
678 cw closeWaiter // closed wait stream transitions to closed state
679 ctx context.Context
680 cancelCtx func()
681 682 // owned by serverConn's serve loop:
683 bodyBytes int64 // body bytes seen so far
684 declBodyBytes int64 // or -1 if undeclared
685 flow outflow // limits writing from Handler to client
686 inflow inflow // what the client is allowed to POST/etc to us
687 state streamState
688 resetQueued bool // RST_STREAM queued for write; set by sc.resetStream
689 gotTrailerHeader bool // HEADER frame for trailers was seen
690 wroteHeaders bool // whether we wrote headers (not status 100)
691 readDeadline *time.Timer // nil if unused
692 writeDeadline *time.Timer // nil if unused
693 closeErr error // set before cw is closed
694 695 trailer http.Header // accumulated trailers
696 reqTrailer http.Header // handler's Request.Trailer
697 }
698 699 func (sc *serverConn) Framer() *Framer { return sc.framer }
700 func (sc *serverConn) CloseConn() error { return sc.conn.Close() }
701 func (sc *serverConn) Flush() error { return sc.bw.Flush() }
702 func (sc *serverConn) HeaderEncoder() (*hpack.Encoder, *bytes.Buffer) {
703 return sc.hpackEncoder, &sc.headerWriteBuf
704 }
705 706 func (sc *serverConn) state(streamID uint32) (streamState, *stream) {
707 sc.serveG.check()
708 // http://tools.ietf.org/html/rfc7540#section-5.1
709 if st, ok := sc.streams[streamID]; ok {
710 return st.state, st
711 }
712 // "The first use of a new stream identifier implicitly closes all
713 // streams in the "idle" state that might have been initiated by
714 // that peer with a lower-valued stream identifier. For example, if
715 // a client sends a HEADERS frame on stream 7 without ever sending a
716 // frame on stream 5, then stream 5 transitions to the "closed"
717 // state when the first frame for stream 7 is sent or received."
718 if streamID%2 == 1 {
719 if streamID <= sc.maxClientStreamID {
720 return stateClosed, nil
721 }
722 } else {
723 if streamID <= sc.maxPushPromiseID {
724 return stateClosed, nil
725 }
726 }
727 return stateIdle, nil
728 }
729 730 // setConnState calls the net/http ConnState hook for this connection, if configured.
731 // Note that the net/http package does StateNew and StateClosed for us.
732 // There is currently no plan for StateHijacked or hijacking HTTP/2 connections.
733 func (sc *serverConn) setConnState(state http.ConnState) {
734 if sc.hs.ConnState != nil {
735 sc.hs.ConnState(sc.conn, state)
736 }
737 }
738 739 func (sc *serverConn) vlogf(format string, args ...interface{}) {
740 if VerboseLogs {
741 sc.logf(format, args...)
742 }
743 }
744 745 func (sc *serverConn) logf(format string, args ...interface{}) {
746 if lg := sc.hs.ErrorLog; lg != nil {
747 lg.Printf(format, args...)
748 } else {
749 log.Printf(format, args...)
750 }
751 }
752 753 // errno returns v's underlying uintptr, else 0.
754 //
755 // TODO: remove this helper function once http2 can use build
756 // tags. See comment in isClosedConnError.
757 func errno(v error) uintptr {
758 if rv := reflect.ValueOf(v); rv.Kind() == reflect.Uintptr {
759 return uintptr(rv.Uint())
760 }
761 return 0
762 }
763 764 // isClosedConnError reports whether err is an error from use of a closed
765 // network connection.
766 func isClosedConnError(err error) bool {
767 if err == nil {
768 return false
769 }
770 771 if errors.Is(err, net.ErrClosed) {
772 return true
773 }
774 775 // TODO(bradfitz): x/tools/cmd/bundle doesn't really support
776 // build tags, so I can't make an http2_windows.go file with
777 // Windows-specific stuff. Fix that and move this, once we
778 // have a way to bundle this into std's net/http somehow.
779 if runtime.GOOS == "windows" {
780 if oe, ok := err.(*net.OpError); ok && oe.Op == "read" {
781 if se, ok := oe.Err.(*os.SyscallError); ok && se.Syscall == "wsarecv" {
782 const WSAECONNABORTED = 10053
783 const WSAECONNRESET = 10054
784 if n := errno(se.Err); n == WSAECONNRESET || n == WSAECONNABORTED {
785 return true
786 }
787 }
788 }
789 }
790 return false
791 }
792 793 func (sc *serverConn) condlogf(err error, format string, args ...interface{}) {
794 if err == nil {
795 return
796 }
797 if err == io.EOF || err == io.ErrUnexpectedEOF || isClosedConnError(err) || err == errPrefaceTimeout {
798 // Boring, expected errors.
799 sc.vlogf(format, args...)
800 } else {
801 sc.logf(format, args...)
802 }
803 }
804 805 // maxCachedCanonicalHeadersKeysSize is an arbitrarily-chosen limit on the size
806 // of the entries in the canonHeader cache.
807 // This should be larger than the size of unique, uncommon header keys likely to
808 // be sent by the peer, while not so high as to permit unreasonable memory usage
809 // if the peer sends an unbounded number of unique header keys.
810 const maxCachedCanonicalHeadersKeysSize = 2048
811 812 func (sc *serverConn) canonicalHeader(v string) string {
813 sc.serveG.check()
814 cv, ok := httpcommon.CachedCanonicalHeader(v)
815 if ok {
816 return cv
817 }
818 cv, ok = sc.canonHeader[v]
819 if ok {
820 return cv
821 }
822 if sc.canonHeader == nil {
823 sc.canonHeader = make(map[string]string)
824 }
825 cv = http.CanonicalHeaderKey(v)
826 size := 100 + len(v)*2 // 100 bytes of map overhead + key + value
827 if sc.canonHeaderKeysSize+size <= maxCachedCanonicalHeadersKeysSize {
828 sc.canonHeader[v] = cv
829 sc.canonHeaderKeysSize += size
830 }
831 return cv
832 }
833 834 type readFrameResult struct {
835 f Frame // valid until readMore is called
836 err error
837 838 // readMore should be called once the consumer no longer needs or
839 // retains f. After readMore, f is invalid and more frames can be
840 // read.
841 readMore func()
842 }
843 844 // readFrames is the loop that reads incoming frames.
845 // It takes care to only read one frame at a time, blocking until the
846 // consumer is done with the frame.
847 // It's run on its own goroutine.
848 func (sc *serverConn) readFrames() {
849 gate := make(chan struct{})
850 gateDone := func() { gate <- struct{}{} }
851 for {
852 f, err := sc.framer.ReadFrame()
853 select {
854 case sc.readFrameCh <- readFrameResult{f, err, gateDone}:
855 case <-sc.doneServing:
856 return
857 }
858 select {
859 case <-gate:
860 case <-sc.doneServing:
861 return
862 }
863 if terminalReadFrameError(err) {
864 return
865 }
866 }
867 }
868 869 // frameWriteResult is the message passed from writeFrameAsync to the serve goroutine.
870 type frameWriteResult struct {
871 _ incomparable
872 wr FrameWriteRequest // what was written (or attempted)
873 err error // result of the writeFrame call
874 }
875 876 // writeFrameAsync runs in its own goroutine and writes a single frame
877 // and then reports when it's done.
878 // At most one goroutine can be running writeFrameAsync at a time per
879 // serverConn.
880 func (sc *serverConn) writeFrameAsync(wr FrameWriteRequest, wd *writeData) {
881 var err error
882 if wd == nil {
883 err = wr.write.writeFrame(sc)
884 } else {
885 err = sc.framer.endWrite()
886 }
887 sc.wroteFrameCh <- frameWriteResult{wr: wr, err: err}
888 }
889 890 func (sc *serverConn) closeAllStreamsOnConnClose() {
891 sc.serveG.check()
892 for _, st := range sc.streams {
893 sc.closeStream(st, errClientDisconnected)
894 }
895 }
896 897 func (sc *serverConn) stopShutdownTimer() {
898 sc.serveG.check()
899 if t := sc.shutdownTimer; t != nil {
900 t.Stop()
901 }
902 }
903 904 func (sc *serverConn) notePanic() {
905 // Note: this is for serverConn.serve panicking, not http.Handler code.
906 if testHookOnPanicMu != nil {
907 testHookOnPanicMu.Lock()
908 defer testHookOnPanicMu.Unlock()
909 }
910 if testHookOnPanic != nil {
911 if e := recover(); e != nil {
912 if testHookOnPanic(sc, e) {
913 panic(e)
914 }
915 }
916 }
917 }
918 919 func (sc *serverConn) serve(conf http2Config) {
920 sc.serveG.check()
921 defer sc.notePanic()
922 defer sc.conn.Close()
923 defer sc.closeAllStreamsOnConnClose()
924 defer sc.stopShutdownTimer()
925 defer close(sc.doneServing) // unblocks handlers trying to send
926 927 if VerboseLogs {
928 sc.vlogf("http2: server connection from %v on %p", sc.conn.RemoteAddr(), sc.hs)
929 }
930 931 settings := writeSettings{
932 {SettingMaxFrameSize, conf.MaxReadFrameSize},
933 {SettingMaxConcurrentStreams, sc.advMaxStreams},
934 {SettingMaxHeaderListSize, sc.maxHeaderListSize()},
935 {SettingHeaderTableSize, conf.MaxDecoderHeaderTableSize},
936 {SettingInitialWindowSize, uint32(sc.initialStreamRecvWindowSize)},
937 }
938 if !disableExtendedConnectProtocol {
939 settings = append(settings, Setting{SettingEnableConnectProtocol, 1})
940 }
941 sc.writeFrame(FrameWriteRequest{
942 write: settings,
943 })
944 sc.unackedSettings++
945 946 // Each connection starts with initialWindowSize inflow tokens.
947 // If a higher value is configured, we add more tokens.
948 if diff := conf.MaxUploadBufferPerConnection - initialWindowSize; diff > 0 {
949 sc.sendWindowUpdate(nil, int(diff))
950 }
951 952 if err := sc.readPreface(); err != nil {
953 sc.condlogf(err, "http2: server: error reading preface from client %v: %v", sc.conn.RemoteAddr(), err)
954 return
955 }
956 // Now that we've got the preface, get us out of the
957 // "StateNew" state. We can't go directly to idle, though.
958 // Active means we read some data and anticipate a request. We'll
959 // do another Active when we get a HEADERS frame.
960 sc.setConnState(http.StateActive)
961 sc.setConnState(http.StateIdle)
962 963 if sc.srv.IdleTimeout > 0 {
964 sc.idleTimer = time.AfterFunc(sc.srv.IdleTimeout, sc.onIdleTimer)
965 defer sc.idleTimer.Stop()
966 }
967 968 if conf.SendPingTimeout > 0 {
969 sc.readIdleTimeout = conf.SendPingTimeout
970 sc.readIdleTimer = time.AfterFunc(conf.SendPingTimeout, sc.onReadIdleTimer)
971 defer sc.readIdleTimer.Stop()
972 }
973 974 go sc.readFrames() // closed by defer sc.conn.Close above
975 976 settingsTimer := time.AfterFunc(firstSettingsTimeout, sc.onSettingsTimer)
977 defer settingsTimer.Stop()
978 979 lastFrameTime := time.Now()
980 loopNum := 0
981 for {
982 loopNum++
983 select {
984 case wr := <-sc.wantWriteFrameCh:
985 if se, ok := wr.write.(StreamError); ok {
986 sc.resetStream(se)
987 break
988 }
989 sc.writeFrame(wr)
990 case res := <-sc.wroteFrameCh:
991 sc.wroteFrame(res)
992 case res := <-sc.readFrameCh:
993 lastFrameTime = time.Now()
994 // Process any written frames before reading new frames from the client since a
995 // written frame could have triggered a new stream to be started.
996 if sc.writingFrameAsync {
997 select {
998 case wroteRes := <-sc.wroteFrameCh:
999 sc.wroteFrame(wroteRes)
1000 default:
1001 }
1002 }
1003 if !sc.processFrameFromReader(res) {
1004 return
1005 }
1006 res.readMore()
1007 if settingsTimer != nil {
1008 settingsTimer.Stop()
1009 settingsTimer = nil
1010 }
1011 case m := <-sc.bodyReadCh:
1012 sc.noteBodyRead(m.st, m.n)
1013 case msg := <-sc.serveMsgCh:
1014 switch v := msg.(type) {
1015 case func(int):
1016 v(loopNum) // for testing
1017 case *serverMessage:
1018 switch v {
1019 case settingsTimerMsg:
1020 sc.logf("timeout waiting for SETTINGS frames from %v", sc.conn.RemoteAddr())
1021 return
1022 case idleTimerMsg:
1023 sc.vlogf("connection is idle")
1024 sc.goAway(ErrCodeNo)
1025 case readIdleTimerMsg:
1026 sc.handlePingTimer(lastFrameTime)
1027 case shutdownTimerMsg:
1028 sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr())
1029 return
1030 case gracefulShutdownMsg:
1031 sc.startGracefulShutdownInternal()
1032 case handlerDoneMsg:
1033 sc.handlerDone()
1034 default:
1035 panic("unknown timer")
1036 }
1037 case *startPushRequest:
1038 sc.startPush(v)
1039 case func(*serverConn):
1040 v(sc)
1041 default:
1042 panic(fmt.Sprintf("unexpected type %T", v))
1043 }
1044 }
1045 1046 // If the peer is causing us to generate a lot of control frames,
1047 // but not reading them from us, assume they are trying to make us
1048 // run out of memory.
1049 if sc.queuedControlFrames > maxQueuedControlFrames {
1050 sc.vlogf("http2: too many control frames in send queue, closing connection")
1051 return
1052 }
1053 1054 // Start the shutdown timer after sending a GOAWAY. When sending GOAWAY
1055 // with no error code (graceful shutdown), don't start the timer until
1056 // all open streams have been completed.
1057 sentGoAway := sc.inGoAway && !sc.needToSendGoAway && !sc.writingFrame
1058 gracefulShutdownComplete := sc.goAwayCode == ErrCodeNo && sc.curOpenStreams() == 0
1059 if sentGoAway && sc.shutdownTimer == nil && (sc.goAwayCode != ErrCodeNo || gracefulShutdownComplete) {
1060 sc.shutDownIn(goAwayTimeout)
1061 }
1062 }
1063 }
1064 1065 func (sc *serverConn) handlePingTimer(lastFrameReadTime time.Time) {
1066 if sc.pingSent {
1067 sc.logf("timeout waiting for PING response")
1068 if f := sc.countErrorFunc; f != nil {
1069 f("conn_close_lost_ping")
1070 }
1071 sc.conn.Close()
1072 return
1073 }
1074 1075 pingAt := lastFrameReadTime.Add(sc.readIdleTimeout)
1076 now := time.Now()
1077 if pingAt.After(now) {
1078 // We received frames since arming the ping timer.
1079 // Reset it for the next possible timeout.
1080 sc.readIdleTimer.Reset(pingAt.Sub(now))
1081 return
1082 }
1083 1084 sc.pingSent = true
1085 // Ignore crypto/rand.Read errors: It generally can't fail, and worse case if it does
1086 // is we send a PING frame containing 0s.
1087 _, _ = rand.Read(sc.sentPingData[:])
1088 sc.writeFrame(FrameWriteRequest{
1089 write: &writePing{data: sc.sentPingData},
1090 })
1091 sc.readIdleTimer.Reset(sc.pingTimeout)
1092 }
1093 1094 type serverMessage int
1095 1096 // Message values sent to serveMsgCh.
1097 var (
1098 settingsTimerMsg = new(serverMessage)
1099 idleTimerMsg = new(serverMessage)
1100 readIdleTimerMsg = new(serverMessage)
1101 shutdownTimerMsg = new(serverMessage)
1102 gracefulShutdownMsg = new(serverMessage)
1103 handlerDoneMsg = new(serverMessage)
1104 )
1105 1106 func (sc *serverConn) onSettingsTimer() { sc.sendServeMsg(settingsTimerMsg) }
1107 func (sc *serverConn) onIdleTimer() { sc.sendServeMsg(idleTimerMsg) }
1108 func (sc *serverConn) onReadIdleTimer() { sc.sendServeMsg(readIdleTimerMsg) }
1109 func (sc *serverConn) onShutdownTimer() { sc.sendServeMsg(shutdownTimerMsg) }
1110 1111 func (sc *serverConn) sendServeMsg(msg interface{}) {
1112 sc.serveG.checkNotOn() // NOT
1113 select {
1114 case sc.serveMsgCh <- msg:
1115 case <-sc.doneServing:
1116 }
1117 }
1118 1119 var errPrefaceTimeout = errors.New("timeout waiting for client preface")
1120 1121 // readPreface reads the ClientPreface greeting from the peer or
1122 // returns errPrefaceTimeout on timeout, or an error if the greeting
1123 // is invalid.
1124 func (sc *serverConn) readPreface() error {
1125 if sc.sawClientPreface {
1126 return nil
1127 }
1128 errc := make(chan error, 1)
1129 go func() {
1130 // Read the client preface
1131 buf := make([]byte, len(ClientPreface))
1132 if _, err := io.ReadFull(sc.conn, buf); err != nil {
1133 errc <- err
1134 } else if !bytes.Equal(buf, clientPreface) {
1135 errc <- fmt.Errorf("bogus greeting %q", buf)
1136 } else {
1137 errc <- nil
1138 }
1139 }()
1140 timer := time.NewTimer(prefaceTimeout) // TODO: configurable on *Server?
1141 defer timer.Stop()
1142 select {
1143 case <-timer.C:
1144 return errPrefaceTimeout
1145 case err := <-errc:
1146 if err == nil {
1147 if VerboseLogs {
1148 sc.vlogf("http2: server: client %v said hello", sc.conn.RemoteAddr())
1149 }
1150 }
1151 return err
1152 }
1153 }
1154 1155 var writeDataPool = sync.Pool{
1156 New: func() interface{} { return new(writeData) },
1157 }
1158 1159 // writeDataFromHandler writes DATA response frames from a handler on
1160 // the given stream.
1161 func (sc *serverConn) writeDataFromHandler(stream *stream, data []byte, endStream bool) error {
1162 ch := sc.srv.state.getErrChan()
1163 writeArg := writeDataPool.Get().(*writeData)
1164 *writeArg = writeData{stream.id, data, endStream}
1165 err := sc.writeFrameFromHandler(FrameWriteRequest{
1166 write: writeArg,
1167 stream: stream,
1168 done: ch,
1169 })
1170 if err != nil {
1171 return err
1172 }
1173 var frameWriteDone bool // the frame write is done (successfully or not)
1174 select {
1175 case err = <-ch:
1176 frameWriteDone = true
1177 case <-sc.doneServing:
1178 return errClientDisconnected
1179 case <-stream.cw:
1180 // If both ch and stream.cw were ready (as might
1181 // happen on the final Write after an http.Handler
1182 // ends), prefer the write result. Otherwise this
1183 // might just be us successfully closing the stream.
1184 // The writeFrameAsync and serve goroutines guarantee
1185 // that the ch send will happen before the stream.cw
1186 // close.
1187 select {
1188 case err = <-ch:
1189 frameWriteDone = true
1190 default:
1191 return errStreamClosed
1192 }
1193 }
1194 sc.srv.state.putErrChan(ch)
1195 if frameWriteDone {
1196 writeDataPool.Put(writeArg)
1197 }
1198 return err
1199 }
1200 1201 // writeFrameFromHandler sends wr to sc.wantWriteFrameCh, but aborts
1202 // if the connection has gone away.
1203 //
1204 // This must not be run from the serve goroutine itself, else it might
1205 // deadlock writing to sc.wantWriteFrameCh (which is only mildly
1206 // buffered and is read by serve itself). If you're on the serve
1207 // goroutine, call writeFrame instead.
1208 func (sc *serverConn) writeFrameFromHandler(wr FrameWriteRequest) error {
1209 sc.serveG.checkNotOn() // NOT
1210 select {
1211 case sc.wantWriteFrameCh <- wr:
1212 return nil
1213 case <-sc.doneServing:
1214 // Serve loop is gone.
1215 // Client has closed their connection to the server.
1216 return errClientDisconnected
1217 }
1218 }
1219 1220 // writeFrame schedules a frame to write and sends it if there's nothing
1221 // already being written.
1222 //
1223 // There is no pushback here (the serve goroutine never blocks). It's
1224 // the http.Handlers that block, waiting for their previous frames to
1225 // make it onto the wire
1226 //
1227 // If you're not on the serve goroutine, use writeFrameFromHandler instead.
1228 func (sc *serverConn) writeFrame(wr FrameWriteRequest) {
1229 sc.serveG.check()
1230 1231 // If true, wr will not be written and wr.done will not be signaled.
1232 var ignoreWrite bool
1233 1234 // We are not allowed to write frames on closed streams. RFC 7540 Section
1235 // 5.1.1 says: "An endpoint MUST NOT send frames other than PRIORITY on
1236 // a closed stream." Our server never sends PRIORITY, so that exception
1237 // does not apply.
1238 //
1239 // The serverConn might close an open stream while the stream's handler
1240 // is still running. For example, the server might close a stream when it
1241 // receives bad data from the client. If this happens, the handler might
1242 // attempt to write a frame after the stream has been closed (since the
1243 // handler hasn't yet been notified of the close). In this case, we simply
1244 // ignore the frame. The handler will notice that the stream is closed when
1245 // it waits for the frame to be written.
1246 //
1247 // As an exception to this rule, we allow sending RST_STREAM after close.
1248 // This allows us to immediately reject new streams without tracking any
1249 // state for those streams (except for the queued RST_STREAM frame). This
1250 // may result in duplicate RST_STREAMs in some cases, but the client should
1251 // ignore those.
1252 if wr.StreamID() != 0 {
1253 _, isReset := wr.write.(StreamError)
1254 if state, _ := sc.state(wr.StreamID()); state == stateClosed && !isReset {
1255 ignoreWrite = true
1256 }
1257 }
1258 1259 // Don't send a 100-continue response if we've already sent headers.
1260 // See golang.org/issue/14030.
1261 switch wr.write.(type) {
1262 case *writeResHeaders:
1263 wr.stream.wroteHeaders = true
1264 case write100ContinueHeadersFrame:
1265 if wr.stream.wroteHeaders {
1266 // We do not need to notify wr.done because this frame is
1267 // never written with wr.done != nil.
1268 if wr.done != nil {
1269 panic("wr.done != nil for write100ContinueHeadersFrame")
1270 }
1271 ignoreWrite = true
1272 }
1273 }
1274 1275 if !ignoreWrite {
1276 if wr.isControl() {
1277 sc.queuedControlFrames++
1278 // For extra safety, detect wraparounds, which should not happen,
1279 // and pull the plug.
1280 if sc.queuedControlFrames < 0 {
1281 sc.conn.Close()
1282 }
1283 }
1284 sc.writeSched.Push(wr)
1285 }
1286 sc.scheduleFrameWrite()
1287 }
1288 1289 // startFrameWrite starts a goroutine to write wr (in a separate
1290 // goroutine since that might block on the network), and updates the
1291 // serve goroutine's state about the world, updated from info in wr.
1292 func (sc *serverConn) startFrameWrite(wr FrameWriteRequest) {
1293 sc.serveG.check()
1294 if sc.writingFrame {
1295 panic("internal error: can only be writing one frame at a time")
1296 }
1297 1298 st := wr.stream
1299 if st != nil {
1300 switch st.state {
1301 case stateHalfClosedLocal:
1302 switch wr.write.(type) {
1303 case StreamError, handlerPanicRST, writeWindowUpdate:
1304 // RFC 7540 Section 5.1 allows sending RST_STREAM, PRIORITY, and WINDOW_UPDATE
1305 // in this state. (We never send PRIORITY from the server, so that is not checked.)
1306 default:
1307 panic(fmt.Sprintf("internal error: attempt to send frame on a half-closed-local stream: %v", wr))
1308 }
1309 case stateClosed:
1310 panic(fmt.Sprintf("internal error: attempt to send frame on a closed stream: %v", wr))
1311 }
1312 }
1313 if wpp, ok := wr.write.(*writePushPromise); ok {
1314 var err error
1315 wpp.promisedID, err = wpp.allocatePromisedID()
1316 if err != nil {
1317 sc.writingFrameAsync = false
1318 wr.replyToWriter(err)
1319 return
1320 }
1321 }
1322 1323 sc.writingFrame = true
1324 sc.needsFrameFlush = true
1325 if wr.write.staysWithinBuffer(sc.bw.Available()) {
1326 sc.writingFrameAsync = false
1327 err := wr.write.writeFrame(sc)
1328 sc.wroteFrame(frameWriteResult{wr: wr, err: err})
1329 } else if wd, ok := wr.write.(*writeData); ok {
1330 // Encode the frame in the serve goroutine, to ensure we don't have
1331 // any lingering asynchronous references to data passed to Write.
1332 // See https://go.dev/issue/58446.
1333 sc.framer.startWriteDataPadded(wd.streamID, wd.endStream, wd.p, nil)
1334 sc.writingFrameAsync = true
1335 go sc.writeFrameAsync(wr, wd)
1336 } else {
1337 sc.writingFrameAsync = true
1338 go sc.writeFrameAsync(wr, nil)
1339 }
1340 }
1341 1342 // errHandlerPanicked is the error given to any callers blocked in a read from
1343 // Request.Body when the main goroutine panics. Since most handlers read in the
1344 // main ServeHTTP goroutine, this will show up rarely.
1345 var errHandlerPanicked = errors.New("http2: handler panicked")
1346 1347 // wroteFrame is called on the serve goroutine with the result of
1348 // whatever happened on writeFrameAsync.
1349 func (sc *serverConn) wroteFrame(res frameWriteResult) {
1350 sc.serveG.check()
1351 if !sc.writingFrame {
1352 panic("internal error: expected to be already writing a frame")
1353 }
1354 sc.writingFrame = false
1355 sc.writingFrameAsync = false
1356 1357 if res.err != nil {
1358 sc.conn.Close()
1359 }
1360 1361 wr := res.wr
1362 1363 if writeEndsStream(wr.write) {
1364 st := wr.stream
1365 if st == nil {
1366 panic("internal error: expecting non-nil stream")
1367 }
1368 switch st.state {
1369 case stateOpen:
1370 // Here we would go to stateHalfClosedLocal in
1371 // theory, but since our handler is done and
1372 // the net/http package provides no mechanism
1373 // for closing a ResponseWriter while still
1374 // reading data (see possible TODO at top of
1375 // this file), we go into closed state here
1376 // anyway, after telling the peer we're
1377 // hanging up on them. We'll transition to
1378 // stateClosed after the RST_STREAM frame is
1379 // written.
1380 st.state = stateHalfClosedLocal
1381 // Section 8.1: a server MAY request that the client abort
1382 // transmission of a request without error by sending a
1383 // RST_STREAM with an error code of NO_ERROR after sending
1384 // a complete response.
1385 sc.resetStream(streamError(st.id, ErrCodeNo))
1386 case stateHalfClosedRemote:
1387 sc.closeStream(st, errHandlerComplete)
1388 }
1389 } else {
1390 switch v := wr.write.(type) {
1391 case StreamError:
1392 // st may be unknown if the RST_STREAM was generated to reject bad input.
1393 if st, ok := sc.streams[v.StreamID]; ok {
1394 sc.closeStream(st, v)
1395 }
1396 case handlerPanicRST:
1397 sc.closeStream(wr.stream, errHandlerPanicked)
1398 }
1399 }
1400 1401 // Reply (if requested) to unblock the ServeHTTP goroutine.
1402 wr.replyToWriter(res.err)
1403 1404 sc.scheduleFrameWrite()
1405 }
1406 1407 // scheduleFrameWrite tickles the frame writing scheduler.
1408 //
1409 // If a frame is already being written, nothing happens. This will be called again
1410 // when the frame is done being written.
1411 //
1412 // If a frame isn't being written and we need to send one, the best frame
1413 // to send is selected by writeSched.
1414 //
1415 // If a frame isn't being written and there's nothing else to send, we
1416 // flush the write buffer.
1417 func (sc *serverConn) scheduleFrameWrite() {
1418 sc.serveG.check()
1419 if sc.writingFrame || sc.inFrameScheduleLoop {
1420 return
1421 }
1422 sc.inFrameScheduleLoop = true
1423 for !sc.writingFrameAsync {
1424 if sc.needToSendGoAway {
1425 sc.needToSendGoAway = false
1426 sc.startFrameWrite(FrameWriteRequest{
1427 write: &writeGoAway{
1428 maxStreamID: sc.maxClientStreamID,
1429 code: sc.goAwayCode,
1430 },
1431 })
1432 continue
1433 }
1434 if sc.needToSendSettingsAck {
1435 sc.needToSendSettingsAck = false
1436 sc.startFrameWrite(FrameWriteRequest{write: writeSettingsAck{}})
1437 continue
1438 }
1439 if !sc.inGoAway || sc.goAwayCode == ErrCodeNo {
1440 if wr, ok := sc.writeSched.Pop(); ok {
1441 if wr.isControl() {
1442 sc.queuedControlFrames--
1443 }
1444 sc.startFrameWrite(wr)
1445 continue
1446 }
1447 }
1448 if sc.needsFrameFlush {
1449 sc.startFrameWrite(FrameWriteRequest{write: flushFrameWriter{}})
1450 sc.needsFrameFlush = false // after startFrameWrite, since it sets this true
1451 continue
1452 }
1453 break
1454 }
1455 sc.inFrameScheduleLoop = false
1456 }
1457 1458 // startGracefulShutdown gracefully shuts down a connection. This
1459 // sends GOAWAY with ErrCodeNo to tell the client we're gracefully
1460 // shutting down. The connection isn't closed until all current
1461 // streams are done.
1462 //
1463 // startGracefulShutdown returns immediately; it does not wait until
1464 // the connection has shut down.
1465 func (sc *serverConn) startGracefulShutdown() {
1466 sc.serveG.checkNotOn() // NOT
1467 sc.shutdownOnce.Do(func() { sc.sendServeMsg(gracefulShutdownMsg) })
1468 }
1469 1470 // After sending GOAWAY with an error code (non-graceful shutdown), the
1471 // connection will close after goAwayTimeout.
1472 //
1473 // If we close the connection immediately after sending GOAWAY, there may
1474 // be unsent data in our kernel receive buffer, which will cause the kernel
1475 // to send a TCP RST on close() instead of a FIN. This RST will abort the
1476 // connection immediately, whether or not the client had received the GOAWAY.
1477 //
1478 // Ideally we should delay for at least 1 RTT + epsilon so the client has
1479 // a chance to read the GOAWAY and stop sending messages. Measuring RTT
1480 // is hard, so we approximate with 1 second. See golang.org/issue/18701.
1481 //
1482 // This is a var so it can be shorter in tests, where all requests uses the
1483 // loopback interface making the expected RTT very small.
1484 //
1485 // TODO: configurable?
1486 var goAwayTimeout = 1 * time.Second
1487 1488 func (sc *serverConn) startGracefulShutdownInternal() {
1489 sc.goAway(ErrCodeNo)
1490 }
1491 1492 func (sc *serverConn) goAway(code ErrCode) {
1493 sc.serveG.check()
1494 if sc.inGoAway {
1495 if sc.goAwayCode == ErrCodeNo {
1496 sc.goAwayCode = code
1497 }
1498 return
1499 }
1500 sc.inGoAway = true
1501 sc.needToSendGoAway = true
1502 sc.goAwayCode = code
1503 sc.scheduleFrameWrite()
1504 }
1505 1506 func (sc *serverConn) shutDownIn(d time.Duration) {
1507 sc.serveG.check()
1508 sc.shutdownTimer = time.AfterFunc(d, sc.onShutdownTimer)
1509 }
1510 1511 func (sc *serverConn) resetStream(se StreamError) {
1512 sc.serveG.check()
1513 sc.writeFrame(FrameWriteRequest{write: se})
1514 if st, ok := sc.streams[se.StreamID]; ok {
1515 st.resetQueued = true
1516 }
1517 }
1518 1519 // processFrameFromReader processes the serve loop's read from readFrameCh from the
1520 // frame-reading goroutine.
1521 // processFrameFromReader returns whether the connection should be kept open.
1522 func (sc *serverConn) processFrameFromReader(res readFrameResult) bool {
1523 sc.serveG.check()
1524 err := res.err
1525 if err != nil {
1526 if err == ErrFrameTooLarge {
1527 sc.goAway(ErrCodeFrameSize)
1528 return true // goAway will close the loop
1529 }
1530 clientGone := err == io.EOF || err == io.ErrUnexpectedEOF || isClosedConnError(err)
1531 if clientGone {
1532 // TODO: could we also get into this state if
1533 // the peer does a half close
1534 // (e.g. CloseWrite) because they're done
1535 // sending frames but they're still wanting
1536 // our open replies? Investigate.
1537 // TODO: add CloseWrite to crypto/tls.Conn first
1538 // so we have a way to test this? I suppose
1539 // just for testing we could have a non-TLS mode.
1540 return false
1541 }
1542 } else {
1543 f := res.f
1544 if VerboseLogs {
1545 sc.vlogf("http2: server read frame %v", summarizeFrame(f))
1546 }
1547 err = sc.processFrame(f)
1548 if err == nil {
1549 return true
1550 }
1551 }
1552 1553 switch ev := err.(type) {
1554 case StreamError:
1555 sc.resetStream(ev)
1556 return true
1557 case goAwayFlowError:
1558 sc.goAway(ErrCodeFlowControl)
1559 return true
1560 case ConnectionError:
1561 if res.f != nil {
1562 if id := res.f.Header().StreamID; id > sc.maxClientStreamID {
1563 sc.maxClientStreamID = id
1564 }
1565 }
1566 sc.logf("http2: server connection error from %v: %v", sc.conn.RemoteAddr(), ev)
1567 sc.goAway(ErrCode(ev))
1568 return true // goAway will handle shutdown
1569 default:
1570 if res.err != nil {
1571 sc.vlogf("http2: server closing client connection; error reading frame from client %s: %v", sc.conn.RemoteAddr(), err)
1572 } else {
1573 sc.logf("http2: server closing client connection: %v", err)
1574 }
1575 return false
1576 }
1577 }
1578 1579 func (sc *serverConn) processFrame(f Frame) error {
1580 sc.serveG.check()
1581 1582 // First frame received must be SETTINGS.
1583 if !sc.sawFirstSettings {
1584 if _, ok := f.(*SettingsFrame); !ok {
1585 return sc.countError("first_settings", ConnectionError(ErrCodeProtocol))
1586 }
1587 sc.sawFirstSettings = true
1588 }
1589 1590 // Discard frames for streams initiated after the identified last
1591 // stream sent in a GOAWAY, or all frames after sending an error.
1592 // We still need to return connection-level flow control for DATA frames.
1593 // RFC 9113 Section 6.8.
1594 if sc.inGoAway && (sc.goAwayCode != ErrCodeNo || f.Header().StreamID > sc.maxClientStreamID) {
1595 1596 if f, ok := f.(*DataFrame); ok {
1597 if !sc.inflow.take(f.Length) {
1598 return sc.countError("data_flow", streamError(f.Header().StreamID, ErrCodeFlowControl))
1599 }
1600 sc.sendWindowUpdate(nil, int(f.Length)) // conn-level
1601 }
1602 return nil
1603 }
1604 1605 switch f := f.(type) {
1606 case *SettingsFrame:
1607 return sc.processSettings(f)
1608 case *MetaHeadersFrame:
1609 return sc.processHeaders(f)
1610 case *WindowUpdateFrame:
1611 return sc.processWindowUpdate(f)
1612 case *PingFrame:
1613 return sc.processPing(f)
1614 case *DataFrame:
1615 return sc.processData(f)
1616 case *RSTStreamFrame:
1617 return sc.processResetStream(f)
1618 case *PriorityFrame:
1619 return sc.processPriority(f)
1620 case *GoAwayFrame:
1621 return sc.processGoAway(f)
1622 case *PushPromiseFrame:
1623 // A client cannot push. Thus, servers MUST treat the receipt of a PUSH_PROMISE
1624 // frame as a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
1625 return sc.countError("push_promise", ConnectionError(ErrCodeProtocol))
1626 default:
1627 sc.vlogf("http2: server ignoring frame: %v", f.Header())
1628 return nil
1629 }
1630 }
1631 1632 func (sc *serverConn) processPing(f *PingFrame) error {
1633 sc.serveG.check()
1634 if f.IsAck() {
1635 if sc.pingSent && sc.sentPingData == f.Data {
1636 // This is a response to a PING we sent.
1637 sc.pingSent = false
1638 sc.readIdleTimer.Reset(sc.readIdleTimeout)
1639 }
1640 // 6.7 PING: " An endpoint MUST NOT respond to PING frames
1641 // containing this flag."
1642 return nil
1643 }
1644 if f.StreamID != 0 {
1645 // "PING frames are not associated with any individual
1646 // stream. If a PING frame is received with a stream
1647 // identifier field value other than 0x0, the recipient MUST
1648 // respond with a connection error (Section 5.4.1) of type
1649 // PROTOCOL_ERROR."
1650 return sc.countError("ping_on_stream", ConnectionError(ErrCodeProtocol))
1651 }
1652 sc.writeFrame(FrameWriteRequest{write: writePingAck{f}})
1653 return nil
1654 }
1655 1656 func (sc *serverConn) processWindowUpdate(f *WindowUpdateFrame) error {
1657 sc.serveG.check()
1658 switch {
1659 case f.StreamID != 0: // stream-level flow control
1660 state, st := sc.state(f.StreamID)
1661 if state == stateIdle {
1662 // Section 5.1: "Receiving any frame other than HEADERS
1663 // or PRIORITY on a stream in this state MUST be
1664 // treated as a connection error (Section 5.4.1) of
1665 // type PROTOCOL_ERROR."
1666 return sc.countError("stream_idle", ConnectionError(ErrCodeProtocol))
1667 }
1668 if st == nil {
1669 // "WINDOW_UPDATE can be sent by a peer that has sent a
1670 // frame bearing the END_STREAM flag. This means that a
1671 // receiver could receive a WINDOW_UPDATE frame on a "half
1672 // closed (remote)" or "closed" stream. A receiver MUST
1673 // NOT treat this as an error, see Section 5.1."
1674 return nil
1675 }
1676 if !st.flow.add(int32(f.Increment)) {
1677 return sc.countError("bad_flow", streamError(f.StreamID, ErrCodeFlowControl))
1678 }
1679 default: // connection-level flow control
1680 if !sc.flow.add(int32(f.Increment)) {
1681 return goAwayFlowError{}
1682 }
1683 }
1684 sc.scheduleFrameWrite()
1685 return nil
1686 }
1687 1688 func (sc *serverConn) processResetStream(f *RSTStreamFrame) error {
1689 sc.serveG.check()
1690 1691 state, st := sc.state(f.StreamID)
1692 if state == stateIdle {
1693 // 6.4 "RST_STREAM frames MUST NOT be sent for a
1694 // stream in the "idle" state. If a RST_STREAM frame
1695 // identifying an idle stream is received, the
1696 // recipient MUST treat this as a connection error
1697 // (Section 5.4.1) of type PROTOCOL_ERROR.
1698 return sc.countError("reset_idle_stream", ConnectionError(ErrCodeProtocol))
1699 }
1700 if st != nil {
1701 st.cancelCtx()
1702 sc.closeStream(st, streamError(f.StreamID, f.ErrCode))
1703 }
1704 return nil
1705 }
1706 1707 func (sc *serverConn) closeStream(st *stream, err error) {
1708 sc.serveG.check()
1709 if st.state == stateIdle || st.state == stateClosed {
1710 panic(fmt.Sprintf("invariant; can't close stream in state %v", st.state))
1711 }
1712 st.state = stateClosed
1713 if st.readDeadline != nil {
1714 st.readDeadline.Stop()
1715 }
1716 if st.writeDeadline != nil {
1717 st.writeDeadline.Stop()
1718 }
1719 if st.isPushed() {
1720 sc.curPushedStreams--
1721 } else {
1722 sc.curClientStreams--
1723 }
1724 delete(sc.streams, st.id)
1725 if len(sc.streams) == 0 {
1726 sc.setConnState(http.StateIdle)
1727 if sc.srv.IdleTimeout > 0 && sc.idleTimer != nil {
1728 sc.idleTimer.Reset(sc.srv.IdleTimeout)
1729 }
1730 if h1ServerKeepAlivesDisabled(sc.hs) {
1731 sc.startGracefulShutdownInternal()
1732 }
1733 }
1734 if p := st.body; p != nil {
1735 // Return any buffered unread bytes worth of conn-level flow control.
1736 // See golang.org/issue/16481
1737 sc.sendWindowUpdate(nil, p.Len())
1738 1739 p.CloseWithError(err)
1740 }
1741 if e, ok := err.(StreamError); ok {
1742 if e.Cause != nil {
1743 err = e.Cause
1744 } else {
1745 err = errStreamClosed
1746 }
1747 }
1748 st.closeErr = err
1749 st.cancelCtx()
1750 st.cw.Close() // signals Handler's CloseNotifier, unblocks writes, etc
1751 sc.writeSched.CloseStream(st.id)
1752 }
1753 1754 func (sc *serverConn) processSettings(f *SettingsFrame) error {
1755 sc.serveG.check()
1756 if f.IsAck() {
1757 sc.unackedSettings--
1758 if sc.unackedSettings < 0 {
1759 // Why is the peer ACKing settings we never sent?
1760 // The spec doesn't mention this case, but
1761 // hang up on them anyway.
1762 return sc.countError("ack_mystery", ConnectionError(ErrCodeProtocol))
1763 }
1764 return nil
1765 }
1766 if f.NumSettings() > 100 || f.HasDuplicates() {
1767 // This isn't actually in the spec, but hang up on
1768 // suspiciously large settings frames or those with
1769 // duplicate entries.
1770 return sc.countError("settings_big_or_dups", ConnectionError(ErrCodeProtocol))
1771 }
1772 if err := f.ForeachSetting(sc.processSetting); err != nil {
1773 return err
1774 }
1775 // TODO: judging by RFC 7540, Section 6.5.3 each SETTINGS frame should be
1776 // acknowledged individually, even if multiple are received before the ACK.
1777 sc.needToSendSettingsAck = true
1778 sc.scheduleFrameWrite()
1779 return nil
1780 }
1781 1782 func (sc *serverConn) processSetting(s Setting) error {
1783 sc.serveG.check()
1784 if err := s.Valid(); err != nil {
1785 return err
1786 }
1787 if VerboseLogs {
1788 sc.vlogf("http2: server processing setting %v", s)
1789 }
1790 switch s.ID {
1791 case SettingHeaderTableSize:
1792 sc.hpackEncoder.SetMaxDynamicTableSize(s.Val)
1793 case SettingEnablePush:
1794 sc.pushEnabled = s.Val != 0
1795 case SettingMaxConcurrentStreams:
1796 sc.clientMaxStreams = s.Val
1797 case SettingInitialWindowSize:
1798 return sc.processSettingInitialWindowSize(s.Val)
1799 case SettingMaxFrameSize:
1800 sc.maxFrameSize = int32(s.Val) // the maximum valid s.Val is < 2^31
1801 case SettingMaxHeaderListSize:
1802 sc.peerMaxHeaderListSize = s.Val
1803 case SettingEnableConnectProtocol:
1804 // Receipt of this parameter by a server does not
1805 // have any impact
1806 default:
1807 // Unknown setting: "An endpoint that receives a SETTINGS
1808 // frame with any unknown or unsupported identifier MUST
1809 // ignore that setting."
1810 if VerboseLogs {
1811 sc.vlogf("http2: server ignoring unknown setting %v", s)
1812 }
1813 }
1814 return nil
1815 }
1816 1817 func (sc *serverConn) processSettingInitialWindowSize(val uint32) error {
1818 sc.serveG.check()
1819 // Note: val already validated to be within range by
1820 // processSetting's Valid call.
1821 1822 // "A SETTINGS frame can alter the initial flow control window
1823 // size for all current streams. When the value of
1824 // SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST
1825 // adjust the size of all stream flow control windows that it
1826 // maintains by the difference between the new value and the
1827 // old value."
1828 old := sc.initialStreamSendWindowSize
1829 sc.initialStreamSendWindowSize = int32(val)
1830 growth := int32(val) - old // may be negative
1831 for _, st := range sc.streams {
1832 if !st.flow.add(growth) {
1833 // 6.9.2 Initial Flow Control Window Size
1834 // "An endpoint MUST treat a change to
1835 // SETTINGS_INITIAL_WINDOW_SIZE that causes any flow
1836 // control window to exceed the maximum size as a
1837 // connection error (Section 5.4.1) of type
1838 // FLOW_CONTROL_ERROR."
1839 return sc.countError("setting_win_size", ConnectionError(ErrCodeFlowControl))
1840 }
1841 }
1842 return nil
1843 }
1844 1845 func (sc *serverConn) processData(f *DataFrame) error {
1846 sc.serveG.check()
1847 id := f.Header().StreamID
1848 1849 data := f.Data()
1850 state, st := sc.state(id)
1851 if id == 0 || state == stateIdle {
1852 // Section 6.1: "DATA frames MUST be associated with a
1853 // stream. If a DATA frame is received whose stream
1854 // identifier field is 0x0, the recipient MUST respond
1855 // with a connection error (Section 5.4.1) of type
1856 // PROTOCOL_ERROR."
1857 //
1858 // Section 5.1: "Receiving any frame other than HEADERS
1859 // or PRIORITY on a stream in this state MUST be
1860 // treated as a connection error (Section 5.4.1) of
1861 // type PROTOCOL_ERROR."
1862 return sc.countError("data_on_idle", ConnectionError(ErrCodeProtocol))
1863 }
1864 1865 // "If a DATA frame is received whose stream is not in "open"
1866 // or "half closed (local)" state, the recipient MUST respond
1867 // with a stream error (Section 5.4.2) of type STREAM_CLOSED."
1868 if st == nil || state != stateOpen || st.gotTrailerHeader || st.resetQueued {
1869 // This includes sending a RST_STREAM if the stream is
1870 // in stateHalfClosedLocal (which currently means that
1871 // the http.Handler returned, so it's done reading &
1872 // done writing). Try to stop the client from sending
1873 // more DATA.
1874 1875 // But still enforce their connection-level flow control,
1876 // and return any flow control bytes since we're not going
1877 // to consume them.
1878 if !sc.inflow.take(f.Length) {
1879 return sc.countError("data_flow", streamError(id, ErrCodeFlowControl))
1880 }
1881 sc.sendWindowUpdate(nil, int(f.Length)) // conn-level
1882 1883 if st != nil && st.resetQueued {
1884 // Already have a stream error in flight. Don't send another.
1885 return nil
1886 }
1887 return sc.countError("closed", streamError(id, ErrCodeStreamClosed))
1888 }
1889 if st.body == nil {
1890 panic("internal error: should have a body in this state")
1891 }
1892 1893 // Sender sending more than they'd declared?
1894 if st.declBodyBytes != -1 && st.bodyBytes+int64(len(data)) > st.declBodyBytes {
1895 if !sc.inflow.take(f.Length) {
1896 return sc.countError("data_flow", streamError(id, ErrCodeFlowControl))
1897 }
1898 sc.sendWindowUpdate(nil, int(f.Length)) // conn-level
1899 1900 st.body.CloseWithError(fmt.Errorf("sender tried to send more than declared Content-Length of %d bytes", st.declBodyBytes))
1901 // RFC 7540, sec 8.1.2.6: A request or response is also malformed if the
1902 // value of a content-length header field does not equal the sum of the
1903 // DATA frame payload lengths that form the body.
1904 return sc.countError("send_too_much", streamError(id, ErrCodeProtocol))
1905 }
1906 if f.Length > 0 {
1907 // Check whether the client has flow control quota.
1908 if !takeInflows(&sc.inflow, &st.inflow, f.Length) {
1909 return sc.countError("flow_on_data_length", streamError(id, ErrCodeFlowControl))
1910 }
1911 1912 if len(data) > 0 {
1913 st.bodyBytes += int64(len(data))
1914 wrote, err := st.body.Write(data)
1915 if err != nil {
1916 // The handler has closed the request body.
1917 // Return the connection-level flow control for the discarded data,
1918 // but not the stream-level flow control.
1919 sc.sendWindowUpdate(nil, int(f.Length)-wrote)
1920 return nil
1921 }
1922 if wrote != len(data) {
1923 panic("internal error: bad Writer")
1924 }
1925 }
1926 1927 // Return any padded flow control now, since we won't
1928 // refund it later on body reads.
1929 // Call sendWindowUpdate even if there is no padding,
1930 // to return buffered flow control credit if the sent
1931 // window has shrunk.
1932 pad := int32(f.Length) - int32(len(data))
1933 sc.sendWindowUpdate32(nil, pad)
1934 sc.sendWindowUpdate32(st, pad)
1935 }
1936 if f.StreamEnded() {
1937 st.endStream()
1938 }
1939 return nil
1940 }
1941 1942 func (sc *serverConn) processGoAway(f *GoAwayFrame) error {
1943 sc.serveG.check()
1944 if f.ErrCode != ErrCodeNo {
1945 sc.logf("http2: received GOAWAY %+v, starting graceful shutdown", f)
1946 } else {
1947 sc.vlogf("http2: received GOAWAY %+v, starting graceful shutdown", f)
1948 }
1949 sc.startGracefulShutdownInternal()
1950 // http://tools.ietf.org/html/rfc7540#section-6.8
1951 // We should not create any new streams, which means we should disable push.
1952 sc.pushEnabled = false
1953 return nil
1954 }
1955 1956 // isPushed reports whether the stream is server-initiated.
1957 func (st *stream) isPushed() bool {
1958 return st.id%2 == 0
1959 }
1960 1961 // endStream closes a Request.Body's pipe. It is called when a DATA
1962 // frame says a request body is over (or after trailers).
1963 func (st *stream) endStream() {
1964 sc := st.sc
1965 sc.serveG.check()
1966 1967 if st.declBodyBytes != -1 && st.declBodyBytes != st.bodyBytes {
1968 st.body.CloseWithError(fmt.Errorf("request declared a Content-Length of %d but only wrote %d bytes",
1969 st.declBodyBytes, st.bodyBytes))
1970 } else {
1971 st.body.closeWithErrorAndCode(io.EOF, st.copyTrailersToHandlerRequest)
1972 st.body.CloseWithError(io.EOF)
1973 }
1974 st.state = stateHalfClosedRemote
1975 }
1976 1977 // copyTrailersToHandlerRequest is run in the Handler's goroutine in
1978 // its Request.Body.Read just before it gets io.EOF.
1979 func (st *stream) copyTrailersToHandlerRequest() {
1980 for k, vv := range st.trailer {
1981 if _, ok := st.reqTrailer[k]; ok {
1982 // Only copy it over it was pre-declared.
1983 st.reqTrailer[k] = vv
1984 }
1985 }
1986 }
1987 1988 // onReadTimeout is run on its own goroutine (from time.AfterFunc)
1989 // when the stream's ReadTimeout has fired.
1990 func (st *stream) onReadTimeout() {
1991 if st.body != nil {
1992 // Wrap the ErrDeadlineExceeded to avoid callers depending on us
1993 // returning the bare error.
1994 st.body.CloseWithError(fmt.Errorf("%w", os.ErrDeadlineExceeded))
1995 }
1996 }
1997 1998 // onWriteTimeout is run on its own goroutine (from time.AfterFunc)
1999 // when the stream's WriteTimeout has fired.
2000 func (st *stream) onWriteTimeout() {
2001 st.sc.writeFrameFromHandler(FrameWriteRequest{write: StreamError{
2002 StreamID: st.id,
2003 Code: ErrCodeInternal,
2004 Cause: os.ErrDeadlineExceeded,
2005 }})
2006 }
2007 2008 func (sc *serverConn) processHeaders(f *MetaHeadersFrame) error {
2009 sc.serveG.check()
2010 id := f.StreamID
2011 // http://tools.ietf.org/html/rfc7540#section-5.1.1
2012 // Streams initiated by a client MUST use odd-numbered stream
2013 // identifiers. [...] An endpoint that receives an unexpected
2014 // stream identifier MUST respond with a connection error
2015 // (Section 5.4.1) of type PROTOCOL_ERROR.
2016 if id%2 != 1 {
2017 return sc.countError("headers_even", ConnectionError(ErrCodeProtocol))
2018 }
2019 // A HEADERS frame can be used to create a new stream or
2020 // send a trailer for an open one. If we already have a stream
2021 // open, let it process its own HEADERS frame (trailers at this
2022 // point, if it's valid).
2023 if st := sc.streams[f.StreamID]; st != nil {
2024 if st.resetQueued {
2025 // We're sending RST_STREAM to close the stream, so don't bother
2026 // processing this frame.
2027 return nil
2028 }
2029 // RFC 7540, sec 5.1: If an endpoint receives additional frames, other than
2030 // WINDOW_UPDATE, PRIORITY, or RST_STREAM, for a stream that is in
2031 // this state, it MUST respond with a stream error (Section 5.4.2) of
2032 // type STREAM_CLOSED.
2033 if st.state == stateHalfClosedRemote {
2034 return sc.countError("headers_half_closed", streamError(id, ErrCodeStreamClosed))
2035 }
2036 return st.processTrailerHeaders(f)
2037 }
2038 2039 // [...] The identifier of a newly established stream MUST be
2040 // numerically greater than all streams that the initiating
2041 // endpoint has opened or reserved. [...] An endpoint that
2042 // receives an unexpected stream identifier MUST respond with
2043 // a connection error (Section 5.4.1) of type PROTOCOL_ERROR.
2044 if id <= sc.maxClientStreamID {
2045 return sc.countError("stream_went_down", ConnectionError(ErrCodeProtocol))
2046 }
2047 sc.maxClientStreamID = id
2048 2049 if sc.idleTimer != nil {
2050 sc.idleTimer.Stop()
2051 }
2052 2053 // http://tools.ietf.org/html/rfc7540#section-5.1.2
2054 // [...] Endpoints MUST NOT exceed the limit set by their peer. An
2055 // endpoint that receives a HEADERS frame that causes their
2056 // advertised concurrent stream limit to be exceeded MUST treat
2057 // this as a stream error (Section 5.4.2) of type PROTOCOL_ERROR
2058 // or REFUSED_STREAM.
2059 if sc.curClientStreams+1 > sc.advMaxStreams {
2060 if sc.unackedSettings == 0 {
2061 // They should know better.
2062 return sc.countError("over_max_streams", streamError(id, ErrCodeProtocol))
2063 }
2064 // Assume it's a network race, where they just haven't
2065 // received our last SETTINGS update. But actually
2066 // this can't happen yet, because we don't yet provide
2067 // a way for users to adjust server parameters at
2068 // runtime.
2069 return sc.countError("over_max_streams_race", streamError(id, ErrCodeRefusedStream))
2070 }
2071 2072 initialState := stateOpen
2073 if f.StreamEnded() {
2074 initialState = stateHalfClosedRemote
2075 }
2076 st := sc.newStream(id, 0, initialState)
2077 2078 if f.HasPriority() {
2079 if err := sc.checkPriority(f.StreamID, f.Priority); err != nil {
2080 return err
2081 }
2082 sc.writeSched.AdjustStream(st.id, f.Priority)
2083 }
2084 2085 rw, req, err := sc.newWriterAndRequest(st, f)
2086 if err != nil {
2087 return err
2088 }
2089 st.reqTrailer = req.Trailer
2090 if st.reqTrailer != nil {
2091 st.trailer = make(http.Header)
2092 }
2093 st.body = req.Body.(*requestBody).pipe // may be nil
2094 st.declBodyBytes = req.ContentLength
2095 2096 handler := sc.handler.ServeHTTP
2097 if f.Truncated {
2098 // Their header list was too long. Send a 431 error.
2099 handler = handleHeaderListTooLong
2100 } else if err := checkValidHTTP2RequestHeaders(req.Header); err != nil {
2101 handler = new400Handler(err)
2102 }
2103 2104 // The net/http package sets the read deadline from the
2105 // http.Server.ReadTimeout during the TLS handshake, but then
2106 // passes the connection off to us with the deadline already
2107 // set. Disarm it here after the request headers are read,
2108 // similar to how the http1 server works. Here it's
2109 // technically more like the http1 Server's ReadHeaderTimeout
2110 // (in Go 1.8), though. That's a more sane option anyway.
2111 if sc.hs.ReadTimeout > 0 {
2112 sc.conn.SetReadDeadline(time.Time{})
2113 st.readDeadline = time.AfterFunc(sc.hs.ReadTimeout, st.onReadTimeout)
2114 }
2115 2116 return sc.scheduleHandler(id, rw, req, handler)
2117 }
2118 2119 func (sc *serverConn) upgradeRequest(req *http.Request) {
2120 sc.serveG.check()
2121 id := uint32(1)
2122 sc.maxClientStreamID = id
2123 st := sc.newStream(id, 0, stateHalfClosedRemote)
2124 st.reqTrailer = req.Trailer
2125 if st.reqTrailer != nil {
2126 st.trailer = make(http.Header)
2127 }
2128 rw := sc.newResponseWriter(st, req)
2129 2130 // Disable any read deadline set by the net/http package
2131 // prior to the upgrade.
2132 if sc.hs.ReadTimeout > 0 {
2133 sc.conn.SetReadDeadline(time.Time{})
2134 }
2135 2136 // This is the first request on the connection,
2137 // so start the handler directly rather than going
2138 // through scheduleHandler.
2139 sc.curHandlers++
2140 go sc.runHandler(rw, req, sc.handler.ServeHTTP)
2141 }
2142 2143 func (st *stream) processTrailerHeaders(f *MetaHeadersFrame) error {
2144 sc := st.sc
2145 sc.serveG.check()
2146 if st.gotTrailerHeader {
2147 return sc.countError("dup_trailers", ConnectionError(ErrCodeProtocol))
2148 }
2149 st.gotTrailerHeader = true
2150 if !f.StreamEnded() {
2151 return sc.countError("trailers_not_ended", streamError(st.id, ErrCodeProtocol))
2152 }
2153 2154 if len(f.PseudoFields()) > 0 {
2155 return sc.countError("trailers_pseudo", streamError(st.id, ErrCodeProtocol))
2156 }
2157 if st.trailer != nil {
2158 for _, hf := range f.RegularFields() {
2159 key := sc.canonicalHeader(hf.Name)
2160 if !httpguts.ValidTrailerHeader(key) {
2161 // TODO: send more details to the peer somehow. But http2 has
2162 // no way to send debug data at a stream level. Discuss with
2163 // HTTP folk.
2164 return sc.countError("trailers_bogus", streamError(st.id, ErrCodeProtocol))
2165 }
2166 st.trailer[key] = append(st.trailer[key], hf.Value)
2167 }
2168 }
2169 st.endStream()
2170 return nil
2171 }
2172 2173 func (sc *serverConn) checkPriority(streamID uint32, p PriorityParam) error {
2174 if streamID == p.StreamDep {
2175 // Section 5.3.1: "A stream cannot depend on itself. An endpoint MUST treat
2176 // this as a stream error (Section 5.4.2) of type PROTOCOL_ERROR."
2177 // Section 5.3.3 says that a stream can depend on one of its dependencies,
2178 // so it's only self-dependencies that are forbidden.
2179 return sc.countError("priority", streamError(streamID, ErrCodeProtocol))
2180 }
2181 return nil
2182 }
2183 2184 func (sc *serverConn) processPriority(f *PriorityFrame) error {
2185 if err := sc.checkPriority(f.StreamID, f.PriorityParam); err != nil {
2186 return err
2187 }
2188 sc.writeSched.AdjustStream(f.StreamID, f.PriorityParam)
2189 return nil
2190 }
2191 2192 func (sc *serverConn) newStream(id, pusherID uint32, state streamState) *stream {
2193 sc.serveG.check()
2194 if id == 0 {
2195 panic("internal error: cannot create stream with id 0")
2196 }
2197 2198 ctx, cancelCtx := context.WithCancel(sc.baseCtx)
2199 st := &stream{
2200 sc: sc,
2201 id: id,
2202 state: state,
2203 ctx: ctx,
2204 cancelCtx: cancelCtx,
2205 }
2206 st.cw.Init()
2207 st.flow.conn = &sc.flow // link to conn-level counter
2208 st.flow.add(sc.initialStreamSendWindowSize)
2209 st.inflow.init(sc.initialStreamRecvWindowSize)
2210 if sc.hs.WriteTimeout > 0 {
2211 st.writeDeadline = time.AfterFunc(sc.hs.WriteTimeout, st.onWriteTimeout)
2212 }
2213 2214 sc.streams[id] = st
2215 sc.writeSched.OpenStream(st.id, OpenStreamOptions{PusherID: pusherID})
2216 if st.isPushed() {
2217 sc.curPushedStreams++
2218 } else {
2219 sc.curClientStreams++
2220 }
2221 if sc.curOpenStreams() == 1 {
2222 sc.setConnState(http.StateActive)
2223 }
2224 2225 return st
2226 }
2227 2228 func (sc *serverConn) newWriterAndRequest(st *stream, f *MetaHeadersFrame) (*responseWriter, *http.Request, error) {
2229 sc.serveG.check()
2230 2231 rp := httpcommon.ServerRequestParam{
2232 Method: f.PseudoValue("method"),
2233 Scheme: f.PseudoValue("scheme"),
2234 Authority: f.PseudoValue("authority"),
2235 Path: f.PseudoValue("path"),
2236 Protocol: f.PseudoValue("protocol"),
2237 }
2238 2239 // extended connect is disabled, so we should not see :protocol
2240 if disableExtendedConnectProtocol && rp.Protocol != "" {
2241 return nil, nil, sc.countError("bad_connect", streamError(f.StreamID, ErrCodeProtocol))
2242 }
2243 2244 isConnect := rp.Method == "CONNECT"
2245 if isConnect {
2246 if rp.Protocol == "" && (rp.Path != "" || rp.Scheme != "" || rp.Authority == "") {
2247 return nil, nil, sc.countError("bad_connect", streamError(f.StreamID, ErrCodeProtocol))
2248 }
2249 } else if rp.Method == "" || rp.Path == "" || (rp.Scheme != "https" && rp.Scheme != "http") {
2250 // See 8.1.2.6 Malformed Requests and Responses:
2251 //
2252 // Malformed requests or responses that are detected
2253 // MUST be treated as a stream error (Section 5.4.2)
2254 // of type PROTOCOL_ERROR."
2255 //
2256 // 8.1.2.3 Request Pseudo-Header Fields
2257 // "All HTTP/2 requests MUST include exactly one valid
2258 // value for the :method, :scheme, and :path
2259 // pseudo-header fields"
2260 return nil, nil, sc.countError("bad_path_method", streamError(f.StreamID, ErrCodeProtocol))
2261 }
2262 2263 header := make(http.Header)
2264 rp.Header = header
2265 for _, hf := range f.RegularFields() {
2266 header.Add(sc.canonicalHeader(hf.Name), hf.Value)
2267 }
2268 if rp.Authority == "" {
2269 rp.Authority = header.Get("Host")
2270 }
2271 if rp.Protocol != "" {
2272 header.Set(":protocol", rp.Protocol)
2273 }
2274 2275 rw, req, err := sc.newWriterAndRequestNoBody(st, rp)
2276 if err != nil {
2277 return nil, nil, err
2278 }
2279 bodyOpen := !f.StreamEnded()
2280 if bodyOpen {
2281 if vv, ok := rp.Header["Content-Length"]; ok {
2282 if cl, err := strconv.ParseUint(vv[0], 10, 63); err == nil {
2283 req.ContentLength = int64(cl)
2284 } else {
2285 req.ContentLength = 0
2286 }
2287 } else {
2288 req.ContentLength = -1
2289 }
2290 req.Body.(*requestBody).pipe = &pipe{
2291 b: &dataBuffer{expected: req.ContentLength},
2292 }
2293 }
2294 return rw, req, nil
2295 }
2296 2297 func (sc *serverConn) newWriterAndRequestNoBody(st *stream, rp httpcommon.ServerRequestParam) (*responseWriter, *http.Request, error) {
2298 sc.serveG.check()
2299 2300 var tlsState *tls.ConnectionState // nil if not scheme https
2301 if rp.Scheme == "https" {
2302 tlsState = sc.tlsState
2303 }
2304 2305 res := httpcommon.NewServerRequest(rp)
2306 if res.InvalidReason != "" {
2307 return nil, nil, sc.countError(res.InvalidReason, streamError(st.id, ErrCodeProtocol))
2308 }
2309 2310 body := &requestBody{
2311 conn: sc,
2312 stream: st,
2313 needsContinue: res.NeedsContinue,
2314 }
2315 req := (&http.Request{
2316 Method: rp.Method,
2317 URL: res.URL,
2318 RemoteAddr: sc.remoteAddrStr,
2319 Header: rp.Header,
2320 RequestURI: res.RequestURI,
2321 Proto: "HTTP/2.0",
2322 ProtoMajor: 2,
2323 ProtoMinor: 0,
2324 TLS: tlsState,
2325 Host: rp.Authority,
2326 Body: body,
2327 Trailer: res.Trailer,
2328 }).WithContext(st.ctx)
2329 rw := sc.newResponseWriter(st, req)
2330 return rw, req, nil
2331 }
2332 2333 func (sc *serverConn) newResponseWriter(st *stream, req *http.Request) *responseWriter {
2334 rws := responseWriterStatePool.Get().(*responseWriterState)
2335 bwSave := rws.bw
2336 *rws = responseWriterState{} // zero all the fields
2337 rws.conn = sc
2338 rws.bw = bwSave
2339 rws.bw.Reset(chunkWriter{rws})
2340 rws.stream = st
2341 rws.req = req
2342 return &responseWriter{rws: rws}
2343 }
2344 2345 type unstartedHandler struct {
2346 streamID uint32
2347 rw *responseWriter
2348 req *http.Request
2349 handler func(http.ResponseWriter, *http.Request)
2350 }
2351 2352 // scheduleHandler starts a handler goroutine,
2353 // or schedules one to start as soon as an existing handler finishes.
2354 func (sc *serverConn) scheduleHandler(streamID uint32, rw *responseWriter, req *http.Request, handler func(http.ResponseWriter, *http.Request)) error {
2355 sc.serveG.check()
2356 maxHandlers := sc.advMaxStreams
2357 if sc.curHandlers < maxHandlers {
2358 sc.curHandlers++
2359 go sc.runHandler(rw, req, handler)
2360 return nil
2361 }
2362 if len(sc.unstartedHandlers) > int(4*sc.advMaxStreams) {
2363 return sc.countError("too_many_early_resets", ConnectionError(ErrCodeEnhanceYourCalm))
2364 }
2365 sc.unstartedHandlers = append(sc.unstartedHandlers, unstartedHandler{
2366 streamID: streamID,
2367 rw: rw,
2368 req: req,
2369 handler: handler,
2370 })
2371 return nil
2372 }
2373 2374 func (sc *serverConn) handlerDone() {
2375 sc.serveG.check()
2376 sc.curHandlers--
2377 i := 0
2378 maxHandlers := sc.advMaxStreams
2379 for ; i < len(sc.unstartedHandlers); i++ {
2380 u := sc.unstartedHandlers[i]
2381 if sc.streams[u.streamID] == nil {
2382 // This stream was reset before its goroutine had a chance to start.
2383 continue
2384 }
2385 if sc.curHandlers >= maxHandlers {
2386 break
2387 }
2388 sc.curHandlers++
2389 go sc.runHandler(u.rw, u.req, u.handler)
2390 sc.unstartedHandlers[i] = unstartedHandler{} // don't retain references
2391 }
2392 sc.unstartedHandlers = sc.unstartedHandlers[i:]
2393 if len(sc.unstartedHandlers) == 0 {
2394 sc.unstartedHandlers = nil
2395 }
2396 }
2397 2398 // Run on its own goroutine.
2399 func (sc *serverConn) runHandler(rw *responseWriter, req *http.Request, handler func(http.ResponseWriter, *http.Request)) {
2400 defer sc.sendServeMsg(handlerDoneMsg)
2401 didPanic := true
2402 defer func() {
2403 rw.rws.stream.cancelCtx()
2404 if req.MultipartForm != nil {
2405 req.MultipartForm.RemoveAll()
2406 }
2407 if didPanic {
2408 e := recover()
2409 sc.writeFrameFromHandler(FrameWriteRequest{
2410 write: handlerPanicRST{rw.rws.stream.id},
2411 stream: rw.rws.stream,
2412 })
2413 // Same as net/http:
2414 if e != nil && e != http.ErrAbortHandler {
2415 const size = 64 << 10
2416 buf := make([]byte, size)
2417 buf = buf[:runtime.Stack(buf, false)]
2418 sc.logf("http2: panic serving %v: %v\n%s", sc.conn.RemoteAddr(), e, buf)
2419 }
2420 return
2421 }
2422 rw.handlerDone()
2423 }()
2424 handler(rw, req)
2425 didPanic = false
2426 }
2427 2428 func handleHeaderListTooLong(w http.ResponseWriter, r *http.Request) {
2429 // 10.5.1 Limits on Header Block Size:
2430 // .. "A server that receives a larger header block than it is
2431 // willing to handle can send an HTTP 431 (Request Header Fields Too
2432 // Large) status code"
2433 const statusRequestHeaderFieldsTooLarge = 431 // only in Go 1.6+
2434 w.WriteHeader(statusRequestHeaderFieldsTooLarge)
2435 io.WriteString(w, "<h1>HTTP Error 431</h1><p>Request Header Field(s) Too Large</p>")
2436 }
2437 2438 // called from handler goroutines.
2439 // h may be nil.
2440 func (sc *serverConn) writeHeaders(st *stream, headerData *writeResHeaders) error {
2441 sc.serveG.checkNotOn() // NOT on
2442 var errc chan error
2443 if headerData.h != nil {
2444 // If there's a header map (which we don't own), so we have to block on
2445 // waiting for this frame to be written, so an http.Flush mid-handler
2446 // writes out the correct value of keys, before a handler later potentially
2447 // mutates it.
2448 errc = sc.srv.state.getErrChan()
2449 }
2450 if err := sc.writeFrameFromHandler(FrameWriteRequest{
2451 write: headerData,
2452 stream: st,
2453 done: errc,
2454 }); err != nil {
2455 return err
2456 }
2457 if errc != nil {
2458 select {
2459 case err := <-errc:
2460 sc.srv.state.putErrChan(errc)
2461 return err
2462 case <-sc.doneServing:
2463 return errClientDisconnected
2464 case <-st.cw:
2465 return errStreamClosed
2466 }
2467 }
2468 return nil
2469 }
2470 2471 // called from handler goroutines.
2472 func (sc *serverConn) write100ContinueHeaders(st *stream) {
2473 sc.writeFrameFromHandler(FrameWriteRequest{
2474 write: write100ContinueHeadersFrame{st.id},
2475 stream: st,
2476 })
2477 }
2478 2479 // A bodyReadMsg tells the server loop that the http.Handler read n
2480 // bytes of the DATA from the client on the given stream.
2481 type bodyReadMsg struct {
2482 st *stream
2483 n int
2484 }
2485 2486 // called from handler goroutines.
2487 // Notes that the handler for the given stream ID read n bytes of its body
2488 // and schedules flow control tokens to be sent.
2489 func (sc *serverConn) noteBodyReadFromHandler(st *stream, n int, err error) {
2490 sc.serveG.checkNotOn() // NOT on
2491 if n > 0 {
2492 select {
2493 case sc.bodyReadCh <- bodyReadMsg{st, n}:
2494 case <-sc.doneServing:
2495 }
2496 }
2497 }
2498 2499 func (sc *serverConn) noteBodyRead(st *stream, n int) {
2500 sc.serveG.check()
2501 sc.sendWindowUpdate(nil, n) // conn-level
2502 if st.state != stateHalfClosedRemote && st.state != stateClosed {
2503 // Don't send this WINDOW_UPDATE if the stream is closed
2504 // remotely.
2505 sc.sendWindowUpdate(st, n)
2506 }
2507 }
2508 2509 // st may be nil for conn-level
2510 func (sc *serverConn) sendWindowUpdate32(st *stream, n int32) {
2511 sc.sendWindowUpdate(st, int(n))
2512 }
2513 2514 // st may be nil for conn-level
2515 func (sc *serverConn) sendWindowUpdate(st *stream, n int) {
2516 sc.serveG.check()
2517 var streamID uint32
2518 var send int32
2519 if st == nil {
2520 send = sc.inflow.add(n)
2521 } else {
2522 streamID = st.id
2523 send = st.inflow.add(n)
2524 }
2525 if send == 0 {
2526 return
2527 }
2528 sc.writeFrame(FrameWriteRequest{
2529 write: writeWindowUpdate{streamID: streamID, n: uint32(send)},
2530 stream: st,
2531 })
2532 }
2533 2534 // requestBody is the Handler's Request.Body type.
2535 // Read and Close may be called concurrently.
2536 type requestBody struct {
2537 _ incomparable
2538 stream *stream
2539 conn *serverConn
2540 closeOnce sync.Once // for use by Close only
2541 sawEOF bool // for use by Read only
2542 pipe *pipe // non-nil if we have an HTTP entity message body
2543 needsContinue bool // need to send a 100-continue
2544 }
2545 2546 func (b *requestBody) Close() error {
2547 b.closeOnce.Do(func() {
2548 if b.pipe != nil {
2549 b.pipe.BreakWithError(errClosedBody)
2550 }
2551 })
2552 return nil
2553 }
2554 2555 func (b *requestBody) Read(p []byte) (n int, err error) {
2556 if b.needsContinue {
2557 b.needsContinue = false
2558 b.conn.write100ContinueHeaders(b.stream)
2559 }
2560 if b.pipe == nil || b.sawEOF {
2561 return 0, io.EOF
2562 }
2563 n, err = b.pipe.Read(p)
2564 if err == io.EOF {
2565 b.sawEOF = true
2566 }
2567 if b.conn == nil {
2568 return
2569 }
2570 b.conn.noteBodyReadFromHandler(b.stream, n, err)
2571 return
2572 }
2573 2574 // responseWriter is the http.ResponseWriter implementation. It's
2575 // intentionally small (1 pointer wide) to minimize garbage. The
2576 // responseWriterState pointer inside is zeroed at the end of a
2577 // request (in handlerDone) and calls on the responseWriter thereafter
2578 // simply crash (caller's mistake), but the much larger responseWriterState
2579 // and buffers are reused between multiple requests.
2580 type responseWriter struct {
2581 rws *responseWriterState
2582 }
2583 2584 // Optional http.ResponseWriter interfaces implemented.
2585 var (
2586 _ http.CloseNotifier = (*responseWriter)(nil)
2587 _ http.Flusher = (*responseWriter)(nil)
2588 _ stringWriter = (*responseWriter)(nil)
2589 )
2590 2591 type responseWriterState struct {
2592 // immutable within a request:
2593 stream *stream
2594 req *http.Request
2595 conn *serverConn
2596 2597 // TODO: adjust buffer writing sizes based on server config, frame size updates from peer, etc
2598 bw *bufio.Writer // writing to a chunkWriter{this *responseWriterState}
2599 2600 // mutated by http.Handler goroutine:
2601 handlerHeader http.Header // nil until called
2602 snapHeader http.Header // snapshot of handlerHeader at WriteHeader time
2603 trailers []string // set in writeChunk
2604 status int // status code passed to WriteHeader
2605 wroteHeader bool // WriteHeader called (explicitly or implicitly). Not necessarily sent to user yet.
2606 sentHeader bool // have we sent the header frame?
2607 handlerDone bool // handler has finished
2608 2609 sentContentLen int64 // non-zero if handler set a Content-Length header
2610 wroteBytes int64
2611 2612 closeNotifierMu sync.Mutex // guards closeNotifierCh
2613 closeNotifierCh chan bool // nil until first used
2614 }
2615 2616 type chunkWriter struct{ rws *responseWriterState }
2617 2618 func (cw chunkWriter) Write(p []byte) (n int, err error) {
2619 n, err = cw.rws.writeChunk(p)
2620 if err == errStreamClosed {
2621 // If writing failed because the stream has been closed,
2622 // return the reason it was closed.
2623 err = cw.rws.stream.closeErr
2624 }
2625 return n, err
2626 }
2627 2628 func (rws *responseWriterState) hasTrailers() bool { return len(rws.trailers) > 0 }
2629 2630 func (rws *responseWriterState) hasNonemptyTrailers() bool {
2631 for _, trailer := range rws.trailers {
2632 if _, ok := rws.handlerHeader[trailer]; ok {
2633 return true
2634 }
2635 }
2636 return false
2637 }
2638 2639 // declareTrailer is called for each Trailer header when the
2640 // response header is written. It notes that a header will need to be
2641 // written in the trailers at the end of the response.
2642 func (rws *responseWriterState) declareTrailer(k string) {
2643 k = http.CanonicalHeaderKey(k)
2644 if !httpguts.ValidTrailerHeader(k) {
2645 // Forbidden by RFC 7230, section 4.1.2.
2646 rws.conn.logf("ignoring invalid trailer %q", k)
2647 return
2648 }
2649 if !strSliceContains(rws.trailers, k) {
2650 rws.trailers = append(rws.trailers, k)
2651 }
2652 }
2653 2654 // writeChunk writes chunks from the bufio.Writer. But because
2655 // bufio.Writer may bypass its chunking, sometimes p may be
2656 // arbitrarily large.
2657 //
2658 // writeChunk is also responsible (on the first chunk) for sending the
2659 // HEADER response.
2660 func (rws *responseWriterState) writeChunk(p []byte) (n int, err error) {
2661 if !rws.wroteHeader {
2662 rws.writeHeader(200)
2663 }
2664 2665 if rws.handlerDone {
2666 rws.promoteUndeclaredTrailers()
2667 }
2668 2669 isHeadResp := rws.req.Method == "HEAD"
2670 if !rws.sentHeader {
2671 rws.sentHeader = true
2672 var ctype, clen string
2673 if clen = rws.snapHeader.Get("Content-Length"); clen != "" {
2674 rws.snapHeader.Del("Content-Length")
2675 if cl, err := strconv.ParseUint(clen, 10, 63); err == nil {
2676 rws.sentContentLen = int64(cl)
2677 } else {
2678 clen = ""
2679 }
2680 }
2681 _, hasContentLength := rws.snapHeader["Content-Length"]
2682 if !hasContentLength && clen == "" && rws.handlerDone && bodyAllowedForStatus(rws.status) && (len(p) > 0 || !isHeadResp) {
2683 clen = strconv.Itoa(len(p))
2684 }
2685 _, hasContentType := rws.snapHeader["Content-Type"]
2686 // If the Content-Encoding is non-blank, we shouldn't
2687 // sniff the body. See Issue golang.org/issue/31753.
2688 ce := rws.snapHeader.Get("Content-Encoding")
2689 hasCE := len(ce) > 0
2690 if !hasCE && !hasContentType && bodyAllowedForStatus(rws.status) && len(p) > 0 {
2691 ctype = http.DetectContentType(p)
2692 }
2693 var date string
2694 if _, ok := rws.snapHeader["Date"]; !ok {
2695 // TODO(bradfitz): be faster here, like net/http? measure.
2696 date = time.Now().UTC().Format(http.TimeFormat)
2697 }
2698 2699 for _, v := range rws.snapHeader["Trailer"] {
2700 foreachHeaderElement(v, rws.declareTrailer)
2701 }
2702 2703 // "Connection" headers aren't allowed in HTTP/2 (RFC 7540, 8.1.2.2),
2704 // but respect "Connection" == "close" to mean sending a GOAWAY and tearing
2705 // down the TCP connection when idle, like we do for HTTP/1.
2706 // TODO: remove more Connection-specific header fields here, in addition
2707 // to "Connection".
2708 if _, ok := rws.snapHeader["Connection"]; ok {
2709 v := rws.snapHeader.Get("Connection")
2710 delete(rws.snapHeader, "Connection")
2711 if v == "close" {
2712 rws.conn.startGracefulShutdown()
2713 }
2714 }
2715 2716 endStream := (rws.handlerDone && !rws.hasTrailers() && len(p) == 0) || isHeadResp
2717 err = rws.conn.writeHeaders(rws.stream, &writeResHeaders{
2718 streamID: rws.stream.id,
2719 httpResCode: rws.status,
2720 h: rws.snapHeader,
2721 endStream: endStream,
2722 contentType: ctype,
2723 contentLength: clen,
2724 date: date,
2725 })
2726 if err != nil {
2727 return 0, err
2728 }
2729 if endStream {
2730 return 0, nil
2731 }
2732 }
2733 if isHeadResp {
2734 return len(p), nil
2735 }
2736 if len(p) == 0 && !rws.handlerDone {
2737 return 0, nil
2738 }
2739 2740 // only send trailers if they have actually been defined by the
2741 // server handler.
2742 hasNonemptyTrailers := rws.hasNonemptyTrailers()
2743 endStream := rws.handlerDone && !hasNonemptyTrailers
2744 if len(p) > 0 || endStream {
2745 // only send a 0 byte DATA frame if we're ending the stream.
2746 if err := rws.conn.writeDataFromHandler(rws.stream, p, endStream); err != nil {
2747 return 0, err
2748 }
2749 }
2750 2751 if rws.handlerDone && hasNonemptyTrailers {
2752 err = rws.conn.writeHeaders(rws.stream, &writeResHeaders{
2753 streamID: rws.stream.id,
2754 h: rws.handlerHeader,
2755 trailers: rws.trailers,
2756 endStream: true,
2757 })
2758 return len(p), err
2759 }
2760 return len(p), nil
2761 }
2762 2763 // TrailerPrefix is a magic prefix for ResponseWriter.Header map keys
2764 // that, if present, signals that the map entry is actually for
2765 // the response trailers, and not the response headers. The prefix
2766 // is stripped after the ServeHTTP call finishes and the values are
2767 // sent in the trailers.
2768 //
2769 // This mechanism is intended only for trailers that are not known
2770 // prior to the headers being written. If the set of trailers is fixed
2771 // or known before the header is written, the normal Go trailers mechanism
2772 // is preferred:
2773 //
2774 // https://golang.org/pkg/net/http/#ResponseWriter
2775 // https://golang.org/pkg/net/http/#example_ResponseWriter_trailers
2776 const TrailerPrefix = "Trailer:"
2777 2778 // promoteUndeclaredTrailers permits http.Handlers to set trailers
2779 // after the header has already been flushed. Because the Go
2780 // ResponseWriter interface has no way to set Trailers (only the
2781 // Header), and because we didn't want to expand the ResponseWriter
2782 // interface, and because nobody used trailers, and because RFC 7230
2783 // says you SHOULD (but not must) predeclare any trailers in the
2784 // header, the official ResponseWriter rules said trailers in Go must
2785 // be predeclared, and then we reuse the same ResponseWriter.Header()
2786 // map to mean both Headers and Trailers. When it's time to write the
2787 // Trailers, we pick out the fields of Headers that were declared as
2788 // trailers. That worked for a while, until we found the first major
2789 // user of Trailers in the wild: gRPC (using them only over http2),
2790 // and gRPC libraries permit setting trailers mid-stream without
2791 // predeclaring them. So: change of plans. We still permit the old
2792 // way, but we also permit this hack: if a Header() key begins with
2793 // "Trailer:", the suffix of that key is a Trailer. Because ':' is an
2794 // invalid token byte anyway, there is no ambiguity. (And it's already
2795 // filtered out) It's mildly hacky, but not terrible.
2796 //
2797 // This method runs after the Handler is done and promotes any Header
2798 // fields to be trailers.
2799 func (rws *responseWriterState) promoteUndeclaredTrailers() {
2800 for k, vv := range rws.handlerHeader {
2801 if !strings.HasPrefix(k, TrailerPrefix) {
2802 continue
2803 }
2804 trailerKey := strings.TrimPrefix(k, TrailerPrefix)
2805 rws.declareTrailer(trailerKey)
2806 rws.handlerHeader[http.CanonicalHeaderKey(trailerKey)] = vv
2807 }
2808 2809 if len(rws.trailers) > 1 {
2810 sorter := sorterPool.Get().(*sorter)
2811 sorter.SortStrings(rws.trailers)
2812 sorterPool.Put(sorter)
2813 }
2814 }
2815 2816 func (w *responseWriter) SetReadDeadline(deadline time.Time) error {
2817 st := w.rws.stream
2818 if !deadline.IsZero() && deadline.Before(time.Now()) {
2819 // If we're setting a deadline in the past, reset the stream immediately
2820 // so writes after SetWriteDeadline returns will fail.
2821 st.onReadTimeout()
2822 return nil
2823 }
2824 w.rws.conn.sendServeMsg(func(sc *serverConn) {
2825 if st.readDeadline != nil {
2826 if !st.readDeadline.Stop() {
2827 // Deadline already exceeded, or stream has been closed.
2828 return
2829 }
2830 }
2831 if deadline.IsZero() {
2832 st.readDeadline = nil
2833 } else if st.readDeadline == nil {
2834 st.readDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onReadTimeout)
2835 } else {
2836 st.readDeadline.Reset(deadline.Sub(time.Now()))
2837 }
2838 })
2839 return nil
2840 }
2841 2842 func (w *responseWriter) SetWriteDeadline(deadline time.Time) error {
2843 st := w.rws.stream
2844 if !deadline.IsZero() && deadline.Before(time.Now()) {
2845 // If we're setting a deadline in the past, reset the stream immediately
2846 // so writes after SetWriteDeadline returns will fail.
2847 st.onWriteTimeout()
2848 return nil
2849 }
2850 w.rws.conn.sendServeMsg(func(sc *serverConn) {
2851 if st.writeDeadline != nil {
2852 if !st.writeDeadline.Stop() {
2853 // Deadline already exceeded, or stream has been closed.
2854 return
2855 }
2856 }
2857 if deadline.IsZero() {
2858 st.writeDeadline = nil
2859 } else if st.writeDeadline == nil {
2860 st.writeDeadline = time.AfterFunc(deadline.Sub(time.Now()), st.onWriteTimeout)
2861 } else {
2862 st.writeDeadline.Reset(deadline.Sub(time.Now()))
2863 }
2864 })
2865 return nil
2866 }
2867 2868 func (w *responseWriter) EnableFullDuplex() error {
2869 // We always support full duplex responses, so this is a no-op.
2870 return nil
2871 }
2872 2873 func (w *responseWriter) Flush() {
2874 w.FlushError()
2875 }
2876 2877 func (w *responseWriter) FlushError() error {
2878 rws := w.rws
2879 if rws == nil {
2880 panic("Header called after Handler finished")
2881 }
2882 var err error
2883 if rws.bw.Buffered() > 0 {
2884 err = rws.bw.Flush()
2885 } else {
2886 // The bufio.Writer won't call chunkWriter.Write
2887 // (writeChunk with zero bytes), so we have to do it
2888 // ourselves to force the HTTP response header and/or
2889 // final DATA frame (with END_STREAM) to be sent.
2890 _, err = chunkWriter{rws}.Write(nil)
2891 if err == nil {
2892 select {
2893 case <-rws.stream.cw:
2894 err = rws.stream.closeErr
2895 default:
2896 }
2897 }
2898 }
2899 return err
2900 }
2901 2902 func (w *responseWriter) CloseNotify() <-chan bool {
2903 rws := w.rws
2904 if rws == nil {
2905 panic("CloseNotify called after Handler finished")
2906 }
2907 rws.closeNotifierMu.Lock()
2908 ch := rws.closeNotifierCh
2909 if ch == nil {
2910 ch = make(chan bool, 1)
2911 rws.closeNotifierCh = ch
2912 cw := rws.stream.cw
2913 go func() {
2914 cw.Wait() // wait for close
2915 ch <- true
2916 }()
2917 }
2918 rws.closeNotifierMu.Unlock()
2919 return ch
2920 }
2921 2922 func (w *responseWriter) Header() http.Header {
2923 rws := w.rws
2924 if rws == nil {
2925 panic("Header called after Handler finished")
2926 }
2927 if rws.handlerHeader == nil {
2928 rws.handlerHeader = make(http.Header)
2929 }
2930 return rws.handlerHeader
2931 }
2932 2933 // checkWriteHeaderCode is a copy of net/http's checkWriteHeaderCode.
2934 func checkWriteHeaderCode(code int) {
2935 // Issue 22880: require valid WriteHeader status codes.
2936 // For now we only enforce that it's three digits.
2937 // In the future we might block things over 599 (600 and above aren't defined
2938 // at http://httpwg.org/specs/rfc7231.html#status.codes).
2939 // But for now any three digits.
2940 //
2941 // We used to send "HTTP/1.1 000 0" on the wire in responses but there's
2942 // no equivalent bogus thing we can realistically send in HTTP/2,
2943 // so we'll consistently panic instead and help people find their bugs
2944 // early. (We can't return an error from WriteHeader even if we wanted to.)
2945 if code < 100 || code > 999 {
2946 panic(fmt.Sprintf("invalid WriteHeader code %v", code))
2947 }
2948 }
2949 2950 func (w *responseWriter) WriteHeader(code int) {
2951 rws := w.rws
2952 if rws == nil {
2953 panic("WriteHeader called after Handler finished")
2954 }
2955 rws.writeHeader(code)
2956 }
2957 2958 func (rws *responseWriterState) writeHeader(code int) {
2959 if rws.wroteHeader {
2960 return
2961 }
2962 2963 checkWriteHeaderCode(code)
2964 2965 // Handle informational headers
2966 if code >= 100 && code <= 199 {
2967 // Per RFC 8297 we must not clear the current header map
2968 h := rws.handlerHeader
2969 2970 _, cl := h["Content-Length"]
2971 _, te := h["Transfer-Encoding"]
2972 if cl || te {
2973 h = h.Clone()
2974 h.Del("Content-Length")
2975 h.Del("Transfer-Encoding")
2976 }
2977 2978 rws.conn.writeHeaders(rws.stream, &writeResHeaders{
2979 streamID: rws.stream.id,
2980 httpResCode: code,
2981 h: h,
2982 endStream: rws.handlerDone && !rws.hasTrailers(),
2983 })
2984 2985 return
2986 }
2987 2988 rws.wroteHeader = true
2989 rws.status = code
2990 if len(rws.handlerHeader) > 0 {
2991 rws.snapHeader = cloneHeader(rws.handlerHeader)
2992 }
2993 }
2994 2995 func cloneHeader(h http.Header) http.Header {
2996 h2 := make(http.Header, len(h))
2997 for k, vv := range h {
2998 vv2 := make([]string, len(vv))
2999 copy(vv2, vv)
3000 h2[k] = vv2
3001 }
3002 return h2
3003 }
3004 3005 // The Life Of A Write is like this:
3006 //
3007 // * Handler calls w.Write or w.WriteString ->
3008 // * -> rws.bw (*bufio.Writer) ->
3009 // * (Handler might call Flush)
3010 // * -> chunkWriter{rws}
3011 // * -> responseWriterState.writeChunk(p []byte)
3012 // * -> responseWriterState.writeChunk (most of the magic; see comment there)
3013 func (w *responseWriter) Write(p []byte) (n int, err error) {
3014 return w.write(len(p), p, "")
3015 }
3016 3017 func (w *responseWriter) WriteString(s string) (n int, err error) {
3018 return w.write(len(s), nil, s)
3019 }
3020 3021 // either dataB or dataS is non-zero.
3022 func (w *responseWriter) write(lenData int, dataB []byte, dataS string) (n int, err error) {
3023 rws := w.rws
3024 if rws == nil {
3025 panic("Write called after Handler finished")
3026 }
3027 if !rws.wroteHeader {
3028 w.WriteHeader(200)
3029 }
3030 if !bodyAllowedForStatus(rws.status) {
3031 return 0, http.ErrBodyNotAllowed
3032 }
3033 rws.wroteBytes += int64(len(dataB)) + int64(len(dataS)) // only one can be set
3034 if rws.sentContentLen != 0 && rws.wroteBytes > rws.sentContentLen {
3035 // TODO: send a RST_STREAM
3036 return 0, errors.New("http2: handler wrote more than declared Content-Length")
3037 }
3038 3039 if dataB != nil {
3040 return rws.bw.Write(dataB)
3041 } else {
3042 return rws.bw.WriteString(dataS)
3043 }
3044 }
3045 3046 func (w *responseWriter) handlerDone() {
3047 rws := w.rws
3048 rws.handlerDone = true
3049 w.Flush()
3050 w.rws = nil
3051 responseWriterStatePool.Put(rws)
3052 }
3053 3054 // Push errors.
3055 var (
3056 ErrRecursivePush = errors.New("http2: recursive push not allowed")
3057 ErrPushLimitReached = errors.New("http2: push would exceed peer's SETTINGS_MAX_CONCURRENT_STREAMS")
3058 )
3059 3060 var _ http.Pusher = (*responseWriter)(nil)
3061 3062 func (w *responseWriter) Push(target string, opts *http.PushOptions) error {
3063 st := w.rws.stream
3064 sc := st.sc
3065 sc.serveG.checkNotOn()
3066 3067 // No recursive pushes: "PUSH_PROMISE frames MUST only be sent on a peer-initiated stream."
3068 // http://tools.ietf.org/html/rfc7540#section-6.6
3069 if st.isPushed() {
3070 return ErrRecursivePush
3071 }
3072 3073 if opts == nil {
3074 opts = new(http.PushOptions)
3075 }
3076 3077 // Default options.
3078 if opts.Method == "" {
3079 opts.Method = "GET"
3080 }
3081 if opts.Header == nil {
3082 opts.Header = http.Header{}
3083 }
3084 wantScheme := "http"
3085 if w.rws.req.TLS != nil {
3086 wantScheme = "https"
3087 }
3088 3089 // Validate the request.
3090 u, err := url.Parse(target)
3091 if err != nil {
3092 return err
3093 }
3094 if u.Scheme == "" {
3095 if !strings.HasPrefix(target, "/") {
3096 return fmt.Errorf("target must be an absolute URL or an absolute path: %q", target)
3097 }
3098 u.Scheme = wantScheme
3099 u.Host = w.rws.req.Host
3100 } else {
3101 if u.Scheme != wantScheme {
3102 return fmt.Errorf("cannot push URL with scheme %q from request with scheme %q", u.Scheme, wantScheme)
3103 }
3104 if u.Host == "" {
3105 return errors.New("URL must have a host")
3106 }
3107 }
3108 for k := range opts.Header {
3109 if strings.HasPrefix(k, ":") {
3110 return fmt.Errorf("promised request headers cannot include pseudo header %q", k)
3111 }
3112 // These headers are meaningful only if the request has a body,
3113 // but PUSH_PROMISE requests cannot have a body.
3114 // http://tools.ietf.org/html/rfc7540#section-8.2
3115 // Also disallow Host, since the promised URL must be absolute.
3116 if asciiEqualFold(k, "content-length") ||
3117 asciiEqualFold(k, "content-encoding") ||
3118 asciiEqualFold(k, "trailer") ||
3119 asciiEqualFold(k, "te") ||
3120 asciiEqualFold(k, "expect") ||
3121 asciiEqualFold(k, "host") {
3122 return fmt.Errorf("promised request headers cannot include %q", k)
3123 }
3124 }
3125 if err := checkValidHTTP2RequestHeaders(opts.Header); err != nil {
3126 return err
3127 }
3128 3129 // The RFC effectively limits promised requests to GET and HEAD:
3130 // "Promised requests MUST be cacheable [GET, HEAD, or POST], and MUST be safe [GET or HEAD]"
3131 // http://tools.ietf.org/html/rfc7540#section-8.2
3132 if opts.Method != "GET" && opts.Method != "HEAD" {
3133 return fmt.Errorf("method %q must be GET or HEAD", opts.Method)
3134 }
3135 3136 msg := &startPushRequest{
3137 parent: st,
3138 method: opts.Method,
3139 url: u,
3140 header: cloneHeader(opts.Header),
3141 done: sc.srv.state.getErrChan(),
3142 }
3143 3144 select {
3145 case <-sc.doneServing:
3146 return errClientDisconnected
3147 case <-st.cw:
3148 return errStreamClosed
3149 case sc.serveMsgCh <- msg:
3150 }
3151 3152 select {
3153 case <-sc.doneServing:
3154 return errClientDisconnected
3155 case <-st.cw:
3156 return errStreamClosed
3157 case err := <-msg.done:
3158 sc.srv.state.putErrChan(msg.done)
3159 return err
3160 }
3161 }
3162 3163 type startPushRequest struct {
3164 parent *stream
3165 method string
3166 url *url.URL
3167 header http.Header
3168 done chan error
3169 }
3170 3171 func (sc *serverConn) startPush(msg *startPushRequest) {
3172 sc.serveG.check()
3173 3174 // http://tools.ietf.org/html/rfc7540#section-6.6.
3175 // PUSH_PROMISE frames MUST only be sent on a peer-initiated stream that
3176 // is in either the "open" or "half-closed (remote)" state.
3177 if msg.parent.state != stateOpen && msg.parent.state != stateHalfClosedRemote {
3178 // responseWriter.Push checks that the stream is peer-initiated.
3179 msg.done <- errStreamClosed
3180 return
3181 }
3182 3183 // http://tools.ietf.org/html/rfc7540#section-6.6.
3184 if !sc.pushEnabled {
3185 msg.done <- http.ErrNotSupported
3186 return
3187 }
3188 3189 // PUSH_PROMISE frames must be sent in increasing order by stream ID, so
3190 // we allocate an ID for the promised stream lazily, when the PUSH_PROMISE
3191 // is written. Once the ID is allocated, we start the request handler.
3192 allocatePromisedID := func() (uint32, error) {
3193 sc.serveG.check()
3194 3195 // Check this again, just in case. Technically, we might have received
3196 // an updated SETTINGS by the time we got around to writing this frame.
3197 if !sc.pushEnabled {
3198 return 0, http.ErrNotSupported
3199 }
3200 // http://tools.ietf.org/html/rfc7540#section-6.5.2.
3201 if sc.curPushedStreams+1 > sc.clientMaxStreams {
3202 return 0, ErrPushLimitReached
3203 }
3204 3205 // http://tools.ietf.org/html/rfc7540#section-5.1.1.
3206 // Streams initiated by the server MUST use even-numbered identifiers.
3207 // A server that is unable to establish a new stream identifier can send a GOAWAY
3208 // frame so that the client is forced to open a new connection for new streams.
3209 if sc.maxPushPromiseID+2 >= 1<<31 {
3210 sc.startGracefulShutdownInternal()
3211 return 0, ErrPushLimitReached
3212 }
3213 sc.maxPushPromiseID += 2
3214 promisedID := sc.maxPushPromiseID
3215 3216 // http://tools.ietf.org/html/rfc7540#section-8.2.
3217 // Strictly speaking, the new stream should start in "reserved (local)", then
3218 // transition to "half closed (remote)" after sending the initial HEADERS, but
3219 // we start in "half closed (remote)" for simplicity.
3220 // See further comments at the definition of stateHalfClosedRemote.
3221 promised := sc.newStream(promisedID, msg.parent.id, stateHalfClosedRemote)
3222 rw, req, err := sc.newWriterAndRequestNoBody(promised, httpcommon.ServerRequestParam{
3223 Method: msg.method,
3224 Scheme: msg.url.Scheme,
3225 Authority: msg.url.Host,
3226 Path: msg.url.RequestURI(),
3227 Header: cloneHeader(msg.header), // clone since handler runs concurrently with writing the PUSH_PROMISE
3228 })
3229 if err != nil {
3230 // Should not happen, since we've already validated msg.url.
3231 panic(fmt.Sprintf("newWriterAndRequestNoBody(%+v): %v", msg.url, err))
3232 }
3233 3234 sc.curHandlers++
3235 go sc.runHandler(rw, req, sc.handler.ServeHTTP)
3236 return promisedID, nil
3237 }
3238 3239 sc.writeFrame(FrameWriteRequest{
3240 write: &writePushPromise{
3241 streamID: msg.parent.id,
3242 method: msg.method,
3243 url: msg.url,
3244 h: msg.header,
3245 allocatePromisedID: allocatePromisedID,
3246 },
3247 stream: msg.parent,
3248 done: msg.done,
3249 })
3250 }
3251 3252 // foreachHeaderElement splits v according to the "#rule" construction
3253 // in RFC 7230 section 7 and calls fn for each non-empty element.
3254 func foreachHeaderElement(v string, fn func(string)) {
3255 v = textproto.TrimString(v)
3256 if v == "" {
3257 return
3258 }
3259 if !strings.Contains(v, ",") {
3260 fn(v)
3261 return
3262 }
3263 for _, f := range strings.Split(v, ",") {
3264 if f = textproto.TrimString(f); f != "" {
3265 fn(f)
3266 }
3267 }
3268 }
3269 3270 // From http://httpwg.org/specs/rfc7540.html#rfc.section.8.1.2.2
3271 var connHeaders = []string{
3272 "Connection",
3273 "Keep-Alive",
3274 "Proxy-Connection",
3275 "Transfer-Encoding",
3276 "Upgrade",
3277 }
3278 3279 // checkValidHTTP2RequestHeaders checks whether h is a valid HTTP/2 request,
3280 // per RFC 7540 Section 8.1.2.2.
3281 // The returned error is reported to users.
3282 func checkValidHTTP2RequestHeaders(h http.Header) error {
3283 for _, k := range connHeaders {
3284 if _, ok := h[k]; ok {
3285 return fmt.Errorf("request header %q is not valid in HTTP/2", k)
3286 }
3287 }
3288 te := h["Te"]
3289 if len(te) > 0 && (len(te) > 1 || (te[0] != "trailers" && te[0] != "")) {
3290 return errors.New(`request header "TE" may only be "trailers" in HTTP/2`)
3291 }
3292 return nil
3293 }
3294 3295 func new400Handler(err error) http.HandlerFunc {
3296 return func(w http.ResponseWriter, r *http.Request) {
3297 http.Error(w, err.Error(), http.StatusBadRequest)
3298 }
3299 }
3300 3301 // h1ServerKeepAlivesDisabled reports whether hs has its keep-alives
3302 // disabled. See comments on h1ServerShutdownChan above for why
3303 // the code is written this way.
3304 func h1ServerKeepAlivesDisabled(hs *http.Server) bool {
3305 var x interface{} = hs
3306 type I interface {
3307 doKeepAlives() bool
3308 }
3309 if hs, ok := x.(I); ok {
3310 return !hs.doKeepAlives()
3311 }
3312 return false
3313 }
3314 3315 func (sc *serverConn) countError(name string, err error) error {
3316 if sc == nil || sc.srv == nil {
3317 return err
3318 }
3319 f := sc.countErrorFunc
3320 if f == nil {
3321 return err
3322 }
3323 var typ string
3324 var code ErrCode
3325 switch e := err.(type) {
3326 case ConnectionError:
3327 typ = "conn"
3328 code = ErrCode(e)
3329 case StreamError:
3330 typ = "stream"
3331 code = ErrCode(e.Code)
3332 default:
3333 return err
3334 }
3335 codeStr := errCodeName[code]
3336 if codeStr == "" {
3337 codeStr = strconv.Itoa(int(code))
3338 }
3339 f(fmt.Sprintf("%s_%s_%s", typ, codeStr, name))
3340 return err
3341 }
3342