GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2025.2.1

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/http2_test.go

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