GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2021.7.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

289lines · modecode

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