1
package reverseproxy
2
3
import (
4
+ "bufio"
5
"bytes"
6
+ "errors"
7
"io"
8
+ "net"
9
+ "net/http"
10
"net/http/httptest"
11
+ "net/url"
12
+ "runtime"
13
"strings"
14
"sync"
15
"testing"
16
+ "time"
17
18
"github.com/caddyserver/caddy/v2"
19
+ "github.com/caddyserver/caddy/v2/modules/caddyhttp"
20
)
21
22
func TestHandlerCopyResponse(t *testing.T) {
66
go copier.copyToBackend(errc)
67
wg.Wait()
68
61
- if err := <-errc; err != nil {
62
- t.Fatalf("copyToBackend() error = %v", err)
69
+ // the destination cannot be half-closed, so a clean copy reports errCopyDone
70
+ if err := <-errc; !errors.Is(err, errCopyDone) {
71
+ t.Fatalf("copyToBackend() error = %v, want %v", err, errCopyDone)
72
}
73
if got := dst.String(); got != "hello" {
74
t.Fatalf("copied data = %q, want %q", got, "hello")
75
}
76
}
77
78
+// A clean EOF in one direction must be propagated to the destination as a
79
+// write-half close instead of tearing the whole tunnel down.
80
+// See https://github.com/caddyserver/caddy/issues/8026.
81
+func TestSwitchProtocolCopierHalfClose(t *testing.T) {
82
+ closeWriteErr := errors.New("no half-close")
83
+
84
+ for _, tc := range []struct {
85
+ name string
86
+ dst io.ReadWriteCloser
87
+ wantErr error // nil means the half-close was propagated
88
+ }{
89
+ {
90
+ name: "destination supports CloseWrite",
91
+ dst: &closeWriteRecorder{},
92
+ wantErr: nil,
93
+ },
94
+ {
95
+ name: "destination does not support CloseWrite",
96
+ dst: nopReadWriteCloser{Writer: io.Discard},
97
+ wantErr: errCopyDone,
98
+ },
99
+ {
100
+ name: "CloseWrite fails",
101
+ dst: &closeWriteRecorder{err: closeWriteErr},
102
+ wantErr: closeWriteErr,
103
+ },
104
+ } {
105
+ t.Run(tc.name, func(t *testing.T) {
106
+ var wg sync.WaitGroup
107
+ errc := make(chan error, 1)
108
+ copier := switchProtocolCopier{
109
+ user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
110
+ backend: tc.dst,
111
+ wg: &wg,
112
+ }
113
+
114
+ wg.Add(1)
115
+ go copier.copyToBackend(errc)
116
+ wg.Wait()
117
+
118
+ if err := <-errc; !errors.Is(err, tc.wantErr) {
119
+ t.Fatalf("copyToBackend() error = %v, want %v", err, tc.wantErr)
120
+ }
121
+ if rec, ok := tc.dst.(*closeWriteRecorder); ok && !rec.called {
122
+ t.Fatal("CloseWrite() was not called on a destination that supports it")
123
+ }
124
+ })
125
+ }
126
+}
127
+
128
+// A copy error must be reported as-is and must not be mistaken for a clean
129
+// half-close, so the tunnel is still torn down immediately.
130
+func TestSwitchProtocolCopierErrorIsNotHalfClose(t *testing.T) {
131
+ var wg sync.WaitGroup
132
+ errc := make(chan error, 1)
133
+ wantErr := errors.New("write failed")
134
+ dst := &closeWriteRecorder{writeErr: wantErr}
135
+
136
+ copier := switchProtocolCopier{
137
+ user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
138
+ backend: dst,
139
+ wg: &wg,
140
+ }
141
+
142
+ wg.Add(1)
143
+ go copier.copyToBackend(errc)
144
+ wg.Wait()
145
+
146
+ if err := <-errc; !errors.Is(err, wantErr) {
147
+ t.Fatalf("error = %v, want %v", err, wantErr)
148
+ }
149
+ if dst.called {
150
+ t.Fatal("CloseWrite() was called after a copy error")
151
+ }
152
+}
153
+
154
func TestSwitchProtocolCopierDefaultBufferSize(t *testing.T) {
155
copier := switchProtocolCopier{}
156
buf := copier.buffer()
165
}
166
167
func (nopReadWriteCloser) Close() error { return nil }
168
+
169
+// closeWriteRecorder is a destination that supports closing only its write
170
+// half, and records whether that happened.
171
+type closeWriteRecorder struct {
172
+ called bool
173
+ err error
174
+ writeErr error
175
+}
176
+
177
+func (c *closeWriteRecorder) Read([]byte) (int, error) { return 0, io.EOF }
178
+
179
+func (c *closeWriteRecorder) Write(p []byte) (int, error) {
180
+ if c.writeErr != nil {
181
+ return 0, c.writeErr
182
+ }
183
+ return len(p), nil
184
+}
185
+
186
+func (c *closeWriteRecorder) Close() error { return nil }
187
+
188
+func (c *closeWriteRecorder) CloseWrite() error {
189
+ c.called = true
190
+ return c.err
191
+}
192
+
193
+// TestHandlerUpgradedStreamHalfClose drives a real upgraded stream through the
194
+// handler and checks that closing one direction is propagated to the other side
195
+// instead of tearing the whole tunnel down.
196
+//
197
+// It mirrors net/http/httputil's TestReverseProxyWebSocketHalfTCP, which covers
198
+// the same class upstream (https://go.dev/issue/35892). Before the fix, the
199
+// "close write" cases failed: the handler returned as soon as the first
200
+// direction reported, and the deferred closes killed the other direction before
201
+// its pending bytes could be delivered.
202
+//
203
+// See https://github.com/caddyserver/caddy/issues/8026.
204
+func TestHandlerUpgradedStreamHalfClose(t *testing.T) {
205
+ switch runtime.GOOS {
206
+ case "plan9", "js", "wasip1":
207
+ t.Skipf("not supported on %s", runtime.GOOS)
208
+ }
209
+
210
+ // Reads carry a deadline so a regression fails the test promptly instead of
211
+ // blocking until the whole package times out.
212
+ const readTimeout = 10 * time.Second
213
+
214
+ mustRead := func(t *testing.T, conn *net.TCPConn, msg string) {
215
+ t.Helper()
216
+ if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
217
+ t.Fatalf("failed to set read deadline: %v", err)
218
+ }
219
+ b := make([]byte, len(msg))
220
+ if _, err := io.ReadFull(conn, b); err != nil {
221
+ t.Fatalf("failed to read: %v", err)
222
+ }
223
+ if got, want := string(b), msg; got != want {
224
+ t.Fatalf("got %#q, want %#q", got, want)
225
+ }
226
+ }
227
+
228
+ mustReadEOF := func(t *testing.T, conn *net.TCPConn) {
229
+ t.Helper()
230
+ if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
231
+ t.Fatalf("failed to set read deadline: %v", err)
232
+ }
233
+ b := make([]byte, 1)
234
+ if _, err := conn.Read(b); !errors.Is(err, io.EOF) {
235
+ t.Fatalf("read after peer half-close: got %v, want EOF", err)
236
+ }
237
+ }
238
+
239
+ mustWrite := func(t *testing.T, conn *net.TCPConn, msg string) {
240
+ t.Helper()
241
+ if _, err := conn.Write([]byte(msg)); err != nil {
242
+ t.Fatalf("failed to write: %v", err)
243
+ }
244
+ }
245
+
246
+ mustCloseRead := func(t *testing.T, conn *net.TCPConn) {
247
+ t.Helper()
248
+ if err := conn.CloseRead(); err != nil {
249
+ t.Fatalf("failed to CloseRead: %v", err)
250
+ }
251
+ }
252
+
253
+ mustCloseWrite := func(t *testing.T, conn *net.TCPConn) {
254
+ t.Helper()
255
+ if err := conn.CloseWrite(); err != nil {
256
+ t.Fatalf("failed to CloseWrite: %v", err)
257
+ }
258
+ }
259
+
260
+ tests := map[string]func(t *testing.T, cli, srv *net.TCPConn){
261
+ "backend close read": func(t *testing.T, cli, srv *net.TCPConn) {
262
+ mustCloseRead(t, srv)
263
+ mustWrite(t, srv, "backend sends")
264
+ mustRead(t, cli, "backend sends")
265
+ },
266
+ "backend close write": func(t *testing.T, cli, srv *net.TCPConn) {
267
+ mustCloseWrite(t, srv)
268
+ mustWrite(t, cli, "client sends")
269
+ mustRead(t, srv, "client sends")
270
+ mustReadEOF(t, cli)
271
+ },
272
+ "client close read": func(t *testing.T, cli, srv *net.TCPConn) {
273
+ mustCloseRead(t, cli)
274
+ mustWrite(t, cli, "client sends")
275
+ mustRead(t, srv, "client sends")
276
+ },
277
+ "client close write": func(t *testing.T, cli, srv *net.TCPConn) {
278
+ mustCloseWrite(t, cli)
279
+ mustWrite(t, srv, "backend sends")
280
+ mustRead(t, cli, "backend sends")
281
+ mustReadEOF(t, srv)
282
+ },
283
+ }
284
+
285
+ for name, test := range tests {
286
+ t.Run(name, func(t *testing.T) {
287
+ srvc := make(chan *net.TCPConn, 1)
288
+
289
+ backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
290
+ conn, _, err := http.NewResponseController(w).Hijack()
291
+ if err != nil {
292
+ t.Errorf("backend hijack failed: %v", err)
293
+ return
294
+ }
295
+ tcp, ok := conn.(*net.TCPConn)
296
+ if !ok {
297
+ conn.Close()
298
+ t.Errorf("backend conn is %T, want *net.TCPConn", conn)
299
+ return
300
+ }
301
+ if _, err := io.WriteString(tcp, "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {
302
+ tcp.Close()
303
+ t.Errorf("backend upgrade write failed: %v", err)
304
+ return
305
+ }
306
+ srvc <- tcp
307
+ }))
308
+ defer backend.Close()
309
+
310
+ h := minimalHandler(0, &Upstream{
311
+ Host: new(Host),
312
+ Dial: backend.Listener.Addr().String(),
313
+ })
314
+ h.connections = make(map[io.ReadWriteCloser]openConnection)
315
+ h.connectionsMu = new(sync.Mutex)
316
+
317
+ frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
318
+ _ = h.ServeHTTP(w, prepareTestRequest(r), caddyhttp.HandlerFunc(func(http.ResponseWriter, *http.Request) error {
319
+ return nil
320
+ }))
321
+ }))
322
+ defer frontend.Close()
323
+
324
+ frontendURL, err := url.Parse(frontend.URL)
325
+ if err != nil {
326
+ t.Fatalf("failed to parse frontend URL: %v", err)
327
+ }
328
+ addr, err := net.ResolveTCPAddr("tcp", frontendURL.Host)
329
+ if err != nil {
330
+ t.Fatalf("failed to resolve TCP address: %v", err)
331
+ }
332
+ cli, err := net.DialTCP("tcp", nil, addr)
333
+ if err != nil {
334
+ t.Fatalf("failed to dial frontend: %v", err)
335
+ }
336
+ defer cli.Close()
337
+
338
+ req, _ := http.NewRequest(http.MethodGet, frontend.URL, nil)
339
+ req.Header.Set("Connection", "Upgrade")
340
+ req.Header.Set("Upgrade", "websocket")
341
+ if err := req.Write(cli); err != nil {
342
+ t.Fatalf("failed to write upgrade request: %v", err)
343
+ }
344
+
345
+ resp, err := http.ReadResponse(bufio.NewReader(cli), &http.Request{Method: http.MethodGet})
346
+ if err != nil {
347
+ t.Fatalf("failed to read upgrade response: %v", err)
348
+ }
349
+ if resp.StatusCode != http.StatusSwitchingProtocols {
350
+ t.Fatalf("status = %d, want 101", resp.StatusCode)
351
+ }
352
+
353
+ var srv *net.TCPConn
354
+ select {
355
+ case srv = <-srvc:
356
+ case <-time.After(5 * time.Second):
357
+ t.Fatal("timed out waiting for the backend connection")
358
+ }
359
+ defer srv.Close()
360
+
361
+ test(t, cli, srv)
362
+ })
363
+ }
364
+}