http2.go 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910
  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. // See https://code.google.com/p/go/source/browse/CONTRIBUTORS
  5. // Licensed under the same terms as Go itself:
  6. // https://code.google.com/p/go/source/browse/LICENSE
  7. // Package http2 implements the HTTP/2 protocol.
  8. //
  9. // This is a work in progress. This package is low-level and intended
  10. // to be used directly by very few people. Most users will use it
  11. // indirectly through integration with the net/http package. See
  12. // ConfigureServer. That ConfigureServer call will likely be automatic
  13. // or available via an empty import in the future.
  14. //
  15. // This package currently targets draft-14. See http://http2.github.io/
  16. package http2
  17. // TODO: finish GOAWAY support. Consider each incoming frame type and whether
  18. // it should be ignored during a shutdown race.
  19. import (
  20. "bytes"
  21. "crypto/tls"
  22. "errors"
  23. "fmt"
  24. "io"
  25. "log"
  26. "net"
  27. "net/http"
  28. "net/url"
  29. "strconv"
  30. "strings"
  31. "github.com/bradfitz/http2/hpack"
  32. )
  33. var VerboseLogs = false
  34. const (
  35. // ClientPreface is the string that must be sent by new
  36. // connections from clients.
  37. ClientPreface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
  38. // SETTINGS_MAX_FRAME_SIZE default
  39. // http://http2.github.io/http2-spec/#rfc.section.6.5.2
  40. initialMaxFrameSize = 16384
  41. npnProto = "h2-14"
  42. // http://http2.github.io/http2-spec/#SettingValues
  43. initialHeaderTableSize = 4096
  44. initialWindowSize = 65535 // 6.9.2 Initial Flow Control Window Size
  45. )
  46. var (
  47. clientPreface = []byte(ClientPreface)
  48. )
  49. // Server is an HTTP2 server.
  50. type Server struct {
  51. // MaxStreams optionally ...
  52. MaxStreams int
  53. }
  54. func (srv *Server) handleConn(hs *http.Server, c net.Conn, h http.Handler) {
  55. sc := &serverConn{
  56. hs: hs,
  57. conn: c,
  58. handler: h,
  59. framer: NewFramer(c, c),
  60. streams: make(map[uint32]*stream),
  61. canonHeader: make(map[string]string),
  62. readFrameCh: make(chan frameAndProcessed),
  63. readFrameErrCh: make(chan error, 1),
  64. writeHeaderCh: make(chan headerWriteReq), // must not be buffered
  65. flow: newFlow(initialWindowSize),
  66. doneServing: make(chan struct{}),
  67. maxWriteFrameSize: initialMaxFrameSize,
  68. initialWindowSize: initialWindowSize,
  69. serveG: newGoroutineLock(),
  70. }
  71. sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf)
  72. sc.hpackDecoder = hpack.NewDecoder(initialHeaderTableSize, sc.onNewHeaderField)
  73. sc.serve()
  74. }
  75. // frameAndProcessed coordinates the readFrames and serve goroutines, since
  76. // the Framer interface only permits the most recently-read Frame from being
  77. // accessed. The serve goroutine sends on processed to signal to the readFrames
  78. // goroutine that another frame may be read.
  79. type frameAndProcessed struct {
  80. f Frame
  81. processed chan struct{}
  82. }
  83. type serverConn struct {
  84. // Immutable:
  85. hs *http.Server
  86. conn net.Conn
  87. handler http.Handler
  88. framer *Framer
  89. hpackDecoder *hpack.Decoder
  90. hpackEncoder *hpack.Encoder
  91. doneServing chan struct{} // closed when serverConn.serve ends
  92. readFrameCh chan frameAndProcessed // written by serverConn.readFrames
  93. readFrameErrCh chan error
  94. writeHeaderCh chan headerWriteReq // must not be buffered
  95. serveG goroutineLock // used to verify funcs are on serve()
  96. flow *flow // the connection-wide one
  97. // Everything following is owned by the serve loop; use serveG.check()
  98. maxStreamID uint32 // max ever seen
  99. streams map[uint32]*stream
  100. maxWriteFrameSize uint32 // TODO: update this when settings come in
  101. initialWindowSize int32
  102. canonHeader map[string]string // http2-lower-case -> Go-Canonical-Case
  103. sentGoAway bool
  104. req requestParam // non-zero while reading request headers
  105. headerWriteBuf bytes.Buffer // used to write response headers
  106. }
  107. // requestParam is the state of the next request, initialized over
  108. // potentially several frames HEADERS + zero or more CONTINUATION
  109. // frames.
  110. type requestParam struct {
  111. // stream is non-nil if we're reading (HEADER or CONTINUATION)
  112. // frames for a request (but not DATA).
  113. stream *stream
  114. header http.Header
  115. method, path string
  116. scheme, authority string
  117. sawRegularHeader bool // saw a non-pseudo header already
  118. invalidHeader bool // an invalid header was seen
  119. }
  120. type streamState int
  121. const (
  122. stateIdle streamState = iota
  123. stateOpen
  124. stateHalfClosedLocal
  125. stateHalfClosedRemote
  126. stateResvLocal
  127. stateResvRemote
  128. stateClosed
  129. )
  130. type stream struct {
  131. id uint32
  132. state streamState // owned by serverConn's processing loop
  133. flow *flow // limits writing from Handler to client
  134. body *pipe // non-nil if expecting DATA frames
  135. bodyBytes int64 // body bytes seen so far
  136. declBodyBytes int64 // or -1 if undeclared
  137. }
  138. func (sc *serverConn) state(streamID uint32) streamState {
  139. sc.serveG.check()
  140. // http://http2.github.io/http2-spec/#rfc.section.5.1
  141. if st, ok := sc.streams[streamID]; ok {
  142. return st.state
  143. }
  144. // "The first use of a new stream identifier implicitly closes all
  145. // streams in the "idle" state that might have been initiated by
  146. // that peer with a lower-valued stream identifier. For example, if
  147. // a client sends a HEADERS frame on stream 7 without ever sending a
  148. // frame on stream 5, then stream 5 transitions to the "closed"
  149. // state when the first frame for stream 7 is sent or received."
  150. if streamID <= sc.maxStreamID {
  151. return stateClosed
  152. }
  153. return stateIdle
  154. }
  155. func (sc *serverConn) vlogf(format string, args ...interface{}) {
  156. if VerboseLogs {
  157. sc.logf(format, args...)
  158. }
  159. }
  160. func (sc *serverConn) logf(format string, args ...interface{}) {
  161. if lg := sc.hs.ErrorLog; lg != nil {
  162. lg.Printf(format, args...)
  163. } else {
  164. log.Printf(format, args...)
  165. }
  166. }
  167. func (sc *serverConn) condlogf(err error, format string, args ...interface{}) {
  168. if err == nil {
  169. return
  170. }
  171. str := err.Error()
  172. if strings.Contains(str, "use of closed network connection") {
  173. // Boring, expected errors.
  174. sc.vlogf(format, args...)
  175. } else {
  176. sc.logf(format, args...)
  177. }
  178. }
  179. func (sc *serverConn) onNewHeaderField(f hpack.HeaderField) {
  180. sc.serveG.check()
  181. switch {
  182. case !validHeader(f.Name):
  183. sc.req.invalidHeader = true
  184. case strings.HasPrefix(f.Name, ":"):
  185. if sc.req.sawRegularHeader {
  186. sc.logf("pseudo-header after regular header")
  187. sc.req.invalidHeader = true
  188. return
  189. }
  190. var dst *string
  191. switch f.Name {
  192. case ":method":
  193. dst = &sc.req.method
  194. case ":path":
  195. dst = &sc.req.path
  196. case ":scheme":
  197. dst = &sc.req.scheme
  198. case ":authority":
  199. dst = &sc.req.authority
  200. default:
  201. // 8.1.2.1 Pseudo-Header Fields
  202. // "Endpoints MUST treat a request or response
  203. // that contains undefined or invalid
  204. // pseudo-header fields as malformed (Section
  205. // 8.1.2.6)."
  206. sc.logf("invalid pseudo-header %q", f.Name)
  207. sc.req.invalidHeader = true
  208. return
  209. }
  210. if *dst != "" {
  211. sc.logf("duplicate pseudo-header %q sent", f.Name)
  212. sc.req.invalidHeader = true
  213. return
  214. }
  215. *dst = f.Value
  216. case f.Name == "cookie":
  217. sc.req.sawRegularHeader = true
  218. if s, ok := sc.req.header["Cookie"]; ok && len(s) == 1 {
  219. s[0] = s[0] + "; " + f.Value
  220. } else {
  221. sc.req.header.Add("Cookie", f.Value)
  222. }
  223. default:
  224. sc.req.sawRegularHeader = true
  225. sc.req.header.Add(sc.canonicalHeader(f.Name), f.Value)
  226. }
  227. }
  228. func (sc *serverConn) canonicalHeader(v string) string {
  229. sc.serveG.check()
  230. // TODO: use a sync.Pool instead of putting the cache on *serverConn?
  231. cv, ok := sc.canonHeader[v]
  232. if !ok {
  233. cv = http.CanonicalHeaderKey(v)
  234. sc.canonHeader[v] = cv
  235. }
  236. return cv
  237. }
  238. // readFrames is the loop that reads incoming frames.
  239. // It's run on its own goroutine.
  240. func (sc *serverConn) readFrames() {
  241. processed := make(chan struct{}, 1)
  242. for {
  243. f, err := sc.framer.ReadFrame()
  244. if err != nil {
  245. close(sc.readFrameCh)
  246. sc.readFrameErrCh <- err
  247. return
  248. }
  249. sc.readFrameCh <- frameAndProcessed{f, processed}
  250. <-processed
  251. }
  252. }
  253. func (sc *serverConn) serve() {
  254. sc.serveG.check()
  255. defer sc.conn.Close()
  256. defer close(sc.doneServing)
  257. sc.vlogf("HTTP/2 connection from %v on %p", sc.conn.RemoteAddr(), sc.hs)
  258. // Read the client preface
  259. buf := make([]byte, len(ClientPreface))
  260. // TODO: timeout reading from the client
  261. if _, err := io.ReadFull(sc.conn, buf); err != nil {
  262. sc.logf("error reading client preface: %v", err)
  263. return
  264. }
  265. if !bytes.Equal(buf, clientPreface) {
  266. sc.logf("bogus greeting from client: %q", buf)
  267. return
  268. }
  269. sc.vlogf("client %v said hello", sc.conn.RemoteAddr())
  270. f, err := sc.framer.ReadFrame()
  271. if err != nil {
  272. sc.logf("error reading initial frame from client: %v", err)
  273. return
  274. }
  275. sf, ok := f.(*SettingsFrame)
  276. if !ok {
  277. sc.logf("invalid initial frame type %T received from client", f)
  278. return
  279. }
  280. if err := sf.ForeachSetting(sc.processSetting); err != nil {
  281. sc.logf("initial settings error: %v", err)
  282. return
  283. }
  284. // TODO: don't send two network packets for our SETTINGS + our
  285. // ACK of their settings. But if we make framer write to a
  286. // *bufio.Writer, that increases the per-connection memory
  287. // overhead, and there could be many idle conns. So maybe some
  288. // liveswitchWriter-like thing where we only switch to a
  289. // *bufio Writer when we really need one temporarily, else go
  290. // back to an unbuffered writes by default.
  291. if err := sc.framer.WriteSettings( /* TODO: actual settings */ ); err != nil {
  292. sc.logf("error writing server's initial settings: %v", err)
  293. return
  294. }
  295. if err := sc.framer.WriteSettingsAck(); err != nil {
  296. sc.logf("error writing server's ack of client's settings: %v", err)
  297. return
  298. }
  299. go sc.readFrames()
  300. for {
  301. select {
  302. case hr := <-sc.writeHeaderCh:
  303. if err := sc.writeHeaderInLoop(hr); err != nil {
  304. sc.condlogf(err, "error writing response header: %v", err)
  305. return
  306. }
  307. case fp, ok := <-sc.readFrameCh:
  308. if !ok {
  309. err := <-sc.readFrameErrCh
  310. if err != io.EOF {
  311. errstr := err.Error()
  312. if !strings.Contains(errstr, "use of closed network connection") {
  313. sc.logf("client %s stopped sending frames: %v", sc.conn.RemoteAddr(), errstr)
  314. }
  315. }
  316. return
  317. }
  318. f := fp.f
  319. sc.vlogf("got %v: %#v", f.Header(), f)
  320. err := sc.processFrame(f)
  321. fp.processed <- struct{}{} // let readFrames proceed
  322. switch ev := err.(type) {
  323. case nil:
  324. // nothing.
  325. case StreamError:
  326. if err := sc.resetStreamInLoop(ev); err != nil {
  327. sc.logf("Error writing RSTSTream: %v", err)
  328. return
  329. }
  330. case ConnectionError:
  331. sc.logf("Disconnecting; %v", ev)
  332. return
  333. case goAwayFlowError:
  334. if err := sc.goAway(ErrCodeFlowControl); err != nil {
  335. sc.condlogf(err, "failed to GOAWAY: %v", err)
  336. return
  337. }
  338. default:
  339. sc.logf("Disconnection due to other error: %v", err)
  340. return
  341. }
  342. }
  343. }
  344. }
  345. func (sc *serverConn) goAway(code ErrCode) error {
  346. sc.serveG.check()
  347. sc.sentGoAway = true
  348. return sc.framer.WriteGoAway(sc.maxStreamID, code, nil)
  349. }
  350. func (sc *serverConn) resetStreamInLoop(se StreamError) error {
  351. sc.serveG.check()
  352. if err := sc.framer.WriteRSTStream(se.streamID, uint32(se.code)); err != nil {
  353. return err
  354. }
  355. delete(sc.streams, se.streamID)
  356. return nil
  357. }
  358. func (sc *serverConn) curHeaderStreamID() uint32 {
  359. sc.serveG.check()
  360. st := sc.req.stream
  361. if st == nil {
  362. return 0
  363. }
  364. return st.id
  365. }
  366. func (sc *serverConn) processFrame(f Frame) error {
  367. sc.serveG.check()
  368. if s := sc.curHeaderStreamID(); s != 0 {
  369. if cf, ok := f.(*ContinuationFrame); !ok {
  370. return ConnectionError(ErrCodeProtocol)
  371. } else if cf.Header().StreamID != s {
  372. return ConnectionError(ErrCodeProtocol)
  373. }
  374. }
  375. switch f := f.(type) {
  376. case *SettingsFrame:
  377. return sc.processSettings(f)
  378. case *HeadersFrame:
  379. return sc.processHeaders(f)
  380. case *ContinuationFrame:
  381. return sc.processContinuation(f)
  382. case *WindowUpdateFrame:
  383. return sc.processWindowUpdate(f)
  384. case *PingFrame:
  385. return sc.processPing(f)
  386. case *DataFrame:
  387. return sc.processData(f)
  388. default:
  389. log.Printf("Ignoring unknown frame %#v", f)
  390. return nil
  391. }
  392. }
  393. func (sc *serverConn) processPing(f *PingFrame) error {
  394. sc.serveG.check()
  395. if f.Flags.Has(FlagSettingsAck) {
  396. // 6.7 PING: " An endpoint MUST NOT respond to PING frames
  397. // containing this flag."
  398. return nil
  399. }
  400. if f.StreamID != 0 {
  401. // "PING frames are not associated with any individual
  402. // stream. If a PING frame is received with a stream
  403. // identifier field value other than 0x0, the recipient MUST
  404. // respond with a connection error (Section 5.4.1) of type
  405. // PROTOCOL_ERROR."
  406. return ConnectionError(ErrCodeProtocol)
  407. }
  408. return sc.framer.WritePing(true, f.Data)
  409. }
  410. func (sc *serverConn) processWindowUpdate(f *WindowUpdateFrame) error {
  411. sc.serveG.check()
  412. switch {
  413. case f.StreamID != 0: // stream-level flow control
  414. st := sc.streams[f.StreamID]
  415. if st == nil {
  416. // "WINDOW_UPDATE can be sent by a peer that has sent a
  417. // frame bearing the END_STREAM flag. This means that a
  418. // receiver could receive a WINDOW_UPDATE frame on a "half
  419. // closed (remote)" or "closed" stream. A receiver MUST
  420. // NOT treat this as an error, see Section 5.1."
  421. return nil
  422. }
  423. if !st.flow.add(int32(f.Increment)) {
  424. return StreamError{f.StreamID, ErrCodeFlowControl}
  425. }
  426. default: // connection-level flow control
  427. if !sc.flow.add(int32(f.Increment)) {
  428. return goAwayFlowError{}
  429. }
  430. }
  431. return nil
  432. }
  433. func (sc *serverConn) processSettings(f *SettingsFrame) error {
  434. sc.serveG.check()
  435. return f.ForeachSetting(sc.processSetting)
  436. }
  437. func (sc *serverConn) processSetting(s Setting) error {
  438. sc.serveG.check()
  439. sc.vlogf("processing setting %v", s)
  440. switch s.ID {
  441. case SettingInitialWindowSize:
  442. return sc.processSettingInitialWindowSize(s.Val)
  443. }
  444. log.Printf("TODO: handle %v", s)
  445. return nil
  446. }
  447. func (sc *serverConn) processSettingInitialWindowSize(val uint32) error {
  448. sc.serveG.check()
  449. if val > (1<<31 - 1) {
  450. // 6.5.2 Defined SETTINGS Parameters
  451. // "Values above the maximum flow control window size of
  452. // 231-1 MUST be treated as a connection error (Section
  453. // 5.4.1) of type FLOW_CONTROL_ERROR."
  454. return ConnectionError(ErrCodeFlowControl)
  455. }
  456. // "A SETTINGS frame can alter the initial flow control window
  457. // size for all current streams. When the value of
  458. // SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST
  459. // adjust the size of all stream flow control windows that it
  460. // maintains by the difference between the new value and the
  461. // old value."
  462. old := sc.initialWindowSize
  463. sc.initialWindowSize = int32(val)
  464. growth := sc.initialWindowSize - old // may be negative
  465. for _, st := range sc.streams {
  466. if !st.flow.add(growth) {
  467. // 6.9.2 Initial Flow Control Window Size
  468. // "An endpoint MUST treat a change to
  469. // SETTINGS_INITIAL_WINDOW_SIZE that causes any flow
  470. // control window to exceed the maximum size as a
  471. // connection error (Section 5.4.1) of type
  472. // FLOW_CONTROL_ERROR."
  473. return ConnectionError(ErrCodeFlowControl)
  474. }
  475. }
  476. return nil
  477. }
  478. func (sc *serverConn) processData(f *DataFrame) error {
  479. sc.serveG.check()
  480. // "If a DATA frame is received whose stream is not in "open"
  481. // or "half closed (local)" state, the recipient MUST respond
  482. // with a stream error (Section 5.4.2) of type STREAM_CLOSED."
  483. id := f.Header().StreamID
  484. st, ok := sc.streams[id]
  485. if !ok || (st.state != stateOpen && st.state != stateHalfClosedLocal) {
  486. return StreamError{id, ErrCodeStreamClosed}
  487. }
  488. if st.body == nil {
  489. // Not expecting data.
  490. // TODO: which error code?
  491. return StreamError{id, ErrCodeStreamClosed}
  492. }
  493. data := f.Data()
  494. // Sender sending more than they'd declared?
  495. if st.declBodyBytes != -1 && st.bodyBytes+int64(len(data)) > st.declBodyBytes {
  496. st.body.Close(fmt.Errorf("Sender tried to send more than declared Content-Length of %d bytes", st.declBodyBytes))
  497. return StreamError{id, ErrCodeStreamClosed}
  498. }
  499. if len(data) > 0 {
  500. // TODO: verify they're allowed to write with the flow control
  501. // window we'd advertised to them.
  502. // TODO: verify n from Write
  503. if _, err := st.body.Write(data); err != nil {
  504. return StreamError{id, ErrCodeStreamClosed}
  505. }
  506. st.bodyBytes += int64(len(data))
  507. }
  508. if f.Header().Flags.Has(FlagDataEndStream) {
  509. if st.declBodyBytes != -1 && st.declBodyBytes != st.bodyBytes {
  510. st.body.Close(fmt.Errorf("Request declared a Content-Length of %d but only wrote %d bytes",
  511. st.declBodyBytes, st.bodyBytes))
  512. } else {
  513. st.body.Close(io.EOF)
  514. }
  515. }
  516. return nil
  517. }
  518. func (sc *serverConn) processHeaders(f *HeadersFrame) error {
  519. sc.serveG.check()
  520. id := f.Header().StreamID
  521. if sc.sentGoAway {
  522. // Ignore.
  523. return nil
  524. }
  525. // http://http2.github.io/http2-spec/#rfc.section.5.1.1
  526. if id%2 != 1 || id <= sc.maxStreamID || sc.req.stream != nil {
  527. // Streams initiated by a client MUST use odd-numbered
  528. // stream identifiers. [...] The identifier of a newly
  529. // established stream MUST be numerically greater than all
  530. // streams that the initiating endpoint has opened or
  531. // reserved. [...] An endpoint that receives an unexpected
  532. // stream identifier MUST respond with a connection error
  533. // (Section 5.4.1) of type PROTOCOL_ERROR.
  534. return ConnectionError(ErrCodeProtocol)
  535. }
  536. if id > sc.maxStreamID {
  537. sc.maxStreamID = id
  538. }
  539. st := &stream{
  540. id: id,
  541. state: stateOpen,
  542. flow: newFlow(sc.initialWindowSize),
  543. }
  544. if f.Header().Flags.Has(FlagHeadersEndStream) {
  545. st.state = stateHalfClosedRemote
  546. }
  547. sc.streams[id] = st
  548. sc.req = requestParam{
  549. stream: st,
  550. header: make(http.Header),
  551. }
  552. return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded())
  553. }
  554. func (sc *serverConn) processContinuation(f *ContinuationFrame) error {
  555. sc.serveG.check()
  556. st := sc.streams[f.Header().StreamID]
  557. if st == nil || sc.curHeaderStreamID() != st.id {
  558. return ConnectionError(ErrCodeProtocol)
  559. }
  560. return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded())
  561. }
  562. func (sc *serverConn) processHeaderBlockFragment(st *stream, frag []byte, end bool) error {
  563. sc.serveG.check()
  564. if _, err := sc.hpackDecoder.Write(frag); err != nil {
  565. // TODO: convert to stream error I assume?
  566. return err
  567. }
  568. if !end {
  569. return nil
  570. }
  571. if err := sc.hpackDecoder.Close(); err != nil {
  572. // TODO: convert to stream error I assume?
  573. return err
  574. }
  575. rw, req, err := sc.newWriterAndRequest()
  576. sc.req = requestParam{}
  577. if err != nil {
  578. return err
  579. }
  580. st.body = req.Body.(*requestBody).pipe // may be nil
  581. st.declBodyBytes = req.ContentLength
  582. go sc.runHandler(rw, req)
  583. return nil
  584. }
  585. func (sc *serverConn) newWriterAndRequest() (*responseWriter, *http.Request, error) {
  586. sc.serveG.check()
  587. rp := &sc.req
  588. if rp.invalidHeader || rp.method == "" || rp.path == "" ||
  589. (rp.scheme != "https" && rp.scheme != "http") {
  590. // See 8.1.2.6 Malformed Requests and Responses:
  591. //
  592. // Malformed requests or responses that are detected
  593. // MUST be treated as a stream error (Section 5.4.2)
  594. // of type PROTOCOL_ERROR."
  595. //
  596. // 8.1.2.3 Request Pseudo-Header Fields
  597. // "All HTTP/2 requests MUST include exactly one valid
  598. // value for the :method, :scheme, and :path
  599. // pseudo-header fields"
  600. return nil, nil, StreamError{rp.stream.id, ErrCodeProtocol}
  601. }
  602. var tlsState *tls.ConnectionState // make this non-nil if https
  603. if rp.scheme == "https" {
  604. // TODO: get from sc's ConnectionState
  605. tlsState = &tls.ConnectionState{}
  606. }
  607. authority := rp.authority
  608. if authority == "" {
  609. authority = rp.header.Get("Host")
  610. }
  611. bodyOpen := rp.stream.state == stateOpen
  612. body := &requestBody{
  613. sc: sc,
  614. streamID: rp.stream.id,
  615. }
  616. req := &http.Request{
  617. Method: rp.method,
  618. URL: &url.URL{},
  619. RemoteAddr: sc.conn.RemoteAddr().String(),
  620. Header: rp.header,
  621. RequestURI: rp.path,
  622. Proto: "HTTP/2.0",
  623. ProtoMajor: 2,
  624. ProtoMinor: 0,
  625. TLS: tlsState,
  626. Host: authority,
  627. Body: body,
  628. }
  629. if bodyOpen {
  630. body.pipe = &pipe{
  631. b: buffer{buf: make([]byte, 65536)}, // TODO: share/remove
  632. }
  633. body.pipe.c.L = &body.pipe.m
  634. if vv, ok := rp.header["Content-Length"]; ok {
  635. req.ContentLength, _ = strconv.ParseInt(vv[0], 10, 64)
  636. } else {
  637. req.ContentLength = -1
  638. }
  639. }
  640. rw := &responseWriter{
  641. sc: sc,
  642. streamID: rp.stream.id,
  643. req: req,
  644. body: body,
  645. }
  646. return rw, req, nil
  647. }
  648. // Run on its own goroutine.
  649. func (sc *serverConn) runHandler(rw *responseWriter, req *http.Request) {
  650. defer rw.handlerDone()
  651. // TODO: catch panics like net/http.Server
  652. sc.handler.ServeHTTP(rw, req)
  653. }
  654. // called from handler goroutines
  655. func (sc *serverConn) writeData(streamID uint32, p []byte) (n int, err error) {
  656. // TODO: implement
  657. log.Printf("WRITE on %d: %q", streamID, p)
  658. return len(p), nil
  659. }
  660. // headerWriteReq is a request to write an HTTP response header from a server Handler.
  661. type headerWriteReq struct {
  662. streamID uint32
  663. httpResCode int
  664. h http.Header // may be nil
  665. endStream bool
  666. }
  667. // called from handler goroutines.
  668. // h may be nil.
  669. func (sc *serverConn) writeHeader(req headerWriteReq) {
  670. sc.writeHeaderCh <- req
  671. }
  672. func (sc *serverConn) writeHeaderInLoop(req headerWriteReq) error {
  673. sc.serveG.check()
  674. sc.headerWriteBuf.Reset()
  675. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: ":status", Value: httpCodeString(req.httpResCode)})
  676. for k, vv := range req.h {
  677. for _, v := range vv {
  678. // TODO: for gargage, cache lowercase copies of headers at
  679. // least for common ones and/or popular recent ones for
  680. // this serverConn. LRU?
  681. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: strings.ToLower(k), Value: v})
  682. }
  683. }
  684. headerBlock := sc.headerWriteBuf.Bytes()
  685. if len(headerBlock) > int(sc.maxWriteFrameSize) {
  686. // we'll need continuation ones.
  687. panic("TODO")
  688. }
  689. return sc.framer.WriteHeaders(HeadersFrameParam{
  690. StreamID: req.streamID,
  691. BlockFragment: headerBlock,
  692. EndStream: req.endStream,
  693. EndHeaders: true, // no continuation yet
  694. })
  695. }
  696. // ConfigureServer adds HTTP/2 support to a net/http Server.
  697. //
  698. // The configuration conf may be nil.
  699. //
  700. // ConfigureServer must be called before s begins serving.
  701. func ConfigureServer(s *http.Server, conf *Server) {
  702. if conf == nil {
  703. conf = new(Server)
  704. }
  705. if s.TLSConfig == nil {
  706. s.TLSConfig = new(tls.Config)
  707. }
  708. haveNPN := false
  709. for _, p := range s.TLSConfig.NextProtos {
  710. if p == npnProto {
  711. haveNPN = true
  712. break
  713. }
  714. }
  715. if !haveNPN {
  716. s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, npnProto)
  717. }
  718. if s.TLSNextProto == nil {
  719. s.TLSNextProto = map[string]func(*http.Server, *tls.Conn, http.Handler){}
  720. }
  721. s.TLSNextProto[npnProto] = func(hs *http.Server, c *tls.Conn, h http.Handler) {
  722. if testHookOnConn != nil {
  723. testHookOnConn()
  724. }
  725. conf.handleConn(hs, c, h)
  726. }
  727. }
  728. type requestBody struct {
  729. sc *serverConn
  730. streamID uint32
  731. closed bool
  732. pipe *pipe // non-nil if we have a HTTP entity message body
  733. }
  734. var errClosedBody = errors.New("body closed by handler")
  735. func (b *requestBody) Close() error {
  736. if b.pipe != nil {
  737. b.pipe.Close(errClosedBody)
  738. }
  739. b.closed = true
  740. return nil
  741. }
  742. func (b *requestBody) Read(p []byte) (n int, err error) {
  743. if b.pipe == nil {
  744. return 0, io.EOF
  745. }
  746. n, err = b.pipe.Read(p)
  747. if n > 0 {
  748. // TODO: tell b.sc to send back 'n' flow control quota credits to the sender
  749. }
  750. return
  751. }
  752. type responseWriter struct {
  753. sc *serverConn
  754. streamID uint32
  755. wroteHeaders bool
  756. h http.Header
  757. req *http.Request
  758. body *requestBody // to close at end of request, if DATA frames didn't
  759. }
  760. // TODO: bufio writing of responseWriter. add Flush, add pools of
  761. // bufio.Writers, adjust bufio writer sized based on frame size
  762. // updates from peer? For now: naive.
  763. func (w *responseWriter) Header() http.Header {
  764. if w.h == nil {
  765. w.h = make(http.Header)
  766. }
  767. return w.h
  768. }
  769. func (w *responseWriter) WriteHeader(code int) {
  770. if w.wroteHeaders {
  771. return
  772. }
  773. // TODO: defer actually writing this frame until a Flush or
  774. // handlerDone, like net/http's Server. then we can coalesce
  775. // e.g. a 204 response to have a Header response frame with
  776. // END_STREAM set, without a separate frame being sent in
  777. // handleDone.
  778. w.wroteHeaders = true
  779. w.sc.writeHeader(headerWriteReq{
  780. streamID: w.streamID,
  781. httpResCode: code,
  782. h: w.h,
  783. })
  784. }
  785. // TODO: responseWriter.WriteString too?
  786. func (w *responseWriter) Write(p []byte) (n int, err error) {
  787. if !w.wroteHeaders {
  788. w.WriteHeader(200)
  789. }
  790. return w.sc.writeData(w.streamID, p) // blocks waiting for tokens
  791. }
  792. func (w *responseWriter) handlerDone() {
  793. if !w.wroteHeaders {
  794. w.sc.writeHeader(headerWriteReq{
  795. streamID: w.streamID,
  796. httpResCode: 200,
  797. h: w.h,
  798. endStream: true, // handler has finished; can't be any data.
  799. })
  800. }
  801. }
  802. var testHookOnConn func() // for testing
  803. func validHeader(v string) bool {
  804. if len(v) == 0 {
  805. return false
  806. }
  807. for _, r := range v {
  808. // "Just as in HTTP/1.x, header field names are
  809. // strings of ASCII characters that are compared in a
  810. // case-insensitive fashion. However, header field
  811. // names MUST be converted to lowercase prior to their
  812. // encoding in HTTP/2. "
  813. if r >= 127 || ('A' <= r && r <= 'Z') {
  814. return false
  815. }
  816. }
  817. return true
  818. }
  819. var httpCodeStringCommon = map[int]string{} // n -> strconv.Itoa(n)
  820. func init() {
  821. for i := 100; i <= 999; i++ {
  822. if v := http.StatusText(i); v != "" {
  823. httpCodeStringCommon[i] = strconv.Itoa(i)
  824. }
  825. }
  826. }
  827. func httpCodeString(code int) string {
  828. if s, ok := httpCodeStringCommon[code]; ok {
  829. return s
  830. }
  831. return strconv.Itoa(code)
  832. }