servewaf answers rproxy's question about each request and takes its report (closes #170)
check / check (push) Successful in 10m27s
check / check (push) Successful in 10m27s
`smallwebwaf servewaf` runs today's checks on the visitor's request each question carries, answers rproxy, and passes nothing to an app. rproxy reports every request servewaf answered. A report counts an allowed request's bytes and writes servewaf's second log line. The table holds a copy of what a report needs, never the request, so a report cannot reach a request being handled. Judgement call: a question's body is read before any check; over 64 KiB gets 400. Judgement call: a report's bytes and durations have ceilings, so they cannot overflow. Judgement call: a `close` disposition for SWWAF_BAN_RESPONSE=close. Model: opus-5-5
This commit is contained in:
@@ -41,6 +41,15 @@ var (
|
||||
errNotAddress = errors.New("is not an address, such as 203.0.113.9")
|
||||
)
|
||||
|
||||
// answerHealth answers a request for the health endpoint.
|
||||
func (rq *request) answerHealth() {
|
||||
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")
|
||||
}
|
||||
|
||||
// answerAdmin answers a request for smallwebwaf itself, under
|
||||
// /_smallwebwaf/, once it has passed the checks. Each endpoint needs a
|
||||
// token, sent as Authorization: Bearer <token>: the metrics
|
||||
|
||||
@@ -15,9 +15,11 @@ func TestWithEveryAnomalyThresholdOffARequestIsNotCounted(t *testing.T) {
|
||||
// anomaly counters, nor a clock, and the request no response: reading
|
||||
// any of them to count the request panics.
|
||||
rq := &request{
|
||||
h: &handler{config: &config.Config{LookupSource: "geojs"}},
|
||||
client: netip.MustParseAddr("203.0.113.9"),
|
||||
lookedUp: true,
|
||||
decision: decision{
|
||||
client: netip.MustParseAddr("203.0.113.9"),
|
||||
lookedUp: true,
|
||||
},
|
||||
h: &handler{config: &config.Config{LookupSource: "geojs"}},
|
||||
}
|
||||
|
||||
defer func() {
|
||||
|
||||
@@ -323,6 +323,7 @@ func (rq *request) alertBan(ban bans.Ban) {
|
||||
|
||||
rq.h.alerts.Raise(alerts.Alert{
|
||||
Event: event,
|
||||
Host: rq.site,
|
||||
Client: rq.client,
|
||||
Netblock: ban.Netblock,
|
||||
ASN: ban.Notes.ASN,
|
||||
@@ -339,11 +340,11 @@ func (rq *request) alertBan(ban bans.Ban) {
|
||||
func (rq *request) noted(now time.Time, status int) bans.Request {
|
||||
return bans.Request{
|
||||
Time: now,
|
||||
Method: rq.in.Method,
|
||||
Host: rq.in.Host,
|
||||
Path: rq.in.URL.RequestURI(),
|
||||
Method: rq.line.Method,
|
||||
Host: rq.line.Host,
|
||||
Path: rq.requestURI,
|
||||
Status: status,
|
||||
UserAgent: rq.in.UserAgent(),
|
||||
UserAgent: rq.line.UserAgent,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -25,6 +25,8 @@ const (
|
||||
userAgent = "ban-test/1.0"
|
||||
// permanent is the log line's ban_expires for a permanent ban.
|
||||
permanent = "permanent"
|
||||
// banResponseClose is the SWWAF_BAN_RESPONSE that closes the connection.
|
||||
banResponseClose = "close"
|
||||
)
|
||||
|
||||
func TestBrokenLimitBansTheClient(t *testing.T) {
|
||||
@@ -237,7 +239,7 @@ func TestBanResponseAnswersEveryRefusalButTheSizeLimits(t *testing.T) {
|
||||
{"", http.StatusForbidden},
|
||||
{"403", http.StatusForbidden},
|
||||
{"429", http.StatusTooManyRequests},
|
||||
{"close", 0},
|
||||
{banResponseClose, 0},
|
||||
} {
|
||||
t.Run(banResponse+"="+tc.setting, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -113,6 +113,7 @@ func (rq *request) alertWAFBlock(result waf.Result) {
|
||||
|
||||
rq.h.alerts.Raise(alerts.Alert{
|
||||
Event: alerts.EventWAFBlock,
|
||||
Host: rq.site,
|
||||
Client: rq.client,
|
||||
Netblock: rq.h.clientGroup(rq.client),
|
||||
ASN: rq.line.ASN,
|
||||
|
||||
@@ -207,7 +207,7 @@ func TestMetricsCountLimitsAndBans(t *testing.T) {
|
||||
rateLimitPerMinute: "1",
|
||||
rateLimitExemptNets: scraper,
|
||||
denyNets: denied,
|
||||
banResponse: "close",
|
||||
banResponse: banResponseClose,
|
||||
limitBanDuration: "1h",
|
||||
maxBanDuration: "2h",
|
||||
})
|
||||
|
||||
+52
-33
@@ -1,6 +1,8 @@
|
||||
// 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 runs the checks on each request and writes its request log
|
||||
// line. New passes each request it lets through to the app and the app's
|
||||
// answer back, unchanged, within the size and time limits. NewServeWAF instead
|
||||
// answers rproxy's question about each request, and takes rproxy's report once
|
||||
// the visitor's answer has ended, which writes the request a second log line.
|
||||
package proxy
|
||||
|
||||
import (
|
||||
@@ -11,6 +13,8 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
lru "github.com/hashicorp/golang-lru/v2"
|
||||
|
||||
"sneak.berlin/go/smallwebwaf/internal/alerts"
|
||||
"sneak.berlin/go/smallwebwaf/internal/anomaly"
|
||||
"sneak.berlin/go/smallwebwaf/internal/bans"
|
||||
@@ -19,7 +23,6 @@ import (
|
||||
"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"
|
||||
"sneak.berlin/go/smallwebwaf/internal/waf"
|
||||
)
|
||||
@@ -53,7 +56,8 @@ const ClientsPath = "/_smallwebwaf/clients/"
|
||||
// Params are what New needs.
|
||||
type Params struct {
|
||||
Config *config.Config
|
||||
// RequestLog receives one JSON line per request.
|
||||
// RequestLog receives one JSON line per request, and from servewaf a second
|
||||
// when the request's report comes.
|
||||
RequestLog io.Writer
|
||||
// ProcessLog receives the process's own messages.
|
||||
ProcessLog *slog.Logger
|
||||
@@ -109,6 +113,23 @@ type Server struct {
|
||||
// SWWAF_CLIENT_REQUEST_TIMEOUT while the headers arrive; the proxy
|
||||
// applies the timeouts and size limits from then on.
|
||||
func New(params Params) *Server {
|
||||
h := newHandler(params)
|
||||
|
||||
return h.server(&http.Server{
|
||||
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),
|
||||
})
|
||||
}
|
||||
|
||||
// newHandler returns the handler with what every request shares, as the
|
||||
// settings in params make it.
|
||||
func newHandler(params Params) *handler {
|
||||
errorLog := slog.NewLogLogger(params.ProcessLog.Handler(), slog.LevelWarn)
|
||||
m := metrics.New(params.Config.MetricsTopN, params.Config.InstanceName)
|
||||
lists, dnsbl, abuseIPDB := newReputation(params, m)
|
||||
@@ -171,29 +192,7 @@ func New(params Params) *Server {
|
||||
m.AddBansAndClients(h.ledger, h.limiter, params.Now)
|
||||
m.AddRules(params.Rules)
|
||||
|
||||
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,
|
||||
AbuseIPDB: h.abuseIPDB,
|
||||
Metrics: m,
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// newReputation returns the lists fetched from URLs, the DNSBL zones'
|
||||
@@ -274,6 +273,10 @@ type handler struct {
|
||||
rules *rules.Files
|
||||
coreRuleSet *waf.CoreRuleSet
|
||||
alerts *alerts.Queue
|
||||
// awaitingReport holds, in servewaf, the decision on each answered
|
||||
// question by its id until its report comes, and nil for a question still
|
||||
// being handled.
|
||||
awaitingReport *lru.Cache[string, *decision]
|
||||
}
|
||||
|
||||
// newTransport returns what carries requests to the app. It never goes
|
||||
@@ -299,11 +302,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
// 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")
|
||||
rq.answerHealth()
|
||||
|
||||
return
|
||||
}
|
||||
@@ -339,3 +338,23 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
rq.forward(r.Context())
|
||||
}
|
||||
|
||||
// server returns the Server of h, which listens at SWWAF_LISTEN_ADDR and
|
||||
// logs through h, with what httpServer sets.
|
||||
func (h *handler) server(httpServer *http.Server) *Server {
|
||||
httpServer.Addr = h.config.ListenAddr
|
||||
httpServer.ErrorLog = h.errorLog
|
||||
|
||||
return &Server{
|
||||
Server: httpServer,
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -284,6 +284,16 @@ func newProxy(
|
||||
) (*proxy.Server, *output, *alerts.Queue) {
|
||||
t.Helper()
|
||||
|
||||
return newServerWith(t, proxy.New, appURL, geojsURL, now, env)
|
||||
}
|
||||
|
||||
// newServerWith is newProxy for the server build makes.
|
||||
func newServerWith(
|
||||
t *testing.T, build func(proxy.Params) *proxy.Server,
|
||||
appURL, geojsURL string, now func() time.Time, env map[string]string,
|
||||
) (*proxy.Server, *output, *alerts.Queue) {
|
||||
t.Helper()
|
||||
|
||||
settings := map[string]string{
|
||||
"SWWAF_UPSTREAM_URL": appURL, rulesDir: t.TempDir(), instanceName: "app",
|
||||
wafMode: off,
|
||||
@@ -334,7 +344,7 @@ func newProxy(
|
||||
}
|
||||
}
|
||||
|
||||
server := proxy.New(proxy.Params{
|
||||
server := build(proxy.Params{
|
||||
Config: cfg,
|
||||
RequestLog: out,
|
||||
ProcessLog: processLog,
|
||||
|
||||
@@ -110,6 +110,7 @@ func (rq *request) noteHit(hit bans.ReputationHit, reason string) {
|
||||
rq.h.metrics.ReputationHit(hit.Source)
|
||||
rq.h.alerts.Raise(alerts.Alert{
|
||||
Event: alerts.EventReputationHit,
|
||||
Host: rq.site,
|
||||
Client: rq.client,
|
||||
Netblock: rq.h.clientGroup(rq.client),
|
||||
ASN: rq.line.ASN,
|
||||
|
||||
+97
-55
@@ -39,20 +39,20 @@ type refusal struct {
|
||||
limit string
|
||||
}
|
||||
|
||||
// request is one request on its way through smallwebwaf, from the moment
|
||||
// its headers have been read to its log line.
|
||||
type request struct {
|
||||
h *handler
|
||||
in *http.Request
|
||||
// rc sets the deadlines of the connection to the client.
|
||||
rc *http.ResponseController
|
||||
out *responseWriter
|
||||
body *requestBody // nil for a request without a body
|
||||
line requestlog.Line
|
||||
|
||||
client netip.Addr
|
||||
peer netip.Addr
|
||||
peerTrusted bool
|
||||
// decision is what the checks note of a request: its log line, and what
|
||||
// counting the request needs once it has ended. servewaf keeps a copy of it
|
||||
// until the request's report comes. It holds no credentials, since the log line
|
||||
// never does, and neither the request's objects nor their bodies.
|
||||
type decision struct {
|
||||
line requestlog.Line
|
||||
client netip.Addr
|
||||
// requestURI is the path with the query, as a ban's notes keep it.
|
||||
requestURI string
|
||||
// site is, in servewaf, the host the visitor asked for, which the
|
||||
// alerts name, since one servewaf serves many sites. It is "" otherwise.
|
||||
site string
|
||||
// allowed is true for a question answered with DispositionAllow.
|
||||
allowed bool
|
||||
// lookedUp is true once the client's AS number and country have been
|
||||
// looked up, whether or not an answer was there, and lookupAnswer is
|
||||
// what the lookup gave then, the zero Answer while GeoJS had given none.
|
||||
@@ -65,18 +65,37 @@ type request struct {
|
||||
counted bool
|
||||
limitPercent, bytesPercent percentage
|
||||
// attack is true for a request that matched a ban rule or asked for a
|
||||
// trap path, ruleBlocked for one a block rule refused, wafBlocked for
|
||||
// one the Core Rule Set refused, and tokenRefused for one refused for a
|
||||
// missing or wrong token, each an offence its client's history counts.
|
||||
attack, ruleBlocked, wafBlocked, tokenRefused bool
|
||||
// trap path, ruleBlocked for one a block rule refused, and wafBlocked for
|
||||
// one the Core Rule Set refused, each an offence its client's history
|
||||
// counts.
|
||||
attack, ruleBlocked, wafBlocked bool
|
||||
// reputation is the reputation sources that list the client, for the
|
||||
// notes of a ban the request makes.
|
||||
reputation []bans.ReputationHit
|
||||
}
|
||||
|
||||
// request is one request on its way through smallwebwaf, from the moment
|
||||
// its headers have been read to its log line.
|
||||
type request struct {
|
||||
decision
|
||||
|
||||
h *handler
|
||||
in *http.Request
|
||||
// rc sets the deadlines of the connection to the client.
|
||||
rc *http.ResponseController
|
||||
out *responseWriter
|
||||
body *requestBody // nil for a request without a body
|
||||
|
||||
peer netip.Addr
|
||||
peerTrusted bool
|
||||
// tokenRefused is true for a request refused for a missing or wrong
|
||||
// token, an offence its client's history counts.
|
||||
tokenRefused bool
|
||||
// blocklisted is true once a blocklist is found to list the client,
|
||||
// dnsblListed once a DNSBL zone's verdict is, and abuseIPDBHit once
|
||||
// AbuseIPDB's score of it is a hit.
|
||||
blocklisted, dnsblListed, abuseIPDBHit bool
|
||||
// reputation is the reputation sources that list the client, for the
|
||||
// notes of a ban the request makes.
|
||||
reputation []bans.ReputationHit
|
||||
start time.Time
|
||||
start time.Time
|
||||
// checked is when the checks were done, and upstreamStart when the
|
||||
// request was handed to the app.
|
||||
checked time.Time
|
||||
@@ -108,50 +127,62 @@ type request struct {
|
||||
answerStarted time.Time
|
||||
}
|
||||
|
||||
// newRequest starts handling r: it notes the time, counts the request as
|
||||
// under way, works out the client, and starts the log line with what is
|
||||
// known of the request.
|
||||
// newRequest starts handling r: it works out the client from the TCP peer
|
||||
// and X-Forwarded-For, and goes on as startRequest does.
|
||||
func (h *handler) newRequest(w http.ResponseWriter, r *http.Request) *request {
|
||||
peer := peerAddress(r)
|
||||
trusted := h.config.TrustedProxies
|
||||
client := clientAddress(peer, r.Header.Values("X-Forwarded-For"), trusted)
|
||||
|
||||
return h.startRequest(w, r, client, peer, isInside(peer, trusted))
|
||||
}
|
||||
|
||||
// startRequest starts handling r from client, which came through peer: it
|
||||
// notes the time, counts the request as under way, and starts the log line
|
||||
// with what is known of the request. peerTrusted says whether to believe
|
||||
// the request's id and scheme, which r's headers give.
|
||||
func (h *handler) startRequest(
|
||||
w http.ResponseWriter, r *http.Request, client, peer netip.Addr, peerTrusted bool,
|
||||
) *request {
|
||||
h.metrics.RequestStarted()
|
||||
|
||||
start := time.Now()
|
||||
peer := peerAddress(r)
|
||||
trusted := h.config.TrustedProxies
|
||||
peerTrusted := isInside(peer, trusted)
|
||||
forwardedFor := r.Header.Values("X-Forwarded-For")
|
||||
client := clientAddress(peer, forwardedFor, trusted)
|
||||
|
||||
rq := &request{
|
||||
decision: decision{
|
||||
client: client,
|
||||
requestURI: r.URL.RequestURI(),
|
||||
line: requestlog.Line{
|
||||
Time: requestlog.FormatTime(start),
|
||||
Instance: h.config.InstanceName,
|
||||
ClientIP: client.String(),
|
||||
Method: r.Method,
|
||||
Scheme: scheme(r, peerTrusted),
|
||||
Host: r.Host,
|
||||
Path: rules.Path(r),
|
||||
Query: r.URL.RawQuery,
|
||||
Protocol: r.Proto,
|
||||
Referer: r.Referer(),
|
||||
UserAgent: r.UserAgent(),
|
||||
RequestID: requestID(r, peerTrusted),
|
||||
PeerIP: peer.String(),
|
||||
ForwardedFor: strings.Join(forwardedFor, ", "),
|
||||
ClientGroup: h.clientGroup(client).String(),
|
||||
ContentType: r.Header.Get("Content-Type"),
|
||||
RequestHeaders: requestHeaders(r, h.config.LogRequestHeaders),
|
||||
HasAuthorization: len(r.Header.Values("Authorization")) > 0,
|
||||
HasCookie: len(r.Header.Values("Cookie")) > 0,
|
||||
Action: requestlog.ActionForward,
|
||||
},
|
||||
},
|
||||
h: h,
|
||||
in: r,
|
||||
rc: http.NewResponseController(w),
|
||||
out: &responseWriter{ResponseWriter: w},
|
||||
client: client,
|
||||
peer: peer,
|
||||
peerTrusted: peerTrusted,
|
||||
start: start,
|
||||
line: requestlog.Line{
|
||||
Time: requestlog.FormatTime(start),
|
||||
Instance: h.config.InstanceName,
|
||||
ClientIP: client.String(),
|
||||
Method: r.Method,
|
||||
Scheme: scheme(r, peerTrusted),
|
||||
Host: r.Host,
|
||||
Path: rules.Path(r),
|
||||
Query: r.URL.RawQuery,
|
||||
Protocol: r.Proto,
|
||||
Referer: r.Referer(),
|
||||
UserAgent: r.UserAgent(),
|
||||
RequestID: requestID(r, peerTrusted),
|
||||
PeerIP: peer.String(),
|
||||
ForwardedFor: strings.Join(forwardedFor, ", "),
|
||||
ClientGroup: h.clientGroup(client).String(),
|
||||
ContentType: r.Header.Get("Content-Type"),
|
||||
RequestHeaders: requestHeaders(r, h.config.LogRequestHeaders),
|
||||
HasAuthorization: len(r.Header.Values("Authorization")) > 0,
|
||||
HasCookie: len(r.Header.Values("Cookie")) > 0,
|
||||
Action: requestlog.ActionForward,
|
||||
},
|
||||
}
|
||||
|
||||
// A length of -1 is a body whose length was not announced.
|
||||
@@ -569,10 +600,20 @@ func (rq *request) finish() {
|
||||
}
|
||||
|
||||
// Counted before the log line is written, so that the metrics count
|
||||
// every request whose line is out.
|
||||
rq.h.metrics.RequestEnded(line, limit, duration, upstreamDuration)
|
||||
// every request whose line is out. An allowed request is the exception:
|
||||
// its report gives the status, bytes and times the metrics count.
|
||||
if rq.allowed {
|
||||
rq.h.metrics.RequestAllowed()
|
||||
} else {
|
||||
rq.h.metrics.RequestEnded(line, limit, duration, upstreamDuration)
|
||||
}
|
||||
|
||||
err := requestlog.Write(rq.h.requestLog, line)
|
||||
rq.writeLine()
|
||||
}
|
||||
|
||||
// writeLine writes the request's log line.
|
||||
func (rq *request) writeLine() {
|
||||
err := requestlog.Write(rq.h.requestLog, &rq.line)
|
||||
if err != nil {
|
||||
rq.h.processLog.Error("writing the request log failed", "error", err.Error())
|
||||
}
|
||||
@@ -634,6 +675,7 @@ func (rq *request) countAnomalies() {
|
||||
answer, _ := rq.answerAtTheEnd()
|
||||
|
||||
rq.h.anomalies.Count(rq.h.now(), anomaly.Request{
|
||||
Host: rq.site,
|
||||
Client: rq.client,
|
||||
ClientGroup: rq.h.clientGroup(rq.client),
|
||||
ASN: answer.ASN,
|
||||
|
||||
@@ -0,0 +1,464 @@
|
||||
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
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
// 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()
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user