GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

mirrored from https://github.com/cloudflare/cloudflaredAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2018.10.5

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

cloudflared/h2mux

h2mux/h2mux.go

434lines · modecode

RawDownload
Latest commit unavailable.
1package h2mux
2
3import (
4 "bytes"
5 "context"
6 "io"
7 "strings"
8 "sync"
9 "time"
10
11 log "github.com/sirupsen/logrus"
12 "golang.org/x/net/http2"
13 "golang.org/x/net/http2/hpack"
14 "golang.org/x/sync/errgroup"
15)
16
17const (
18 defaultFrameSize uint32 = 1 << 14 // Minimum frame size in http2 spec
19 defaultWindowSize uint32 = (1 << 16) - 1 // Minimum window size in http2 spec
20 maxWindowSize uint32 = (1 << 31) - 1 // 2^31-1 = 2147483647, max window size in http2 spec
21 defaultTimeout time.Duration = 5 * time.Second
22 defaultRetries uint64 = 5
23 defaultWriteBufferMaxLen int = 1024 * 1024 * 512 // 500mb
24
25 SettingMuxerMagic http2.SettingID = 0x42db
26 MuxerMagicOrigin uint32 = 0xa2e43c8b
27 MuxerMagicEdge uint32 = 0x1088ebf9
28)
29
30type MuxedStreamHandler interface {
31 ServeStream(*MuxedStream) error
32}
33
34type MuxedStreamFunc func(stream *MuxedStream) error
35
36func (f MuxedStreamFunc) ServeStream(stream *MuxedStream) error {
37 return f(stream)
38}
39
40type MuxerConfig struct {
41 Timeout time.Duration
42 Handler MuxedStreamHandler
43 IsClient bool
44 // Name is used to identify this muxer instance when logging.
45 Name string
46 // The minimum time this connection can be idle before sending a heartbeat.
47 HeartbeatInterval time.Duration
48 // The minimum number of heartbeats to send before terminating the connection.
49 MaxHeartbeats uint64
50 // Logger to use
51 Logger *log.Entry
52 CompressionQuality CompressionSetting
53 // Initial size for HTTP2 flow control windows
54 DefaultWindowSize uint32
55 // Largest allowable size for HTTP2 flow control windows
56 MaxWindowSize uint32
57 // Largest allowable capacity for the buffer of data to be sent
58 StreamWriteBufferMaxLen int
59}
60
61type Muxer struct {
62 // f is used to read and write HTTP2 frames on the wire.
63 f *http2.Framer
64 // config is the MuxerConfig given in Handshake.
65 config MuxerConfig
66 // w, r are references to the underlying connection used.
67 w io.WriteCloser
68 r io.ReadCloser
69 // muxReader is the read process.
70 muxReader *MuxReader
71 // muxWriter is the write process.
72 muxWriter *MuxWriter
73 // muxMetricsUpdater is the process to update metrics
74 muxMetricsUpdater *muxMetricsUpdater
75 // newStreamChan is used to create new streams on the writer thread.
76 // The writer will assign the next available stream ID.
77 newStreamChan chan MuxedStreamRequest
78 // abortChan is used to abort the writer event loop.
79 abortChan chan struct{}
80 // abortOnce is used to ensure abortChan is closed once only.
81 abortOnce sync.Once
82 // readyList is used to signal writable streams.
83 readyList *ReadyList
84 // streams tracks currently-open streams.
85 streams *activeStreamMap
86 // explicitShutdown records whether the Muxer is closing because Shutdown was called, or due to another
87 // error.
88 explicitShutdown *BooleanFuse
89
90 compressionQuality CompressionPreset
91}
92
93type Header struct {
94 Name, Value string
95}
96
97// Handshake establishes a muxed connection with the peer.
98// After the handshake completes, it is possible to open and accept streams.
99func Handshake(
100 w io.WriteCloser,
101 r io.ReadCloser,
102 config MuxerConfig,
103) (*Muxer, error) {
104 // Set default config values
105 if config.Timeout == 0 {
106 config.Timeout = defaultTimeout
107 }
108 if config.DefaultWindowSize == 0 {
109 config.DefaultWindowSize = defaultWindowSize
110 }
111 if config.MaxWindowSize == 0 {
112 config.MaxWindowSize = maxWindowSize
113 }
114 if config.StreamWriteBufferMaxLen == 0 {
115 config.StreamWriteBufferMaxLen = defaultWriteBufferMaxLen
116 }
117 // Initialise connection state fields
118 m := &Muxer{
119 f: http2.NewFramer(w, r), // A framer that writes to w and reads from r
120 config: config,
121 w: w,
122 r: r,
123 newStreamChan: make(chan MuxedStreamRequest),
124 abortChan: make(chan struct{}),
125 readyList: NewReadyList(),
126 streams: newActiveStreamMap(config.IsClient),
127 }
128
129 m.f.ReadMetaHeaders = hpack.NewDecoder(4096, func(hpack.HeaderField) {})
130 // Initialise the settings to identify this connection and confirm the other end is sane.
131 handshakeSetting := http2.Setting{ID: SettingMuxerMagic, Val: MuxerMagicEdge}
132 compressionSetting := http2.Setting{ID: SettingCompression, Val: config.CompressionQuality.toH2Setting()}
133 if CompressionIsSupported() {
134 log.Debug("Compression is supported")
135 m.compressionQuality = config.CompressionQuality.getPreset()
136 } else {
137 log.Debug("Compression is not supported")
138 compressionSetting = http2.Setting{ID: SettingCompression, Val: 0}
139 }
140
141 expectedMagic := MuxerMagicOrigin
142 if config.IsClient {
143 handshakeSetting.Val = MuxerMagicOrigin
144 expectedMagic = MuxerMagicEdge
145 }
146 errChan := make(chan error, 2)
147 // Simultaneously send our settings and verify the peer's settings.
148 go func() { errChan <- m.f.WriteSettings(handshakeSetting, compressionSetting) }()
149 go func() { errChan <- m.readPeerSettings(expectedMagic) }()
150 err := joinErrorsWithTimeout(errChan, 2, config.Timeout, ErrHandshakeTimeout)
151 if err != nil {
152 return nil, err
153 }
154 // Confirm sanity by ACKing the frame and expecting an ACK for our frame.
155 // Not strictly necessary, but let's pretend to be H2-like.
156 go func() { errChan <- m.f.WriteSettingsAck() }()
157 go func() { errChan <- m.readPeerSettingsAck() }()
158 err = joinErrorsWithTimeout(errChan, 2, config.Timeout, ErrHandshakeTimeout)
159 if err != nil {
160 return nil, err
161 }
162
163 // set up reader/writer pair ready for serve
164 streamErrors := NewStreamErrorMap()
165 goAwayChan := make(chan http2.ErrCode, 1)
166 updateRTTChan := make(chan *roundTripMeasurement, 1)
167 updateReceiveWindowChan := make(chan uint32, 1)
168 updateSendWindowChan := make(chan uint32, 1)
169 updateInBoundBytesChan := make(chan uint64)
170 updateOutBoundBytesChan := make(chan uint64)
171 inBoundCounter := NewAtomicCounter(0)
172 outBoundCounter := NewAtomicCounter(0)
173 pingTimestamp := NewPingTimestamp()
174 connActive := NewSignal()
175 idleDuration := config.HeartbeatInterval
176 // Sanity check to enusre idelDuration is sane
177 if idleDuration == 0 || idleDuration < defaultTimeout {
178 idleDuration = defaultTimeout
179 config.Logger.Warn("Minimum idle time has been adjusted to ", defaultTimeout)
180 }
181 maxRetries := config.MaxHeartbeats
182 if maxRetries == 0 {
183 maxRetries = defaultRetries
184 config.Logger.Warn("Minimum number of unacked heartbeats to send before closing the connection has been adjusted to ", maxRetries)
185 }
186
187 m.explicitShutdown = NewBooleanFuse()
188 m.muxReader = &MuxReader{
189 f: m.f,
190 handler: m.config.Handler,
191 streams: m.streams,
192 readyList: m.readyList,
193 streamErrors: streamErrors,
194 goAwayChan: goAwayChan,
195 abortChan: m.abortChan,
196 pingTimestamp: pingTimestamp,
197 connActive: connActive,
198 initialStreamWindow: m.config.DefaultWindowSize,
199 streamWindowMax: m.config.MaxWindowSize,
200 streamWriteBufferMaxLen: m.config.StreamWriteBufferMaxLen,
201 r: m.r,
202 updateRTTChan: updateRTTChan,
203 updateReceiveWindowChan: updateReceiveWindowChan,
204 updateSendWindowChan: updateSendWindowChan,
205 bytesRead: inBoundCounter,
206 updateInBoundBytesChan: updateInBoundBytesChan,
207 }
208 m.muxWriter = &MuxWriter{
209 f: m.f,
210 streams: m.streams,
211 streamErrors: streamErrors,
212 readyStreamChan: m.readyList.ReadyChannel(),
213 newStreamChan: m.newStreamChan,
214 goAwayChan: goAwayChan,
215 abortChan: m.abortChan,
216 pingTimestamp: pingTimestamp,
217 idleTimer: NewIdleTimer(idleDuration, maxRetries),
218 connActiveChan: connActive.WaitChannel(),
219 maxFrameSize: defaultFrameSize,
220 updateReceiveWindowChan: updateReceiveWindowChan,
221 updateSendWindowChan: updateSendWindowChan,
222 bytesWrote: outBoundCounter,
223 updateOutBoundBytesChan: updateOutBoundBytesChan,
224 }
225 m.muxWriter.headerEncoder = hpack.NewEncoder(&m.muxWriter.headerBuffer)
226
227 compBytesBefore, compBytesAfter := NewAtomicCounter(0), NewAtomicCounter(0)
228
229 m.muxMetricsUpdater = newMuxMetricsUpdater(
230 updateRTTChan,
231 updateReceiveWindowChan,
232 updateSendWindowChan,
233 updateInBoundBytesChan,
234 updateOutBoundBytesChan,
235 m.abortChan,
236 compBytesBefore,
237 compBytesAfter,
238 )
239
240 if m.compressionQuality.dictSize > 0 && m.compressionQuality.nDicts > 0 {
241 nd, sz := m.compressionQuality.nDicts, m.compressionQuality.dictSize
242 writeDicts, dictChan := newH2WriteDictionaries(
243 nd,
244 sz,
245 m.compressionQuality.quality,
246 compBytesBefore,
247 compBytesAfter,
248 )
249 readDicts := newH2ReadDictionaries(nd, sz)
250 m.muxReader.dictionaries = h2Dictionaries{read: &readDicts, write: writeDicts}
251 m.muxWriter.useDictChan = dictChan
252 }
253
254 return m, nil
255}
256
257func (m *Muxer) readPeerSettings(magic uint32) error {
258 frame, err := m.f.ReadFrame()
259 if err != nil {
260 return err
261 }
262 settingsFrame, ok := frame.(*http2.SettingsFrame)
263 if !ok {
264 return ErrBadHandshakeNotSettings
265 }
266 if settingsFrame.Header().Flags != 0 {
267 return ErrBadHandshakeUnexpectedAck
268 }
269 peerMagic, ok := settingsFrame.Value(SettingMuxerMagic)
270 if !ok {
271 return ErrBadHandshakeNoMagic
272 }
273 if magic != peerMagic {
274 return ErrBadHandshakeWrongMagic
275 }
276 peerCompression, ok := settingsFrame.Value(SettingCompression)
277 if !ok {
278 m.compressionQuality = compressionPresets[CompressionNone]
279 return nil
280 }
281 ver, fmt, sz, nd := parseCompressionSettingVal(peerCompression)
282 if ver != compressionVersion || fmt != compressionFormat || sz == 0 || nd == 0 {
283 m.compressionQuality = compressionPresets[CompressionNone]
284 return nil
285 }
286 // Values used for compression are the mimimum between the two peers
287 if sz < m.compressionQuality.dictSize {
288 m.compressionQuality.dictSize = sz
289 }
290 if nd < m.compressionQuality.nDicts {
291 m.compressionQuality.nDicts = nd
292 }
293 return nil
294}
295
296func (m *Muxer) readPeerSettingsAck() error {
297 frame, err := m.f.ReadFrame()
298 if err != nil {
299 return err
300 }
301 settingsFrame, ok := frame.(*http2.SettingsFrame)
302 if !ok {
303 return ErrBadHandshakeNotSettingsAck
304 }
305 if settingsFrame.Header().Flags != http2.FlagSettingsAck {
306 return ErrBadHandshakeUnexpectedSettings
307 }
308 return nil
309}
310
311func joinErrorsWithTimeout(errChan <-chan error, receiveCount int, timeout time.Duration, timeoutError error) error {
312 for i := 0; i < receiveCount; i++ {
313 select {
314 case err := <-errChan:
315 if err != nil {
316 return err
317 }
318 case <-time.After(timeout):
319 return timeoutError
320 }
321 }
322 return nil
323}
324
325func (m *Muxer) Serve(ctx context.Context) error {
326 errGroup, _ := errgroup.WithContext(ctx)
327 errGroup.Go(func() error {
328 err := m.muxReader.run(m.config.Logger)
329 m.explicitShutdown.Fuse(false)
330 m.r.Close()
331 m.abort()
332 return err
333 })
334
335 errGroup.Go(func() error {
336 err := m.muxWriter.run(m.config.Logger)
337 m.explicitShutdown.Fuse(false)
338 m.w.Close()
339 m.abort()
340 return err
341 })
342
343 errGroup.Go(func() error {
344 err := m.muxMetricsUpdater.run(m.config.Logger)
345 return err
346 })
347
348 err := errGroup.Wait()
349 if isUnexpectedTunnelError(err, m.explicitShutdown.Value()) {
350 return err
351 }
352 return nil
353}
354
355func (m *Muxer) Shutdown() {
356 m.explicitShutdown.Fuse(true)
357 m.muxReader.Shutdown()
358}
359
360// IsUnexpectedTunnelError identifies errors that are expected when shutting down the h2mux tunnel.
361// The set of expected errors change depending on whether we initiated shutdown or not.
362func isUnexpectedTunnelError(err error, expectedShutdown bool) bool {
363 if err == nil {
364 return false
365 }
366 if !expectedShutdown {
367 return true
368 }
369 return !isConnectionClosedError(err)
370}
371
372func isConnectionClosedError(err error) bool {
373 if err == io.EOF {
374 return true
375 }
376 if err == io.ErrClosedPipe {
377 return true
378 }
379 if err.Error() == "tls: use of closed connection" {
380 return true
381 }
382 if strings.HasSuffix(err.Error(), "use of closed network connection") {
383 return true
384 }
385 return false
386}
387
388// OpenStream opens a new data stream with the given headers.
389// Called by proxy server and tunnel
390func (m *Muxer) OpenStream(headers []Header, body io.Reader) (*MuxedStream, error) {
391 stream := &MuxedStream{
392 responseHeadersReceived: make(chan struct{}),
393 readBuffer: NewSharedBuffer(),
394 writeBuffer: &bytes.Buffer{},
395 writeBufferMaxLen: m.config.StreamWriteBufferMaxLen,
396 writeBufferHasSpace: make(chan struct{}, 1),
397 receiveWindow: m.config.DefaultWindowSize,
398 receiveWindowCurrentMax: m.config.DefaultWindowSize,
399 receiveWindowMax: m.config.MaxWindowSize,
400 sendWindow: m.config.DefaultWindowSize,
401 readyList: m.readyList,
402 writeHeaders: headers,
403 dictionaries: m.muxReader.dictionaries,
404 }
405
406 select {
407 // Will be received by mux writer
408 case m.newStreamChan <- MuxedStreamRequest{stream: stream, body: body}:
409 case <-m.abortChan:
410 return nil, ErrConnectionClosed
411 }
412 select {
413 case <-stream.responseHeadersReceived:
414 return stream, nil
415 case <-m.abortChan:
416 return nil, ErrConnectionClosed
417 }
418}
419
420func (m *Muxer) Metrics() *MuxerMetrics {
421 return m.muxMetricsUpdater.Metrics()
422}
423
424func (m *Muxer) abort() {
425 m.abortOnce.Do(func() {
426 close(m.abortChan)
427 m.streams.Abort()
428 })
429}
430
431// Return how many retries/ticks since the connection was last marked active
432func (m *Muxer) TimerRetries() uint64 {
433 return m.muxWriter.idleTimer.RetryCount()
434}
435