GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
a8ae6de213f88b22047f0ca6275751eb8bc00a85

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

244lines · 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/cloudflare/cloudflared/h2mux"
14 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
15
16 "github.com/rs/zerolog"
17 "golang.org/x/net/http2"
18)
19
20const (
21 internalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
22 websocketUpgrade = "websocket"
23 controlStreamUpgrade = "control-stream"
24)
25
26var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
27
28type http2Connection struct {
29 conn net.Conn
30 server *http2.Server
31 config *Config
32 namedTunnel *NamedTunnelConfig
33 connOptions *tunnelpogs.ConnectionOptions
34 observer *Observer
35 connIndexStr string
36 connIndex uint8
37 // newRPCClientFunc allows us to mock RPCs during testing
38 newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
39
40 activeRequestsWG sync.WaitGroup
41 connectedFuse ConnectedFuse
42 gracefulShutdownC <-chan struct{}
43 stoppedGracefully bool
44 controlStreamErr error // result of running control stream handler
45}
46
47func NewHTTP2Connection(
48 conn net.Conn,
49 config *Config,
50 namedTunnelConfig *NamedTunnelConfig,
51 connOptions *tunnelpogs.ConnectionOptions,
52 observer *Observer,
53 connIndex uint8,
54 connectedFuse ConnectedFuse,
55 gracefulShutdownC <-chan struct{},
56) *http2Connection {
57 return &http2Connection{
58 conn: conn,
59 server: &http2.Server{
60 MaxConcurrentStreams: math.MaxUint32,
61 },
62 config: config,
63 namedTunnel: namedTunnelConfig,
64 connOptions: connOptions,
65 observer: observer,
66 connIndexStr: uint8ToString(connIndex),
67 connIndex: connIndex,
68 newRPCClientFunc: newRegistrationRPCClient,
69 connectedFuse: connectedFuse,
70 gracefulShutdownC: gracefulShutdownC,
71 }
72}
73
74func (c *http2Connection) Serve(ctx context.Context) error {
75 go func() {
76 <-ctx.Done()
77 c.close()
78 }()
79 c.server.ServeConn(c.conn, &http2.ServeConnOpts{
80 Context: ctx,
81 Handler: c,
82 })
83
84 switch {
85 case c.stoppedGracefully:
86 return nil
87 case c.controlStreamErr != nil:
88 return c.controlStreamErr
89 default:
90 c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Lost connection with the edge")
91 return errEdgeConnectionClosed
92 }
93}
94
95func (c *http2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
96 c.activeRequestsWG.Add(1)
97 defer c.activeRequestsWG.Done()
98
99 respWriter := &http2RespWriter{
100 r: r.Body,
101 w: w,
102 }
103 flusher, isFlusher := w.(http.Flusher)
104 if !isFlusher {
105 c.observer.log.Error().Msgf("%T doesn't implement http.Flusher", w)
106 respWriter.WriteErrorResponse()
107 return
108 }
109 respWriter.flusher = flusher
110 var err error
111 if isControlStreamUpgrade(r) {
112 respWriter.shouldFlush = true
113 err = c.serveControlStream(r.Context(), respWriter)
114 c.controlStreamErr = err
115 } else if isWebsocketUpgrade(r) {
116 respWriter.shouldFlush = true
117 stripWebsocketUpgradeHeader(r)
118 err = c.config.OriginClient.Proxy(respWriter, r, true)
119 } else {
120 err = c.config.OriginClient.Proxy(respWriter, r, false)
121 }
122
123 if err != nil {
124 respWriter.WriteErrorResponse()
125 }
126}
127
128func (c *http2Connection) serveControlStream(ctx context.Context, respWriter *http2RespWriter) error {
129 rpcClient := c.newRPCClientFunc(ctx, respWriter, c.observer.log)
130 defer rpcClient.Close()
131
132 if err := rpcClient.RegisterConnection(ctx, c.namedTunnel, c.connOptions, c.connIndex, c.observer); err != nil {
133 return err
134 }
135 c.connectedFuse.Connected()
136
137 // wait for connection termination or start of graceful shutdown
138 select {
139 case <-ctx.Done():
140 break
141 case <-c.gracefulShutdownC:
142 c.stoppedGracefully = true
143 }
144
145 c.observer.sendUnregisteringEvent(c.connIndex)
146 rpcClient.GracefulShutdown(ctx, c.config.GracePeriod)
147 c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Unregistered tunnel connection")
148 return nil
149}
150
151func (c *http2Connection) close() {
152 // Wait for all serve HTTP handlers to return
153 c.activeRequestsWG.Wait()
154 c.conn.Close()
155}
156
157type http2RespWriter struct {
158 r io.Reader
159 w http.ResponseWriter
160 flusher http.Flusher
161 shouldFlush bool
162}
163
164func (rp *http2RespWriter) WriteRespHeaders(resp *http.Response) error {
165 dest := rp.w.Header()
166 userHeaders := make(http.Header, len(resp.Header))
167 for header, values := range resp.Header {
168 // Since these are http2 headers, they're required to be lowercase
169 h2name := strings.ToLower(header)
170 for _, v := range values {
171 if h2name == "content-length" {
172 // This header has meaning in HTTP/2 and will be used by the edge,
173 // so it should be sent as an HTTP/2 response header.
174 dest.Add(h2name, v)
175 // Since these are http2 headers, they're required to be lowercase
176 } else if !h2mux.IsControlHeader(h2name) || h2mux.IsWebsocketClientHeader(h2name) {
177 // User headers, on the other hand, must all be serialized so that
178 // HTTP/2 header validation won't be applied to HTTP/1 header values
179 userHeaders.Add(h2name, v)
180 }
181 }
182 }
183
184 // Perform user header serialization and set them in the single header
185 dest.Set(canonicalResponseUserHeadersField, h2mux.SerializeHeaders(userHeaders))
186 rp.setResponseMetaHeader(responseMetaHeaderOrigin)
187 status := resp.StatusCode
188 // HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
189 if status == http.StatusSwitchingProtocols {
190 status = http.StatusOK
191 }
192 rp.w.WriteHeader(status)
193 if IsServerSentEvent(resp.Header) {
194 rp.shouldFlush = true
195 }
196 if rp.shouldFlush {
197 rp.flusher.Flush()
198 }
199 return nil
200}
201
202func (rp *http2RespWriter) WriteErrorResponse() {
203 rp.setResponseMetaHeader(responseMetaHeaderCfd)
204 rp.w.WriteHeader(http.StatusBadGateway)
205}
206
207func (rp *http2RespWriter) setResponseMetaHeader(value string) {
208 rp.w.Header().Set(canonicalResponseMetaHeaderField, value)
209}
210
211func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
212 return rp.r.Read(p)
213}
214
215func (rp *http2RespWriter) Write(p []byte) (n int, err error) {
216 defer func() {
217 // Implementer of OriginClient should make sure it doesn't write to the connection after Proxy returns
218 // Register a recover routine just in case.
219 if r := recover(); r != nil {
220 println("Recover from http2 response writer panic, error", r)
221 }
222 }()
223 n, err = rp.w.Write(p)
224 if err == nil && rp.shouldFlush {
225 rp.flusher.Flush()
226 }
227 return n, err
228}
229
230func (rp *http2RespWriter) Close() error {
231 return nil
232}
233
234func isControlStreamUpgrade(r *http.Request) bool {
235 return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == controlStreamUpgrade
236}
237
238func isWebsocketUpgrade(r *http.Request) bool {
239 return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == websocketUpgrade
240}
241
242func stripWebsocketUpgradeHeader(r *http.Request) {
243 r.Header.Del(internalUpgradeHeader)
244}
245