smallwebwaf/internal/servewaf/servewaf.go

package servewaf

import (
	"bytes"
	"context"
	"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

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: the heuristic crowdsec, header_limits, blocklists or rate_limits
// will refuse it, as a function in that heuristic's file says. 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
	}

	_, decided := rq.crowdSecDecisionInForce(now)

	return decided || overHeaderLimits(rq.headerCounts()) ||
		rq.refusedByBlocklists(rq.h.lists.ListedBy(rq.client)) ||
		rq.overAWholeRateLimit(now)
}

// 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.runAfterAnswer(r.Context())

		rq.allow()

		return
	}

	// The last deferred runs first: runAfterAnswer before addToHistory, so that
	// a broken refusal limit is in the client's history.
	defer rq.addToHistory()
	defer rq.countAnomalies()
	defer rq.runAfterAnswer(r.Context())

	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(r.Context(), 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 runs the heuristics of onReport, and adds the
// request 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. ctx is the report's own context.
func (h *handler) finishWith(ctx context.Context, 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.runBanningHeuristics(ctx, onReport())
		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()
}