servewaf answers rproxy's question about each request and takes its report (closes #170)

`smallwebwaf servewaf` runs today's checks on the visitor's request each question carries, answers rproxy, and passes nothing to an app. rproxy reports every request servewaf answered. A report counts an allowed request's bytes and writes servewaf's second log line. The table holds a copy of what a report needs, never the request, so a report cannot reach a request being handled. Judgement call: a question's body is read before any check; over 64 KiB gets 400. Judgement call: a report's bytes and durations have ceilings, so they cannot overflow. Judgement call: a `close` disposition for SWWAF_BAN_RESPONSE=close. Model: opus-5-5
2026-11-05 00:32:08 +01:00
parent 718a2c94b2
commit 2eeea136e8
28 changed files with 2919 additions and 143 deletions
@@ -0,0 +1,464 @@
+package proxy
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/netip"
+	"strconv"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	lru "github.com/hashicorp/golang-lru/v2"
+
+	"sneak.berlin/go/smallwebwaf/internal/requestlog"
+)
+
+// The headers of rproxy's question and of servewaf's answer, as
+// "servewaf" in SPEC.md gives them.
+const (
+	// ClientHeader marks a request as rproxy's question, and gives the
+	// visitor's address.
+	ClientHeader = "X-Smallwebwaf-Client"
+	// PeerHeader gives the address of rproxy's own TCP peer.
+	PeerHeader = "X-Smallwebwaf-Peer"
+	// ContentLengthHeader gives the body length the visitor announced.
+	ContentLengthHeader = "X-Smallwebwaf-Content-Length"
+	// DispositionHeader gives servewaf's answer.
+	DispositionHeader = "X-Smallwebwaf-Disposition"
+)
+
+// The values of DispositionHeader.
+const (
+	// DispositionAllow tells rproxy to pass the request to the app.
+	DispositionAllow = "allow"
+	// DispositionDeny tells rproxy to give the visitor the answer's status,
+	// headers and body.
+	DispositionDeny = "deny"
+	// DispositionClose tells rproxy to close the visitor's connection
+	// without an answer, which SWWAF_BAN_RESPONSE asks for when it is close.
+	// servewaf closing the question's connection instead would read as
+	// servewaf being down, and rproxy would then let the request through.
+	DispositionClose = "close"
+)
+
+// ReportPath is where rproxy sends the report of each request servewaf
+// answered.
+const ReportPath = "/_smallwebwaf/report"
+
+// QuestionBodyMaxBytes is the most of a visitor's body a question carries.
+const QuestionBodyMaxBytes = 64 << 10
+
+const (
+	// maxAwaitingReport is the most answered requests held until their
+	// reports come. Past it, the request answered longest ago is dropped.
+	maxAwaitingReport = 10000
+	// maxRequestIDBytes is the longest id a question may give.
+	maxRequestIDBytes = 128
+	// reportBodyMaxBytes is the most of a report's body that is read. It is
+	// far above any report's.
+	reportBodyMaxBytes = 1 << 20
+	// maxHTTPStatus is the highest status a report may give.
+	maxHTTPStatus = 599
+	// maxReportedBytes is the most bytes a report may give for each way, 1
+	// PiB, so that adding them up cannot overflow.
+	maxReportedBytes = 1 << 50
+	// maxReportedMilliseconds is the longest duration a report may give,
+	// about 31 years, so that it fits a time.Duration.
+	maxReportedMilliseconds = 1e12
+	// serveWAFTimeout bounds how long a request to servewaf takes to arrive,
+	// and its answer to leave, and serveWAFIdleTimeout how long a kept-open
+	// connection waits for its next request.
+	serveWAFTimeout     = 60 * time.Second
+	serveWAFIdleTimeout = 120 * time.Second
+)
+
+var (
+	errReportFieldMissing = errors.New("a field is missing")
+	errReportValue        = errors.New("a value is out of range")
+)
+
+// Report is the JSON object rproxy posts to ReportPath once the visitor's
+// answer has ended, its fields named as the request log's of the same
+// meaning. RequestID is the question's id. Status is what the visitor got, 0
+// if nothing. The durations are in milliseconds.
+//
+//nolint:tagliatelle // docs/request-log.md names the fields in snake_case
+type Report struct {
+	RequestID             string  `json:"request_id"`
+	Status                int     `json:"status"`
+	RequestBytes          int64   `json:"request_bytes"`
+	ResponseBytes         int64   `json:"response_bytes"`
+	DurationTotal         float64 `json:"duration_total"`
+	DurationUpstreamTotal float64 `json:"duration_upstream_total"`
+}
+
+// NewServeWAF returns the server `smallwebwaf servewaf` runs. It answers
+// rproxy's questions with the checks New runs, takes rproxy's reports and
+// answers its own endpoints. It passes nothing to an app. rproxy applies
+// the size and time limits to the visitor, so their settings do not apply
+// here. Go's default limit on a request's line and headers, 1 MiB, stays:
+// it is far above rproxy's, and a question refused for its size would have
+// rproxy let the request through.
+func NewServeWAF(params Params) *Server {
+	cfg := *params.Config
+	cfg.RequestMaxBytes = 0
+	cfg.ClientRequestTimeout = serveWAFTimeout
+	// The server's WriteTimeout does this. It sets a new deadline for each
+	// request, where a deadline a handler sets stays on a kept-open
+	// connection, and would run out under a later request's answer.
+	cfg.ClientResponseTimeout = 0
+	// A question's body of QuestionBodyMaxBytes may be cut short, and the
+	// Core Rule Set does not read a JSON or XML body over its limit.
+	cfg.WAFBodyLimit = min(cfg.WAFBodyLimit, QuestionBodyMaxBytes-1)
+	params.Config = &cfg
+
+	h := newHandler(params)
+
+	awaitingReport, err := lru.New[string, *decision](maxAwaitingReport)
+	if err != nil {
+		panic(err) // New fails only for a size below one
+	}
+
+	h.awaitingReport = awaitingReport
+
+	return h.server(&http.Server{
+		Handler:      http.HandlerFunc(h.serveWAF),
+		ReadTimeout:  serveWAFTimeout,
+		WriteTimeout: serveWAFTimeout,
+		IdleTimeout:  serveWAFIdleTimeout,
+	})
+}
+
+// serveWAF handles one request to servewaf: a request carrying
+// ClientHeader is rproxy's question, and any other is for servewaf itself.
+func (h *handler) serveWAF(w http.ResponseWriter, r *http.Request) {
+	if len(r.Header.Values(ClientHeader)) > 0 {
+		h.answerQuestion(w, r)
+
+		return
+	}
+
+	h.answerOwnEndpoint(w, r)
+}
+
+// answerQuestion runs the checks on the visitor's request a question
+// carries, and answers rproxy what to do with it. A denied request's counting
+// is finished here, and an allowed one's by its report.
+func (h *handler) answerQuestion(w http.ResponseWriter, r *http.Request) {
+	client, body, ok := h.readQuestion(w, r)
+	if !ok {
+		return
+	}
+
+	// One that cannot be read leaves the log line's peer_ip invalid.
+	peer, _ := netip.ParseAddr(r.Header.Get(PeerHeader))
+
+	rq := h.startRequest(w, asTheVisitorSentIt(r, body), client.Unmap(), peer.Unmap(),
+		true)
+	// The last deferred runs first, so that what is held has the finished log
+	// line. Go's server sends the answer only once this function has
+	// returned, so the report cannot come before it is held.
+	defer rq.holdForReport()
+	defer rq.finish()
+
+	rq.site = r.Host
+	rq.line.ServeWAFLine = requestlog.ServeWAFLineDecision
+	refused := rq.check(r.Context())
+	rq.checked = time.Now()
+	underOwnPrefix := strings.HasPrefix(r.URL.Path, adminPrefix)
+
+	if refused == nil && !underOwnPrefix {
+		// In observe mode, what enforce mode would have refused counts here.
+		defer rq.countRefusal()
+
+		rq.allow()
+
+		return
+	}
+
+	// The last deferred runs first, as in ServeHTTP.
+	defer rq.addToHistory()
+	defer rq.countAnomalies()
+	defer rq.countRefusal()
+
+	if refused != nil {
+		rq.deny(*refused)
+
+		return
+	}
+
+	// The app never sees this prefix. servewaf's own endpoints are at its
+	// own address, so this answers as a path with no endpoint does.
+	rq.line.Action = requestlog.ActionAdmin
+	rq.out.Header().Set(DispositionHeader, DispositionDeny)
+	http.Error(rq.out, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+}
+
+// readQuestion returns the visitor's address and the body a question carries,
+// and whether servewaf takes the question. It answers a question it does not
+// take with 400 and no disposition, which rproxy treats as it treats no
+// answer. The id of a question it takes is held from here on, without a
+// decision yet, so that another question with that id is refused while this
+// one is being handled.
+//
+// The body is read before any check, since Go's server does not send an answer
+// until it has read the body. A question whose body stops arriving gets no
+// answer, and is not looked at.
+func (h *handler) readQuestion(
+	w http.ResponseWriter, r *http.Request,
+) (netip.Addr, []byte, bool) {
+	client, err := netip.ParseAddr(r.Header.Get(ClientHeader))
+	if err != nil || client.Zone() != "" {
+		h.refuseQuestion(w, ClientHeader+" is not an address without a zone")
+
+		return netip.Addr{}, nil, false
+	}
+
+	id := r.Header.Get(requestIDHeader)
+	if id == "" || len(id) > maxRequestIDBytes || !utf8.ValidString(id) {
+		h.refuseQuestion(w, requestIDHeader+" is not a valid id")
+
+		return netip.Addr{}, nil, false
+	}
+
+	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, QuestionBodyMaxBytes))
+
+	if _, tooLarge := errors.AsType[*http.MaxBytesError](err); tooLarge {
+		h.refuseQuestion(w, "the body is over 64 KiB")
+
+		return netip.Addr{}, nil, false
+	}
+
+	if err != nil {
+		h.processLog.Warn("a question's body stopped arriving", "error", err.Error())
+		// This panic has Go's server close the connection without an answer,
+		// and log nothing.
+		panic(http.ErrAbortHandler)
+	}
+
+	held, _ := h.awaitingReport.ContainsOrAdd(id, nil)
+	if held {
+		h.refuseQuestion(w, requestIDHeader+" is the id of a request servewaf holds")
+
+		return netip.Addr{}, nil, false
+	}
+
+	return client, body, true
+}
+
+// refuseQuestion answers a question servewaf does not take, for reason.
+func (h *handler) refuseQuestion(w http.ResponseWriter, reason string) {
+	h.processLog.Warn("a question is refused", "reason", reason)
+	http.Error(w, reason, http.StatusBadRequest)
+}
+
+// asTheVisitorSentIt returns a copy of r, a question, as the checks and the
+// log line are to see it: without the headers rproxy adds for servewaf, with
+// body, which has been read of r, and with the length the visitor announced,
+// as -1 when it announced none. The question's own length covers only the part
+// of the body it carries.
+func asTheVisitorSentIt(r *http.Request, body []byte) *http.Request {
+	sent := r.Clone(r.Context())
+	announced := sent.Header.Get(ContentLengthHeader)
+
+	sent.Header.Del(ClientHeader)
+	sent.Header.Del(PeerHeader)
+	sent.Header.Del(ContentLengthHeader)
+	sent.Header.Del("Content-Length")
+
+	sent.ContentLength = -1
+
+	length, err := strconv.ParseInt(announced, 10, 64)
+	if err == nil && length >= 0 {
+		sent.ContentLength = length
+		sent.Header.Set("Content-Length", announced)
+	}
+
+	sent.Body = http.NoBody
+	if len(body) > 0 {
+		sent.Body = io.NopCloser(bytes.NewReader(body))
+	}
+
+	return sent
+}
+
+// allow answers a question with DispositionAllow and, while
+// SWWAF_ADD_LOOKUP_HEADERS is set, the client's AS number and country.
+func (rq *request) allow() {
+	rq.allowed = true
+
+	header := rq.out.Header()
+	header.Set(DispositionHeader, DispositionAllow)
+
+	if rq.h.config.AddLookupHeaders {
+		setLookupHeaders(header, rq.line.ASN, rq.line.Country)
+	}
+
+	rq.out.WriteHeader(http.StatusOK)
+}
+
+// holdForReport holds a copy of what the report's steps need, by the request's
+// id. A report only ever reaches that copy, never the request being handled.
+func (rq *request) holdForReport() {
+	held := rq.decision
+	rq.h.awaitingReport.Add(rq.line.RequestID, &held)
+}
+
+// deny answers a question with what the visitor is to get for r.
+func (rq *request) deny(r refusal) {
+	if r.status == 0 {
+		// SWWAF_BAN_RESPONSE is close. Written past rq.out, since the log line's
+		// status stays 0: the visitor is sent nothing.
+		rq.refused.CompareAndSwap(nil, &r)
+		rq.out.Header().Set(DispositionHeader, DispositionClose)
+		rq.out.ResponseWriter.WriteHeader(http.StatusOK)
+
+		return
+	}
+
+	rq.out.Header().Set(DispositionHeader, DispositionDeny)
+	rq.answer(r)
+}
+
+// answerOwnEndpoint answers a request to servewaf's own address. No check
+// runs on it: the address is on the private network, where the peers are
+// rproxy and the admin's tools, not visitors.
+func (h *handler) answerOwnEndpoint(w http.ResponseWriter, r *http.Request) {
+	if r.Method == http.MethodPost && r.URL.Path == ReportPath {
+		h.takeReport(w, r)
+
+		return
+	}
+
+	peer := peerAddress(r)
+
+	rq := h.startRequest(w, r, peer, peer, false)
+	defer rq.finish()
+
+	if r.Method == http.MethodGet && r.URL.Path == HealthPath {
+		rq.answerHealth()
+
+		return
+	}
+
+	rq.answerAdmin()
+}
+
+// takeReport answers POST ReportPath: it finishes the request the report
+// names, if it holds that request's decision, and ignores the report
+// otherwise.
+func (h *handler) takeReport(w http.ResponseWriter, r *http.Request) {
+	report, err := readReport(w, r)
+	if err != nil {
+		h.processLog.Warn("a report is refused", "error", err.Error())
+		http.Error(w, "the body is not a report", http.StatusBadRequest)
+
+		return
+	}
+
+	held, found := h.awaitingReport.Peek(report.RequestID)
+	// Remove reports true to one report only, of several that name the id.
+	if found && held != nil && h.awaitingReport.Remove(report.RequestID) {
+		h.finishWith(held, report)
+	}
+
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// readReport reads a report from r's body, which must be one JSON object with
+// all of Report's fields, in the ranges SPEC.md gives.
+func readReport(w http.ResponseWriter, r *http.Request) (Report, error) {
+	var report Report
+
+	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, reportBodyMaxBytes))
+	if err != nil {
+		return report, fmt.Errorf("read the body: %w", err)
+	}
+
+	var fields map[string]json.RawMessage
+
+	err = json.Unmarshal(body, &fields)
+	if err != nil {
+		return report, fmt.Errorf("decode the body: %w", err)
+	}
+
+	for _, name := range []string{
+		"request_id", "status", "request_bytes", "response_bytes",
+		"duration_total", "duration_upstream_total",
+	} {
+		value, there := fields[name]
+		if !there || string(value) == "null" {
+			return report, fmt.Errorf("%w: %s", errReportFieldMissing, name)
+		}
+	}
+
+	err = json.Unmarshal(body, &report)
+	if err != nil {
+		return report, fmt.Errorf("decode the body: %w", err)
+	}
+
+	return report, report.checkRanges()
+}
+
+// checkRanges returns an error for a value of the report that cannot be true.
+func (report *Report) checkRanges() error {
+	switch {
+	case report.Status != 0 &&
+		(report.Status < http.StatusContinue || report.Status > maxHTTPStatus):
+		return fmt.Errorf("%w: status", errReportValue)
+	case report.RequestBytes < 0 || report.RequestBytes > maxReportedBytes:
+		return fmt.Errorf("%w: request_bytes", errReportValue)
+	case report.ResponseBytes < 0 || report.ResponseBytes > maxReportedBytes:
+		return fmt.Errorf("%w: response_bytes", errReportValue)
+	case report.DurationTotal < 0 || report.DurationTotal > maxReportedMilliseconds:
+		return fmt.Errorf("%w: duration_total", errReportValue)
+	case report.DurationUpstreamTotal < 0 ||
+		report.DurationUpstreamTotal > maxReportedMilliseconds:
+		return fmt.Errorf("%w: duration_upstream_total", errReportValue)
+	}
+
+	return nil
+}
+
+// finishWith finishes a request with what its report says, and writes the
+// request's last log line, which holds its decision's fields and the report's.
+// For an allowed request it first takes the steps ServeHTTP takes once a
+// request passed to the app has ended, with the status and bytes of the report
+// in place of those it would have seen. A denied request's counting was
+// finished at its decision.
+func (h *handler) finishWith(held *decision, report Report) {
+	rq := &request{
+		decision: *held,
+		h:        h,
+		out:      &responseWriter{status: report.Status, bytes: report.ResponseBytes},
+		body:     &requestBody{},
+	}
+	rq.body.bytes.Store(report.RequestBytes)
+
+	rq.line.ServeWAFLine = requestlog.ServeWAFLineReport
+	rq.line.Status = report.Status
+	rq.line.RequestBytes = report.RequestBytes
+	rq.line.ResponseBytes = report.ResponseBytes
+	rq.line.DurationTotal = report.DurationTotal
+
+	if rq.allowed {
+		rq.line.DurationUpstreamTotal = &report.DurationUpstreamTotal
+		// addToHistory goes by this for whether the request reached the app.
+		rq.upstreamStart = time.Now()
+
+		rq.countBytes()
+		rq.countAnomalies()
+		rq.addToHistory()
+
+		rq.h.metrics.RequestReported(&rq.line,
+			time.Duration(report.DurationTotal*float64(time.Millisecond)),
+			time.Duration(report.DurationUpstreamTotal*float64(time.Millisecond)))
+	}
+
+	rq.writeLine()
+}