rproxy, the proxy in front of each app, asks servewaf and fails open (closes #171)

`smallwebwaf rproxy` reads the first 64 KiB of each body, asks servewaf over
HTTP/1.1, and passes what servewaf allows to the app within its limits. Any
other answer goes to the visitor as servewaf gave it. SWRP_DEFAULT_DISPOSITION
decides only when servewaf cannot be connected to, does not answer at all, or
does not answer within SWRP_TIMEOUT_MS. A method SPEC.md does not list gets 405.

Duplication: internal/rproxy copies passing-on code from internal/proxy until
#172 removes the originals.
Rule suppressed: gosec G704 and G705; gosec at
internal/smallwebwaf/rproxy_test.go:431; tagliatelle at
internal/requestlog/requestlog.go:201; paralleltest on one test.
Judgement call: an answer over 1 MiB, or with a status below 200, gets 502.
Judgement call: every target not starting with one / gets 400.

Model: opus-5-5

internal/rproxy/bodies.go

@@ -0,0 +1,211 @@
+package rproxy
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"sync/atomic"
+)
+
+// errResponseTooLarge ends an app's response body that is longer than
+// SWRP_MAX_RESPONSE_SIZE.
+var errResponseTooLarge = errors.New(
+	"the response body is over SWRP_MAX_RESPONSE_SIZE")
+
+// requestBody is the visitor's request body. rproxy first reads up to the first
+// 64 KiB of it into held, and then, on its way to the app, the transport reads
+// it on a goroutine of its own.
+type requestBody struct {
+	// body is the visitor's body, ending in an *http.MaxBytesError past
+	// SWRP_MAX_REQUEST_SIZE.
+	body io.ReadCloser
+	rq   *request
+	// held is the part read before the request goes to the app, which Read gives
+	// first.
+	held []byte
+	// waiting is true while a Read waits for the visitor to send more.
+	waiting atomic.Bool
+	// received is true once the visitor has sent the whole body.
+	received atomic.Bool
+	// bytes is how much of the body has been read.
+	bytes atomic.Int64
+}
+
+// Read reads from the visitor's body, after the part held, which has been
+// counted already.
+func (b *requestBody) Read(p []byte) (int, error) {
+	if len(b.held) > 0 {
+		n := copy(p, b.held)
+		b.held = b.held[n:]
+
+		return n, nil
+	}
+
+	b.waiting.Store(true)
+	n, err := b.body.Read(p)
+	b.waiting.Store(false)
+	b.bytes.Add(int64(n))
+
+	var tooLarge *http.MaxBytesError
+
+	switch {
+	case errors.Is(err, io.EOF):
+		b.received.Store(true)
+	case errors.As(err, &tooLarge):
+		b.rq.refuse(http.StatusRequestEntityTooLarge)
+	}
+
+	return n, err
+}
+
+// Close closes the visitor's body.
+func (b *requestBody) Close() error {
+	return b.body.Close()
+}
+
+// responseBody is the app's response body on its way to the visitor.
+type responseBody struct {
+	// body is the app's body, ending in an *http.MaxBytesError past
+	// SWRP_MAX_RESPONSE_SIZE.
+	body io.ReadCloser
+	rq   *request
+}
+
+// Read reads from the app's body.
+func (b *responseBody) Read(p []byte) (int, error) {
+	n, err := b.body.Read(p)
+	if err == nil {
+		return n, nil
+	}
+
+	var tooLarge *http.MaxBytesError
+
+	switch {
+	case errors.Is(err, io.EOF):
+		b.rq.responseReceived()
+	case errors.As(err, &tooLarge):
+		b.rq.refuse(http.StatusBadGateway)
+
+		return n, errResponseTooLarge
+	case b.rq.in.Context().Err() == nil:
+		// The answer broke off, not because the visitor went away. If a time
+		// limit cut it, that refusal came first and is the one kept.
+		b.rq.refuse(http.StatusBadGateway)
+	}
+
+	return n, err
+}
+
+// Close closes the app's body.
+func (b *responseBody) Close() error {
+	return b.body.Close()
+}
+
+// upgradedConn is the connection to the app once the app has switched
+// protocols, as for a WebSocket. ReverseProxy writes to it what the visitor
+// sends and reads from it what the app sends, on goroutines of its own, until
+// the connection closes; it counts the bytes each way, for the report.
+type upgradedConn struct {
+	io.ReadWriteCloser
+
+	// fromApp is how many bytes the app has sent, and toApp how many the
+	// visitor has.
+	fromApp atomic.Int64
+	toApp   atomic.Int64
+}
+
+// Read reads what the app sends.
+func (c *upgradedConn) Read(p []byte) (int, error) {
+	n, err := c.ReadWriteCloser.Read(p)
+	c.fromApp.Add(int64(n))
+
+	return n, err
+}
+
+// Write sends the app what the visitor sent.
+func (c *upgradedConn) Write(p []byte) (int, error) {
+	n, err := c.ReadWriteCloser.Write(p)
+	c.toApp.Add(int64(n))
+
+	return n, err
+}
+
+// CloseWrite tells the app that the visitor sends no more, while what the app
+// sends still passes. ReverseProxy calls it once the visitor has stopped
+// sending, and closes the connection there if it is not supported.
+func (c *upgradedConn) CloseWrite() error {
+	conn, ok := c.ReadWriteCloser.(interface{ CloseWrite() error })
+	if !ok {
+		return http.ErrNotSupported
+	}
+
+	return conn.CloseWrite()
+}
+
+// limitBody returns body, cut off with an *http.MaxBytesError after maxBytes, or
+// unchanged if maxBytes is zero, which is off.
+func limitBody(body io.ReadCloser, maxBytes int64) io.ReadCloser {
+	if maxBytes == 0 {
+		return body
+	}
+
+	// Without a ResponseWriter, MaxBytesReader only counts and cuts off.
+	return http.MaxBytesReader(nil, body, maxBytes)
+}
+
+// responseWriter is the response to the visitor. It notes the status and size
+// for the log line and the report.
+type responseWriter struct {
+	http.ResponseWriter
+
+	// status is the final status sent, or zero before one is.
+	status int
+	bytes  int64
+}
+
+// WriteHeader sends the status and headers. An informational 1xx status is
+// passed on and the final status still comes later.
+func (w *responseWriter) WriteHeader(status int) {
+	if status >= http.StatusOK && w.status == 0 {
+		w.status = status
+	}
+
+	w.ResponseWriter.WriteHeader(status)
+}
+
+// Write sends part of the body.
+func (w *responseWriter) Write(p []byte) (int, error) {
+	if w.status == 0 {
+		w.status = http.StatusOK
+	}
+
+	n, err := w.ResponseWriter.Write(p) //nolint:gosec // the answer, as it was given
+	w.bytes += int64(n)
+
+	return n, err
+}
+
+// FlushError sends what has been written so far. http.ResponseController calls
+// it, as ReverseProxy does after each write.
+func (w *responseWriter) FlushError() error {
+	return http.NewResponseController(w.ResponseWriter).Flush()
+}
+
+// Hijack takes over the connection to the visitor, which ReverseProxy does only
+// to pass on the app's 101 Switching Protocols: that is then the status sent.
+func (w *responseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	conn, buffered, err := http.NewResponseController(w.ResponseWriter).Hijack()
+	if err == nil {
+		w.status = http.StatusSwitchingProtocols
+	}
+
+	return conn, buffered, err
+}
+
+// Unwrap lets http.ResponseController reach net/http's own ResponseWriter for
+// what responseWriter does not do itself.
+func (w *responseWriter) Unwrap() http.ResponseWriter {
+	return w.ResponseWriter
+}

internal/rproxy/client.go

@@ -0,0 +1,122 @@
+package rproxy
+
+import (
+	"net/http"
+	"net/netip"
+	"slices"
+	"strings"
+)
+
+// The headers rproxy sets on the requests it passes to the app. Go writes every
+// header name in this form, as it sends it and as it receives it, so
+// X-Client-ASN arrives as X-Client-Asn, and Del removes a visitor's own whatever
+// their case; header names are not case-sensitive.
+const (
+	requestIDHeader = "X-Request-ID"
+	asnHeader       = "X-Client-Asn"
+	countryHeader   = "X-Client-Country"
+)
+
+// peerAddress is the address of the request's TCP peer, normally traefik, without
+// any IPv6 zone, which servewaf refuses.
+func peerAddress(r *http.Request) netip.Addr {
+	addrPort, err := netip.ParseAddrPort(r.RemoteAddr)
+	if err != nil {
+		return netip.Addr{}
+	}
+
+	return addrPort.Addr().Unmap().WithZone("")
+}
+
+// clientAddress works out who the visitor is. A peer outside the trusted
+// proxies is the visitor, and what it says in X-Forwarded-For is ignored. For a
+// peer inside them, X-Forwarded-For is read from the right, and the first
+// address outside them is the visitor; if every address in it is inside, the
+// leftmost is, and with no header, the peer. An entry that is not an address
+// ends the reading, since nothing to its left can be believed. The visitor's
+// address never has an IPv6 zone, which servewaf refuses.
+func clientAddress(
+	peer netip.Addr, forwardedFor []string, trusted []netip.Prefix,
+) netip.Addr {
+	client := peer
+	if !isInside(peer, trusted) {
+		return client
+	}
+
+	entries := strings.Split(strings.Join(forwardedFor, ","), ",")
+	for _, entry := range slices.Backward(entries) {
+		addr, err := netip.ParseAddr(strings.TrimSpace(entry))
+		if err != nil {
+			break
+		}
+
+		client = addr.Unmap().WithZone("")
+		if !isInside(client, trusted) {
+			break
+		}
+	}
+
+	return client
+}
+
+// scheme is how the visitor reached traefik, as a trusted proxy says in
+// X-Forwarded-Proto, or otherwise http, the only scheme rproxy serves.
+func scheme(r *http.Request, peerTrusted bool) string {
+	proto := r.Header.Get("X-Forwarded-Proto")
+	if !peerTrusted || proto == "" {
+		return "http"
+	}
+
+	return proto
+}
+
+// isInside reports whether addr is in one of the netblocks.
+func isInside(addr netip.Addr, netblocks []netip.Prefix) bool {
+	return slices.ContainsFunc(netblocks, func(netblock netip.Prefix) bool {
+		return netblock.Contains(addr)
+	})
+}
+
+// setForwardedHeaders sets the headers in which the app learns about the
+// visitor, so that it sees what it would see from traefik directly. A trusted
+// proxy's forwarded headers pass on, with the proxy's own address added to
+// X-Forwarded-For. Those of any other peer are its own claims and are replaced:
+// X-Forwarded-For names the peer, X-Forwarded-Host the host it asked for, and
+// X-Forwarded-Proto plain http, which is how it reached rproxy.
+func setForwardedHeaders(in, out *http.Request, peer netip.Addr, trusted bool) {
+	forwardedFor := peer.String()
+
+	if trusted {
+		// ReverseProxy removes these from out before Rewrite.
+		for _, name := range []string{"Forwarded", "X-Forwarded-Host", "X-Forwarded-Proto"} {
+			values, ok := in.Header[name]
+			if ok {
+				out.Header[name] = values
+			}
+		}
+
+		prior := in.Header.Values("X-Forwarded-For")
+		if len(prior) > 0 {
+			forwardedFor = strings.Join(prior, ", ") + ", " + forwardedFor
+		}
+
+		out.Header.Set("X-Forwarded-For", forwardedFor)
+
+		return
+	}
+
+	// ReverseProxy has removed Forwarded and the three set below; these are the
+	// other headers in which traefik tells the app about the visitor and its
+	// request.
+	for _, name := range []string{
+		"X-Forwarded-Port", "X-Forwarded-Server", "X-Forwarded-Uri",
+		"X-Forwarded-Method", "X-Forwarded-Prefix", "X-Forwarded-Tls-Client-Cert",
+		"X-Forwarded-Tls-Client-Cert-Info", "X-Real-Ip",
+	} {
+		out.Header.Del(name)
+	}
+
+	out.Header.Set("X-Forwarded-For", forwardedFor)
+	out.Header.Set("X-Forwarded-Host", in.Host)
+	out.Header.Set("X-Forwarded-Proto", "http")
+}

internal/rproxy/limits_test.go

@@ -0,0 +1,811 @@
+package rproxy_test
+
+import (
+	"bytes"
+	"io"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/config"
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+	"sneak.berlin/go/smallwebwaf/internal/rproxy"
+)
+
+// post is the start of a POST whose headers end with what follows it.
+const post = "POST / HTTP/1.1\r\nHost: site.example\r\n"
+
+// announcedLength is the header that announces a body of size bytes, and the
+// end of the headers.
+func announcedLength(size int) string {
+	return "Content-Length: " + strconv.Itoa(size) + "\r\n\r\n"
+}
+
+func TestServerHasTheFixedLimits(t *testing.T) {
+	t.Parallel()
+
+	cfg, err := config.RProxyFromEnvironment(func(name string) (string, bool) {
+		if name == "SWRP_WAF_BACKEND" {
+			return nowhere, true
+		}
+
+		return "", false
+	})
+	if err != nil {
+		t.Fatalf("settings: %v", err)
+	}
+
+	server := rproxy.New(rproxy.Params{
+		Config:     cfg,
+		RequestLog: io.Discard,
+		ProcessLog: requestlog.NewProcessLogger(io.Discard, "test", slog.LevelInfo),
+	})
+
+	if server.Addr != ":8080" || server.MaxHeaderBytes != 28<<10 ||
+		server.IdleTimeout != 2*time.Minute || server.ReadHeaderTimeout != time.Minute {
+		t.Errorf("server listens on %q with header limit %d, idle time %s and "+
+			"header timeout %s", server.Addr, server.MaxHeaderBytes,
+			server.IdleTimeout, server.ReadHeaderTimeout)
+	}
+}
+
+func TestABodyAnnouncedOverItsLimitIsRefusedOnceServeWAFAllowsIt(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, limit string
+		size, asked int
+	}{
+		{"with the limit within the first part", "1K", 1025, 1024},
+		{"with the limit past the first part", "128K", 129 << 10, questionBodyMaxBytes},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, func(http.ResponseWriter, *http.Request) {
+				t.Error("the app was asked")
+			})
+			addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				maxRequestSize: tc.limit,
+			})
+
+			got := visit(t, newRequest(t, http.MethodPost, addr, "/",
+				bytes.NewReader(make([]byte, tc.size))))
+			if got.status != http.StatusRequestEntityTooLarge {
+				t.Errorf("answered %+v, want 413", got)
+			}
+
+			if asked := servewaf.question(t); len(asked.body) != tc.asked {
+				t.Errorf("the question carries %d bytes, want %d", len(asked.body), tc.asked)
+			}
+
+			report := servewaf.report(t)
+			if report.Status != http.StatusRequestEntityTooLarge {
+				t.Errorf("report %+v, want status 413", report)
+			}
+		})
+	}
+}
+
+func TestADenyOutranksTheBodySizeLimit(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, deny)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxRequestSize: "1K",
+	})
+
+	got := visit(t, newRequest(t, http.MethodPost, addr, "/",
+		bytes.NewReader(make([]byte, 1025))))
+	if got.status != http.StatusForbidden {
+		t.Errorf("answered %+v, want servewaf's 403", got)
+	}
+}
+
+func TestABodyNotAnnouncedOverItsLimitIsRefused(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, limit string
+		size        int
+		// mayReachApp is true when the part of the body the limit lets through
+		// goes on to the app before the limit is met. The app may still not have
+		// started on the request when it is cut off.
+		mayReachApp bool
+	}{
+		{"within the first part", "1K", 2 << 10, false},
+		{"past the first part", "128K", 129 << 10, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			var asked atomic.Bool
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+				asked.Store(true)
+				greet(w, r)
+			})
+			addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				maxRequestSize: tc.limit,
+			})
+
+			conn := dial(t, addr)
+			send(t, conn, post+"Transfer-Encoding: chunked\r\n\r\n"+
+				strconv.FormatInt(int64(tc.size), 16)+"\r\n"+strings.Repeat("x", tc.size)+
+				"\r\n0\r\n\r\n")
+
+			if status := readStatus(t, conn); status != http.StatusRequestEntityTooLarge {
+				t.Errorf("status %d, want 413", status)
+			}
+
+			if asked.Load() && !tc.mayReachApp {
+				t.Error("the app was asked")
+			}
+		})
+	}
+}
+
+func TestAResponseOverItsLimit(t *testing.T) {
+	t.Parallel()
+
+	large := strings.Repeat("x", 2<<10)
+
+	t.Run("announced, is refused", func(t *testing.T) {
+		t.Parallel()
+
+		servewaf := startServeWAF(t, allow)
+		app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set(contentLengthHeader, strconv.Itoa(len(large)))
+			_, _ = io.WriteString(w, large)
+		})
+		addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+			maxResponseSize: "1K",
+		})
+
+		got := get(t, addr, "/")
+		if got.status != http.StatusBadGateway {
+			t.Errorf("answered %d, want 502", got.status)
+		}
+
+		if report := servewaf.report(t); report.Status != http.StatusBadGateway {
+			t.Errorf("report %+v, want status 502", report)
+		}
+	})
+
+	t.Run("not announced, is cut off", func(t *testing.T) {
+		t.Parallel()
+
+		servewaf := startServeWAF(t, allow)
+		app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = io.WriteString(w, large[:1])
+			_ = http.NewResponseController(w).Flush()
+			_, _ = io.WriteString(w, large[1:])
+		})
+		addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+			maxResponseSize: "1K",
+		})
+
+		conn := dial(t, addr)
+		send(t, conn, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+
+		text, err := io.ReadAll(conn)
+		if err != nil || bytes.Contains(text, []byte(large)) {
+			t.Errorf("read %d bytes (%v), want the answer cut off", len(text), err)
+		}
+
+		report := servewaf.report(t)
+		if report.Status != http.StatusOK || report.ResponseBytes > 1<<10 {
+			t.Errorf("report %+v, want status 200 and at most 1K of the answer", report)
+		}
+	})
+}
+
+func TestAHeadWhoseAnswerAnnouncesMoreThanTheResponseLimitIsPassedOn(t *testing.T) {
+	t.Parallel()
+
+	const announced = "2048"
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set(contentLengthHeader, announced)
+	})
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxResponseSize: "1K",
+	})
+
+	got := visit(t, newRequest(t, http.MethodHead, addr, "/", http.NoBody))
+	if got.status != http.StatusOK || got.header.Get(contentLengthHeader) != announced {
+		t.Errorf("answered %+v, want the app's 200 announcing %s bytes", got, announced)
+	}
+}
+
+func TestASettingOfOffSetsNoLimit(t *testing.T) {
+	t.Parallel()
+
+	for _, setting := range []string{
+		maxRequestSize, maxResponseSize, maxRequestTime, maxResponseTime,
+	} {
+		t.Run(setting, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, greet)
+			addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				setting: "off",
+			})
+
+			// A body and an answer that a limit of zero would cut off, as it would the
+			// next answer, which rproxy gives itself.
+			wantGreeting(t, visit(t, newRequest(t, http.MethodPost, addr, "/",
+				strings.NewReader("a body"))))
+
+			if got := get(t, addr, proxy.HealthPath); got.status != http.StatusOK {
+				t.Errorf("answered %+v, want rproxy's own 200", got)
+			}
+		})
+	}
+}
+
+func TestAVisitorTooSlowToSendTheFirstPartIsAnswered408(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, rest string
+	}{
+		{"with its length announced", announcedLength(100) + "some of it"},
+		{"with no length announced", "Transfer-Encoding: chunked\r\n\r\na\r\nsome of it\r\n"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, greet)
+			addr, out := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				maxRequestTime: "500",
+			})
+
+			conn := dial(t, addr)
+			send(t, conn, post+tc.rest)
+
+			if status := readStatus(t, conn); status != http.StatusRequestTimeout {
+				t.Errorf("status %d, want 408", status)
+			}
+
+			line := out.requestLine(t)
+			if line["disposition"] != nil || line["request_bytes"] != 10.0 {
+				t.Errorf("line %v, want no disposition and 10 bytes", line)
+			}
+
+			if len(servewaf.questions) != 0 {
+				t.Error("servewaf was asked")
+			}
+
+			servewaf.wantNoReport(t)
+		})
+	}
+}
+
+func TestAVisitorThatGoesAwayBeforeTheFirstPartIsNotAsked(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, greet)
+	addr, out := startRProxy(t, servewaf.URL, app.URL, nil)
+
+	conn := dial(t, addr)
+	send(t, conn, post+announcedLength(100)+"some of it")
+	_ = conn.Close()
+
+	line := out.requestLine(t)
+	if line["status"] != 0.0 || line["disposition"] != nil {
+		t.Errorf("line %v, want no answer and no disposition", line)
+	}
+
+	if len(servewaf.questions) != 0 {
+		t.Error("servewaf was asked")
+	}
+}
+
+func TestABodyThatCannotBeReadClosesTheConnectionWithoutAnAnswer(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, request string
+		stopsSending  bool
+	}{
+		{
+			"a malformed chunk", post + "Transfer-Encoding: chunked\r\n\r\nnot a chunk\r\n",
+			false,
+		},
+		{"a visitor that stops sending", post + announcedLength(100) + "some of it", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, greet)
+			addr, out := startRProxy(t, servewaf.URL, app.URL, nil)
+
+			conn := dial(t, addr)
+			send(t, conn, tc.request)
+
+			if tc.stopsSending {
+				tcpConn, ok := conn.(*net.TCPConn)
+				if !ok {
+					t.Fatalf("the connection is a %T", conn)
+				}
+
+				err := tcpConn.CloseWrite()
+				if err != nil {
+					t.Fatalf("stop sending: %v", err)
+				}
+			}
+
+			text, err := io.ReadAll(conn)
+			if err != nil || len(text) != 0 {
+				t.Errorf("read %q (%v), want the connection closed without an answer",
+					text, err)
+			}
+
+			line := out.requestLine(t)
+			if line["status"] != 0.0 || line["disposition"] != nil {
+				t.Errorf("line %v, want no answer and no disposition", line)
+			}
+
+			if len(servewaf.questions) != 0 {
+				t.Error("servewaf was asked")
+			}
+		})
+	}
+}
+
+// deadlineRecorder notes the time limits rproxy sets on reading the visitor's body
+// and on sending its answer.
+type deadlineRecorder struct {
+	*httptest.ResponseRecorder
+
+	readDeadline  time.Time
+	writeDeadline time.Time
+}
+
+// SetReadDeadline notes deadline.
+func (w *deadlineRecorder) SetReadDeadline(deadline time.Time) error {
+	w.readDeadline = deadline
+
+	return nil
+}
+
+// SetWriteDeadline notes deadline.
+func (w *deadlineRecorder) SetWriteDeadline(deadline time.Time) error {
+	w.writeDeadline = deadline
+
+	return nil
+}
+
+func TestABodyLeftUnreadIsReadOnlyUntilTheRequestTimeLimit(t *testing.T) {
+	t.Parallel()
+
+	const (
+		limit = 5 * time.Minute
+		// announced is over what Go's server reads of a body left unread, so that
+		// the app's own server answers without waiting for the rest.
+		announced = 100 * questionBodyMaxBytes
+	)
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusNoContent)
+	})
+	_, _, server := startServer(t, servewaf.URL, app.URL, map[string]string{
+		maxRequestTime: strconv.FormatInt(limit.Milliseconds(), 10),
+	})
+
+	// The visitor sends the first part and a little more, and then nothing.
+	stalled, stop := io.Pipe()
+
+	t.Cleanup(func() { _ = stop.Close() })
+
+	req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/",
+		io.MultiReader(bytes.NewReader(make([]byte, questionBodyMaxBytes+1)), stalled))
+	req.ContentLength = announced
+	req.Header.Set(contentLengthHeader, strconv.Itoa(announced))
+
+	w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+
+	before := time.Now()
+
+	server.Handler.ServeHTTP(w, req)
+
+	if w.Code != http.StatusNoContent {
+		t.Errorf("status %d, want the app's 204", w.Code)
+	}
+
+	if w.readDeadline.Before(before.Add(limit)) ||
+		w.readDeadline.After(time.Now().Add(limit)) {
+		t.Errorf("the reading of the body ends at %s, want %s from the request",
+			w.readDeadline, limit)
+	}
+}
+
+func TestAnAnswerRProxyGivesItselfHasTheResponseTimeToBeSent(t *testing.T) {
+	t.Parallel()
+
+	const limit = 5 * time.Minute
+
+	for _, tc := range []struct {
+		name, target string
+		status       int
+	}{
+		{"servewaf's deny", "/", http.StatusForbidden},
+		{"rproxy's own 400", "x:admin", http.StatusBadRequest},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, deny)
+			app := startApp(t, greet)
+			_, _, server := startServer(t, servewaf.URL, app.URL, map[string]string{
+				maxResponseTime: strconv.FormatInt(limit.Milliseconds(), 10),
+			})
+
+			req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, tc.target,
+				http.NoBody)
+			w := &deadlineRecorder{ResponseRecorder: httptest.NewRecorder()}
+
+			before := time.Now()
+
+			server.Handler.ServeHTTP(w, req)
+
+			if w.Code != tc.status {
+				t.Errorf("status %d, want %d", w.Code, tc.status)
+			}
+
+			if w.writeDeadline.Before(before.Add(limit)) ||
+				w.writeDeadline.After(time.Now().Add(limit)) {
+				t.Errorf("the time limit on sending ends at %s, want %s from the answer",
+					w.writeDeadline, limit)
+			}
+		})
+	}
+}
+
+func TestAVisitorTooSlowToSendTheRestIsAnswered408(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxRequestTime: "3000",
+	})
+
+	conn := dial(t, addr)
+	send(t, conn, post+announcedLength(2*questionBodyMaxBytes)+
+		strings.Repeat("x", questionBodyMaxBytes+1))
+
+	if status := readStatus(t, conn); status != http.StatusRequestTimeout {
+		t.Errorf("status %d, want 408", status)
+	}
+
+	if asked := servewaf.question(t); len(asked.body) != questionBodyMaxBytes {
+		t.Errorf("the question carries %d bytes", len(asked.body))
+	}
+
+	if report := servewaf.report(t); report.Status != http.StatusRequestTimeout {
+		t.Errorf("report %+v, want status 408", report)
+	}
+}
+
+// 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)
+	}
+
+	stopped := make(chan struct{})
+
+	t.Cleanup(func() {
+		close(stopped)
+
+		_ = listener.Close()
+	})
+
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				return
+			}
+
+			go func() {
+				<-stopped
+
+				_ = conn.Close()
+			}()
+		}
+	}()
+
+	return "http://" + listener.Addr().String()
+}
+
+func TestAnAppTooSlowToTakeTheRequestIsAnswered504(t *testing.T) {
+	t.Parallel()
+
+	// More than the connections hold while nobody reads, written a little at a
+	// time, so that rproxy soon waits for the app.
+	const (
+		bodySize  = 64 << 20
+		writeSize = 32 << 10
+	)
+
+	servewaf := startServeWAF(t, allow)
+	addr, _ := startRProxy(t, servewaf.URL, startAppThatTakesNothing(t),
+		map[string]string{maxRequestTime: "3000"})
+
+	conn := dial(t, addr)
+	send(t, conn, post+announcedLength(bodySize))
+
+	go func() {
+		part := make([]byte, writeSize)
+
+		for range bodySize / writeSize {
+			_, err := conn.Write(part)
+			if err != nil {
+				return
+			}
+		}
+	}()
+
+	if status := readStatus(t, conn); status != http.StatusGatewayTimeout {
+		t.Errorf("status %d, want 504", status)
+	}
+}
+
+func TestAnAppTooSlowToAnswerIsAnswered504(t *testing.T) {
+	t.Parallel()
+
+	release := make(chan struct{})
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(http.ResponseWriter, *http.Request) {
+		<-release
+	})
+
+	t.Cleanup(func() { close(release) })
+
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxResponseTime: "300",
+	})
+
+	got := get(t, addr, "/")
+	if got.status != http.StatusGatewayTimeout {
+		t.Errorf("answered %d, want 504", got.status)
+	}
+
+	if report := servewaf.report(t); report.Status != http.StatusGatewayTimeout {
+		t.Errorf("report %+v, want status 504", report)
+	}
+}
+
+func TestAnAppTooSlowToFinishItsAnswerIsCutOff(t *testing.T) {
+	t.Parallel()
+
+	release := make(chan struct{})
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, "the start")
+		_ = http.NewResponseController(w).Flush()
+
+		<-release
+	})
+
+	t.Cleanup(func() { close(release) })
+
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxResponseTime: "300",
+	})
+
+	conn := dial(t, addr)
+	send(t, conn, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+
+	// dial's deadline ends this read with an error if the connection stays open.
+	_, err := io.ReadAll(conn)
+	if err != nil {
+		t.Errorf("read: %v, want the connection closed", err)
+	}
+
+	if report := servewaf.report(t); report.Status != http.StatusOK {
+		t.Errorf("report %+v, want the status the visitor got", report)
+	}
+}
+
+func TestAVisitorTooSlowToTakeTheAnswerIsCutOff(t *testing.T) {
+	t.Parallel()
+
+	// More than the connections hold while nobody reads.
+	const (
+		bodySize  = 64 << 20
+		writeSize = 32 << 10
+	)
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+		part := make([]byte, writeSize)
+
+		for range bodySize / writeSize {
+			_, err := w.Write(part)
+			if err != nil {
+				return
+			}
+		}
+	})
+	addr, out := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxResponseTime: "500",
+	})
+
+	// The visitor asks, and never reads the answer.
+	conn := dial(t, addr)
+	send(t, conn, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+
+	line := out.requestLine(t)
+
+	responseBytes, _ := line["response_bytes"].(float64)
+	if line["status"] != float64(http.StatusOK) || responseBytes >= bodySize {
+		t.Errorf("line %v, want status 200 and the answer cut off", line)
+	}
+
+	report := servewaf.report(t)
+	if report.Status != http.StatusOK || report.ResponseBytes >= bodySize {
+		t.Errorf("report %+v, want status 200 and the answer cut off", report)
+	}
+}
+
+func TestAnAppThatCannotBeReachedIsAnswered502WithAReport(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, allow)
+	addr, _ := startRProxy(t, servewaf.URL, nowhere, nil)
+
+	got := get(t, addr, "/")
+	if got.status != http.StatusBadGateway {
+		t.Errorf("answered %d, want 502", got.status)
+	}
+
+	if report := servewaf.report(t); report.Status != http.StatusBadGateway {
+		t.Errorf("report %+v, want status 502", report)
+	}
+}
+
+func TestAVisitorThatStopsSendingAfterTheAppHasAnsweredGetsTheAnswer(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, greeting)
+	})
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxRequestTime: "3000",
+	})
+
+	// Far more announced than sent, so that Go's server answers for the app
+	// without waiting for the rest.
+	conn := dial(t, addr)
+	send(t, conn, post+announcedLength(100*questionBodyMaxBytes)+
+		strings.Repeat("x", questionBodyMaxBytes))
+
+	if status := readStatus(t, conn); status != http.StatusOK {
+		t.Errorf("status %d, want the app's 200", status)
+	}
+}
+
+func TestAVisitorThatStopsSendingWhileTheAppWorksGetsNoAnswer(t *testing.T) {
+	t.Parallel()
+
+	working := make(chan struct{})
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(_ http.ResponseWriter, r *http.Request) {
+		close(working)
+		<-r.Context().Done()
+	})
+	addr, out := startRProxy(t, servewaf.URL, app.URL, nil)
+
+	conn := dial(t, addr)
+	send(t, conn, "GET / HTTP/1.1\r\nHost: site.example\r\n\r\n")
+
+	select {
+	case <-working:
+	case <-time.After(waitLimit):
+		t.Fatal("the app was not asked")
+	}
+
+	tcpConn, ok := conn.(*net.TCPConn)
+	if !ok {
+		t.Fatalf("the connection is a %T", conn)
+	}
+
+	err := tcpConn.CloseWrite()
+	if err != nil {
+		t.Fatalf("stop sending: %v", err)
+	}
+
+	text, err := io.ReadAll(conn)
+	if err != nil || len(text) != 0 {
+		t.Errorf("read %q (%v), want the connection closed without an answer", text, err)
+	}
+
+	if line := out.requestLine(t); line["status"] != 0.0 {
+		t.Errorf("line %v, want status 0", line)
+	}
+}
+
+func TestTheRequestTimeLimitDoesNotCutAnAnswerOnceTheAppHasTheRequest(t *testing.T) {
+	t.Parallel()
+
+	const answer = "the end of a long answer"
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.Copy(io.Discard, r.Body)
+		_ = http.NewResponseController(w).Flush()
+
+		time.Sleep(2100 * time.Millisecond)
+
+		_, _ = io.WriteString(w, answer)
+	})
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		maxRequestTime: "2000",
+	})
+
+	got := visit(t, newRequest(t, http.MethodPost, addr, "/", strings.NewReader("a body")))
+	if got.status != http.StatusOK || got.body != answer {
+		t.Errorf("answered %d %q, want the app's whole answer", got.status, got.body)
+	}
+}
+
+func TestAVisitorThatStopsSendingGetsAnAnswerRProxyGivesByTheDeadline(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, start string
+		status      int
+	}{
+		{"a deny", post, http.StatusForbidden},
+		{
+			"the health check",
+			"GET " + proxy.HealthPath + " HTTP/1.1\r\nHost: site.example\r\n",
+			http.StatusOK,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, deny)
+			app := startApp(t, greet)
+			addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				maxRequestTime: "500",
+			})
+
+			// Go's server waits for a little of the body left before it sends an answer,
+			// and the time for sending the request ends that wait.
+			conn := dial(t, addr)
+			send(t, conn, tc.start+announcedLength(questionBodyMaxBytes+100)+
+				strings.Repeat("x", questionBodyMaxBytes))
+
+			if status := readStatus(t, conn); status != tc.status {
+				t.Errorf("status %d, want %d", status, tc.status)
+			}
+		})
+	}
+}

internal/rproxy/question_test.go

@@ -0,0 +1,477 @@
+package rproxy_test
+
+import (
+	"bytes"
+	"crypto/rand"
+	"io"
+	"maps"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strconv"
+	"strings"
+	"testing"
+
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+)
+
+// questionBodyMaxBytes is the most of a body a question carries.
+const questionBodyMaxBytes = proxy.QuestionBodyMaxBytes
+
+// chunked is the encoding of a body whose length the visitor did not announce.
+const chunked = "chunked"
+
+// pathGoReencodes is a path Go's client would decode and escape again, %2F turned
+// into / and | into %7C, unless told to send it as it is. A test sets it as the
+// visitor's URL.Opaque, so that the visitor sends it unchanged too.
+const pathGoReencodes = "/some%2Fpath/with%20space|pipe"
+
+func TestQuestionIsTheVisitorsRequestWithRProxysHeaders(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, nil)
+
+	const target = pathGoReencodes + "?b=2&a=1&bad=%zz;x"
+
+	req := newRequest(t, http.MethodDelete, addr, target, http.NoBody)
+	req.URL.Opaque = pathGoReencodes
+	req.Host = site
+	req.Header.Set("X-Test", "from the visitor")
+	req.Header.Set(forwardedForHeader, "198.51.100.1")
+	req.Header.Set(forwardedProtoHeader, "gopher")
+	req.Header.Set(requestIDHeader, "the-visitor's")
+	req.Header.Set(proxy.ClientHeader, "198.51.100.2")
+	req.Header.Set(proxy.PeerHeader, "198.51.100.3")
+	req.Header.Set(proxy.ContentLengthHeader, "12345")
+	// Go's client then sends no User-Agent, so that one in the question would be
+	// Go's own.
+	req.Header["User-Agent"] = nil
+
+	wantGreeting(t, visit(t, req))
+
+	asked := servewaf.question(t)
+	if asked.method != http.MethodDelete || asked.target != target ||
+		asked.host != site {
+		t.Errorf("asked about %s %s at %s", asked.method, asked.target, asked.host)
+	}
+
+	if asked.header.Get(requestIDHeader) == "the-visitor's" {
+		t.Error("the question carries the visitor's id")
+	}
+
+	// Exactly these: visit's connection has the visitor send Connection: close.
+	want := http.Header{}
+	want.Set("Connection", "close")
+	want.Set("X-Test", "from the visitor")
+	want.Set(forwardedForHeader, "198.51.100.1")
+	want.Set(forwardedProtoHeader, "http")
+	want.Set(requestIDHeader, asked.header.Get(requestIDHeader))
+	want.Set(proxy.ClientHeader, localhost)
+	want.Set(proxy.PeerHeader, localhost)
+
+	if !maps.EqualFunc(asked.header, want, slices.Equal) {
+		t.Errorf("the question's headers are %v, want %v", asked.header, want)
+	}
+}
+
+func TestQuestionNamesTheVisitorATrustedProxySpeaksFor(t *testing.T) {
+	t.Parallel()
+
+	const visitor = "203.0.113.9"
+
+	for _, tc := range []struct {
+		name, trusted, client, proto, forwarded string
+	}{
+		{
+			"from a trusted proxy", localhost + "/32", visitor, "https",
+			visitor + ", " + localhost,
+		},
+		{"from a peer not trusted", "10.0.0.0/8", localhost, "http", localhost},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			forwarded := make(chan http.Header, 1)
+			app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+				forwarded <- r.Header.Clone()
+
+				greet(w, r)
+			})
+			addr, out := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+				trustedProxies: tc.trusted,
+			})
+
+			req := newRequest(t, http.MethodGet, addr, "/", http.NoBody)
+			req.Header.Set(forwardedForHeader, visitor)
+			req.Header.Set(forwardedProtoHeader, "https")
+			wantGreeting(t, visit(t, req))
+
+			asked := servewaf.question(t)
+			if asked.header.Get(proxy.ClientHeader) != tc.client ||
+				asked.header.Get(proxy.PeerHeader) != localhost ||
+				asked.header.Get(forwardedForHeader) != visitor ||
+				asked.header.Get(forwardedProtoHeader) != tc.proto {
+				t.Errorf("the question's headers are %v", asked.header)
+			}
+
+			if line := out.requestLine(t); line["client_ip"] != tc.client {
+				t.Errorf("line %v, want client_ip %s", line, tc.client)
+			}
+
+			if got := (<-forwarded).Get(forwardedForHeader); got != tc.forwarded {
+				t.Errorf("the app saw %s %q, want %q", forwardedForHeader, got, tc.forwarded)
+			}
+		})
+	}
+}
+
+// randomBytes returns n random bytes.
+func randomBytes(n int) []byte {
+	body := make([]byte, n)
+	_, _ = rand.Read(body) // it never returns an error
+
+	return body
+}
+
+func TestFirst64KiBReachServeWAFAndTheWholeBodyTheApp(t *testing.T) {
+	t.Parallel()
+
+	for _, size := range []int{
+		1, questionBodyMaxBytes - 1, questionBodyMaxBytes, questionBodyMaxBytes + 1,
+		5 * questionBodyMaxBytes,
+	} {
+		for _, announced := range []bool{true, false} {
+			name := strconv.Itoa(size) + " bytes, length announced: " +
+				strconv.FormatBool(announced)
+
+			t.Run(name, func(t *testing.T) {
+				t.Parallel()
+
+				body := randomBytes(size)
+				servewaf := startServeWAF(t, allow)
+				received := make(chan []byte, 1)
+				app := startApp(t, func(w http.ResponseWriter, r *http.Request) {
+					whole, _ := io.ReadAll(r.Body)
+					received <- whole
+
+					greet(w, r)
+				})
+				addr, _ := startRProxy(t, servewaf.URL, app.URL, nil)
+
+				req := newRequest(t, http.MethodPost, addr, "/", bytes.NewReader(body))
+				if !announced {
+					// Go's client sends a body of unknown length in chunks.
+					req.ContentLength = -1
+				}
+
+				wantGreeting(t, visit(t, req))
+
+				asked := servewaf.question(t)
+				want := body[:min(size, questionBodyMaxBytes)]
+				wantQuestionBody(t, asked, want, size, announced)
+
+				if !bytes.Equal(<-received, body) {
+					t.Error("the app did not get the whole body")
+				}
+
+				if report := servewaf.report(t); report.RequestBytes != int64(size) {
+					t.Errorf("report %+v, of %d bytes", report, size)
+				}
+			})
+		}
+	}
+}
+
+// wantQuestionBody checks the body a question carries, want, and how it gives
+// the length of the visitor's body, of size bytes.
+func wantQuestionBody(
+	t *testing.T, asked question, want []byte, size int, announced bool,
+) {
+	t.Helper()
+
+	if !bytes.Equal(asked.body, want) {
+		t.Errorf("the question carries %d bytes, want the first %d", len(asked.body),
+			len(want))
+	}
+
+	values := asked.header.Values(proxy.ContentLengthHeader)
+
+	if announced {
+		if !slices.Equal(values, []string{strconv.Itoa(size)}) ||
+			asked.contentLength != int64(len(want)) {
+			t.Errorf("announced %v and length %d, want %d and %d", values,
+				asked.contentLength, size, len(want))
+		}
+
+		return
+	}
+
+	if len(values) != 0 || !slices.Equal(asked.transferEncoding, []string{chunked}) {
+		t.Errorf("announced %v and encoded %v, want none and chunked", values,
+			asked.transferEncoding)
+	}
+}
+
+func TestDenyNeedsOnlyTheFirstPartOfALargeBody(t *testing.T) {
+	t.Parallel()
+
+	servewaf := startServeWAF(t, deny)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, nil)
+
+	// Far more announced than sent. Go's server does not wait for that much, so
+	// the answer comes at once.
+	conn := dial(t, addr)
+	send(t, conn, post+announcedLength(100*questionBodyMaxBytes)+
+		strings.Repeat("x", questionBodyMaxBytes))
+
+	if status := readStatus(t, conn); status != http.StatusForbidden {
+		t.Errorf("status %d, want servewaf's 403", status)
+	}
+
+	if report := servewaf.report(t); report.RequestBytes != questionBodyMaxBytes {
+		t.Errorf("report %+v, want the first part's bytes", report)
+	}
+}
+
+func TestAQuestionCarriesItsBodyWhateverTheMethod(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, framing, body string
+		encoding            []string
+	}{
+		{
+			chunked, "Transfer-Encoding: chunked\r\n\r\n4\r\nbody\r\n0\r\n\r\n", "body",
+			[]string{chunked},
+		},
+		{
+			"chunked and empty", "Transfer-Encoding: chunked\r\n\r\n0\r\n\r\n", "",
+			[]string{chunked},
+		},
+		{"with its length", announcedLength(4) + "body", "body", nil},
+	} {
+		for _, method := range []string{
+			http.MethodGet, http.MethodDelete, http.MethodPost,
+		} {
+			t.Run(method+" "+tc.name, func(t *testing.T) {
+				t.Parallel()
+
+				servewaf := startServeWAF(t, deny)
+				addr, _ := startRProxy(t, servewaf.URL, nowhere, nil)
+
+				conn := dial(t, addr)
+				send(t, conn, method+" /x HTTP/1.1\r\nHost: site.example\r\n"+tc.framing)
+
+				if status := readStatus(t, conn); status != http.StatusForbidden {
+					t.Errorf("status %d, want servewaf's 403", status)
+				}
+
+				asked := servewaf.question(t)
+				if string(asked.body) != tc.body ||
+					!slices.Equal(asked.transferEncoding, tc.encoding) {
+					t.Errorf("the question carries %q, encoded %v, want %q, encoded %v",
+						asked.body, asked.transferEncoding, tc.body, tc.encoding)
+				}
+			})
+		}
+	}
+}
+
+func TestAMethodNotPassedOnIsAnswered405(t *testing.T) {
+	t.Parallel()
+
+	for _, method := range []string{http.MethodTrace, "PROPFIND", "LOCK", "BREW"} {
+		t.Run(method, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, func(http.ResponseWriter, *http.Request) {
+				t.Error("the app was asked")
+			})
+			addr, out := startRProxy(t, servewaf.URL, app.URL, nil)
+
+			got := visit(t, newRequest(t, method, addr, "/", http.NoBody))
+			if got.status != http.StatusMethodNotAllowed ||
+				got.header.Get("Allow") != "GET, HEAD, POST, OPTIONS, PUT, DELETE, PATCH, QUERY" {
+				t.Errorf("answered %+v, want 405 naming the methods passed on", got)
+			}
+
+			line := out.requestLine(t)
+			if line["status"] != float64(http.StatusMethodNotAllowed) ||
+				line["disposition"] != nil || line["default_disposition"] != nil {
+				t.Errorf("line %v, want status 405 and no disposition", line)
+			}
+
+			if len(servewaf.questions) != 0 {
+				t.Error("servewaf was asked")
+			}
+
+			servewaf.wantNoReport(t)
+		})
+	}
+}
+
+func TestEachMethodPassedOnReachesServeWAFAndTheApp(t *testing.T) {
+	t.Parallel()
+
+	for _, method := range []string{
+		http.MethodGet, http.MethodHead, http.MethodPost, http.MethodOptions,
+		http.MethodPut, http.MethodDelete, http.MethodPatch, "QUERY",
+	} {
+		t.Run(method, func(t *testing.T) {
+			t.Parallel()
+
+			reached := make(chan string, 1)
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, func(_ http.ResponseWriter, r *http.Request) {
+				reached <- r.Method
+			})
+			addr, _ := startRProxy(t, servewaf.URL, app.URL, nil)
+
+			got := visit(t, newRequest(t, method, addr, "/", http.NoBody))
+			if got.status != http.StatusOK {
+				t.Errorf("answered %+v, want the app's 200", got)
+			}
+
+			if asked := servewaf.question(t); asked.method != method {
+				t.Errorf("servewaf was asked about %s", asked.method)
+			}
+
+			if atApp := <-reached; atApp != method {
+				t.Errorf("the app was asked %s", atApp)
+			}
+		})
+	}
+}
+
+func TestARequestThatCannotGoToServeWAFAsSentIsRefused(t *testing.T) {
+	t.Parallel()
+
+	const host = "Host: site.example\r\n\r\n"
+
+	for name, request := range map[string]string{
+		"a target of a scheme":    "GET x:admin HTTP/1.1\r\n" + host,
+		"a target of a URL":       "GET http://site.example/ HTTP/1.1\r\n" + host,
+		"a host Go will not send": "GET http://a<b/ HTTP/1.1\r\n" + host,
+		"a target of a host":      "GET site.example:443 HTTP/1.1\r\n" + host,
+		"a target of *":           "GET * HTTP/1.1\r\n" + host,
+		"a target of //":          "GET //x/y HTTP/1.1\r\n" + host,
+		"a target of // escaped":  "GET //x%2Fadmin| HTTP/1.1\r\n" + host,
+		"an empty Host":           "GET / HTTP/1.1\r\nHost:\r\n\r\n",
+		"HTTP/1.0 with no Host":   "GET / HTTP/1.0\r\n\r\n",
+		"a Host with a zone":      "GET / HTTP/1.1\r\nHost: [fe80::1%25eth0]:80\r\n\r\n",
+		"two User-Agents": "GET / HTTP/1.1\r\nUser-Agent: a\r\nUser-Agent: b\r\n" +
+			host,
+	} {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, allow)
+			app := startApp(t, greet)
+			addr, out := startRProxy(t, servewaf.URL, app.URL, nil)
+
+			conn := dial(t, addr)
+			send(t, conn, request)
+
+			if status := readStatus(t, conn); status != http.StatusBadRequest {
+				t.Errorf("status %d, want 400", status)
+			}
+
+			line := out.requestLine(t)
+			if line["status"] != float64(http.StatusBadRequest) ||
+				line["disposition"] != nil || line["default_disposition"] != nil {
+				t.Errorf("line %v, want status 400 and no disposition", line)
+			}
+
+			if len(servewaf.questions) != 0 {
+				t.Error("servewaf was asked")
+			}
+
+			servewaf.wantNoReport(t)
+		})
+	}
+}
+
+func TestAHostWithAPercentSignButNoBracketGoesToServeWAF(t *testing.T) {
+	t.Parallel()
+
+	const host = "a%b"
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, nil)
+
+	conn := dial(t, addr)
+	send(t, conn, "GET / HTTP/1.1\r\nHost: "+host+"\r\n\r\n")
+
+	if status := readStatus(t, conn); status != http.StatusOK {
+		t.Errorf("status %d, want the app's 200", status)
+	}
+
+	if asked := servewaf.question(t); asked.host != host {
+		t.Errorf("asked about the host %q, want %q", asked.host, host)
+	}
+}
+
+func TestATrustedPeerWithNoForwardedProtoIsAskedAboutAsHTTP(t *testing.T) {
+	t.Parallel()
+
+	const want = "http"
+
+	servewaf := startServeWAF(t, allow)
+	app := startApp(t, greet)
+	addr, _ := startRProxy(t, servewaf.URL, app.URL, map[string]string{
+		trustedProxies: localhost + "/32",
+	})
+
+	wantGreeting(t, get(t, addr, "/"))
+
+	asked := servewaf.question(t)
+	if got := asked.header.Get(forwardedProtoHeader); got != want {
+		t.Errorf("the question's %s is %q, want %s", forwardedProtoHeader, got, want)
+	}
+}
+
+func TestAQuestionNamesNoZone(t *testing.T) {
+	t.Parallel()
+
+	for _, tc := range []struct {
+		name, remoteAddr, forwardedFor, client, peer string
+	}{
+		{"from a link-local peer", "[fe80::1%eth0]:1234", "", "fe80::1", "fe80::1"},
+		{
+			"from X-Forwarded-For through a trusted peer", localhost + ":1234",
+			"2001:db8::1%eth0", "2001:db8::1", localhost,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			servewaf := startServeWAF(t, deny)
+			app := startApp(t, greet)
+			_, out, server := startServer(t, servewaf.URL, app.URL, map[string]string{
+				trustedProxies: localhost + "/32",
+			})
+
+			req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", http.NoBody)
+			req.RemoteAddr = tc.remoteAddr
+			req.Header.Set(forwardedForHeader, tc.forwardedFor)
+
+			server.Handler.ServeHTTP(httptest.NewRecorder(), req)
+
+			asked := servewaf.question(t)
+			if asked.header.Get(proxy.ClientHeader) != tc.client ||
+				asked.header.Get(proxy.PeerHeader) != tc.peer {
+				t.Errorf("the question's headers are %v", asked.header)
+			}
+
+			if line := out.requestLine(t); line["client_ip"] != tc.client {
+				t.Errorf("line %v, want client_ip %s", line, tc.client)
+			}
+		})
+	}
+}

internal/rproxy/request.go

@@ -0,0 +1,593 @@
+package rproxy
+
+import (
+	"context"
+	"crypto/rand"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptrace"
+	"net/http/httputil"
+	"net/netip"
+	"net/url"
+	"os"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+	"sneak.berlin/go/smallwebwaf/internal/rules"
+)
+
+// flushAfterEachWrite has ReverseProxy pass on each part of the app's answer as
+// soon as it arrives.
+const flushAfterEachWrite time.Duration = -1
+
+// reportStatusMax is the highest status servewaf takes in a report. An app can
+// send a higher one, up to 999, and the visitor gets that as it is.
+const reportStatusMax = 599
+
+// plainText is the Content-Type of the answers rproxy gives itself.
+const plainText = "text/plain; charset=utf-8"
+
+// request is one request on its way through rproxy, from the moment its headers
+// have been read to its log line and report.
+type request struct {
+	h  *handler
+	in *http.Request
+	// rc sets the deadlines of the connection to the visitor.
+	rc   *http.ResponseController
+	out  *responseWriter
+	body *requestBody // nil for a request without a body
+
+	// id is the fresh id made for the request when it arrived.
+	id          string
+	client      netip.Addr
+	peer        netip.Addr
+	peerTrusted bool
+	start       time.Time
+
+	// disposition is what is done with the request: servewaf's valid disposition,
+	// or SWRP_DEFAULT_DISPOSITION. It is "" when neither gave one.
+	disposition string
+	// answered is true when servewaf gave disposition, which makes a report due,
+	// and byDefault when SWRP_DEFAULT_DISPOSITION did.
+	answered  bool
+	byDefault bool
+	// asn and country are servewaf's X-Client-ASN and X-Client-Country, for the
+	// app.
+	asn, country string
+
+	// upstreamStart is when the request was handed to the app.
+	upstreamStart time.Time
+	// cancel ends the request to the app.
+	cancel context.CancelFunc
+	// refused is the status of the first refusal, from whichever goroutine meets
+	// it: what the visitor is answered if the response has not started yet.
+	refused atomic.Int64
+	// upgraded is the connection to the app once the app has switched protocols,
+	// as for a WebSocket, and nil otherwise.
+	upgraded *upgradedConn
+
+	// mu guards what follows. The time limits run on goroutines of their own,
+	// and the transport starts and stops them, and notes the times below, from
+	// its own; once timersStopped is set, neither time limit acts any more.
+	mu            sync.Mutex
+	timersStopped bool
+	requestTimer  *time.Timer
+	responseTimer *time.Timer
+	// requestEnd is when the app had been sent the whole request, or the answer
+	// started, if that was sooner.
+	requestEnd time.Time
+	// answerStarted is when the first byte of the app's answer arrived.
+	answerStarted time.Time
+}
+
+// newRequest starts handling r: it notes the time, works out the visitor from
+// the TCP peer and X-Forwarded-For, and makes the request's id.
+func (h *handler) newRequest(w http.ResponseWriter, r *http.Request) *request {
+	peer := peerAddress(r)
+	trusted := h.config.TrustedProxies
+
+	rq := &request{
+		h:           h,
+		in:          r,
+		rc:          http.NewResponseController(w),
+		out:         &responseWriter{ResponseWriter: w},
+		id:          rand.Text(),
+		client:      clientAddress(peer, r.Header.Values("X-Forwarded-For"), trusted),
+		peer:        peer,
+		peerTrusted: isInside(peer, trusted),
+		start:       time.Now(),
+	}
+
+	if r.Body != http.NoBody {
+		rq.body = &requestBody{body: limitBody(r.Body, h.config.MaxRequestSize), rq: rq}
+	}
+
+	return rq
+}
+
+// answerHealth answers the health check.
+func (rq *request) answerHealth() {
+	rq.prepareOwnAnswer(rq.requestDeadline())
+	rq.out.Header().Set("Content-Type", plainText)
+	rq.send(http.StatusOK, []byte("ok\n"))
+}
+
+// readFirstPartOfBody reads up to the first 64 KiB of the body into memory,
+// within SWRP_MAX_REQUEST_TIME_MS, and reports whether the request goes on. A
+// visitor too slow to send it is answered 408. For a malformed chunk, or a
+// visitor that went away or stopped its sending side, the connection is closed
+// without an answer. A body over SWRP_MAX_REQUEST_SIZE goes on with the part
+// read, so that servewaf is asked first; passToApp then refuses it.
+func (rq *request) readFirstPartOfBody() bool {
+	if rq.body == nil {
+		return true
+	}
+
+	_ = rq.rc.SetReadDeadline(rq.requestDeadline())
+	held, err := io.ReadAll(io.LimitReader(rq.body, proxy.QuestionBodyMaxBytes))
+	// From here on, the request timer ends a read of the rest of the body, so that
+	// the answer can name the side rproxy was waiting on.
+	_ = rq.rc.SetReadDeadline(time.Time{})
+
+	rq.body.held = held
+
+	switch {
+	case err == nil || rq.refused.Load() != 0:
+		return true
+	case errors.Is(err, os.ErrDeadlineExceeded):
+		rq.answer(http.StatusRequestTimeout)
+
+		return false
+	default:
+		// This panic has Go's server close the connection without an answer, and
+		// log nothing; the deferred finish still runs.
+		panic(http.ErrAbortHandler)
+	}
+}
+
+// setPathAsSent has out, the URL of a request made from in, send the path as the
+// visitor sent it. Go sends the path URL.EscapedPath gives, which, for a path
+// holding a character Go escapes, such as |, is the whole path decoded and
+// escaped again, %2F turned into /. Opaque is sent as it is.
+func setPathAsSent(out *url.URL, in *http.Request) {
+	out.Opaque = rules.Path(in)
+}
+
+// forward passes the request to the app, the part of the body held and then the
+// rest, and the app's answer back. ctx is the request's own context.
+func (rq *request) forward(ctx context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	rq.cancel = cancel
+	ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{
+		WroteRequest:         rq.wroteRequest,
+		GotFirstResponseByte: rq.gotFirstResponseByte,
+	})
+
+	out := rq.in.WithContext(ctx)
+	if rq.body != nil {
+		out.Body = rq.body
+	}
+
+	reverseProxy := &httputil.ReverseProxy{
+		Rewrite:        rq.rewrite,
+		Transport:      rq.h.transport,
+		FlushInterval:  flushAfterEachWrite,
+		ErrorLog:       rq.h.errorLog,
+		ModifyResponse: rq.modifyResponse,
+		ErrorHandler:   rq.answerError,
+	}
+
+	rq.startRequestTimer()
+	rq.upstreamStart = time.Now()
+	reverseProxy.ServeHTTP(rq.out, out)
+}
+
+// rewrite makes the request the app receives: the visitor's request, unchanged,
+// sent to SWRP_UPSTREAM_URL, with the forwarded headers and the request's id set,
+// without any X-Client-ASN or X-Client-Country the visitor sent, and with
+// servewaf's.
+func (rq *request) rewrite(pr *httputil.ProxyRequest) {
+	upstream := rq.h.config.UpstreamURL
+	pr.Out.URL.Scheme = upstream.Scheme
+	pr.Out.URL.Host = upstream.Host
+	// ReverseProxy drops query parameters it cannot parse; the app gets the
+	// query as the visitor sent it.
+	pr.Out.URL.RawQuery = pr.In.URL.RawQuery
+	setPathAsSent(pr.Out.URL, pr.In)
+
+	setForwardedHeaders(pr.In, pr.Out, rq.peer, rq.peerTrusted)
+	pr.Out.Header.Set(requestIDHeader, rq.id)
+	pr.Out.Header.Del(asnHeader)
+	pr.Out.Header.Del(countryHeader)
+
+	if rq.asn != "" {
+		pr.Out.Header.Set(asnHeader, rq.asn)
+	}
+
+	if rq.country != "" {
+		pr.Out.Header.Set(countryHeader, rq.country)
+	}
+}
+
+// modifyResponse looks at the app's answer before ReverseProxy passes it on.
+func (rq *request) modifyResponse(res *http.Response) error {
+	if res.StatusCode == http.StatusSwitchingProtocols {
+		// An upgraded connection, such as a WebSocket, is not cut by the time
+		// limits. ReverseProxy writes this answer straight to the connection it
+		// takes over, not through rq.out, and then copies what passes each way
+		// through res.Body, the connection to the app. It may still refuse the
+		// switch, as when the app's Upgrade is not the visitor's, so the status
+		// is noted only once it has taken over the connection, by rq.out's
+		// Hijack.
+		rq.stopTimers()
+
+		conn, ok := res.Body.(io.ReadWriteCloser)
+		if ok {
+			rq.upgraded = &upgradedConn{ReadWriteCloser: conn}
+			res.Body = rq.upgraded
+		}
+
+		return nil
+	}
+
+	maxBytes := rq.h.config.MaxResponseSize
+	if maxBytes > 0 && res.Body != http.NoBody && res.ContentLength > maxBytes {
+		rq.refuse(http.StatusBadGateway)
+
+		return errResponseTooLarge
+	}
+
+	// Go's server adds a Content-Type it guesses from the body to a response
+	// whose headers have no Content-Type key. The key with no value sends none,
+	// as the app sent none.
+	if res.Header.Get("Content-Type") == "" {
+		rq.out.Header()["Content-Type"] = nil
+	}
+
+	res.Body = &responseBody{body: limitBody(res.Body, maxBytes), rq: rq}
+	rq.setResponseDeadline()
+
+	return nil
+}
+
+// answerError is ReverseProxy's ErrorHandler: the request could not be passed to
+// the app, or the app's answer cannot be passed on.
+func (rq *request) answerError(_ http.ResponseWriter, _ *http.Request, err error) {
+	if rq.upgraded != nil {
+		// The app switched protocols, but the switch was not passed on, and
+		// ReverseProxy leaves the app's connection open when it refuses it.
+		_ = rq.upgraded.Close()
+	}
+
+	status := int(rq.refused.Load())
+	if status == 0 {
+		if rq.in.Context().Err() != nil {
+			// The visitor went away or stopped its sending side. This panic has Go's
+			// server close the connection without an answer, and log nothing; the
+			// deferred finish still runs.
+			panic(http.ErrAbortHandler)
+		}
+
+		rq.h.processLog.Warn("request to the app failed", "error", err.Error())
+
+		status = http.StatusBadGateway
+	}
+
+	rq.answer(status)
+}
+
+// answer sends rproxy's own answer, unless the response has already started.
+func (rq *request) answer(status int) {
+	rq.refused.CompareAndSwap(0, int64(status))
+
+	if rq.out.status != 0 {
+		return // too late to answer: the connection can only be cut
+	}
+
+	// A visitor found too slow is read no more; any other may go on sending until
+	// its time is up, so that Go's server can read the rest of the body and end
+	// the request cleanly.
+	deadline := rq.requestDeadline()
+	if status == http.StatusRequestTimeout {
+		deadline = time.Now()
+	}
+
+	rq.prepareOwnAnswer(deadline)
+
+	header := rq.out.Header()
+	header.Set("Content-Type", plainText)
+	header.Set("X-Content-Type-Options", "nosniff")
+	rq.send(status, []byte(http.StatusText(status)+"\n"))
+}
+
+// send gives the visitor an answer rproxy holds whole: status, the headers set so
+// far and body. Go's server would otherwise send so short an answer only once the
+// handler has returned, after finish has written the log line and sent the report.
+// Without Content-Length, the flush would send the body in chunks, the last of
+// which would still wait for the handler. A HEAD's answer has no body, so any
+// Content-Length it has stays as servewaf gave it.
+func (rq *request) send(status int, body []byte) {
+	if rq.in.Method != http.MethodHead {
+		rq.out.Header().Set("Content-Length", strconv.Itoa(len(body)))
+	}
+
+	rq.out.WriteHeader(status)
+	_, _ = rq.out.Write(body)
+	_ = rq.rc.Flush()
+}
+
+// prepareOwnAnswer readies the connection for an answer rproxy gives itself: the
+// reading of a body that has not arrived whole ends at deadline, and the answer
+// has SWRP_MAX_RESPONSE_TIME_MS from now to reach the visitor, even when the app
+// used that time up first.
+func (rq *request) prepareOwnAnswer(deadline time.Time) {
+	rq.stopReadingBody(deadline)
+	rq.endRequest()
+
+	limit := rq.h.config.MaxResponseTime
+	if limit > 0 {
+		_ = rq.rc.SetWriteDeadline(time.Now().Add(limit))
+	}
+}
+
+// refuse records status, unless an earlier refusal was, and ends the request to
+// the app, if one was made.
+func (rq *request) refuse(status int) {
+	rq.refused.CompareAndSwap(0, int64(status))
+
+	if rq.cancel != nil {
+		rq.cancel()
+	}
+}
+
+// finish ends the request's time limits, writes its log line and, if servewaf
+// answered the question, sends the report. ctx is the request's own context.
+func (rq *request) finish(ctx context.Context) {
+	rq.stopTimers()
+
+	if rq.refused.Load() == 0 {
+		rq.stopReadingBody(rq.requestDeadline())
+	}
+
+	requestEnd := rq.endRequest()
+	end := time.Now()
+	requestBytes, responseBytes := rq.bytesEachWay()
+
+	err := requestlog.WriteRProxyLine(rq.h.requestLog, &requestlog.RProxyLine{
+		Time:               requestlog.FormatTime(rq.start),
+		RequestID:          rq.id,
+		ClientIP:           rq.client.String(),
+		Host:               rq.in.Host,
+		Method:             rq.in.Method,
+		Path:               rules.Path(rq.in),
+		Query:              rq.in.URL.RawQuery,
+		Status:             rq.out.status,
+		RequestBytes:       requestBytes,
+		ResponseBytes:      responseBytes,
+		DurationRequest:    requestlog.Milliseconds(requestEnd.Sub(rq.start)),
+		DurationResponse:   requestlog.Milliseconds(end.Sub(requestEnd)),
+		Disposition:        rq.disposition,
+		DefaultDisposition: rq.byDefault,
+	})
+	if err != nil {
+		rq.h.processLog.Error("writing the request log failed", "error", err.Error())
+	}
+
+	if !rq.answered {
+		return
+	}
+
+	report := &proxy.Report{
+		RequestID:     rq.id,
+		Status:        min(rq.out.status, reportStatusMax),
+		RequestBytes:  requestBytes,
+		ResponseBytes: responseBytes,
+		DurationTotal: requestlog.Milliseconds(end.Sub(rq.start)),
+	}
+	if !rq.upstreamStart.IsZero() {
+		report.DurationUpstreamTotal = requestlog.Milliseconds(end.Sub(rq.upstreamStart))
+	}
+
+	rq.h.sendReport(ctx, report)
+}
+
+// bytesEachWay returns the body bytes of the request and of the answer. For an
+// upgraded connection, such as a WebSocket, which has closed by then, what it
+// carried from the visitor counts with the request's and what it carried from the
+// app with the answer's.
+func (rq *request) bytesEachWay() (int64, int64) {
+	var requestBytes int64
+	if rq.body != nil {
+		requestBytes = rq.body.bytes.Load()
+	}
+
+	responseBytes := rq.out.bytes
+
+	if rq.upgraded != nil {
+		requestBytes += rq.upgraded.toApp.Load()
+		responseBytes += rq.upgraded.fromApp.Load()
+	}
+
+	return requestBytes, responseBytes
+}
+
+// requestDeadline is when the visitor must have sent its whole request, or zero
+// when SWRP_MAX_REQUEST_TIME_MS is off.
+func (rq *request) requestDeadline() time.Time {
+	limit := rq.h.config.MaxRequestTime
+	if limit == 0 {
+		return time.Time{}
+	}
+
+	return rq.start.Add(limit)
+}
+
+// stopReadingBody ends, at deadline, the reading of a visitor's body that has not
+// arrived whole: Go's server then reads no more of it, and closes the connection
+// after the answer.
+func (rq *request) stopReadingBody(deadline time.Time) {
+	if rq.body == nil || rq.body.received.Load() {
+		return
+	}
+
+	_ = rq.rc.SetReadDeadline(deadline)
+}
+
+// startRequestTimer starts SWRP_MAX_REQUEST_TIME_MS, which runs from the
+// request's headers being read until the app has been sent the whole request.
+func (rq *request) startRequestTimer() {
+	deadline := rq.requestDeadline()
+	if deadline.IsZero() {
+		return
+	}
+
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	rq.requestTimer = time.AfterFunc(time.Until(deadline), rq.requestTimedOut)
+}
+
+// requestTimedOut is called when SWRP_MAX_REQUEST_TIME_MS runs out while the
+// request is still on its way to the app. The answer names the side rproxy was
+// waiting on at that moment: 408 when it was waiting for the visitor to send more
+// of its body, 504 when it was waiting for the app to be reached or to take what
+// it had.
+func (rq *request) requestTimedOut() {
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	if rq.timersStopped {
+		return
+	}
+
+	if rq.body == nil || !rq.body.waiting.Load() {
+		// Once the app has answered, the transport, ReverseProxy and Go's server
+		// each close the visitor's body, and closing it reads what is left of a
+		// body under 256 KiB without requestBody.Read, so waiting is false while
+		// that read waits for the visitor. That read ends now, before refuse:
+		// should rproxy still answer itself, answer then sets the deadline it
+		// gives its own answers.
+		if !rq.answerStarted.IsZero() {
+			rq.stopReadingBody(time.Now())
+		}
+
+		rq.refuse(http.StatusGatewayTimeout)
+
+		return
+	}
+
+	rq.refuse(http.StatusRequestTimeout)
+	// The transport gives up on the app only once its Read of the visitor's body
+	// returns, so that Read is ended now. The lock keeps this from reaching the
+	// connection after the request is handled.
+	_ = rq.rc.SetReadDeadline(time.Now())
+}
+
+// gotFirstResponseByte is called once the first byte of the app's answer has
+// arrived.
+func (rq *request) gotFirstResponseByte() {
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	rq.answerStarted = time.Now()
+}
+
+// wroteRequest is called once the app has been sent the whole request:
+// SWRP_MAX_REQUEST_TIME_MS ends and SWRP_MAX_RESPONSE_TIME_MS starts.
+func (rq *request) wroteRequest(info httptrace.WroteRequestInfo) {
+	if info.Err != nil {
+		return // the transport gives up, or tries again
+	}
+
+	requestEnd := rq.endRequest()
+
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	if rq.timersStopped {
+		return
+	}
+
+	stopTimer(rq.requestTimer)
+
+	limit := rq.h.config.MaxResponseTime
+	if limit > 0 {
+		rq.responseTimer = time.AfterFunc(time.Until(requestEnd.Add(limit)),
+			rq.responseTimedOut)
+	}
+}
+
+// responseTimedOut is called when SWRP_MAX_RESPONSE_TIME_MS runs out before the
+// app has sent its whole answer.
+func (rq *request) responseTimedOut() {
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	if !rq.timersStopped {
+		rq.refuse(http.StatusGatewayTimeout)
+	}
+}
+
+// responseReceived is called once the app has sent its whole answer. An app may
+// answer before the visitor has sent its whole body, as with a 404 it sends
+// without reading it. The time limits then run on: until the transport's read of
+// the body returns, Go's server cannot send the answer, and
+// SWRP_MAX_REQUEST_TIME_MS is what ends that read for a visitor that stops
+// sending.
+func (rq *request) responseReceived() {
+	if rq.body != nil && !rq.body.received.Load() {
+		return
+	}
+
+	rq.stopTimers()
+}
+
+// setResponseDeadline sets SWRP_MAX_RESPONSE_TIME_MS on the connection to the
+// visitor: the app's answer must reach the visitor within it of the end of the
+// request.
+func (rq *request) setResponseDeadline() {
+	requestEnd := rq.endRequest()
+
+	limit := rq.h.config.MaxResponseTime
+	if limit > 0 {
+		_ = rq.rc.SetWriteDeadline(requestEnd.Add(limit))
+	}
+}
+
+// endRequest notes now as the end of the request, unless an end is noted, and
+// returns the end noted.
+func (rq *request) endRequest() time.Time {
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	if rq.requestEnd.IsZero() {
+		rq.requestEnd = time.Now()
+	}
+
+	return rq.requestEnd
+}
+
+// stopTimers stops the request's time limits and keeps any from starting later:
+// the app's answer is complete, the connection upgraded, or the request handled.
+func (rq *request) stopTimers() {
+	rq.mu.Lock()
+	defer rq.mu.Unlock()
+
+	rq.timersStopped = true
+	stopTimer(rq.requestTimer)
+	stopTimer(rq.responseTimer)
+}
+
+// stopTimer stops t, which is nil when its time limit is off.
+func stopTimer(t *time.Timer) {
+	if t != nil {
+		t.Stop()
+	}
+}

internal/rproxy/rproxy.go

@@ -0,0 +1,228 @@
+// Package rproxy is the reverse proxy in front of one app that asks servewaf
+// about each request. It passes a request servewaf allows to the app and the
+// app's answer back, within the size and time limits. Any other answer from
+// servewaf goes to the visitor as servewaf gave it. SWRP_DEFAULT_DISPOSITION
+// decides only when servewaf cannot be connected to, does not answer at all, or
+// does not answer within SWRP_TIMEOUT_MS. rproxy itself refuses a request with a
+// method it does not pass on, or one it cannot put to servewaf as the visitor
+// sent it. It keeps nothing between requests and never touches disk. "rproxy" in
+// SPEC.md says what it does.
+package rproxy
+
+import (
+	"context"
+	"io"
+	"log"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	"sneak.berlin/go/smallwebwaf/internal/config"
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+)
+
+// How rproxy keeps connections to the app and to servewaf open between
+// requests.
+const (
+	idleConns       = 100
+	idleConnTimeout = 90 * time.Second
+)
+
+// The limits on a connection from a visitor that have no setting.
+const (
+	// requestHeaderMaxBytes is the largest request line and headers a visitor
+	// may send.
+	requestHeaderMaxBytes = 32 << 10
+	// goServerExtraHeaderBytes is how far past MaxHeaderBytes Go's server reads
+	// before it refuses.
+	goServerExtraHeaderBytes = 4 << 10
+	// idleTimeout is how long a kept-open connection may wait for its next
+	// request.
+	idleTimeout = 120 * time.Second
+)
+
+// adminPrefix starts the path of every request for smallwebwaf itself, which
+// never reaches the app.
+const adminPrefix = "/_smallwebwaf/"
+
+// allowedMethods lists the methods rproxy passes on, as an Allow header gives
+// them. It answers any other method 405 itself.
+const allowedMethods = "GET, HEAD, POST, OPTIONS, PUT, DELETE, PATCH, QUERY"
+
+// Params are what New needs.
+type Params struct {
+	Config *config.RProxy
+	// RequestLog receives one short JSON line per request.
+	RequestLog io.Writer
+	// ProcessLog receives the process's own messages.
+	ProcessLog *slog.Logger
+}
+
+// Server is the server `smallwebwaf rproxy` runs.
+type Server struct {
+	*http.Server
+
+	h *handler
+}
+
+// New returns the server `smallwebwaf rproxy` runs. Go's server itself refuses
+// a request line and headers over requestHeaderMaxBytes, with 431, closes a
+// connection idle for idleTimeout, and applies SWRP_MAX_REQUEST_TIME_MS while
+// the headers arrive. The handler applies the time and size limits from then
+// on.
+func New(params Params) *Server {
+	backend := params.Config.WAFBackend
+	errorLog := slog.NewLogLogger(params.ProcessLog.Handler(), slog.LevelWarn)
+	h := &handler{
+		config:            params.Config,
+		requestLog:        params.RequestLog,
+		processLog:        params.ProcessLog,
+		errorLog:          errorLog,
+		transport:         newTransport(),
+		serveWAFTransport: newServeWAFTransport(),
+		reportURL: (&url.URL{
+			Scheme: backend.Scheme, Host: backend.Host, Path: proxy.ReportPath,
+		}).String(),
+	}
+
+	return &Server{
+		Server: &http.Server{
+			Addr:              params.Config.ListenAddr,
+			Handler:           h,
+			ErrorLog:          errorLog,
+			ReadHeaderTimeout: params.Config.MaxRequestTime,
+			IdleTimeout:       idleTimeout,
+			MaxHeaderBytes:    requestHeaderMaxBytes - goServerExtraHeaderBytes,
+		},
+		h: h,
+	}
+}
+
+// WaitForReports returns once the reports being sent have gone or failed,
+// which SWRP_TIMEOUT_MS bounds. Any report after this call is dropped: that of
+// a WebSocket that closes while rproxy stops, for one.
+func (s *Server) WaitForReports() {
+	s.h.reportsMu.Lock()
+	s.h.reportsStopped = true
+	s.h.reportsMu.Unlock()
+
+	s.h.reports.Wait()
+}
+
+// handler is the proxy. It holds what every request shares; what belongs to
+// one request is in a request.
+type handler struct {
+	config     *config.RProxy
+	requestLog io.Writer
+	processLog *slog.Logger
+	errorLog   *log.Logger
+	// transport carries requests to the app, and serveWAFTransport questions and
+	// reports to servewaf.
+	transport         http.RoundTripper
+	serveWAFTransport http.RoundTripper
+	// reportURL is where the reports go.
+	reportURL string
+	// reports waits for the reports being sent. reportsMu guards starting one
+	// against reportsStopped, which is set once nothing is to wait for another.
+	reports        sync.WaitGroup
+	reportsMu      sync.Mutex
+	reportsStopped bool
+}
+
+// newTransport returns what carries requests to the app. It never goes through
+// a proxy named in the environment, and leaves answers compressed or not as they
+// were sent.
+func newTransport() *http.Transport {
+	return &http.Transport{
+		MaxIdleConns:        idleConns,
+		MaxIdleConnsPerHost: idleConns,
+		IdleConnTimeout:     idleConnTimeout,
+		DisableCompression:  true,
+	}
+}
+
+// newServeWAFTransport returns what carries questions and reports to servewaf.
+// It speaks HTTP/1.1 only. Go's HTTP/2 client refuses to send a visitor's Upgrade
+// header, or a Connection header other than close or keep-alive, and that would
+// read as servewaf giving no answer.
+func newServeWAFTransport() *http.Transport {
+	transport := newTransport()
+	transport.Protocols = new(http.Protocols)
+	transport.Protocols.SetHTTP1(true)
+
+	return transport
+}
+
+// ServeHTTP handles one request: it answers the health check itself, refuses a
+// request with a method it does not pass on or one it cannot put to servewaf as
+// the visitor sent it, and for any other request reads the first part of the
+// body, asks servewaf, and does what the answer says. Once the answer has ended,
+// it writes the request's log line and sends its report.
+func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	rq := h.newRequest(w, r)
+	defer rq.finish(r.Context())
+
+	if r.Method == http.MethodGet && r.URL.Path == proxy.HealthPath {
+		rq.answerHealth()
+
+		return
+	}
+
+	if !slices.Contains(strings.Split(allowedMethods, ", "), r.Method) {
+		rq.out.Header().Set("Allow", allowedMethods)
+		rq.answer(http.StatusMethodNotAllowed)
+
+		return
+	}
+
+	if !canBeAsked(r) {
+		rq.answer(http.StatusBadRequest)
+
+		return
+	}
+
+	if !rq.readFirstPartOfBody() {
+		return
+	}
+
+	answer, err := rq.ask(r.Context())
+	if err != nil {
+		rq.handleNoUsableAnswer(r.Context(), err)
+
+		return
+	}
+
+	rq.noteAnswer(answer)
+
+	switch rq.disposition {
+	case proxy.DispositionAllow:
+		rq.passToApp(r.Context())
+	case proxy.DispositionClose:
+		// This panic has Go's server close the connection without an answer,
+		// and log nothing; the deferred finish still runs.
+		panic(http.ErrAbortHandler)
+	default: // a deny, an error, or an answer without a valid disposition
+		rq.answerWith(answer)
+	}
+}
+
+// passToApp passes the request on to the app, unless a limit or its path keeps
+// it from the app. servewaf answers a path under adminPrefix itself, so such a
+// request gets here only when SWRP_DEFAULT_DISPOSITION allowed it. ctx is the
+// request's own context.
+func (rq *request) passToApp(ctx context.Context) {
+	maxBytes := rq.h.config.MaxRequestSize
+
+	switch {
+	case rq.refused.Load() != 0 || (maxBytes > 0 && rq.in.ContentLength > maxBytes):
+		rq.answer(http.StatusRequestEntityTooLarge)
+	case strings.HasPrefix(rq.in.URL.Path, adminPrefix):
+		rq.answer(http.StatusNotFound)
+	default:
+		rq.forward(ctx)
+	}
+}

internal/rproxy/servewaf.go

@@ -0,0 +1,266 @@
+package rproxy
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"net/http"
+	"strings"
+
+	"sneak.berlin/go/smallwebwaf/internal/proxy"
+)
+
+// answerBodyMaxBytes is the most of the body of servewaf's answer that is read.
+// It is far above any answer's.
+const answerBodyMaxBytes = 1 << 20
+
+// reportAnswerMaxBytes is the most of the answer to a report that is read,
+// which has no body when servewaf takes the report.
+const reportAnswerMaxBytes = 4 << 10
+
+var (
+	errCannotPassOn  = errors.New("the answer cannot be passed to the visitor")
+	errReportRefused = errors.New("the report was answered")
+)
+
+// servewafAnswer is what servewaf answered to a question.
+type servewafAnswer struct {
+	status int
+	header http.Header
+	body   []byte
+}
+
+// noteAnswer notes what servewaf answered. An answer without a valid disposition
+// notes nothing, and no report is due for it: servewaf holds no decision to report
+// on.
+func (rq *request) noteAnswer(answer *servewafAnswer) {
+	disposition := answer.header.Get(proxy.DispositionHeader)
+
+	switch disposition {
+	case proxy.DispositionAllow, proxy.DispositionDeny, proxy.DispositionClose:
+		rq.disposition = disposition
+		rq.answered = true
+		rq.asn = answer.header.Get(asnHeader)
+		rq.country = answer.header.Get(countryHeader)
+	}
+}
+
+// handleNoUsableAnswer deals with a request whose question ended in err, which
+// leaves no answer to act on. ctx is the request's own context.
+func (rq *request) handleNoUsableAnswer(ctx context.Context, err error) {
+	switch {
+	case ctx.Err() != nil:
+		// The visitor went away while servewaf was being asked. servewaf has not
+		// failed, and nothing is noted. This panic has Go's server close the
+		// connection without an answer, and log nothing; the deferred finish still
+		// runs.
+		panic(http.ErrAbortHandler)
+	case errors.Is(err, errCannotPassOn):
+		rq.h.processLog.Warn("servewaf's answer is dropped", "error", err.Error())
+		rq.answer(http.StatusBadGateway)
+	default:
+		// servewaf cannot be connected to, does not answer at all, or does not answer
+		// within SWRP_TIMEOUT_MS. A request that keeps servewaf busy past
+		// SWRP_TIMEOUT_MS, like any request while servewaf is overloaded, gets
+		// SWRP_DEFAULT_DISPOSITION here. deny closes this, at the cost of answering 503
+		// whenever servewaf is slow. Making such requests hard to send is
+		// https://git.eeqj.de/sneak/smallwebwaf/issues/182.
+		rq.h.processLog.Warn("servewaf gave no answer", "error", err.Error())
+
+		rq.disposition = rq.h.config.DefaultDisposition
+		rq.byDefault = true
+
+		if rq.disposition == proxy.DispositionAllow {
+			rq.passToApp(ctx)
+		} else {
+			rq.answer(http.StatusServiceUnavailable)
+		}
+	}
+}
+
+// ask sends servewaf the question, and returns its answer, which must have come
+// whole within SWRP_TIMEOUT_MS. It ends when ctx does, as when the visitor goes
+// away.
+func (rq *request) ask(ctx context.Context) (*servewafAnswer, error) {
+	ctx, cancel := context.WithTimeout(ctx, rq.h.config.Timeout)
+	defer cancel()
+
+	question, err := rq.newQuestion(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := rq.h.serveWAFTransport.RoundTrip(question)
+	if err != nil {
+		return nil, fmt.Errorf("send the question: %w", err)
+	}
+
+	defer func() {
+		_ = res.Body.Close()
+	}()
+
+	// The body of a 101 is the connection itself, and the time limit does not end
+	// a read of that. Go's client also takes a status below 100, which Go's server
+	// panics when asked to send.
+	if res.StatusCode < http.StatusOK {
+		return nil, fmt.Errorf("%w: its status is %d", errCannotPassOn, res.StatusCode)
+	}
+
+	body, err := io.ReadAll(io.LimitReader(res.Body, answerBodyMaxBytes+1))
+	if err != nil {
+		return nil, fmt.Errorf("read the answer: %w", err)
+	}
+
+	if len(body) > answerBodyMaxBytes {
+		return nil, fmt.Errorf("%w: its body is over 1 MiB", errCannotPassOn)
+	}
+
+	return &servewafAnswer{res.StatusCode, res.Header, body}, nil
+}
+
+// canBeAsked reports whether Go's client can send servewaf r as the visitor sent
+// it. It cannot send these as they came:
+//   - a target that does not start with /, or that starts with //;
+//   - an empty Host, for which it sends servewaf's own address;
+//   - a Host holding an IPv6 zone, which it removes;
+//   - more than one User-Agent, of which it sends the first;
+//   - an empty User-Agent, which it leaves out, the one case that still goes to
+//     servewaf.
+func canBeAsked(r *http.Request) bool {
+	target, host := r.RequestURI, r.Host
+	hostHasZone := strings.HasPrefix(host, "[") && strings.Contains(host, "%")
+
+	return strings.HasPrefix(target, "/") && !strings.HasPrefix(target, "//") &&
+		host != "" && !hostHasZone && len(r.Header.Values("User-Agent")) <= 1
+}
+
+// newQuestion returns the question about the request, as "The question" in
+// SPEC.md gives it: the visitor's request as the visitor sent it, with the part
+// of the body it holds and rproxy's headers.
+func (rq *request) newQuestion(ctx context.Context) (*http.Request, error) {
+	in := rq.in
+
+	var held []byte
+	if rq.body != nil {
+		held = rq.body.held
+	}
+
+	question, err := http.NewRequestWithContext( //nolint:gosec // a setting's URL
+		ctx, in.Method, rq.h.config.WAFBackend.String(), bytes.NewReader(held))
+	if err != nil {
+		return nil, fmt.Errorf("make the question: %w", err)
+	}
+
+	backend := question.URL
+	target := *in.URL
+	target.Scheme = backend.Scheme
+	target.Host = backend.Host
+	setPathAsSent(&target, in)
+	question.URL = &target
+	question.Host = in.Host
+
+	// The length NewRequestWithContext set is that of what is held. This line makes
+	// a body whose length the visitor did not announce go chunked, as "The question"
+	// in SPEC.md requires.
+	question.TransferEncoding = in.TransferEncoding
+
+	question.Header = in.Header.Clone()
+	question.Header.Set(proxy.ClientHeader, rq.client.String())
+	question.Header.Set(proxy.PeerHeader, rq.peer.String())
+	question.Header.Del(proxy.ContentLengthHeader)
+
+	announced := in.Header.Get("Content-Length")
+	if announced != "" {
+		question.Header.Set(proxy.ContentLengthHeader, announced)
+	}
+
+	question.Header.Set(requestIDHeader, rq.id)
+	question.Header.Set("X-Forwarded-Proto", scheme(in, rq.peerTrusted))
+
+	// Go's client adds a User-Agent of its own to a request without the key.
+	_, sent := question.Header["User-Agent"]
+	if !sent {
+		question.Header["User-Agent"] = nil
+	}
+
+	return question, nil
+}
+
+// answerWith gives the visitor servewaf's answer, less its disposition.
+func (rq *request) answerWith(answer *servewafAnswer) {
+	rq.prepareOwnAnswer(rq.requestDeadline())
+
+	header := rq.out.Header()
+	maps.Copy(header, answer.header)
+	header.Del(proxy.DispositionHeader)
+
+	// Go's server adds a Content-Type it guesses from the body to a response whose
+	// headers have no Content-Type key. The key with no value sends none.
+	if header.Get("Content-Type") == "" {
+		header["Content-Type"] = nil
+	}
+
+	rq.send(answer.status, answer.body)
+}
+
+// sendReport sends servewaf report on a goroutine of its own, so that the
+// visitor does not wait for it. The report goes on after ctx, the request's own
+// context, has ended.
+func (h *handler) sendReport(ctx context.Context, report *proxy.Report) {
+	h.reportsMu.Lock()
+	defer h.reportsMu.Unlock()
+
+	if h.reportsStopped {
+		return
+	}
+
+	h.reports.Go(func() {
+		err := h.postReport(context.WithoutCancel(ctx), report)
+		if err != nil {
+			h.processLog.Warn("a report is dropped", "error", err.Error())
+		}
+	})
+}
+
+// postReport posts report to servewaf, and waits for its answer for up to
+// SWRP_TIMEOUT_MS.
+func (h *handler) postReport(ctx context.Context, report *proxy.Report) error {
+	body, err := json.Marshal(report)
+	if err != nil {
+		return fmt.Errorf("encode the report: %w", err)
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, h.config.Timeout)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext( //nolint:gosec // a setting's URL
+		ctx, http.MethodPost, h.reportURL, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("make the request: %w", err)
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+
+	res, err := h.serveWAFTransport.RoundTrip(req)
+	if err != nil {
+		return fmt.Errorf("send the report: %w", err)
+	}
+
+	// The body of a 101 is the connection itself, and the time limit does not end
+	// a read of that.
+	if res.StatusCode >= http.StatusOK {
+		_, _ = io.Copy(io.Discard, io.LimitReader(res.Body, reportAnswerMaxBytes))
+	}
+
+	_ = res.Body.Close()
+
+	if res.StatusCode != http.StatusNoContent {
+		return fmt.Errorf("%w %s", errReportRefused, res.Status)
+	}
+
+	return nil
+}