smallwebwaf/internal/rproxy/rproxy.go

// Package rproxy is the reverse proxy in front of one app that asks discriminator
// about each request. It passes a request discriminator allows to the app and the
// app's answer back, within the size and time limits. Any other answer from
// discriminator goes to the visitor as discriminator gave it. SWRP_DEFAULT_DISPOSITION
// decides only when discriminator 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 discriminator 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/discriminator"
)

// How rproxy keeps connections to the app and to discriminator 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 sww itself, which
// never reaches the app.
const adminPrefix = "/_sww/"

// 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 `sww rproxy` runs.
type Server struct {
	*http.Server

	h *handler
}

// New returns the server `sww 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(),
		discriminatorTransport: newDiscriminatorTransport(),
		reportURL: (&url.URL{
			Scheme: backend.Scheme, Host: backend.Host, Path: discriminator.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 discriminatorTransport questions and
	// reports to discriminator.
	transport              http.RoundTripper
	discriminatorTransport 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,
	}
}

// newDiscriminatorTransport returns what carries questions and reports to
// discriminator. 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 discriminator giving no answer.
func newDiscriminatorTransport() *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 discriminator as
// the visitor sent it, and for any other request reads the first part of the
// body, asks discriminator, 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 == discriminator.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 discriminator.DispositionAllow:
		rq.passToApp(r.Context())
	case discriminator.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. discriminator 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)
	}
}