Rename internal/proxy to internal/servewaf (closes #173)

The package passes nothing on since the one-process mode went: it answers rproxy's questions, takes its reports and answers servewaf's own endpoints. It is now named after the command it serves, with `git mv`, and its constructor is `servewaf.New`. `SPEC.md` and the other documents follow. No change of behaviour.

Model: opus-5-5

internal/servewaf/servewaf.go

@@ -0,0 +1,509 @@
+package servewaf
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/netip"
+	"strconv"
+	"strings"
+	"time"
+	"unicode/utf8"
+
+	"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
+
+// The header limits: the most cookies and header fields a request may have
+// when the checks reach the Core Rule Set. Real clients never reach them.
+const (
+	maxCookies      = 50
+	maxHeaderFields = 100
+)
+
+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"`
+}
+
+// New returns the server `smallwebwaf servewaf` runs. It answers
+// rproxy's questions, 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. 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 New(params Params) *Server {
+	h := newHandler(params)
+
+	return &Server{
+		Server: &http.Server{
+			Addr:         params.Config.ListenAddr,
+			Handler:      h,
+			ErrorLog:     slog.NewLogLogger(params.ProcessLog.Handler(), slog.LevelWarn),
+			ReadTimeout:  serveWAFTimeout,
+			WriteTimeout: serveWAFTimeout,
+			IdleTimeout:  serveWAFIdleTimeout,
+		},
+		Ledger:     h.ledger,
+		Limiter:    h.limiter,
+		GeoJS:      h.geojs,
+		Anomalies:  h.anomalies,
+		LookupFile: h.lookupFile,
+		Lists:      h.lists,
+		DNSBL:      h.dnsbl,
+		AbuseIPDB:  h.abuseIPDB,
+		Metrics:    h.metrics,
+	}
+}
+
+// ServeHTTP handles one request to servewaf: a request carrying ClientHeader is
+// rproxy's question, and any other is for servewaf itself.
+func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if len(r.Header.Values(ClientHeader)) > 0 {
+		h.answerQuestion(w, r)
+
+		return
+	}
+
+	h.answerOwnEndpoint(w, r)
+}
+
+// refusedWhateverTheAnswer reports whether the request is one that does not wait
+// for GeoJS, since the checks will refuse it at now whatever GeoJS answers about
+// its client: it is over the header limits, a blocklist lists its client while
+// SWWAF_BLOCKLIST_ACTION is deny, the CrowdSec decision list lists it, or it is
+// over a whole rate limit. In observe mode, which refuses nothing, none is.
+// rproxy lets a request through when servewaf answers too late.
+func (rq *request) refusedWhateverTheAnswer(now time.Time) bool {
+	if rq.h.config.Observe {
+		return false
+	}
+
+	lists := rq.h.lists
+	_, decided := lists.CrowdSecDecision(rq.client, now)
+
+	return decided || rq.overHeaderLimits() ||
+		(rq.h.config.BlocklistAction == deny && len(lists.ListedBy(rq.client)) > 0) ||
+		(rq.countedByRateLimits() &&
+			rq.h.limiter.OverARateLimit(rq.h.clientGroup(rq.client), now))
+}
+
+// overHeaderLimits reports whether servewaf refuses the request for its
+// headers: it has more than maxHeaderFields fields, each value of a header sent
+// twice being one, or more than maxCookies cookies, each part of a Cookie header
+// between semicolons being one unless it is empty or only spaces and tabs. The
+// Core Rule Set runs each of its rules on each cookie, so that thousands of small
+// cookies keep it busy for over a second. It trims only spaces and control
+// characters, of which a header value can hold only tabs, so a part of Unicode
+// spaces is a cookie to it.
+func (rq *request) overHeaderLimits() bool {
+	fields, cookies := 0, 0
+
+	for _, values := range rq.in.Header {
+		fields += len(values)
+	}
+
+	for _, value := range rq.in.Header.Values("Cookie") {
+		for part := range strings.SplitSeq(value, ";") {
+			if strings.Trim(part, " \t") != "" {
+				cookies++
+			}
+		}
+	}
+
+	return fields > maxHeaderFields || cookies > maxCookies
+}
+
+// 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: countRefusal before addToHistory, so that a
+	// broken error burst is in the client's history.
+	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) {
+	rq.refused = &r
+
+	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.out.Header().Set(DispositionHeader, DispositionClose)
+		rq.out.ResponseWriter.WriteHeader(http.StatusOK)
+
+		return
+	}
+
+	rq.out.Header().Set(DispositionHeader, DispositionDeny)
+	http.Error(rq.out, http.StatusText(r.status), r.status)
+}
+
+// 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. A
+	// question still being handled is held as nil.
+	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 counts the request's bytes, and adds it to its
+// client's history, the anomaly counters and the metrics, with the status and
+// bytes the report gives. 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},
+		requestBytes: 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
+
+		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()
+}