server.go 100 KB

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