clawbot
d9f7344ae6
check / check (push) Canceled after 0s
When less than 256 KiB of a client's body is left after the app has answered, the rest is read by closing the body, which the transport, ReverseProxy and Go's server each do, not through requestBody.Read. waiting was then false, so a request timeout that ran out took the 504 branch, which set no read deadline, and the request held its connection for as long as the client kept it open. Once the app has answered, that branch now ends reading the client's body at once, whichever of SWWAF_CLIENT_REQUEST_TIMEOUT and SWWAF_UPSTREAM_REQUEST_TIMEOUT ran out. Model: opus-5-5
454 lines
12 KiB
Go
454 lines
12 KiB
Go
package proxy_test
|
|
|
|
import (
|
|
"errors"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"sneak.berlin/go/smallwebwaf/internal/requestlog"
|
|
)
|
|
|
|
// largeBodySize is more than the connections between the client,
|
|
// smallwebwaf and the app can hold while nobody reads, so that a sender
|
|
// soon waits.
|
|
const largeBodySize = 64 << 20
|
|
|
|
// writeSize is how much a test sender writes at a time.
|
|
const writeSize = 32 << 10
|
|
|
|
func TestRequestTimeouts(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
// limit is the setting set to shortTimeout, which runs out; long
|
|
// is one set to longTimeoutSetting, which does not, or "".
|
|
limit, long string
|
|
// appTakesNothing has the app never read, while the client sends
|
|
// as fast as it can; otherwise the app reads, and the client
|
|
// stops sending halfway.
|
|
appTakesNothing bool
|
|
want int
|
|
}{
|
|
{
|
|
name: "client request timeout, waiting on the client",
|
|
limit: clientRequestTimeout,
|
|
want: http.StatusRequestTimeout,
|
|
},
|
|
{
|
|
name: "upstream request timeout, waiting on the client",
|
|
limit: upstreamRequestTimeout,
|
|
long: clientRequestTimeout,
|
|
want: http.StatusRequestTimeout,
|
|
},
|
|
{
|
|
name: "upstream request timeout, waiting on the app",
|
|
limit: upstreamRequestTimeout,
|
|
appTakesNothing: true,
|
|
want: http.StatusGatewayTimeout,
|
|
},
|
|
{
|
|
name: "client request timeout, waiting on the app",
|
|
limit: clientRequestTimeout,
|
|
long: upstreamRequestTimeout,
|
|
appTakesNothing: true,
|
|
want: http.StatusGatewayTimeout,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
var (
|
|
app *httptest.Server
|
|
appURL string
|
|
sendRequest func(*testing.T, string) net.Conn
|
|
appGotBody atomic.Bool
|
|
)
|
|
|
|
if tc.appTakesNothing {
|
|
appURL, sendRequest = startAppThatTakesNothing(t), sendLargeBody
|
|
} else {
|
|
app = startApp(t, func(_ http.ResponseWriter, r *http.Request) {
|
|
n, _ := io.Copy(io.Discard, r.Body)
|
|
appGotBody.Store(n > 0)
|
|
})
|
|
appURL, sendRequest = app.URL, sendPartOfBody
|
|
}
|
|
|
|
env := map[string]string{tc.limit: shortTimeoutSetting, metricsToken: token}
|
|
if tc.long != "" {
|
|
env[tc.long] = longTimeoutSetting
|
|
}
|
|
|
|
addr, out := startProxy(t, appURL, env)
|
|
start := time.Now()
|
|
got := readResponse(t, sendRequest(t, addr))
|
|
wantTimedOut(t, start)
|
|
|
|
want := tc.want
|
|
|
|
if app != nil {
|
|
// Close returns once the app has finished with the request.
|
|
app.Close()
|
|
|
|
// Until some of the body has reached the app, smallwebwaf
|
|
// waits on the app, and SPEC.md asks for 504; the timeout
|
|
// runs out then only if the test process is held up.
|
|
if !appGotBody.Load() {
|
|
want = http.StatusGatewayTimeout
|
|
}
|
|
}
|
|
|
|
wantStatus(t, got, want)
|
|
wantLine(t, out.requestLine(t), want, requestlog.ActionTimedOut)
|
|
wantLimitHits(t, addr, tc.limit, 1)
|
|
})
|
|
}
|
|
}
|
|
|
|
// startAppThatTakesNothing starts an app that accepts connections and
|
|
// never reads from them, and returns its URL.
|
|
func startAppThatTakesNothing(t *testing.T) string {
|
|
t.Helper()
|
|
|
|
listener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0")
|
|
if err != nil {
|
|
t.Fatalf("listen: %v", err)
|
|
}
|
|
|
|
var (
|
|
mu sync.Mutex
|
|
held []net.Conn
|
|
)
|
|
|
|
hold := func(conn net.Conn) {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
held = append(held, conn)
|
|
}
|
|
|
|
go func() {
|
|
for {
|
|
conn, err := listener.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
hold(conn)
|
|
}
|
|
}()
|
|
|
|
t.Cleanup(func() {
|
|
_ = listener.Close()
|
|
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
|
|
for _, conn := range held {
|
|
_ = conn.Close()
|
|
}
|
|
})
|
|
|
|
return "http://" + listener.Addr().String()
|
|
}
|
|
|
|
// sendPartOfBody sends a request that announces a large body, and only
|
|
// the first bytes of it.
|
|
func sendPartOfBody(t *testing.T, addr string) net.Conn {
|
|
t.Helper()
|
|
|
|
conn := dial(t, addr)
|
|
send(t, conn, "POST /upload HTTP/1.1\r\nHost: app\r\nContent-Length: "+
|
|
strconv.Itoa(largeBodySize)+"\r\n\r\nthe first bytes")
|
|
|
|
return conn
|
|
}
|
|
|
|
// sendLargeBody sends a request with a large body, as fast as smallwebwaf
|
|
// takes it, from a goroutine of its own.
|
|
func sendLargeBody(t *testing.T, addr string) net.Conn {
|
|
t.Helper()
|
|
|
|
conn := dial(t, addr)
|
|
send(t, conn, "POST /upload HTTP/1.1\r\nHost: app\r\nContent-Length: "+
|
|
strconv.Itoa(largeBodySize)+"\r\n\r\n")
|
|
|
|
go func() {
|
|
chunk := make([]byte, writeSize)
|
|
for range largeBodySize / writeSize {
|
|
_, err := conn.Write(chunk)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
return conn
|
|
}
|
|
|
|
func TestClientStopsSendingAfterTheAppHasAnswered(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
// answeredAt is when the app answered, or nil if the bytes the client
|
|
// sent never reached it.
|
|
var answeredAt atomic.Pointer[time.Time]
|
|
|
|
// The app answers without reading the whole body, as an app does for
|
|
// a path it does not serve, and only once smallwebwaf waits for the
|
|
// rest of it: the app takes the bytes the client sent, and then waits
|
|
// half the timeout, far longer than smallwebwaf takes to go on to
|
|
// wait.
|
|
app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
|
|
_, err := io.ReadFull(r.Body, make([]byte, len("the first bytes")))
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
time.Sleep(shortTimeout / 2)
|
|
answeredAt.Store(new(time.Now()))
|
|
http.NotFound(w, r)
|
|
})
|
|
addr, out := startProxy(t, app.URL, map[string]string{
|
|
clientRequestTimeout: shortTimeoutSetting,
|
|
metricsToken: token,
|
|
})
|
|
|
|
start := time.Now()
|
|
got := readResponse(t, sendPartOfBody(t, addr))
|
|
wantTimedOut(t, start)
|
|
|
|
// Close returns once the app has finished with the request.
|
|
app.Close()
|
|
|
|
// The app's answer came before the time ran out, so the client is
|
|
// sent that answer, not smallwebwaf's 408. Only a hold-up of the test
|
|
// process can make it come later, and SPEC.md then asks for 408, as
|
|
// smallwebwaf was waiting for the client, or for 504 if the bytes the
|
|
// client sent had not reached the app.
|
|
at := answeredAt.Load()
|
|
want := http.StatusNotFound
|
|
|
|
switch {
|
|
case at == nil:
|
|
want = http.StatusGatewayTimeout
|
|
case at.Sub(start) >= shortTimeout:
|
|
want = http.StatusRequestTimeout
|
|
}
|
|
|
|
wantStatus(t, got, want)
|
|
wantLine(t, out.requestLine(t), want, requestlog.ActionTimedOut)
|
|
wantLimitHits(t, addr, clientRequestTimeout, 1)
|
|
}
|
|
|
|
func TestClientStopsSendingWithLittleOfItsBodyLeft(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
// limit is the setting set to shortTimeout, which runs out, and
|
|
// clientTimeout what SWWAF_CLIENT_REQUEST_TIMEOUT is set to when it
|
|
// is not limit.
|
|
limit, clientTimeout string
|
|
}{
|
|
{
|
|
name: "client request timeout",
|
|
limit: clientRequestTimeout,
|
|
},
|
|
{
|
|
name: "upstream request timeout, client request timeout off",
|
|
limit: upstreamRequestTimeout,
|
|
clientTimeout: off,
|
|
},
|
|
{
|
|
name: "upstream request timeout, client request timeout longer",
|
|
limit: upstreamRequestTimeout,
|
|
clientTimeout: longTimeoutSetting,
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
// appClosed is closed once smallwebwaf has closed its
|
|
// connection to the app, which it does once it has the app's
|
|
// answer.
|
|
appClosed := make(chan struct{})
|
|
|
|
// The app answers without reading the rest of the body, and
|
|
// says it closes the connection, only once smallwebwaf waits
|
|
// for the client to send more: it takes the bytes the client
|
|
// sent, and then waits half the timeout, far longer than
|
|
// smallwebwaf takes to go on to wait.
|
|
app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
|
|
conn, buf, err := http.NewResponseController(w).Hijack()
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
_, _ = io.ReadFull(buf.Reader, make([]byte, len("the first bytes")))
|
|
|
|
time.Sleep(shortTimeout / 2)
|
|
|
|
_, _ = buf.WriteString("HTTP/1.1 404 Not Found\r\n" +
|
|
"Content-Length: 0\r\nConnection: close\r\n\r\n")
|
|
_ = buf.Flush()
|
|
_, _ = io.Copy(io.Discard, buf.Reader)
|
|
_ = conn.Close()
|
|
|
|
close(appClosed)
|
|
})
|
|
|
|
env := map[string]string{tc.limit: shortTimeoutSetting, metricsToken: token}
|
|
if tc.clientTimeout != "" {
|
|
env[clientRequestTimeout] = tc.clientTimeout
|
|
}
|
|
|
|
addr, out := startProxy(t, app.URL, env)
|
|
|
|
// Less than 256 KiB of the body is left after the first bytes,
|
|
// little enough for Go's server in smallwebwaf to read the rest
|
|
// itself before it sends the app's answer.
|
|
start := time.Now()
|
|
conn := dial(t, addr)
|
|
send(t, conn, "POST /upload HTTP/1.1\r\nHost: app\r\nContent-Length: 200000\r\n\r\n"+
|
|
"the first bytes")
|
|
|
|
select {
|
|
case <-appClosed:
|
|
case <-time.After(waitLimit):
|
|
t.Fatal("smallwebwaf did not close its connection to the app")
|
|
}
|
|
|
|
// More of the body, which can no longer go to the app, and then
|
|
// nothing.
|
|
send(t, conn, strings.Repeat("x", 8<<10))
|
|
|
|
got := readResponse(t, conn)
|
|
wantTimedOut(t, start)
|
|
wantStatus(t, got, http.StatusNotFound)
|
|
wantLine(t, out.requestLine(t), http.StatusNotFound, requestlog.ActionTimedOut)
|
|
wantLimitHits(t, addr, tc.limit, 1)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestAppTooSlowToAnswer(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
app := startApp(t, func(_ http.ResponseWriter, r *http.Request) {
|
|
<-r.Context().Done()
|
|
})
|
|
addr, out := startProxy(t, app.URL, map[string]string{
|
|
upstreamResponseTimeout: shortTimeoutSetting,
|
|
metricsToken: token,
|
|
})
|
|
|
|
start := time.Now()
|
|
|
|
wantStatus(t, get(t, addr, "/slow"), http.StatusGatewayTimeout)
|
|
wantTimedOut(t, start)
|
|
|
|
line := out.requestLine(t)
|
|
wantLine(t, line, http.StatusGatewayTimeout, requestlog.ActionTimedOut)
|
|
|
|
_, answered := line.fields["upstream_status"]
|
|
if answered {
|
|
t.Errorf("log line has upstream_status %v for an app that never answered",
|
|
line.fields["upstream_status"])
|
|
}
|
|
|
|
wantLimitHits(t, addr, upstreamResponseTimeout, 1)
|
|
}
|
|
|
|
func TestAppTooSlowToFinishItsAnswer(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
|
|
_, _ = io.WriteString(w, "the first part")
|
|
_ = http.NewResponseController(w).Flush()
|
|
|
|
<-r.Context().Done()
|
|
})
|
|
addr, out := startProxy(t, app.URL, map[string]string{
|
|
upstreamResponseTimeout: shortTimeoutSetting,
|
|
})
|
|
|
|
start := time.Now()
|
|
got := get(t, addr, "/slow")
|
|
wantStatus(t, got, http.StatusOK)
|
|
|
|
if string(got.body) != "the first part" || !errors.Is(got.err, io.ErrUnexpectedEOF) {
|
|
t.Errorf("client read %q (%v), want the first part cut off", got.body, got.err)
|
|
}
|
|
|
|
wantTimedOut(t, start)
|
|
|
|
line := out.requestLine(t)
|
|
wantLine(t, line, http.StatusOK, requestlog.ActionTimedOut)
|
|
|
|
if line.UpstreamStatus != http.StatusOK {
|
|
t.Errorf("log line has upstream_status %d, want %d",
|
|
line.UpstreamStatus, http.StatusOK)
|
|
}
|
|
}
|
|
|
|
func TestClientTooSlowToTakeTheAnswer(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
|
|
chunk := make([]byte, writeSize)
|
|
for range largeBodySize / writeSize {
|
|
_, err := w.Write(chunk)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
})
|
|
addr, out := startProxy(t, app.URL, map[string]string{
|
|
clientResponseTimeout: shortTimeoutSetting,
|
|
metricsToken: token,
|
|
})
|
|
|
|
start := time.Now()
|
|
|
|
// The client asks, and never reads the answer.
|
|
conn := dial(t, addr)
|
|
send(t, conn, "GET /large HTTP/1.1\r\nHost: app\r\n\r\n")
|
|
|
|
line := out.requestLine(t)
|
|
wantTimedOut(t, start)
|
|
wantLine(t, line, http.StatusOK, requestlog.ActionTimedOut)
|
|
wantLimitHits(t, addr, clientResponseTimeout, 1)
|
|
}
|
|
|
|
func TestClosesAnIdleConnection(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
app := startApp(t, func(http.ResponseWriter, *http.Request) {})
|
|
addr, _ := startProxy(t, app.URL, map[string]string{
|
|
clientIdleTimeout: shortTimeoutSetting,
|
|
})
|
|
|
|
// The idle time starts once the answer is sent, so after start.
|
|
start := time.Now()
|
|
conn := dial(t, addr)
|
|
send(t, conn, "GET / HTTP/1.1\r\nHost: app\r\n\r\n")
|
|
wantStatus(t, readResponse(t, conn), http.StatusOK)
|
|
|
|
// The read deadline readResponse set still bounds this read.
|
|
_, err := conn.Read(make([]byte, 1))
|
|
if !errors.Is(err, io.EOF) {
|
|
t.Fatalf("read on the idle connection: %v, want it closed", err)
|
|
}
|
|
|
|
wantTimedOut(t, start)
|
|
}
|