From 55a3b5cae461add83ab65255cb0eb72525eb77e5 Mon Sep 17 00:00:00 2001 From: Quang <20378quang@gmail.com> Date: Tue, 25 Aug 2026 00:11:01 -0400 Subject: [PATCH] feat(ingestion): rate limit public form submissions per IP and per endpoint --- .env.example | 35 + README.md | 277 +++++- src/hymical_forms/api/submissions.py | 233 ++++- src/hymical_forms/config.py | 76 +- .../0004_20260824_rate_limit_counters.py | 79 ++ src/hymical_forms/models.py | 44 +- src/hymical_forms/ratelimit.py | 168 ++++ src/hymical_forms/storage.py | 128 ++- tests/integration/test_migrations_postgres.py | 132 ++- .../test_rate_limiting_postgres.py | 279 ++++++ tests/test_openapi.py | 11 + tests/test_rate_limiting.py | 856 ++++++++++++++++++ 12 files changed, 2267 insertions(+), 51 deletions(-) create mode 100644 src/hymical_forms/migrations/versions/0004_20260824_rate_limit_counters.py create mode 100644 src/hymical_forms/ratelimit.py create mode 100644 tests/integration/test_rate_limiting_postgres.py create mode 100644 tests/test_rate_limiting.py diff --git a/.env.example b/.env.example index 4a8c591..720b848 100644 --- a/.env.example +++ b/.env.example @@ -62,3 +62,38 @@ FORMS_DATABASE_URL=postgresql+psycopg://forms:forms@localhost:5432/forms # point at a server on your own machine. Development only: enabling this in # production lets anyone who can create an endpoint reach your internal network. # FORMS_ALLOW_PRIVATE_WEBHOOK_TARGETS=false + +# --- public ingestion rate limiting ------------------------------------------- +# +# These apply only to POST /f/{endpoint_id}. Management routes and /health are +# not affected. Counters live in PostgreSQL, so every API process enforces the +# same limit rather than one each. + +# Enforce the public ingestion rate limits. On by default, because a public route +# with no limit is the exposure this exists to close. Turn it off only for local +# development or a test that is about something else. +# FORMS_RATE_LIMIT_ENABLED=true + +# Submission attempts one source address may make per window, and how long that +# window lasts in seconds. +# FORMS_RATE_LIMIT_IP_REQUESTS=60 +# FORMS_RATE_LIMIT_IP_WINDOW_SECONDS=60 + +# Submission attempts one endpoint may receive per window, from every source +# together, and how long that window lasts in seconds. This is the limit that +# answers an attack spread across many addresses. +# FORMS_RATE_LIMIT_ENDPOINT_REQUESTS=600 +# FORMS_RATE_LIMIT_ENDPOINT_WINDOW_SECONDS=60 + +# Secret keying the digest that client addresses are counted under. Optional. +# Without it the digest is unkeyed, which keeps addresses out of the table but is +# not privacy against anyone who can read it, because the address space is small +# enough to enumerate. Every API process must be given the same value, or they +# will count the same client under different subjects. +# FORMS_RATE_LIMIT_IP_SECRET= + +# How many reverse proxies of your own stand in front of this process. 0, the +# default, means the client address is the socket peer and X-Forwarded-For is +# ignored entirely. Set it to the real number of hops, never higher: a value +# larger than your actual chain lets clients choose their own rate limit bucket. +# FORMS_TRUSTED_PROXY_HOPS=0 diff --git a/README.md b/README.md index 82b260c..55212f6 100644 --- a/README.md +++ b/README.md @@ -45,30 +45,33 @@ same authenticated management API. Everything that administers the service requires a management API key. Form ingestion stays public, because an ingestion URL is meant to sit in the `action` -attribute of somebody's HTML form. There is still no rate limiting and no spam -protection, so a public deployment is exposed to whatever volume the internet -sends it. - -| Capability | Status | -| ----------------------------- | ------------------------- | -| Health endpoint | Implemented | -| Form ingestion + validation | Implemented | -| Request limits + error model | Implemented | -| Endpoint registry | Implemented | -| Submission persistence | Implemented | -| Idempotent retries | Implemented | -| Signed webhook delivery | Implemented | -| Durable delivery queue | Implemented | -| Retries with backoff | Implemented | -| Schema migrations | Implemented | -| API keys / authentication | Implemented | -| Endpoint management | Implemented | -| Delivery inspection | Implemented | -| Manual delivery replay | Implemented | -| Endpoint deletion | **Not implemented** | -| Submission retrieval | **Not implemented** | -| Rate limiting, spam handling | **Not implemented** | -| Export, retention, dashboards | **Not implemented** | +attribute of somebody's HTML form. Public submissions are now rate limited per +source address and per endpoint, which bounds the volume one deployment will +accept. That is traffic protection and nothing more: **there is still no spam +protection**, no CAPTCHA and no content classification, so a public deployment +will accept junk up to the configured rate. + +| Capability | Status | +| ------------------------------ | ------------------------- | +| Health endpoint | Implemented | +| Form ingestion + validation | Implemented | +| Request limits + error model | Implemented | +| Endpoint registry | Implemented | +| Submission persistence | Implemented | +| Idempotent retries | Implemented | +| Signed webhook delivery | Implemented | +| Durable delivery queue | Implemented | +| Retries with backoff | Implemented | +| Schema migrations | Implemented | +| API keys / authentication | Implemented | +| Endpoint management | Implemented | +| Delivery inspection | Implemented | +| Manual delivery replay | Implemented | +| Public ingestion rate limiting | Implemented | +| Endpoint deletion | **Not implemented** | +| Submission retrieval | **Not implemented** | +| Spam handling, CAPTCHA | **Not implemented** | +| Export, retention, dashboards | **Not implemented** | ## Requirements @@ -134,7 +137,7 @@ database is reachable and at the migration revision the build was written against, and refuse to start otherwise: ``` -the database is at migration '0002' but this build expects '0003'. +the database is at migration '0003' but this build expects '0004'. Run 'alembic upgrade head' before starting. ``` @@ -540,6 +543,11 @@ Accepts a form submission for a registered endpoint and stores it. has no way to send. No management credential is read here, and one sent anyway is ignored rather than forwarded anywhere. +**It is rate limited**, per source address and per endpoint, and an attempt over +either limit is refused with `429 rate_limit_exceeded` and a `Retry-After` +header. See [Rate limits](#rate-limits) for the defaults, what counts as an +attempt, and how the client address is determined. + A submission to an ID that does not exist is rejected with `404 endpoint_not_found`, and one to an inactive endpoint with `409 endpoint_inactive`. Neither leaves anything in the database. @@ -575,6 +583,160 @@ No outbound request is made during this request, so the response says nothing about whether a destination is reachable: that is the worker's business, and a destination being down can no longer affect whether a form is accepted. +### Rate limits + +`POST /f/{endpoint_id}` is public and stays public, which means anyone who can +reach it can send it traffic. Two limits bound how much. + +| Limit | Counts | Default | +| --------------- | --------------------------------------------------- | ------------------ | +| Per source | Attempts one client address makes, across every endpoint | 60 per 60 seconds | +| Per endpoint | Attempts one endpoint receives, from every source together | 600 per 60 seconds | + +A submission must satisfy **both**. The per-source limit stops one client +flooding many endpoints; the per-endpoint limit stops one endpoint consuming the +whole deployment's capacity, including under an attack spread across thousands of +addresses that each stay under the per-source limit. + +Neither limit applies to `GET /health` or to any management route. Ingestion +traffic cannot lock an operator out of their own service. + +#### Being refused + +``` +HTTP/1.1 429 Too Many Requests +Retry-After: 30 +``` + +```json +{ + "error": { + "code": "rate_limit_exceeded", + "message": "Too many submission attempts. Try again in 30 seconds.", + "details": { + "scope": "ip", + "limit": 60, + "window_seconds": 60, + "retry_after_seconds": 30 + } + } +} +``` + +`scope` is `ip` or `endpoint`, and `Retry-After` is whole seconds until the +window that refused you ends. Which limit tripped is told to you deliberately: a +developer whose own client is looping and one whose form is being flooded from +elsewhere need to do completely different things about it, and anybody could +distinguish the two anyway by trying the same endpoint from a second address. +What is never returned is the counter, the subject it is keyed by, or anything +naming a column. + +#### What counts as an attempt + +Every request that reaches the ingestion route spends a unit of budget, **whether +or not it is accepted**. A malformed body, an unsupported content type, an empty +submission and a submission to a disabled endpoint all cost the sender the same +as a successful one, because they all cost this service the same work. Abuse +traffic that is invalid is still abuse traffic. + +The order is fixed and worth knowing: + +| Step | Effect | +| ----------------------------------- | ---------------------------------------------------------- | +| Body size cap, in middleware | An oversized body is refused with `413` and spends nothing | +| Per-source limit | Always spent, before the endpoint ID is even checked | +| Endpoint lookup | An unknown endpoint returns `404` | +| Per-endpoint limit | Spent for any endpoint that exists, active or not | +| Content type, body parse, storage | Only reached once both limits have allowed the attempt | + +Two consequences follow from that order, and both are deliberate: + +- **An attempt the endpoint limit refuses has already spent the source's + budget.** Otherwise hammering a saturated endpoint would be free, and an + attacker could keep a source address permanently under its own limit while + doing nothing but flooding. +- **An attempt the source limit refuses does not spend an endpoint's budget.** + The endpoint is never looked up, so a blocked address cannot burn through the + budget of an endpoint it is not being allowed to reach. A submission to an + endpoint ID that does not exist spends the source's budget and creates no + endpoint counter, so guessing identifiers cannot be used to choose how much + this table grows. + +**Idempotent replays count.** A repeated `Idempotency-Key` is still a request +that crosses the network and reaches the database, and exempting it would make +one leaked key an unlimited way past both limits. A replay the limits allow +behaves exactly as it did before: it returns the original submission and queues +no second delivery. + +#### Which address you are counted as + +By default the client address is the **socket peer address** the ASGI server +reports, and `X-Forwarded-For` is ignored entirely. That header is text the +client writes, so trusting it by default would hand every client its own private +rate limit. + +**If you run this behind a reverse proxy, the socket peer is your proxy**, and +without configuration every visitor would share one bucket. Set +`FORMS_TRUSTED_PROXY_HOPS` to the number of proxies of your own in front of the +process: + +```bash +FORMS_TRUSTED_PROXY_HOPS=1 +``` + +Each proxy appends the address it saw, so the entry that many places from the +**right** of `X-Forwarded-For` is the one your outermost proxy observed; +everything to the left of it was written by somebody who is not yours to trust. +Set it to the real number of hops and never higher: a value larger than your +actual chain lets a client insert entries and pick its own bucket. If the header +is missing, or carries fewer entries than you configured, the socket peer is used +instead rather than the header being half believed. Make sure your proxy is +actually appending the header (nginx: `proxy_set_header X-Forwarded-For +$proxy_add_x_forwarded_for`). + +Addresses are stored as a SHA-256 digest, never as text, and no route or log line +returns one. Setting `FORMS_RATE_LIMIT_IP_SECRET` keys that digest with HMAC and +makes it genuinely one way; without it the digest is obfuscation only, because the +IPv4 space is small enough for anybody holding the table to enumerate. The usual +argument against adding a second secret does not apply here: these counters live +for one window, so changing or losing the secret costs at most one window of +accounting. Every API process must be given the same value. + +#### How the limits are enforced + +Counters are rows in PostgreSQL, keyed by limiter, subject and window start, and +incremented with a single `INSERT ... ON CONFLICT DO UPDATE ... RETURNING`. That +one statement is the whole concurrency argument: reading a counter, comparing it +in Python and writing it back would let two simultaneous requests both see room +and both pass. Here the database settles it and hands each request a different +number, so at most one of them can be the last one under the limit. This is +tested against real PostgreSQL with several independently built applications, +each with its own engine and connection pool, submitting at the same instant. + +**The limit is shared, not per process.** Two API replicas enforce one limit +between them rather than one each, which is the entire reason the state is in the +database. There is no Redis, no external rate-limit service and no sticky-session +requirement. + +The algorithm is a **fixed window**: the current window is `now` floored to a +multiple of the window length, against the Unix epoch, so every process derives +the same boundary. It is deterministic and cheap, and its known weakness is the +boundary. A client that spends a whole window just before it ends and another +just after can make twice the configured requests across those two windows. A +sliding window or token bucket would smooth that out at the cost of keeping a log +of request instants or a refill timestamp, which is not worth it for a first +layer whose job is to stop unbounded traffic rather than shape well-behaved +traffic. + +Old windows are removed opportunistically: a small fraction of submission +attempts also delete counters whose window ended several windows ago. The cutoff +is far enough back that a sweep can never take a window still being counted in, +and the delete rides an index on the window column rather than scanning. There is +no extra daemon to deploy for it. + +Set `FORMS_RATE_LIMIT_ENABLED=false` to turn all of this off for local +development. It is on by default, and should stay on in production. + ### Retrying safely with `Idempotency-Key` A client that never sees a response cannot tell whether the submission landed. @@ -1033,6 +1195,7 @@ add. | 409 | `delivery_not_replayable` | Delivery has not terminally failed | | 413 | `request_body_too_large` | Body exceeded `FORMS_MAX_BODY_BYTES` | | 415 | `unsupported_media_type` | Content type is not a supported form encoding | +| 429 | `rate_limit_exceeded` | A public ingestion rate limit was exhausted | | 422 | `empty_submission` | No fields were submitted | | 422 | `invalid_cursor` | Pagination cursor does not continue from a known row | | 422 | `invalid_endpoint_id` | Endpoint ID in a request body breaks the ID rules | @@ -1050,6 +1213,10 @@ Ingestion rule codes are `too_many_fields`, `field_name_too_long`, from: `404` when it arrived as a submission path that addresses nothing, `422` when it arrived as a field in a request body. +`rate_limit_exceeded` is the one error that carries a `Retry-After` header, in +whole seconds. Its `details` name which limit tripped and how long its window is; +see [Rate limits](#rate-limits). + Both `401`s carry `WWW-Authenticate: Bearer`. They are deliberately `401` and not `403`: a `403` says the caller is known and not permitted, which needs a permission model this build does not have. `invalid_api_key` is one answer for @@ -1078,6 +1245,18 @@ All settings are read from `FORMS_`-prefixed environment variables, or from a | `FORMS_WORKER_POLL_SECONDS` | `1` | How often an idle worker looks for work | | `FORMS_WORKER_BATCH_SIZE` | `10` | Deliveries a worker claims at once | | `FORMS_WORKER_LEASE_SECONDS` | `60` | How long a worker's claim holds | +| `FORMS_RATE_LIMIT_ENABLED` | `true` | Enforce the public ingestion rate limits | +| `FORMS_RATE_LIMIT_IP_REQUESTS` | `60` | Attempts one source address may make per window | +| `FORMS_RATE_LIMIT_IP_WINDOW_SECONDS` | `60` | How long the per-address window lasts | +| `FORMS_RATE_LIMIT_ENDPOINT_REQUESTS` | `600` | Attempts one endpoint may receive per window | +| `FORMS_RATE_LIMIT_ENDPOINT_WINDOW_SECONDS` | `60` | How long the per-endpoint window lasts | +| `FORMS_RATE_LIMIT_IP_SECRET` | unset | Secret keying the digest addresses are counted under | +| `FORMS_TRUSTED_PROXY_HOPS` | `0` | Reverse proxies of your own in front of this process | + +The rate limit settings apply only to `POST /f/{endpoint_id}`; see +[Rate limits](#rate-limits). `FORMS_TRUSTED_PROXY_HOPS` is security-sensitive: +leaving it at `0` behind a proxy makes every visitor share one bucket, and +setting it higher than your real chain lets clients pick their own. There is deliberately no setting for a management API key. Keys live in the database so that creating and revoking one needs no restart, and so that a @@ -1118,6 +1297,14 @@ downgrade removes only what the newer revision added, and that the data survives that too. Another settles the manual replay race: several real connections replay one failed delivery at the same instant, and exactly one of them wins. +The rate limit suite there is the one that could not be faked. It builds several +whole applications, each with its own engine and connection pool, and has them +submit at the same instant against one database. Exactly the configured number of +attempts is accepted and the rest are refused, no increment is lost, and a budget +one application spent is already spent for another that has never seen the client +before, which is what "shared enforcement rather than process-local state" +actually has to mean. + CI runs the lint, format and type checks once, the fast suite across Python 3.11 to 3.13, and the PostgreSQL suite once against a PostgreSQL 17 service. @@ -1135,6 +1322,7 @@ src/hymical_forms/ ingestion.py domain rules: endpoint IDs, submission validation middleware.py request body size limit models.py the persisted schema + ratelimit.py rate limit rules: windows, subjects, client address trust storage.py queries and writes webhooks.py webhook rules: URL validation, payload, signature, retry policy worker.py the delivery worker process @@ -1149,8 +1337,8 @@ src/hymical_forms/ migrations/ Alembic environment and revisions ``` -`ingestion.py`, `webhooks.py` and `apikeys.py` hold the domain rules and know -nothing about HTTP or the database. `models.py` and `storage.py` are the only +`ingestion.py`, `webhooks.py`, `apikeys.py` and `ratelimit.py` hold the domain +rules and know nothing about HTTP or the database. `models.py` and `storage.py` are the only modules that write queries, and `delivery.py` is the only one that makes an outbound request. `api/` translates requests into domain rules and storage calls, and their outcomes into responses. `api/security.py` holds the one authentication @@ -1206,6 +1394,17 @@ judging it in Python and then writing it would let both requests pass the check and both reset the retry cycle, which is the duplicated work this exists to prevent. +Rate limit counters are the one table here that records nothing durable. A row is +a limiter, a subject and a window start, all three of which are the primary key, +so the index the key already creates is the index the increment conflicts on and +there is no second structure to keep in agreement with it. The increment is a +single `INSERT ... ON CONFLICT DO UPDATE ... RETURNING`, committed on its own +rather than inside the submission's transaction, so a submission that is refused +or fails to store cannot roll back the accounting that refused it. Every row +stops being consulted the moment its window ends, which is what makes bulk +deletion of old windows safe and why the whole table can be lost for the price of +one window of accounting. + The two attempt counters on a delivery are what let a replay be both honest and useful. `attempts` is the lifetime total and only ever rises, so it can number the audit trail without a number ever being reused. `cycle_attempts` is the @@ -1246,9 +1445,27 @@ database sets one from the other. Treat the current checks as a guardrail against mistakes, not a defence against an attacker who can configure endpoints. Authentication narrows who that is to whoever holds a management key; it does not make the checks complete. -- **Form ingestion is public and unrated.** Anyone who can reach the API can post - to any active endpoint. There is no rate limiting, no spam protection and no - CAPTCHA, so a public deployment accepts whatever volume it is sent. +- **Rate limiting is traffic protection, not spam protection.** It bounds how + much a source or an endpoint can send; it has no opinion whatsoever about what + is in a submission. There is no CAPTCHA, no Turnstile, no content or ML + classification, no honeypot field, no disposable-email detection and no email + verification, so a public deployment still accepts junk up to the configured + rate. Form ingestion is public by design and stays that way. +- **The rate limit windows are fixed, so the boundary is soft.** A client that + spends a whole window just before it ends and another just after can make twice + the configured requests across those two windows. Set the window shorter if + that burst matters more to you than the smaller counters a longer window keeps. +- **The client address is only as trustworthy as your deployment.** Behind a + reverse proxy, `FORMS_TRUSTED_PROXY_HOPS` must match your real chain. Left at + `0` every visitor shares your proxy's bucket, and set too high a client can + forge `X-Forwarded-For` entries and pick its own. The default trusts nothing + but the socket peer, which is the safe end to fail towards but is wrong behind + a proxy. +- **Rate limiting adds writes to the ingestion path.** Every public attempt costs + one upsert per limiter, committed before the body is parsed. That is the price + of a limit that is shared across processes rather than enforced per process, + and it means the limiter fails closed: if the database is unreachable, the + attempt is refused with `503` rather than let through uncounted. - **A lost management key cannot be recovered,** only replaced. The server holds a digest and nothing else. Create a new key, move your callers onto it, and revoke the old one by the key ID `list-keys` still shows. diff --git a/src/hymical_forms/api/submissions.py b/src/hymical_forms/api/submissions.py index 8276cb3..da64f43 100644 --- a/src/hymical_forms/api/submissions.py +++ b/src/hymical_forms/api/submissions.py @@ -4,21 +4,26 @@ from __future__ import annotations +import logging import math -from datetime import datetime +import random +from datetime import datetime, timedelta from http import HTTPStatus from fastapi import APIRouter, Request from pydantic import BaseModel, Field from python_multipart.exceptions import ParseError +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session from starlette.concurrency import run_in_threadpool from starlette.datastructures import UploadFile from starlette.formparsers import FormParser, MultiPartException, MultiPartParser +from starlette.responses import JSONResponse from hymical_forms import storage from hymical_forms.config import Settings from hymical_forms.db import SessionDep -from hymical_forms.errors import ApiError, ErrorResponse +from hymical_forms.errors import ApiError, ErrorResponse, error_response from hymical_forms.ingestion import ( ENDPOINT_ID_RULE, IDEMPOTENCY_KEY_RULE, @@ -27,8 +32,21 @@ is_valid_idempotency_key, payload_fingerprint, ) +from hymical_forms.models import utcnow +from hymical_forms.ratelimit import ( + FORWARDED_FOR_HEADER, + Limiter, + RateLimit, + RateLimitDecision, + client_address, + ip_subject, + seconds_until_window_ends, + window_start, +) from hymical_forms.webhooks import WebhookTarget +logger = logging.getLogger(__name__) + IDEMPOTENCY_KEY_HEADER = "Idempotency-Key" URLENCODED = "application/x-www-form-urlencoded" @@ -39,6 +57,20 @@ # but only ever a bounded prefix of what the client sent. _MEDIA_TYPE_ECHO_LIMIT = 128 +# Fixed windows leave rows behind, and pretending that is harmless would be +# untrue: one row per source address per window is unbounded in exactly the +# traffic this feature exists to survive. A background process for one DELETE +# would be a whole thing to deploy, so a small fraction of submission attempts +# pay for it instead. At one in a hundred, a service quiet enough to accumulate +# nothing sweeps rarely and a service busy enough to accumulate a lot sweeps +# often, which is the right shape without a schedule to tune. +_SWEEP_PROBABILITY = 0.01 + +# How many of the longest configured window to keep before sweeping. More than +# one, so a sweep can never take a window that is still being counted in, and +# small, because these rows answer nothing once their window has ended. +_SWEEP_RETAINED_WINDOWS = 2 + router = APIRouter(tags=["submissions"]) @@ -199,6 +231,53 @@ def __init__(self, field_name: str) -> None: ) +class RateLimitExceeded(ApiError): + """ + raised when a public submission attempt exhausted one of the traffic limits + """ + + status_code = HTTPStatus.TOO_MANY_REQUESTS + code = "rate_limit_exceeded" + + def __init__(self, decision: RateLimitDecision) -> None: + """ + report which budget ran out and how long it takes to refill + :param decision: the exhausted budget's decision for this attempt + """ + # Which limiter tripped is included on purpose. An integrator whose form + # is being flooded from many addresses and one whose own client is + # looping need to do completely different things about it, and the answer + # is not something the response could keep hidden anyway: anyone can tell + # the two apart by trying the same endpoint from a second address. What is + # not included is the subject, the counter, or anything naming a column. + super().__init__( + f"Too many submission attempts. Try again in {decision.retry_after_seconds} seconds.", + details={ + "scope": str(decision.limiter), + "limit": decision.limit.requests, + "window_seconds": decision.limit.window_seconds, + "retry_after_seconds": decision.retry_after_seconds, + }, + ) + self.retry_after_seconds = decision.retry_after_seconds + + def as_response(self) -> JSONResponse: + """ + render this error with the wait a 429 is not a complete answer without + :returns: a JSONResponse carrying the envelope and a Retry-After header + """ + # Built per instance rather than through ``ApiError.headers``, which is a + # ClassVar for statuses whose header never varies. This one carries a + # number worked out for the request being refused. + return error_response( + status_code=self.status_code, + code=self.code, + message=self.message, + details=self.details, + headers={"Retry-After": str(self.retry_after_seconds)}, + ) + + class DeliveryStatus(BaseModel): """ whether this submission owes a webhook delivery @@ -251,6 +330,10 @@ class SubmissionAccepted(BaseModel): 413: {"model": ErrorResponse, "description": "Request body too large"}, 415: {"model": ErrorResponse, "description": "Unsupported content type"}, 422: {"model": ErrorResponse, "description": "Submission rejected by an ingestion rule"}, + 429: { + "model": ErrorResponse, + "description": "Rate limited by source address or by endpoint", + }, 503: {"model": ErrorResponse, "description": "Database unavailable"}, }, ) @@ -265,33 +348,79 @@ async def submit(endpoint_id: str, request: Request, session: SessionDep) -> Sub # The response is 202 Accepted rather than 201 Created: the submission is # stored, but the delivery it was accepted for has not happened yet. # - # The endpoint is resolved before the body is parsed, so an unknown endpoint - # costs one indexed lookup rather than a full parse of a body we would throw - # away. This handler must stay ``async`` to stream the body, so each blocking + # The order of this handler is the abuse-protection design, so it is worth + # stating plainly. The body-size cap has already run, in middleware, before + # this function exists: an oversized body is refused without being read and + # without touching the database at all. Then the source address spends a unit + # of its budget, before anything is looked up. Then the endpoint is resolved, + # and if it exists it spends a unit of its own budget. Only after both + # limiters have allowed the attempt is the body parsed and stored. + # + # This handler must stay ``async`` to stream the body, so each blocking # database call is handed to a worker thread instead of stalling the loop. + settings: Settings = request.app.state.settings + now = utcnow() + + # Charged before the identifier is even checked for syntax, because an + # attempt costs this service something whether or not it turns out to be + # well formed, and because the cheapest place to refuse a flood is the + # earliest one. Every attempt that reaches this handler is charged, including + # ones that go on to be refused as malformed, unsupported or unacceptable. + if settings.rate_limit_enabled: + await _consume( + session, + limiter=Limiter.IP, + subject=_address_subject(request, settings), + limit=settings.ip_rate_limit(), + now=now, + ) + if not is_valid_endpoint_id(endpoint_id): raise InvalidEndpointId() + # The endpoint is resolved before the body is parsed, so an unknown endpoint + # costs one indexed lookup rather than a full parse of a body we would throw + # away. endpoint = await run_in_threadpool(storage.get_endpoint, session, endpoint_id) if endpoint is None: + # A guessed identifier spends the guesser's own budget and nothing else. + # Charging a per-endpoint counter here would mean inventing a row for + # every string an attacker tries, which hands them control of how much + # this table grows. raise EndpointNotFound(endpoint_id) - if not endpoint.is_active: - raise EndpointInactive(endpoint_id) - # Read the webhook configuration off the row now, while the session is known - # to be clean. Storing the submission can roll back to settle an idempotency - # race, and a rollback expires loaded objects, so touching the endpoint later - # would silently issue a refresh query from this async handler. + # Read the endpoint's configuration off the row now, while the session is + # known to be clean. The limiter below commits, and storing the submission + # can roll back to settle an idempotency race; a rollback expires loaded + # objects, so touching the endpoint later would silently issue a refresh + # query from this async handler. + is_active = endpoint.is_active webhook_url = endpoint.webhook_url webhook_secret = endpoint.webhook_secret + # Charged for a resolved endpoint whether or not the attempt is going to be + # accepted, so that an endpoint somebody has disabled cannot be used as a + # free target either. Its budget is shared by every source, which is what + # makes it the limit that answers an attack spread across many addresses. + if settings.rate_limit_enabled: + await _consume( + session, + limiter=Limiter.ENDPOINT, + subject=endpoint_id, + limit=settings.endpoint_rate_limit(), + now=now, + ) + await _sweep_old_windows(session, settings, now=now) + + if not is_active: + raise EndpointInactive(endpoint_id) + media_type = _media_type(request.headers.get("content-type")) if media_type not in SUPPORTED_MEDIA_TYPES: raise UnsupportedMediaType(media_type) idempotency_key = _idempotency_key(request) - settings: Settings = request.app.state.settings submission = build_submission( endpoint_id, await _parse_form(request, media_type, settings), @@ -342,6 +471,86 @@ async def submit(endpoint_id: str, request: Request, session: SessionDep) -> Sub ) +async def _consume( + session: Session, + *, + limiter: Limiter, + subject: str, + limit: RateLimit, + now: datetime, +) -> None: + """ + spend one unit of a budget and refuse the attempt if it was already spent + :param session: the session this request does its database work through + :param limiter: which budget is being drawn from + :param subject: the value that budget is keyed by + :param limit: how many attempts the window allows and how long it lasts + :param now: the instant this attempt arrived + :raises RateLimitExceeded: if the subject has already spent this window's budget + """ + # The unit is spent first and judged afterwards, which is what makes a + # refused attempt still count against the sender. Deciding first and then + # charging only the attempts that passed would let a saturated subject keep + # sending for free, and free is the one thing abuse traffic must not be. + start = window_start(now, limit.window_seconds) + used = await run_in_threadpool( + storage.consume_rate_limit, + session, + limiter=limiter, + subject=subject, + window_start=start, + ) + decision = RateLimitDecision( + limiter=limiter, + limit=limit, + used=used, + retry_after_seconds=seconds_until_window_ends(now, start, limit.window_seconds), + ) + if not decision.allowed: + raise RateLimitExceeded(decision) + + +async def _sweep_old_windows(session: Session, settings: Settings, *, now: datetime) -> None: + """ + occasionally remove counters whose window ended long ago + :param session: the session this request does its database work through + :param settings: active configuration, read for the window lengths + :param now: the instant this attempt arrived + """ + if random.random() >= _SWEEP_PROBABILITY: + return + + oldest = max( + settings.rate_limit_ip_window_seconds, + settings.rate_limit_endpoint_window_seconds, + ) + before = now - timedelta(seconds=_SWEEP_RETAINED_WINDOWS * oldest) + try: + await run_in_threadpool(storage.delete_expired_rate_limit_counters, session, before=before) + except SQLAlchemyError: + # Housekeeping, and it runs after the decision this request needed has + # already been made and committed. A database that cannot tidy up must + # not be able to turn an otherwise fine submission into a 503, and the + # rollback is what hands the rest of the handler a usable session. + session.rollback() + logger.warning("could not sweep expired rate limit counters") + + +def _address_subject(request: Request, settings: Settings) -> str: + """ + work out the value this request's source address is counted under + :param request: the incoming request + :param settings: active configuration, read for the proxy and privacy settings + :returns: a hex digest of the resolved client address + """ + address = client_address( + peer=request.client.host if request.client is not None else None, + forwarded_for=request.headers.get(FORWARDED_FOR_HEADER), + trusted_proxy_hops=settings.trusted_proxy_hops, + ) + return ip_subject(address, settings.rate_limit_ip_secret) + + def _idempotency_key(request: Request) -> str | None: """ read and validate the retry key a client may have sent diff --git a/src/hymical_forms/config.py b/src/hymical_forms/config.py index 71482b6..5371616 100644 --- a/src/hymical_forms/config.py +++ b/src/hymical_forms/config.py @@ -1,8 +1,9 @@ """ application settings, read from ``FORMS_``-prefixed environment variables -Settings are added only when the code actually uses them, so this model is -currently limited to the ingestion boundary's protective limits. +Settings are added only when the code actually uses them, so this model covers +the ingestion boundary's protective limits, the traffic limits that guard public +ingestion, and what the delivery worker needs, and nothing speculative. """ from __future__ import annotations @@ -10,6 +11,7 @@ from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict +from hymical_forms.ratelimit import RateLimit from hymical_forms.webhooks import RetryPolicy @@ -105,6 +107,56 @@ class Settings(BaseSettings): ), ) + rate_limit_enabled: bool = Field( + default=True, + description=( + "Enforce the public ingestion rate limits. On by default, because a " + "public route with no limit is the exposure this exists to close. " + "Turn it off only for local development or a test that is about " + "something else." + ), + ) + rate_limit_ip_requests: int = Field( + default=60, + ge=1, + description="Public submission attempts one source address may make per window.", + ) + rate_limit_ip_window_seconds: int = Field( + default=60, + ge=1, + description="How long the per-address window lasts, in seconds.", + ) + rate_limit_endpoint_requests: int = Field( + default=600, + ge=1, + description="Public submission attempts one endpoint may receive per window.", + ) + rate_limit_endpoint_window_seconds: int = Field( + default=60, + ge=1, + description="How long the per-endpoint window lasts, in seconds.", + ) + rate_limit_ip_secret: str | None = Field( + default=None, + min_length=16, + description=( + "Secret keying the digest that client addresses are counted under. " + "Optional: without it the digest is unkeyed, which keeps addresses out " + "of the table but is not privacy against anyone who can read it. Every " + "API process must be given the same value." + ), + ) + trusted_proxy_hops: int = Field( + default=0, + ge=0, + description=( + "How many reverse proxies of your own stand in front of this process. " + "0, the default, means the client address is the socket peer and " + "X-Forwarded-For is ignored. Set it to the real number of hops, never " + "higher, or clients can choose their own rate limit bucket." + ), + ) + def retry_policy(self) -> RetryPolicy: """ gather the retry settings into the value the delivery code works with @@ -115,3 +167,23 @@ def retry_policy(self) -> RetryPolicy: initial_seconds=self.webhook_retry_initial_seconds, max_seconds=self.webhook_retry_max_seconds, ) + + def ip_rate_limit(self) -> RateLimit: + """ + gather the per-address limit into the value the limiter works with + :returns: the configured per-address rate limit + """ + return RateLimit( + requests=self.rate_limit_ip_requests, + window_seconds=self.rate_limit_ip_window_seconds, + ) + + def endpoint_rate_limit(self) -> RateLimit: + """ + gather the per-endpoint limit into the value the limiter works with + :returns: the configured per-endpoint rate limit + """ + return RateLimit( + requests=self.rate_limit_endpoint_requests, + window_seconds=self.rate_limit_endpoint_window_seconds, + ) diff --git a/src/hymical_forms/migrations/versions/0004_20260824_rate_limit_counters.py b/src/hymical_forms/migrations/versions/0004_20260824_rate_limit_counters.py new file mode 100644 index 0000000..9922c4c --- /dev/null +++ b/src/hymical_forms/migrations/versions/0004_20260824_rate_limit_counters.py @@ -0,0 +1,79 @@ +""" +public ingestion rate limit counters + +Adds the table the two public ingestion limiters count in. Nothing an earlier +revision created is touched, so a database already holding endpoints, +submissions, deliveries, attempts and management keys gains a table and loses +nothing, and the downgrade removes exactly that table again. + +The identity of a counter is its limiter, its subject and the start of the fixed +window it counts, and all three are the primary key. That is deliberate: the +increment is written as an upsert, so the index the primary key already creates +is the index the conflict resolves on, and there is no second structure to keep +in agreement with it. The window column is indexed on its own as well, because +cleanup ranges over it without knowing a limiter or a subject. + +Nothing in this table is durable in the sense the rest of the schema is. Every +row stops being consulted the moment its window ends, and losing the whole table +costs at most one window of accounting, which is why the downgrade drops it +without a backfill and why deleting old rows in bulk is safe. + +No raw address is stored here. The per-IP subject is a digest, and no submitted +field name, field value or credential reaches this table at all. + +Timestamps are written as ``sa.DateTime(timezone=True)`` rather than the +application's ``UtcDateTime`` decorator, for the reason given in ``0001``: a +migration is a frozen record of a change, not a view of the current models. + +revision: 0004 +revises: 0003 +created: 2026-08-24 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0004" +down_revision: str | None = "0003" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """ + create the table the public ingestion limiters count in + """ + op.create_table( + "rate_limit_counters", + sa.Column("limiter", sa.String(length=16), nullable=False), + sa.Column("subject", sa.String(length=64), nullable=False), + sa.Column("window_start", sa.DateTime(timezone=True), nullable=False), + sa.Column("attempts", sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint( + "limiter", + "subject", + "window_start", + name=op.f("pk_rate_limit_counters"), + ), + ) + op.create_index( + op.f("ix_rate_limit_counters_window_start"), + "rate_limit_counters", + ["window_start"], + unique=False, + ) + + +def downgrade() -> None: + """ + remove the schema this revision created + """ + # Nothing references this table and nothing else references it, so dropping + # it leaves every earlier revision's data exactly as it was. A build at 0003 + # simply enforces no ingestion rate limit again. + op.drop_index(op.f("ix_rate_limit_counters_window_start"), table_name="rate_limit_counters") + op.drop_table("rate_limit_counters") diff --git a/src/hymical_forms/models.py b/src/hymical_forms/models.py index de0e4c8..e3539af 100644 --- a/src/hymical_forms/models.py +++ b/src/hymical_forms/models.py @@ -1,6 +1,7 @@ """ -the persisted schema: endpoints, the submissions addressed to them, and the -management credentials that administer the service +the persisted schema: endpoints, the submissions addressed to them, the +management credentials that administer the service, and the traffic counters +that protect public ingestion """ from __future__ import annotations @@ -33,6 +34,7 @@ SUBMISSION_ID_MAX_LENGTH, ) from hymical_forms.ingestion import Submission as DomainSubmission +from hymical_forms.ratelimit import LIMITER_MAX_LENGTH, SUBJECT_MAX_LENGTH from hymical_forms.webhooks import ( DELIVERY_ATTEMPT_ID_MAX_LENGTH, DELIVERY_ERROR_MAX_LENGTH, @@ -391,3 +393,41 @@ def is_active(self) -> bool: :returns: True if the key has not been revoked """ return self.revoked_at is None + + +class RateLimitCounter(Base): + """ + how many public submission attempts one subject made inside one fixed window + """ + + # The shared state the public ingestion limiters decide on. It is here rather + # than in process memory because a limit enforced per process is not a limit: + # two API replicas would each allow the configured number and the service + # would accept twice it, and an autoscaler would raise the real ceiling every + # time it added a replica. + # + # This is the one table in the schema that holds no durable record of + # anything. Every row is disposable, it stops being consulted the moment its + # window ends, and losing the whole table costs at most one window of + # accounting. That is what makes the counters cheap to write on the hot path + # and safe to delete in bulk. + __tablename__ = "rate_limit_counters" + + # The three parts of the identity are the primary key, which makes the index + # that the atomic upsert conflicts on the same index the read rides on. There + # is no surrogate key because nothing ever refers to one of these rows. + limiter: Mapped[str] = mapped_column(String(LIMITER_MAX_LENGTH), primary_key=True) + + # An endpoint identifier for the per-endpoint limiter, and a digest of the + # client address for the per-IP one. No raw address is stored here, and + # neither is anything a submission carried: this table sees a subject and a + # count, never a field name, a value or a credential. + subject: Mapped[str] = mapped_column(String(SUBJECT_MAX_LENGTH), primary_key=True) + + # Floored to the window boundary, so every process writing this row derives + # the same value from the same instant. Indexed on its own as well as being + # the last part of the key, because cleanup ranges over it without knowing a + # limiter or a subject and the composite key cannot answer that. + window_start: Mapped[datetime] = mapped_column(UtcDateTime, primary_key=True, index=True) + + attempts: Mapped[int] = mapped_column() diff --git a/src/hymical_forms/ratelimit.py b/src/hymical_forms/ratelimit.py new file mode 100644 index 0000000..4648f44 --- /dev/null +++ b/src/hymical_forms/ratelimit.py @@ -0,0 +1,168 @@ +""" +traffic rate limiting rules for public form ingestion + +Nothing in this module performs I/O or knows about HTTP or the database. It +answers which fixed window an instant falls in, how long that window has left, +which address a request should be counted against, and what value each limiter +is keyed by. :mod:`hymical_forms.storage` owns the atomic counter, and the +ingestion route owns the order the two limiters run in. + +The algorithm is a fixed window on purpose. It is one row and one statement per +decision, every process computes the same boundary from the same clock, and what +it does is explainable in a sentence. Its known weakness is the boundary: a +client that spends a whole window just before it ends and a whole window just +after can make twice the configured requests across those two windows. A sliding +window or a token bucket would smooth that out, at the cost of either keeping a +log of request instants or a second column that has to be refilled from a +timestamp, and neither is worth it for a first layer of abuse protection whose +job is to stop unbounded traffic rather than to shape well-behaved traffic. +""" + +from __future__ import annotations + +import hashlib +import hmac +import math +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from enum import StrEnum + +# One column holds both kinds of subject. An endpoint identifier is at most 64 +# characters and a hashed address is a 64-character hex digest, so they fit the +# same width without either being padded or truncated. +SUBJECT_MAX_LENGTH = 64 +LIMITER_MAX_LENGTH = 16 + +FORWARDED_FOR_HEADER = "X-Forwarded-For" + +# What a request is counted under when the ASGI server reported no peer address. +# Such requests share one bucket rather than escaping the limiter: a request +# nothing can attribute is exactly what abuse looks like, so the safe reading is +# the strict one. +UNKNOWN_CLIENT = "unknown" + + +class Limiter(StrEnum): + """ + which budget a rate limit decision is drawn from + """ + + # These strings are stored and are reported back in the body of a 429, so + # they are a public contract rather than an implementation detail. + IP = "ip" + ENDPOINT = "endpoint" + + +@dataclass(frozen=True, slots=True) +class RateLimit: + """ + how many attempts one subject may make within one fixed window + """ + + requests: int + window_seconds: int + + +@dataclass(frozen=True, slots=True) +class RateLimitDecision: + """ + the outcome of spending one unit of one subject's budget + """ + + limiter: Limiter + limit: RateLimit + used: int + retry_after_seconds: int + + @property + def allowed(self) -> bool: + """ + report whether the attempt that spent this unit may proceed + :returns: True if the budget was not already exhausted + """ + # The unit is spent either way, so the request that takes a subject + # exactly to its limit is the last one allowed through. + return self.used <= self.limit.requests + + +def window_start(now: datetime, window_seconds: int) -> datetime: + """ + find the start of the fixed window an instant falls in + :param now: the instant to place + :param window_seconds: how long each window lasts + :returns: the window's start, as a timezone-aware UTC timestamp + """ + # Floored against the Unix epoch rather than against anything process-local, + # so every API process derives the same boundary from the same clock and the + # counter they share is the counter they both meant to write. + elapsed = int(now.timestamp()) + return datetime.fromtimestamp(elapsed - elapsed % window_seconds, UTC) + + +def seconds_until_window_ends(now: datetime, start: datetime, window_seconds: int) -> int: + """ + work out how long a refused client has to wait for a fresh window + :param now: the instant the decision was made + :param start: the start of the window the decision was made in + :param window_seconds: how long each window lasts + :returns: whole seconds remaining, never fewer than one + """ + # Rounded up and floored at one, because RFC 9110 wants whole seconds and + # because answering with a truncated value would invite a retry the same + # window is still going to refuse. + remaining = (start + timedelta(seconds=window_seconds) - now).total_seconds() + return max(1, math.ceil(remaining)) + + +def client_address(*, peer: str | None, forwarded_for: str | None, trusted_proxy_hops: int) -> str: + """ + decide which address a request should be rate limited by + :param peer: socket peer address the ASGI server reported, or None if it reported none + :param forwarded_for: raw ``X-Forwarded-For`` header value, or None when absent + :param trusted_proxy_hops: how many proxies of your own stand in front of this process + :returns: the address the per-IP limiter counts against + """ + # The socket peer is the default because it is the one address in a request + # that the client did not write. ``X-Forwarded-For`` is attacker-controlled + # text until a proxy you run appends to it, so treating it as authoritative + # by default would hand every client its own private rate limit for free. + # + # It is read only when an operator has said how many hops of their own to + # skip, and it is counted from the right: each proxy in the chain appends the + # address it saw, so with one trusted proxy the last entry is what that proxy + # observed, with two it is the second from last, and everything to the left of + # that was written by somebody who is not yours to trust. + if trusted_proxy_hops > 0 and forwarded_for: + hops = [entry.strip() for entry in forwarded_for.split(",") if entry.strip()] + if len(hops) >= trusted_proxy_hops: + return hops[-trusted_proxy_hops] + # Fewer entries than configured means the chain is not what the operator + # described, so the header is discarded rather than half believed. + return peer or UNKNOWN_CLIENT + + +def ip_subject(address: str, secret: str | None) -> str: + """ + reduce a client address to the value a counter may be keyed by + :param address: the client address resolved for this request + :param secret: server-side secret to key the digest with, or None for a plain digest + :returns: a hex SHA-256 digest of the address + """ + # The raw address is never stored, never logged and never returned. What is + # stored is a fixed-width digest of it, which is enough for the only thing the + # limiter needs: telling one source from another within a window. + # + # Without a secret this is obfuscation and is documented as exactly that. The + # IPv4 space is small enough to enumerate, so anybody holding the table can + # recover the addresses in it; the digest only keeps them out of a casual dump + # and out of anything that reads the column by eye. + # + # With a secret it is a genuine one-way mapping, and the usual objection to + # introducing a second secret does not apply here. These counters live for one + # window, so changing or losing the secret costs at most one window of + # accounting rather than invalidating anything durable. That is the whole + # operational problem it solves, and it is why the secret is optional rather + # than required. + if secret is None: + return hashlib.sha256(address.encode("utf-8")).hexdigest() + return hmac.new(secret.encode("utf-8"), address.encode("utf-8"), hashlib.sha256).hexdigest() diff --git a/src/hymical_forms/storage.py b/src/hymical_forms/storage.py index f1d2a6a..1f914e3 100644 --- a/src/hymical_forms/storage.py +++ b/src/hymical_forms/storage.py @@ -10,9 +10,11 @@ anything once it is committed; :func:`complete_attempt`, because the audit record and the state it justifies have to land together; :func:`update_endpoint` and :func:`requeue_failed_delivery`, because each is the whole of what its -management request came to do; and the two management key writes, +management request came to do; the two management key writes, :func:`revoke_management_key` and :func:`record_management_key_use`, for the -same reason. +same reason; and the two rate limit operations, :func:`consume_rate_limit` and +:func:`delete_expired_rate_limit_counters`, because abuse accounting has to +outlive the request it was accounting for. """ from __future__ import annotations @@ -21,13 +23,16 @@ from datetime import datetime, timedelta from typing import Any, TypeVar, cast -from sqlalchemy import ColumnElement, Select, and_, or_, select, tuple_, update +from sqlalchemy import ColumnElement, Select, and_, delete, or_, select, tuple_, update +from sqlalchemy.dialects.postgresql import insert as postgresql_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.engine import CursorResult from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from hymical_forms import models from hymical_forms.ingestion import Submission +from hymical_forms.ratelimit import Limiter from hymical_forms.webhooks import ( DeliveryOutcome, DeliveryResult, @@ -734,6 +739,123 @@ def record_management_key_use(session: Session, key_id: str, *, now: datetime) - session.commit() +class UnsupportedRateLimitBackend(Exception): + """ + raised when the configured database cannot perform an atomic counter upsert + """ + + def __init__(self, dialect: str) -> None: + """ + record which backend was asked to enforce a rate limit + :param dialect: the SQLAlchemy dialect name the session is bound to + """ + super().__init__( + f"rate limiting needs PostgreSQL or SQLite, not {dialect!r}, " + "because the counter is incremented with an upsert" + ) + self.dialect = dialect + + +def consume_rate_limit( + session: Session, + *, + limiter: Limiter, + subject: str, + window_start: datetime, +) -> int: + """ + spend one unit of a subject's budget for one window, atomically + :param session: the session to write through + :param limiter: which budget is being drawn from + :param subject: the value that budget is keyed by + :param window_start: the start of the fixed window being counted in + :returns: how many attempts this subject has now made inside that window + :raises UnsupportedRateLimitBackend: if the session is bound to another database + """ + # One statement decides everything. Reading the counter, comparing it in + # Python and writing it back would let two requests that arrive together both + # read the same value, both find room, and both pass, which is precisely the + # hole a rate limiter exists to close. Here the database inserts the row or + # increments the existing one under its own row lock, and hands back the value + # it settled on, so two simultaneous requests get two different numbers and at + # most one of them can be the last one under the limit. + # + # ``ON CONFLICT DO UPDATE`` rather than a lock-then-update, because the row + # for a brand new subject does not exist yet and two requests racing to create + # it have nothing to lock. The upsert makes the create and the increment the + # same operation, so the first request of a window is settled by the primary + # key rather than by whoever inserted first. + counter = models.RateLimitCounter + upsert: Any = _upsert_for(session) + statement = ( + upsert(counter) + .values( + limiter=str(limiter), + subject=subject, + window_start=window_start, + attempts=1, + ) + .on_conflict_do_update( + index_elements=["limiter", "subject", "window_start"], + # The unqualified column on the right is the stored row's value, not + # the one this statement proposed, which is what makes this an + # increment rather than an overwrite. + set_={"attempts": counter.attempts + 1}, + ) + .returning(counter.attempts) + ) + used = cast(int, session.scalars(statement).one()) + + # Committed here, and deliberately not left to the caller. The decision has to + # survive whatever the request does next: a submission that is refused, fails + # validation, or loses an idempotency race rolls its own work back, and abuse + # accounting that rolled back with it would let an attacker send unlimited + # traffic as long as every request was invalid. + session.commit() + return used + + +def delete_expired_rate_limit_counters(session: Session, *, before: datetime) -> int: + """ + remove counters for windows old enough that nothing will consult them again + :param session: the session to write through + :param before: counters whose window starts strictly before this instant are removed + :returns: how many counters were removed + """ + # A range over the indexed window column, so this is a bounded delete rather + # than a scan of the table. The caller chooses a cutoff several windows in the + # past, so a sweep can never take a window that is still being counted in. + counter = models.RateLimitCounter + result = cast( + "CursorResult[Any]", + session.execute( + delete(counter) + .where(counter.window_start < before) + .execution_options(synchronize_session=False) + ), + ) + session.commit() + return result.rowcount + + +def _upsert_for(session: Session) -> Any: + """ + pick the dialect-specific insert that can express an atomic increment + :param session: the session whose backend the statement will run against + :returns: the dialect's ``insert`` construct + :raises UnsupportedRateLimitBackend: if the backend is neither PostgreSQL nor SQLite + """ + # ``ON CONFLICT`` is not in the generic construct, so the dialect has to be + # named. Anything else is refused rather than silently falling back to a + # read-compare-write that would not be atomic. + dialect = session.get_bind().dialect.name + if dialect == "postgresql": + return postgresql_insert + if dialect == "sqlite": + return sqlite_insert + raise UnsupportedRateLimitBackend(dialect) + + def _page_after(session: Session, model: type[_Paged], cursor: str) -> ColumnElement[bool]: """ build the test for rows that come after a cursor, in newest-first order diff --git a/tests/integration/test_migrations_postgres.py b/tests/integration/test_migrations_postgres.py index 739f9b6..eee06dd 100644 --- a/tests/integration/test_migrations_postgres.py +++ b/tests/integration/test_migrations_postgres.py @@ -24,7 +24,7 @@ from integration.support import temporary_database BASELINE_TABLES = {"endpoints", "submissions", "webhook_deliveries", "delivery_attempts"} -EXPECTED_TABLES = BASELINE_TABLES | {"management_api_keys"} +EXPECTED_TABLES = BASELINE_TABLES | {"management_api_keys", "rate_limit_counters"} # Representative interval 6 data: an endpoint with a webhook, a submission sent # with an idempotency key, the delivery it owes, and one recorded attempt. @@ -92,6 +92,7 @@ def test_the_migration_creates_the_constraints_the_application_relies_on( "uq_submissions_endpoint_idempotency_key", "uq_webhook_deliveries_submission", "uq_management_api_keys_key_digest", + "pk_rate_limit_counters", "ck_endpoints_webhook_configuration", "ck_submissions_idempotency_identity", "ck_webhook_deliveries_completion", @@ -302,7 +303,7 @@ def test_downgrading_from_0003_leaves_the_delivery_data_alone(postgres_url: str) def test_a_populated_0002_survives_the_whole_round_trip(postgres_url: str) -> None: """ - populated 0002 to 0003 and back and forward again must end with zero drift + populated 0002 to 0003 and back and forward to head must end with zero drift :param postgres_url: a URL on the PostgreSQL server to work against """ with _database_at_baseline(postgres_url) as (config, engine): @@ -312,7 +313,134 @@ def test_a_populated_0002_survives_the_whole_round_trip(postgres_url: str) -> No command.upgrade(config, "0003") command.downgrade(config, "0002") + command.upgrade(config, "head") + + assert current_revision(engine) == head_revision() + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + assert ( + connection.scalar( + text("select cycle_attempts from webhook_deliveries where id = :id"), + {"id": SEEDED_DELIVERY}, + ) + == 1 + ) + difference = compare_metadata(MigrationContext.configure(connection), Base.metadata) + assert difference == [], f"migrated schema differs from the models: {difference}" + + +# --- the rate limit counters 0004 added -------------------------------------- +# +# 0004 adds a table rather than changing one, so what an upgrade has to prove +# here is that everything an operator's database already held is untouched, and +# that the new table arrives empty and ready rather than needing a backfill. + + +def test_upgrading_a_populated_0003_adds_the_counter_table(postgres_url: str) -> None: + """ + the rate limit table must arrive without disturbing anything already stored + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) command.upgrade(config, "0003") + assert "rate_limit_counters" not in set(inspect(engine).get_table_names()) + + command.upgrade(config, "0004") + + assert current_revision(engine) == "0004" + assert "rate_limit_counters" in set(inspect(engine).get_table_names()) + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + assert connection.scalar(text("select count(*) from rate_limit_counters")) == 0 + + +def test_the_counter_table_is_keyed_by_limiter_subject_and_window(postgres_url: str) -> None: + """ + the upsert conflicts on this key, so the key is what makes the increment atomic + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + command.upgrade(config, "0004") + + with engine.connect() as connection: + key = list( + connection.scalars( + text( + "select a.attname from pg_index i " + "join pg_attribute a on a.attrelid = i.indrelid " + "and a.attnum = any(i.indkey) " + "where i.indrelid = 'rate_limit_counters'::regclass and i.indisprimary" + ) + ) + ) + indexed = set( + connection.scalars( + text("select indexname from pg_indexes where tablename = 'rate_limit_counters'") + ) + ) + + assert set(key) == {"limiter", "subject", "window_start"} + # Cleanup ranges over the window column without knowing a limiter or a + # subject, which the primary key cannot answer. + assert "ix_rate_limit_counters_window_start" in indexed + + +def test_the_counter_window_is_stored_with_a_timezone(postgres_url: str) -> None: + with _database_at_baseline(postgres_url) as (config, engine): + command.upgrade(config, "0004") + + with engine.connect() as connection: + data_type = connection.scalar( + text( + "select data_type from information_schema.columns " + "where table_name = 'rate_limit_counters' and column_name = 'window_start'" + ) + ) + + assert data_type == "timestamp with time zone" + + +def test_downgrading_from_0004_leaves_everything_else_alone(postgres_url: str) -> None: + """ + the downgrade must remove the table 0004 added and nothing else + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) + command.upgrade(config, "0004") + with engine.begin() as connection: + connection.execute( + text( + "insert into rate_limit_counters (limiter, subject, window_start, attempts) " + "values ('ip', :subject, :now, 3)" + ), + {"subject": "e" * 64, "now": SEEDED_AT}, + ) + + command.downgrade(config, "0003") + + assert current_revision(engine) == "0003" + assert "rate_limit_counters" not in set(inspect(engine).get_table_names()) + with engine.connect() as connection: + _assert_baseline_data_intact(connection) + + +def test_a_populated_0003_survives_the_whole_round_trip(postgres_url: str) -> None: + """ + populated 0003 to 0004 and back and forward again must end with zero drift + :param postgres_url: a URL on the PostgreSQL server to work against + """ + with _database_at_baseline(postgres_url) as (config, engine): + with engine.begin() as connection: + _seed_baseline_data(connection) + command.upgrade(config, "0003") + + command.upgrade(config, "0004") + command.downgrade(config, "0003") + command.upgrade(config, "0004") assert current_revision(engine) == head_revision() with engine.connect() as connection: diff --git a/tests/integration/test_rate_limiting_postgres.py b/tests/integration/test_rate_limiting_postgres.py new file mode 100644 index 0000000..79442be --- /dev/null +++ b/tests/integration/test_rate_limiting_postgres.py @@ -0,0 +1,279 @@ +""" +public ingestion rate limiting against real PostgreSQL + +This is the behaviour SQLite cannot show. SQLite serialises writers, so the fast +suite can only demonstrate that the arithmetic is right. What matters in +production is that several API processes holding several connections cannot +between them let more traffic through than one of them would, and that can only +be shown here. + +Every application in this module is built on its own, with its own engine and its +own connection pool, so two of them are as independent as two deployed replicas. +Nothing is mocked and no session is shared: the only thing these applications +have in common is the database, which is exactly the claim being tested. +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import Engine, select +from sqlalchemy.orm import Session, sessionmaker + +from hymical_forms import models +from hymical_forms.app import create_app +from hymical_forms.ratelimit import Limiter +from integration.support import IsolatedSettings, seed_endpoint + +ENDPOINT = "/f/contact-form" +FORM = {"email": "dev@example.com"} + +IP_ALLOWANCE = 3 +ENDPOINT_ALLOWANCE = 4 + +# Long enough that no test can cross a window boundary while it runs, so an exact +# total is an exact total rather than a race with the clock. +WINDOW_SECONDS = 3600 + + +def no_headers(index: int) -> dict[str, str]: + """ + build the headers an attempt that claims no forwarded address sends + :param index: which attempt this is, which makes no difference here + :returns: an empty header mapping + """ + return {} + + +def build_app(postgres_url: str, **overrides: Any) -> FastAPI: + """ + build an application with a connection pool of its own + :param postgres_url: the database every application in a test shares + :param overrides: setting values to replace the built-in defaults + :returns: an application as independent of the others as a separate replica + """ + return create_app( + IsolatedSettings( + database_url=postgres_url, + rate_limit_ip_window_seconds=WINDOW_SECONDS, + rate_limit_endpoint_window_seconds=WINDOW_SECONDS, + **overrides, + ) + ) + + +def fire_together( + postgres_url: str, + attempts: int, + *, + headers_for: Callable[[int], dict[str, str]] = no_headers, + **overrides: Any, +) -> list[int]: + """ + submit once from each of several independent applications at the same instant + :param postgres_url: the database every application shares + :param attempts: how many applications, and therefore how many attempts + :param headers_for: builds the headers one attempt sends, from its index + :param overrides: setting values every application is built with + :returns: the status code each attempt received + """ + # The application and its client are built before the barrier, so what the + # attempts genuinely share is the moment they post rather than the moment they + # started connecting. + barrier = threading.Barrier(attempts) + + def attempt(index: int) -> int: + with TestClient(build_app(postgres_url, **overrides)) as client: + barrier.wait() + return client.post(ENDPOINT, data=FORM, headers=headers_for(index)).status_code + + with ThreadPoolExecutor(max_workers=attempts) as pool: + futures = [pool.submit(attempt, index) for index in range(attempts)] + return [future.result() for future in futures] + + +def counter_total(sessions: sessionmaker[Session], limiter: Limiter) -> int: + """ + total what one limiter recorded, read on a connection of its own + :param sessions: factory handing out independent connections + :param limiter: which limiter's counters to total + :returns: the sum of the attempts it recorded + """ + with sessions() as session: + rows = list(session.scalars(select(models.RateLimitCounter))) + return sum(row.attempts for row in rows if row.limiter == limiter) + + +def test_concurrent_attempts_from_one_source_cannot_exceed_its_allowance( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + ten processes must let exactly the configured number through, not ten + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + attempts = 10 + with sessions() as setup: + seed_endpoint(setup) + + statuses = fire_together( + postgres_url, + attempts, + rate_limit_ip_requests=IP_ALLOWANCE, + rate_limit_endpoint_requests=1000, + ) + + assert statuses.count(202) == IP_ALLOWANCE + assert statuses.count(429) == attempts - IP_ALLOWANCE + assert set(statuses) == {202, 429} + + # And the database agrees, end to end: every attempt was counted, and exactly + # the allowed ones left a submission behind. + assert counter_total(sessions, Limiter.IP) == attempts + with sessions() as session: + stored = list(session.scalars(select(models.Submission))) + assert len(stored) == IP_ALLOWANCE + + +def test_concurrent_attempts_from_many_sources_cannot_exceed_the_endpoint_allowance( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + an attack spread over addresses is exactly what the endpoint limit answers + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + attempts = 10 + with sessions() as setup: + seed_endpoint(setup) + + statuses = fire_together( + postgres_url, + attempts, + # Every attempt arrives from a different address, so no address budget can + # be what refused any of them. + headers_for=lambda index: {"X-Forwarded-For": f"203.0.113.{index + 1}"}, + trusted_proxy_hops=1, + rate_limit_ip_requests=1000, + rate_limit_endpoint_requests=ENDPOINT_ALLOWANCE, + ) + + assert statuses.count(202) == ENDPOINT_ALLOWANCE + assert statuses.count(429) == attempts - ENDPOINT_ALLOWANCE + assert counter_total(sessions, Limiter.ENDPOINT) == attempts + + +def test_no_increment_is_lost_when_many_processes_count_at_once( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + a read-compare-write would lose updates here and the total would come up short + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + attempts = 12 + with sessions() as setup: + seed_endpoint(setup) + + statuses = fire_together( + postgres_url, + attempts, + rate_limit_ip_requests=1000, + rate_limit_endpoint_requests=1000, + ) + + assert statuses == [202] * attempts + assert counter_total(sessions, Limiter.IP) == attempts + assert counter_total(sessions, Limiter.ENDPOINT) == attempts + + +def test_one_source_counter_holds_every_concurrent_attempt( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + the whole point of one shared row is that it is one row, not one per process + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + + fire_together(postgres_url, 8, rate_limit_ip_requests=1000, rate_limit_endpoint_requests=1000) + + with sessions() as session: + rows = list(session.scalars(select(models.RateLimitCounter))) + by_limiter = {row.limiter: row.attempts for row in rows} + assert len(rows) == 2, "the processes did not share one counter per limiter" + assert by_limiter == {str(Limiter.IP): 8, str(Limiter.ENDPOINT): 8} + + +def test_a_budget_one_process_spent_is_already_spent_for_the_next( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + an in-memory limiter would pass this second application every time + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + overrides: dict[str, Any] = { + "rate_limit_ip_requests": IP_ALLOWANCE, + "rate_limit_endpoint_requests": 1000, + } + + with TestClient(build_app(postgres_url, **overrides)) as first: + spent = [first.post(ENDPOINT, data=FORM).status_code for _ in range(IP_ALLOWANCE)] + + # A whole other application, built afterwards, with an engine and a pool it + # does not share with the first one. It has never seen this address before. + with TestClient(build_app(postgres_url, **overrides)) as second: + response = second.post(ENDPOINT, data=FORM) + + assert spent == [202] * IP_ALLOWANCE + assert response.status_code == 429 + assert response.json()["error"]["details"]["scope"] == "ip" + assert response.headers["Retry-After"].isdigit() + + +def test_two_endpoints_do_not_share_a_budget_under_load( + postgres_url: str, migrated_engine: Engine, sessions: sessionmaker[Session] +) -> None: + """ + flooding one endpoint must not refuse traffic addressed to another + :param postgres_url: the database every application shares + :param migrated_engine: unused, but forces the schema to exist first + :param sessions: factory handing out independent connections + """ + with sessions() as setup: + seed_endpoint(setup) + seed_endpoint(setup, endpoint_id="second-form") + + fire_together( + postgres_url, + 6, + headers_for=lambda index: {"X-Forwarded-For": f"203.0.113.{index + 1}"}, + trusted_proxy_hops=1, + rate_limit_ip_requests=1000, + rate_limit_endpoint_requests=ENDPOINT_ALLOWANCE, + ) + + with TestClient( + build_app( + postgres_url, + rate_limit_ip_requests=1000, + rate_limit_endpoint_requests=ENDPOINT_ALLOWANCE, + ) + ) as client: + assert client.post(ENDPOINT, data=FORM).status_code == 429 + assert client.post("/f/second-form", data=FORM).status_code == 202 diff --git a/tests/test_openapi.py b/tests/test_openapi.py index e2134d1..a8be3fd 100644 --- a/tests/test_openapi.py +++ b/tests/test_openapi.py @@ -100,6 +100,17 @@ def test_a_management_route_documents_its_refusal( assert "401" in responses +def test_the_submission_route_documents_being_rate_limited(client: TestClient) -> None: + """ + a public route that can answer 429 has to say so, or a caller cannot handle it + :param client: test client whose app holds the default endpoint + """ + responses = operation(client, "/f/{endpoint_id}", "post")["responses"] + + assert "429" in responses + assert "401" not in responses, "the public submission route advertises authentication" + + def test_no_response_schema_mentions_a_webhook_signing_secret(client: TestClient) -> None: """ a read model that could name the secret is a leak waiting to be written diff --git a/tests/test_rate_limiting.py b/tests/test_rate_limiting.py new file mode 100644 index 0000000..e74f59b --- /dev/null +++ b/tests/test_rate_limiting.py @@ -0,0 +1,856 @@ +""" +traffic rate limiting on public form ingestion + +These tests are about the ordering as much as the arithmetic. What counts as an +attempt, which limiter a refused attempt has already spent, and what is charged +before the endpoint is even known are all decisions rather than accidents, so +each one is asserted rather than assumed. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from pydantic import ValidationError +from sqlalchemy import select + +from conftest import ( + URLENCODED_HEADERS, + ClientFactory, + build_settings, + create_endpoint, + open_session, +) +from hymical_forms import models, storage +from hymical_forms.ratelimit import ( + UNKNOWN_CLIENT, + Limiter, + RateLimit, + client_address, + ip_subject, + seconds_until_window_ends, + window_start, +) + +ENDPOINT = "/f/contact-form" +FORM = {"email": "dev@example.com"} + +# Mid-window for a sixty second window, so the wait a refusal reports is a round +# number and the boundary the next window starts on is unambiguous. +NOON = datetime(2026, 8, 24, 12, 0, 30, tzinfo=UTC) + +MULTIPART_HEADERS = {"content-type": "multipart/form-data; boundary=hymical"} + + +class Clock: + """ + a stand-in for the wall clock that a test moves deliberately + """ + + def __init__(self, start: datetime) -> None: + """ + start the clock at an instant + :param start: the instant the clock reads to begin with + """ + self.now = start + + def advance(self, seconds: float) -> None: + """ + move the clock forward + :param seconds: how far forward to move it + """ + self.now += timedelta(seconds=seconds) + + def __call__(self) -> datetime: + """ + read the clock + :returns: the instant the clock currently reads + """ + return self.now + + +@pytest.fixture +def clock(monkeypatch: pytest.MonkeyPatch) -> Clock: + """ + freeze the clock the ingestion route stamps rate limit windows from + :param monkeypatch: pytest fixture used to replace the route's clock + :returns: the clock, which the test moves itself + """ + # A window is an interval of wall-clock time, so expiry can only be tested by + # controlling the clock. Sleeping through a real window would make the suite + # slow and would still be timing dependent. + frozen = Clock(NOON) + monkeypatch.setattr("hymical_forms.api.submissions.utcnow", frozen) + return frozen + + +def counters(client: TestClient) -> list[models.RateLimitCounter]: + """ + read every rate limit counter behind a client + :param client: the client whose application database should be inspected + :returns: the counter rows + """ + with open_session(client) as session: + return list(session.scalars(select(models.RateLimitCounter))) + + +def spent(client: TestClient, limiter: Limiter) -> int: + """ + total the attempts recorded against one limiter + :param client: the client whose application database should be inspected + :param limiter: which limiter's counters to total + :returns: the sum of the attempts it has recorded + """ + return sum(row.attempts for row in counters(client) if row.limiter == limiter) + + +def submit(client: TestClient, **kwargs: Any) -> int: + """ + post the default form and report only the status + :param client: the client to submit through + :param kwargs: extra arguments passed straight to the request + :returns: the response status code + """ + return client.post(ENDPOINT, data=FORM, **kwargs).status_code + + +def forwarded(address: str) -> dict[str, str]: + """ + build the header a reverse proxy would have appended + :param address: the address the proxy is claiming to have seen + :returns: an ``X-Forwarded-For`` header carrying it + """ + return {"X-Forwarded-For": address} + + +# --- configuration ----------------------------------------------------------- + + +def test_rate_limiting_is_enabled_by_default() -> None: + settings = build_settings() + + assert settings.rate_limit_enabled is True + assert settings.ip_rate_limit() == RateLimit(requests=60, window_seconds=60) + assert settings.endpoint_rate_limit() == RateLimit(requests=600, window_seconds=60) + + +def test_the_default_trust_model_reads_no_forwarding_header() -> None: + """ + a spoofable header must not become authoritative without an operator saying so + """ + settings = build_settings() + + assert settings.trusted_proxy_hops == 0 + assert settings.rate_limit_ip_secret is None + + +@pytest.mark.parametrize( + "override", + [ + {"rate_limit_ip_requests": 0}, + {"rate_limit_ip_requests": -1}, + {"rate_limit_endpoint_requests": 0}, + {"rate_limit_endpoint_requests": -5}, + ], +) +def test_a_limit_that_allows_nothing_is_refused(override: dict[str, Any]) -> None: + with pytest.raises(ValidationError): + build_settings(**override) + + +@pytest.mark.parametrize( + "override", + [ + {"rate_limit_ip_window_seconds": 0}, + {"rate_limit_ip_window_seconds": -60}, + {"rate_limit_endpoint_window_seconds": 0}, + {"rate_limit_endpoint_window_seconds": -1}, + ], +) +def test_a_window_with_no_length_is_refused(override: dict[str, Any]) -> None: + with pytest.raises(ValidationError): + build_settings(**override) + + +def test_a_negative_proxy_hop_count_is_refused() -> None: + with pytest.raises(ValidationError): + build_settings(trusted_proxy_hops=-1) + + +def test_a_pointlessly_short_address_secret_is_refused() -> None: + """ + a secret that is guessable does not make the digest it keys one-way + """ + with pytest.raises(ValidationError): + build_settings(rate_limit_ip_secret="short") + + +def test_disabling_rate_limiting_lets_every_attempt_through(make_client: ClientFactory) -> None: + client = make_client(rate_limit_enabled=False, rate_limit_ip_requests=1) + + statuses = [submit(client) for _ in range(4)] + + assert statuses == [202, 202, 202, 202] + assert counters(client) == [], "a disabled limiter still wrote counters" + + +# --- per source address ------------------------------------------------------ + + +def test_attempts_below_the_limit_are_accepted(make_client: ClientFactory, clock: Clock) -> None: + client = make_client(rate_limit_ip_requests=3) + + assert [submit(client) for _ in range(2)] == [202, 202] + + +def test_the_attempt_that_reaches_the_limit_is_still_accepted( + make_client: ClientFactory, clock: Clock +) -> None: + """ + the boundary is inclusive: a limit of three means three attempts, not two + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=3) + + assert [submit(client) for _ in range(3)] == [202, 202, 202] + assert spent(client, Limiter.IP) == 3 + + +def test_the_attempt_after_the_limit_is_refused(make_client: ClientFactory, clock: Clock) -> None: + client = make_client(rate_limit_ip_requests=3, rate_limit_ip_window_seconds=60) + for _ in range(3): + submit(client) + + response = client.post(ENDPOINT, data=FORM) + + assert response.status_code == 429 + body = response.json() + assert body["error"]["code"] == "rate_limit_exceeded" + assert body["error"]["details"] == { + "scope": "ip", + "limit": 3, + "window_seconds": 60, + "retry_after_seconds": 30, + } + + +def test_a_refusal_says_how_long_to_wait(make_client: ClientFactory, clock: Clock) -> None: + client = make_client(rate_limit_ip_requests=1, rate_limit_ip_window_seconds=60) + submit(client) + + response = client.post(ENDPOINT, data=FORM) + + # The clock sits thirty seconds into a sixty second window, so that is exactly + # how long is left of the window that refused it. + assert response.headers["Retry-After"] == "30" + + +def test_the_refusal_names_no_internal_state(make_client: ClientFactory, clock: Clock) -> None: + """ + a 429 must not become a way to read the counter table + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=1) + submit(client) + + body = client.post(ENDPOINT, data=FORM).json() + + rendered = repr(body) + assert "rate_limit_counters" not in rendered + assert "subject" not in rendered + assert "window_start" not in rendered + assert "testclient" not in rendered + + +def test_the_budget_returns_with_the_next_window(make_client: ClientFactory, clock: Clock) -> None: + """ + a fixed window has to actually end, and only on its boundary + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=1, rate_limit_ip_window_seconds=60) + submit(client) + + # One second short of the boundary the window is still the same window. + clock.advance(29) + assert submit(client) == 429 + + clock.advance(1) + assert submit(client) == 202 + + +def test_one_source_does_not_spend_another_sources_budget( + make_client: ClientFactory, clock: Clock +) -> None: + client = make_client(trusted_proxy_hops=1, rate_limit_ip_requests=2) + for _ in range(2): + submit(client, headers=forwarded("203.0.113.1")) + + assert submit(client, headers=forwarded("203.0.113.1")) == 429 + assert submit(client, headers=forwarded("203.0.113.2")) == 202 + + +# --- per endpoint ------------------------------------------------------------ + + +def test_an_endpoint_reaches_its_own_limit(make_client: ClientFactory, clock: Clock) -> None: + client = make_client(rate_limit_endpoint_requests=2, rate_limit_ip_requests=100) + + statuses = [submit(client) for _ in range(3)] + + assert statuses == [202, 202, 429] + assert client.post(ENDPOINT, data=FORM).json()["error"]["details"]["scope"] == "endpoint" + + +def test_another_endpoint_stays_usable(make_client: ClientFactory, clock: Clock) -> None: + """ + one endpoint being flooded must not take the rest of the service down with it + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_endpoint_requests=2, rate_limit_ip_requests=100) + create_endpoint(client, "second-form", name="Second form") + for _ in range(3): + submit(client) + + assert submit(client) == 429 + assert client.post("/f/second-form", data=FORM).status_code == 202 + + +def test_distributed_sources_still_spend_the_endpoint_budget( + make_client: ClientFactory, clock: Clock +) -> None: + """ + the endpoint limit is what answers an attack spread across many addresses + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client( + trusted_proxy_hops=1, + rate_limit_endpoint_requests=3, + rate_limit_ip_requests=100, + ) + + statuses = [submit(client, headers=forwarded(f"203.0.113.{n}")) for n in range(1, 6)] + + # Every attempt came from a different address and none of them exhausted an + # address budget, so only the shared endpoint budget can have refused them. + assert statuses == [202, 202, 202, 429, 429] + assert spent(client, Limiter.ENDPOINT) == 5 + + +def test_the_endpoint_budget_returns_with_the_next_window( + make_client: ClientFactory, clock: Clock +) -> None: + client = make_client( + rate_limit_endpoint_requests=1, + rate_limit_endpoint_window_seconds=60, + rate_limit_ip_requests=100, + ) + submit(client) + assert submit(client) == 429 + + clock.advance(30) + + assert submit(client) == 202 + + +# --- both limits together ---------------------------------------------------- + + +def test_the_source_limit_can_be_the_one_that_refuses( + make_client: ClientFactory, clock: Clock +) -> None: + client = make_client(rate_limit_ip_requests=1, rate_limit_endpoint_requests=100) + submit(client) + + body = client.post(ENDPOINT, data=FORM).json() + + assert body["error"]["details"]["scope"] == "ip" + + +def test_the_endpoint_limit_can_be_the_one_that_refuses( + make_client: ClientFactory, clock: Clock +) -> None: + client = make_client(rate_limit_ip_requests=100, rate_limit_endpoint_requests=1) + submit(client) + + body = client.post(ENDPOINT, data=FORM).json() + + assert body["error"]["details"]["scope"] == "endpoint" + + +def test_an_attempt_the_endpoint_limit_refuses_still_spends_the_source_budget( + make_client: ClientFactory, clock: Clock +) -> None: + """ + otherwise one saturated endpoint becomes a free target for one address + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=4, rate_limit_endpoint_requests=1) + + statuses = [submit(client) for _ in range(4)] + + # The first attempt spent the endpoint's whole budget and the next three were + # refused by it, but all four spent a unit of the address budget, so the fifth + # is refused by the address limit rather than the endpoint one. + assert statuses == [202, 429, 429, 429] + assert spent(client, Limiter.IP) == 4 + assert client.post(ENDPOINT, data=FORM).json()["error"]["details"]["scope"] == "ip" + + +def test_an_attempt_the_source_limit_refuses_does_not_spend_the_endpoint_budget( + make_client: ClientFactory, clock: Clock +) -> None: + """ + a blocked address must not be able to burn the budget of an endpoint it cannot reach + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=1, rate_limit_endpoint_requests=100) + submit(client) + for _ in range(5): + assert submit(client) == 429 + + assert spent(client, Limiter.IP) == 6 + assert spent(client, Limiter.ENDPOINT) == 1 + + +def test_the_wait_belongs_to_the_limiter_that_refused( + make_client: ClientFactory, clock: Clock +) -> None: + """ + two limiters can run different windows, so the wait must come from the right one + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client( + rate_limit_ip_requests=100, + rate_limit_ip_window_seconds=60, + rate_limit_endpoint_requests=1, + rate_limit_endpoint_window_seconds=10, + ) + submit(client) + + response = client.post(ENDPOINT, data=FORM) + + # The address window has thirty seconds left and the endpoint window has ten. + # The endpoint limiter is the one that refused, so ten is the honest answer. + assert response.headers["Retry-After"] == "10" + assert response.json()["error"]["details"]["scope"] == "endpoint" + + +# --- what an attempt is ------------------------------------------------------ + + +def test_an_oversized_body_is_refused_before_the_limiter_sees_it( + make_client: ClientFactory, clock: Clock +) -> None: + """ + the body cap runs in middleware, so the limiter never has to hold the body + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(max_body_bytes=64, rate_limit_ip_requests=1) + + response = client.post(ENDPOINT, content=b"note=" + b"x" * 200, headers=URLENCODED_HEADERS) + + assert response.status_code == 413 + assert counters(client) == [], "a body the middleware refused reached the limiter" + assert submit(client) == 202, "an oversized body spent a budget it never reached" + + +def test_an_unknown_endpoint_spends_only_the_source_budget( + client: TestClient, clock: Clock +) -> None: + """ + a guessed identifier must not let an attacker choose how much this table grows + :param client: test client whose app holds the default endpoint + :param clock: the frozen clock the route stamps windows from + """ + assert client.post("/f/no-such-form", data=FORM).status_code == 404 + + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 0 + + +def test_a_malformed_endpoint_id_still_spends_the_source_budget( + client: TestClient, clock: Clock +) -> None: + assert client.post("/f/NOPE", data=FORM).status_code == 404 + + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 0 + + +def test_a_malformed_body_still_spends_both_budgets(client: TestClient, clock: Clock) -> None: + """ + invalid traffic still costs this service work, so it still costs the sender budget + :param client: test client whose app holds the default endpoint + :param clock: the frozen clock the route stamps windows from + """ + response = client.post(ENDPOINT, content=b"not multipart", headers=MULTIPART_HEADERS) + + assert response.status_code == 400 + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 1 + + +def test_an_unsupported_content_type_still_spends_both_budgets( + client: TestClient, clock: Clock +) -> None: + response = client.post(ENDPOINT, content=b"{}", headers={"content-type": "application/json"}) + + assert response.status_code == 415 + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 1 + + +def test_an_empty_submission_still_spends_both_budgets(client: TestClient, clock: Clock) -> None: + response = client.post(ENDPOINT, content=b"", headers=URLENCODED_HEADERS) + + assert response.status_code == 422 + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 1 + + +def test_an_inactive_endpoint_still_spends_both_budgets( + make_client: ClientFactory, clock: Clock +) -> None: + """ + a disabled endpoint must not become a free target either + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(seed_endpoint=False) + create_endpoint(client, "contact-form", is_active=False) + + assert submit(client) == 409 + + assert spent(client, Limiter.IP) == 1 + assert spent(client, Limiter.ENDPOINT) == 1 + + +# --- idempotency ------------------------------------------------------------- + + +def key(value: str = "a") -> dict[str, str]: + """ + build an idempotency header long enough to satisfy the key rules + :param value: the character the key is built from + :returns: an ``Idempotency-Key`` header + """ + return {"Idempotency-Key": value * 32} + + +def test_a_replay_the_limiter_allows_still_returns_the_original_submission( + client: TestClient, clock: Clock +) -> None: + first = client.post(ENDPOINT, data=FORM, headers=key()).json() + + second = client.post(ENDPOINT, data=FORM, headers=key()).json() + + assert second["submission_id"] == first["submission_id"] + assert second["idempotent_replay"] is True + + +def test_a_replay_spends_the_budget_like_any_other_attempt( + make_client: ClientFactory, clock: Clock +) -> None: + """ + otherwise one leaked key would be an unlimited way past the limits + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=2) + + assert client.post(ENDPOINT, data=FORM, headers=key()).status_code == 202 + assert client.post(ENDPOINT, data=FORM, headers=key()).status_code == 202 + assert client.post(ENDPOINT, data=FORM, headers=key()).status_code == 429 + + +def test_a_refused_retry_changes_nothing_that_was_already_stored( + make_client: ClientFactory, clock: Clock +) -> None: + """ + a 429 is a refusal to do work, not a partial one + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(seed_endpoint=False, rate_limit_ip_requests=1) + create_endpoint(client, "contact-form", webhook_url="https://example.com/hooks") + stored = client.post(ENDPOINT, data=FORM, headers=key()).json() + + assert client.post(ENDPOINT, data=FORM, headers=key()).status_code == 429 + + with open_session(client) as session: + submissions = list(session.scalars(select(models.Submission))) + deliveries = list(session.scalars(select(models.WebhookDelivery))) + assert [row.id for row in submissions] == [stored["submission_id"]] + assert [row.submission_id for row in deliveries] == [stored["submission_id"]] + assert deliveries[0].attempts == 0 + + +def test_a_conflicting_key_is_still_a_conflict_when_the_limiter_allows_it( + client: TestClient, clock: Clock +) -> None: + client.post(ENDPOINT, data=FORM, headers=key()) + + response = client.post(ENDPOINT, data={"email": "other@example.com"}, headers=key()) + + assert response.status_code == 409 + assert response.json()["error"]["code"] == "idempotency_conflict" + + +# --- routes the form limits must not touch ----------------------------------- + + +def test_management_routes_are_not_bound_by_the_form_limits( + make_client: ClientFactory, clock: Clock +) -> None: + """ + ingestion traffic must not be able to lock an operator out of their own service + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=1) + submit(client) + assert submit(client) == 429 + + assert [client.get("/endpoints").status_code for _ in range(5)] == [200] * 5 + assert client.get("/deliveries").status_code == 200 + assert create_endpoint(client, "another-form")["id"] == "another-form" + + +def test_health_is_unaffected(make_client: ClientFactory, clock: Clock) -> None: + client = make_client(rate_limit_ip_requests=1) + submit(client) + assert submit(client) == 429 + + assert [client.get("/health").status_code for _ in range(5)] == [200] * 5 + + +def test_health_spends_nothing(client: TestClient, clock: Clock) -> None: + for _ in range(5): + client.get("/health") + + assert counters(client) == [] + + +# --- cleanup ----------------------------------------------------------------- + + +def test_old_windows_can_be_removed_without_touching_the_current_one( + client: TestClient, +) -> None: + """ + claiming these rows are harmless would be untrue, so they have to be removable + :param client: test client whose app holds the default endpoint + """ + subject = "a" * 64 + stale = NOON - timedelta(hours=1) + with open_session(client) as session: + for start in (stale, NOON): + storage.consume_rate_limit( + session, limiter=Limiter.IP, subject=subject, window_start=start + ) + + removed = storage.delete_expired_rate_limit_counters( + session, before=NOON - timedelta(minutes=5) + ) + + assert removed == 1 + assert [row.window_start for row in counters(client)] == [NOON] + + +def test_a_submission_sweeps_old_windows_when_it_draws_the_short_straw( + make_client: ClientFactory, clock: Clock, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + the sweep is wired into the request path, not only available to be called + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + :param monkeypatch: pytest fixture used to make the sweep certain rather than rare + """ + client = make_client(rate_limit_ip_window_seconds=60, rate_limit_endpoint_window_seconds=60) + with open_session(client) as session: + storage.consume_rate_limit( + session, + limiter=Limiter.IP, + subject="a" * 64, + window_start=NOON - timedelta(hours=1), + ) + + # A sweep happens on a small fraction of attempts, so a test that wants to see + # one has to stop it being a coin toss. + monkeypatch.setattr("hymical_forms.api.submissions.random.random", lambda: 0.0) + assert submit(client) == 202 + + windows = {row.window_start for row in counters(client)} + assert windows == {window_start(NOON, 60)}, "the sweep took a window still in use" + + +def test_spending_a_budget_reports_a_rising_count(client: TestClient) -> None: + with open_session(client) as session: + counts = [ + storage.consume_rate_limit( + session, limiter=Limiter.ENDPOINT, subject="contact-form", window_start=NOON + ) + for _ in range(4) + ] + + assert counts == [1, 2, 3, 4] + + +def test_the_two_limiters_do_not_share_a_counter(client: TestClient) -> None: + """ + a subject in one limiter must never be the same row as the same string in the other + :param client: test client whose app holds the default endpoint + """ + with open_session(client) as session: + first = storage.consume_rate_limit( + session, limiter=Limiter.IP, subject="shared", window_start=NOON + ) + second = storage.consume_rate_limit( + session, limiter=Limiter.ENDPOINT, subject="shared", window_start=NOON + ) + + assert (first, second) == (1, 1) + assert len(counters(client)) == 2 + + +# --- the client address trust model ------------------------------------------ + + +def test_a_forwarding_header_is_ignored_by_default( + make_client: ClientFactory, clock: Clock +) -> None: + """ + a client that could pick its own bucket would have no rate limit at all + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(rate_limit_ip_requests=1) + + assert submit(client, headers=forwarded("203.0.113.1")) == 202 + assert submit(client, headers=forwarded("203.0.113.2")) == 429 + + +def test_a_trusted_hop_counts_the_address_that_proxy_saw( + make_client: ClientFactory, clock: Clock +) -> None: + """ + everything to the left of your own proxy's entry was written by somebody else + :param make_client: factory for clients bound to a configured app + :param clock: the frozen clock the route stamps windows from + """ + client = make_client(trusted_proxy_hops=1, rate_limit_ip_requests=1) + + assert submit(client, headers=forwarded("1.1.1.1, 203.0.113.9")) == 202 + # A different forged prefix, the same real client: still one bucket. + assert submit(client, headers=forwarded("2.2.2.2, 203.0.113.9")) == 429 + assert submit(client, headers=forwarded("2.2.2.2, 203.0.113.8")) == 202 + + +def test_a_trusted_hop_falls_back_to_the_peer_without_a_header( + make_client: ClientFactory, clock: Clock +) -> None: + client = make_client(trusted_proxy_hops=1, rate_limit_ip_requests=1) + + assert submit(client) == 202 + assert submit(client) == 429 + + +@pytest.mark.parametrize( + ("peer", "forwarded_for", "hops", "expected"), + [ + ("198.51.100.4", "203.0.113.1", 0, "198.51.100.4"), + ("198.51.100.4", None, 0, "198.51.100.4"), + ("198.51.100.4", "203.0.113.1", 1, "203.0.113.1"), + ("198.51.100.4", "1.1.1.1, 203.0.113.1", 1, "203.0.113.1"), + ("198.51.100.4", " 1.1.1.1 , 203.0.113.1 ", 1, "203.0.113.1"), + ("198.51.100.4", "1.1.1.1, 203.0.113.1", 2, "1.1.1.1"), + # A chain shorter than the operator described is not the chain they + # described, so it is discarded rather than half believed. + ("198.51.100.4", "203.0.113.1", 2, "198.51.100.4"), + ("198.51.100.4", "", 1, "198.51.100.4"), + (None, None, 0, UNKNOWN_CLIENT), + (None, "203.0.113.1", 0, UNKNOWN_CLIENT), + ], +) +def test_the_client_address_is_resolved_from_the_declared_trust_model( + peer: str | None, forwarded_for: str | None, hops: int, expected: str +) -> None: + resolved = client_address(peer=peer, forwarded_for=forwarded_for, trusted_proxy_hops=hops) + + assert resolved == expected + + +# --- the rules underneath ---------------------------------------------------- + + +def test_a_window_is_floored_against_the_epoch() -> None: + """ + every process has to derive the same boundary or they are not sharing a counter + """ + start = window_start(datetime(2026, 8, 24, 12, 0, 59, 999999, tzinfo=UTC), 60) + + assert start == datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + assert window_start(datetime(2026, 8, 24, 12, 1, tzinfo=UTC), 60) == datetime( + 2026, 8, 24, 12, 1, tzinfo=UTC + ) + + +def test_the_wait_is_rounded_up_to_whole_seconds() -> None: + start = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + + remaining = seconds_until_window_ends( + datetime(2026, 8, 24, 12, 0, 30, 500000, tzinfo=UTC), start, 60 + ) + + # 29.5 seconds are left, and answering 29 would invite a retry the same window + # is still going to refuse. + assert remaining == 30 + + +def test_the_wait_is_never_shorter_than_a_second() -> None: + start = datetime(2026, 8, 24, 12, 0, tzinfo=UTC) + + remaining = seconds_until_window_ends( + datetime(2026, 8, 24, 12, 0, 59, 999999, tzinfo=UTC), start, 60 + ) + + assert remaining == 1 + + +def test_a_hashed_address_does_not_contain_the_address() -> None: + subject = ip_subject("198.51.100.4", None) + + assert "198.51.100.4" not in subject + assert len(subject) == 64 + + +def test_the_digest_of_an_address_is_stable() -> None: + """ + two processes must agree on the subject or they are counting different things + """ + assert ip_subject("198.51.100.4", None) == ip_subject("198.51.100.4", None) + assert ip_subject("198.51.100.4", "s" * 32) == ip_subject("198.51.100.4", "s" * 32) + + +def test_a_secret_changes_what_an_address_digests_to() -> None: + plain = ip_subject("198.51.100.4", None) + keyed = ip_subject("198.51.100.4", "s" * 32) + + assert plain != keyed + assert keyed != ip_subject("198.51.100.4", "t" * 32) + + +def test_different_addresses_digest_differently() -> None: + assert ip_subject("198.51.100.4", None) != ip_subject("198.51.100.5", None)