rproxy and servewaf replace the one-process smallwebwaf (closes #172, closes #165)
check / check (push) Failing after 23m49s
check / check (push) Failing after 23m49s
`smallwebwaf` with no command now prints the commands and exits with status 2. The image runs rproxy, or servewaf with SMALLWEBWAF_COMMAND=servewaf. The passing-on code, its settings and its log fields and metric go. internal/proxy keeps the checks, answers, reports and own endpoints; its tests run through the real rproxy. The documents describe the pair. Judgement call: clients.json keeps token_refused, since state files refuse unknown fields. Deviation: script/gitea-requests sets SWWAF_LOG_REQUEST_HEADERS, to label requests. Model: opus-5-5
This commit is contained in:
+68
-562
@@ -2,17 +2,10 @@ package proxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptrace"
|
||||
"net/http/httputil"
|
||||
"net/netip"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"sneak.berlin/go/smallwebwaf/internal/anomaly"
|
||||
@@ -24,19 +17,11 @@ import (
|
||||
"sneak.berlin/go/smallwebwaf/internal/rules"
|
||||
)
|
||||
|
||||
// flushAfterEachWrite has ReverseProxy pass on each part of the app's
|
||||
// answer as soon as it arrives.
|
||||
const flushAfterEachWrite time.Duration = -1
|
||||
|
||||
// refusal is smallwebwaf refusing a request, or refusing to go on with it:
|
||||
// the status the client is answered if the response has not started yet,
|
||||
// 0 to close the connection without an answer, the action the log line
|
||||
// names, and the setting whose size or time limit the request passed, if
|
||||
// that is why.
|
||||
// refusal is servewaf refusing a request: the status the visitor is to get, 0
|
||||
// to close its connection without an answer, and the action the log line names.
|
||||
type refusal struct {
|
||||
status int
|
||||
action string
|
||||
limit string
|
||||
}
|
||||
|
||||
// decision is what the checks note of a request: its log line, and what
|
||||
@@ -48,8 +33,8 @@ type decision struct {
|
||||
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 is, for a question, the host the visitor asked for, which the
|
||||
// alerts name, since one servewaf serves many sites.
|
||||
site string
|
||||
// allowed is true for a question answered with DispositionAllow.
|
||||
allowed bool
|
||||
@@ -78,75 +63,34 @@ type decision struct {
|
||||
reputation []bans.ReputationHit
|
||||
}
|
||||
|
||||
// request is one request on its way through smallwebwaf, from the moment
|
||||
// its headers have been read to its log line.
|
||||
// request is one request to servewaf, 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
|
||||
h *handler
|
||||
in *http.Request
|
||||
out *responseWriter
|
||||
|
||||
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
|
||||
start time.Time
|
||||
// checked is when the checks were done, and upstreamStart when the
|
||||
// request was handed to the app.
|
||||
checked time.Time
|
||||
upstreamStart time.Time
|
||||
// cancel ends the request to the app.
|
||||
cancel context.CancelFunc
|
||||
// refused is the first refusal, from whichever goroutine meets it.
|
||||
refused atomic.Pointer[refusal]
|
||||
// complete is true once the app's whole answer has been passed on.
|
||||
complete bool
|
||||
// upgraded is the connection to the app once the app has switched
|
||||
// protocols, as for a WebSocket, and nil otherwise.
|
||||
upgraded *upgradedConn
|
||||
|
||||
// mu guards what follows. The timeouts run on goroutines of their
|
||||
// own, and the transport starts and stops them, and notes the times
|
||||
// below, from its own; once timersStopped is set, none of the timeouts
|
||||
// acts any more.
|
||||
mu sync.Mutex
|
||||
timersStopped bool
|
||||
clientRequestTimer *time.Timer
|
||||
upstreamRequestTimer *time.Timer
|
||||
upstreamResponseTimer *time.Timer
|
||||
// connected is when there was a connection to the app, requestSent
|
||||
// when the app had been sent the whole request, and answerStarted
|
||||
// when the first byte of its answer arrived.
|
||||
connected time.Time
|
||||
requestSent time.Time
|
||||
answerStarted time.Time
|
||||
}
|
||||
|
||||
// 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))
|
||||
// checked is when the checks were done.
|
||||
checked time.Time
|
||||
// refused is what the request was refused with, nil if it was not.
|
||||
refused *refusal
|
||||
// requestBytes is how many bytes of the request's body have been read.
|
||||
requestBytes int64
|
||||
}
|
||||
|
||||
// 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.
|
||||
// with what is known of the request. isQuestion says whether r is a question,
|
||||
// whose headers give the request's id and scheme.
|
||||
func (h *handler) startRequest(
|
||||
w http.ResponseWriter, r *http.Request, client, peer netip.Addr, peerTrusted bool,
|
||||
w http.ResponseWriter, r *http.Request, client, peer netip.Addr, isQuestion bool,
|
||||
) *request {
|
||||
h.metrics.RequestStarted()
|
||||
|
||||
@@ -162,14 +106,14 @@ func (h *handler) startRequest(
|
||||
Instance: h.config.InstanceName,
|
||||
ClientIP: client.String(),
|
||||
Method: r.Method,
|
||||
Scheme: scheme(r, peerTrusted),
|
||||
Scheme: scheme(r, isQuestion),
|
||||
Host: r.Host,
|
||||
Path: rules.Path(r),
|
||||
Query: r.URL.RawQuery,
|
||||
Protocol: r.Proto,
|
||||
Referer: r.Referer(),
|
||||
UserAgent: r.UserAgent(),
|
||||
RequestID: requestID(r, peerTrusted),
|
||||
RequestID: requestID(r, isQuestion),
|
||||
PeerIP: peer.String(),
|
||||
ForwardedFor: strings.Join(forwardedFor, ", "),
|
||||
ClientGroup: h.clientGroup(client).String(),
|
||||
@@ -180,13 +124,10 @@ func (h *handler) startRequest(
|
||||
Action: requestlog.ActionForward,
|
||||
},
|
||||
},
|
||||
h: h,
|
||||
in: r,
|
||||
rc: http.NewResponseController(w),
|
||||
out: &responseWriter{ResponseWriter: w},
|
||||
peer: peer,
|
||||
peerTrusted: peerTrusted,
|
||||
start: start,
|
||||
h: h,
|
||||
in: r,
|
||||
out: &responseWriter{ResponseWriter: w},
|
||||
start: start,
|
||||
}
|
||||
|
||||
// A length of -1 is a body whose length was not announced.
|
||||
@@ -194,10 +135,6 @@ func (h *handler) startRequest(
|
||||
rq.line.ContentLength = r.ContentLength
|
||||
}
|
||||
|
||||
if r.Body != http.NoBody {
|
||||
rq.body = &requestBody{body: limitBody(r.Body, h.config.RequestMaxBytes), rq: rq}
|
||||
}
|
||||
|
||||
return rq
|
||||
}
|
||||
|
||||
@@ -222,29 +159,21 @@ func requestHeaders(r *http.Request, names []string) map[string]string {
|
||||
return headers
|
||||
}
|
||||
|
||||
// check is the one place where a request can be refused once its client
|
||||
// is known, before anything reaches the app, and before its body is read,
|
||||
// but for the part the Core Rule Set reads. It returns nil to let the
|
||||
// request through. The checks of checkClient come first, answered with
|
||||
// SWWAF_BAN_RESPONSE, 403 for a block rule or the Core Rule Set, or 431 for
|
||||
// servewaf's header limits, and then the size limit, so that a request the
|
||||
// rate limits count is counted even when it is refused for its size. In
|
||||
// observe mode a request checkClient refuses goes on to the size limit like
|
||||
// any other. A size or time limit the Core Rule Set's reading of the body
|
||||
// meets ends the request in either mode. ctx is the request's own context.
|
||||
// check is the one place where a request can be refused. It returns nil to let
|
||||
// the request through. The checks of checkClient decide: a request they refuse
|
||||
// is answered with SWWAF_BAN_RESPONSE, 403 for a block rule or the Core Rule Set,
|
||||
// or 431 for the header limits. In observe mode nothing is refused, and the log
|
||||
// line names what enforce mode would have done. ctx is the request's own context.
|
||||
func (rq *request) check(ctx context.Context) *refusal {
|
||||
action := rq.checkClient(ctx)
|
||||
|
||||
refused := rq.refused.Load()
|
||||
if refused != nil {
|
||||
return refused
|
||||
}
|
||||
|
||||
switch {
|
||||
case action == "":
|
||||
return nil
|
||||
case rq.h.config.Observe:
|
||||
// The log line names what enforce mode would have done.
|
||||
rq.line.WouldAction = action
|
||||
|
||||
return nil
|
||||
case action == requestlog.ActionRuleBlocked || action == requestlog.ActionWAFBlocked:
|
||||
return &refusal{status: http.StatusForbidden, action: action}
|
||||
case action == requestlog.ActionTooLarge:
|
||||
@@ -252,17 +181,6 @@ func (rq *request) check(ctx context.Context) *refusal {
|
||||
default:
|
||||
return rq.banResponse(action)
|
||||
}
|
||||
|
||||
maxBytes := rq.h.config.RequestMaxBytes
|
||||
if maxBytes > 0 && rq.in.ContentLength > maxBytes {
|
||||
return &refusal{
|
||||
status: http.StatusRequestEntityTooLarge,
|
||||
action: requestlog.ActionTooLarge,
|
||||
limit: "SWWAF_REQUEST_MAX_BYTES",
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// checkClient runs the checks on the request's client, and returns the
|
||||
@@ -277,10 +195,9 @@ func (rq *request) check(ctx context.Context) *refusal {
|
||||
// rate limits, unless the client is in SWWAF_RATE_LIMIT_EXEMPT_NETS or the
|
||||
// request's path is exempt under SWWAF_RATE_LIMIT_EXEMPT_PATHS, so that
|
||||
// every other request is counted, each of them by the client's limit
|
||||
// percentages, then SWWAF_TRAP_PATHS, then the rule files, then, in
|
||||
// servewaf, the header limits, and last the Core Rule Set. A request exempt
|
||||
// from the rate limits is exempt from the byte limits too. ctx is the
|
||||
// request's own context.
|
||||
// percentages, then SWWAF_TRAP_PATHS, then the rule files, then the header
|
||||
// limits, and last the Core Rule Set. A request exempt from the rate limits
|
||||
// is exempt from the byte limits too. ctx is the request's own context.
|
||||
func (rq *request) checkClient(ctx context.Context) string {
|
||||
cfg := rq.h.config
|
||||
if isInside(rq.client, cfg.AllowNets) {
|
||||
@@ -381,251 +298,32 @@ func pathExempt(r *http.Request, prefixes []string) bool {
|
||||
})
|
||||
}
|
||||
|
||||
// forward passes the request to the app and the app's answer back. ctx
|
||||
// is the request's own context.
|
||||
func (rq *request) forward(ctx context.Context) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
rq.cancel = cancel
|
||||
ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{
|
||||
GotConn: rq.gotConn,
|
||||
WroteRequest: rq.wroteRequest,
|
||||
GotFirstResponseByte: rq.gotFirstResponseByte,
|
||||
})
|
||||
|
||||
out := rq.in.WithContext(ctx)
|
||||
if rq.body != nil {
|
||||
out.Body = rq.body
|
||||
}
|
||||
|
||||
reverseProxy := &httputil.ReverseProxy{
|
||||
Rewrite: rq.rewrite,
|
||||
Transport: rq.h.transport,
|
||||
FlushInterval: flushAfterEachWrite,
|
||||
ErrorLog: rq.h.errorLog,
|
||||
ModifyResponse: rq.modifyResponse,
|
||||
ErrorHandler: rq.answerError,
|
||||
}
|
||||
|
||||
rq.startRequestTimers()
|
||||
rq.upstreamStart = time.Now()
|
||||
reverseProxy.ServeHTTP(rq.out, out)
|
||||
}
|
||||
|
||||
// rewrite makes the request the app receives: the client's request,
|
||||
// unchanged, sent to SWWAF_UPSTREAM_URL, with the forwarded headers and
|
||||
// the request's id set, without any X-Client-ASN or X-Client-Country the
|
||||
// client sent, whatever SWWAF_ADD_LOOKUP_HEADERS says, and, while it is
|
||||
// set, with the client's AS number and country in them.
|
||||
func (rq *request) rewrite(pr *httputil.ProxyRequest) {
|
||||
upstream := rq.h.config.UpstreamURL
|
||||
pr.Out.URL.Scheme = upstream.Scheme
|
||||
pr.Out.URL.Host = upstream.Host
|
||||
// ReverseProxy drops query parameters it cannot parse; the app gets
|
||||
// the query as the client sent it.
|
||||
pr.Out.URL.RawQuery = pr.In.URL.RawQuery
|
||||
// ReverseProxy sends the path URL.EscapedPath gives, which, for a path
|
||||
// holding a character Go escapes, such as |, is the whole path decoded
|
||||
// and escaped again, %2F turned into /. Opaque is sent as it is, so the
|
||||
// app gets the path as the client sent it. Left out are a target that
|
||||
// is no path, such as *, and a path starting with //, which Go would
|
||||
// send in Opaque as a scheme and host.
|
||||
path := rules.Path(pr.In)
|
||||
if strings.HasPrefix(path, "/") && !strings.HasPrefix(path, "//") {
|
||||
pr.Out.URL.Opaque = path
|
||||
}
|
||||
|
||||
setForwardedHeaders(pr.In, pr.Out, rq.peer, rq.peerTrusted)
|
||||
pr.Out.Header.Set(requestIDHeader, rq.line.RequestID)
|
||||
pr.Out.Header.Del(asnHeader)
|
||||
pr.Out.Header.Del(countryHeader)
|
||||
|
||||
if rq.h.config.AddLookupHeaders {
|
||||
setLookupHeaders(pr.Out.Header, rq.line.ASN, rq.line.Country)
|
||||
}
|
||||
}
|
||||
|
||||
// modifyResponse looks at the app's answer before ReverseProxy passes it
|
||||
// on.
|
||||
func (rq *request) modifyResponse(res *http.Response) error {
|
||||
rq.line.UpstreamStatus = res.StatusCode
|
||||
|
||||
if res.StatusCode == http.StatusSwitchingProtocols {
|
||||
// An upgraded connection, such as a WebSocket, is not cut by the
|
||||
// timeouts. ReverseProxy writes this answer straight to the
|
||||
// connection it takes over, not through rq.out, and then copies
|
||||
// what passes each way through res.Body, the connection to the app.
|
||||
// It may still refuse the switch, as when the app's Upgrade is not
|
||||
// the client's, so the status is noted only once it has taken over
|
||||
// the connection, by rq.out's Hijack.
|
||||
rq.stopTimers()
|
||||
rq.line.Websocket = true
|
||||
|
||||
conn, ok := res.Body.(io.ReadWriteCloser)
|
||||
if ok {
|
||||
rq.upgraded = &upgradedConn{ReadWriteCloser: conn}
|
||||
res.Body = rq.upgraded
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
maxBytes := rq.h.config.ResponseMaxBytes
|
||||
if maxBytes > 0 && res.Body != http.NoBody && res.ContentLength > maxBytes {
|
||||
rq.refuse(refusal{
|
||||
status: http.StatusBadGateway,
|
||||
action: requestlog.ActionTooLarge,
|
||||
limit: "SWWAF_RESPONSE_MAX_BYTES",
|
||||
})
|
||||
|
||||
return errResponseTooLarge
|
||||
}
|
||||
|
||||
// Go's server adds a Content-Type it guesses from the body to a
|
||||
// response whose headers have no Content-Type key. The key with no
|
||||
// value sends none, as the app sent none.
|
||||
if res.Header.Get("Content-Type") == "" {
|
||||
rq.out.Header()["Content-Type"] = nil
|
||||
}
|
||||
|
||||
res.Body = &responseBody{body: limitBody(res.Body, maxBytes), rq: rq}
|
||||
rq.startClientResponseTimeout()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// answerError is ReverseProxy's ErrorHandler: the request could not be
|
||||
// passed to the app, or the app's answer cannot be passed on.
|
||||
func (rq *request) answerError(_ http.ResponseWriter, _ *http.Request, err error) {
|
||||
if rq.upgraded != nil {
|
||||
// The app switched protocols, but the switch was not passed on,
|
||||
// and ReverseProxy leaves the app's connection open when it
|
||||
// refuses it.
|
||||
_ = rq.upgraded.Close()
|
||||
}
|
||||
|
||||
refused := rq.refused.Load()
|
||||
if refused == nil {
|
||||
if rq.in.Context().Err() != nil {
|
||||
return // the client has gone, and there is no one to answer
|
||||
}
|
||||
|
||||
rq.h.processLog.Warn("request to the app failed", "error", err.Error())
|
||||
|
||||
refused = &refusal{
|
||||
status: http.StatusBadGateway,
|
||||
action: requestlog.ActionUpstreamError,
|
||||
}
|
||||
}
|
||||
|
||||
rq.answer(*refused)
|
||||
}
|
||||
|
||||
// answer sends smallwebwaf's own answer, unless the response has already
|
||||
// started, and records the refusal for the log line.
|
||||
func (rq *request) answer(r refusal) {
|
||||
rq.refused.CompareAndSwap(nil, &r)
|
||||
|
||||
if rq.out.status != 0 {
|
||||
return // too late to answer: the connection can only be cut
|
||||
}
|
||||
|
||||
if r.status == 0 {
|
||||
// SWWAF_BAN_RESPONSE is close. This panic has Go's server close
|
||||
// the connection without an answer, and log nothing; the log line
|
||||
// is still written as the handler returns.
|
||||
panic(http.ErrAbortHandler)
|
||||
}
|
||||
|
||||
// A client found too slow is read no more; any other may go on
|
||||
// sending until its time is up, so that Go's server can read the
|
||||
// rest of the body and end the request cleanly.
|
||||
deadline := rq.clientRequestDeadline()
|
||||
if r.status == http.StatusRequestTimeout {
|
||||
deadline = time.Now()
|
||||
}
|
||||
|
||||
rq.stopReadingBody(deadline)
|
||||
|
||||
timeout := rq.h.config.ClientResponseTimeout
|
||||
if timeout > 0 {
|
||||
_ = rq.rc.SetWriteDeadline(time.Now().Add(timeout))
|
||||
}
|
||||
|
||||
http.Error(rq.out, http.StatusText(r.status), r.status)
|
||||
}
|
||||
|
||||
// refuse records r, unless an earlier refusal was, and ends the request
|
||||
// to the app, if one was made: smallwebwaf reads the body of a request
|
||||
// it answers itself too.
|
||||
func (rq *request) refuse(r refusal) {
|
||||
rq.refused.CompareAndSwap(nil, &r)
|
||||
|
||||
if rq.cancel != nil {
|
||||
rq.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
// finish ends the request's timeouts, counts it in the metrics and writes
|
||||
// its log line.
|
||||
// finish counts the request in the metrics and writes its log line.
|
||||
func (rq *request) finish() {
|
||||
rq.stopTimers()
|
||||
|
||||
refused := rq.refused.Load()
|
||||
if refused == nil {
|
||||
rq.stopReadingBody(rq.clientRequestDeadline())
|
||||
}
|
||||
|
||||
line := &rq.line
|
||||
line.Status = rq.out.status
|
||||
line.ResponseBytes = rq.out.bytes
|
||||
header := rq.out.Header()
|
||||
line.ResponseContentType = header.Get("Content-Type")
|
||||
line.CacheControl = header.Get("Cache-Control")
|
||||
line.Location = header.Get("Location")
|
||||
line.RequestBytes = rq.requestBytes()
|
||||
|
||||
// limit is the setting whose size or time limit the request passed.
|
||||
var limit string
|
||||
line.ResponseContentType = rq.out.Header().Get("Content-Type")
|
||||
line.RequestBytes = rq.requestBytes
|
||||
|
||||
switch {
|
||||
case refused != nil:
|
||||
line.Action = refused.action
|
||||
limit = refused.limit
|
||||
case errors.Is(rq.out.err, os.ErrDeadlineExceeded):
|
||||
// The client took longer than SWWAF_CLIENT_RESPONSE_TIMEOUT to
|
||||
// take the response.
|
||||
line.Action = requestlog.ActionTimedOut
|
||||
limit = "SWWAF_CLIENT_RESPONSE_TIMEOUT"
|
||||
case !rq.complete && (rq.out.err != nil || rq.in.Context().Err() != nil):
|
||||
case rq.refused != nil:
|
||||
line.Action = rq.refused.action
|
||||
case rq.out.err != nil || rq.in.Context().Err() != nil:
|
||||
line.Aborted = true
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
duration := now.Sub(rq.start)
|
||||
duration := time.Since(rq.start)
|
||||
line.DurationTotal = requestlog.Milliseconds(duration)
|
||||
line.DurationChecks = timing(rq.start, rq.checked)
|
||||
|
||||
var upstreamDuration time.Duration
|
||||
|
||||
if !rq.upstreamStart.IsZero() {
|
||||
upstreamDuration = now.Sub(rq.upstreamStart)
|
||||
line.DurationUpstreamTotal = new(requestlog.Milliseconds(upstreamDuration))
|
||||
|
||||
rq.mu.Lock()
|
||||
line.DurationUpstreamConnect = timing(rq.upstreamStart, rq.connected)
|
||||
line.DurationUpstreamFirstByte = timing(rq.upstreamStart, rq.answerStarted)
|
||||
rq.mu.Unlock()
|
||||
}
|
||||
|
||||
// Counted before the log line is written, so that the metrics count
|
||||
// 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)
|
||||
rq.h.metrics.RequestEnded(line, duration)
|
||||
}
|
||||
|
||||
rq.writeLine()
|
||||
@@ -657,18 +355,16 @@ func timing(start, end time.Time) *float64 {
|
||||
// before either was there, and one from GeoJS that comes later is added
|
||||
// when it comes.
|
||||
func (rq *request) addToHistory() {
|
||||
forwarded := !rq.upstreamStart.IsZero()
|
||||
request := ratelimit.Request{
|
||||
Forwarded: forwarded,
|
||||
Refused: !forwarded && rq.refused.Load() != nil,
|
||||
Forwarded: rq.allowed,
|
||||
Refused: rq.refused != nil,
|
||||
Status: rq.out.status,
|
||||
RequestBytes: rq.requestBytes(),
|
||||
RequestBytes: rq.requestBytes,
|
||||
ResponseBytes: rq.out.bytes,
|
||||
BrokeLimit: rq.line.Offence == requestlog.OffenceLimit,
|
||||
Attack: rq.attack,
|
||||
RuleBlocked: rq.ruleBlocked,
|
||||
WAFBlocked: rq.wafBlocked,
|
||||
TokenRefused: rq.tokenRefused,
|
||||
}
|
||||
|
||||
rq.h.limiter.AddToHistory(rq.h.clientGroup(rq.client), rq.h.now(), request)
|
||||
@@ -729,228 +425,38 @@ func (rq *request) answerAtTheEnd() (lookup.Answer, bool) {
|
||||
return rq.h.geojs.Kept(rq.h.clientGroup(rq.client))
|
||||
}
|
||||
|
||||
// requestBytes is how many bytes of the request's body have been read.
|
||||
func (rq *request) requestBytes() int64 {
|
||||
if rq.body == nil {
|
||||
return 0
|
||||
}
|
||||
// responseWriter is servewaf's answer to a request. It notes the status and
|
||||
// size for the log line, and the first error writing the answer.
|
||||
type responseWriter struct {
|
||||
http.ResponseWriter
|
||||
|
||||
return rq.body.bytes.Load()
|
||||
// status is the status sent, or zero before one is.
|
||||
status int
|
||||
bytes int64
|
||||
err error
|
||||
}
|
||||
|
||||
// clientRequestDeadline is when the client must have sent its whole
|
||||
// request, or zero when SWWAF_CLIENT_REQUEST_TIMEOUT is off.
|
||||
func (rq *request) clientRequestDeadline() time.Time {
|
||||
timeout := rq.h.config.ClientRequestTimeout
|
||||
if timeout == 0 {
|
||||
return time.Time{}
|
||||
// WriteHeader sends the status and headers.
|
||||
func (w *responseWriter) WriteHeader(status int) {
|
||||
if w.status == 0 {
|
||||
w.status = status
|
||||
}
|
||||
|
||||
return rq.start.Add(timeout)
|
||||
w.ResponseWriter.WriteHeader(status)
|
||||
}
|
||||
|
||||
// stopReadingBody ends, at deadline, the reading of a client body that has
|
||||
// not arrived whole: Go's server then reads no more of it, and closes the
|
||||
// connection after the answer.
|
||||
func (rq *request) stopReadingBody(deadline time.Time) {
|
||||
if rq.body == nil || rq.body.received.Load() {
|
||||
return
|
||||
// Write sends part of the body.
|
||||
func (w *responseWriter) Write(p []byte) (int, error) {
|
||||
if w.status == 0 {
|
||||
w.status = http.StatusOK
|
||||
}
|
||||
|
||||
_ = rq.rc.SetReadDeadline(deadline)
|
||||
}
|
||||
n, err := w.ResponseWriter.Write(p)
|
||||
w.bytes += int64(n)
|
||||
|
||||
// startRequestTimers starts the timeouts that run while the request goes
|
||||
// to the app: SWWAF_CLIENT_REQUEST_TIMEOUT until the client has sent its
|
||||
// whole body, and SWWAF_UPSTREAM_REQUEST_TIMEOUT until the app has been
|
||||
// sent the whole request.
|
||||
func (rq *request) startRequestTimers() {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
if rq.body != nil && rq.h.config.ClientRequestTimeout > 0 {
|
||||
rq.clientRequestTimer = time.AfterFunc(
|
||||
time.Until(rq.clientRequestDeadline()), func() {
|
||||
rq.requestTimedOut("SWWAF_CLIENT_REQUEST_TIMEOUT")
|
||||
})
|
||||
if w.err == nil {
|
||||
w.err = err
|
||||
}
|
||||
|
||||
timeout := rq.h.config.UpstreamRequestTimeout
|
||||
if timeout > 0 {
|
||||
rq.upstreamRequestTimer = time.AfterFunc(timeout, func() {
|
||||
rq.requestTimedOut("SWWAF_UPSTREAM_REQUEST_TIMEOUT")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// requestTimedOut is called when limit, SWWAF_CLIENT_REQUEST_TIMEOUT or
|
||||
// SWWAF_UPSTREAM_REQUEST_TIMEOUT, runs out while the request is still on
|
||||
// its way to the app. The answer names the side smallwebwaf was waiting
|
||||
// on at that moment: 408 when it was waiting for the client to send more
|
||||
// of its body, 504 when it was waiting for the app to be reached or to
|
||||
// take what it had.
|
||||
func (rq *request) requestTimedOut(limit string) {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
if rq.timersStopped {
|
||||
return
|
||||
}
|
||||
|
||||
if rq.body == nil || !rq.body.waiting.Load() {
|
||||
// Once the app has answered, the transport, ReverseProxy and Go's
|
||||
// server each close the client's body, and closing it reads what
|
||||
// is left of a body under 256 KiB without requestBody.Read, so
|
||||
// waiting is false while that read waits for the client. That
|
||||
// read ends now, before refuse: should smallwebwaf still answer
|
||||
// itself, answer then sets the deadline it gives its own answers.
|
||||
if !rq.answerStarted.IsZero() {
|
||||
rq.stopReadingBody(time.Now())
|
||||
}
|
||||
|
||||
rq.refuse(refusal{
|
||||
status: http.StatusGatewayTimeout,
|
||||
action: requestlog.ActionTimedOut,
|
||||
limit: limit,
|
||||
})
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
rq.refuse(refusal{
|
||||
status: http.StatusRequestTimeout,
|
||||
action: requestlog.ActionTimedOut,
|
||||
limit: limit,
|
||||
})
|
||||
// The transport gives up on the app only once its Read of the
|
||||
// client's body returns, so that Read is ended now. The lock keeps
|
||||
// this from reaching the connection after the request is handled.
|
||||
_ = rq.rc.SetReadDeadline(time.Now())
|
||||
}
|
||||
|
||||
// bodyReceived is called once the client has sent its whole body.
|
||||
func (rq *request) bodyReceived() {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
stopTimer(rq.clientRequestTimer)
|
||||
}
|
||||
|
||||
// gotConn is called once there is a connection to the app, a new one or
|
||||
// one kept open from an earlier request.
|
||||
func (rq *request) gotConn(httptrace.GotConnInfo) {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
rq.connected = time.Now()
|
||||
}
|
||||
|
||||
// gotFirstResponseByte is called once the first byte of the app's answer
|
||||
// has arrived.
|
||||
func (rq *request) gotFirstResponseByte() {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
rq.answerStarted = time.Now()
|
||||
}
|
||||
|
||||
// wroteRequest is called once the app has been sent the whole request:
|
||||
// the request timeouts end and SWWAF_UPSTREAM_RESPONSE_TIMEOUT starts.
|
||||
func (rq *request) wroteRequest(info httptrace.WroteRequestInfo) {
|
||||
if info.Err != nil {
|
||||
return // the transport gives up, or tries again
|
||||
}
|
||||
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
if rq.timersStopped {
|
||||
return
|
||||
}
|
||||
|
||||
stopTimer(rq.clientRequestTimer)
|
||||
stopTimer(rq.upstreamRequestTimer)
|
||||
rq.requestSent = time.Now()
|
||||
|
||||
timeout := rq.h.config.UpstreamResponseTimeout
|
||||
if timeout > 0 {
|
||||
rq.upstreamResponseTimer = time.AfterFunc(timeout, rq.responseTimedOut)
|
||||
}
|
||||
}
|
||||
|
||||
// responseTimedOut is called when SWWAF_UPSTREAM_RESPONSE_TIMEOUT runs out
|
||||
// before the app has sent its whole answer.
|
||||
func (rq *request) responseTimedOut() {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
if !rq.timersStopped {
|
||||
rq.refuse(refusal{
|
||||
status: http.StatusGatewayTimeout,
|
||||
action: requestlog.ActionTimedOut,
|
||||
limit: "SWWAF_UPSTREAM_RESPONSE_TIMEOUT",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// responseReceived is called once the app has sent its whole answer. An
|
||||
// app may answer before the client has sent its whole body, as with a 404
|
||||
// it sends without reading it. The timeouts then run on: until the
|
||||
// transport's read of the body returns, Go's server cannot send the
|
||||
// answer, and SWWAF_CLIENT_REQUEST_TIMEOUT is what ends that read for a
|
||||
// client that stops sending.
|
||||
func (rq *request) responseReceived() {
|
||||
rq.complete = true
|
||||
|
||||
if rq.body != nil && !rq.body.received.Load() {
|
||||
return
|
||||
}
|
||||
|
||||
rq.stopTimers()
|
||||
}
|
||||
|
||||
// startClientResponseTimeout sets SWWAF_CLIENT_RESPONSE_TIMEOUT on the
|
||||
// connection to the client: the response must reach the client within it
|
||||
// of the end of the request, or of now if the app answers before it has
|
||||
// the whole request.
|
||||
func (rq *request) startClientResponseTimeout() {
|
||||
timeout := rq.h.config.ClientResponseTimeout
|
||||
if timeout == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
from := rq.sentAt()
|
||||
if from.IsZero() {
|
||||
from = time.Now()
|
||||
}
|
||||
|
||||
_ = rq.rc.SetWriteDeadline(from.Add(timeout))
|
||||
}
|
||||
|
||||
// sentAt is when the app had been sent the whole request, or zero.
|
||||
func (rq *request) sentAt() time.Time {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
return rq.requestSent
|
||||
}
|
||||
|
||||
// stopTimers stops the request's timeouts and keeps any from starting
|
||||
// later: the app's answer is complete, the connection upgraded, or the
|
||||
// request handled.
|
||||
func (rq *request) stopTimers() {
|
||||
rq.mu.Lock()
|
||||
defer rq.mu.Unlock()
|
||||
|
||||
rq.timersStopped = true
|
||||
stopTimer(rq.clientRequestTimer)
|
||||
stopTimer(rq.upstreamRequestTimer)
|
||||
stopTimer(rq.upstreamResponseTimer)
|
||||
}
|
||||
|
||||
// stopTimer stops t, which is nil when its timeout is off.
|
||||
func stopTimer(t *time.Timer) {
|
||||
if t != nil {
|
||||
t.Stop()
|
||||
}
|
||||
return n, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user