Add distributed public ingestion rate limiting

.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

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