GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
bf38e5aa1268cd9f5e2e4153774efcefe902507c

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

408lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "io/ioutil"
8 "net"
9 "net/http"
10 "net/http/httptest"
11 "sync"
12 "testing"
13 "time"
14
15 "github.com/gobwas/ws/wsutil"
16 "github.com/rs/zerolog"
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/require"
19 "golang.org/x/net/http2"
20
21 "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
22 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
23)
24
25var (
26 testTransport = http2.Transport{}
27)
28
29func newTestHTTP2Connection() (*http2Connection, net.Conn) {
30 edgeConn, originConn := net.Pipe()
31 var connIndex = uint8(0)
32 return NewHTTP2Connection(
33 originConn,
34 testConfig,
35 &NamedTunnelConfig{},
36 &pogs.ConnectionOptions{},
37 NewObserver(&log, &log, false),
38 connIndex,
39 mockConnectedFuse{},
40 nil,
41 ), edgeConn
42}
43
44func TestServeHTTP(t *testing.T) {
45 tests := []testRequest{
46 {
47 name: "ok",
48 endpoint: "ok",
49 expectedStatus: http.StatusOK,
50 expectedBody: []byte(http.StatusText(http.StatusOK)),
51 },
52 {
53 name: "large_file",
54 endpoint: "large_file",
55 expectedStatus: http.StatusOK,
56 expectedBody: testLargeResp,
57 },
58 {
59 name: "Bad request",
60 endpoint: "400",
61 expectedStatus: http.StatusBadRequest,
62 expectedBody: []byte(http.StatusText(http.StatusBadRequest)),
63 },
64 {
65 name: "Internal server error",
66 endpoint: "500",
67 expectedStatus: http.StatusInternalServerError,
68 expectedBody: []byte(http.StatusText(http.StatusInternalServerError)),
69 },
70 {
71 name: "Proxy error",
72 endpoint: "error",
73 expectedStatus: http.StatusBadGateway,
74 expectedBody: nil,
75 isProxyError: true,
76 },
77 }
78
79 http2Conn, edgeConn := newTestHTTP2Connection()
80
81 ctx, cancel := context.WithCancel(context.Background())
82 var wg sync.WaitGroup
83 wg.Add(1)
84 go func() {
85 defer wg.Done()
86 http2Conn.Serve(ctx)
87 }()
88
89 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
90 require.NoError(t, err)
91
92 for _, test := range tests {
93 endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
94 req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
95 require.NoError(t, err)
96
97 resp, err := edgeHTTP2Conn.RoundTrip(req)
98 require.NoError(t, err)
99 require.Equal(t, test.expectedStatus, resp.StatusCode)
100 if test.expectedBody != nil {
101 respBody, err := ioutil.ReadAll(resp.Body)
102 require.NoError(t, err)
103 require.Equal(t, test.expectedBody, respBody)
104 }
105 if test.isProxyError {
106 require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(ResponseMetaHeader))
107 } else {
108 require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
109 }
110 }
111 cancel()
112 wg.Wait()
113}
114
115type mockNamedTunnelRPCClient struct {
116 shouldFail error
117 registered chan struct{}
118 unregistered chan struct{}
119}
120
121func (mc mockNamedTunnelRPCClient) RegisterConnection(
122 c context.Context,
123 config *NamedTunnelConfig,
124 options *tunnelpogs.ConnectionOptions,
125 connIndex uint8,
126 observer *Observer,
127) error {
128 if mc.shouldFail != nil {
129 return mc.shouldFail
130 }
131 close(mc.registered)
132 return nil
133}
134
135func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
136 close(mc.unregistered)
137}
138
139func (mockNamedTunnelRPCClient) Close() {}
140
141type mockRPCClientFactory struct {
142 shouldFail error
143 registered chan struct{}
144 unregistered chan struct{}
145}
146
147func (mf *mockRPCClientFactory) newMockRPCClient(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient {
148 return mockNamedTunnelRPCClient{
149 shouldFail: mf.shouldFail,
150 registered: mf.registered,
151 unregistered: mf.unregistered,
152 }
153}
154
155type wsRespWriter struct {
156 *httptest.ResponseRecorder
157 readPipe *io.PipeReader
158 writePipe *io.PipeWriter
159}
160
161func newWSRespWriter() *wsRespWriter {
162 readPipe, writePipe := io.Pipe()
163 return &wsRespWriter{
164 httptest.NewRecorder(),
165 readPipe,
166 writePipe,
167 }
168}
169
170func (w *wsRespWriter) RespBody() io.ReadWriter {
171 return nowriter{w.readPipe}
172}
173
174func (w *wsRespWriter) Write(data []byte) (n int, err error) {
175 return w.writePipe.Write(data)
176}
177
178func TestServeWS(t *testing.T) {
179 http2Conn, _ := newTestHTTP2Connection()
180
181 ctx, cancel := context.WithCancel(context.Background())
182 var wg sync.WaitGroup
183 wg.Add(1)
184 go func() {
185 defer wg.Done()
186 http2Conn.Serve(ctx)
187 }()
188
189 respWriter := newWSRespWriter()
190 readPipe, writePipe := io.Pipe()
191
192 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws", readPipe)
193 require.NoError(t, err)
194 req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
195
196 wg.Add(1)
197 go func() {
198 defer wg.Done()
199 http2Conn.ServeHTTP(respWriter, req)
200 }()
201
202 data := []byte("test websocket")
203 err = wsutil.WriteClientText(writePipe, data)
204 require.NoError(t, err)
205
206 respBody, err := wsutil.ReadServerText(respWriter.RespBody())
207 require.NoError(t, err)
208 require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
209
210 cancel()
211 resp := respWriter.Result()
212 // http2RespWriter should rewrite status 101 to 200
213 require.Equal(t, http.StatusOK, resp.StatusCode)
214 require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
215
216 wg.Wait()
217}
218
219func TestServeControlStream(t *testing.T) {
220 http2Conn, edgeConn := newTestHTTP2Connection()
221
222 rpcClientFactory := mockRPCClientFactory{
223 registered: make(chan struct{}),
224 unregistered: make(chan struct{}),
225 }
226 http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
227
228 ctx, cancel := context.WithCancel(context.Background())
229 var wg sync.WaitGroup
230 wg.Add(1)
231 go func() {
232 defer wg.Done()
233 http2Conn.Serve(ctx)
234 }()
235
236 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
237 require.NoError(t, err)
238 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
239
240 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
241 require.NoError(t, err)
242
243 wg.Add(1)
244 go func() {
245 defer wg.Done()
246 edgeHTTP2Conn.RoundTrip(req)
247 }()
248
249 <-rpcClientFactory.registered
250 cancel()
251 <-rpcClientFactory.unregistered
252 assert.False(t, http2Conn.stoppedGracefully)
253
254 wg.Wait()
255}
256
257func TestFailRegistration(t *testing.T) {
258 http2Conn, edgeConn := newTestHTTP2Connection()
259
260 rpcClientFactory := mockRPCClientFactory{
261 shouldFail: errDuplicationConnection,
262 registered: make(chan struct{}),
263 unregistered: make(chan struct{}),
264 }
265 http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
266
267 ctx, cancel := context.WithCancel(context.Background())
268 var wg sync.WaitGroup
269 wg.Add(1)
270 go func() {
271 defer wg.Done()
272 http2Conn.Serve(ctx)
273 }()
274
275 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
276 require.NoError(t, err)
277 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
278
279 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
280 require.NoError(t, err)
281 resp, err := edgeHTTP2Conn.RoundTrip(req)
282 require.NoError(t, err)
283 require.Equal(t, http.StatusBadGateway, resp.StatusCode)
284
285 assert.NotNil(t, http2Conn.controlStreamErr)
286 cancel()
287 wg.Wait()
288}
289
290func TestGracefulShutdownHTTP2(t *testing.T) {
291 http2Conn, edgeConn := newTestHTTP2Connection()
292
293 rpcClientFactory := mockRPCClientFactory{
294 registered: make(chan struct{}),
295 unregistered: make(chan struct{}),
296 }
297 events := &eventCollectorSink{}
298 http2Conn.newRPCClientFunc = rpcClientFactory.newMockRPCClient
299 http2Conn.observer.RegisterSink(events)
300 shutdownC := make(chan struct{})
301 http2Conn.gracefulShutdownC = shutdownC
302
303 ctx, cancel := context.WithCancel(context.Background())
304 var wg sync.WaitGroup
305 wg.Add(1)
306 go func() {
307 defer wg.Done()
308 http2Conn.Serve(ctx)
309 }()
310
311 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
312 require.NoError(t, err)
313 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
314
315 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
316 require.NoError(t, err)
317
318 wg.Add(1)
319 go func() {
320 defer wg.Done()
321 _, _ = edgeHTTP2Conn.RoundTrip(req)
322 }()
323
324 select {
325 case <-rpcClientFactory.registered:
326 break //ok
327 case <-time.Tick(time.Second):
328 t.Fatal("timeout out waiting for registration")
329 }
330
331 // signal graceful shutdown
332 close(shutdownC)
333
334 select {
335 case <-rpcClientFactory.unregistered:
336 break //ok
337 case <-time.Tick(time.Second):
338 t.Fatal("timeout out waiting for unregistered signal")
339 }
340 assert.True(t, http2Conn.stoppedGracefully)
341
342 cancel()
343 wg.Wait()
344
345 events.assertSawEvent(t, Event{
346 Index: http2Conn.connIndex,
347 EventType: Unregistering,
348 })
349}
350
351func benchmarkServeHTTP(b *testing.B, test testRequest) {
352 http2Conn, edgeConn := newTestHTTP2Connection()
353
354 ctx, cancel := context.WithCancel(context.Background())
355 var wg sync.WaitGroup
356 wg.Add(1)
357 go func() {
358 defer wg.Done()
359 http2Conn.Serve(ctx)
360 }()
361
362 endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
363 req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
364 require.NoError(b, err)
365
366 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
367 require.NoError(b, err)
368
369 b.ResetTimer()
370 for i := 0; i < b.N; i++ {
371 b.StartTimer()
372 resp, err := edgeHTTP2Conn.RoundTrip(req)
373 b.StopTimer()
374 require.NoError(b, err)
375 require.Equal(b, test.expectedStatus, resp.StatusCode)
376 if test.expectedBody != nil {
377 respBody, err := ioutil.ReadAll(resp.Body)
378 require.NoError(b, err)
379 require.Equal(b, test.expectedBody, respBody)
380 }
381 resp.Body.Close()
382 }
383
384 cancel()
385 wg.Wait()
386}
387
388func BenchmarkServeHTTPSimple(b *testing.B) {
389 test := testRequest{
390 name: "ok",
391 endpoint: "ok",
392 expectedStatus: http.StatusOK,
393 expectedBody: []byte(http.StatusText(http.StatusOK)),
394 }
395
396 benchmarkServeHTTP(b, test)
397}
398
399func BenchmarkServeHTTPLargeFile(b *testing.B) {
400 test := testRequest{
401 name: "large_file",
402 endpoint: "large_file",
403 expectedStatus: http.StatusOK,
404 expectedBody: testLargeResp,
405 }
406
407 benchmarkServeHTTP(b, test)
408}
409