GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2025.2.1

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

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