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/connection_test.go

211lines · modecode

RawDownload
Latest commit unavailable.
1package connection
2
3import (
4 "context"
5 "crypto/rand"
6 "fmt"
7 "io"
8 "math/big"
9 "net/http"
10 "time"
11
12 pkgerrors "github.com/pkg/errors"
13 "github.com/rs/zerolog"
14
15 cfdflow "github.com/cloudflare/cloudflared/flow"
16
17 "github.com/cloudflare/cloudflared/stream"
18 "github.com/cloudflare/cloudflared/tracing"
19 tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
20 "github.com/cloudflare/cloudflared/websocket"
21)
22
23const (
24 largeFileSize = 2 * 1024 * 1024
25 testGracePeriod = time.Millisecond * 100
26)
27
28var (
29 testOrchestrator = &mockOrchestrator{
30 originProxy: &mockOriginProxy{},
31 }
32 log = zerolog.Nop()
33 testLargeResp = make([]byte, largeFileSize)
34)
35
36var _ ReadWriteAcker = (*HTTPResponseReadWriteAcker)(nil)
37
38type testRequest struct {
39 name string
40 endpoint string
41 expectedStatus int
42 expectedBody []byte
43 isProxyError bool
44}
45
46type mockOrchestrator struct {
47 originProxy OriginProxy
48}
49
50func (mcr *mockOrchestrator) GetConfigJSON() ([]byte, error) {
51 return nil, fmt.Errorf("not implemented")
52}
53
54func (*mockOrchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
55 return &tunnelpogs.UpdateConfigurationResponse{
56 LastAppliedVersion: version,
57 }
58}
59
60func (mcr *mockOrchestrator) GetOriginProxy() (OriginProxy, error) {
61 return mcr.originProxy, nil
62}
63
64func (mcr *mockOrchestrator) WarpRoutingEnabled() (enabled bool) {
65 return true
66}
67
68type mockOriginProxy struct{}
69
70func (moc *mockOriginProxy) ProxyHTTP(
71 w ResponseWriter,
72 tr *tracing.TracedHTTPRequest,
73 isWebsocket bool,
74) error {
75 req := tr.Request
76 if isWebsocket {
77 switch req.URL.Path {
78 case "/ws/echo":
79 return wsEchoEndpoint(w, req)
80 case "/ws/flaky":
81 return wsFlakyEndpoint(w, req)
82 default:
83 originRespEndpoint(w, http.StatusNotFound, []byte("ws endpoint not found"))
84 return fmt.Errorf("unknown websocket endpoint %s", req.URL.Path)
85 }
86 }
87 switch req.URL.Path {
88 case "/ok":
89 originRespEndpoint(w, http.StatusOK, []byte(http.StatusText(http.StatusOK)))
90 case "/large_file":
91 originRespEndpoint(w, http.StatusOK, testLargeResp)
92 case "/400":
93 originRespEndpoint(w, http.StatusBadRequest, []byte(http.StatusText(http.StatusBadRequest)))
94 case "/500":
95 originRespEndpoint(w, http.StatusInternalServerError, []byte(http.StatusText(http.StatusInternalServerError)))
96 case "/error":
97 return fmt.Errorf("Failed to proxy to origin")
98 default:
99 originRespEndpoint(w, http.StatusNotFound, []byte("page not found"))
100 }
101 return nil
102}
103
104func (moc *mockOriginProxy) ProxyTCP(
105 ctx context.Context,
106 rwa ReadWriteAcker,
107 r *TCPRequest,
108) error {
109 if r.CfTraceID == "flow-rate-limited" {
110 return pkgerrors.Wrap(cfdflow.ErrTooManyActiveFlows, "tcp flow rate limited")
111 }
112
113 return nil
114}
115
116type echoPipe struct {
117 reader *io.PipeReader
118 writer *io.PipeWriter
119}
120
121func (ep *echoPipe) Read(p []byte) (int, error) {
122 return ep.reader.Read(p)
123}
124
125func (ep *echoPipe) Write(p []byte) (int, error) {
126 return ep.writer.Write(p)
127}
128
129// A mock origin that echos data by streaming like a tcpOverWSConnection
130// https://github.com/cloudflare/cloudflared/blob/master/ingress/origin_connection.go
131func wsEchoEndpoint(w ResponseWriter, r *http.Request) error {
132 resp := &http.Response{
133 StatusCode: http.StatusSwitchingProtocols,
134 }
135 if err := w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {
136 return err
137 }
138 wsCtx, cancel := context.WithCancel(r.Context())
139 readPipe, writePipe := io.Pipe()
140
141 wsConn := websocket.NewConn(wsCtx, NewHTTPResponseReadWriterAcker(w, w.(http.Flusher), r), &log)
142 go func() {
143 select {
144 case <-wsCtx.Done():
145 case <-r.Context().Done():
146 }
147 readPipe.Close()
148 writePipe.Close()
149 }()
150
151 originConn := &echoPipe{reader: readPipe, writer: writePipe}
152 stream.Pipe(wsConn, originConn, &log)
153 cancel()
154 wsConn.Close()
155 return nil
156}
157
158type flakyConn struct {
159 closeAt time.Time
160}
161
162func (fc *flakyConn) Read(p []byte) (int, error) {
163 if time.Now().After(fc.closeAt) {
164 return 0, io.EOF
165 }
166 n := copy(p, "Read from flaky connection")
167 return n, nil
168}
169
170func (fc *flakyConn) Write(p []byte) (int, error) {
171 if time.Now().After(fc.closeAt) {
172 return 0, fmt.Errorf("flaky connection closed")
173 }
174 return len(p), nil
175}
176
177func wsFlakyEndpoint(w ResponseWriter, r *http.Request) error {
178 resp := &http.Response{
179 StatusCode: http.StatusSwitchingProtocols,
180 }
181 if err := w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {
182 return err
183 }
184 wsCtx, cancel := context.WithCancel(r.Context())
185
186 wsConn := websocket.NewConn(wsCtx, NewHTTPResponseReadWriterAcker(w, w.(http.Flusher), r), &log)
187
188 rInt, _ := rand.Int(rand.Reader, big.NewInt(50))
189 closedAfter := time.Millisecond * time.Duration(rInt.Int64())
190 originConn := &flakyConn{closeAt: time.Now().Add(closedAfter)}
191 stream.Pipe(wsConn, originConn, &log)
192 cancel()
193 wsConn.Close()
194 return nil
195}
196
197func originRespEndpoint(w ResponseWriter, status int, data []byte) {
198 resp := &http.Response{
199 StatusCode: status,
200 }
201 _ = w.WriteRespHeaders(resp.StatusCode, resp.Header)
202 _, _ = w.Write(data)
203}
204
205type mockConnectedFuse struct{}
206
207func (mcf mockConnectedFuse) Connected() {}
208
209func (mcf mockConnectedFuse) IsConnected() bool {
210 return true
211}
212