@marcel91200 / caddy Mirror · read-only
Fork 0

Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS

Mirrored from https://github.com/Slazee/caddy.git

Synced Nov 11, 01:00 UTC

Code Commits Issues Pulls Cloud Agent CICD
master
master
go 525 lines 14.8 KB
Raw
1 // Copyright 2015 Matthew Holt and The Caddy Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 package reverseproxy
16
17 import (
18 "context"
19 "crypto/tls"
20 "errors"
21 "fmt"
22 "io"
23 "net"
24 "net/http"
25 "runtime"
26 "sync"
27 "testing"
28 "time"
29
30 "github.com/quic-go/webtransport-go"
31 "go.uber.org/zap"
32 )
33
34 // pumpTestTopology spins up:
35 //
36 // client → frontend(Server F) → upstream(Server U)
37 //
38 // Server U is the real upstream; its handler is provided by the test.
39 // Server F's handler dials U and runs runWebTransportPump between the two
40 // sessions, so the client (who dials F) effectively talks to U through
41 // the pump.
42 type pumpTestTopology struct {
43 frontendAddr *net.UDPAddr
44 clientTLS *tls.Config
45 shutdown func()
46 }
47
48 func newPumpTestTopology(t *testing.T, upstreamHandler func(*webtransport.Session, *http.Request)) *pumpTestTopology {
49 t.Helper()
50
51 uAddr, uRoot, uShutdown := startTestWebTransportServer(t, upstreamHandler)
52
53 fAddr, fRoot, fShutdown := startTestWebTransportServer(t, func(clientSess *webtransport.Session, _ *http.Request) {
54 // Dial U.
55 dialCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
56 defer cancel()
57 url := fmt.Sprintf("https://localhost:%d/", uAddr.Port)
58 _, upstreamSess, err := dialUpstreamWebTransport(dialCtx, clientTLSFor(uRoot), url, nil, nil, "")
59 if err != nil {
60 t.Errorf("frontend: dial upstream: %v", err)
61 _ = clientSess.CloseWithError(0, "upstream dial failed")
62 return
63 }
64 runWebTransportPump(clientSess, upstreamSess, zap.NewNop())
65 })
66
67 return &pumpTestTopology{
68 frontendAddr: fAddr,
69 clientTLS: clientTLSFor(fRoot),
70 shutdown: func() {
71 fShutdown()
72 uShutdown()
73 },
74 }
75 }
76
77 // dialFrontend returns a fresh session dialed against the frontend server.
78 func (tt *pumpTestTopology) dialFrontend(t *testing.T) *webtransport.Session {
79 t.Helper()
80 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
81 defer cancel()
82 url := fmt.Sprintf("https://localhost:%d/", tt.frontendAddr.Port)
83 _, sess, err := dialUpstreamWebTransport(ctx, tt.clientTLS, url, nil, nil, "")
84 if err != nil {
85 t.Fatalf("client dial frontend: %v", err)
86 }
87 return sess
88 }
89
90 // echoUpstream is a ready-made upstream handler that echoes bytes on every
91 // bidirectional stream it's given.
92 func echoUpstream(sess *webtransport.Session, _ *http.Request) {
93 ctx := sess.Context()
94 for {
95 str, err := sess.AcceptStream(ctx)
96 if err != nil {
97 return
98 }
99 go func(s *webtransport.Stream) {
100 _, _ = io.Copy(s, s)
101 _ = s.Close()
102 }(str)
103 }
104 }
105
106 func TestPump_BidiStreamClientToUpstream(t *testing.T) {
107 if testing.Short() {
108 t.Skip()
109 }
110 tt := newPumpTestTopology(t, echoUpstream)
111 t.Cleanup(tt.shutdown)
112
113 sess := tt.dialFrontend(t)
114 defer sess.CloseWithError(0, "")
115
116 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
117 defer cancel()
118 str, err := sess.OpenStreamSync(ctx)
119 if err != nil {
120 t.Fatalf("open stream: %v", err)
121 }
122
123 const payload = "hello from client"
124 if _, err := io.WriteString(str, payload); err != nil {
125 t.Fatalf("write: %v", err)
126 }
127 if err := str.Close(); err != nil {
128 t.Fatalf("close write: %v", err)
129 }
130 got, err := io.ReadAll(str)
131 if err != nil {
132 t.Fatalf("read: %v", err)
133 }
134 if string(got) != payload {
135 t.Fatalf("echo mismatch: got %q want %q", got, payload)
136 }
137 }
138
139 func TestPump_BidiStreamUpstreamToClient(t *testing.T) {
140 if testing.Short() {
141 t.Skip()
142 }
143 // Upstream opens a stream toward the client and sends data.
144 serverReady := make(chan struct{})
145 const payload = "hello from upstream"
146 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
147 defer close(serverReady)
148 ctx, cancel := context.WithTimeout(sess.Context(), 5*time.Second)
149 defer cancel()
150 str, err := sess.OpenStreamSync(ctx)
151 if err != nil {
152 t.Errorf("upstream open: %v", err)
153 return
154 }
155 if _, err := io.WriteString(str, payload); err != nil {
156 t.Errorf("upstream write: %v", err)
157 return
158 }
159 _ = str.Close()
160 // Keep the session alive briefly so the stream can be drained client-side.
161 <-sess.Context().Done()
162 })
163 t.Cleanup(tt.shutdown)
164
165 sess := tt.dialFrontend(t)
166 defer sess.CloseWithError(0, "")
167
168 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
169 defer cancel()
170 str, err := sess.AcceptStream(ctx)
171 if err != nil {
172 t.Fatalf("client accept: %v", err)
173 }
174 got, err := io.ReadAll(str)
175 if err != nil {
176 t.Fatalf("client read: %v", err)
177 }
178 if string(got) != payload {
179 t.Fatalf("bytes mismatch: got %q want %q", got, payload)
180 }
181 }
182
183 func TestPump_UniStreamClientToUpstream(t *testing.T) {
184 if testing.Short() {
185 t.Skip()
186 }
187 // Upstream: accept one uni stream and echo its bytes on a new uni stream
188 // back to the client.
189 const payload = "uni from client"
190 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
191 ctx := sess.Context()
192 recv, err := sess.AcceptUniStream(ctx)
193 if err != nil {
194 return
195 }
196 data, err := io.ReadAll(recv)
197 if err != nil {
198 t.Errorf("upstream read uni: %v", err)
199 return
200 }
201 send, err := sess.OpenUniStreamSync(ctx)
202 if err != nil {
203 t.Errorf("upstream open uni: %v", err)
204 return
205 }
206 if _, err := send.Write(data); err != nil {
207 t.Errorf("upstream write uni: %v", err)
208 }
209 _ = send.Close()
210 <-sess.Context().Done()
211 })
212 t.Cleanup(tt.shutdown)
213
214 sess := tt.dialFrontend(t)
215 defer sess.CloseWithError(0, "")
216
217 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
218 defer cancel()
219 sendStr, err := sess.OpenUniStreamSync(ctx)
220 if err != nil {
221 t.Fatalf("client open uni: %v", err)
222 }
223 if _, err := sendStr.Write([]byte(payload)); err != nil {
224 t.Fatalf("client write uni: %v", err)
225 }
226 _ = sendStr.Close()
227
228 recvStr, err := sess.AcceptUniStream(ctx)
229 if err != nil {
230 t.Fatalf("client accept uni: %v", err)
231 }
232 got, err := io.ReadAll(recvStr)
233 if err != nil {
234 t.Fatalf("client read uni: %v", err)
235 }
236 if string(got) != payload {
237 t.Fatalf("uni echo mismatch: got %q want %q", got, payload)
238 }
239 }
240
241 func TestPump_Datagram(t *testing.T) {
242 if testing.Short() {
243 t.Skip()
244 }
245 // Upstream echoes whatever datagram it receives.
246 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
247 ctx := sess.Context()
248 for {
249 d, err := sess.ReceiveDatagram(ctx)
250 if err != nil {
251 return
252 }
253 _ = sess.SendDatagram(d)
254 }
255 })
256 t.Cleanup(tt.shutdown)
257
258 sess := tt.dialFrontend(t)
259 defer sess.CloseWithError(0, "")
260
261 // Datagrams are unreliable. Retry a few times to get one round-trip.
262 payload := []byte("dgram")
263 deadline := time.Now().Add(3 * time.Second)
264 for {
265 if time.Now().After(deadline) {
266 t.Fatal("no datagram echo observed within deadline")
267 }
268 if err := sess.SendDatagram(payload); err != nil {
269 t.Fatalf("send datagram: %v", err)
270 }
271 ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
272 got, err := sess.ReceiveDatagram(ctx)
273 cancel()
274 if err == nil && string(got) == string(payload) {
275 return
276 }
277 }
278 }
279
280 func TestPump_CloseWithErrorPropagatesClientToUpstream(t *testing.T) {
281 if testing.Short() {
282 t.Skip()
283 }
284 upstreamSawErr := make(chan error, 1)
285 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
286 // Use a fresh long-lived context so we don't race with
287 // sess.Context() firing and getting a context error instead of
288 // the session-level error.
289 _, err := sess.AcceptStream(context.Background())
290 upstreamSawErr <- err
291 })
292 t.Cleanup(tt.shutdown)
293
294 sess := tt.dialFrontend(t)
295
296 // Client closes with a specific code; pump should propagate to upstream.
297 const code webtransport.SessionErrorCode = 4242
298 const msg = "client bye"
299 if err := sess.CloseWithError(code, msg); err != nil {
300 t.Fatalf("client close: %v", err)
301 }
302
303 select {
304 case err := <-upstreamSawErr:
305 if err == nil {
306 t.Fatal("upstream expected error after client close; got nil")
307 }
308 // Close propagation is best-effort for a client-initiated close:
309 // webtransport-go's Transport tears down the dedicated QUIC connection
310 // immediately after CloseWithError, and on the pump's server-side
311 // session the WT_CLOSE_SESSION capsule can lose the race to the
312 // QUIC close — in which case parseNextCapsule stores a non-
313 // SessionError and the code is unrecoverable. The invariant we
314 // can reliably enforce is "upstream observed a session-terminating
315 // error." If the code did survive, assert it matches.
316 var sessErr *webtransport.SessionError
317 if errors.As(err, &sessErr) && sessErr.ErrorCode != 0 {
318 if sessErr.ErrorCode != code || sessErr.Message != msg {
319 t.Errorf("upstream saw code=%d msg=%q, want code=%d msg=%q",
320 sessErr.ErrorCode, sessErr.Message, code, msg)
321 }
322 } else {
323 t.Logf("upstream saw %T: %v (code lost to QUIC-close race; close propagation itself is verified)", err, err)
324 }
325 case <-time.After(3 * time.Second):
326 t.Fatal("upstream did not observe close in time")
327 }
328 }
329
330 func TestPump_CloseWithErrorPropagatesUpstreamToClient(t *testing.T) {
331 if testing.Short() {
332 t.Skip()
333 }
334 const code webtransport.SessionErrorCode = 9000
335 const msg = "upstream bye"
336 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
337 _ = sess.CloseWithError(code, msg)
338 })
339 t.Cleanup(tt.shutdown)
340
341 sess := tt.dialFrontend(t)
342
343 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
344 defer cancel()
345 _, err := sess.AcceptStream(ctx)
346 var sessErr *webtransport.SessionError
347 if !errors.As(err, &sessErr) {
348 t.Fatalf("expected SessionError, got %T: %v", err, err)
349 }
350 if sessErr.ErrorCode != code || sessErr.Message != msg {
351 t.Errorf("client saw code=%d msg=%q, want code=%d msg=%q",
352 sessErr.ErrorCode, sessErr.Message, code, msg)
353 }
354 }
355
356 // TestPump_SessionLifecycle_NoGoroutineLeak sanity-checks that after both
357 // sessions end, the pump's goroutines unwind. We compare goroutine counts
358 // before and after, with a small tolerance because the Go runtime has
359 // background goroutines we can't synchronize with.
360 func TestPump_SessionLifecycle_NoGoroutineLeak(t *testing.T) {
361 if testing.Short() {
362 t.Skip()
363 }
364 before := runtime.NumGoroutine()
365
366 // Drive a fast session+close cycle.
367 var wg sync.WaitGroup
368 wg.Add(1)
369 go func() {
370 defer wg.Done()
371 tt := newPumpTestTopology(t, echoUpstream)
372 sess := tt.dialFrontend(t)
373 _ = sess.CloseWithError(0, "")
374 // Allow close to propagate and goroutines to exit.
375 time.Sleep(200 * time.Millisecond)
376 tt.shutdown()
377 }()
378 wg.Wait()
379
380 // Give the runtime a moment to finish tearing down.
381 deadline := time.Now().Add(2 * time.Second)
382 var after int
383 for {
384 after = runtime.NumGoroutine()
385 if after <= before+8 || time.Now().After(deadline) {
386 break
387 }
388 time.Sleep(50 * time.Millisecond)
389 }
390 // Allow some slack — test infrastructure itself keeps a few goroutines.
391 if after > before+16 {
392 t.Errorf("possible goroutine leak: before=%d after=%d", before, after)
393 }
394 }
395
396 // TestPump_StreamResetReleasesWorkers resets many bidi streams without
397 // closing the session. The opposite copy must not stay blocked, or each
398 // reset leaks a worker for the life of the session.
399 func TestPump_StreamResetReleasesWorkers(t *testing.T) {
400 if testing.Short() {
401 t.Skip()
402 }
403 tt := newPumpTestTopology(t, func(sess *webtransport.Session, _ *http.Request) {
404 ctx := sess.Context()
405 for {
406 if _, err := sess.AcceptStream(ctx); err != nil {
407 return
408 }
409 }
410 })
411 t.Cleanup(tt.shutdown)
412
413 sess := tt.dialFrontend(t)
414 defer sess.CloseWithError(0, "")
415
416 // Let the session-level accept loops settle before counting.
417 time.Sleep(100 * time.Millisecond)
418 before := runtime.NumGoroutine()
419
420 const n = 15
421 for i := range n {
422 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
423 str, err := sess.OpenStreamSync(ctx)
424 cancel()
425 if err != nil {
426 t.Fatalf("open stream %d: %v", i, err)
427 }
428 if _, err := str.Write([]byte("x")); err != nil {
429 t.Fatalf("write stream %d: %v", i, err)
430 }
431 str.CancelWrite(0)
432 }
433
434 deadline := time.Now().Add(2 * time.Second)
435 var after int
436 for {
437 after = runtime.NumGoroutine()
438 if after <= before+8 || time.Now().After(deadline) {
439 break
440 }
441 time.Sleep(50 * time.Millisecond)
442 }
443 if after > before+8 {
444 t.Fatalf("stream workers still running after reset: before=%d after=%d", before, after)
445 }
446 }
447
448 // TestPump_ShutdownWaitsForBlockedStream closes the session while a stream
449 // copy is still inside the pump. The pump must not return until that worker
450 // finishes.
451 func TestPump_ShutdownWaitsForBlockedStream(t *testing.T) {
452 if testing.Short() {
453 t.Skip()
454 }
455
456 blocked := make(chan struct{})
457 release := make(chan struct{})
458 var once sync.Once
459 orig := streamCopy
460 streamCopy = func(dst io.Writer, src io.Reader) (int64, error) {
461 once.Do(func() { close(blocked) })
462 <-release
463 return orig(dst, src)
464 }
465 t.Cleanup(func() { streamCopy = orig })
466
467 pumpDone := make(chan struct{})
468 uAddr, uRoot, uShutdown := startTestWebTransportServer(t, func(sess *webtransport.Session, _ *http.Request) {
469 <-sess.Context().Done()
470 })
471 t.Cleanup(uShutdown)
472 fAddr, fRoot, fShutdown := startTestWebTransportServer(t, func(clientSess *webtransport.Session, _ *http.Request) {
473 defer close(pumpDone)
474 dialCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
475 defer cancel()
476 url := fmt.Sprintf("https://localhost:%d/", uAddr.Port)
477 _, upstreamSess, err := dialUpstreamWebTransport(dialCtx, clientTLSFor(uRoot), url, nil, nil, "")
478 if err != nil {
479 t.Errorf("frontend: dial upstream: %v", err)
480 _ = clientSess.CloseWithError(0, "upstream dial failed")
481 return
482 }
483 runWebTransportPump(clientSess, upstreamSess, zap.NewNop())
484 })
485 t.Cleanup(fShutdown)
486
487 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
488 defer cancel()
489 url := fmt.Sprintf("https://localhost:%d/", fAddr.Port)
490 _, sess, err := dialUpstreamWebTransport(ctx, clientTLSFor(fRoot), url, nil, nil, "")
491 if err != nil {
492 t.Fatalf("client dial: %v", err)
493 }
494
495 strCtx, strCancel := context.WithTimeout(context.Background(), 2*time.Second)
496 str, err := sess.OpenStreamSync(strCtx)
497 strCancel()
498 if err != nil {
499 t.Fatalf("open stream: %v", err)
500 }
501 if _, err := str.Write([]byte("x")); err != nil {
502 t.Fatalf("write: %v", err)
503 }
504 defer str.CancelRead(0)
505
506 select {
507 case <-blocked:
508 case <-time.After(2 * time.Second):
509 t.Fatal("stream worker did not start")
510 }
511
512 _ = sess.CloseWithError(0, "")
513 select {
514 case <-pumpDone:
515 t.Fatal("pump returned while a stream worker was still blocked")
516 case <-time.After(300 * time.Millisecond):
517 }
518
519 close(release)
520 select {
521 case <-pumpDone:
522 case <-time.After(2 * time.Second):
523 t.Fatal("pump did not return after the stream worker finished")
524 }
525 }