Compare commits
4 Commits
issue-102-
...
b5b3e1a926
| Author | SHA1 | Date | |
|---|---|---|---|
| b5b3e1a926 | |||
| 543005c0c2 | |||
| 9bfd033a29 | |||
| fd6397154a |
16
README.md
16
README.md
@@ -94,7 +94,7 @@ TTY detection, and security headers are always applied.
|
||||
| `METRICS_PASSWORD` | Basic auth password for `/metrics` | `""` |
|
||||
| `SENTRY_DSN` | Sentry error reporting DSN | `""` |
|
||||
| `SESSION_IDLE_TIMEOUT` | Idle session timeout (Go duration) | `24h` |
|
||||
| `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint | `120` |
|
||||
| `RECEIVER_RATE_LIMIT` | Receiver requests/minute per IP per entrypoint (10x that per IP across the route) | `120` |
|
||||
| `TRUSTED_PROXIES` | CIDRs whose forwarded headers are trusted | `""` (none) |
|
||||
|
||||
#### Trusted proxies
|
||||
@@ -851,6 +851,20 @@ legitimate webhook senders). Requests over the limit receive HTTP 429
|
||||
with a `Retry-After` header. A set-but-invalid `RECEIVER_RATE_LIMIT`
|
||||
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.
|
||||
|
||||
Requests to a `/webhook/` path that names no entrypoint are logged at
|
||||
`DEBUG` only, since the path is attacker-controlled; the request itself
|
||||
still appears in the access log.
|
||||
|
||||
Every limiter here — receiver, login, and password change — identifies
|
||||
the client the same way, through one shared key function: the
|
||||
connection's own address, unless the peer is listed in
|
||||
|
||||
59
TODO.md
59
TODO.md
@@ -1,31 +1,62 @@
|
||||
# Workflow
|
||||
|
||||
* branch (from `main`)
|
||||
* do the work in Next Step
|
||||
* move Next Step to the top of Completed Steps
|
||||
* move the top item of Future Steps into Next Step
|
||||
* commit (`TODO.md` changes in the same commit as the work)
|
||||
* merge to `main` if the branch is not protected, otherwise open a PR
|
||||
* push
|
||||
One issue per unit of work, one branch and one PR per issue:
|
||||
|
||||
* ensure a tracked issue exists with a definition of done
|
||||
* branch from `next` (never from `main`)
|
||||
* do the work; open a PR based on `next` (never on `main`)
|
||||
* pass an independent review, then the manager squash-merges into `next`
|
||||
* push; nothing stays local-only
|
||||
|
||||
`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
|
||||
|
||||
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,
|
||||
event retention (#63), the database archiving target (#43), the admin
|
||||
password change flow (#65), policy compliance (#6), pinned lint tooling
|
||||
(#55), and fail-loud configuration parsing (#80). 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.
|
||||
(#55), and fail-loud configuration parsing (#80).
|
||||
|
||||
`next` (9bfd033) holds the completed 1.0.0 milestone: every issue in it
|
||||
is closed, and it is verified green by cache-defeated container runs
|
||||
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
|
||||
|
||||
Manual event redelivery from the web UI (replay is a core promised
|
||||
capability in the README rationale).
|
||||
Tag 1.0.0 from `main` once the milestone PR merges, then repair the CI
|
||||
gate (#119) before the next cycle's work lands — a gate that can report
|
||||
success without running is the one thing every other guarantee here
|
||||
rests on.
|
||||
|
||||
# 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
|
||||
Profile settings placeholder removed, a progressive-enhancement copy
|
||||
button for the entrypoint URL, and retention form copy that states the
|
||||
|
||||
@@ -46,19 +46,8 @@ func (r *RetentionReaper) ExportStart() {
|
||||
}
|
||||
|
||||
// ExportStop stops the reaper's background loop for tests.
|
||||
func (r *RetentionReaper) ExportStop(ctx context.Context) error {
|
||||
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
|
||||
})
|
||||
func (r *RetentionReaper) ExportStop() {
|
||||
r.stop()
|
||||
}
|
||||
|
||||
// ExportSetInterval overrides the sweep interval for tests.
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/lifecycle"
|
||||
"sneak.berlin/go/webhooker/internal/logger"
|
||||
)
|
||||
|
||||
@@ -63,9 +62,8 @@ func NewRetentionReaper(
|
||||
}
|
||||
|
||||
// registerHooks wires the reaper's start and stop into the fx
|
||||
// lifecycle. The start hook's context is deliberately ignored (see
|
||||
// start for why the sweep loop must not inherit it); the stop hook's
|
||||
// context is honoured (see stop).
|
||||
// lifecycle. The start hook's context is deliberately ignored: see
|
||||
// start for why the sweep loop must not inherit it.
|
||||
func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // Not inheriting the hook context is
|
||||
@@ -75,8 +73,10 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(ctx context.Context) error {
|
||||
return r.stop(ctx)
|
||||
OnStop: func(_ context.Context) error {
|
||||
r.stop()
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -105,27 +105,15 @@ func (r *RetentionReaper) start() {
|
||||
)
|
||||
}
|
||||
|
||||
// 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 {
|
||||
func (r *RetentionReaper) stop() {
|
||||
r.log.Info("retention reaper stopping")
|
||||
|
||||
if r.cancel != nil {
|
||||
r.cancel()
|
||||
}
|
||||
|
||||
err := lifecycle.WaitForShutdown(
|
||||
ctx, r.log, "retention reaper", &r.wg,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
r.wg.Wait()
|
||||
r.log.Info("retention reaper stopped")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *RetentionReaper) run(ctx context.Context) {
|
||||
|
||||
@@ -26,13 +26,6 @@ const (
|
||||
// reaperTestRetentionDays is the retention policy the lifecycle
|
||||
// tests give their webhook.
|
||||
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
|
||||
@@ -214,59 +207,3 @@ func TestRetentionReaper_StopHookStopsLoop(t *testing.T) {
|
||||
"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")
|
||||
}
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"go.uber.org/fx"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/lifecycle"
|
||||
"sneak.berlin/go/webhooker/internal/logger"
|
||||
)
|
||||
|
||||
@@ -68,9 +67,10 @@ func NewArchiveSweeper(
|
||||
}
|
||||
|
||||
// registerHooks wires the sweeper's start and stop into the fx
|
||||
// lifecycle. The start hook's context is deliberately ignored
|
||||
// (see start for why the background loop must not inherit it);
|
||||
// the stop hook's context is honoured (see stop).
|
||||
// lifecycle. Both hook contexts are deliberately ignored: see
|
||||
// start for why the background loop must not inherit the start
|
||||
// hook's context, and stop for why shutdown blocks on the loop
|
||||
// rather than on the stop hook's deadline.
|
||||
func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // Not passing the hook context is
|
||||
@@ -80,8 +80,10 @@ func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(ctx context.Context) error {
|
||||
return s.stop(ctx)
|
||||
OnStop: func(_ context.Context) error {
|
||||
s.stop()
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -111,27 +113,15 @@ func (s *ArchiveSweeper) start() {
|
||||
)
|
||||
}
|
||||
|
||||
// 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 {
|
||||
func (s *ArchiveSweeper) stop() {
|
||||
s.log.Info("archive sweeper stopping")
|
||||
|
||||
if s.cancel != nil {
|
||||
s.cancel()
|
||||
}
|
||||
|
||||
err := lifecycle.WaitForShutdown(
|
||||
ctx, s.log, "archive sweeper", &s.wg,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
s.wg.Wait()
|
||||
s.log.Info("archive sweeper stopped")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *ArchiveSweeper) run(ctx context.Context) {
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
"gorm.io/gorm/clause"
|
||||
@@ -225,6 +226,17 @@ func countArchivedRows(path string) (int64, error) {
|
||||
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
|
||||
// regression test for a sweeper that never swept. fx calls
|
||||
// OnStart with a context carrying the application's start
|
||||
@@ -258,7 +270,7 @@ func TestArchiveSweeper_LoopOutlivesStartHookContext(
|
||||
|
||||
// Drive the genuine fx hooks the application registers,
|
||||
// rather than a test-only entry point.
|
||||
lc := &recordingLifecycle{}
|
||||
lc := &captureLifecycle{}
|
||||
env.sweeper.ExportRegisterHooks(lc)
|
||||
require.Len(t, lc.hooks, 1)
|
||||
|
||||
@@ -912,36 +924,7 @@ func TestArchiveSweeper_StopsCleanly(t *testing.T) {
|
||||
env.sweeper.ExportSetInterval(time.Millisecond)
|
||||
env.sweeper.ExportStart()
|
||||
|
||||
// stop blocks on the loop's WaitGroup, so returning without
|
||||
// error proves the loop observed the cancellation and exited
|
||||
// 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")
|
||||
// stop blocks on the loop's WaitGroup, so returning at all
|
||||
// proves the loop observed the cancellation and exited.
|
||||
env.sweeper.ExportStop()
|
||||
}
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"go.uber.org/fx"
|
||||
"gorm.io/gorm"
|
||||
"sneak.berlin/go/webhooker/internal/database"
|
||||
"sneak.berlin/go/webhooker/internal/lifecycle"
|
||||
"sneak.berlin/go/webhooker/internal/logger"
|
||||
)
|
||||
|
||||
@@ -235,9 +234,8 @@ func (e *Engine) ScheduleRetry(
|
||||
}
|
||||
|
||||
// registerHooks wires the engine's start and stop into the fx
|
||||
// lifecycle. The start hook's context is deliberately ignored
|
||||
// (see start for why the worker pool must not inherit it); the
|
||||
// stop hook's context is honoured (see stop).
|
||||
// lifecycle. The start hook's context is deliberately ignored:
|
||||
// see start for why the worker pool must not inherit it.
|
||||
func (e *Engine) registerHooks(lc fx.Lifecycle) {
|
||||
lc.Append(fx.Hook{
|
||||
//nolint:contextcheck // Not inheriting the hook context
|
||||
@@ -247,8 +245,10 @@ func (e *Engine) registerHooks(lc fx.Lifecycle) {
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(ctx context.Context) error {
|
||||
return e.stop(ctx)
|
||||
OnStop: func(_ context.Context) error {
|
||||
e.stop()
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
@@ -289,26 +289,11 @@ func (e *Engine) start() {
|
||||
)
|
||||
}
|
||||
|
||||
// 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 {
|
||||
func (e *Engine) stop() {
|
||||
e.log.Info("delivery engine stopping")
|
||||
|
||||
if e.cancel != nil {
|
||||
e.cancel()
|
||||
}
|
||||
|
||||
err := lifecycle.WaitForShutdown(
|
||||
ctx, e.log, "delivery engine", &e.wg,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
e.cancel()
|
||||
e.wg.Wait()
|
||||
e.log.Info("delivery engine stopped")
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) worker(ctx context.Context) {
|
||||
|
||||
@@ -501,7 +501,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
|
||||
require.NoError(t, s.Engine.ExportStop(context.Background()))
|
||||
s.Engine.ExportStop()
|
||||
}
|
||||
|
||||
// iWaitForDelivered polls until the delivery reaches the
|
||||
@@ -567,7 +567,7 @@ func TestWorkerLifecycle_ProcessesRetryChannel(
|
||||
|
||||
iWaitForDelivered(t, s.WebhookDB, d.ID)
|
||||
|
||||
require.NoError(t, s.Engine.ExportStop(context.Background()))
|
||||
s.Engine.ExportStop()
|
||||
}
|
||||
|
||||
// --- processDelivery: unknown target type ---
|
||||
|
||||
@@ -27,13 +27,6 @@ const (
|
||||
// and a ready deliveryCh are chosen between at random and a
|
||||
// doomed pool still delivers.
|
||||
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
|
||||
@@ -47,44 +40,6 @@ func (l *recordingLifecycle) Append(h fx.Hook) {
|
||||
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
|
||||
// registers for the engine, handing OnStart a context that is
|
||||
// already done, and returns only once a pool that inherited that
|
||||
@@ -242,30 +197,3 @@ func TestEngine_StopHookStopsWorkers(t *testing.T) {
|
||||
"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")
|
||||
}
|
||||
|
||||
@@ -216,19 +216,8 @@ func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
|
||||
}
|
||||
|
||||
// ExportStop exposes stop for testing.
|
||||
func (e *Engine) ExportStop(ctx context.Context) error {
|
||||
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
|
||||
})
|
||||
func (e *Engine) ExportStop() {
|
||||
e.stop()
|
||||
}
|
||||
|
||||
// ExportDeliveryCh returns the delivery channel.
|
||||
@@ -529,19 +518,8 @@ func (s *ArchiveSweeper) ExportRegisterHooks(lc fx.Lifecycle) {
|
||||
}
|
||||
|
||||
// ExportStop stops the sweeper's background loop for tests.
|
||||
func (s *ArchiveSweeper) ExportStop(ctx context.Context) error {
|
||||
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
|
||||
})
|
||||
func (s *ArchiveSweeper) ExportStop() {
|
||||
s.stop()
|
||||
}
|
||||
|
||||
// ExportSetInterval overrides the sweep interval for tests.
|
||||
|
||||
@@ -39,12 +39,6 @@ func (h *Handlers) HandleWebhook() http.HandlerFunc {
|
||||
return
|
||||
}
|
||||
|
||||
h.log.Info("webhook request received",
|
||||
"entrypoint_uuid", entrypointUUID,
|
||||
"method", r.Method,
|
||||
"remote_addr", r.RemoteAddr,
|
||||
)
|
||||
|
||||
entrypoint, ok := h.lookupEntrypoint(
|
||||
w, r, entrypointUUID,
|
||||
)
|
||||
@@ -52,6 +46,18 @@ func (h *Handlers) HandleWebhook() http.HandlerFunc {
|
||||
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 {
|
||||
http.Error(w, "Gone", http.StatusGone)
|
||||
|
||||
|
||||
@@ -1,57 +0,0 @@
|
||||
// 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(),
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -1,62 +0,0 @@
|
||||
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")
|
||||
}
|
||||
@@ -25,6 +25,11 @@ func IPFromHostPort(hp string) string {
|
||||
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.
|
||||
func IsClientTLS(r *http.Request) bool {
|
||||
return isClientTLS(r)
|
||||
@@ -36,3 +41,13 @@ const LoginRateLimitConst = loginRateLimit
|
||||
// PasswordChangeRateLimitConst exposes the
|
||||
// passwordChangeRateLimit constant.
|
||||
const PasswordChangeRateLimitConst = passwordChangeRateLimit
|
||||
|
||||
// ReceiverAggregateMultiplierConst exposes the
|
||||
// receiverAggregateMultiplier constant.
|
||||
const ReceiverAggregateMultiplierConst = receiverAggregateMultiplier
|
||||
|
||||
// ReceiverAggregateLimitForTest exposes receiverAggregateLimit for
|
||||
// testing.
|
||||
func ReceiverAggregateLimitForTest(perEntrypoint int) int {
|
||||
return receiverAggregateLimit(perEntrypoint)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"math"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"slices"
|
||||
@@ -32,6 +33,21 @@ const (
|
||||
// receiver rate limit. The configured limit is expressed in
|
||||
// requests per 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
|
||||
@@ -70,26 +86,49 @@ func (m *Middleware) isTrustedProxy(addr netip.Addr) bool {
|
||||
// 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
|
||||
// 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(
|
||||
r *http.Request,
|
||||
) (netip.Addr, bool) {
|
||||
hops := strings.Split(
|
||||
strings.Join(r.Header.Values("X-Forwarded-For"), ","), ",",
|
||||
)
|
||||
seen := 0
|
||||
|
||||
for _, hop := range slices.Backward(hops) {
|
||||
hop = strings.TrimSpace(hop)
|
||||
if hop == "" {
|
||||
continue
|
||||
}
|
||||
for _, value := range slices.Backward(
|
||||
r.Header.Values("X-Forwarded-For"),
|
||||
) {
|
||||
for last := false; !last && seen < maxForwardedHops; seen++ {
|
||||
hop := value
|
||||
|
||||
addr, err := netip.ParseAddr(hop)
|
||||
if err != nil {
|
||||
return netip.Addr{}, false
|
||||
}
|
||||
comma := strings.LastIndexByte(value, ',')
|
||||
if comma < 0 {
|
||||
last = true
|
||||
} else {
|
||||
hop, value = value[comma+1:], value[:comma]
|
||||
}
|
||||
|
||||
if addr = normalizeAddr(addr); !m.isTrustedProxy(addr) {
|
||||
return addr, true
|
||||
hop = strings.TrimSpace(hop)
|
||||
if hop == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
addr, err := netip.ParseAddr(hop)
|
||||
if err != nil {
|
||||
return netip.Addr{}, false
|
||||
}
|
||||
|
||||
if addr = normalizeAddr(addr); !m.isTrustedProxy(addr) {
|
||||
return addr, true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -113,8 +152,10 @@ func (m *Middleware) clientKey(r *http.Request) string {
|
||||
peer, err := netip.ParseAddr(ipFromHostPort(r.RemoteAddr))
|
||||
if err != nil {
|
||||
// Not an address we can reason about; key on the raw
|
||||
// value rather than collapsing such peers into one
|
||||
// shared bucket.
|
||||
// value, the most specific identity left. On a
|
||||
// Unix-socket listener every peer carries the same
|
||||
// RemoteAddr and so shares one bucket, which is the
|
||||
// fail-closed direction.
|
||||
return r.RemoteAddr
|
||||
}
|
||||
|
||||
@@ -210,15 +251,26 @@ func (m *Middleware) postRateLimit(
|
||||
}
|
||||
}
|
||||
|
||||
// ReceiverRateLimit returns middleware that rate-limits the
|
||||
// public webhook receiver endpoint per client IP per request
|
||||
// path (the path contains the entrypoint UUID, so each sender
|
||||
// is limited per entrypoint without affecting other senders or
|
||||
// other entrypoints). The limit is Config.ReceiverRateLimit
|
||||
// requests per minute. Requests over the limit receive a 429.
|
||||
// Clients are identified by rateLimitKey.
|
||||
// ReceiverRateLimit returns middleware that rate-limits the public
|
||||
// webhook receiver endpoint with two limits in series.
|
||||
//
|
||||
// The inner limit is per client IP per request path: the path
|
||||
// contains the entrypoint UUID, so each sender is limited per
|
||||
// entrypoint without affecting other senders or other entrypoints.
|
||||
// It is Config.ReceiverRateLimit requests per minute.
|
||||
//
|
||||
// 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 {
|
||||
return httprate.Limit(
|
||||
perEntrypoint := httprate.Limit(
|
||||
m.params.Config.ReceiverRateLimit,
|
||||
receiverRateInterval,
|
||||
httprate.WithKeyFuncs(
|
||||
@@ -230,4 +282,31 @@ func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler {
|
||||
"Too many requests. Please slow down.",
|
||||
)),
|
||||
)
|
||||
|
||||
aggregate := httprate.Limit(
|
||||
receiverAggregateLimit(m.params.Config.ReceiverRateLimit),
|
||||
receiverRateInterval,
|
||||
httprate.WithKeyFuncs(m.rateLimitKey),
|
||||
httprate.WithLimitHandler(m.tooManyRequests(
|
||||
"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
|
||||
}
|
||||
|
||||
@@ -4,11 +4,15 @@ import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"math"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"sneak.berlin/go/webhooker/internal/config"
|
||||
@@ -568,6 +572,176 @@ 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",
|
||||
)
|
||||
}
|
||||
|
||||
// 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
|
||||
// the receiver limiter uses the same gated key function as the
|
||||
// POST limiters.
|
||||
|
||||
Reference in New Issue
Block a user