smallwebwaf/internal/proxy/proxy.go

// Package proxy passes each request to the app and the app's answer back,
// unchanged, within the size and time limits, and writes one request log
// line for each request.
package proxy

import (
	"io"
	"log"
	"log/slog"
	"net/http"
	"strings"
	"time"

	"sneak.berlin/go/smallwebwaf/internal/alerts"
	"sneak.berlin/go/smallwebwaf/internal/anomaly"
	"sneak.berlin/go/smallwebwaf/internal/bans"
	"sneak.berlin/go/smallwebwaf/internal/config"
	"sneak.berlin/go/smallwebwaf/internal/lookup"
	"sneak.berlin/go/smallwebwaf/internal/metrics"
	"sneak.berlin/go/smallwebwaf/internal/ratelimit"
	"sneak.berlin/go/smallwebwaf/internal/reputation"
	"sneak.berlin/go/smallwebwaf/internal/requestlog"
	"sneak.berlin/go/smallwebwaf/internal/rules"
)

// How smallwebwaf keeps connections to the app open between requests.
const (
	appIdleConns       = 100
	appIdleConnTimeout = 90 * time.Second
)

// adminPrefix starts the path of every request for smallwebwaf itself,
// which never reaches the app.
const adminPrefix = "/_smallwebwaf/"

// HealthPath is smallwebwaf's health endpoint, which the container's
// health check asks.
const HealthPath = "/_smallwebwaf/healthz"

// MetricsPath is where the metrics are, for a request that carries
// SWWAF_METRICS_TOKEN.
const MetricsPath = "/_smallwebwaf/metrics"

// BansPath is where an admin lists and adds bans, and, followed by / and
// a client's address, lifts them, with SWWAF_ADMIN_TOKEN.
const BansPath = "/_smallwebwaf/bans"

// ClientsPath is where an admin asks what smallwebwaf knows of a client,
// by the client's address after it, with SWWAF_ADMIN_TOKEN.
const ClientsPath = "/_smallwebwaf/clients/"

// Params are what New needs.
type Params struct {
	Config *config.Config
	// RequestLog receives one JSON line per request.
	RequestLog io.Writer
	// ProcessLog receives the process's own messages.
	ProcessLog *slog.Logger
	// GeoJSURL is where clients' AS numbers and countries are looked up
	// while SWWAF_LOOKUP_SOURCE is geojs, normally lookup.URL.
	GeoJSURL string
	// LookupFile is the lookup database they are looked up in while
	// SWWAF_LOOKUP_SOURCE is file, and nil otherwise.
	LookupFile *lookup.File
	// Now tells the time by which requests are counted for the rate
	// limits, bans are made and run out, and GeoJS's answers are kept,
	// normally time.Now in UTC, the time the state files give.
	Now func() time.Time
	// Rules are the rule files' rules, which each request is checked
	// against.
	Rules *rules.Files
	// Alerts receive the alert for each ban the proxy makes or makes
	// permanent, for each count over an anomaly threshold, for each request
	// whose client a blocklist or a DNSBL zone lists, and for GeoJS failing,
	// a fetch of a list failing or a query to a DNSBL zone failing.
	Alerts *alerts.Queue
}

// Server is the server smallwebwaf runs, with the parts of the proxy
// whose state the state files keep, the lookup database, nil unless
// SWWAF_LOOKUP_SOURCE is file, the lists fetched from URLs, which its Run
// fetches, the DNSBL zones' verdicts, and the metrics.
type Server struct {
	*http.Server

	Ledger     *bans.Ledger
	Limiter    *ratelimit.Limiter
	GeoJS      *lookup.GeoJS
	Anomalies  *anomaly.Counters
	LookupFile *lookup.File
	Lists      *reputation.Lists
	DNSBL      *reputation.DNSBL
	Metrics    *metrics.Metrics
}

// New returns the server smallwebwaf runs: each request it reads passes
// through the proxy. Go's server itself refuses a request line and
// headers over SWWAF_CLIENT_REQUEST_HEADER_MAX_BYTES, with 431, closes a
// connection idle for SWWAF_CLIENT_IDLE_TIMEOUT, and applies
// SWWAF_CLIENT_REQUEST_TIMEOUT while the headers arrive; the proxy
// applies the timeouts and size limits from then on.
func New(params Params) *Server {
	errorLog := slog.NewLogLogger(params.ProcessLog.Handler(), slog.LevelWarn)
	m := metrics.New(params.Config.MetricsTopN, params.Config.InstanceName)
	lists, dnsbl := newReputation(params)
	h := &handler{
		config:     params.Config,
		requestLog: params.RequestLog,
		processLog: params.ProcessLog,
		errorLog:   errorLog,
		transport:  newTransport(),
		now:        params.Now,
		metrics:    m,
		limiter: ratelimit.New(ratelimit.Limits{
			PerMinute:      params.Config.RateLimitPerMinute,
			PerHour:        params.Config.RateLimitPerHour,
			PerDay:         params.Config.RateLimitPerDay,
			BytesPerMinute: params.Config.BytesLimitPerMinute,
			BytesPerHour:   params.Config.BytesLimitPerHour,
			BytesPerDay:    params.Config.BytesLimitPerDay,
		}),
		ledger: bans.New(bans.Rules{
			LimitBanDuration:     params.Config.LimitBanDuration,
			LimitBanRepeatWindow: params.Config.LimitBanRepeatWindow,
			MaxBanDuration:       params.Config.MaxBanDuration,
			AttackBanDuration:    params.Config.AttackBanDuration,
			MaxBans:              params.Config.MaxBans,
		}),
		anomalies: anomaly.New(anomaly.Params{
			Client:         params.Config.AnomalyClient,
			Net:            params.Config.AnomalyNet,
			ASN:            params.Config.AnomalyASN,
			Total:          params.Config.AnomalyTotal,
			Watch:          params.Config.AnomalyWatch,
			NetV4Prefix:    params.Config.AnomalyNetV4Prefix,
			NetV6Prefix:    params.Config.AnomalyNetV6Prefix,
			NamedNetblocks: params.Config.WatchNets,
			Alerts:         params.Alerts,
		}),
		lookupFile: params.LookupFile,
		lists:      lists,
		dnsbl:      dnsbl,
		rules:      params.Rules,
		alerts:     params.Alerts,
	}
	h.geojs = lookup.New(lookup.Params{
		URL:     params.GeoJSURL,
		Timeout: params.Config.LookupTimeout,
		// The country lists, the headers and the biased thresholds act on
		// the answer before the request goes on.
		Wait: len(params.Config.DeniedCountries) > 0 ||
			len(params.Config.ExclusivelyAllowedCountries) > 0 ||
			params.Config.AddLookupHeaders || biasedThresholdsSet(params.Config),
		Answered:   h.addLookup,
		Now:        params.Now,
		ProcessLog: params.ProcessLog,
		Metrics:    m,
		Alerts:     params.Alerts,
	})
	m.AddBansAndClients(h.ledger, h.limiter, params.Now)
	m.AddRules(params.Rules)
	m.AddReputation(h.lists, h.dnsbl)

	return &Server{
		Server: &http.Server{
			Addr:              params.Config.ListenAddr,
			Handler:           h,
			ReadHeaderTimeout: params.Config.ClientRequestTimeout,
			// Off is an IdleTimeout of 0, which Go's server replaces with
			// ReadTimeout: no limit, as long as ReadTimeout stays unset.
			IdleTimeout: params.Config.ClientIdleTimeout,
			// Go's server reads 4 KiB past MaxHeaderBytes before it
			// refuses, so the limit a client meets is the setting.
			MaxHeaderBytes: int(params.Config.ClientRequestHeaderMaxBytes - 4<<10),
			ErrorLog:       errorLog,
		},
		Ledger:     h.ledger,
		Limiter:    h.limiter,
		GeoJS:      h.geojs,
		Anomalies:  h.anomalies,
		LookupFile: h.lookupFile,
		Lists:      h.lists,
		DNSBL:      h.dnsbl,
		Metrics:    m,
	}
}

// newReputation returns the lists fetched from URLs and the DNSBL zones'
// verdicts, as the settings in params name them, with none fetched or
// asked for yet.
func newReputation(params Params) (*reputation.Lists, *reputation.DNSBL) {
	cfg := params.Config
	lists := reputation.New(reputation.Params{
		BlocklistURLs: cfg.BlocklistURLs, Refresh: cfg.BlocklistRefresh,
		ASNLimitPercentURL: cfg.ASNLimitPercentURL, Now: params.Now,
		ProcessLog: params.ProcessLog, Alerts: params.Alerts,
	})
	dnsbl := reputation.NewDNSBL(reputation.DNSBLParams{
		Zones: cfg.DNSBLZones, Resolver: cfg.DNSBLResolver, CacheTTL: cfg.ReputationCacheTTL,
		Timeout: cfg.ReputationTimeout, Now: params.Now, ProcessLog: params.ProcessLog,
		Alerts: params.Alerts,
	})

	return lists, dnsbl
}

// handler is the proxy. It holds what every request shares; what belongs
// to one request is in a request.
type handler struct {
	config     *config.Config
	requestLog io.Writer
	processLog *slog.Logger
	errorLog   *log.Logger
	transport  http.RoundTripper
	now        func() time.Time
	metrics    *metrics.Metrics
	limiter    *ratelimit.Limiter
	ledger     *bans.Ledger
	geojs      *lookup.GeoJS
	anomalies  *anomaly.Counters
	lookupFile *lookup.File
	lists      *reputation.Lists
	dnsbl      *reputation.DNSBL
	rules      *rules.Files
	alerts     *alerts.Queue
}

// newTransport returns what carries requests to the app. It never goes
// through a proxy named in the environment, and leaves the app's answers
// compressed or not as the app sent them.
func newTransport() *http.Transport {
	return &http.Transport{
		MaxIdleConns:        appIdleConns,
		MaxIdleConnsPerHost: appIdleConns,
		IdleConnTimeout:     appIdleConnTimeout,
		DisableCompression:  true,
	}
}

// ServeHTTP handles one request: it works out the client, runs the
// checks, passes the request to the app and the answer back within the
// limits, or answers it itself if it is for smallwebwaf, and writes the
// request's log line.
func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	rq := h.newRequest(w, r)
	defer rq.finish()

	// The health endpoint is answered at once, before any check, so that
	// a health checker is never refused. It does not ask the app.
	if r.Method == http.MethodGet && r.URL.Path == HealthPath {
		rq.line.Action = requestlog.ActionAdmin
		// Set here rather than left to Go's server, which would set it only
		// after the log line has taken the response's headers.
		rq.out.Header().Set("Content-Type", "text/plain; charset=utf-8")
		_, _ = io.WriteString(rq.out, "ok\n")

		return
	}

	// Once the request has ended, before its log line is written.
	defer rq.addToHistory()
	defer rq.countAnomalies()

	refused := rq.check(r.Context())
	rq.checked = time.Now()

	if refused != nil {
		rq.answer(*refused)

		return
	}

	// A request for smallwebwaf itself is answered where another would be
	// passed to the app, so that it goes through every check first.
	if strings.HasPrefix(r.URL.Path, adminPrefix) {
		rq.answerAdmin()

		return
	}

	// Once the response has ended, before the request is added to its
	// client's history. Deferred, since ReverseProxy panics to end a
	// response it cannot finish.
	defer rq.countBytes()

	rq.forward(r.Context())
}