GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2024.11.0

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

584lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "net"
10 "net/http"
11 "net/http/httptest"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/gobwas/ws/wsutil"
17 "github.com/google/uuid"
18 "github.com/rs/zerolog"
19 "github.com/stretchr/testify/assert"
20 "github.com/stretchr/testify/require"
21 "golang.org/x/net/http2"
22
23 "github.com/cloudflare/cloudflared/tunnelrpc"
24 "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
25)
26
27var (
28 testTransport = http2.Transport{}
29)
30
31func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
32 edgeConn, cfdConn := net.Pipe()
33 var connIndex = uint8(0)
34 log := zerolog.Nop()
35 obs := NewObserver(&log, &log)
36 controlStream := NewControlStream(
37 obs,
38 mockConnectedFuse{},
39 &TunnelProperties{},
40 connIndex,
41 nil,
42 nil,
43 1*time.Second,
44 nil,
45 1*time.Second,
46 HTTP2,
47 )
48 return NewHTTP2Connection(
49 cfdConn,
50 // OriginProxy is set in testConfigManager
51 testOrchestrator,
52 &pogs.ConnectionOptions{},
53 obs,
54 connIndex,
55 controlStream,
56 &log,
57 ), edgeConn
58}
59
60func TestHTTP2ConfigurationSet(t *testing.T) {
61 http2Conn, edgeConn := newTestHTTP2Connection()
62
63 ctx, cancel := context.WithCancel(context.Background())
64 var wg sync.WaitGroup
65 wg.Add(1)
66 go func() {
67 defer wg.Done()
68 http2Conn.Serve(ctx)
69 }()
70
71 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
72 require.NoError(t, err)
73
74 endpoint := fmt.Sprintf("http://localhost:8080/ok")
75 reqBody := []byte(`{
76"version": 2,
77"config": {"warp-routing": {"enabled": true}, "originRequest" : {"connectTimeout": 10}, "ingress" : [ {"hostname": "test", "service": "https://localhost:8000" } , {"service": "http_status:404"} ]}}
78`)
79 reader := bytes.NewReader(reqBody)
80 req, err := http.NewRequestWithContext(ctx, http.MethodPut, endpoint, reader)
81 require.NoError(t, err)
82 req.Header.Set(InternalUpgradeHeader, ConfigurationUpdate)
83
84 resp, err := edgeHTTP2Conn.RoundTrip(req)
85 require.NoError(t, err)
86 require.Equal(t, http.StatusOK, resp.StatusCode)
87 bdy, err := io.ReadAll(resp.Body)
88 require.NoError(t, err)
89 assert.Equal(t, `{"lastAppliedVersion":2,"err":null}`, string(bdy))
90 cancel()
91 wg.Wait()
92
93}
94
95func TestServeHTTP(t *testing.T) {
96 tests := []testRequest{
97 {
98 name: "ok",
99 endpoint: "ok",
100 expectedStatus: http.StatusOK,
101 expectedBody: []byte(http.StatusText(http.StatusOK)),
102 },
103 {
104 name: "large_file",
105 endpoint: "large_file",
106 expectedStatus: http.StatusOK,
107 expectedBody: testLargeResp,
108 },
109 {
110 name: "Bad request",
111 endpoint: "400",
112 expectedStatus: http.StatusBadRequest,
113 expectedBody: []byte(http.StatusText(http.StatusBadRequest)),
114 },
115 {
116 name: "Internal server error",
117 endpoint: "500",
118 expectedStatus: http.StatusInternalServerError,
119 expectedBody: []byte(http.StatusText(http.StatusInternalServerError)),
120 },
121 {
122 name: "Proxy error",
123 endpoint: "error",
124 expectedStatus: http.StatusBadGateway,
125 expectedBody: nil,
126 isProxyError: true,
127 },
128 }
129
130 http2Conn, edgeConn := newTestHTTP2Connection()
131
132 ctx, cancel := context.WithCancel(context.Background())
133 var wg sync.WaitGroup
134 wg.Add(1)
135 go func() {
136 defer wg.Done()
137 http2Conn.Serve(ctx)
138 }()
139
140 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
141 require.NoError(t, err)
142
143 for _, test := range tests {
144 endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
145 req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
146 require.NoError(t, err)
147
148 resp, err := edgeHTTP2Conn.RoundTrip(req)
149 require.NoError(t, err)
150 require.Equal(t, test.expectedStatus, resp.StatusCode)
151 if test.expectedBody != nil {
152 respBody, err := io.ReadAll(resp.Body)
153 require.NoError(t, err)
154 require.Equal(t, test.expectedBody, respBody)
155 }
156 if test.isProxyError {
157 require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(ResponseMetaHeader))
158 } else {
159 require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
160 }
161 }
162 cancel()
163 wg.Wait()
164}
165
166type mockNamedTunnelRPCClient struct {
167 shouldFail error
168 registered chan struct{}
169 unregistered chan struct{}
170}
171
172func (mc mockNamedTunnelRPCClient) SendLocalConfiguration(c context.Context, config []byte) error {
173 return nil
174}
175
176func (mc mockNamedTunnelRPCClient) RegisterConnection(
177 ctx context.Context,
178 auth pogs.TunnelAuth,
179 tunnelID uuid.UUID,
180 options *pogs.ConnectionOptions,
181 connIndex uint8,
182 edgeAddress net.IP,
183) (*pogs.ConnectionDetails, error) {
184 if mc.shouldFail != nil {
185 return nil, mc.shouldFail
186 }
187 close(mc.registered)
188 return &pogs.ConnectionDetails{
189 Location: "LIS",
190 UUID: uuid.New(),
191 TunnelIsRemotelyManaged: false,
192 }, nil
193}
194
195func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) error {
196 close(mc.unregistered)
197 return nil
198}
199
200func (mockNamedTunnelRPCClient) Close() {}
201
202type mockRPCClientFactory struct {
203 shouldFail error
204 registered chan struct{}
205 unregistered chan struct{}
206}
207
208func (mf *mockRPCClientFactory) newMockRPCClient(context.Context, io.ReadWriteCloser, time.Duration) tunnelrpc.RegistrationClient {
209 return &mockNamedTunnelRPCClient{
210 shouldFail: mf.shouldFail,
211 registered: mf.registered,
212 unregistered: mf.unregistered,
213 }
214}
215
216type wsRespWriter struct {
217 *httptest.ResponseRecorder
218 readPipe *io.PipeReader
219 writePipe *io.PipeWriter
220 closed bool
221 panicked bool
222}
223
224func newWSRespWriter() *wsRespWriter {
225 readPipe, writePipe := io.Pipe()
226 return &wsRespWriter{
227 httptest.NewRecorder(),
228 readPipe,
229 writePipe,
230 false,
231 false,
232 }
233}
234
235type nowriter struct {
236 io.Reader
237}
238
239func (nowriter) Write(_ []byte) (int, error) {
240 return 0, fmt.Errorf("writer not implemented")
241}
242
243func (w *wsRespWriter) RespBody() io.ReadWriter {
244 return nowriter{w.readPipe}
245}
246
247func (w *wsRespWriter) Write(data []byte) (n int, err error) {
248 if w.closed {
249 w.panicked = true
250 return 0, errors.New("wsRespWriter panicked")
251 }
252 return w.writePipe.Write(data)
253}
254
255func (w *wsRespWriter) close() {
256 w.closed = true
257}
258
259func TestServeWS(t *testing.T) {
260 http2Conn, _ := newTestHTTP2Connection()
261
262 ctx, cancel := context.WithCancel(context.Background())
263
264 respWriter := newWSRespWriter()
265 readPipe, writePipe := io.Pipe()
266
267 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws/echo", readPipe)
268 require.NoError(t, err)
269 req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
270
271 serveDone := make(chan struct{})
272 go func() {
273 defer close(serveDone)
274 http2Conn.ServeHTTP(respWriter, req)
275 respWriter.close()
276 }()
277
278 data := []byte("test websocket")
279 err = wsutil.WriteClientBinary(writePipe, data)
280 require.NoError(t, err)
281
282 respBody, err := wsutil.ReadServerBinary(respWriter.RespBody())
283 require.NoError(t, err)
284 require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
285
286 cancel()
287 resp := respWriter.Result()
288 // http2RespWriter should rewrite status 101 to 200
289 require.Equal(t, http.StatusOK, resp.StatusCode)
290 require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
291
292 <-serveDone
293 require.False(t, respWriter.panicked)
294}
295
296// TestNoWriteAfterServeHTTPReturns is a regression test of https://jira.cfops.it/browse/TUN-5184
297// to make sure we don't write to the ResponseWriter after the ServeHTTP method returns
298func TestNoWriteAfterServeHTTPReturns(t *testing.T) {
299 cfdHTTP2Conn, edgeTCPConn := newTestHTTP2Connection()
300
301 ctx, cancel := context.WithCancel(context.Background())
302 var wg sync.WaitGroup
303
304 serverDone := make(chan struct{})
305 go func() {
306 defer close(serverDone)
307 cfdHTTP2Conn.Serve(ctx)
308 }()
309
310 edgeTransport := http2.Transport{}
311 edgeHTTP2Conn, err := edgeTransport.NewClientConn(edgeTCPConn)
312 require.NoError(t, err)
313 message := []byte(t.Name())
314
315 for i := 0; i < 100; i++ {
316 wg.Add(1)
317 go func() {
318 defer wg.Done()
319 readPipe, writePipe := io.Pipe()
320 reqCtx, reqCancel := context.WithCancel(ctx)
321 req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, "http://localhost:8080/ws/flaky", readPipe)
322 require.NoError(t, err)
323 req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
324
325 resp, err := edgeHTTP2Conn.RoundTrip(req)
326 require.NoError(t, err)
327 // http2RespWriter should rewrite status 101 to 200
328 require.Equal(t, http.StatusOK, resp.StatusCode)
329
330 wg.Add(1)
331 go func() {
332 defer wg.Done()
333 for {
334 select {
335 case <-reqCtx.Done():
336 return
337 default:
338 }
339 _ = wsutil.WriteClientBinary(writePipe, message)
340 }
341 }()
342
343 time.Sleep(time.Millisecond * 100)
344 reqCancel()
345 }()
346 }
347
348 wg.Wait()
349 cancel()
350 <-serverDone
351}
352
353func TestServeControlStream(t *testing.T) {
354 http2Conn, edgeConn := newTestHTTP2Connection()
355
356 rpcClientFactory := mockRPCClientFactory{
357 registered: make(chan struct{}),
358 unregistered: make(chan struct{}),
359 }
360
361 obs := NewObserver(&log, &log)
362 controlStream := NewControlStream(
363 obs,
364 mockConnectedFuse{},
365 &TunnelProperties{},
366 1,
367 nil,
368 rpcClientFactory.newMockRPCClient,
369 1*time.Second,
370 nil,
371 1*time.Second,
372 HTTP2,
373 )
374 http2Conn.controlStreamHandler = controlStream
375
376 ctx, cancel := context.WithCancel(context.Background())
377 var wg sync.WaitGroup
378 wg.Add(1)
379 go func() {
380 defer wg.Done()
381 http2Conn.Serve(ctx)
382 }()
383
384 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
385 require.NoError(t, err)
386 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
387
388 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
389 require.NoError(t, err)
390
391 wg.Add(1)
392 go func() {
393 defer wg.Done()
394 edgeHTTP2Conn.RoundTrip(req)
395 }()
396
397 <-rpcClientFactory.registered
398 cancel()
399 <-rpcClientFactory.unregistered
400 assert.False(t, http2Conn.stoppedGracefully)
401
402 wg.Wait()
403}
404
405func TestFailRegistration(t *testing.T) {
406 http2Conn, edgeConn := newTestHTTP2Connection()
407
408 rpcClientFactory := mockRPCClientFactory{
409 shouldFail: errDuplicationConnection,
410 registered: make(chan struct{}),
411 unregistered: make(chan struct{}),
412 }
413
414 obs := NewObserver(&log, &log)
415 controlStream := NewControlStream(
416 obs,
417 mockConnectedFuse{},
418 &TunnelProperties{},
419 http2Conn.connIndex,
420 nil,
421 rpcClientFactory.newMockRPCClient,
422 1*time.Second,
423 nil,
424 1*time.Second,
425 HTTP2,
426 )
427 http2Conn.controlStreamHandler = controlStream
428
429 ctx, cancel := context.WithCancel(context.Background())
430 var wg sync.WaitGroup
431 wg.Add(1)
432 go func() {
433 defer wg.Done()
434 http2Conn.Serve(ctx)
435 }()
436
437 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
438 require.NoError(t, err)
439 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
440
441 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
442 require.NoError(t, err)
443 resp, err := edgeHTTP2Conn.RoundTrip(req)
444 require.NoError(t, err)
445 require.Equal(t, http.StatusBadGateway, resp.StatusCode)
446
447 assert.NotNil(t, http2Conn.controlStreamErr)
448 cancel()
449 wg.Wait()
450}
451
452func TestGracefulShutdownHTTP2(t *testing.T) {
453 http2Conn, edgeConn := newTestHTTP2Connection()
454
455 rpcClientFactory := mockRPCClientFactory{
456 registered: make(chan struct{}),
457 unregistered: make(chan struct{}),
458 }
459 events := &eventCollectorSink{}
460
461 shutdownC := make(chan struct{})
462 obs := NewObserver(&log, &log)
463 obs.RegisterSink(events)
464 controlStream := NewControlStream(
465 obs,
466 mockConnectedFuse{},
467 &TunnelProperties{},
468 http2Conn.connIndex,
469 nil,
470 rpcClientFactory.newMockRPCClient,
471 1*time.Second,
472 shutdownC,
473 1*time.Second,
474 HTTP2,
475 )
476
477 http2Conn.controlStreamHandler = controlStream
478
479 ctx, cancel := context.WithCancel(context.Background())
480 var wg sync.WaitGroup
481 wg.Add(1)
482 go func() {
483 defer wg.Done()
484 http2Conn.Serve(ctx)
485 }()
486
487 req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
488 require.NoError(t, err)
489 req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
490
491 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
492 require.NoError(t, err)
493
494 wg.Add(1)
495 go func() {
496 defer wg.Done()
497 _, _ = edgeHTTP2Conn.RoundTrip(req)
498 }()
499
500 select {
501 case <-rpcClientFactory.registered:
502 break // ok
503 case <-time.Tick(time.Second):
504 t.Fatal("timeout out waiting for registration")
505 }
506
507 // signal graceful shutdown
508 close(shutdownC)
509
510 select {
511 case <-rpcClientFactory.unregistered:
512 break // ok
513 case <-time.Tick(time.Second):
514 t.Fatal("timeout out waiting for unregistered signal")
515 }
516 assert.True(t, controlStream.IsStopped())
517
518 cancel()
519 wg.Wait()
520
521 events.assertSawEvent(t, Event{
522 Index: http2Conn.connIndex,
523 EventType: Unregistering,
524 })
525}
526
527func benchmarkServeHTTP(b *testing.B, test testRequest) {
528 http2Conn, edgeConn := newTestHTTP2Connection()
529
530 ctx, cancel := context.WithCancel(context.Background())
531 var wg sync.WaitGroup
532 wg.Add(1)
533 go func() {
534 defer wg.Done()
535 http2Conn.Serve(ctx)
536 }()
537
538 endpoint := fmt.Sprintf("http://localhost:8080/%s", test.endpoint)
539 req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
540 require.NoError(b, err)
541
542 edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
543 require.NoError(b, err)
544
545 b.ResetTimer()
546 for i := 0; i < b.N; i++ {
547 b.StartTimer()
548 resp, err := edgeHTTP2Conn.RoundTrip(req)
549 b.StopTimer()
550 require.NoError(b, err)
551 require.Equal(b, test.expectedStatus, resp.StatusCode)
552 if test.expectedBody != nil {
553 respBody, err := io.ReadAll(resp.Body)
554 require.NoError(b, err)
555 require.Equal(b, test.expectedBody, respBody)
556 }
557 resp.Body.Close()
558 }
559
560 cancel()
561 wg.Wait()
562}
563
564func BenchmarkServeHTTPSimple(b *testing.B) {
565 test := testRequest{
566 name: "ok",
567 endpoint: "ok",
568 expectedStatus: http.StatusOK,
569 expectedBody: []byte(http.StatusText(http.StatusOK)),
570 }
571
572 benchmarkServeHTTP(b, test)
573}
574
575func BenchmarkServeHTTPLargeFile(b *testing.B) {
576 test := testRequest{
577 name: "large_file",
578 endpoint: "large_file",
579 expectedStatus: http.StatusOK,
580 expectedBody: testLargeResp,
581 }
582
583 benchmarkServeHTTP(b, test)
584}
585