mirror of
https://github.com/caddyserver/caddy.git
synced 2026-11-03 00:35:05 -05:00
401 lines
11 KiB
Go
401 lines
11 KiB
Go
package reverseproxy
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"runtime"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/caddyserver/caddy/v2"
|
|
"github.com/caddyserver/caddy/v2/modules/caddyhttp"
|
|
)
|
|
|
|
func TestHandlerCopyResponse(t *testing.T) {
|
|
h := Handler{}
|
|
testdata := []string{
|
|
"",
|
|
strings.Repeat("a", defaultBufferSize),
|
|
strings.Repeat("123456789 123456789 123456789 12", 3000),
|
|
}
|
|
|
|
dst := bytes.NewBuffer(nil)
|
|
recorder := httptest.NewRecorder()
|
|
recorder.Body = dst
|
|
|
|
for _, d := range testdata {
|
|
src := bytes.NewBuffer([]byte(d))
|
|
dst.Reset()
|
|
err := h.copyResponse(recorder, src, 0, caddy.Log())
|
|
if err != nil {
|
|
t.Errorf("failed with error: %v", err)
|
|
}
|
|
out := dst.String()
|
|
if out != d {
|
|
t.Errorf("bad read: got %q", out)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestSwitchProtocolCopierBufferSize(t *testing.T) {
|
|
var wg sync.WaitGroup
|
|
var errc = make(chan error, 1)
|
|
var dst bytes.Buffer
|
|
|
|
copier := switchProtocolCopier{
|
|
user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
|
|
backend: nopReadWriteCloser{Writer: &dst},
|
|
wg: &wg,
|
|
bufferSize: 7,
|
|
}
|
|
|
|
buf := copier.buffer()
|
|
if got := len(buf); got != 7 {
|
|
t.Fatalf("buffer len = %d, want 7", got)
|
|
}
|
|
|
|
wg.Add(1)
|
|
go copier.copyToBackend(errc)
|
|
wg.Wait()
|
|
|
|
// the destination cannot be half-closed, so a clean copy reports errCopyDone
|
|
if err := <-errc; !errors.Is(err, errCopyDone) {
|
|
t.Fatalf("copyToBackend() error = %v, want %v", err, errCopyDone)
|
|
}
|
|
if got := dst.String(); got != "hello" {
|
|
t.Fatalf("copied data = %q, want %q", got, "hello")
|
|
}
|
|
}
|
|
|
|
// A clean EOF in one direction must be propagated to the destination as a
|
|
// write-half close instead of tearing the whole tunnel down.
|
|
// See https://github.com/caddyserver/caddy/issues/8026.
|
|
func TestSwitchProtocolCopierHalfClose(t *testing.T) {
|
|
closeWriteErr := errors.New("no half-close")
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
dst io.ReadWriteCloser
|
|
wantErr error // nil means the half-close was propagated
|
|
}{
|
|
{
|
|
name: "destination supports CloseWrite",
|
|
dst: &closeWriteRecorder{},
|
|
wantErr: nil,
|
|
},
|
|
{
|
|
name: "destination does not support CloseWrite",
|
|
dst: nopReadWriteCloser{Writer: io.Discard},
|
|
wantErr: errCopyDone,
|
|
},
|
|
{
|
|
name: "CloseWrite fails",
|
|
dst: &closeWriteRecorder{err: closeWriteErr},
|
|
wantErr: closeWriteErr,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
var wg sync.WaitGroup
|
|
errc := make(chan error, 1)
|
|
copier := switchProtocolCopier{
|
|
user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
|
|
backend: tc.dst,
|
|
wg: &wg,
|
|
}
|
|
|
|
wg.Add(1)
|
|
go copier.copyToBackend(errc)
|
|
wg.Wait()
|
|
|
|
if err := <-errc; !errors.Is(err, tc.wantErr) {
|
|
t.Fatalf("copyToBackend() error = %v, want %v", err, tc.wantErr)
|
|
}
|
|
if rec, ok := tc.dst.(*closeWriteRecorder); ok && !rec.called {
|
|
t.Fatal("CloseWrite() was not called on a destination that supports it")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// A copy error must be reported as-is and must not be mistaken for a clean
|
|
// half-close, so the tunnel is still torn down immediately.
|
|
func TestSwitchProtocolCopierErrorIsNotHalfClose(t *testing.T) {
|
|
var wg sync.WaitGroup
|
|
errc := make(chan error, 1)
|
|
wantErr := errors.New("write failed")
|
|
dst := &closeWriteRecorder{writeErr: wantErr}
|
|
|
|
copier := switchProtocolCopier{
|
|
user: nopReadWriteCloser{Reader: strings.NewReader("hello")},
|
|
backend: dst,
|
|
wg: &wg,
|
|
}
|
|
|
|
wg.Add(1)
|
|
go copier.copyToBackend(errc)
|
|
wg.Wait()
|
|
|
|
if err := <-errc; !errors.Is(err, wantErr) {
|
|
t.Fatalf("error = %v, want %v", err, wantErr)
|
|
}
|
|
if dst.called {
|
|
t.Fatal("CloseWrite() was called after a copy error")
|
|
}
|
|
}
|
|
|
|
func TestSwitchProtocolCopierDefaultBufferSize(t *testing.T) {
|
|
copier := switchProtocolCopier{}
|
|
buf := copier.buffer()
|
|
if got := len(buf); got != defaultBufferSize {
|
|
t.Fatalf("buffer len = %d, want %d", got, defaultBufferSize)
|
|
}
|
|
}
|
|
|
|
type nopReadWriteCloser struct {
|
|
io.Reader
|
|
io.Writer
|
|
}
|
|
|
|
func (nopReadWriteCloser) Close() error { return nil }
|
|
|
|
// closeWriteRecorder is a destination that supports closing only its write
|
|
// half, and records whether that happened.
|
|
type closeWriteRecorder struct {
|
|
called bool
|
|
err error
|
|
writeErr error
|
|
}
|
|
|
|
func (c *closeWriteRecorder) Read([]byte) (int, error) { return 0, io.EOF }
|
|
|
|
func (c *closeWriteRecorder) Write(p []byte) (int, error) {
|
|
if c.writeErr != nil {
|
|
return 0, c.writeErr
|
|
}
|
|
return len(p), nil
|
|
}
|
|
|
|
func (c *closeWriteRecorder) Close() error { return nil }
|
|
|
|
func (c *closeWriteRecorder) CloseWrite() error {
|
|
c.called = true
|
|
return c.err
|
|
}
|
|
|
|
// TestHandlerUpgradedStreamHalfClose drives a real upgraded stream through the
|
|
// handler and checks that closing one direction is propagated to the other side
|
|
// instead of tearing the whole tunnel down.
|
|
//
|
|
// It mirrors net/http/httputil's TestReverseProxyWebSocketHalfTCP, which covers
|
|
// the same class upstream (https://go.dev/issue/35892). Before the fix, the
|
|
// "close write" cases failed: the handler returned as soon as the first
|
|
// direction reported, and the deferred closes killed the other direction before
|
|
// its pending bytes could be delivered.
|
|
//
|
|
// See https://github.com/caddyserver/caddy/issues/8026.
|
|
func TestHandlerUpgradedStreamHalfClose(t *testing.T) {
|
|
switch runtime.GOOS {
|
|
case "plan9", "js", "wasip1":
|
|
t.Skipf("not supported on %s", runtime.GOOS)
|
|
}
|
|
|
|
// Reads carry a deadline so a regression fails the test promptly instead of
|
|
// blocking until the whole package times out.
|
|
const readTimeout = 10 * time.Second
|
|
|
|
mustRead := func(t *testing.T, conn *net.TCPConn, msg string) {
|
|
t.Helper()
|
|
if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
|
|
t.Fatalf("failed to set read deadline: %v", err)
|
|
}
|
|
b := make([]byte, len(msg))
|
|
if _, err := io.ReadFull(conn, b); err != nil {
|
|
t.Fatalf("failed to read: %v", err)
|
|
}
|
|
if got, want := string(b), msg; got != want {
|
|
t.Fatalf("got %#q, want %#q", got, want)
|
|
}
|
|
}
|
|
|
|
mustReadEOF := func(t *testing.T, conn *net.TCPConn) {
|
|
t.Helper()
|
|
if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
|
|
t.Fatalf("failed to set read deadline: %v", err)
|
|
}
|
|
b := make([]byte, 1)
|
|
if _, err := conn.Read(b); !errors.Is(err, io.EOF) {
|
|
t.Fatalf("read after peer half-close: got %v, want EOF", err)
|
|
}
|
|
}
|
|
|
|
mustWrite := func(t *testing.T, conn *net.TCPConn, msg string) {
|
|
t.Helper()
|
|
if _, err := conn.Write([]byte(msg)); err != nil {
|
|
t.Fatalf("failed to write: %v", err)
|
|
}
|
|
}
|
|
|
|
mustCloseRead := func(t *testing.T, conn *net.TCPConn) {
|
|
t.Helper()
|
|
if err := conn.CloseRead(); err != nil {
|
|
t.Fatalf("failed to CloseRead: %v", err)
|
|
}
|
|
}
|
|
|
|
mustCloseWrite := func(t *testing.T, conn *net.TCPConn) {
|
|
t.Helper()
|
|
if err := conn.CloseWrite(); err != nil {
|
|
t.Fatalf("failed to CloseWrite: %v", err)
|
|
}
|
|
}
|
|
|
|
tests := map[string]func(t *testing.T, cli, srv *net.TCPConn){
|
|
"backend close read": func(t *testing.T, cli, srv *net.TCPConn) {
|
|
mustCloseRead(t, srv)
|
|
mustWrite(t, srv, "backend sends")
|
|
mustRead(t, cli, "backend sends")
|
|
},
|
|
"backend close write": func(t *testing.T, cli, srv *net.TCPConn) {
|
|
mustCloseWrite(t, srv)
|
|
mustReadEOF(t, cli)
|
|
mustWrite(t, cli, "client sends")
|
|
mustRead(t, srv, "client sends")
|
|
},
|
|
"client close read": func(t *testing.T, cli, srv *net.TCPConn) {
|
|
mustCloseRead(t, cli)
|
|
mustWrite(t, cli, "client sends")
|
|
mustRead(t, srv, "client sends")
|
|
},
|
|
"client close write": func(t *testing.T, cli, srv *net.TCPConn) {
|
|
mustCloseWrite(t, cli)
|
|
mustReadEOF(t, srv)
|
|
mustWrite(t, srv, "backend sends")
|
|
mustRead(t, cli, "backend sends")
|
|
},
|
|
}
|
|
|
|
for name, test := range tests {
|
|
t.Run(name, func(t *testing.T) {
|
|
srvc := make(chan *net.TCPConn, 1)
|
|
|
|
backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
conn, _, err := http.NewResponseController(w).Hijack()
|
|
if err != nil {
|
|
t.Errorf("backend hijack failed: %v", err)
|
|
return
|
|
}
|
|
tcp, ok := conn.(*net.TCPConn)
|
|
if !ok {
|
|
conn.Close()
|
|
t.Errorf("backend conn is %T, want *net.TCPConn", conn)
|
|
return
|
|
}
|
|
if _, err := io.WriteString(tcp, "HTTP/1.1 101 Switching Protocols\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {
|
|
tcp.Close()
|
|
t.Errorf("backend upgrade write failed: %v", err)
|
|
return
|
|
}
|
|
srvc <- tcp
|
|
}))
|
|
defer backend.Close()
|
|
|
|
h := minimalHandler(0, &Upstream{
|
|
Host: new(Host),
|
|
Dial: backend.Listener.Addr().String(),
|
|
})
|
|
h.connections = make(map[io.ReadWriteCloser]openConnection)
|
|
h.connectionsMu = new(sync.Mutex)
|
|
|
|
frontend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
// Isolate half-close propagation from request cancellation which has
|
|
// its own backend-close path.
|
|
r = r.WithContext(context.WithoutCancel(r.Context()))
|
|
_ = h.ServeHTTP(w, prepareTestRequest(r), caddyhttp.HandlerFunc(func(http.ResponseWriter, *http.Request) error {
|
|
return nil
|
|
}))
|
|
}))
|
|
defer frontend.Close()
|
|
|
|
frontendURL, err := url.Parse(frontend.URL)
|
|
if err != nil {
|
|
t.Fatalf("failed to parse frontend URL: %v", err)
|
|
}
|
|
addr, err := net.ResolveTCPAddr("tcp", frontendURL.Host)
|
|
if err != nil {
|
|
t.Fatalf("failed to resolve TCP address: %v", err)
|
|
}
|
|
cli, err := net.DialTCP("tcp", nil, addr)
|
|
if err != nil {
|
|
t.Fatalf("failed to dial frontend: %v", err)
|
|
}
|
|
defer cli.Close()
|
|
|
|
req, _ := http.NewRequest(http.MethodGet, frontend.URL, nil)
|
|
req.Header.Set("Connection", "Upgrade")
|
|
req.Header.Set("Upgrade", "websocket")
|
|
if err := req.Write(cli); err != nil {
|
|
t.Fatalf("failed to write upgrade request: %v", err)
|
|
}
|
|
|
|
resp, err := http.ReadResponse(bufio.NewReader(cli), &http.Request{Method: http.MethodGet})
|
|
if err != nil {
|
|
t.Fatalf("failed to read upgrade response: %v", err)
|
|
}
|
|
if resp.StatusCode != http.StatusSwitchingProtocols {
|
|
t.Fatalf("status = %d, want 101", resp.StatusCode)
|
|
}
|
|
|
|
var srv *net.TCPConn
|
|
select {
|
|
case srv = <-srvc:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("timed out waiting for the backend connection")
|
|
}
|
|
defer srv.Close()
|
|
|
|
test(t, cli, srv)
|
|
})
|
|
}
|
|
}
|
|
|
|
// A response carrying the Incremental header field (RFC 10036) must be
|
|
// forwarded without buffering, whatever its Content-Type and Content-Length.
|
|
func TestFlushIntervalIncremental(t *testing.T) {
|
|
h := Handler{FlushInterval: caddy.Duration(time.Second)}
|
|
req := httptest.NewRequest(http.MethodGet, "/", nil)
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
incremental string
|
|
receivedIncremental bool
|
|
want time.Duration
|
|
}{
|
|
{name: "incremental", incremental: "?1", want: -1},
|
|
{name: "not incremental", incremental: "?0", want: time.Second},
|
|
{name: "absent", want: time.Second},
|
|
{name: "removed by header operations", receivedIncremental: true, want: -1},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
res := &http.Response{
|
|
Header: http.Header{"Content-Type": []string{"application/json"}},
|
|
ContentLength: 42,
|
|
}
|
|
if tc.incremental != "" {
|
|
res.Header.Set("Incremental", tc.incremental)
|
|
}
|
|
if got := h.flushInterval(req, res, tc.receivedIncremental); got != tc.want {
|
|
t.Errorf("flushInterval() = %v, want %v", got, tc.want)
|
|
}
|
|
})
|
|
}
|
|
}
|