9 Commits

Author SHA1 Message Date
d8f9d149b5 Warn when TRUSTED_PROXIES is empty in production (closes #149)
All checks were successful
check / check (push) Successful in 2m40s
With no trusted proxies configured, every client behind the reverse proxy production requires shares one rate-limit bucket per limit, so five POSTs per minute from anywhere holds the login limit full and denies the admin login until restart. The default is still correct; it was the consequence that was invisible. Startup now warns, and the README no longer claims the login limit is per-IP unconditionally.
2026-08-12 13:49:39 +02:00
339548d794 Record the last four milestone units in TODO.md
All checks were successful
check / check (push) Successful in 6s
Adds Completed Steps for the receiver aggregate rate limit, the documentation accuracy pass, the CI gate repair and the RETENTION_SWEEP_INTERVAL bound. Drops the commit hash that pinned the Status paragraph to a specific next head, and rewrites Next Step now that the gate repair it named has landed.
2026-08-12 13:21:38 +02:00
95161c7768 Bound the receiver rate limit per client IP across /webhook/* (closes #139)
All checks were successful
check / check (push) Superseded by a newer commit; never tested
The receiver limiter keyed on the request path, and /webhook/{uuid} matches any single segment, so a client minted a fresh bucket per invented path and had unlimited aggregate rate against the only unauthenticated endpoint. An outer limiter keyed on the client address alone now bounds that, chained in front of the unchanged per-entrypoint limiter. Its rejections log at DEBUG without the path, and the README states what each limit does and does not bound.
2026-08-12 13:19:43 +02:00
0e397b3174 Correct release-blocking documentation inaccuracies (closes #141)
All checks were successful
check / check (push) Superseded by a newer commit; never tested
The README env table was missing RETENTION_SWEEP_INTERVAL, TODO.md omitted five landed units, and three passages sold manual redelivery in the present tense when nothing implements it. The same false claim was corrected in the doc comment on failUnretryableRetry, which was its source text. Also removes a console.log from the shipped static asset.
2026-08-12 13:15:04 +02:00
be576096aa Make the CI gate execute the checks it reports on (closes #119)
All checks were successful
check / check (push) Successful in 2m52s
The workflow writes a build-context fingerprint before calling script/cibuild, so a code commit invalidates the COPY layer of the lint and builder stages and the checks really run, while a docs-only commit still replays from cache. A superseding run also rewrites the exact failure/Has been cancelled status left on commits that were never tested to skipped, so cancellation no longer reads as red. script/cibuild itself is untouched.
2026-08-12 13:00:51 +02:00
3941f0b0ff Require a positive RETENTION_SWEEP_INTERVAL (closes #140)
All checks were successful
check / check (push) Successful in 6s
A non-positive value reached time.NewTicker in the retention reaper and the archive sweeper, panicking both goroutines after startup had already reported success. envPositiveDuration now rejects it in loadFromEnv, matching how PORT and RECEIVER_RATE_LIMIT fail. SESSION_IDLE_TIMEOUT keeps treating non-positive as disabled, which is guarded at every use site.
2026-08-12 12:46:39 +02:00
543005c0c2 Update TODO.md for the completed 1.0.0 milestone
All checks were successful
check / check (push) Successful in 5s
Records the trusted-proxy gating, hop cap and bounded scan, and corrects the Workflow section, which still described branching from main and committing TODO.md alongside the work.
2026-08-12 12:20:34 +02:00
9bfd033a29 Bound X-Forwarded-For scanning allocation to the hop cap (closes #133)
All checks were successful
check / check (push) Superseded by a newer commit; never tested
forwardedClientAddr now walks the header values in reverse with strings.LastIndexByte instead of joining and splitting, so allocation is bounded by the 64-hop cap rather than by header length: 1.6 MB per call becomes 16 bytes for a 1 MB chain. Semantics are unchanged, verified by differential testing against the previous implementation.
2026-08-12 12:19:14 +02:00
fd6397154a Cap the X-Forwarded-For hop walk at 64 entries (closes #124)
All checks were successful
check / check (push) Successful in 7s
The walk now keeps only the rightmost 64 hops, so an attacker-supplied chain cannot burn unbounded CPU in the rate-limit key function. Running off the end of the truncated slice falls back to the peer address, the same fail-closed direction the rest of the function takes. Also corrects the unparseable-RemoteAddr comment, which overclaimed about Unix-socket peers.
2026-08-12 11:53:48 +02:00
25 changed files with 928 additions and 465 deletions

View File

@@ -1,3 +1,6 @@
# .ci-fingerprint is deliberately NOT excluded: it is the CI cache barrier
# that keeps the check stages from replaying a cached pass. See the lint
# stage of the Dockerfile.
.git/ .git/
bin/ bin/
*.md *.md

View File

@@ -11,5 +11,53 @@ jobs:
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 2024-10-23 uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 2024-10-23
with:
# The fingerprint step below needs history to find the last commit
# that touched the Docker build context.
fetch-depth: 0
- name: Neutralize superseded run statuses
# Gitea cancels the in-flight run when another commit is pushed to the
# same branch and records the cancellation as `failure`, so a commit
# that was never tested reads red. The cancellation is unconditional
# server-side for push events and cannot be disabled from a workflow
# file, so the superseding run rewrites those statuses to `skipped`.
# Only the exact cancellation status is touched; a real failure is
# left alone.
env:
GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }}
run: |
set -eu
api="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}"
ctx='check / check (push)'
for sha in $(git rev-list --max-count=20 "${GITHUB_SHA}^" || true); do
latest="$(curl -sf "${api}/commits/${sha}/status" | jq -r \
--arg c "$ctx" \
'[.statuses[] | select(.context == $c)][0] // empty
| "\(.status)|\(.description)"')" || continue
[ "$latest" = 'failure|Has been cancelled' ] || continue
curl -sf -X POST "${api}/statuses/${sha}" \
-H "Authorization: token ${GITEA_TOKEN}" \
-H 'Content-Type: application/json' \
-d "$(jq -nc --arg c "$ctx" '{
context: $c,
state: "skipped",
description: "Superseded by a newer commit; never tested"
}')" >/dev/null
echo "neutralized superseded status on ${sha}"
done
- name: Fingerprint the build context
# `.dockerignore` keeps docs out of the build context, so a docs-only
# commit legitimately replays the whole image from cache and stays
# cheap. Every other commit writes a new fingerprint into the context,
# which invalidates the `COPY . .` layer of both check stages: a
# commit that was never linted, formatted-checked, tested and built
# cannot report success from cache.
run: |
set -eu
fp="$(git log -1 --format=%H -- . ':!*.md' ':!LICENSE' ':!.editorconfig')"
printf '%s\n' "${fp:-$GITHUB_SHA}" > .ci-fingerprint
- name: Build Docker image (runs make check) - name: Build Docker image (runs make check)
run: script/cibuild run: script/cibuild

3
.gitignore vendored
View File

@@ -42,3 +42,6 @@ data/
# Temporary files # Temporary files
tmp/ tmp/
temp/ temp/
# CI cache barrier, written into the build context by the check workflow
.ci-fingerprint

View File

@@ -12,7 +12,11 @@ WORKDIR /src
COPY go.mod go.sum ./ COPY go.mod go.sum ./
RUN go mod download RUN go mod download
# Copy source code # Copy source code. In CI the context also carries .ci-fingerprint, whose
# value changes with every commit that touches the build context (see
# .gitea/workflows/check.yml). That invalidates this layer, so the checks
# below cannot report success by replaying a cached pass. Do not add it to
# .dockerignore.
COPY . . COPY . .
# Run formatting check and linter # Run formatting check and linter
@@ -36,7 +40,8 @@ WORKDIR /build
COPY go.mod go.sum ./ COPY go.mod go.sum ./
RUN go mod download RUN go mod download
# Copy source code # Copy source code, including the .ci-fingerprint cache barrier described in
# the lint stage above.
COPY . . COPY . .
# Run tests and build # Run tests and build

133
README.md
View File

@@ -93,9 +93,10 @@ 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` | How often the retention reaper and archive sweeper run (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 | `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` | CIDRs whose forwarded headers are trusted (unset: all clients behind a proxy share one rate-limit bucket) | `""` (none) |
#### Trusted proxies #### Trusted proxies
@@ -114,6 +115,23 @@ or draining someone else's. Set it to the address of your reverse
proxy, and to nothing wider. A set but unparseable value aborts proxy, and to nothing wider. A set but unparseable value aborts
startup. startup.
That default is safe against forged headers, but leaving it unset in
production has a cost you must know about. Production runs behind a
TLS-terminating reverse proxy, so with `TRUSTED_PROXIES` unset every
request keys on the proxy's own address and all clients share a single
bucket per limit. For the login and password-change limits that is a
denial of service anyone can perform: a steady five POSTs per minute
from any address on the internet keeps the shared login bucket full,
and the operator's own login then returns HTTP 429 for as long as the
trickle continues. There is no second administrative path and no
bypass. Restarting the service clears the in-memory buckets, but a
sustained trickle re-locks them immediately.
The remedy is to set `TRUSTED_PROXIES` to your reverse proxy's
address, which restores per-client buckets. webhooker logs a warning
at startup when `WEBHOOKER_ENVIRONMENT=prod` and `TRUSTED_PROXIES` is
empty. See [Rate Limiting](#rate-limiting) for what each limit shares.
`X-Real-IP` and `True-Client-IP` are **never** read, from any peer. `X-Real-IP` and `True-Client-IP` are **never** read, from any peer.
Reverse proxies append to `X-Forwarded-For` but forward other client Reverse proxies append to `X-Forwarded-For` but forward other client
headers verbatim, so a single-valued header is client-controlled even headers verbatim, so a single-valued header is client-controlled even
@@ -173,8 +191,12 @@ 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, and every entry in `RECEIVER_RATE_LIMIT` must be at least 1,
`TRUSTED_PROXIES` must be a CIDR block or a bare IP address. `RETENTION_SWEEP_INTERVAL` must be greater than zero (it is a ticker
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`,
@@ -260,9 +282,10 @@ webhooker solves this by acting as a durable intermediary:
targets simultaneously. This enables patterns like forwarding a targets simultaneously. This enables patterns like forwarding a
GitHub webhook to both a deployment service and a Slack channel. GitHub webhook to both a deployment service and a Slack channel.
5. **Replay** — Stored events can be manually redelivered for debugging 5. **Replay** (not yet implemented) — Every received event is stored in
or testing, without requiring the original sender to fire the webhook full, which is what manual redelivery for debugging or testing will
again. be built on. No redelivery exists today, in the web UI or the API;
see [TODO.md](TODO.md).
### Use Cases ### Use Cases
@@ -272,6 +295,7 @@ webhooker solves this by acting as a durable intermediary:
size, and delivery performance size, and delivery performance
- **Debugging** and introspection of webhook payloads in the web UI - **Debugging** and introspection of webhook payloads in the web UI
- **Replay** of webhook events for application testing and development - **Replay** of webhook events for application testing and development
(planned; not yet implemented)
- **Fan-out** delivery of a single webhook to multiple downstream - **Fan-out** delivery of a single webhook to multiple downstream
targets targets
- **High-availability ingestion** for delivery to less reliable backend - **High-availability ingestion** for delivery to less reliable backend
@@ -497,7 +521,7 @@ A programmatic access credential for API authentication.
#### Event #### Event
A captured incoming webhook request. Stores the complete HTTP request A captured incoming webhook request. Stores the complete HTTP request
data for replay and auditing. data for auditing and for the planned replay capability.
| Field | Type | Description | | Field | Type | Description |
| -------------- | ------ | ----------- | | -------------- | ------ | ----------- |
@@ -779,9 +803,10 @@ unknown) one while one of its deliveries is still `retrying`, both
recovery paths above terminally mark that delivery `failed` and record a recovery paths above terminally mark that delivery `failed` and record a
`DeliveryResult` naming the current target type as the reason, logging it `DeliveryResult` naming the current target type as the reason, logging it
at warn level. The delivery is not re-dispatched under the new type — the at warn level. The delivery is not re-dispatched under the new type — the
operator never asked for that delivery — and the event itself remains operator never asked for that delivery — and while the event itself
stored in the per-webhook event database, so it can be redelivered remains stored in the per-webhook event database, there is no way to
manually. redeliver it: manual redelivery is planned, not implemented (see
[TODO.md](TODO.md)).
### Circuit Breaker (HTTP Targets with Retries) ### Circuit Breaker (HTTP Targets with Retries)
@@ -851,15 +876,60 @@ 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; log volume it caps rather than eliminates. A path that names no
entrypoint is recorded by the handler at `DEBUG`, and the aggregate
limiter logs its own rejections at `DEBUG` and without the path, so
neither appears at all under the default level. The per-entrypoint
limiter is the loud one: it still logs every rejection at `WARN` with
the request path, which on this route is attacker-controlled text. A
client hammering a single invented path is served `RECEIVER_RATE_LIMIT`
requests and has the rest of its aggregate budget rejected there, so
the aggregate limit is what bounds those `WARN` lines — to under ten
times `RECEIVER_RATE_LIMIT` per minute per client IP, 1080 at the
defaults, where before it there was no bound at all. The access log is
bounded by neither limit: every request is recorded once at `INFO` with
its full URL, served or rejected alike.
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
`TRUSTED_PROXIES`, in which case the forwarded client address is used `TRUSTED_PROXIES`, in which case the forwarded client address is used
instead. See [Trusted proxies](#trusted-proxies). Deployed without that 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. Set `TRUSTED_PROXIES` to the
to be wrong in: set `TRUSTED_PROXIES` to the proxy's address to get proxy's address to get per-client limits back. What the shared bucket
per-client limits back. costs is not the same for every limiter, and the two cases pull in
opposite directions:
- For the **receiver** limits it costs throughput, which is the safe
direction to be wrong in: sharing can only make a limit bind sooner,
never let a sender past it. It matters more for the 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`.
- For the **login and password-change** limits it costs availability of
the only administrative path, which is not safe at all. Five POSTs
per minute from any address on the internet keeps the single shared
login bucket full, and the operator's own login returns HTTP 429 for
as long as that trickle continues. A restart clears the in-memory
buckets and a resumed trickle re-locks them. Production deployments
must set `TRUSTED_PROXIES`; webhooker warns at startup when it is
empty in `prod`.
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
@@ -1078,8 +1148,11 @@ downstream at form-parse time.
(custom HTTP transport with SSRF-safe dialer that validates resolved (custom HTTP transport with SSRF-safe dialer that validates resolved
IPs before connecting, preventing DNS rebinding attacks) IPs before connecting, preventing DNS rebinding attacks)
- **Login rate limiting** via [go-chi/httprate](https://github.com/go-chi/httprate): - **Login rate limiting** via [go-chi/httprate](https://github.com/go-chi/httprate):
per-IP sliding-window rate limiter on the login endpoint (5 POST sliding-window rate limiter on the login endpoint, 5 POST attempts
attempts per minute per IP) to prevent brute-force attacks per minute per bucket, to slow brute-force attacks. The bucket is per
client IP only when `TRUSTED_PROXIES` names the reverse proxy;
unset, every client shares one bucket and the login becomes remotely
deniable (see [Rate Limiting](#rate-limiting))
- Prometheus metrics behind basic auth - Prometheus metrics behind basic auth
- Static assets embedded in binary (no filesystem access needed at - Static assets embedded in binary (no filesystem access needed at
runtime) runtime)
@@ -1104,6 +1177,34 @@ binary is statically linked and runs on Alpine.
`docker build .` is the CI gate — if it passes, the code is formatted, `docker build .` is the CI gate — if it passes, the code is formatted,
linted, tested, and compiled. linted, tested, and compiled.
#### CI gate honesty
A layer cache lets `docker build .` exit 0 in seconds with the lint and
test stages replayed rather than executed, which would make a green
check meaningless. The `check` workflow therefore writes
`.ci-fingerprint` into the build context before building. Its value is
the hash of the last commit that touched the build context, so:
- Any commit that changes code (including a squash merge whose tree
matches an already-built branch) gets a new fingerprint, invalidates
the `COPY . .` layer of both check stages, and really runs
`make fmt-check`, `make lint`, `make test`, and `make build`. A run
that reports success ran them.
- A docs-only commit leaves the fingerprint unchanged — `.dockerignore`
excludes `*.md` and `LICENSE` from the context anyway — so the image
replays from cache and costs seconds.
The module download layer sits above `COPY . .` and stays cached either
way.
The workflow's first step covers a second way the gate lied: Gitea
cancels an in-flight run when a newer commit lands on the same branch
and records that cancellation as a `failure` status, marking a commit
red that was never tested. Cancellation is unconditional server-side for
push events, so the superseding run rewrites the exact
`Has been cancelled` status to `skipped`. Genuine failures are never
touched.
## TODO ## TODO
See [TODO.md](TODO.md). See [TODO.md](TODO.md).

120
TODO.md
View File

@@ -1,35 +1,124 @@
# Workflow # Workflow
* branch (from `main`) One issue per unit of work, one branch and one PR per issue:
* do the work in Next Step
* move Next Step to the top of Completed Steps * ensure a tracked issue exists with a definition of done
* move the top item of Future Steps into Next Step * branch from `next` (never from `main`)
* commit (`TODO.md` changes in the same commit as the work) * do the work; open a PR based on `next` (never on `main`)
* merge to `main` if the branch is not protected, otherwise open a PR * pass an independent review, then the manager squash-merges into `next`
* push * 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 # 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). Note: TODO.md was (#55), and fail-loud configuration parsing (#80).
deliberately deleted from this repo in f9a9569 (2026-03-01, #6); its
content was folded into the README TODO section, which this draft `next` holds the completed 1.0.0 milestone: every issue in it is closed,
reconstructs as of 2026-07-06. and it is verified green both by CI and by cache-defeated container
runs. The two were only made to mean the same thing this cycle — before
#119, a warm layer cache let the gate report success without executing
anything, and replayed the previous build's console log so the lie
looked like a real run. 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
Manual event redelivery from the web UI (replay is a core promised Merge the milestone PR to `main` and tag 1.0.0 from it.
capability in the README rationale).
Two decisions are open and belong to the owner, neither blocking the
tag: #115 (mask the `http` target's destination URL, implemented
speculatively and awaiting a yes or no) and #125 (whether IPv6
rate-limit keys should bucket by `/64`).
# Completed Steps # Completed Steps
- 2026-08-12 Bound the receiver rate limit per client IP across the
whole `/webhook/*` route. The existing limiter keyed on the request
path and `/webhook/{uuid}` matches any single segment, so a client
that invented a fresh path per request minted a fresh bucket per
request: the limit on the only unauthenticated endpoint bounded
nothing in aggregate, and every request still cost an entrypoint
lookup before it 404ed. An outer limiter keyed on the client address
alone now bounds that, chained in front of the unchanged
per-entrypoint limiter (#139)
- 2026-08-12 Correct release-blocking documentation inaccuracies: the
README promised manual redelivery in the present tense in three
places when nothing implements it (the same false claim also sat in
the doc comment that was its source text), the env table omitted
`RETENTION_SWEEP_INTERVAL`, and `TODO.md` itself omitted five landed
units (#141)
- 2026-08-12 Make the CI gate execute the checks it reports on. The
workflow now writes a build-context fingerprint before calling
`script/cibuild`, so a code commit invalidates the `COPY` layer of
the lint and builder stages while a docs-only commit still replays
from cache; a superseding run also rewrites the `failure` status
Gitea leaves on commits it cancelled and never tested. Verified by
pushing a deliberately broken test and watching CI go red (#119)
- 2026-08-12 Require a positive `RETENTION_SWEEP_INTERVAL`: a
non-positive value reached `time.NewTicker` in both the retention
reaper and the archive sweeper, panicking two goroutines with no
recover after startup had already reported success (#140)
- 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
actual policy (deletion by the reaper, 0 retains forever) (#57) actual policy (deletion by the reaper, 0 retains forever) (#57)
- 2026-08-11 Mask the webhook credential in delivery errors and logs:
Go embeds the request URL in `*url.Error`, so every transport failure
persisted the full Slack webhook URL into the per-webhook event
database via `DeliveryResult.Error`, a field a future REST API would
have served. `maskURLError` drops path, query and userinfo while
preserving the wrapped cause, so `errors.Is`/`As` and `Timeout()`
still work and DNS, TLS and timeout failures still read differently
(#118)
- 2026-08-11 Rate-limit the public webhook receiver endpoint
(`RECEIVER_RATE_LIMIT`, default 120/min), keyed on client IP plus
entrypoint path so one entrypoint cannot exhaust another's budget;
over-limit requests get 429 with `Retry-After`. It was the one
unauthenticated, internet-facing endpoint with no limit at all (#64)
- 2026-08-11 Enforce the body size limit before CSRF parses the form:
`MaxBodySize` is now first in all four form-parsing route groups, so
an oversized request is rejected with 413 instead of being read in
full by the CSRF middleware before any cap applied (#90)
- 2026-08-11 Mask target config on the source detail page, which
rendered the stored blob verbatim and so exposed the Slack
incoming-webhook URL — a bearer credential that cannot be revoked
per-holder. Config reaches the template only as a `TargetView` of
labelled fields, and header values are rendered as a count (#113)
- 2026-08-11 Allow `retention_days` of 0 to mean retain forever, via a
sentinel written in `BeforeSave` so the GORM column default cannot
win the race. Also bounds the reaper's cutoff arithmetic: day counts
above 106751 overflowed `time.Duration` and wrapped the cutoff into
the future, where every row matched and the sweep deleted everything
(#79)
- 2026-08-09 Inactivity-based session timeout: sliding idle expiry - 2026-08-09 Inactivity-based session timeout: sliding idle expiry
(`SESSION_IDLE_TIMEOUT`, default `24h`) refreshed on authenticated (`SESSION_IDLE_TIMEOUT`, default `24h`) refreshed on authenticated
requests, with the 7-day absolute cap kept as an independent requests, with the 7-day absolute cap kept as an independent
@@ -84,6 +173,9 @@ capability in the README rationale).
# Future Steps # Future Steps
- Manual event redelivery from the web UI — the "Replay" capability the
README describes as planned. No redelivery code exists anywhere in the
tree; events are stored in full, which is all it would be built on
- Delivery status and retry management UI - Delivery status and retry management UI
- Per-webhook rate limiting in the receiver handler (per-webhook config - Per-webhook rate limiting in the receiver handler (per-webhook config
plus handler enforcement; global limits must not apply to receiver plus handler enforcement; global limits must not apply to receiver

View File

@@ -92,6 +92,7 @@ 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
@@ -235,6 +236,34 @@ 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).
@@ -346,7 +375,7 @@ func loadFromEnv() (*Config, error) {
return nil, err return nil, err
} }
retentionSweepInterval, err := envDuration( retentionSweepInterval, err := envPositiveDuration(
"RETENTION_SWEEP_INTERVAL", "RETENTION_SWEEP_INTERVAL",
defaultRetentionSweepInterval, defaultRetentionSweepInterval,
) )
@@ -354,6 +383,8 @@ 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,
@@ -391,6 +422,38 @@ func loadFromEnv() (*Config, error) {
}, nil }, nil
} }
// warnSharedRateLimitBucket logs a startup warning when a production
// deployment leaves TRUSTED_PROXIES empty.
//
// With no trusted proxies every rate limiter keys on the connecting
// peer's address. A production deployment is required to run behind a
// TLS-terminating reverse proxy, and the peer is then that proxy for
// every request, so all clients share one bucket per limiter. The
// login limiter's bucket is the dangerous one: any remote client can
// keep it full, which denies the only administrative login to
// everyone until the process restarts.
//
// The default of trusting nobody is deliberate — trusting forwarded
// headers from arbitrary peers lets any client choose its own bucket —
// so this warns rather than failing startup or changing the key.
func (c *Config) warnSharedRateLimitBucket(log *slog.Logger) {
if !c.IsProd() || len(c.TrustedProxies) > 0 {
return
}
log.Warn(
"TRUSTED_PROXIES is empty: rate limits key on the "+
"connecting peer, so behind the reverse proxy a "+
"production deployment runs behind, every client "+
"shares one bucket per limit. Any remote client can "+
"then keep the login limit full and deny the admin "+
"login, the only administrative path, until restart. "+
"Set TRUSTED_PROXIES to your reverse proxy's address.",
"environment", c.Environment,
"trustedProxies", len(c.TrustedProxies),
)
}
// New creates a Config by reading environment variables. // New creates a Config by reading environment variables.
// //
//nolint:revive // lc parameter is required by fx even if unused. //nolint:revive // lc parameter is required by fx even if unused.
@@ -435,5 +498,7 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
s.MetricsUsername != "" && s.MetricsPassword != "", s.MetricsUsername != "" && s.MetricsPassword != "",
) )
s.warnSharedRateLimitBucket(log)
return s, nil return s, nil
} }

View File

@@ -1,6 +1,8 @@
package config_test package config_test
import ( import (
"bytes"
"log/slog"
"os" "os"
"testing" "testing"
"time" "time"
@@ -139,7 +141,11 @@ func TestRetentionSweepInterval(t *testing.T) {
set bool set bool
value string value string
expectError bool expectError bool
expected time.Duration // 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
}{ }{
{ {
name: caseUnsetUsesDefault, name: caseUnsetUsesDefault,
@@ -158,6 +164,24 @@ 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 {
@@ -175,7 +199,9 @@ func TestRetentionSweepInterval(t *testing.T) {
} }
if tt.expectError { if tt.expectError {
expectStartupError(t) expectStartupErrorFor(
t, "RETENTION_SWEEP_INTERVAL", tt.sentinel,
)
} else { } else {
testRetentionSweepIntervalSuccess(t, tt.expected) testRetentionSweepIntervalSuccess(t, tt.expected)
} }
@@ -281,6 +307,22 @@ 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 {
@@ -584,3 +626,79 @@ func testTrustedProxiesSuccess(
assert.Equal(t, expected, got) assert.Equal(t, expected, got)
} }
// TestSharedRateLimitBucketWarning covers the startup warning that
// tells an operator their production deployment shares one rate-limit
// bucket between every client, which makes the admin login remotely
// deniable. It must fire when TRUSTED_PROXIES is empty in production
// and stay quiet otherwise.
func TestSharedRateLimitBucketWarning(t *testing.T) {
tests := []struct {
name string
environment string
trustedProxies string
expectWarning bool
}{
{
name: "prod without trusted proxies warns",
environment: config.EnvironmentProd,
expectWarning: true,
},
{
name: "prod with trusted proxies is quiet",
environment: config.EnvironmentProd,
trustedProxies: cidrPrivateV4,
expectWarning: false,
},
{
// Development is not required to run behind a
// reverse proxy, so the shared bucket the warning
// describes is not the expected shape there.
name: "dev without trusted proxies is quiet",
environment: config.EnvironmentDev,
expectWarning: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
// Cannot use t.Parallel() here because t.Setenv
// is incompatible with parallel subtests.
t.Setenv("WEBHOOKER_ENVIRONMENT", tt.environment)
if tt.trustedProxies == "" {
require.NoError(
t, os.Unsetenv("TRUSTED_PROXIES"),
)
} else {
t.Setenv("TRUSTED_PROXIES", tt.trustedProxies)
}
var buf bytes.Buffer
log := slog.New(slog.NewJSONHandler(
&buf, &slog.HandlerOptions{
Level: slog.LevelDebug,
},
))
require.NoError(
t,
config.WarnSharedRateLimitBucketForTest(log),
)
if !tt.expectWarning {
assert.Empty(t, buf.String())
return
}
logged := buf.String()
assert.Contains(t, logged, `"level":"WARN"`)
assert.Contains(t, logged, "TRUSTED_PROXIES")
assert.Contains(t, logged, "shares one bucket")
assert.Contains(t, logged, "deny the admin login")
})
}
}

View File

@@ -1,9 +1,26 @@
package config package config
import "log/slog"
// This file exposes the unexported environment parsing helpers to // This file exposes the unexported environment parsing helpers to
// the external config_test package so each helper can be covered by // the external config_test package so each helper can be covered by
// its own table-driven test without weakening the package API. // its own table-driven test without weakening the package API.
// WarnSharedRateLimitBucketForTest loads a Config from the current
// environment and emits its startup warnings to log. The real logger
// writes to stdout, so this lets the warning's firing condition be
// asserted against a handler the test controls.
func WarnSharedRateLimitBucketForTest(log *slog.Logger) error {
c, err := loadFromEnv()
if err != nil {
return err
}
c.warnSharedRateLimitBucket(log)
return nil
}
// EnvBoolForTest exposes envBool. // EnvBoolForTest exposes envBool.
func EnvBoolForTest(key string, defaultValue bool) (bool, error) { func EnvBoolForTest(key string, defaultValue bool) (bool, error) {
return envBool(key, defaultValue) return envBool(key, defaultValue)

View File

@@ -46,19 +46,8 @@ 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(ctx context.Context) error { func (r *RetentionReaper) ExportStop() {
return r.stop(ctx) r.stop()
}
// 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,7 +10,6 @@ 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"
) )
@@ -63,9 +62,8 @@ 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); the stop hook's // start for why the sweep loop must not inherit it.
// 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
@@ -75,8 +73,10 @@ func (r *RetentionReaper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(ctx context.Context) error { OnStop: func(_ context.Context) error {
return r.stop(ctx) r.stop()
return nil
}, },
}) })
} }
@@ -105,27 +105,15 @@ func (r *RetentionReaper) start() {
) )
} }
// stop cancels the sweep loop's context and waits for it to func (r *RetentionReaper) stop() {
// 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()
} }
err := lifecycle.WaitForShutdown( r.wg.Wait()
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,13 +26,6 @@ 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
@@ -214,59 +207,3 @@ 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,7 +10,6 @@ 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"
) )
@@ -68,9 +67,10 @@ 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. The start hook's context is deliberately ignored // lifecycle. Both hook contexts are deliberately ignored: see
// (see start for why the background loop must not inherit it); // start for why the background loop must not inherit the start
// the stop hook's context is honoured (see stop). // 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) { 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,8 +80,10 @@ func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(ctx context.Context) error { OnStop: func(_ context.Context) error {
return s.stop(ctx) s.stop()
return nil
}, },
}) })
} }
@@ -111,27 +113,15 @@ func (s *ArchiveSweeper) start() {
) )
} }
// stop cancels the sweep loop's context and waits for it to func (s *ArchiveSweeper) stop() {
// 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()
} }
err := lifecycle.WaitForShutdown( s.wg.Wait()
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,6 +14,7 @@ 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"
@@ -225,6 +226,17 @@ 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
@@ -258,7 +270,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 := &recordingLifecycle{} lc := &captureLifecycle{}
env.sweeper.ExportRegisterHooks(lc) env.sweeper.ExportRegisterHooks(lc)
require.Len(t, lc.hooks, 1) require.Len(t, lc.hooks, 1)
@@ -912,36 +924,7 @@ 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 without // stop blocks on the loop's WaitGroup, so returning at all
// error proves the loop observed the cancellation and exited // proves the loop observed the cancellation and exited.
// well inside the stop context. env.sweeper.ExportStop()
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,7 +13,6 @@ 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"
) )
@@ -235,9 +234,8 @@ 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); the // see start for why the worker pool must not inherit it.
// 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
@@ -247,8 +245,10 @@ func (e *Engine) registerHooks(lc fx.Lifecycle) {
return nil return nil
}, },
OnStop: func(ctx context.Context) error { OnStop: func(_ context.Context) error {
return e.stop(ctx) e.stop()
return nil
}, },
}) })
} }
@@ -289,26 +289,11 @@ func (e *Engine) start() {
) )
} }
// stop cancels the worker pool's context and waits for the pool func (e *Engine) stop() {
// 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")
e.cancel()
if e.cancel != nil { e.wg.Wait()
e.cancel()
}
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) {
@@ -800,9 +785,9 @@ func (e *Engine) sweepSingleRetry(
// status retrying themselves. Re-dispatching under the new type // status retrying themselves. Re-dispatching under the new type
// would be a delivery the operator never asked for, and leaving // would be a delivery the operator never asked for, and leaving
// the row retrying strands it forever, so the delivery is // the row retrying strands it forever, so the delivery is
// failed with a recorded reason and can be redelivered // failed with a recorded reason. The event stays stored, but
// manually. Logged at warn, not error: this is operator-caused // nothing redelivers it today. Logged at warn, not error: this
// state, not a system fault. // is operator-caused state, not a system fault.
func (e *Engine) failUnretryableRetry( func (e *Engine) failUnretryableRetry(
webhookDB *gorm.DB, webhookDB *gorm.DB,
webhookID string, webhookID string,

View File

@@ -501,7 +501,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
iWaitForDelivered(t, s.WebhookDB, d.ID) iWaitForDelivered(t, s.WebhookDB, d.ID)
require.NoError(t, s.Engine.ExportStop(context.Background())) s.Engine.ExportStop()
} }
// 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)
require.NoError(t, s.Engine.ExportStop(context.Background())) s.Engine.ExportStop()
} }
// --- processDelivery: unknown target type --- // --- processDelivery: unknown target type ---

View File

@@ -27,13 +27,6 @@ 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
@@ -47,44 +40,6 @@ 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
@@ -242,30 +197,3 @@ 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,19 +216,8 @@ func (e *Engine) ExportRegisterHooks(lc fx.Lifecycle) {
} }
// ExportStop exposes stop for testing. // ExportStop exposes stop for testing.
func (e *Engine) ExportStop(ctx context.Context) error { func (e *Engine) ExportStop() {
return e.stop(ctx) e.stop()
}
// 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.
@@ -529,19 +518,8 @@ 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(ctx context.Context) error { func (s *ArchiveSweeper) ExportStop() {
return s.stop(ctx) s.stop()
}
// 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,12 +39,6 @@ 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,
) )
@@ -52,6 +46,18 @@ 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

@@ -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(),
)
}
}

View File

@@ -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")
}

View File

@@ -25,6 +25,11 @@ 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)
@@ -36,3 +41,13 @@ 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,6 +1,7 @@
package middleware package middleware
import ( import (
"math"
"net/http" "net/http"
"net/netip" "net/netip"
"slices" "slices"
@@ -32,6 +33,21 @@ 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
@@ -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 // 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) {
hops := strings.Split( seen := 0
strings.Join(r.Header.Values("X-Forwarded-For"), ","), ",",
)
for _, hop := range slices.Backward(hops) { for _, value := range slices.Backward(
hop = strings.TrimSpace(hop) r.Header.Values("X-Forwarded-For"),
if hop == "" { ) {
continue for last := false; !last && seen < maxForwardedHops; seen++ {
} hop := value
addr, err := netip.ParseAddr(hop) comma := strings.LastIndexByte(value, ',')
if err != nil { if comma < 0 {
return netip.Addr{}, false last = true
} } else {
hop, value = value[comma+1:], value[:comma]
}
if addr = normalizeAddr(addr); !m.isTrustedProxy(addr) { hop = strings.TrimSpace(hop)
return addr, true 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)) 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 rather than collapsing such peers into one // value, the most specific identity left. On a
// shared bucket. // Unix-socket listener every peer carries the same
// RemoteAddr and so shares one bucket, which is the
// fail-closed direction.
return r.RemoteAddr return r.RemoteAddr
} }
@@ -130,9 +171,11 @@ func (m *Middleware) clientKey(r *http.Request) string {
return peer.String() return peer.String()
} }
// tooManyRequests returns the 429 handler shared by every limiter: // tooManyRequests returns the 429 handler used by the login,
// it logs the rejection with logMessage and answers with // password-change and per-entrypoint receiver limiters: it logs the
// responseMessage. httprate adds the Retry-After header (RFC 6585). // rejection with logMessage and answers with responseMessage.
// httprate adds the Retry-After header (RFC 6585). The aggregate
// receiver limiter uses floodTooManyRequests instead.
func (m *Middleware) tooManyRequests( func (m *Middleware) tooManyRequests(
logMessage, responseMessage string, logMessage, responseMessage string,
) http.HandlerFunc { ) http.HandlerFunc {
@@ -142,6 +185,31 @@ 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
@@ -210,15 +278,26 @@ func (m *Middleware) postRateLimit(
} }
} }
// ReceiverRateLimit returns middleware that rate-limits the // ReceiverRateLimit returns middleware that rate-limits the public
// public webhook receiver endpoint per client IP per request // webhook receiver endpoint with two limits in series.
// path (the path contains the entrypoint UUID, so each sender //
// is limited per entrypoint without affecting other senders or // The inner limit is per client IP per request path: the path
// other entrypoints). The limit is Config.ReceiverRateLimit // contains the entrypoint UUID, so each sender is limited per
// requests per minute. Requests over the limit receive a 429. // entrypoint without affecting other senders or other entrypoints.
// Clients are identified by rateLimitKey. // 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 { func (m *Middleware) ReceiverRateLimit() func(http.Handler) http.Handler {
return httprate.Limit( perEntrypoint := httprate.Limit(
m.params.Config.ReceiverRateLimit, m.params.Config.ReceiverRateLimit,
receiverRateInterval, receiverRateInterval,
httprate.WithKeyFuncs( httprate.WithKeyFuncs(
@@ -230,4 +309,31 @@ 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,11 +4,15 @@ 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"
@@ -568,6 +572,228 @@ 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.

View File

@@ -1,5 +1,4 @@
// Webhooker client-side JavaScript // Webhooker client-side JavaScript
console.log("Webhooker loaded");
// Copy-to-clipboard, as progressive enhancement. // Copy-to-clipboard, as progressive enhancement.
// //