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/h2mux.go

261lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "io"
6 "net"
7 "net/http"
8 "time"
9
10 "github.com/pkg/errors"
11 "github.com/rs/zerolog"
12 "golang.org/x/sync/errgroup"
13
14 "github.com/cloudflare/cloudflared/h2mux"
15 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
16 "github.com/cloudflare/cloudflared/websocket"
17)
18
19const (
20 muxerTimeout = 5 * time.Second
21 openStreamTimeout = 30 * time.Second
22)
23
24type h2muxConnection struct {
25 config *Config
26 muxerConfig *MuxerConfig
27 muxer *h2mux.Muxer
28 // connectionID is only used by metrics, and prometheus requires labels to be string
29 connIndexStr string
30 connIndex uint8
31
32 observer *Observer
33 gracefulShutdownC <-chan struct{}
34 stoppedGracefully bool
35
36 // newRPCClientFunc allows us to mock RPCs during testing
37 newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
38}
39
40type MuxerConfig struct {
41 HeartbeatInterval time.Duration
42 MaxHeartbeats uint64
43 CompressionSetting h2mux.CompressionSetting
44 MetricsUpdateFreq time.Duration
45}
46
47func (mc *MuxerConfig) H2MuxerConfig(h h2mux.MuxedStreamHandler, log *zerolog.Logger) *h2mux.MuxerConfig {
48 return &h2mux.MuxerConfig{
49 Timeout: muxerTimeout,
50 Handler: h,
51 IsClient: true,
52 HeartbeatInterval: mc.HeartbeatInterval,
53 MaxHeartbeats: mc.MaxHeartbeats,
54 Log: log,
55 CompressionQuality: mc.CompressionSetting,
56 }
57}
58
59// NewTunnelHandler returns a TunnelHandler, origin LAN IP and error
60func NewH2muxConnection(
61 config *Config,
62 muxerConfig *MuxerConfig,
63 edgeConn net.Conn,
64 connIndex uint8,
65 observer *Observer,
66 gracefulShutdownC <-chan struct{},
67) (*h2muxConnection, error, bool) {
68 h := &h2muxConnection{
69 config: config,
70 muxerConfig: muxerConfig,
71 connIndexStr: uint8ToString(connIndex),
72 connIndex: connIndex,
73 observer: observer,
74 gracefulShutdownC: gracefulShutdownC,
75 newRPCClientFunc: newRegistrationRPCClient,
76 }
77
78 // Establish a muxed connection with the edge
79 // Client mux handshake with agent server
80 muxer, err := h2mux.Handshake(edgeConn, edgeConn, *muxerConfig.H2MuxerConfig(h, observer.logTransport), h2mux.ActiveStreams)
81 if err != nil {
82 recoverable := isHandshakeErrRecoverable(err, connIndex, observer)
83 return nil, err, recoverable
84 }
85 h.muxer = muxer
86 return h, nil, false
87}
88
89func (h *h2muxConnection) ServeNamedTunnel(ctx context.Context, namedTunnel *NamedTunnelConfig, connOptions *tunnelpogs.ConnectionOptions, connectedFuse ConnectedFuse) error {
90 errGroup, serveCtx := errgroup.WithContext(ctx)
91 errGroup.Go(func() error {
92 return h.serveMuxer(serveCtx)
93 })
94
95 errGroup.Go(func() error {
96 if err := h.registerNamedTunnel(serveCtx, namedTunnel, connOptions); err != nil {
97 return err
98 }
99 connectedFuse.Connected()
100 return nil
101 })
102
103 errGroup.Go(func() error {
104 h.controlLoop(serveCtx, connectedFuse, true)
105 return nil
106 })
107
108 err := errGroup.Wait()
109 if err == errMuxerStopped {
110 if h.stoppedGracefully {
111 return nil
112 }
113 h.observer.log.Info().Uint8(LogFieldConnIndex, h.connIndex).Msg("Unexpected muxer shutdown")
114 }
115 return err
116}
117
118func (h *h2muxConnection) ServeClassicTunnel(ctx context.Context, classicTunnel *ClassicTunnelConfig, credentialManager CredentialManager, registrationOptions *tunnelpogs.RegistrationOptions, connectedFuse ConnectedFuse) error {
119 errGroup, serveCtx := errgroup.WithContext(ctx)
120 errGroup.Go(func() error {
121 return h.serveMuxer(serveCtx)
122 })
123
124 errGroup.Go(func() (err error) {
125 defer func() {
126 if err == nil {
127 connectedFuse.Connected()
128 }
129 }()
130 if classicTunnel.UseReconnectToken && connectedFuse.IsConnected() {
131 err := h.reconnectTunnel(ctx, credentialManager, classicTunnel, registrationOptions)
132 if err == nil {
133 return nil
134 }
135 // log errors and proceed to RegisterTunnel
136 h.observer.log.Err(err).
137 Uint8(LogFieldConnIndex, h.connIndex).
138 Msg("Couldn't reconnect connection. Re-registering it instead.")
139 }
140 return h.registerTunnel(ctx, credentialManager, classicTunnel, registrationOptions)
141 })
142
143 errGroup.Go(func() error {
144 h.controlLoop(serveCtx, connectedFuse, false)
145 return nil
146 })
147
148 err := errGroup.Wait()
149 if err == errMuxerStopped {
150 if h.stoppedGracefully {
151 return nil
152 }
153 h.observer.log.Info().Uint8(LogFieldConnIndex, h.connIndex).Msg("Unexpected muxer shutdown")
154 }
155 return err
156}
157
158func (h *h2muxConnection) serveMuxer(ctx context.Context) error {
159 // All routines should stop when muxer finish serving. When muxer is shutdown
160 // gracefully, it doesn't return an error, so we need to return errMuxerShutdown
161 // here to notify other routines to stop
162 err := h.muxer.Serve(ctx)
163 if err == nil {
164 return errMuxerStopped
165 }
166 return err
167}
168
169func (h *h2muxConnection) controlLoop(ctx context.Context, connectedFuse ConnectedFuse, isNamedTunnel bool) {
170 updateMetricsTicker := time.NewTicker(h.muxerConfig.MetricsUpdateFreq)
171 defer updateMetricsTicker.Stop()
172 var shutdownCompleted <-chan struct{}
173 for {
174 select {
175 case <-h.gracefulShutdownC:
176 if connectedFuse.IsConnected() {
177 h.unregister(isNamedTunnel)
178 }
179 h.stoppedGracefully = true
180 h.gracefulShutdownC = nil
181 shutdownCompleted = h.muxer.Shutdown()
182
183 case <-shutdownCompleted:
184 return
185
186 case <-ctx.Done():
187 // UnregisterTunnel blocks until the RPC call returns
188 if !h.stoppedGracefully && connectedFuse.IsConnected() {
189 h.unregister(isNamedTunnel)
190 }
191 h.muxer.Shutdown()
192 // don't wait for shutdown to finish when context is closed, this is the hard termination path
193 return
194
195 case <-updateMetricsTicker.C:
196 h.observer.metrics.updateMuxerMetrics(h.connIndexStr, h.muxer.Metrics())
197 }
198 }
199}
200
201func (h *h2muxConnection) newRPCStream(ctx context.Context, rpcName rpcName) (*h2mux.MuxedStream, error) {
202 openStreamCtx, openStreamCancel := context.WithTimeout(ctx, openStreamTimeout)
203 defer openStreamCancel()
204 stream, err := h.muxer.OpenRPCStream(openStreamCtx)
205 if err != nil {
206 return nil, err
207 }
208 return stream, nil
209}
210
211func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
212 respWriter := &h2muxRespWriter{stream}
213
214 req, reqErr := h.newRequest(stream)
215 if reqErr != nil {
216 respWriter.WriteErrorResponse()
217 return reqErr
218 }
219
220 var sourceConnectionType = TypeHTTP
221 if websocket.IsWebSocketUpgrade(req) {
222 sourceConnectionType = TypeWebsocket
223 }
224
225 err := h.config.OriginProxy.Proxy(respWriter, req, sourceConnectionType)
226 if err != nil {
227 respWriter.WriteErrorResponse()
228 return err
229 }
230 return nil
231}
232
233func (h *h2muxConnection) newRequest(stream *h2mux.MuxedStream) (*http.Request, error) {
234 req, err := http.NewRequest("GET", "http://localhost:8080", h2mux.MuxedStreamReader{MuxedStream: stream})
235 if err != nil {
236 return nil, errors.Wrap(err, "Unexpected error from http.NewRequest")
237 }
238 err = H2RequestHeadersToH1Request(stream.Headers, req)
239 if err != nil {
240 return nil, errors.Wrap(err, "invalid request received")
241 }
242 return req, nil
243}
244
245type h2muxRespWriter struct {
246 *h2mux.MuxedStream
247}
248
249func (rp *h2muxRespWriter) WriteRespHeaders(status int, header http.Header) error {
250 headers := H1ResponseToH2ResponseHeaders(status, header)
251 headers = append(headers, h2mux.Header{Name: ResponseMetaHeader, Value: responseMetaHeaderOrigin})
252 return rp.WriteHeaders(headers)
253}
254
255func (rp *h2muxRespWriter) WriteErrorResponse() {
256 _ = rp.WriteHeaders([]h2mux.Header{
257 {Name: ":status", Value: "502"},
258 {Name: ResponseMetaHeader, Value: responseMetaHeaderCfd},
259 })
260 _, _ = rp.Write([]byte("502 Bad Gateway"))
261}
262