smallwebwaf/internal/proxy/servewaf.go

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

// 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"`
}

// 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)
}

// answersQuestions reports whether h is servewaf, which alone holds requests
// for their reports.
func (h *handler) answersQuestions() bool {
	return h.awaitingReport != nil
}

// refusedWhateverTheAnswer reports whether the checks will refuse the request
// 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. rproxy
// lets a request through when servewaf answers too late, so such a request
// does not wait for GeoJS.
func (rq *request) refusedWhateverTheAnswer(now time.Time) bool {
	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 that is not empty being one. 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.
func (rq *request) overHeaderLimits() bool {
	if !rq.h.answersQuestions() {
		return false
	}

	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.TrimSpace(part) != "" {
				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, 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()
}