GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
d7da74cb9ee92ec09c110d93b0df9233b8f24d37

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

302lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "math"
8 "net"
9 "net/http"
10 "runtime/debug"
11 "strings"
12 "sync"
13
14 "github.com/pkg/errors"
15 "github.com/rs/zerolog"
16 "golang.org/x/net/http2"
17
18 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
19)
20
21// note: these constants are exported so we can reuse them in the edge-side code
22const (
23 InternalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
24 InternalTCPProxySrcHeader = "Cf-Cloudflared-Proxy-Src"
25 WebsocketUpgrade = "websocket"
26 ControlStreamUpgrade = "control-stream"
27)
28
29var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
30
31// HTTP2Connection represents a net.Conn that uses HTTP2 frames to proxy traffic from the edge to cloudflared on the
32// origin.
33type HTTP2Connection struct {
34 conn net.Conn
35 server *http2.Server
36 config *Config
37 connOptions *tunnelpogs.ConnectionOptions
38 observer *Observer
39 connIndex uint8
40 // newRPCClientFunc allows us to mock RPCs during testing
41 newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
42
43 log *zerolog.Logger
44 activeRequestsWG sync.WaitGroup
45 controlStreamHandler ControlStreamHandler
46 stoppedGracefully bool
47 controlStreamErr error // result of running control stream handler
48}
49
50// NewHTTP2Connection returns a new instance of HTTP2Connection.
51func NewHTTP2Connection(
52 conn net.Conn,
53 config *Config,
54 connOptions *tunnelpogs.ConnectionOptions,
55 observer *Observer,
56 connIndex uint8,
57 controlStreamHandler ControlStreamHandler,
58 log *zerolog.Logger,
59) *HTTP2Connection {
60 return &HTTP2Connection{
61 conn: conn,
62 server: &http2.Server{
63 MaxConcurrentStreams: math.MaxUint32,
64 },
65 config: config,
66 connOptions: connOptions,
67 observer: observer,
68 connIndex: connIndex,
69 newRPCClientFunc: newRegistrationRPCClient,
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)
105 if err != nil {
106 c.observer.log.Error().Msg(err.Error())
107 return
108 }
109
110 switch connType {
111 case TypeControlStream:
112 if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, true); err != nil {
113 c.controlStreamErr = err
114 c.log.Error().Err(err)
115 respWriter.WriteErrorResponse()
116 }
117
118 case TypeWebsocket, TypeHTTP:
119 stripWebsocketUpgradeHeader(r)
120 if err := c.config.OriginProxy.ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
121 err := fmt.Errorf("Failed to proxy HTTP: %w", err)
122 c.log.Error().Err(err)
123 respWriter.WriteErrorResponse()
124 }
125
126 case TypeTCP:
127 host, err := getRequestHost(r)
128 if err != nil {
129 err := fmt.Errorf(`cloudflared recieved a warp-routing request with an empty host value: %w`, err)
130 c.log.Error().Err(err)
131 respWriter.WriteErrorResponse()
132 }
133
134 rws := NewHTTPResponseReadWriterAcker(respWriter, r)
135 if err := c.config.OriginProxy.ProxyTCP(r.Context(), rws, &TCPRequest{
136 Dest: host,
137 CFRay: FindCfRayHeader(r),
138 LBProbe: IsLBProbeRequest(r),
139 }); err != nil {
140 respWriter.WriteErrorResponse()
141 }
142
143 default:
144 err := fmt.Errorf("Received unknown connection type: %s", connType)
145 c.log.Error().Err(err)
146 respWriter.WriteErrorResponse()
147 }
148}
149
150func (c *HTTP2Connection) close() {
151 // Wait for all serve HTTP handlers to return
152 c.activeRequestsWG.Wait()
153 c.conn.Close()
154}
155
156type http2RespWriter struct {
157 r io.Reader
158 w http.ResponseWriter
159 flusher http.Flusher
160 shouldFlush bool
161}
162
163func NewHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (*http2RespWriter, error) {
164 flusher, isFlusher := w.(http.Flusher)
165 if !isFlusher {
166 respWriter := &http2RespWriter{
167 r: r.Body,
168 w: w,
169 }
170 respWriter.WriteErrorResponse()
171 return nil, fmt.Errorf("%T doesn't implement http.Flusher", w)
172 }
173
174 return &http2RespWriter{
175 r: r.Body,
176 w: w,
177 flusher: flusher,
178 shouldFlush: connType.shouldFlush(),
179 }, nil
180}
181
182func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) error {
183 dest := rp.w.Header()
184 userHeaders := make(http.Header, len(header))
185 for name, values := range header {
186 // Since these are http2 headers, they're required to be lowercase
187 h2name := strings.ToLower(name)
188 if h2name == "content-length" {
189 // This header has meaning in HTTP/2 and will be used by the edge,
190 // so it should be sent as an HTTP/2 response header.
191 dest[name] = values
192 // Since these are http2 headers, they're required to be lowercase
193 } else if !IsControlResponseHeader(h2name) || IsWebsocketClientHeader(h2name) {
194 // User headers, on the other hand, must all be serialized so that
195 // HTTP/2 header validation won't be applied to HTTP/1 header values
196 userHeaders[name] = values
197 }
198 }
199
200 // Perform user header serialization and set them in the single header
201 dest.Set(CanonicalResponseUserHeaders, SerializeHeaders(userHeaders))
202 rp.setResponseMetaHeader(responseMetaHeaderOrigin)
203 // HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
204 if status == http.StatusSwitchingProtocols {
205 status = http.StatusOK
206 }
207 rp.w.WriteHeader(status)
208 if IsServerSentEvent(header) {
209 rp.shouldFlush = true
210 }
211 if rp.shouldFlush {
212 rp.flusher.Flush()
213 }
214 return nil
215}
216
217func (rp *http2RespWriter) WriteErrorResponse() {
218 rp.setResponseMetaHeader(responseMetaHeaderCfd)
219 rp.w.WriteHeader(http.StatusBadGateway)
220}
221
222func (rp *http2RespWriter) setResponseMetaHeader(value string) {
223 rp.w.Header().Set(CanonicalResponseMetaHeader, value)
224}
225
226func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
227 return rp.r.Read(p)
228}
229
230func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
231 defer func() {
232 // Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
233 // Register a recover routine just in case.
234 if r := recover(); r != nil {
235 println(fmt.Sprintf("Recover from http2 response writer panic, error %s", debug.Stack()))
236 }
237 }()
238 n, err = rp.w.Write(p)
239 if err == nil && rp.shouldFlush {
240 rp.flusher.Flush()
241 }
242 return n, err
243}
244
245func (rp *http2RespWriter) Close() error {
246 return nil
247}
248
249func determineHTTP2Type(r *http.Request) Type {
250 switch {
251 case isWebsocketUpgrade(r):
252 return TypeWebsocket
253 case IsTCPStream(r):
254 return TypeTCP
255 case isControlStreamUpgrade(r):
256 return TypeControlStream
257 default:
258 return TypeHTTP
259 }
260}
261
262func handleMissingRequestParts(connType Type, r *http.Request) {
263 if connType == TypeHTTP {
264 // http library has no guarantees that we receive a filled URL. If not, then we fill it, as we reuse the request
265 // for proxying. We use the same values as we used to in h2mux. For proxying they should not matter since we
266 // control the dialer on every egress proxied.
267 if len(r.URL.Scheme) == 0 {
268 r.URL.Scheme = "http"
269 }
270 if len(r.URL.Host) == 0 {
271 r.URL.Host = "localhost:8080"
272 }
273 }
274}
275
276func isControlStreamUpgrade(r *http.Request) bool {
277 return r.Header.Get(InternalUpgradeHeader) == ControlStreamUpgrade
278}
279
280func isWebsocketUpgrade(r *http.Request) bool {
281 return r.Header.Get(InternalUpgradeHeader) == WebsocketUpgrade
282}
283
284// IsTCPStream discerns if the connection request needs a tcp stream proxy.
285func IsTCPStream(r *http.Request) bool {
286 return r.Header.Get(InternalTCPProxySrcHeader) != ""
287}
288
289func stripWebsocketUpgradeHeader(r *http.Request) {
290 r.Header.Del(InternalUpgradeHeader)
291}
292
293// getRequestHost returns the host of the http.Request.
294func getRequestHost(r *http.Request) (string, error) {
295 if r.Host != "" {
296 return r.Host, nil
297 }
298 if r.URL != nil {
299 return r.URL.Host, nil
300 }
301 return "", errors.New("host not set in incoming request")
302}
303