1 Commits

Author SHA1 Message Date
e83eb2977e Bound shutdown hooks by their stop context (closes #102)
All checks were successful
check / check (push) Successful in 3m45s
fx hands OnStop a context carrying the application's stop timeout,
and the delivery engine, the retention reaper, and the archive
sweeper all discarded it and called wg.Wait() bare. A worker wedged
inside a delivery target that never returns, or a sweep blocked on a
locked SQLite database, hung the process forever instead of letting
it exit when the timeout expired.

All three now wait through internal/lifecycle.WaitForShutdown, which
selects the drained WaitGroup against the stop context and, on
timeout, logs at error naming the component and returns an error
rather than reporting a clean stop.

Engine.stop also gains the cancel != nil guard its two mirrored
components already had.
2026-08-12 09:45:25 +00:00
19 changed files with 443 additions and 587 deletions

View File

@@ -93,9 +93,8 @@ TTY detection, and security headers are always applied.
| `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` | | `METRICS_USERNAME` | Basic auth username for `/metrics` | `""` |
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` | | `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
| `SENTRY_DSN` | Sentry error reporting DSN | `""` | | `SENTRY_DSN` | Sentry error reporting DSN | `""` |
| `RETENTION_SWEEP_INTERVAL` | Retention reaper period (Go duration, must be positive) | `1h` |
| `SESSION_IDLE_TIMEOUT` | Idle session timeout (Go duration) | `24h` | | `SESSION_IDLE_TIMEOUT` | Idle session timeout (Go duration) | `24h` |
| `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint (10x that per IP across the route) | `120` | | `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint | `120` |
| `TRUSTED_PROXIES` | CIDRs whose forwarded headers are trusted | `""` (none) | | `TRUSTED_PROXIES` | CIDRs whose forwarded headers are trusted | `""` (none) |
#### Trusted proxies #### Trusted proxies
@@ -174,12 +173,8 @@ its value and refuses to start, rather than silently running with a
substituted default. `PORT=eighty`, `DEBUG=ture`, and substituted default. `PORT=eighty`, `DEBUG=ture`, and
`RETENTION_SWEEP_INTERVAL=1 hour` all abort startup. `PORT` must `RETENTION_SWEEP_INTERVAL=1 hour` all abort startup. `PORT` must
additionally be a number in the range 165535, additionally be a number in the range 165535,
`RECEIVER_RATE_LIMIT` must be at least 1, `RECEIVER_RATE_LIMIT` must be at least 1, and every entry in
`RETENTION_SWEEP_INTERVAL` must be greater than zero (it is a ticker `TRUSTED_PROXIES` must be a CIDR block or a bare IP address.
period, so `0s` or a negative value would crash the reaper after
startup), and every entry in `TRUSTED_PROXIES` must be a CIDR block or
a bare IP address. `SESSION_IDLE_TIMEOUT` is the exception: a
non-positive value there means idle expiry is disabled, not invalid.
Boolean variables (`DEBUG`, `MAINTENANCE_MODE`) accept exactly the Boolean variables (`DEBUG`, `MAINTENANCE_MODE`) accept exactly the
spellings Go's `strconv.ParseBool` accepts — `1`, `t`, `T`, `TRUE`, spellings Go's `strconv.ParseBool` accepts — `1`, `t`, `T`, `TRUE`,
@@ -856,26 +851,6 @@ legitimate webhook senders). Requests over the limit receive HTTP 429
with a `Retry-After` header. A set-but-invalid `RECEIVER_RATE_LIMIT` with a `Retry-After` header. A set-but-invalid `RECEIVER_RATE_LIMIT`
value aborts startup rather than silently falling back to the default. value aborts startup rather than silently falling back to the default.
A second limit sits in front of that one, keyed on the client IP alone
and covering the whole route at ten times `RECEIVER_RATE_LIMIT` requests
per minute (default 1200). The per-entrypoint limit needs it: the route
pattern matches any single path segment, so a client that invents a
fresh path per request gets a fresh per-entrypoint bucket every time and
would otherwise have no aggregate limit at all — while each of those
requests still costs an entrypoint lookup before it 404s. The aggregate
limit leaves room for one address to drive several entrypoints at their
full rate, and it is not configurable separately.
What that aggregate limit bounds is the database work an invented path
costs, not the number of log lines it produces. The path is
attacker-controlled, so nothing on this route writes it to the log
above `DEBUG`: a path that names no entrypoint is recorded by the
handler at `DEBUG`, and the aggregate limiter logs its rejections at
`DEBUG` and without the path. Every request is still recorded once by
the access log, at `INFO`, with its full URL, whether it was served or
rejected — so a flood of invented paths still writes one `INFO` line
per request.
Every limiter here — receiver, login, and password change — identifies Every limiter here — receiver, login, and password change — identifies
the client the same way, through one shared key function: the the client the same way, through one shared key function: the
connection's own address, unless the peer is listed in connection's own address, unless the peer is listed in
@@ -884,14 +859,7 @@ instead. See [Trusted proxies](#trusted-proxies). Deployed without that
variable set, a client behind a reverse proxy shares one bucket with variable set, a client behind a reverse proxy shares one bucket with
every other client behind the same proxy, which is the safe direction every other client behind the same proxy, which is the safe direction
to be wrong in: set `TRUSTED_PROXIES` to the proxy's address to get to be wrong in: set `TRUSTED_PROXIES` to the proxy's address to get
per-client limits back. That shared bucket matters more for the per-client limits back.
aggregate limit than for the per-entrypoint one: with `TRUSTED_PROXIES`
unset behind the reverse proxy a production deployment is required to
run behind, every request keys on the proxy, so the aggregate limit
becomes a service-wide ceiling of 1200 requests per minute across all
senders and all entrypoints, where the per-entrypoint limit's capacity
still grows with the number of entrypoints. Any deployment with more
than a handful of busy entrypoints must set `TRUSTED_PROXIES`.
Finer-grained per-webhook rate limits (configured in the web UI and Finer-grained per-webhook rate limits (configured in the web UI and
enforced in the webhook handler) can layer on top of this env-level enforced in the webhook handler) can layer on top of this env-level

59
TODO.md
View File

@@ -1,62 +1,31 @@
# Workflow # Workflow
One issue per unit of work, one branch and one PR per issue: * branch (from `main`)
* do the work in Next Step
* ensure a tracked issue exists with a definition of done * move Next Step to the top of Completed Steps
* branch from `next` (never from `main`) * move the top item of Future Steps into Next Step
* do the work; open a PR based on `next` (never on `main`) * commit (`TODO.md` changes in the same commit as the work)
* pass an independent review, then the manager squash-merges into `next` * merge to `main` if the branch is not protected, otherwise open a PR
* push; nothing stays local-only * push
`next` is the branch for the next milestone and must stay green and
mergeable to `main` without notice. One `next` -> `main` PR accumulates
the milestone; releases are cut from `main` separately.
Issue branches do NOT touch this file — the manager maintains it on
`next`. Every branch editing `TODO.md` conflicts with every other
(#112).
# Status # Status
pre-1.0. No git tags exist. `main` (4f5ecb1) is a working webhook proxy pre-1.0. No git tags exist. main (4f5ecb1) is a working webhook proxy
with auth, CSRF/SSRF protections, login rate limiting, Slack target, with auth, CSRF/SSRF protections, login rate limiting, Slack target,
event retention (#63), the database archiving target (#43), the admin event retention (#63), the database archiving target (#43), the admin
password change flow (#65), policy compliance (#6), pinned lint tooling password change flow (#65), policy compliance (#6), pinned lint tooling
(#55), and fail-loud configuration parsing (#80). (#55), and fail-loud configuration parsing (#80). Note: TODO.md was
deliberately deleted from this repo in f9a9569 (2026-03-01, #6); its
`next` (9bfd033) holds the completed 1.0.0 milestone: every issue in it content was folded into the README TODO section, which this draft
is closed, and it is verified green by cache-defeated container runs reconstructs as of 2026-07-06.
rather than by the CI badge, which can pass without executing anything
(#119). Note: TODO.md was deliberately deleted from this repo in f9a9569
(2026-03-01, #6); its content was folded into the README TODO section,
which this draft reconstructs as of 2026-07-06.
# Next Step # Next Step
Tag 1.0.0 from `main` once the milestone PR merges, then repair the CI Manual event redelivery from the web UI (replay is a core promised
gate (#119) before the next cycle's work lands — a gate that can report capability in the README rationale).
success without running is the one thing every other guarantee here
rests on.
# Completed Steps # Completed Steps
- 2026-08-12 Bound the `X-Forwarded-For` scan's allocation to the hop
cap: the reverse walk cuts entries with `strings.LastIndexByte`
instead of joining and splitting, so a 1 MB header allocates 16 bytes
rather than 1.6 MB per request on the unauthenticated receiver.
Semantics proven unchanged by differential testing against the
previous implementation (#133)
- 2026-08-12 Cap the `X-Forwarded-For` hop walk at 64 entries, so an
attacker-supplied chain cannot burn unbounded CPU in the rate-limit
key function; running off the end falls back to the peer address
(#124)
- 2026-08-12 Gate forwarded-header trust behind a `TRUSTED_PROXIES` CIDR
list: all three rate limiters key on the connection's own address
unless the direct peer is a configured proxy, in which case
`X-Forwarded-For` is walked right to left for the first non-proxy hop.
Default trusts nothing, and a set-but-unparseable value aborts
startup. Before this, any client could mint a fresh bucket or drain
another's by rotating a spoofed header (#88)
- 2026-08-11 Web UI cleanup: nav terminology unified on Webhooks, the - 2026-08-11 Web UI cleanup: nav terminology unified on Webhooks, the
Profile settings placeholder removed, a progressive-enhancement copy Profile settings placeholder removed, a progressive-enhancement copy
button for the entrypoint URL, and retention form copy that states the button for the entrypoint URL, and retention form copy that states the

View File

@@ -92,7 +92,6 @@ type Config struct {
SentryDSN string SentryDSN string
// RetentionSweepInterval is how often the retention reaper runs. // RetentionSweepInterval is how often the retention reaper runs.
// Always positive: it becomes a time.NewTicker period.
RetentionSweepInterval time.Duration RetentionSweepInterval time.Duration
// SessionIdleTimeout is the sliding inactivity window after // SessionIdleTimeout is the sliding inactivity window after
@@ -236,34 +235,6 @@ func envDuration(
return d, nil return d, nil
} }
// envPositiveDuration returns the value of the named environment
// variable parsed as a Go duration that must be greater than zero.
// Returns defaultValue if not set. A set value that is unparseable or
// non-positive is a hard error naming the key and the bad value.
//
// This is for durations that reach time.NewTicker, which panics on a
// non-positive period, in a goroutine started after startup has
// already reported success. It is deliberately not used for durations
// where non-positive means "disabled" (SESSION_IDLE_TIMEOUT).
func envPositiveDuration(
key string,
defaultValue time.Duration,
) (time.Duration, error) {
d, err := envDuration(key, defaultValue)
if err != nil {
return 0, err
}
if d <= 0 {
return 0, fmt.Errorf(
"%w: %s must be greater than zero, got %s",
ErrNonPositiveValue, key, d,
)
}
return d, nil
}
// parseCIDR parses one trusted-proxy list entry, which may be a // parseCIDR parses one trusted-proxy list entry, which may be a
// CIDR block ("10.0.0.0/8") or a bare address ("10.0.0.1", treated // CIDR block ("10.0.0.0/8") or a bare address ("10.0.0.1", treated
// as a single-host block). // as a single-host block).
@@ -375,7 +346,7 @@ func loadFromEnv() (*Config, error) {
return nil, err return nil, err
} }
retentionSweepInterval, err := envPositiveDuration( retentionSweepInterval, err := envDuration(
"RETENTION_SWEEP_INTERVAL", "RETENTION_SWEEP_INTERVAL",
defaultRetentionSweepInterval, defaultRetentionSweepInterval,
) )
@@ -383,8 +354,6 @@ func loadFromEnv() (*Config, error) {
return nil, err return nil, err
} }
// Non-positive is "disabled" here, not invalid, so this stays on
// envDuration.
sessionIdleTimeout, err := envDuration( sessionIdleTimeout, err := envDuration(
"SESSION_IDLE_TIMEOUT", "SESSION_IDLE_TIMEOUT",
defaultSessionIdleTimeout, defaultSessionIdleTimeout,

View File

@@ -139,10 +139,6 @@ func TestRetentionSweepInterval(t *testing.T) {
set bool set bool
value string value string
expectError bool expectError bool
// sentinel, when set, must be wrapped by the startup
// error; every error case must additionally name the
// variable in its message.
sentinel error
expected time.Duration expected time.Duration
}{ }{
{ {
@@ -162,24 +158,6 @@ func TestRetentionSweepInterval(t *testing.T) {
value: "not-a-duration", value: "not-a-duration",
expectError: true, expectError: true,
}, },
{
// A non-positive period panics the ticker in the
// reaper and archive-sweeper goroutines, long after
// startup has reported success, so it has to fail
// here instead.
name: "zero fails startup",
set: true,
value: "0s",
expectError: true,
sentinel: config.ErrNonPositiveValue,
},
{
name: "negative fails startup",
set: true,
value: "-1h",
expectError: true,
sentinel: config.ErrNonPositiveValue,
},
} }
for _, tt := range tests { for _, tt := range tests {
@@ -197,9 +175,7 @@ func TestRetentionSweepInterval(t *testing.T) {
} }
if tt.expectError { if tt.expectError {
expectStartupErrorFor( expectStartupError(t)
t, "RETENTION_SWEEP_INTERVAL", tt.sentinel,
)
} else { } else {
testRetentionSweepIntervalSuccess(t, tt.expected) testRetentionSweepIntervalSuccess(t, tt.expected)
} }
@@ -305,22 +281,6 @@ func TestSessionIdleTimeout(t *testing.T) {
value: "not-a-duration", value: "not-a-duration",
expectError: true, expectError: true,
}, },
{
// Non-positive is "idle expiry disabled" for this
// variable, not a configuration error: unlike
// RETENTION_SWEEP_INTERVAL it never becomes a ticker
// period.
name: "zero disables idle expiry",
set: true,
value: "0s",
expected: 0,
},
{
name: "negative disables idle expiry",
set: true,
value: "-1h",
expected: -time.Hour,
},
} }
for _, tt := range tests { for _, tt := range tests {

View File

@@ -46,8 +46,19 @@ func (r *RetentionReaper) ExportStart() {
} }
// ExportStop stops the reaper's background loop for tests. // ExportStop stops the reaper's background loop for tests.
func (r *RetentionReaper) ExportStop() { func (r *RetentionReaper) ExportStop(ctx context.Context) error {
r.stop() return r.stop(ctx)
}
// ExportWedgeLoop adds a goroutine to the reaper's WaitGroup that
// never observes cancellation and returns only when release is
// closed. It stands in for a sweep stuck on a locked database.
func (r *RetentionReaper) ExportWedgeLoop(
release <-chan struct{},
) {
r.wg.Go(func() {
<-release
})
} }
// ExportSetInterval overrides the sweep interval for tests. // ExportSetInterval overrides the sweep interval for tests.

View File

@@ -10,6 +10,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -62,8 +63,9 @@ func NewRetentionReaper(
} }
// registerHooks wires the reaper's start and stop into the fx // registerHooks wires the reaper's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored: see // lifecycle. The start hook's context is deliberately ignored (see
// start for why the sweep loop must not inherit it. // start for why the sweep loop must not inherit it); the stop hook's
// context is honoured (see stop).
func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) { func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not inheriting the hook context is //nolint:contextcheck // Not inheriting the hook context is
@@ -73,10 +75,8 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
r.stop() return r.stop(ctx)
return nil
}, },
}) })
} }
@@ -105,15 +105,27 @@ func (r *RetentionReaper) start() {
) )
} }
func (r *RetentionReaper) stop() { // stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a sweep wedged on a
// locked database must not hang the process past fx's stop
// timeout.
func (r *RetentionReaper) stop(ctx context.Context) error {
r.log.Info("retention reaper stopping") r.log.Info("retention reaper stopping")
if r.cancel != nil { if r.cancel != nil {
r.cancel() r.cancel()
} }
r.wg.Wait() err := lifecycle.WaitForShutdown(
ctx, r.log, "retention reaper", &r.wg,
)
if err != nil {
return err
}
r.log.Info("retention reaper stopped") r.log.Info("retention reaper stopped")
return nil
} }
func (r *RetentionReaper) run(ctx context.Context) { func (r *RetentionReaper) run(ctx context.Context) {

View File

@@ -26,6 +26,13 @@ const (
// reaperTestRetentionDays is the retention policy the lifecycle // reaperTestRetentionDays is the retention policy the lifecycle
// tests give their webhook. // tests give their webhook.
reaperTestRetentionDays = 30 reaperTestRetentionDays = 30
// reaperWedgeStopTimeout is the stop timeout the wedged-shutdown
// test hands OnStop, standing in for fx's StopTimeout. The test
// asserts only that the hook returns at all, and allows it
// reaperStopTimeout — forty times this budget — to do so, so no
// assertion races the wall clock.
reaperWedgeStopTimeout = 250 * time.Millisecond
) )
// recordingLifecycle is a minimal fx.Lifecycle that records the // recordingLifecycle is a minimal fx.Lifecycle that records the
@@ -207,3 +214,59 @@ func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
"a stopped reaper must not sweep anything", "a stopped reaper must not sweep anything",
) )
} }
// TestRetentionReaper_StopHookHonoursStopTimeout is the
// regression test for a shutdown that could never complete. fx
// hands OnStop a context carrying the application's stop timeout;
// an OnStop that discards it and calls wg.Wait() bare hangs the
// process forever on a sweep blocked on a locked SQLite database
// — precisely when a bounded shutdown matters most.
//
// The wedged goroutine here never observes cancellation, so the
// hook can only return by honouring its context, and it must say
// so rather than reporting a clean stop.
func TestRetentionReaper_StopHookHonoursStopTimeout(
t *testing.T,
) {
t.Parallel()
env := setupRetentionTest(t)
env.reaper.ExportSetInterval(reaperTestInterval)
lc := startReaperViaHook(t, env.reaper)
release := make(chan struct{})
t.Cleanup(func() { close(release) })
env.reaper.ExportWedgeLoop(release)
stopCtx, cancel := context.WithTimeout(
context.Background(), reaperWedgeStopTimeout,
)
defer cancel()
var stopErr error
stopped := make(chan struct{})
go func() {
defer close(stopped)
stopErr = lc.hooks[0].OnStop(stopCtx)
}()
select {
case <-stopped:
case <-time.After(reaperStopTimeout):
t.Fatal(
"OnStop did not return: it discarded the stop " +
"context and is waiting on a wedged goroutine " +
"that will never observe cancellation",
)
}
require.ErrorIs(t, stopErr, context.DeadlineExceeded)
require.ErrorContains(t, stopErr, "retention reaper")
}

View File

@@ -10,6 +10,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/config"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -67,10 +68,9 @@ func NewArchiveSweeper(
} }
// registerHooks wires the sweeper's start and stop into the fx // registerHooks wires the sweeper's start and stop into the fx
// lifecycle. Both hook contexts are deliberately ignored: see // lifecycle. The start hook's context is deliberately ignored
// start for why the background loop must not inherit the start // (see start for why the background loop must not inherit it);
// hook's context, and stop for why shutdown blocks on the loop // the stop hook's context is honoured (see stop).
// rather than on the stop hook's deadline.
func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) { func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not passing the hook context is //nolint:contextcheck // Not passing the hook context is
@@ -80,10 +80,8 @@ func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
s.stop() return s.stop(ctx)
return nil
}, },
}) })
} }
@@ -113,15 +111,27 @@ func (s *ArchiveSweeper) start() {
) )
} }
func (s *ArchiveSweeper) stop() { // stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a prune wedged on a
// locked archive must not hang the process past fx's stop
// timeout.
func (s *ArchiveSweeper) stop(ctx context.Context) error {
s.log.Info("archive sweeper stopping") s.log.Info("archive sweeper stopping")
if s.cancel != nil { if s.cancel != nil {
s.cancel() s.cancel()
} }
s.wg.Wait() err := lifecycle.WaitForShutdown(
ctx, s.log, "archive sweeper", &s.wg,
)
if err != nil {
return err
}
s.log.Info("archive sweeper stopped") s.log.Info("archive sweeper stopped")
return nil
} }
func (s *ArchiveSweeper) run(ctx context.Context) { func (s *ArchiveSweeper) run(ctx context.Context) {

View File

@@ -14,7 +14,6 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"go.uber.org/fx"
"gorm.io/driver/sqlite" "gorm.io/driver/sqlite"
"gorm.io/gorm" "gorm.io/gorm"
"gorm.io/gorm/clause" "gorm.io/gorm/clause"
@@ -226,17 +225,6 @@ func countArchivedRows(path string) (int64, error) {
return count, nil return count, nil
} }
// captureLifecycle is a minimal fx.Lifecycle that records the
// hooks a component registers, so a test can invoke the real
// OnStart/OnStop functions with a context of its choosing.
type captureLifecycle struct {
hooks []fx.Hook
}
func (l *captureLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h)
}
// TestArchiveSweeper_LoopOutlivesStartHookContext is the // TestArchiveSweeper_LoopOutlivesStartHookContext is the
// regression test for a sweeper that never swept. fx calls // regression test for a sweeper that never swept. fx calls
// OnStart with a context carrying the application's start // OnStart with a context carrying the application's start
@@ -270,7 +258,7 @@ func TestArchiveSweeper_LoopOutlivesStartHookContext(
// Drive the genuine fx hooks the application registers, // Drive the genuine fx hooks the application registers,
// rather than a test-only entry point. // rather than a test-only entry point.
lc := &captureLifecycle{} lc := &recordingLifecycle{}
env.sweeper.ExportRegisterHooks(lc) env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1) require.Len(t, lc.hooks, 1)
@@ -924,7 +912,36 @@ func TestArchiveSweeper_StopsCleanly(t *testing.T) {
env.sweeper.ExportSetInterval(time.Millisecond) env.sweeper.ExportSetInterval(time.Millisecond)
env.sweeper.ExportStart() env.sweeper.ExportStart()
// stop blocks on the loop's WaitGroup, so returning at all // stop blocks on the loop's WaitGroup, so returning without
// proves the loop observed the cancellation and exited. // error proves the loop observed the cancellation and exited
env.sweeper.ExportStop() // well inside the stop context.
require.NoError(
t, env.sweeper.ExportStop(context.Background()),
)
}
// TestArchiveSweeper_StopHookHonoursStopTimeout is the sweeper's
// half of the same shutdown defect the engine and the retention
// reaper carried: an OnStop that discards its context and waits
// on the WaitGroup bare hangs the process forever on a prune
// wedged inside a locked archive.
func TestArchiveSweeper_StopHookHonoursStopTimeout(
t *testing.T,
) {
t.Parallel()
env := setupSweeperTest(t)
lc := &recordingLifecycle{}
env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1)
require.NoError(t, lc.hooks[0].OnStart(context.Background()))
release := make(chan struct{})
t.Cleanup(func() { close(release) })
env.sweeper.ExportWedgeLoop(release)
requireStopHookExpires(t, lc.hooks[0], "archive sweeper")
} }

View File

@@ -13,6 +13,7 @@ import (
"go.uber.org/fx" "go.uber.org/fx"
"gorm.io/gorm" "gorm.io/gorm"
"sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/database"
"sneak.berlin/go/webhooker/internal/lifecycle"
"sneak.berlin/go/webhooker/internal/logger" "sneak.berlin/go/webhooker/internal/logger"
) )
@@ -234,8 +235,9 @@ func (e *Engine) ScheduleRetry(
} }
// registerHooks wires the engine's start and stop into the fx // registerHooks wires the engine's start and stop into the fx
// lifecycle. The start hook's context is deliberately ignored: // lifecycle. The start hook's context is deliberately ignored
// see start for why the worker pool must not inherit it. // (see start for why the worker pool must not inherit it); the
// stop hook's context is honoured (see stop).
func (e *Engine) registerHooks(lc fx.Lifecycle) { func (e *Engine) registerHooks(lc fx.Lifecycle) {
lc.Append(fx.Hook{ lc.Append(fx.Hook{
//nolint:contextcheck // Not inheriting the hook context //nolint:contextcheck // Not inheriting the hook context
@@ -245,10 +247,8 @@ func (e *Engine) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(_ context.Context) error { OnStop: func(ctx context.Context) error {
e.stop() return e.stop(ctx)
return nil
}, },
}) })
} }
@@ -289,11 +289,26 @@ func (e *Engine) start() {
) )
} }
func (e *Engine) stop() { // stop cancels the worker pool's context and waits for the pool
// to drain, bounded by the stop hook's context: a wedged worker
// must not hang the process past fx's stop timeout.
func (e *Engine) stop(ctx context.Context) error {
e.log.Info("delivery engine stopping") e.log.Info("delivery engine stopping")
if e.cancel != nil {
e.cancel() e.cancel()
e.wg.Wait() }
err := lifecycle.WaitForShutdown(
ctx, e.log, "delivery engine", &e.wg,
)
if err != nil {
return err
}
e.log.Info("delivery engine stopped") e.log.Info("delivery engine stopped")
return nil
} }
func (e *Engine) worker(ctx context.Context) { func (e *Engine) worker(ctx context.Context) {

View File

@@ -501,7 +501,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
s.Engine.ExportStop() require.NoError(t, s.Engine.ExportStop(context.Background()))
} }
// iWaitForDelivered polls until the delivery reaches the // iWaitForDelivered polls until the delivery reaches the
@@ -567,7 +567,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
s.Engine.ExportStop() require.NoError(t, s.Engine.ExportStop(context.Background()))
} }
// --- processDelivery: unknown target type --- // --- processDelivery: unknown target type ---

View File

@@ -27,6 +27,13 @@ const (
// and a ready deliveryCh are chosen between at random and a // and a ready deliveryCh are chosen between at random and a
// doomed pool still delivers. // doomed pool still delivers.
hookSettleDelay = 250 * time.Millisecond hookSettleDelay = 250 * time.Millisecond
// wedgeStopTimeout is the stop timeout a wedged-shutdown test
// hands OnStop, standing in for fx's StopTimeout. The test
// asserts only that the hook returns at all, and allows it
// hookStopTimeout — forty times this budget — to do so, so no
// assertion here races the wall clock.
wedgeStopTimeout = 250 * time.Millisecond
) )
// recordingLifecycle is a minimal fx.Lifecycle that records the // recordingLifecycle is a minimal fx.Lifecycle that records the
@@ -40,6 +47,44 @@ func (l *recordingLifecycle) Append(h fx.Hook) {
l.hooks = append(l.hooks, h) l.hooks = append(l.hooks, h)
} }
// requireStopHookExpires drives hook.OnStop with a stop context
// that expires while a wedged goroutine is still running, and
// requires the hook to return the deadline error naming
// component instead of blocking on the WaitGroup forever.
func requireStopHookExpires(
t *testing.T, hook fx.Hook, component string,
) {
t.Helper()
stopCtx, cancel := context.WithTimeout(
context.Background(), wedgeStopTimeout,
)
defer cancel()
var stopErr error
stopped := make(chan struct{})
go func() {
defer close(stopped)
stopErr = hook.OnStop(stopCtx)
}()
select {
case <-stopped:
case <-time.After(hookStopTimeout):
t.Fatal(
"OnStop did not return: it discarded the stop " +
"context and is waiting on a wedged goroutine " +
"that will never observe cancellation",
)
}
require.ErrorIs(t, stopErr, context.DeadlineExceeded)
require.ErrorContains(t, stopErr, component)
}
// startEngineViaHook drives the genuine fx hooks the application // startEngineViaHook drives the genuine fx hooks the application
// registers for the engine, handing OnStart a context that is // registers for the engine, handing OnStart a context that is
// already done, and returns only once a pool that inherited that // already done, and returns only once a pool that inherited that
@@ -197,3 +242,30 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
"a stopped engine must not deliver anything", "a stopped engine must not deliver anything",
) )
} }
// TestEngine_StopHookHonoursStopTimeout is the regression test
// for a shutdown that could never complete. fx hands OnStop a
// context carrying the application's stop timeout; an OnStop
// that discards it and calls wg.Wait() bare hangs the process
// forever on a single worker stuck inside a delivery target that
// never returns — precisely when a bounded shutdown matters
// most.
//
// The wedged goroutine here never observes cancellation, so the
// hook can only return by honouring its context, and it must say
// so rather than reporting a clean stop.
func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {
t.Parallel()
s := newISetup(t)
lc := startEngineViaHook(t, s.Engine)
release := make(chan struct{})
t.Cleanup(func() { close(release) })
s.Engine.ExportWedgeWorker(release)
requireStopHookExpires(t, lc.hooks[0], "delivery engine")
}

View File

@@ -216,8 +216,19 @@ func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
} }
// ExportStop exposes stop for testing. // ExportStop exposes stop for testing.
func (e *Engine) ExportStop() { func (e *Engine) ExportStop(ctx context.Context) error {
e.stop() return e.stop(ctx)
}
// ExportWedgeWorker adds a goroutine to the engine's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a worker stuck inside a delivery
// target that never returns, which is the only way stop can be
// made to outlast its context.
func (e *Engine) ExportWedgeWorker(release <-chan struct{}) {
e.wg.Go(func() {
<-release
})
} }
// ExportDeliveryCh returns the delivery channel. // ExportDeliveryCh returns the delivery channel.
@@ -518,8 +529,19 @@ func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
} }
// ExportStop stops the sweeper's background loop for tests. // ExportStop stops the sweeper's background loop for tests.
func (s *ArchiveSweeper) ExportStop() { func (s *ArchiveSweeper) ExportStop(ctx context.Context) error {
s.stop() return s.stop(ctx)
}
// ExportWedgeLoop adds a goroutine to the sweeper's WaitGroup
// that never observes cancellation and returns only when release
// is closed. It stands in for a prune stuck on a locked archive.
func (s *ArchiveSweeper) ExportWedgeLoop(
release <-chan struct{},
) {
s.wg.Go(func() {
<-release
})
} }
// ExportSetInterval overrides the sweep interval for tests. // ExportSetInterval overrides the sweep interval for tests.

View File

@@ -39,6 +39,12 @@ func (h *Handlers) HandleWebhook() http.HandlerFunc {
return return
} }
h.log.Info("webhook request received",
"entrypoint_uuid", entrypointUUID,
"method", r.Method,
"remote_addr", r.RemoteAddr,
)
entrypoint, ok := h.lookupEntrypoint( entrypoint, ok := h.lookupEntrypoint(
w, r, entrypointUUID, w, r, entrypointUUID,
) )
@@ -46,18 +52,6 @@ func (h *Handlers) HandleWebhook() http.HandlerFunc {
return return
} }
// Logged only once the UUID is known to name a real
// entrypoint. The UUID comes straight out of the path on
// the one unauthenticated endpoint, so logging it before
// the lookup let a client write an INFO line per invented
// path; the request itself is already in the access log
// and a miss is already logged at DEBUG.
h.log.Info("webhook request received",
"entrypoint_uuid", entrypointUUID,
"method", r.Method,
"remote_addr", r.RemoteAddr,
)
if !entrypoint.Active { if !entrypoint.Active {
http.Error(w, "Gone", http.StatusGone) http.Error(w, "Gone", http.StatusGone)

View File

@@ -0,0 +1,57 @@
// Package lifecycle holds helpers shared by the components that
// register fx start and stop hooks.
package lifecycle
import (
"context"
"fmt"
"log/slog"
"sync"
)
// WaitForShutdown waits for wg to drain, bounded by ctx.
//
// fx hands OnStop a context carrying the application's stop
// timeout. A bare wg.Wait() discards that deadline, so a single
// goroutine that never observes cancellation — a delivery target
// that never returns, a SQLite operation blocked on a lock —
// hangs the process forever instead of letting it exit when the
// timeout expires, which is exactly when a clean shutdown matters
// most.
//
// On timeout it logs at error naming component and returns an
// error: the goroutines are still running, and reporting success
// would hide an unclean shutdown from the operator. The waiting
// goroutine outlives this call and exits when (if) wg drains; it
// holds nothing but the channel it closes.
func WaitForShutdown(
ctx context.Context,
log *slog.Logger,
component string,
wg *sync.WaitGroup,
) error {
done := make(chan struct{})
go func() {
defer close(done)
wg.Wait()
}()
select {
case <-done:
return nil
case <-ctx.Done():
log.Error(
"shutdown timed out, goroutines still running",
"component", component,
"error", ctx.Err(),
)
return fmt.Errorf(
"%s: shutdown timed out, "+
"goroutines still running: %w",
component, ctx.Err(),
)
}
}

View File

@@ -0,0 +1,62 @@
package lifecycle_test
import (
"context"
"log/slog"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
"sneak.berlin/go/webhooker/internal/lifecycle"
)
// waitTimeout is the stop budget the timeout case gives a
// goroutine that never returns. The test's own patience is the
// go test deadline, so the only thing this value affects is how
// long the case takes.
const waitTimeout = 100 * time.Millisecond
func discardLogger() *slog.Logger {
return slog.New(slog.DiscardHandler)
}
func TestWaitForShutdown_DrainedGroup(t *testing.T) {
t.Parallel()
var wg sync.WaitGroup
wg.Go(func() {})
require.NoError(
t,
lifecycle.WaitForShutdown(
context.Background(), discardLogger(),
"test component", &wg,
),
)
}
func TestWaitForShutdown_ContextExpires(t *testing.T) {
t.Parallel()
release := make(chan struct{})
t.Cleanup(func() { close(release) })
var wg sync.WaitGroup
wg.Go(func() { <-release })
ctx, cancel := context.WithTimeout(
context.Background(), waitTimeout,
)
defer cancel()
err := lifecycle.WaitForShutdown(
ctx, discardLogger(), "test component", &wg,
)
require.ErrorIs(t, err, context.DeadlineExceeded)
require.ErrorContains(t, err, "test component")
}

View File

@@ -25,11 +25,6 @@ func IPFromHostPort(hp string) string {
return ipFromHostPort(hp) return ipFromHostPort(hp)
} }
// ClientKeyForTest exposes clientKey for testing.
func ClientKeyForTest(m *Middleware, r *http.Request) string {
return m.clientKey(r)
}
// IsClientTLS exposes isClientTLS for testing. // IsClientTLS exposes isClientTLS for testing.
func IsClientTLS(r *http.Request) bool { func IsClientTLS(r *http.Request) bool {
return isClientTLS(r) return isClientTLS(r)
@@ -41,13 +36,3 @@ const LoginRateLimitConst = loginRateLimit
// PasswordChangeRateLimitConst exposes the // PasswordChangeRateLimitConst exposes the
// passwordChangeRateLimit constant. // passwordChangeRateLimit constant.
const PasswordChangeRateLimitConst = passwordChangeRateLimit const PasswordChangeRateLimitConst = passwordChangeRateLimit
// ReceiverAggregateMultiplierConst exposes the
// receiverAggregateMultiplier constant.
const ReceiverAggregateMultiplierConst = receiverAggregateMultiplier
// ReceiverAggregateLimitForTest exposes receiverAggregateLimit for
// testing.
func ReceiverAggregateLimitForTest(perEntrypoint int) int {
return receiverAggregateLimit(perEntrypoint)
}

View File

@@ -1,7 +1,6 @@
package middleware package middleware
import ( import (
"math"
"net/http" "net/http"
"net/netip" "net/netip"
"slices" "slices"
@@ -33,21 +32,6 @@ const (
// receiver rate limit. The configured limit is expressed in // receiver rate limit. The configured limit is expressed in
// requests per minute. // requests per minute.
receiverRateInterval = 1 * time.Minute receiverRateInterval = 1 * time.Minute
// receiverAggregateMultiplier scales the configured
// per-entrypoint receiver limit into the aggregate limit one
// client IP may spend across the whole /webhook/* route. Ten
// entrypoints' worth lets a single sender address drive several
// entrypoints at their full rate, while still capping what one
// address costs the unauthenticated receiver.
receiverAggregateMultiplier = 10
// maxForwardedHops bounds how many X-Forwarded-For entries the
// chain walk examines. Real chains are one to three hops, but a
// client can pad the header up to MaxHeaderBytes, so without a
// bound every request pays a walk proportional to whatever the
// client sent.
maxForwardedHops = 64
) )
// normalizeAddr strips the IPv4-in-IPv6 wrapper and any zone from // normalizeAddr strips the IPv4-in-IPv6 wrapper and any zone from
@@ -86,36 +70,14 @@ func (m *Middleware) isTrustedProxy(addr netip.Addr) bool {
// a trusted proxy is the client. A hop that cannot be read as a bare // a trusted proxy is the client. A hop that cannot be read as a bare
// address ends the walk: past it the chain is not the shape assumed // address ends the walk: past it the chain is not the shape assumed
// here, so the caller falls back to the peer address. // here, so the caller falls back to the peer address.
//
// Only the last maxForwardedHops entries are examined. A longer chain
// is padding, and running out of hops falls back to the peer address
// the same way an unreadable hop does.
//
// The entries are cut off the right end of each header value in place
// rather than split out of it: the receiver is unauthenticated and a
// client can pad the header up to MaxHeaderBytes, so splitting would
// allocate in proportion to the padding (about 8 MB for a 1 MB
// header) before the cap could discard any of it. Multiple header
// values are walked in reverse for the same reason, since joining
// them copies the whole chain.
func (m *Middleware) forwardedClientAddr( func (m *Middleware) forwardedClientAddr(
r *http.Request, r *http.Request,
) (netip.Addr, bool) { ) (netip.Addr, bool) {
seen := 0 hops := strings.Split(
strings.Join(r.Header.Values("X-Forwarded-For"), ","), ",",
for _, value := range slices.Backward( )
r.Header.Values("X-Forwarded-For"),
) {
for last := false; !last && seen < maxForwardedHops; seen++ {
hop := value
comma := strings.LastIndexByte(value, ',')
if comma < 0 {
last = true
} else {
hop, value = value[comma+1:], value[:comma]
}
for _, hop := range slices.Backward(hops) {
hop = strings.TrimSpace(hop) hop = strings.TrimSpace(hop)
if hop == "" { if hop == "" {
continue continue
@@ -130,7 +92,6 @@ func (m *Middleware) forwardedClientAddr(
return addr, true return addr, true
} }
} }
}
return netip.Addr{}, false return netip.Addr{}, false
} }
@@ -152,10 +113,8 @@ func (m *Middleware) clientKey(r *http.Request) string {
peer, err := netip.ParseAddr(ipFromHostPort(r.RemoteAddr)) peer, err := netip.ParseAddr(ipFromHostPort(r.RemoteAddr))
if err != nil { if err != nil {
// Not an address we can reason about; key on the raw // Not an address we can reason about; key on the raw
// value, the most specific identity left. On a // value rather than collapsing such peers into one
// Unix-socket listener every peer carries the same // shared bucket.
// RemoteAddr and so shares one bucket, which is the
// fail-closed direction.
return r.RemoteAddr return r.RemoteAddr
} }
@@ -183,31 +142,6 @@ func (m *Middleware) tooManyRequests(
} }
} }
// floodTooManyRequests returns the 429 handler for a limiter whose
// rejections are themselves the flood: it logs at DEBUG and without
// the path, then answers with responseMessage.
//
// The aggregate receiver limiter trips exactly when one address is
// sending faster than the receiver wants to serve, so its rejection
// log is one line per request of that flood. At WARN with "path" that
// hands a client a way to write its own text into the operator's log,
// at a level that trips alerting, once per request — the log-volume
// problem this limiter exists to bound. DEBUG is off in production by
// default, so a flood costs nothing here; the path is dropped so that
// turning DEBUG on to diagnose one does not restore the problem.
//
// This limiter bounds the database work an invented path costs, not
// the number of log lines it produces: the access log in
// middleware.go still records every request, served or rejected.
func (m *Middleware) floodTooManyRequests(
logMessage, responseMessage string,
) http.HandlerFunc {
return func(w http.ResponseWriter, _ *http.Request) {
m.log.Debug(logMessage)
http.Error(w, responseMessage, http.StatusTooManyRequests)
}
}
// LoginRateLimit returns middleware that enforces per-IP rate // LoginRateLimit returns middleware that enforces per-IP rate
// limiting on login attempts using go-chi/httprate. Only POST // limiting on login attempts using go-chi/httprate. Only POST
// requests are rate-limited; GET requests (rendering the login // requests are rate-limited; GET requests (rendering the login
@@ -276,26 +210,15 @@ func (m *Middleware) postRateLimit(
} }
} }
// ReceiverRateLimit returns middleware that rate-limits the public // ReceiverRateLimit returns middleware that rate-limits the
// webhook receiver endpoint with two limits in series. // public webhook receiver endpoint per client IP per request
// // path (the path contains the entrypoint UUID, so each sender
// The inner limit is per client IP per request path: the path // is limited per entrypoint without affecting other senders or
// contains the entrypoint UUID, so each sender is limited per // other entrypoints). The limit is Config.ReceiverRateLimit
// entrypoint without affecting other senders or other entrypoints. // requests per minute. Requests over the limit receive a 429.
// It is Config.ReceiverRateLimit requests per minute. // Clients are identified by rateLimitKey.
//
// That limit alone bounds nothing in aggregate. The route pattern
// /webhook/{uuid} matches any single segment, so a client that
// invents a fresh path per request mints a fresh bucket per request
// and never refills one — and every such request still reaches the
// handler's entrypoint lookup before it 404s. The outer limit is
// therefore keyed on the client IP alone, capping what one address
// can spend across the whole route however it varies the path.
//
// Requests over either limit receive a 429. Clients are identified
// by rateLimitKey.
func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler { func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler {
perEntrypoint := httprate.Limit( return httprate.Limit(
m.params.Config.ReceiverRateLimit, m.params.Config.ReceiverRateLimit,
receiverRateInterval, receiverRateInterval,
httprate.WithKeyFuncs( httprate.WithKeyFuncs(
@@ -307,31 +230,4 @@ func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler {
"Too many requests. Please slow down.", "Too many requests. Please slow down.",
)), )),
) )
aggregate := httprate.Limit(
receiverAggregateLimit(m.params.Config.ReceiverRateLimit),
receiverRateInterval,
httprate.WithKeyFuncs(m.rateLimitKey),
httprate.WithLimitHandler(m.floodTooManyRequests(
"webhook receiver aggregate rate limit exceeded",
"Too many requests. Please slow down.",
)),
)
return func(next http.Handler) http.Handler {
return aggregate(perEntrypoint(next))
}
}
// receiverAggregateLimit is the per-IP aggregate limit derived from
// the configured per-entrypoint limit. The operator sets the latter
// and nothing bounds it from above, so the multiplication is
// saturated rather than allowed to wrap into a negative limit that
// would reject every request.
func receiverAggregateLimit(perEntrypoint int) int {
if perEntrypoint > math.MaxInt/receiverAggregateMultiplier {
return math.MaxInt
}
return perEntrypoint * receiverAggregateMultiplier
} }

View File

@@ -4,15 +4,11 @@ import (
"context" "context"
"fmt" "fmt"
"log/slog" "log/slog"
"math"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"net/netip" "net/netip"
"os" "os"
"runtime"
"strings"
"testing" "testing"
"time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/config"
@@ -572,228 +568,6 @@ func TestRateLimitKey_ChainWalkSkipsClientPrepended(t *testing.T) {
) )
} }
// TestRateLimitKey_LongChainCapsWalkAndFallsBackToPeer covers the
// hop-walk cap. A client behind the trusted proxy can pad
// X-Forwarded-For with tens of thousands of trusted-looking hops,
// which costs a walk proportional to the padding and, once the walk
// runs off the left end of the chain, reaches the entry the client
// put there. Capping the walk stops both: the key falls back to the
// peer address, so rotating the head of the chain mints no bucket,
// and the run does not scale with the chain length.
func TestRateLimitKey_LongChainCapsWalkAndFallsBackToPeer(
t *testing.T,
) {
t.Parallel()
// 50k hops is roughly 0.9 MB, within the default
// MaxHeaderBytes.
const hops = 50000
padding := strings.Repeat(", 10.0.0.2", hops-1)
start := time.Now()
assertSharedBucket(
t, trustedProxies("10.0.0.0/8"), "10.0.0.1:44444",
func(i int) map[string]string {
return map[string]string{
headerXFF: fmt.Sprintf("9.9.9.%d%s", i+1, padding),
}
},
"a padded X-Forwarded-For chain must fall back to the "+
"peer address, not reach the client-controlled entry "+
"at the head of the chain",
)
assert.Less(
t, time.Since(start), 2*time.Second,
"the capped walk must not scale with the chain length",
)
}
// TestRateLimitKey_LongChainAllocationIsBounded is the allocation
// half of the hop cap. Capping the walk still left every request
// paying for the whole header the client sent, because the chain was
// split before it was capped: about 8 MB of []string for the 1 MB a
// default MaxHeaderBytes allows, on the unauthenticated receiver.
//
// Bytes are the measurement, not allocation count: strings.Split of a
// 1 MB chain is a single allocation, so testing.AllocsPerRun scores
// it as cheap. The test is deliberately sequential — it reads
// process-wide counters, and Go runs this package's parallel tests
// only after the sequential ones finish.
//
//nolint:paralleltest // reads process-wide allocation counters
func TestRateLimitKey_LongChainAllocationIsBounded(t *testing.T) {
// 100k hops of ", 10.0.0.2" is roughly 1 MB.
const (
hops = 100000
iterations = 50
maxBytesPerCall = 4096
)
m := rateLimitMiddleware(t, &config.Config{
TrustedProxies: trustedProxies("10.0.0.0/8"),
})
req := httptest.NewRequestWithContext(
context.Background(), http.MethodPost, loginPath, nil,
)
req.RemoteAddr = "10.0.0.1:44444"
req.Header.Set(
headerXFF, "9.9.9.9"+strings.Repeat(", 10.0.0.2", hops),
)
var before, after runtime.MemStats
var key string
runtime.ReadMemStats(&before)
for range iterations {
key = middleware.ClientKeyForTest(m, req)
}
runtime.ReadMemStats(&after)
perCall := (after.TotalAlloc - before.TotalAlloc) / iterations
assert.Less(
t, perCall, uint64(maxBytesPerCall),
"a %d-byte X-Forwarded-For must not allocate in proportion "+
"to its length, but cost %d bytes per call",
len(req.Header.Get(headerXFF)), perCall,
)
assert.Equal(
t, "10.0.0.1", key,
"the padded chain must still fall back to the peer address",
)
}
// TestReceiverRateLimit_LimitsAggregateAcrossInventedPaths is the
// regression test for the per-path bucket key. The route pattern
// matches any single segment, so a client that never reuses a path
// never reuses a per-entrypoint bucket either, and its aggregate
// rate against the receiver is whatever it likes — with every
// request reaching an entrypoint lookup before it 404s. The IP-only
// aggregate limiter is what bounds that, so this must fail if the
// aggregate limiter is removed.
func TestReceiverRateLimit_LimitsAggregateAcrossInventedPaths(
t *testing.T,
) {
t.Parallel()
const (
limit = 3
ip = "6.6.6.6:1234"
)
aggregate := limit * middleware.ReceiverAggregateMultiplierConst
handler := receiverLimitedHandler(t, limit)
// Every request goes to a path this client has never used, so
// none of them shares a per-entrypoint bucket with another.
for i := range aggregate {
w := receiverPost(
handler, ip, fmt.Sprintf("/webhook/invented-%d", i),
)
assert.Equal(
t, http.StatusOK, w.Code,
"request %d to a distinct path should pass", i,
)
}
w := receiverPost(
handler, ip, fmt.Sprintf("/webhook/invented-%d", aggregate),
)
assert.Equal(
t, http.StatusTooManyRequests, w.Code,
"a client must not be able to raise its aggregate rate "+
"against /webhook/* by varying the path",
)
// The aggregate limit is still per client IP: exhausting one
// address must not throttle another.
w = receiverPost(handler, "6.6.6.7:1234", "/webhook/invented-0")
assert.Equal(
t, http.StatusOK, w.Code,
"a different client IP must not be affected",
)
}
// TestReceiverRateLimit_RejectedRequestsCountTowardAggregate pins the
// order the two limiters are chained in. The aggregate limiter has to
// be the outer one, so that it counts requests the per-entrypoint
// limiter rejects: those requests still arrive, and the aggregate
// limit exists to bound what one address can make the receiver do.
//
// One path is hammered past the per-entrypoint limit, which alone
// would leave the aggregate budget almost untouched; then a path the
// client has never used must be rejected, which only the aggregate
// limiter can do. Swap the two limiters and that last request is
// served, because the rejected ones never reached the aggregate
// limiter to be counted.
func TestReceiverRateLimit_RejectedRequestsCountTowardAggregate(
t *testing.T,
) {
t.Parallel()
const (
limit = 3
ip = "6.6.6.8:1234"
)
aggregate := limit * middleware.ReceiverAggregateMultiplierConst
handler := receiverLimitedHandler(t, limit)
// Spend the whole aggregate budget on one path. Only the first
// limit requests are served; the rest are rejected by the
// per-entrypoint limiter but still count against the aggregate.
for i := range aggregate {
w := receiverPost(handler, ip, "/webhook/exhausted")
want := http.StatusTooManyRequests
if i < limit {
want = http.StatusOK
}
assert.Equal(
t, want, w.Code,
"request %d to the exhausted path", i,
)
}
w := receiverPost(handler, ip, "/webhook/never-used")
assert.Equal(
t, http.StatusTooManyRequests, w.Code,
"requests rejected per entrypoint must still count "+
"toward the aggregate limit, so the aggregate "+
"limiter has to run first",
)
}
// TestReceiverAggregateLimit_SaturatesOnOverflow covers the derived
// aggregate limit for a configured per-entrypoint limit large enough
// that multiplying it would wrap negative, which httprate would read
// as a limit that rejects every request.
func TestReceiverAggregateLimit_SaturatesOnOverflow(t *testing.T) {
t.Parallel()
assert.Equal(
t, 1200,
middleware.ReceiverAggregateLimitForTest(120),
"the default limit scales by the multiplier",
)
assert.Equal(
t, math.MaxInt,
middleware.ReceiverAggregateLimitForTest(math.MaxInt),
"an overflowing limit saturates instead of wrapping",
)
}
// TestReceiverRateLimit_IgnoresForwardedFromUntrustedPeer proves // TestReceiverRateLimit_IgnoresForwardedFromUntrustedPeer proves
// the receiver limiter uses the same gated key function as the // the receiver limiter uses the same gated key function as the
// POST limiters. // POST limiters.