GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2024.10.1

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

428lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "bufio"
5 "context"
6 gojson "encoding/json"
7 "fmt"
8 "io"
9 "net"
10 "net/http"
11 "runtime/debug"
12 "strings"
13 "sync"
14
15 "github.com/pkg/errors"
16 "github.com/rs/zerolog"
17 "golang.org/x/net/http2"
18
19 "github.com/cloudflare/cloudflared/tracing"
20 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
21)
22
23// note: these constants are exported so we can reuse them in the edge-side code
24const (
25 InternalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
26 InternalTCPProxySrcHeader = "Cf-Cloudflared-Proxy-Src"
27 WebsocketUpgrade = "websocket"
28 ControlStreamUpgrade = "control-stream"
29 ConfigurationUpdate = "update-configuration"
30)
31
32var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
33
34// HTTP2Connection represents a net.Conn that uses HTTP2 frames to proxy traffic from the edge to cloudflared on the
35// origin.
36type HTTP2Connection struct {
37 conn net.Conn
38 server *http2.Server
39 orchestrator Orchestrator
40 connOptions *tunnelpogs.ConnectionOptions
41 observer *Observer
42 connIndex uint8
43
44 log *zerolog.Logger
45 activeRequestsWG sync.WaitGroup
46 controlStreamHandler ControlStreamHandler
47 stoppedGracefully bool
48 controlStreamErr error // result of running control stream handler
49}
50
51// NewHTTP2Connection returns a new instance of HTTP2Connection.
52func NewHTTP2Connection(
53 conn net.Conn,
54 orchestrator Orchestrator,
55 connOptions *tunnelpogs.ConnectionOptions,
56 observer *Observer,
57 connIndex uint8,
58 controlStreamHandler ControlStreamHandler,
59 log *zerolog.Logger,
60) *HTTP2Connection {
61 return &HTTP2Connection{
62 conn: conn,
63 server: &http2.Server{
64 MaxConcurrentStreams: MaxConcurrentStreams,
65 },
66 orchestrator: orchestrator,
67 connOptions: connOptions,
68 observer: observer,
69 connIndex: connIndex,
70 controlStreamHandler: controlStreamHandler,
71 log: log,
72 }
73}
74
75// Serve serves an HTTP2 server that the edge can talk to.
76func (c *HTTP2Connection) Serve(ctx context.Context) error {
77 go func() {
78 <-ctx.Done()
79 c.close()
80 }()
81 c.server.ServeConn(c.conn, &http2.ServeConnOpts{
82 Context: ctx,
83 Handler: c,
84 })
85
86 switch {
87 case c.controlStreamHandler.IsStopped():
88 return nil
89 case c.controlStreamErr != nil:
90 return c.controlStreamErr
91 default:
92 c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Lost connection with the edge")
93 return errEdgeConnectionClosed
94 }
95}
96
97func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
98 c.activeRequestsWG.Add(1)
99 defer c.activeRequestsWG.Done()
100
101 connType := determineHTTP2Type(r)
102 handleMissingRequestParts(connType, r)
103
104 respWriter, err := NewHTTP2RespWriter(r, w, connType, c.log)
105 if err != nil {
106 c.observer.log.Error().Msg(err.Error())
107 return
108 }
109
110 originProxy, err := c.orchestrator.GetOriginProxy()
111 if err != nil {
112 c.observer.log.Error().Msg(err.Error())
113 return
114 }
115
116 var requestErr error
117 switch connType {
118 case TypeControlStream:
119 requestErr = c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, c.orchestrator)
120 if requestErr != nil {
121 c.controlStreamErr = requestErr
122 }
123
124 case TypeConfiguration:
125 requestErr = c.handleConfigurationUpdate(respWriter, r)
126
127 case TypeWebsocket, TypeHTTP:
128 stripWebsocketUpgradeHeader(r)
129 // Check for tracing on request
130 tr := tracing.NewTracedHTTPRequest(r, c.connIndex, c.log)
131 if err := originProxy.ProxyHTTP(respWriter, tr, connType == TypeWebsocket); err != nil {
132 requestErr = fmt.Errorf("Failed to proxy HTTP: %w", err)
133 }
134
135 case TypeTCP:
136 host, err := getRequestHost(r)
137 if err != nil {
138 requestErr = fmt.Errorf(`cloudflared received a warp-routing request with an empty host value: %w`, err)
139 break
140 }
141
142 rws := NewHTTPResponseReadWriterAcker(respWriter, respWriter, r)
143 requestErr = originProxy.ProxyTCP(r.Context(), rws, &TCPRequest{
144 Dest: host,
145 CFRay: FindCfRayHeader(r),
146 LBProbe: IsLBProbeRequest(r),
147 CfTraceID: r.Header.Get(tracing.TracerContextName),
148 ConnIndex: c.connIndex,
149 })
150
151 default:
152 requestErr = fmt.Errorf("Received unknown connection type: %s", connType)
153 }
154
155 if requestErr != nil {
156 c.log.Error().Err(requestErr).Msg("failed to serve incoming request")
157
158 // WriteErrorResponse will return false if status was already written. we need to abort handler.
159 if !respWriter.WriteErrorResponse() {
160 c.log.Debug().Msg("Handler aborted due to failure to write error response after status already sent")
161 panic(http.ErrAbortHandler)
162 }
163 }
164}
165
166// ConfigurationUpdateBody is the representation followed by the edge to send updates to cloudflared.
167type ConfigurationUpdateBody struct {
168 Version int32 `json:"version"`
169 Config gojson.RawMessage `json:"config"`
170}
171
172func (c *HTTP2Connection) handleConfigurationUpdate(respWriter *http2RespWriter, r *http.Request) error {
173 var configBody ConfigurationUpdateBody
174 if err := json.NewDecoder(r.Body).Decode(&configBody); err != nil {
175 return err
176 }
177 resp := c.orchestrator.UpdateConfig(configBody.Version, configBody.Config)
178 bdy, err := json.Marshal(resp)
179 if err != nil {
180 return err
181 }
182 _, err = respWriter.Write(bdy)
183 return err
184}
185
186func (c *HTTP2Connection) close() {
187 // Wait for all serve HTTP handlers to return
188 c.activeRequestsWG.Wait()
189 c.conn.Close()
190}
191
192type http2RespWriter struct {
193 r io.Reader
194 w http.ResponseWriter
195 flusher http.Flusher
196 shouldFlush bool
197 statusWritten bool
198 respHeaders http.Header
199 hijackedMutex sync.Mutex
200 hijackedv bool
201 log *zerolog.Logger
202}
203
204func NewHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type, log *zerolog.Logger) (*http2RespWriter, error) {
205 flusher, isFlusher := w.(http.Flusher)
206 if !isFlusher {
207 respWriter := &http2RespWriter{
208 r: r.Body,
209 w: w,
210 log: log,
211 }
212 respWriter.WriteErrorResponse()
213 return nil, fmt.Errorf("%T doesn't implement http.Flusher", w)
214 }
215
216 return &http2RespWriter{
217 r: r.Body,
218 w: w,
219 flusher: flusher,
220 shouldFlush: connType.shouldFlush(),
221 respHeaders: make(http.Header),
222 log: log,
223 }, nil
224}
225
226func (rp *http2RespWriter) AddTrailer(trailerName, trailerValue string) {
227 if !rp.statusWritten {
228 rp.log.Warn().Msg("Tried to add Trailer to response before status written. Ignoring...")
229 return
230 }
231
232 rp.w.Header().Add(http2.TrailerPrefix+trailerName, trailerValue)
233}
234
235func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) error {
236 if rp.hijacked() {
237 rp.log.Warn().Msg("WriteRespHeaders after hijack")
238 return nil
239 }
240 dest := rp.w.Header()
241 userHeaders := make(http.Header, len(header))
242 for name, values := range header {
243 // lowercase headers for simplicity check
244 h2name := strings.ToLower(name)
245
246 if h2name == "content-length" {
247 // This header has meaning in HTTP/2 and will be used by the edge,
248 // so it should be sent *also* as an HTTP/2 response header.
249 dest[name] = values
250 }
251
252 if h2name == tracing.IntCloudflaredTracingHeader {
253 // Add cf-int-cloudflared-tracing header outside of serialized userHeaders
254 dest[tracing.CanonicalCloudflaredTracingHeader] = values
255 continue
256 }
257
258 if !IsControlResponseHeader(h2name) || IsWebsocketClientHeader(h2name) {
259 // User headers, on the other hand, must all be serialized so that
260 // HTTP/2 header validation won't be applied to HTTP/1 header values
261 userHeaders[name] = values
262 }
263 }
264
265 // Perform user header serialization and set them in the single header
266 dest.Set(CanonicalResponseUserHeaders, SerializeHeaders(userHeaders))
267
268 rp.setResponseMetaHeader(responseMetaHeaderOrigin)
269 // HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
270 if status == http.StatusSwitchingProtocols {
271 status = http.StatusOK
272 }
273 rp.w.WriteHeader(status)
274 if shouldFlush(header) {
275 rp.shouldFlush = true
276 }
277 if rp.shouldFlush {
278 rp.flusher.Flush()
279 }
280
281 rp.statusWritten = true
282 return nil
283}
284
285func (rp *http2RespWriter) Header() http.Header {
286 return rp.respHeaders
287}
288
289func (rp *http2RespWriter) Flush() {
290 rp.flusher.Flush()
291}
292
293func (rp *http2RespWriter) WriteHeader(status int) {
294 if rp.hijacked() {
295 rp.log.Warn().Msg("WriteHeader after hijack")
296 return
297 }
298 rp.WriteRespHeaders(status, rp.respHeaders)
299}
300
301func (rp *http2RespWriter) hijacked() bool {
302 rp.hijackedMutex.Lock()
303 defer rp.hijackedMutex.Unlock()
304 return rp.hijackedv
305}
306
307func (rp *http2RespWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
308 if !rp.statusWritten {
309 return nil, nil, fmt.Errorf("status not yet written before attempting to hijack connection")
310 }
311 // Make sure to flush anything left in the buffer before hijacking
312 if rp.shouldFlush {
313 rp.flusher.Flush()
314 }
315 rp.hijackedMutex.Lock()
316 defer rp.hijackedMutex.Unlock()
317 if rp.hijackedv {
318 return nil, nil, http.ErrHijacked
319 }
320 rp.hijackedv = true
321 conn := &localProxyConnection{rp}
322 // We return the http2RespWriter here because we want to make sure that we flush after every write
323 // otherwise the HTTP2 write buffer waits a few seconds before sending.
324 readWriter := bufio.NewReadWriter(
325 bufio.NewReader(rp),
326 bufio.NewWriter(rp),
327 )
328 return conn, readWriter, nil
329}
330
331func (rp *http2RespWriter) WriteErrorResponse() bool {
332 if rp.statusWritten {
333 return false
334 }
335
336 rp.setResponseMetaHeader(responseMetaHeaderCfd)
337 rp.w.WriteHeader(http.StatusBadGateway)
338 rp.statusWritten = true
339
340 return true
341}
342
343func (rp *http2RespWriter) setResponseMetaHeader(value string) {
344 rp.w.Header().Set(CanonicalResponseMetaHeader, value)
345}
346
347func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
348 return rp.r.Read(p)
349}
350
351func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
352 defer func() {
353 // Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
354 // Register a recover routine just in case.
355 if r := recover(); r != nil {
356 rp.log.Debug().Msgf("Recover from http2 response writer panic, error %s", debug.Stack())
357 }
358 }()
359 n, err = rp.w.Write(p)
360 if err == nil && rp.shouldFlush {
361 rp.flusher.Flush()
362 }
363 return n, err
364}
365
366func (rp *http2RespWriter) Close() error {
367 return nil
368}
369
370func determineHTTP2Type(r *http.Request) Type {
371 switch {
372 case isConfigurationUpdate(r):
373 return TypeConfiguration
374 case isWebsocketUpgrade(r):
375 return TypeWebsocket
376 case IsTCPStream(r):
377 return TypeTCP
378 case isControlStreamUpgrade(r):
379 return TypeControlStream
380 default:
381 return TypeHTTP
382 }
383}
384
385func handleMissingRequestParts(connType Type, r *http.Request) {
386 if connType == TypeHTTP {
387 // http library has no guarantees that we receive a filled URL. If not, then we fill it, as we reuse the request
388 // for proxying. For proxying they should not matter since we control the dialer on every egress proxied.
389 if len(r.URL.Scheme) == 0 {
390 r.URL.Scheme = "http"
391 }
392 if len(r.URL.Host) == 0 {
393 r.URL.Host = "localhost:8080"
394 }
395 }
396}
397
398func isControlStreamUpgrade(r *http.Request) bool {
399 return r.Header.Get(InternalUpgradeHeader) == ControlStreamUpgrade
400}
401
402func isWebsocketUpgrade(r *http.Request) bool {
403 return r.Header.Get(InternalUpgradeHeader) == WebsocketUpgrade
404}
405
406func isConfigurationUpdate(r *http.Request) bool {
407 return r.Header.Get(InternalUpgradeHeader) == ConfigurationUpdate
408}
409
410// IsTCPStream discerns if the connection request needs a tcp stream proxy.
411func IsTCPStream(r *http.Request) bool {
412 return r.Header.Get(InternalTCPProxySrcHeader) != ""
413}
414
415func stripWebsocketUpgradeHeader(r *http.Request) {
416 r.Header.Del(InternalUpgradeHeader)
417}
418
419// getRequestHost returns the host of the http.Request.
420func getRequestHost(r *http.Request) (string, error) {
421 if r.Host != "" {
422 return r.Host, nil
423 }
424 if r.URL != nil {
425 return r.URL.Host, nil
426 }
427 return "", errors.New("host not set in incoming request")
428}
429