GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
cd4af5696d0db81caae903b33e4bbddfb5aff9ff

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

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