@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 617 lines 20 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 "encoding/json"
20 "errors"
21 "fmt"
22 "io"
23 "net"
24 "net/http"
25 "net/http/httptest"
26 "os"
27 "path/filepath"
28 "reflect"
29 "runtime"
30 "strings"
31 "sync"
32 "testing"
33 "time"
34
35 "github.com/caddyserver/caddy/v2"
36 )
37
38 // trailerTestBody models net/http populating the original request's trailers
39 // during Read, including readers which return final bytes together with EOF.
40 type trailerTestBody struct {
41 io.Reader
42 finish func()
43 err error
44 closed bool
45 }
46
47 func (b *trailerTestBody) Read(p []byte) (int, error) {
48 if b.closed {
49 return 0, errors.New("body closed")
50 }
51 n, err := b.Reader.Read(p)
52 if err == io.EOF {
53 if b.finish != nil {
54 b.finish()
55 b.finish = nil
56 }
57 if b.err != nil {
58 err = b.err
59 }
60 }
61 return n, err
62 }
63
64 func (b *trailerTestBody) Close() error { b.closed = true; return nil }
65
66 type finalBytesReader struct{ *strings.Reader }
67
68 func (r finalBytesReader) Read(p []byte) (int, error) {
69 n, err := r.Reader.Read(p)
70 if r.Len() == 0 {
71 err = io.EOF
72 }
73 return n, err
74 }
75
76 func TestRequestTrailersEOF(t *testing.T) {
77 for _, limit := range []int64{0, 2, 32, -1} {
78 for _, declared := range []bool{false, true} {
79 for _, finalBytes := range []bool{false, true} {
80 t.Run(fmt.Sprintf("buffer=%d/declared=%t/final-bytes=%t", limit, declared, finalBytes), func(t *testing.T) {
81 req := prepareTestRequest(httptest.NewRequest(http.MethodPut, "http://example.com/", nil))
82 req.ContentLength = -1
83 req.ProtoMajor = 2
84 if declared {
85 req.Trailer = http.Header{"Content-Digest": nil, "X-Checksum": nil, "X-Absent": nil}
86 }
87 want := http.Header{"Content-Digest": {"digest-one", "digest-two"}, "X-Checksum": {"checksum"}}
88 if declared {
89 want["X-Absent"] = nil
90 }
91 var reader io.Reader = strings.NewReader("hello")
92 if finalBytes {
93 reader = finalBytesReader{strings.NewReader("hello")}
94 }
95 body := &trailerTestBody{Reader: reader, finish: func() { req.Trailer = want.Clone() }}
96 req.Body = body
97 h := Handler{RequestBuffers: limit}
98 out, err := h.prepareRequest(req, caddy.NewReplacer())
99 if err != nil {
100 t.Fatal(err)
101 }
102 defer out.Body.Close()
103 if limit == 0 && body.finish == nil {
104 t.Fatal("prepareRequest consumed streaming body")
105 }
106 got, err := io.ReadAll(out.Body)
107 if err != nil || string(got) != "hello" {
108 t.Fatalf("body=%q err=%v", got, err)
109 }
110 for range 2 {
111 n, err := out.Body.Read(make([]byte, 8))
112 if n != 0 || err != io.EOF {
113 t.Fatalf("repeated read: n=%d err=%v", n, err)
114 }
115 }
116 if !reflect.DeepEqual(out.Trailer, want) {
117 t.Fatalf("trailers=%v, want %v", out.Trailer, want)
118 }
119 out.Trailer["Content-Digest"][0] = "outbound change"
120 out.Trailer.Set("X-New", "outbound only")
121 if !reflect.DeepEqual(req.Trailer, want) {
122 t.Fatalf("outbound changes mutated incoming trailers: %v", req.Trailer)
123 }
124 })
125 }
126 }
127 }
128 }
129
130 func TestRequestTrailersReadError(t *testing.T) {
131 for _, readErr := range []error{io.ErrUnexpectedEOF, context.Canceled} {
132 for _, limit := range []int64{0, 2, 32, -1} {
133 t.Run(fmt.Sprintf("%v/buffer=%d", readErr, limit), func(t *testing.T) {
134 req := prepareTestRequest(httptest.NewRequest(http.MethodPut, "http://example.com/", nil))
135 req.ContentLength = -1
136 req.Trailer = http.Header{"Content-Digest": nil}
137 req.Body = &trailerTestBody{Reader: strings.NewReader("partial"), err: readErr, finish: func() {
138 req.Trailer.Set("Content-Digest", "must not forward without EOF")
139 }}
140 out, err := (Handler{RequestBuffers: limit}).prepareRequest(req, caddy.NewReplacer())
141 if err != nil {
142 t.Fatal(err)
143 }
144 defer out.Body.Close()
145 _, err = io.ReadAll(out.Body)
146 // Unlimited buffering currently discards read errors; do not change
147 // that existing policy here. It must still not publish trailers.
148 if limit >= 0 && !errors.Is(err, readErr) {
149 t.Fatalf("read error=%v, want %v", err, readErr)
150 }
151 if len(out.Trailer.Values("Content-Digest")) != 0 {
152 t.Fatalf("published trailers after failed read: %v", out.Trailer)
153 }
154 })
155 }
156 }
157 }
158
159 type trailerObservation struct {
160 Body string
161 Trailer http.Header
162 Header string
163 Proto string
164 }
165
166 func TestRequestTrailersNetwork(t *testing.T) {
167 for _, network := range []string{"tcp", "unix"} {
168 t.Run(network, func(t *testing.T) {
169 if network == "unix" && runtime.GOOS == "windows" {
170 t.Skip("Unix sockets are not supported by this test on Windows")
171 }
172 backend := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
173 body, err := io.ReadAll(r.Body)
174 if err != nil {
175 http.Error(w, err.Error(), 500)
176 return
177 }
178 _ = json.NewEncoder(w).Encode(trailerObservation{string(body), r.Trailer, r.Header.Get("X-Control"), r.Proto})
179 }))
180 dial := backend.Listener.Addr().String()
181 if network == "unix" {
182 _ = backend.Listener.Close()
183 // Keep the socket path below the platform's sockaddr_un limit.
184 dir, err := os.MkdirTemp("", "caddy-trailers-")
185 if err != nil {
186 t.Fatal(err)
187 }
188 defer os.RemoveAll(dir)
189 backend.Listener, err = net.Listen("unix", filepath.Join(dir, "upstream.sock"))
190 if err != nil {
191 t.Fatal(err)
192 }
193 dial = "unix/" + backend.Listener.Addr().String()
194 }
195 backend.Start()
196 defer backend.Close()
197 for _, limit := range []int64{0, 2, 1 << 20, -1} {
198 t.Run(fmt.Sprintf("buffer=%d", limit), func(t *testing.T) {
199 ctx, cancel := caddy.NewContext(caddy.Context{Context: context.Background()})
200 defer cancel()
201 transport := &HTTPTransport{}
202 var err error
203 transport.Transport, err = transport.NewTransport(ctx)
204 if err != nil {
205 t.Fatal(err)
206 }
207 defer transport.Transport.CloseIdleConnections()
208 h := minimalHandler(0, &Upstream{Host: new(Host), Dial: dial})
209 h.Transport = transport
210 h.RequestBuffers = limit
211 front := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
212 w.Header().Set("X-Client-Protocol", r.Proto)
213 if r.ProtoMajor == 1 && len(r.TransferEncoding) != 1 {
214 t.Errorf("HTTP/1 request is not chunked: %v", r.TransferEncoding)
215 }
216 if err := h.ServeHTTP(w, prepareTestRequest(r), nil); err != nil {
217 http.Error(w, err.Error(), 502)
218 }
219 }))
220 front.EnableHTTP2 = true
221 front.StartTLS()
222 defer front.Close()
223 for _, proto := range []int{1, 2} {
224 tr := front.Client().Transport.(*http.Transport).Clone()
225 tr.ForceAttemptHTTP2 = proto == 2
226 tr.Protocols = new(http.Protocols)
227 tr.Protocols.SetHTTP1(proto == 1)
228 tr.Protocols.SetHTTP2(proto == 2)
229 tr.TLSClientConfig.NextProtos = nil
230 client := &http.Client{Transport: tr, Timeout: 5 * time.Second}
231 defer tr.CloseIdleConnections()
232 for _, body := range []string{"", "hello, trailers"} {
233 for _, trailers := range []http.Header{
234 nil,
235 {"Content-Digest": {"first", "second"}, "X-Checksum": {"checksum"}, "X-Absent": nil},
236 } {
237 req, err := http.NewRequest(http.MethodPut, front.URL, io.NopCloser(strings.NewReader(body)))
238 if err != nil {
239 t.Fatal(err)
240 }
241 req.ContentLength = -1
242 req.Trailer = trailers.Clone()
243 req.Header.Set("X-Control", "ordinary header")
244 res, err := client.Do(req)
245 if err != nil {
246 t.Fatal(err)
247 }
248 var got trailerObservation
249 err = json.NewDecoder(res.Body).Decode(&got)
250 _ = res.Body.Close()
251 if err != nil {
252 t.Fatalf("decode status=%d: %v", res.StatusCode, err)
253 }
254 wantProto := fmt.Sprintf("HTTP/%d.%d", proto, 1)
255 if proto == 2 {
256 wantProto = "HTTP/2.0"
257 }
258 if res.Proto != wantProto || res.Header.Get("X-Client-Protocol") != wantProto {
259 t.Fatalf("negotiated protocol: response=%s request=%s want=%s", res.Proto, res.Header.Get("X-Client-Protocol"), wantProto)
260 }
261 if res.StatusCode != 200 || got.Body != body || got.Header != "ordinary header" || got.Proto != "HTTP/1.1" {
262 t.Fatalf("status=%d observation=%+v", res.StatusCode, got)
263 }
264 // An announced but absent field can be represented as nil
265 // or one empty value by the HTTP/1 transport.
266 if len(trailers) == 0 {
267 if len(got.Trailer) != 0 {
268 t.Fatalf("invented trailers: %v", got.Trailer)
269 }
270 } else {
271 want := trailers.Clone()
272 delete(want, "X-Absent")
273 if got.Trailer.Get("X-Absent") != "" {
274 t.Fatalf("absent trailer has value: %v", got.Trailer)
275 }
276 delete(got.Trailer, "X-Absent")
277 if !reflect.DeepEqual(got.Trailer, want) {
278 t.Fatalf("protocol=%s body=%q trailers=%v want=%v", wantProto, body, got.Trailer, want)
279 }
280 }
281 }
282 }
283 }
284 })
285 }
286 })
287 }
288 }
289
290 // trailerRoundTripper makes retry failures deterministic without relying on
291 // racing the operating system to reserve an unused port.
292 type trailerRoundTripper func(*http.Request) (*http.Response, error)
293
294 func (f trailerRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
295
296 func TestRequestTrailersRetry(t *testing.T) {
297 for _, tc := range []struct {
298 name string
299 limit int64
300 dialFailure bool
301 partial bool
302 }{
303 {name: "dial failure", dialFailure: true},
304 {name: "dial failure with partial buffering", limit: 2, dialFailure: true},
305 {name: "fully buffered replay", limit: 32},
306 {name: "unlimited buffered replay", limit: -1},
307 {name: "consumed stream cannot replay", partial: true},
308 {name: "partially buffered stream cannot replay", limit: 2, partial: true},
309 } {
310 t.Run(tc.name, func(t *testing.T) {
311 const payload = "hello, retry trailers"
312 want := http.Header{"Content-Digest": {"first", "second"}, "X-Checksum": {"checksum"}}
313 observed := make(chan trailerObservation, 1)
314 backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
315 body, err := io.ReadAll(r.Body)
316 if err != nil {
317 http.Error(w, err.Error(), 500)
318 return
319 }
320 observed <- trailerObservation{Body: string(body), Trailer: r.Trailer}
321 w.WriteHeader(http.StatusOK)
322 }))
323 defer backend.Close()
324 h := minimalHandler(1, &Upstream{Host: new(Host), Dial: backend.Listener.Addr().String()})
325 h.RequestBuffers = tc.limit
326 transport := testTransport{&http.Transport{}}
327 defer transport.CloseIdleConnections()
328 attempts := 0
329 h.Transport = trailerRoundTripper(func(r *http.Request) (*http.Response, error) {
330 attempts++
331 if attempts == 1 {
332 if !tc.dialFailure {
333 if tc.partial {
334 if _, err := io.ReadFull(r.Body, make([]byte, 3)); err != nil {
335 t.Errorf("initial partial read: %v", err)
336 }
337 } else {
338 got, err := io.ReadAll(r.Body)
339 if err != nil || string(got) != payload || !reflect.DeepEqual(r.Trailer, want) {
340 t.Errorf("initial attempt: body=%q trailers=%v err=%v", got, r.Trailer, err)
341 }
342 }
343 }
344 _ = r.Body.Close()
345 err := errors.New("controlled first-attempt failure")
346 if tc.dialFailure {
347 return nil, DialError{err}
348 }
349 return nil, err
350 }
351 if tc.partial {
352 _, err := io.ReadAll(r.Body)
353 if err == nil {
354 t.Error("partially consumed request body became replayable")
355 }
356 return nil, errors.New("consumed body is not replayable")
357 }
358 return transport.RoundTrip(r)
359 })
360 req := prepareTestRequest(httptest.NewRequest(http.MethodGet, "http://example.com/", nil))
361 req.ContentLength = -1
362 req.Trailer = http.Header{"Content-Digest": nil, "X-Checksum": nil}
363 req.Body = &trailerTestBody{Reader: strings.NewReader(payload), finish: func() { req.Trailer = want.Clone() }}
364 defer req.Body.Close()
365 ctx, cancel := context.WithTimeout(req.Context(), 5*time.Second)
366 defer cancel()
367 req = req.WithContext(ctx)
368 rec := httptest.NewRecorder()
369 err := h.ServeHTTP(rec, req, nil)
370 if attempts != 2 {
371 t.Fatalf("attempts=%d, want 2", attempts)
372 }
373 if tc.partial {
374 if err == nil {
375 t.Fatal("partially consumed body was accepted for replay")
376 }
377 select {
378 case got := <-observed:
379 t.Fatalf("unexpected successful request: %+v", got)
380 default:
381 }
382 return
383 }
384 if err != nil {
385 t.Fatal(err)
386 }
387 select {
388 case got := <-observed:
389 if got.Body != payload || !reflect.DeepEqual(got.Trailer, want) {
390 t.Fatalf("replay=%+v, want body=%q trailers=%v", got, payload, want)
391 }
392 default:
393 t.Fatal("upstream did not receive the replay")
394 }
395 })
396 }
397 }
398
399 func TestRequestTrailersStreaming(t *testing.T) {
400 for _, limit := range []int64{0, 2} {
401 t.Run(fmt.Sprintf("buffer=%d", limit), func(t *testing.T) {
402 prefixRead := make(chan struct{})
403 backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
404 prefix := make([]byte, 5)
405 if _, err := io.ReadFull(r.Body, prefix); err != nil {
406 http.Error(w, err.Error(), 500)
407 return
408 }
409 close(prefixRead)
410 rest, err := io.ReadAll(r.Body)
411 if err != nil {
412 http.Error(w, err.Error(), 500)
413 return
414 }
415 _ = json.NewEncoder(w).Encode(trailerObservation{Body: string(prefix) + string(rest), Trailer: r.Trailer})
416 }))
417 defer backend.Close()
418 h := minimalHandler(0, &Upstream{Host: new(Host), Dial: backend.Listener.Addr().String()})
419 defer h.Transport.(testTransport).CloseIdleConnections()
420 h.RequestBuffers = limit
421 front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
422 if err := h.ServeHTTP(w, prepareTestRequest(r), nil); err != nil {
423 http.Error(w, err.Error(), 502)
424 }
425 }))
426 defer front.Close()
427 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
428 defer cancel()
429 reader, writer := io.Pipe()
430 defer reader.Close()
431 defer writer.Close()
432 writeDone := make(chan error, 1)
433 go func() {
434 _, err := io.WriteString(writer, "hello")
435 if err == nil {
436 select {
437 case <-prefixRead:
438 _, err = io.WriteString(writer, ", streamed")
439 case <-ctx.Done():
440 err = ctx.Err()
441 }
442 }
443 _ = writer.CloseWithError(err)
444 writeDone <- err
445 }()
446 req, err := http.NewRequestWithContext(ctx, http.MethodPut, front.URL, reader)
447 if err != nil {
448 t.Fatal(err)
449 }
450 req.Trailer = http.Header{"Content-Digest": {"stream digest"}}
451 client := front.Client()
452 client.Timeout = 5 * time.Second
453 res, err := client.Do(req)
454 if err != nil {
455 t.Fatal(err)
456 }
457 defer res.Body.Close()
458 var got trailerObservation
459 if err := json.NewDecoder(res.Body).Decode(&got); err != nil {
460 t.Fatal(err)
461 }
462 if got.Body != "hello, streamed" || !reflect.DeepEqual(got.Trailer, req.Trailer) {
463 t.Fatalf("streamed request=%+v", got)
464 }
465 if err := <-writeDone; err != nil {
466 t.Fatal(err)
467 }
468 })
469 }
470 }
471
472 // trailerCloseTracker lets tests wait for transport cleanup, including its
473 // background request writer, without polling goroutine counts.
474 type trailerCloseTracker struct {
475 io.ReadCloser
476 once sync.Once
477 readOnce sync.Once
478 closed chan struct{}
479 readDone chan struct{}
480 }
481
482 func (b *trailerCloseTracker) Read(p []byte) (int, error) {
483 n, err := b.ReadCloser.Read(p)
484 if err != nil {
485 b.readOnce.Do(func() { close(b.readDone) })
486 }
487 return n, err
488 }
489
490 func (b *trailerCloseTracker) Close() error {
491 err := b.ReadCloser.Close()
492 b.once.Do(func() { close(b.closed) })
493 return err
494 }
495
496 func TestRequestTrailersEarlyResponse(t *testing.T) {
497 backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
498 if err := http.NewResponseController(w).EnableFullDuplex(); err != nil {
499 http.Error(w, err.Error(), 500)
500 return
501 }
502 // Ensure the upload has started, then respond while the rest is blocked.
503 if _, err := io.ReadFull(r.Body, make([]byte, 1)); err != nil {
504 http.Error(w, err.Error(), 500)
505 return
506 }
507 w.WriteHeader(http.StatusForbidden)
508 _ = http.NewResponseController(w).Flush()
509 }))
510 defer backend.Close()
511 h := minimalHandler(0, &Upstream{Host: new(Host), Dial: backend.Listener.Addr().String()})
512 defer h.Transport.(testTransport).CloseIdleConnections()
513 reader, writer := io.Pipe()
514 defer reader.Close()
515 defer writer.Close()
516 req := prepareTestRequest(httptest.NewRequest(http.MethodPut, "http://example.com/", reader))
517 body := &trailerCloseTracker{ReadCloser: reader, closed: make(chan struct{}), readDone: make(chan struct{})}
518 req.Body = body
519 req.ContentLength = -1
520 req.Trailer = http.Header{"Content-Digest": nil}
521 ctx, cancel := context.WithTimeout(req.Context(), 5*time.Second)
522 defer cancel()
523 req = req.WithContext(ctx)
524 writeDone := make(chan error, 1)
525 go func() {
526 _, err := io.WriteString(writer, "x")
527 writeDone <- err
528 }()
529 rec := httptest.NewRecorder()
530 if err := h.ServeHTTP(rec, req, nil); err != nil {
531 t.Fatal(err)
532 }
533 if rec.Code != http.StatusForbidden {
534 t.Fatalf("status=%d", rec.Code)
535 }
536 if err := <-writeDone; err != nil {
537 t.Fatal(err)
538 }
539 // The HTTP server owns the incoming body and closes it after the
540 // proxy returns. Model that ownership here; the transport may still be
541 // reading the unfinished upload when an early response arrives.
542 _ = req.Body.Close()
543 select {
544 case <-body.readDone:
545 case <-ctx.Done():
546 t.Fatal("interrupted upload read did not stop")
547 }
548 if req.Trailer.Get("Content-Digest") != "" {
549 t.Fatal("early response invented trailers")
550 }
551 }
552
553 func TestRequestTrailersCancellation(t *testing.T) {
554 started := make(chan struct{})
555 backendDone := make(chan struct{})
556 backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
557 defer close(backendDone)
558 if _, err := io.ReadFull(r.Body, make([]byte, 1)); err != nil {
559 return
560 }
561 close(started)
562 _, err := io.Copy(io.Discard, r.Body)
563 if err == nil {
564 t.Error("canceled upstream body unexpectedly reached EOF")
565 }
566 if r.Trailer.Get("Content-Digest") != "" {
567 t.Error("cancellation published trailer values")
568 }
569 }))
570 defer backend.Close()
571 h := minimalHandler(0, &Upstream{Host: new(Host), Dial: backend.Listener.Addr().String()})
572 defer h.Transport.(testTransport).CloseIdleConnections()
573 reader, writer := io.Pipe()
574 defer reader.Close()
575 defer writer.Close()
576 body := &trailerCloseTracker{ReadCloser: reader, closed: make(chan struct{}), readDone: make(chan struct{})}
577 req := prepareTestRequest(httptest.NewRequest(http.MethodPut, "http://example.com/", body))
578 req.ContentLength = -1
579 req.Trailer = http.Header{"Content-Digest": nil}
580 timeout, stop := context.WithTimeout(req.Context(), 5*time.Second)
581 defer stop()
582 ctx, cancel := context.WithCancel(timeout)
583 defer cancel()
584 req = req.WithContext(ctx)
585 writeDone := make(chan error, 1)
586 go func() { _, err := io.WriteString(writer, "x"); writeDone <- err }()
587 rec := httptest.NewRecorder()
588 done := make(chan error, 1)
589 go func() { done <- h.ServeHTTP(rec, req, nil) }()
590 select {
591 case <-started:
592 case <-timeout.Done():
593 t.Fatal("upstream never received upload prefix")
594 }
595 cancel()
596 // A real client disconnect also breaks the incoming socket read. Canceling
597 // only a synthetic request context cannot unblock an unrelated io.Pipe.
598 _ = writer.CloseWithError(context.Canceled)
599 select {
600 case err := <-done:
601 if err != nil || rec.Code != 499 {
602 t.Fatalf("cancellation: status=%d error=%v", rec.Code, err)
603 }
604 case <-timeout.Done():
605 t.Fatal("proxy did not stop after cancellation")
606 }
607 for _, ch := range []<-chan struct{}{body.closed, body.readDone, backendDone} {
608 select {
609 case <-ch:
610 case <-timeout.Done():
611 t.Fatal("upload cleanup did not complete")
612 }
613 }
614 if err := <-writeDone; err != nil {
615 t.Fatal(err)
616 }
617 }