GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

cloudflare/cloudflared

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
2023.4.2

Branches

Tags

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

Clone

HTTPS

Download ZIP

cloudflared/connection

connection/connection_test.go

202lines · modecode

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