GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
6b86f81c4a158b0e65043cfaf84231c5fb1ea90c

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2.go

248lines · modecode

RawDownload
TUN-3458: Upgrade to http2 when available, fallback to h2mux when we reach max retriesa490443
1package connection
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "math"
8 "net"
9 "net/http"
10 "net/url"
11 "strings"
12 "sync"
13
14 "github.com/cloudflare/cloudflared/h2mux"
15 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
16
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
26type HTTP2Connection struct {
27 conn net.Conn
28 server *http2.Server
29 config *Config
30 originURL *url.URL
31 namedTunnel *NamedTunnelConfig
32 connOptions *tunnelpogs.ConnectionOptions
33 observer *Observer
34 connIndexStr string
35 connIndex uint8
36 wg *sync.WaitGroup
37 connectedFuse ConnectedFuse
38}
39
40func NewHTTP2Connection(conn net.Conn, config *Config, originURL *url.URL, namedTunnelConfig *NamedTunnelConfig, connOptions *tunnelpogs.ConnectionOptions, observer *Observer, connIndex uint8, connectedFuse ConnectedFuse) *HTTP2Connection {
41 return &HTTP2Connection{
42 conn: conn,
43 server: &http2.Server{
44 MaxConcurrentStreams: math.MaxUint32,
45 },
46 config: config,
47 originURL: originURL,
48 namedTunnel: namedTunnelConfig,
49 connOptions: connOptions,
50 observer: observer,
51 connIndexStr: uint8ToString(connIndex),
52 connIndex: connIndex,
53 wg: &sync.WaitGroup{},
54 connectedFuse: connectedFuse,
55 }
56}
57
58func (c *HTTP2Connection) Serve(ctx context.Context) {
59 go func() {
60 <-ctx.Done()
61 c.close()
62 }()
63 c.server.ServeConn(c.conn, &http2.ServeConnOpts{
64 Context: ctx,
65 Handler: c,
66 })
67}
68
69func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
70 c.wg.Add(1)
71 defer c.wg.Done()
72
73 r.URL.Scheme = c.originURL.Scheme
74 r.URL.Host = c.originURL.Host
75
76 respWriter := &http2RespWriter{
77 r: r.Body,
78 w: w,
79 }
80 if isControlStreamUpgrade(r) {
81 err := c.serveControlStream(r.Context(), respWriter)
82 if err != nil {
83 respWriter.WriteErrorResponse(err)
84 }
85 } else if isWebsocketUpgrade(r) {
86 wsRespWriter, err := newWSRespWriter(respWriter)
87 if err != nil {
88 respWriter.WriteErrorResponse(err)
89 return
90 }
91 stripWebsocketUpgradeHeader(r)
92 c.config.OriginClient.Proxy(wsRespWriter, r, true)
93 } else {
94 c.config.OriginClient.Proxy(respWriter, r, false)
95 }
96}
97
98func (c *HTTP2Connection) serveControlStream(ctx context.Context, h2RespWriter *http2RespWriter) error {
99 stream, err := newWSRespWriter(h2RespWriter)
100 if err != nil {
101 return err
102 }
103
104 rpcClient := newRegistrationRPCClient(ctx, stream, c.observer)
105 defer rpcClient.close()
106
107 if err = registerConnection(ctx, rpcClient, c.namedTunnel, c.connOptions, c.connIndex, c.observer); err != nil {
108 return err
109 }
110 c.connectedFuse.Connected()
111
112 <-ctx.Done()
113 c.gracefulShutdown(ctx, rpcClient)
114 return nil
115}
116
117func (c *HTTP2Connection) registerConnection(
118 ctx context.Context,
119 rpcClient tunnelpogs.RegistrationServer_PogsClient,
120) error {
121 connDetail, err := rpcClient.RegisterConnection(
122 ctx,
123 c.namedTunnel.Auth,
124 c.namedTunnel.ID,
125 c.connIndex,
126 c.connOptions,
127 )
128 if err != nil {
129 c.observer.Errorf("Cannot register connection, err: %v", err)
130 return err
131 }
132 c.observer.Infof("Connection %s registered with %s using ID %s", c.connIndexStr, connDetail.Location, connDetail.UUID)
133 return nil
134}
135
136func (c *HTTP2Connection) gracefulShutdown(ctx context.Context, rpcClient *registrationServerClient) {
137 ctx, cancel := context.WithTimeout(ctx, c.config.GracePeriod)
138 defer cancel()
139 rpcClient.client.UnregisterConnection(ctx)
140}
141
142func (c *HTTP2Connection) close() {
143 // Wait for all serve HTTP handlers to return
144 c.wg.Wait()
145 c.conn.Close()
146}
147
148type http2RespWriter struct {
149 r io.Reader
150 w http.ResponseWriter
151}
152
153func (rp *http2RespWriter) WriteRespHeaders(resp *http.Response) error {
154 dest := rp.w.Header()
155 userHeaders := make(http.Header, len(resp.Header))
156 for header, values := range resp.Header {
157 // Since these are http2 headers, they're required to be lowercase
158 h2name := strings.ToLower(header)
159 for _, v := range values {
160 if h2name == "content-length" {
161 // This header has meaning in HTTP/2 and will be used by the edge,
162 // so it should be sent as an HTTP/2 response header.
163 dest.Add(h2name, v)
164 // Since these are http2 headers, they're required to be lowercase
165 } else if !h2mux.IsControlHeader(h2name) || h2mux.IsWebsocketClientHeader(h2name) {
166 // User headers, on the other hand, must all be serialized so that
167 // HTTP/2 header validation won't be applied to HTTP/1 header values
168 userHeaders.Add(h2name, v)
169 }
170 }
171 }
172
173 // Perform user header serialization and set them in the single header
174 dest.Set(canonicalResponseUserHeadersField, h2mux.SerializeHeaders(userHeaders))
175 rp.setResponseMetaHeader(responseMetaHeaderCfd)
176 status := resp.StatusCode
177 // HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
178 if status == http.StatusSwitchingProtocols {
179 status = http.StatusOK
180 }
181 rp.w.WriteHeader(status)
182 return nil
183}
184
185func (rp *http2RespWriter) WriteErrorResponse(err error) {
186 rp.setResponseMetaHeader(responseMetaHeaderCfd)
187 rp.w.WriteHeader(http.StatusBadGateway)
188}
189
190func (rp *http2RespWriter) setResponseMetaHeader(value string) {
191 rp.w.Header().Set(canonicalResponseMetaHeaderField, value)
192}
193
194func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
195 return rp.r.Read(p)
196}
197
198func (wr *http2RespWriter) Write(p []byte) (n int, err error) {
199 return wr.w.Write(p)
200}
201
202type wsRespWriter struct {
203 *http2RespWriter
204 flusher http.Flusher
205}
206
207func newWSRespWriter(h2 *http2RespWriter) (*wsRespWriter, error) {
208 flusher, ok := h2.w.(http.Flusher)
209 if !ok {
210 return nil, fmt.Errorf("ResponseWriter doesn't implement http.Flusher")
211 }
212 return &wsRespWriter{
213 h2,
214 flusher,
215 }, nil
216}
217
218func (rw *wsRespWriter) WriteRespHeaders(resp *http.Response) (err error) {
219 err = rw.http2RespWriter.WriteRespHeaders(resp)
220 if err == nil {
221 rw.flusher.Flush()
222 }
223 return
224}
225
226func (rw *wsRespWriter) Write(p []byte) (n int, err error) {
227 n, err = rw.http2RespWriter.Write(p)
228 if err == nil {
229 rw.flusher.Flush()
230 }
231 return
232}
233
234func (rw *wsRespWriter) Close() error {
235 return nil
236}
237
238func isControlStreamUpgrade(r *http.Request) bool {
239 return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == controlStreamUpgrade
240}
241
242func isWebsocketUpgrade(r *http.Request) bool {
243 return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == websocketUpgrade
244}
245
246func stripWebsocketUpgradeHeader(r *http.Request) {
247 r.Header.Del(internalUpgradeHeader)
248}