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.3.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/h2mux.go

260lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "io"
6 "net"
7 "net/http"
8 "time"
9
10 "github.com/cloudflare/cloudflared/h2mux"
11 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
12 "github.com/cloudflare/cloudflared/websocket"
13
14 "github.com/pkg/errors"
15 "github.com/rs/zerolog"
16 "golang.org/x/sync/errgroup"
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 updateMetricsTickC := time.Tick(h.muxerConfig.MetricsUpdateFreq)
171 var shutdownCompleted <-chan struct{}
172 for {
173 select {
174 case <-h.gracefulShutdownC:
175 if connectedFuse.IsConnected() {
176 h.unregister(isNamedTunnel)
177 }
178 h.stoppedGracefully = true
179 h.gracefulShutdownC = nil
180 shutdownCompleted = h.muxer.Shutdown()
181
182 case <-shutdownCompleted:
183 return
184
185 case <-ctx.Done():
186 // UnregisterTunnel blocks until the RPC call returns
187 if !h.stoppedGracefully && connectedFuse.IsConnected() {
188 h.unregister(isNamedTunnel)
189 }
190 h.muxer.Shutdown()
191 // don't wait for shutdown to finish when context is closed, this is the hard termination path
192 return
193
194 case <-updateMetricsTickC:
195 h.observer.metrics.updateMuxerMetrics(h.connIndexStr, h.muxer.Metrics())
196 }
197 }
198}
199
200func (h *h2muxConnection) newRPCStream(ctx context.Context, rpcName rpcName) (*h2mux.MuxedStream, error) {
201 openStreamCtx, openStreamCancel := context.WithTimeout(ctx, openStreamTimeout)
202 defer openStreamCancel()
203 stream, err := h.muxer.OpenRPCStream(openStreamCtx)
204 if err != nil {
205 return nil, err
206 }
207 return stream, nil
208}
209
210func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
211 respWriter := &h2muxRespWriter{stream}
212
213 req, reqErr := h.newRequest(stream)
214 if reqErr != nil {
215 respWriter.WriteErrorResponse()
216 return reqErr
217 }
218
219 var sourceConnectionType = TypeHTTP
220 if websocket.IsWebSocketUpgrade(req) {
221 sourceConnectionType = TypeWebsocket
222 }
223
224 err := h.config.OriginProxy.Proxy(respWriter, req, sourceConnectionType)
225 if err != nil {
226 respWriter.WriteErrorResponse()
227 return err
228 }
229 return nil
230}
231
232func (h *h2muxConnection) newRequest(stream *h2mux.MuxedStream) (*http.Request, error) {
233 req, err := http.NewRequest("GET", "http://localhost:8080", h2mux.MuxedStreamReader{MuxedStream: stream})
234 if err != nil {
235 return nil, errors.Wrap(err, "Unexpected error from http.NewRequest")
236 }
237 err = h2mux.H2RequestHeadersToH1Request(stream.Headers, req)
238 if err != nil {
239 return nil, errors.Wrap(err, "invalid request received")
240 }
241 return req, nil
242}
243
244type h2muxRespWriter struct {
245 *h2mux.MuxedStream
246}
247
248func (rp *h2muxRespWriter) WriteRespHeaders(status int, header http.Header) error {
249 headers := h2mux.H1ResponseToH2ResponseHeaders(status, header)
250 headers = append(headers, h2mux.Header{Name: ResponseMetaHeaderField, Value: responseMetaHeaderOrigin})
251 return rp.WriteHeaders(headers)
252}
253
254func (rp *h2muxRespWriter) WriteErrorResponse() {
255 _ = rp.WriteHeaders([]h2mux.Header{
256 {Name: ":status", Value: "502"},
257 {Name: ResponseMetaHeaderField, Value: responseMetaHeaderCfd},
258 })
259 _, _ = rp.Write([]byte("502 Bad Gateway"))
260}
261