How to Harden a FastAPI Rate Limiter Against Spoofed Client IPs, IPv6 Address Rotation, and Wrong Retry-After Headers

A rate limiter looks like a solved problem as soon as you have a token bucket. The arithmetic is about twenty lines of Python, and our earlier tutorial on building rate limiters walks through the common algorithms from scratch. What goes wrong in production is everything around the arithmetic: the limiter counts the wrong thing as a client, it tells refused clients to come back at the wrong moment, or it quietly stops being correct after a refactor and nobody notices.

In this tutorial you will harden a small FastAPI service against those three failures, and you will measure every fix instead of taking it on trust. You do not need the earlier tutorial, because Step 2 recaps the one algorithm we use. If you also want the fixed-window and sliding-window variants, read that one first.

What you will build, and the words you need first

Rate limiting means giving each client a budget of requests and refusing the ones that go over it. A refusal uses the HTTP status code 429 (Too Many Requests, defined in RFC 6585). The hard word in that sentence is client. Your server does not know who is calling. It has the address of the machine at the other end of the connection and some headers the caller wrote, and the limiter needs a key, a short string such as an IP address, so that it can keep one budget per key. RFC 6585 says as much: “this specification does not define how the origin server identifies the user, nor how it counts requests”. Choosing the key is your job, and most of the damage happens when it is chosen badly.

By the end you will have:

• a token bucket, and an equivalent version that keeps one number per client (it is called GCRA);
• a pure ASGI middleware that puts the limiter in front of FastAPI, with a background task that forgets idle clients;
• a client key that a forged X-Forwarded-For header cannot talk its way around, and that treats an IPv6 /64 as one client, checked against six proxy setups on a real uvicorn server;
• a Retry-After header that a well-behaved client can obey;
• 29 tests, including property-based tests that run on a fake clock with exact arithmetic, and ten deliberate bugs that those tests catch.

This is not a defence against a flood. A limiter inside your application still spends CPU on every request it refuses, so volumetric attacks need limits at the edge (a CDN, a web application firewall or your reverse proxy). And the counters here live in the memory of one process, which the last section comes back to.

Prerequisites

• Python 3.11 or newer. Every command below ran on Python 3.13.14 on Windows 11. I read Requires-Python in each pinned package’s metadata: Starlette 1.9.0, httpx2 2.16.0 and Hypothesis 6.170.8 need 3.11, the others 3.10.
• A bash shell and curl. On Windows I used Git Bash. Linux and macOS need nothing extra.
• Packages: FastAPI 0.161.1 (which brings Starlette 1.9.0), Uvicorn 0.54.0, httpx2 2.16.0 (Starlette’s test client refuses to start without it), requests 2.34.2, pytest 9.1.1 and Hypothesis 6.170.8.
• Background: you can read a Python class and a dictionary, and you know what an HTTP request, a header and a status code are. No Redis, no Docker and no cloud account are needed.

Step 1: Create the project and a clock you can move by hand

Start with an empty folder and a virtual environment (a private set of installed packages for this project):

$ mkdir ratelimit-lab
$ cd ratelimit-lab
$ python -m venv .venv
$ source .venv/bin/activate
$ python -m pip install fastapi==0.161.1 uvicorn==0.54.0 httpx2==2.16.0 requests==2.34.2 pytest==9.1.1 hypothesis==6.170.8

On Windows in Git Bash, the activation line is source .venv/Scripts/activate. All the files below go in this one folder. The first line of every file is a comment with its name, so you can tell which file a code block belongs to, and a block whose comment says (continued) goes at the end of the file.

A rate limiter is all about time, and code that reads the clock is hard to test: you would have to sleep through real seconds, and the results would change from run to run. So every limiter in this tutorial takes a clock argument, a function that returns the current time in seconds. In production it is time.monotonic, which the Python documentation describes as “a clock that cannot go backwards”, adding that “The clock is not affected by system clock updates”. That matters here, because the wall clock time.time() can jump when a machine’s clock is corrected: a jump forward would refill every bucket at once, and a jump backward would make the elapsed time negative. In tests the clock is a FakeClock that only moves when you tell it to. Create labkit.py:

# labkit.py
class FakeClock:
    """A clock you move by hand, so every experiment is exact and repeatable."""

    def __init__(self, start=0):
        self.now = start

    def __call__(self):
        return self.now

    def set(self, moment):
        self.now = moment

    def advance(self, seconds):
        self.now += seconds

FakeClock is callable, so it can stand in for time.monotonic. The method set jumps to a moment and advance moves forward. Check that it works:

$ python -c "from labkit import FakeClock; c = FakeClock(); c.advance(2.5); print(c())"
2.5

Step 2: A token bucket in twenty lines

Picture a bucket that holds at most capacity tokens. Tokens drip in at a steady rate, and every request must take one token out. If the bucket has a token, the request goes through; if it is empty, the request is refused. Wikipedia’s description of the algorithm is short: when a request arrives, “the bucket is inspected to see if it contains sufficient tokens at that time”. An idle client slowly refills its bucket, so it earns the right to a short burst, but it can never go faster than the refill rate for long.

A real implementation does not need a timer that adds tokens now and then. It only needs to know how long it has been since it last looked: when a request arrives, add elapsed × rate tokens (never more than the capacity) and then decide. That is lazy refill, and it means an idle client costs nothing. Create limiters.py:

# limiters.py
"""Two rate limiters with one interface: limiter.check(key) -> Decision."""
import math
import time
from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class Decision:
    allowed: bool
    retry_after: float  # seconds until a retry could succeed (0.0 when allowed)
    remaining: int      # further requests that would be allowed right now


class Limiter:
    """Shared plumbing: a clock you can replace in tests and one state entry per key."""

    def __init__(self, clock=time.monotonic):
        self.clock = clock
        self.state = {}

    def purge(self):
        """Forget keys whose state equals a brand new key's state. Returns how many were dropped."""
        now = self.clock()
        idle = [key for key, saved in self.state.items() if self._idle(saved, now)]
        for key in idle:
            del self.state[key]
        return len(idle)


class TokenBucket(Limiter):
    """A bucket holds up to `capacity` tokens and refills at `rate` tokens per second; a request takes one."""

    def __init__(self, rate, capacity, clock=time.monotonic):
        super().__init__(clock)
        self.rate, self.capacity = rate, capacity

    def check(self, key):
        now = self.clock()
        tokens, stamp = self.state.get(key, (self.capacity, now))
        tokens = min(self.capacity, tokens + (now - stamp) * self.rate)
        if tokens >= 1:
            self.state[key] = (tokens - 1, now)
            return Decision(True, 0.0, int(tokens - 1))
        self.state[key] = (tokens, now)
        return Decision(False, (1 - tokens) / self.rate, 0)

    def _idle(self, saved, now):  # used by purge() in Step 7
        tokens, stamp = saved
        return tokens + (now - stamp) * self.rate >= self.capacity

Read it from the top. Decision is the answer every limiter gives: whether the request is allowed, how many seconds the client should wait before trying again (retry_after, zero when allowed) and how many more requests would be accepted right now (remaining). Limiter holds the clock and a dictionary called state with one entry per key; its purge method and the _idle methods belong to Step 4, so skip them for now. In TokenBucket.check a client we have never seen starts with a full bucket, the refill is computed from the elapsed time, and the two branches either spend a token or work out the wait: the missing part of a token divided by the rate, (1 - tokens) / self.rate.

Now a first experiment. step02_bucket.py asks a bucket for seven requests at the same instant, then lets time pass:

# step02_bucket.py
from labkit import FakeClock
from limiters import TokenBucket

clock = FakeClock(0.0)
bucket = TokenBucket(rate=0.5, capacity=5, clock=clock)  # one token every 2 seconds, at most 5 saved up


def ask(label):
    decision = bucket.check("client")
    print(f"t={clock():4.1f} s  {label:20} allowed={decision.allowed!s:5}  remaining={decision.remaining}  retry_after={decision.retry_after:.1f}")


for number in range(1, 8):
    ask(f"request {number} at once")
clock.advance(1.0)
ask("1 s later")
clock.advance(1.0)
ask("1 s after that")

Run it with python step02_bucket.py. You should see:

t= 0.0 s  request 1 at once    allowed=True   remaining=4  retry_after=0.0
t= 0.0 s  request 2 at once    allowed=True   remaining=3  retry_after=0.0
t= 0.0 s  request 3 at once    allowed=True   remaining=2  retry_after=0.0
t= 0.0 s  request 4 at once    allowed=True   remaining=1  retry_after=0.0
t= 0.0 s  request 5 at once    allowed=True   remaining=0  retry_after=0.0
t= 0.0 s  request 6 at once    allowed=False  remaining=0  retry_after=2.0
t= 0.0 s  request 7 at once    allowed=False  remaining=0  retry_after=2.0
t= 1.0 s  1 s later            allowed=False  remaining=0  retry_after=1.0
t= 2.0 s  1 s after that       allowed=True   remaining=0  retry_after=0.0

The bucket holds 5 tokens and refills one every 2 seconds. Requests 1 to 5 drain it, and remaining counts down from 4 to 0. Requests 6 and 7 are refused, and each is told to wait 2.0 seconds, which is one token at 0.5 tokens per second. After one second the bucket holds half a token, so the request is still refused, but the advertised wait has shrunk to 1.0 second. After one more second it holds exactly one token and the request is allowed. The numbers are exact here because 0.5 and 1.0 happen to be exactly representable as binary floating point numbers; Step 3 shows what happens when they are not.

Remember the promise this bucket makes, because we will test it in Step 9: in any stretch of T seconds it lets through at most capacity + rate × T requests. With a capacity of 5 and a rate of 0.5 per second, that is at most 10 requests in any 10 seconds, even though the sustained speed is only one request every 2 seconds. A token bucket is a long-run rate (here 5 requests per 10 seconds) plus a burst allowance (5 more). If you want to promise no more than 10 in any 10 seconds, pick the capacity and the rate so that capacity + rate × 10 comes to 10.

Check before moving on

Your output should match the block above line for line. If the sixth request is allowed, check that capacity is 5 and that the experiment uses FakeClock, not the real clock.

Step 3: GCRA, the same bucket with one number per client

The token bucket stores two things per client: how many tokens it has and when we last looked. There is an equivalent algorithm that stores one. The generic cell rate algorithm (GCRA) was designed for network traffic control, and Wikipedia says that its standards “describe the GCRA in two equivalent ways: as a virtual scheduling algorithm and as a continuous state leaky bucket algorithm”. The same article adds that “The GCRA should be considered as a version of the leaky bucket as a meter rather than the leaky bucket as a queue”. In other words, it counts requests and decides yes or no; it never holds requests back in a queue.

Here is the idea in plain words. Give every client a schedule. If a client is allowed one request every interval seconds, then after each accepted request the next request is on schedule one interval later. GCRA stores that moment, the theoretical arrival time (TAT). A request that arrives after the TAT is late and always welcome. A request that arrives before it is early, and it is accepted only if it is not too early: the tolerance, which is the burst size minus one, times the interval, says how far ahead of schedule a client may be. Each accepted request pushes the TAT one interval into the future, so a client that bursts runs far ahead of its schedule and then has to wait for the schedule to catch up. Add the class to the end of limiters.py:

# limiters.py (continued)
class GCRA(Limiter):
    """Generic cell rate algorithm: one number per key, the theoretical arrival time (TAT) of the next request."""

    def __init__(self, rate, burst, clock=time.monotonic):
        super().__init__(clock)
        self.interval = 1 / rate                      # one request every `interval` seconds on average
        self.tolerance = (burst - 1) * self.interval  # how far ahead of schedule a request may arrive

    def check(self, key):
        now = self.clock()
        tat = max(self.state.get(key, now), now)
        if now < tat - self.tolerance:
            return Decision(False, tat - self.tolerance - now, 0)
        tat += self.interval
        self.state[key] = tat
        remaining = math.floor((self.tolerance + now - tat) / self.interval) + 1
        return Decision(True, 0.0, max(0, remaining))

    def _idle(self, tat, now):  # used by purge() in Step 7
        return tat <= now

Two lines in check carry the idea. max(self.state.get(key, now), now) says that a client that has been quiet cannot bank unlimited credit: its schedule never lags behind the present. The if line refuses a request that is earlier than tat - self.tolerance, and how much earlier it is is exactly the wait. The two lines after it push the schedule one interval forward and save it, and the rest works out remaining.

Why is this the same as the bucket? A bucket with burst tokens that refills one token per interval holds burst - (TAT - now) / interval tokens at any moment, and a request needs at least one token. Rearranging that inequality gives now >= TAT - tolerance, the GCRA rule. Claims like that deserve a test, so step03_gcra.py feeds the same 100,000 requests from three clients to both limiters and compares every answer, once with exact fractions and once with ordinary floating point numbers:

# step03_gcra.py
import random
from fractions import Fraction

from labkit import FakeClock
from limiters import GCRA, TokenBucket

rng = random.Random(11)
# 100,000 requests from 3 clients, 0 to 400 ms apart: about 1.7 requests per second per client against a rate of 1
steps = [(rng.randrange(0, 400), rng.randrange(3)) for _ in range(100_000)]


def run(number):
    """Replay the steps with times of type `number`: Fraction is exact arithmetic, float is the usual kind."""
    bucket_clock, gcra_clock = FakeClock(number(0)), FakeClock(number(0))
    bucket = TokenBucket(number(1), 10, bucket_clock)
    gcra = GCRA(number(1), 10, gcra_clock)
    now = number(0)
    allowed = flags = remaining = retries = 0
    biggest_gap = 0.0
    for millis, key in steps:
        now = now + number(millis) / 1000
        bucket_clock.set(now)
        gcra_clock.set(now)
        a, b = bucket.check(key), gcra.check(key)
        allowed += a.allowed
        flags += a.allowed != b.allowed
        remaining += a.remaining != b.remaining
        retries += a.retry_after != b.retry_after
        biggest_gap = max(biggest_gap, abs(float(a.retry_after - b.retry_after)))
    return allowed, flags, remaining, retries, biggest_gap


for label, number in (("exact (Fraction)", Fraction), ("float", float)):
    allowed, flags, remaining, retries, gap = run(number)
    print(f"{label:17} allowed {allowed} of {len(steps)}; allow/refuse differs {flags}, remaining differs {remaining}, "
          f"retry_after differs {retries} (largest gap {gap:.1e} s)")

Python’s fractions module “provides support for rational number arithmetic”, so Fraction(1, 3) is exactly one third, with no rounding. Run python step03_gcra.py:

exact (Fraction)  allowed 59900 of 100000; allow/refuse differs 0, remaining differs 0, retry_after differs 0 (largest gap 0.0e+00 s)
float             allowed 59900 of 100000; allow/refuse differs 0, remaining differs 1, retry_after differs 36934 (largest gap 5.8e-13 s)

With exact arithmetic the two limiters agreed on all 100,000 requests: same allow or refuse answer, same retry_after, same remaining. With floats the allow-or-refuse answers still matched every time, but the retry_after values differed in 36,934 decisions, by at most 5.8e-13 seconds (that is, 0.00000000000058 s), and remaining differed once. Nothing is wrong. Two formulas that are equal on paper round differently in the last binary digit, and that is the reason the test suite in Step 9 uses Fraction: a test should fail because the code is wrong, not because 0.1 + 0.2 is not 0.3.

Step 4: What a client costs in memory, and how to forget idle ones

Which one should you ship? Both make the same decisions, so compare what they cost. step04_memory.py measures the bytes of state per client with tracemalloc (the same tool we used in our tutorial on finding memory leaks), measures the time of one check, and then floods a limiter with 200,000 clients that each make a single request:

# step04_memory.py
import time
import tracemalloc

from labkit import FakeClock
from limiters import GCRA, TokenBucket

MAKERS = {
    "token bucket": lambda clock: TokenBucket(1.0, 10, clock),
    "GCRA": lambda clock: GCRA(1.0, 10, clock),
}
N = 50_000
KEYS = [f"client-{i}" for i in range(N)]  # built first, so the key strings are not counted


def bytes_per_key(make, requests_per_key):
    clock = FakeClock(1000.0)
    limiter = make(clock)
    tracemalloc.start()
    before = tracemalloc.get_traced_memory()[0]
    for key in KEYS:
        for _ in range(requests_per_key):
            clock.advance(0.001)  # distinct timestamps, like real traffic
            limiter.check(key)
    after = tracemalloc.get_traced_memory()[0]
    tracemalloc.stop()
    return (after - before) / N


def nanoseconds_per_check(make, calls=200_000):
    clock = FakeClock(0.0)
    limiter = make(clock)
    names = [f"c{i}" for i in range(1000)]
    start = time.perf_counter()
    for i in range(calls):
        clock.now = i * 0.001
        limiter.check(names[i % 1000])
    return (time.perf_counter() - start) / calls * 1e9


print(f"{'':18}{'1 request/key':>15}{'10 requests/key':>17}{'time per check':>16}")
for name, make in MAKERS.items():
    print(f"{name:18}{bytes_per_key(make, 1):>13.0f} B{bytes_per_key(make, 10):>15.0f} B{nanoseconds_per_check(make):>13.0f} ns")

# a flood of one-request clients, then purge
clock = FakeClock(0.0)
bucket = TokenBucket(rate=2, capacity=5, clock=clock)
for i in range(200_000):
    bucket.check(f"spoofed-{i}")
print("\nkeys after a flood of one-request clients:", len(bucket.state))
clock.advance(0.2)
print("purge 0.2 s later:", bucket.purge(), "dropped (their buckets are not full yet)")
clock.advance(2.0)
for _ in range(30):
    bucket.check("heavy")  # one client keeps draining its bucket
print("purge 2.2 s later:", bucket.purge(), "dropped; keys left:", list(bucket.state))

Run it with python step04_memory.py:

                    1 request/key  10 requests/key  time per check
token bucket                118 B            140 B          420 ns
GCRA                         62 B             62 B          404 ns

keys after a flood of one-request clients: 200000
purge 0.2 s later: 0 dropped (their buckets are not full yet)
purge 2.2 s later: 200000 dropped; keys left: ['heavy']

GCRA needs 62 bytes per client against 118 to 140 for the bucket, because it keeps one float where the bucket keeps a tuple of two. Both take about 0.4 microseconds per check on this machine, so the processor is not the problem. The number of clients is. Every key you have ever seen stays in the dictionary, and if an attacker can invent keys, the dictionary grows without bound. Steps 7 and 8 show how easily keys can be invented: a header you believe, or an IPv6 address from a range the attacker controls.

The fix rests on one observation: a client whose bucket is full is indistinguishable from a client you have never seen. Both get a full burst on their next request, so you can delete the first kind without changing a single answer. That is what the _idle methods in limiters.py decide (for the bucket: would the refill since the last visit have filled it?; for GCRA: is the schedule already in the past?) and what Limiter.purge acts on.

In the output, the first purge, 0.2 seconds after the flood, dropped nothing, because each of the 200,000 buckets was still missing 0.6 of a token. The second purge, 2.2 seconds after the flood, dropped all 200,000 and kept only heavy, the one client that was still draining its bucket. Step 6 calls purge from a background task, and Step 9 proves that purging never changes an answer.

Step 5: Tell the client how long to wait

A refusal is only half of the answer. RFC 6585 says a 429 response “MAY include a Retry-After header indicating how long to wait before making a new request”, and RFC 9110 defines what the header may contain: “A delay-seconds value is a non-negative decimal integer, representing time in seconds.” That sentence hides a trap. Our limiters compute waits like 0.32 seconds, and the header can only carry whole seconds. Add the conversion function to the end of limiters.py:

# limiters.py (continued)
def retry_after_header(seconds):
    """Retry-After takes whole seconds (RFC 9110 delay-seconds), so round up, never down, and never to 0."""
    return str(max(1, math.ceil(seconds)))

The tempting version is str(int(seconds)), which rounds down. A wait of 0.32 seconds becomes Retry-After: 0, which tells the client to come back immediately. To see what that does, step05_retry_after.py simulates a client that empties a one-token bucket and then retries exactly as the header says, taking 50 ms per round trip, once with the rounded-down header and once with retry_after_header:

# step05_retry_after.py
from labkit import FakeClock
from limiters import TokenBucket, retry_after_header

ROUND_TRIP = 0.05  # a client needs 50 ms to send a request and read the answer


def round_down(seconds):
    return str(int(seconds))


def attempts_until_allowed(header, wait):
    """A client empties a one-token bucket, then retries exactly as the Retry-After header says."""
    clock = FakeClock(0.0)
    bucket = TokenBucket(rate=1 / wait, capacity=1, clock=clock)
    bucket.check("client")  # takes the only token
    attempts = 0
    while True:
        attempts += 1
        decision = bucket.check("client")
        if decision.allowed:
            return attempts, clock()
        clock.advance(int(header(decision.retry_after)) + ROUND_TRIP)


print(f"{'true wait':>10}{'int() header':>24}{'ceil() header':>24}")
for wait in (0.32, 1.72, 59.23):
    down = attempts_until_allowed(round_down, wait)
    up = attempts_until_allowed(retry_after_header, wait)
    print(f"{wait:>8} s{down[0]:>10} requests, {down[1]:>6.2f} s{up[0]:>10} requests, {up[1]:>6.2f} s")

Run python step05_retry_after.py:

 true wait            int() header           ceil() header
    0.32 s         8 requests,   0.35 s         2 requests,   1.05 s
    1.72 s        16 requests,   1.75 s         2 requests,   2.05 s
   59.23 s         6 requests,  59.25 s         2 requests,  60.05 s

With the rounded-down header the client sent 8, 16 and 6 requests before it got through, because a header of 0, or one that is short by a fraction of a second, sends it straight back. With the rounded-up header it sent exactly two: the refused one, and the one that succeeded. The price is honest: the polite client finished later, 1.05 seconds instead of 0.35 in the first row, because a whole-second header cannot express 0.32. That is the right trade. A server that is already refusing traffic should not also be hit by every refused client retrying in a tight loop, and max(1, ...) makes sure the header never says zero.

The IETF is working on richer headers. The draft RateLimit header fields for HTTP (revision 11 in May 2026, still an Internet-Draft) defines RateLimit-Policy and RateLimit fields that advertise the quota and what is left of it. Two of its sentences are worth knowing today: clients “MUST NOT assume that a positive available quota is a guarantee that further requests will be served”, and when a response carries both fields, “the Retry-After field value SHOULD NOT reference a point in time earlier than the end of the effective window”. Until it is final, Retry-After is the header every HTTP client library already understands.

Step 6: Put the limiter in front of FastAPI

Now connect the limiter to a web application. You could add a dependency to each route, but a middleware (code that wraps every request) is better here: it sees the request before routing, so a refused request costs the application almost nothing, and it also covers routes you add next month without anyone remembering to decorate them. FastAPI runs on ASGI, the Python standard for asynchronous web servers, and a plain ASGI middleware is a class with an __call__(scope, receive, send) method. The scope dictionary describes the connection; its type is http, websocket or lifespan (the server telling the app that it is starting up or shutting down). Our middleware limits only HTTP requests and passes everything else through untouched, so WebSocket connections are not limited by it; limit those separately if you expose any. Create middleware.py:

# middleware.py
import ipaddress
import json

from limiters import retry_after_header


class RateLimitMiddleware:
    """Pure ASGI middleware: ask the limiter first, so a refused request costs the app almost nothing."""

    def __init__(self, app, limiter, key_func, exempt=("/healthz",)):
        self.app = app
        self.limiter = limiter
        self.key_func = key_func
        self.exempt = set(exempt)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http" or scope["path"] in self.exempt:
            await self.app(scope, receive, send)
            return
        decision = self.limiter.check(self.key_func(scope))
        if decision.allowed:
            await self.app(scope, receive, send)
            return
        wait = retry_after_header(decision.retry_after)
        body = json.dumps({
            "type": "about:blank",
            "title": "Too Many Requests",
            "status": 429,
            "detail": f"Rate limit exceeded. Try again in {wait} second(s).",
        }).encode()
        await send({
            "type": "http.response.start",
            "status": 429,
            "headers": [
                (b"content-type", b"application/problem+json"),
                (b"content-length", str(len(body)).encode()),
                (b"retry-after", wait.encode()),
            ],
        })
        await send({"type": "http.response.body", "body": body})

Three details deserve a sentence each. Health checks are exempt, because a load balancer that polls /healthz every few seconds must not use up anyone’s budget. A refusal is a JSON document in the problem details format, with type set to about:blank. RFC 9457 explains that this value “indicates that the problem has no additional semantics beyond that of the HTTP status code” and that the title “SHOULD be the same as the recommended HTTP status phrase for that code”, which is why the title is Too Many Requests; our tutorial on problem details goes further. And the middleware asks self.key_func(scope) who the client is, a function we pass in, which brings us to the second half of the file. Add these two key functions to the end of middleware.py:

# middleware.py (continued)
def naive_key(scope):
    """The tempting version: believe whatever the X-Forwarded-For header says."""
    headers = dict(scope["headers"])
    forwarded = headers.get(b"x-forwarded-for")
    if forwarded:
        return forwarded.decode("latin-1").split(",")[0].strip()
    return scope["client"][0]


def client_key(scope):
    """Key a request by the peer address the ASGI server reports. IPv6 peers share one key per /64."""
    client = scope.get("client")
    if not client:
        return "unknown"
    try:
        address = ipaddress.ip_address(client[0])
    except ValueError:
        return client[0]  # not an IP address (a unix socket path or the test client's name)
    if address.version == 6 and address.ipv4_mapped:
        address = address.ipv4_mapped
    if address.version == 6:
        return str(ipaddress.ip_network((address, 64), strict=False))
    return str(address)

Do not worry about them yet. naive_key is the tempting wrong answer and client_key is the better one; Steps 7 and 8 explain both, and for now read client_key as the address of the peer that connected to us. Now the application itself. Create app.py:

# app.py
import asyncio
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

from limiters import TokenBucket
from middleware import RateLimitMiddleware, client_key, naive_key


def make_app(limiter, key_func=client_key, cors_origins=(), purge_every=60):
    @asynccontextmanager
    async def lifespan(app):
        async def sweep():
            while True:
                await asyncio.sleep(purge_every)
                limiter.purge()  # forget clients whose buckets are full again

        task = asyncio.create_task(sweep())
        try:
            yield
        finally:
            task.cancel()

    app = FastAPI(lifespan=lifespan)
    # Starlette makes the LAST middleware you add the OUTERMOST one, so add the limiter first:
    # CORS then wraps it and a 429 still gets its CORS headers.
    app.add_middleware(RateLimitMiddleware, limiter=limiter, key_func=key_func)
    if cors_origins:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=list(cors_origins),
            expose_headers=["Retry-After"],
        )

    @app.get("/healthz")
    def healthz():
        return {"ok": True}

    @app.get("/v1/hello")
    def hello(request: Request):
        return {"hello": "world", "peer": request.client.host if request.client else None}

    return app


RATE = float(os.environ.get("RATE", "2"))
BURST = int(os.environ.get("BURST", "5"))
KEY_FUNC = naive_key if os.environ.get("KEY_MODE") == "header" else client_key
app = make_app(TokenBucket(rate=RATE, capacity=BURST), KEY_FUNC)

make_app builds the FastAPI application around any limiter. Its lifespan function starts a background task when the server starts, and every purge_every seconds the task calls limiter.purge(), the Step 4 fix for the memory growth; it cancels the task when the server stops. The comment about middleware order is a measured fact, not a guess. FastAPI’s documentation says that “The last middleware added is the outermost, and the first is the innermost”, so adding the limiter first and the CORS middleware second puts CORS on the outside, and a 429 response still gets its CORS headers. Browsers hide the details of a response without them, and a front end could not read Retry-After (our CORS tutorial explains why). The three environment variables at the bottom, RATE, BURST and KEY_MODE, only exist so that the experiments can reconfigure the app; KEY_MODE=header switches to the wrong key function on purpose.

You can try the app by hand. A bucket of 5 tokens that refills one token every 2 seconds is slow enough to watch, so start the server with those settings in one terminal and send requests from a second one with curl:

$ RATE=0.5 BURST=5 python -m uvicorn app:app --port 8000

To make the session repeatable, step06_live_demo.py starts the server itself, sends eight requests, looks at a refusal, waits, and checks that the health endpoint is never refused. It prints each command and what came back. (It uses port 8000, so stop the server from the first terminal before you run it.)

# step06_live_demo.py
"""Start the real app with uvicorn and talk to it with curl, printing each command and what it answered."""
import os
import subprocess
import sys
import time

import requests

HERE = os.path.dirname(os.path.abspath(__file__))
env = dict(os.environ, RATE="0.5", BURST="5", PYTHONDONTWRITEBYTECODE="1")
server = subprocess.Popen([sys.executable, "-m", "uvicorn", "app:app", "--port", "8000", "--log-level", "warning"],
                          cwd=HERE, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def show(command):
    print(f"$ {command.rstrip()}")
    result = subprocess.run(["bash", "-c", command], capture_output=True, text=True, cwd=HERE)
    print((result.stdout + result.stderr).rstrip("\n"))


try:
    for _ in range(80):
        try:
            if requests.get("http://127.0.0.1:8000/healthz", timeout=1).status_code == 200:
                break
        except requests.RequestException:
            time.sleep(0.25)
    show(r'for i in 1 2 3 4 5 6 7 8; do curl -s -o /dev/null -w "%{http_code} " http://127.0.0.1:8000/v1/hello; done; echo')
    show(r'''curl -si http://127.0.0.1:8000/v1/hello | tr -d "\r" | grep -iE '^(HTTP|content-type|retry-after|\{)' ''')
    show(r'sleep 2; curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/v1/hello')
    show(r'for i in 1 2 3 4 5 6 7 8; do curl -s -o /dev/null -w "%{http_code} " http://127.0.0.1:8000/healthz; done; echo')
finally:
    server.terminate()
    server.wait(timeout=10)

Run python step06_live_demo.py:

$ for i in 1 2 3 4 5 6 7 8; do curl -s -o /dev/null -w "%{http_code} " http://127.0.0.1:8000/v1/hello; done; echo
200 200 200 200 200 429 429 429 
$ curl -si http://127.0.0.1:8000/v1/hello | tr -d "\r" | grep -iE '^(HTTP|content-type|retry-after|\{)'
HTTP/1.1 429 Too Many Requests
content-type: application/problem+json
retry-after: 2
{"type": "about:blank", "title": "Too Many Requests", "status": 429, "detail": "Rate limit exceeded. Try again in 2 second(s)."}
$ sleep 2; curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:8000/v1/hello
200
$ for i in 1 2 3 4 5 6 7 8; do curl -s -o /dev/null -w "%{http_code} " http://127.0.0.1:8000/healthz; done; echo
200 200 200 200 200 200 200 200 

Line by line: the first eight requests got 200 200 200 200 200 429 429 429, the five tokens of the burst and then three refusals. The second command prints the interesting parts of a refusal: the status line, the problem details content type, retry-after: 2 (the real wait was a little under 2 seconds, rounded up) and the JSON body. After sleep 2 the same request succeeds, and eight requests to /healthz all return 200 even though the bucket had just been empty.

Step 7: Who is the client? X-Forwarded-For and uvicorn’s trust list

Production traffic rarely reaches your application directly. It goes through a reverse proxy (nginx, a cloud load balancer, a CDN), a server that accepts the connection from the user and opens a second one to your app. From the app’s point of view every request now comes from the proxy’s address, so a limiter that keys on the connection’s address gives all your users one shared budget. Proxies solve that by adding a header, X-Forwarded-For, that lists the addresses the request has passed through, each proxy appending the address it saw. The nginx documentation describes the variable commonly used to do it, $proxy_add_x_forwarded_for, as the client’s own header “with the $remote_addr variable appended to it, separated by a comma”.

That header is also the single easiest thing to forge, because the user can send one too. The tempting limiter reads it directly, which is what naive_key does: take the first address in the header. Anyone with curl then gets a new budget for every request by sending a new address each time. The better design is to let the web server work out the real client address and to read only that. Uvicorn can do it: with --proxy-headers, which its documentation says is “enabled, but is restricted to only trusting connecting IPs in the forwarded-allow-ips configuration”. The default allow list is short: “127.0.0.1 and ::1 are trusted”. When the connection comes from a trusted address, uvicorn reads X-Forwarded-For, and its source explains how: “each proxy appends to the header list so check it in reverse order”. It walks the list from the right and takes the first address that is not itself a trusted proxy, so the addresses a user typed at the left end are ignored.

That is the theory. step07_proxy_matrix.py tests it against six setups on a real uvicorn 0.54.0 server, each with a bucket of 5 tokens that refills one token every 5 seconds, so that one shared budget allows 5 requests and independent budgets allow all of them. It starts a fresh server for each case, sends requests with chosen X-Forwarded-For values and counts the successes:

# step07_proxy_matrix.py
import os
import subprocess
import sys
import time

import requests

HERE = os.path.dirname(os.path.abspath(__file__))
port = 8120


def start_server(port, key_mode, flags, extra_env):
    env = dict(os.environ, KEY_MODE=key_mode, RATE="0.2", BURST="5", PYTHONDONTWRITEBYTECODE="1", **extra_env)
    command = [sys.executable, "-m", "uvicorn", "app:app", "--port", str(port), "--log-level", "warning", *flags]
    process = subprocess.Popen(command, cwd=HERE, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    for _ in range(80):
        try:
            if requests.get(f"http://127.0.0.1:{port}/healthz", timeout=1).status_code == 200:
                return process
        except requests.RequestException:
            time.sleep(0.25)
    process.kill()
    raise RuntimeError("the server did not start")


def stop_server(process):
    process.terminate()
    process.wait(timeout=10)


def run_case(label, key_mode, flags, forwarded_for, count, extra_env=None):
    global port
    port += 1
    process = start_server(port, key_mode, flags, extra_env or {})
    try:
        session = requests.Session()
        allowed, seen = 0, None
        for i in range(count):
            reply = session.get(f"http://127.0.0.1:{port}/v1/hello", headers={"X-Forwarded-For": forwarded_for(i)})
            if reply.status_code == 200:
                allowed += 1
                seen = seen or reply.json()["peer"]
        print(f"{label:46}{seen or '-':>14}{allowed:>8} of {count}")
    finally:
        stop_server(process)


forged = lambda i: f"203.0.113.{i}"                         # one forged address per request
proxy_appends = lambda i: f"203.0.113.{i}, 198.51.100.9"    # forged address, then the real peer a proxy appended
three_users = lambda i: f"198.51.100.{i % 3 + 1}"           # three honest users, as a proxy reports them

print(f"{'':46}{'app sees':>14}{'allowed':>12}")
run_case("A  app reads X-Forwarded-For itself", "header", [], forged, 20)
run_case("B  peer key, default uvicorn, lone header", "peer", [], forged, 20)
run_case("C  peer key, default uvicorn, proxy appends", "peer", [], proxy_appends, 20)
run_case("D  peer key, FORWARDED_ALLOW_IPS='*'", "peer", [], proxy_appends, 20, {"FORWARDED_ALLOW_IPS": "*"})
run_case("E  3 users, --no-proxy-headers", "peer", ["--no-proxy-headers"], three_users, 15)
run_case("F  3 users, default uvicorn", "peer", [], three_users, 15)

Run python step07_proxy_matrix.py. The six servers start and stop within a few seconds:

                                                    app sees     allowed
A  app reads X-Forwarded-For itself              203.0.113.0      20 of 20
B  peer key, default uvicorn, lone header        203.0.113.0      20 of 20
C  peer key, default uvicorn, proxy appends     198.51.100.9       5 of 20
D  peer key, FORWARDED_ALLOW_IPS='*'             203.0.113.0      20 of 20
E  3 users, --no-proxy-headers                     127.0.0.1       5 of 15
F  3 users, default uvicorn                     198.51.100.1      15 of 15

The column app sees is the client address the application received for the first allowed request. Here is what each row means.

• A: the app reads the header itself. 20 of 20 requests allowed. Every forged address was a new client, so the limit did nothing. This is the bug to avoid.
• B: the app uses the peer address, but the header arrives alone from a trusted address. 20 of 20 again. With uvicorn’s defaults, 127.0.0.1 is trusted, so a header that reaches uvicorn from there is believed. This is what happens when a proxy on the same machine passes the user’s own header through unchanged, which a proxy that forwards the original request headers will do unless it is told to set the header itself. nginx’s proxy_pass_request_headers directive, for example, defaults to on and “Indicates whether the header fields of the original request are passed to the proxied server”.
• C: the proxy appends the address it saw. 5 of 20, the correct answer. The request carried a forged address on the left and the real one, 198.51.100.9, appended on the right. Uvicorn walked the list from the right, and the first address it did not trust was 198.51.100.9, so all twenty requests shared one budget.
• D: the trust list is *. 20 of 20. A literal * means trust everything in the uvicorn documentation, so it takes the leftmost address, the one the user typed. Never use it unless the app can be reached only by your proxy, for example because a firewall or a private network enforces that.
• E: three honest users, proxy headers switched off. 5 of 15. The app sees only the proxy’s address, 127.0.0.1, so the three users share one budget and one noisy user can lock out the other two. This is the opposite failure, and it is what you get if you deploy behind a proxy and forget to configure the app for it.
• F: the same three users with the default trust list. 15 of 15, each user with a separate budget of 5. This is the configuration to aim for.

Two practical notes. First, case D sets FORWARDED_ALLOW_IPS as an environment variable, which uvicorn’s documentation supports, instead of passing --forwarded-allow-ips * on the command line. On my Windows machine the bare * was expanded into the names of the files in the current folder before uvicorn saw it (it answered “Got unexpected extra arguments” and listed my files), so check what your shell does with a star. Second, the allow list should name the addresses of your own proxies, nothing else. If several proxies stand in a chain, list all of them so that uvicorn can walk past each one. If you use another server, such as Gunicorn or Hypercorn, its rules and defaults are its own: run this six-case check against yours before you trust it. The check is cheap enough to keep as a test in your deployment pipeline.

Step 8: One IPv6 /64 is one client

IPv6 breaks a habit from the IPv4 world: that one address means one machine or one customer. RFC 4291 says that for unicast addresses (with one exception that does not matter here) “Interface IDs are required to be 64 bits long”. In practice that means the first 64 bits identify a network (a /64) and the last 64 bits are chosen by the machine on it, so whoever controls one /64 can send from any of 2⁶⁴ addresses. A limiter that keys on the full address gives that attacker a new budget for every address, and a new entry in your dictionary too, which is the memory problem of Step 4 again.

The fix is in the second half of client_key: an IPv6 address is replaced by its /64 network, and an IPv4-mapped IPv6 address (::ffff:203.0.113.7, which some servers use to report an IPv4 client on a dual-stack socket) is turned back into plain IPv4 first. Python’s ipaddress module does the work: ip_network((address, 64), strict=False) clears the host bits. step08_ipv6.py tries 1,000 addresses from one /64 against a limiter that keys on the full address and against client_key, then prints a few keys:

# step08_ipv6.py
from labkit import FakeClock
from limiters import TokenBucket
from middleware import client_key


def scope(address):
    return {"client": (address, 50000)}


# an attacker who controls one IPv6 /64 can send from any of its 2**64 addresses; here, 1,000 of them
addresses = [f"2001:db8:abcd:12::{n:x}" for n in range(1, 1001)]
for label, key_func in (("one key per address", lambda s: s["client"][0]), ("client_key (one key per /64)", client_key)):
    limiter = TokenBucket(rate=0.2, capacity=5, clock=FakeClock(0.0))
    allowed = sum(limiter.check(key_func(scope(a))).allowed for a in addresses)
    print(f"{label:30} allowed {allowed} of {len(addresses)}")

print()
for peer in ("203.0.113.7", "::ffff:203.0.113.7", "2001:db8:abcd:12::1", "2001:db8:abcd:12:ffff::9", "2001:db8:abcd:13::1", "testclient"):
    print(f"{peer:28} -> {client_key(scope(peer))}")
print(f"{'(no peer information)':28} -> {client_key({})}")

Run python step08_ipv6.py:

one key per address            allowed 1000 of 1000
client_key (one key per /64)   allowed 5 of 1000

203.0.113.7                  -> 203.0.113.7
::ffff:203.0.113.7           -> 203.0.113.7
2001:db8:abcd:12::1          -> 2001:db8:abcd:12::/64
2001:db8:abcd:12:ffff::9     -> 2001:db8:abcd:12::/64
2001:db8:abcd:13::1          -> 2001:db8:abcd:13::/64
testclient                   -> testclient
(no peer information)        -> unknown

Keyed by address, the attacker got 1,000 of 1,000 requests through. Keyed by /64, the same traffic got the 5 requests of one bucket. In the second part of the output, the two addresses in 2001:db8:abcd:12:: map to the same key and the one in 2001:db8:abcd:13:: to another. The last two rows are the defaults: a peer that is not an IP address (the test client’s name, or a unix socket path) is used as it is, and a request with no peer information at all shares the single key unknown. There is a trade-off. If a provider puts many unrelated customers inside one /64, they will share a budget. A /64 is the standard size of one subnet, but for logged-in users the better key is the account or the API key anyway, applied after authentication; keep the address-based limit for the routes that need no login, such as signing in or asking for a password reset (see our password reset tutorial).

Step 9: Prove it with tests that can fail

You now have the code and some measurements. Measurements prove that the code worked once; tests prove that it keeps working. We will write two kinds. Property-based tests state a rule that must hold for every possible input, and the Hypothesis library then generates many inputs, looks for one that breaks the rule and shrinks it to the smallest example (our Hypothesis tutorial covers the basics). Example tests check one scenario through the real application. Everything runs on the fake clock, with Fraction numbers so that ties and rounding cannot cause false alarms.

First, a small configuration file that defines two profiles of Hypothesis settings: a quick one (100 generated cases per test) for every run, and a thorough one (3,000) for a nightly job. Create conftest.py:

# conftest.py
from hypothesis import settings

settings.register_profile("lab", max_examples=100, deadline=None)
settings.register_profile("thorough", max_examples=3000, deadline=None)
settings.load_profile("lab")

Then test_properties.py. It starts with the setup shared by all properties: both limiters in one dictionary, so each property runs against both, a strategy for lists of gaps between requests (many zeros, so that requests arrive together, and round numbers, so that exact ties happen) and two helpers. probe asks a copy of the limiter what it would answer some time later, so a test can look into the future without disturbing the present.

# test_properties.py
import copy
from fractions import Fraction

import pytest
from hypothesis import given, strategies as st

from labkit import FakeClock
from limiters import GCRA, TokenBucket

HALF = Fraction(1, 2)
EPS = Fraction(1, 10**6)
FACTORIES = {
    "bucket": lambda clock: TokenBucket(HALF, 5, clock),  # one token every 2 s, at most 5 saved up
    "gcra": lambda clock: GCRA(HALF, 5, clock),
}
# gaps in milliseconds; plenty of zeros, so requests arrive together, and round numbers, so exact ties happen
gaps = st.lists(st.sampled_from([0, 0, 1, 7, 100, 999, 1000, 2000, 4000, 12000]), min_size=1, max_size=60)
# (gap in ms, which client, purge first?) for the tests that mix clients and purging
steps = st.lists(st.tuples(st.sampled_from([0, 1, 500, 3000, 20000]), st.sampled_from("abc"), st.booleans()), min_size=1, max_size=60)


def arrival_times(gap_list):
    now, times = Fraction(0), []
    for millis in gap_list:
        now += Fraction(millis, 1000)
        times.append(now)
    return times


def probe(limiter, delay, key="k"):
    """What would the limiter say `delay` seconds from now, if nothing else arrived? (Asks a copy.)"""
    twin = copy.deepcopy(limiter)
    twin.clock.advance(delay)
    return twin.check(key)

Property 1: the wait is honest and tight

Whenever a request is refused, the advertised retry_after must be neither too short nor too long: a request one microsecond early must still be refused, and one a microsecond late must be allowed.

# test_properties.py (continued)
@pytest.mark.parametrize("name", FACTORIES)
@given(gaps)
def test_retry_after_is_neither_too_short_nor_too_long(name, gap_list):
    clock = FakeClock(Fraction(0))
    limiter = FACTORIES[name](clock)
    for moment in arrival_times(gap_list):
        clock.set(moment)
        decision = limiter.check("k")
        if decision.allowed:
            continue
        assert decision.retry_after > 0
        if decision.retry_after > EPS:
            assert not probe(limiter, decision.retry_after - EPS).allowed  # not too optimistic
        assert probe(limiter, decision.retry_after + EPS).allowed          # not too pessimistic

Properties 2 and 3: remaining is exact, and the bucket keeps its promise

If an allowed request reports remaining of k, exactly k more requests at the same instant must be allowed and the next one refused. And the promise from Step 2 becomes executable: among any run of accepted requests, the count never exceeds capacity + rate × elapsed.

# test_properties.py (continued)
@pytest.mark.parametrize("name", FACTORIES)
@given(gaps)
def test_remaining_is_exactly_how_many_more_fit_right_now(name, gap_list):
    clock = FakeClock(Fraction(0))
    limiter = FACTORIES[name](clock)
    for moment in arrival_times(gap_list):
        clock.set(moment)
        decision = limiter.check("k")
        if not decision.allowed:
            assert decision.remaining == 0
            continue
        twin = copy.deepcopy(limiter)
        for _ in range(decision.remaining):
            assert twin.check("k").allowed
        assert not twin.check("k").allowed


@pytest.mark.parametrize("name", FACTORIES)
@given(gaps)
def test_never_more_than_burst_plus_rate_times_time(name, gap_list):
    clock = FakeClock(Fraction(0))
    limiter = FACTORIES[name](clock)
    accepted = []
    for moment in arrival_times(gap_list):
        clock.set(moment)
        if limiter.check("k").allowed:
            accepted.append(moment)
    for i in range(len(accepted)):
        for j in range(i, len(accepted)):
            assert j - i + 1 <= 5 + HALF * (accepted[j] - accepted[i])

Property 4: GCRA and the token bucket are the same limiter

The claim from Step 3, now checked on every generated schedule with exact arithmetic, comparing the whole Decision object.

# test_properties.py (continued)
@given(gaps)
def test_gcra_makes_exactly_the_decisions_of_the_token_bucket(gap_list):
    bucket_clock, gcra_clock = FakeClock(Fraction(0)), FakeClock(Fraction(0))
    bucket, gcra = FACTORIES["bucket"](bucket_clock), FACTORIES["gcra"](gcra_clock)
    for moment in arrival_times(gap_list):
        bucket_clock.set(moment)
        gcra_clock.set(moment)
        assert bucket.check("k") == gcra.check("k")

Properties 5 to 7: purging is invisible, and clients do not affect each other

Run one limiter that purges at random moments next to one that never purges, and require identical answers: that is the claim of Step 4. A second test checks that a quiet limiter really does empty. The last property says that a client’s answers depend only on that client’s own traffic.

# test_properties.py (continued)
@pytest.mark.parametrize("name", FACTORIES)
@given(steps)
def test_purging_never_changes_a_decision(name, step_list):
    clock_a, clock_b = FakeClock(Fraction(0)), FakeClock(Fraction(0))
    with_purge, without = FACTORIES[name](clock_a), FACTORIES[name](clock_b)
    now = Fraction(0)
    for millis, key, purge_first in step_list:
        now += Fraction(millis, 1000)
        clock_a.set(now)
        clock_b.set(now)
        if purge_first:
            with_purge.purge()
        assert with_purge.check(key) == without.check(key)


@pytest.mark.parametrize("name", FACTORIES)
def test_purge_empties_a_limiter_whose_clients_have_gone_quiet(name):
    clock = FakeClock(Fraction(0))
    limiter = FACTORIES[name](clock)
    for key in "abcde":
        limiter.check(key)
    clock.advance(1000)
    assert limiter.purge() == 5
    assert limiter.state == {}


@pytest.mark.parametrize("name", FACTORIES)
@given(steps)
def test_one_clients_traffic_never_changes_another_clients_answers(name, step_list):
    clock_both, clock_alone = FakeClock(Fraction(0)), FakeClock(Fraction(0))
    both, alone = FACTORIES[name](clock_both), FACTORIES[name](clock_alone)
    now = Fraction(0)
    for millis, key, _ in step_list:
        now += Fraction(millis, 1000)
        clock_both.set(now)
        clock_alone.set(now)
        decision = both.check(key)
        if key == "a":
            assert decision == alone.check("a")

The application tests

Now the example tests in test_app.py, which exercise the real FastAPI app through Starlette’s test client, one behaviour each. Create the file with the first part:

# test_app.py
import time

import pytest
from fastapi.testclient import TestClient

from app import make_app
from labkit import FakeClock
from limiters import TokenBucket, retry_after_header
from middleware import client_key, naive_key


def test_retry_after_header_rounds_up_and_never_says_zero():
    assert retry_after_header(0.2) == "1"
    assert retry_after_header(1.2) == "2"
    assert retry_after_header(3) == "3"
    assert retry_after_header(59.01) == "60"


def build(rate=0.8, capacity=2, **options):
    clock = FakeClock(0.0)
    client = TestClient(make_app(TokenBucket(rate=rate, capacity=capacity, clock=clock), **options))
    return client, clock

The test client does not use a network, so the fake clock still controls time. The scenarios, in order: the third request gets a problem document with Retry-After: 2 (the real wait is 1.25 seconds, so a rounded-down header would say 1); waiting the advertised time works; health checks are exempt and cost nothing; each client has its own budget; and the application’s startup and shutdown pass through the middleware.

# test_app.py (continued)
def test_the_third_request_gets_a_problem_document_and_a_rounded_up_wait():
    client, _ = build()
    assert [client.get("/v1/hello").status_code for _ in range(2)] == [200, 200]
    refused = client.get("/v1/hello")
    assert refused.status_code == 429
    assert refused.headers["retry-after"] == "2"  # the real wait is 1.25 s
    assert refused.headers["content-type"] == "application/problem+json"
    assert refused.json()["status"] == 429


def test_waiting_for_the_advertised_time_works():
    client, clock = build()
    for _ in range(3):
        client.get("/v1/hello")
    clock.advance(2)
    assert client.get("/v1/hello").status_code == 200


def test_health_checks_are_exempt_and_cost_nothing():
    client, _ = build(capacity=1)
    assert [client.get("/healthz").status_code for _ in range(10)] == [200] * 10
    assert client.get("/v1/hello").status_code == 200


def test_each_client_has_its_own_budget():
    app = make_app(TokenBucket(rate=0.8, capacity=1, clock=FakeClock(0.0)))
    first = TestClient(app, client=("203.0.113.1", 5000))
    second = TestClient(app, client=("203.0.113.2", 5000))
    assert first.get("/v1/hello").status_code == 200
    assert first.get("/v1/hello").status_code == 429
    assert second.get("/v1/hello").status_code == 200


def test_lifespan_events_pass_through_the_middleware():
    client, _ = build()
    with client:  # starting and stopping the app sends a lifespan scope through every middleware
        assert client.get("/v1/hello").status_code == 200

The rest cover the background purge task, the CORS headers on a refusal, the key function and the spoofing hole. The background test advances the fake clock by 1,000 seconds, then waits (up to 5 seconds) for the purge task to empty the dictionary on its own:

# test_app.py (continued)
def test_the_app_purges_idle_clients_in_the_background():
    clock = FakeClock(0.0)
    limiter = TokenBucket(rate=0.8, capacity=2, clock=clock)
    with TestClient(make_app(limiter, purge_every=0.01)) as client:
        client.get("/v1/hello")
        assert len(limiter.state) == 1
        clock.advance(1000)  # the client has long since refilled its bucket
        deadline = time.monotonic() + 5
        while limiter.state and time.monotonic() < deadline:
            time.sleep(0.01)
    assert limiter.state == {}


def test_a_refusal_keeps_its_cors_headers_when_cors_wraps_the_limiter():
    client, _ = build(capacity=1, cors_origins=["https://app.example"])
    headers = {"Origin": "https://app.example"}
    client.get("/v1/hello", headers=headers)
    refused = client.get("/v1/hello", headers=headers)
    assert refused.status_code == 429
    assert refused.headers["access-control-allow-origin"] == "https://app.example"
    assert "retry-after" in refused.headers["access-control-expose-headers"].lower()


@pytest.mark.parametrize("peer, expected", [
    ("203.0.113.7", "203.0.113.7"),
    ("::ffff:203.0.113.7", "203.0.113.7"),
    ("2001:db8:abcd:12::1", "2001:db8:abcd:12::/64"),
    ("2001:db8:abcd:12:ffff:ffff:ffff:ffff", "2001:db8:abcd:12::/64"),
    ("2001:db8:abcd:13::1", "2001:db8:abcd:13::/64"),
    ("testclient", "testclient"),
])
def test_client_key(peer, expected):
    assert client_key({"client": (peer, 1)}) == expected


def test_a_request_without_peer_information_gets_one_shared_key():
    assert client_key({}) == "unknown"
    assert client_key({"client": None}) == "unknown"


def test_naive_key_can_be_spoofed_by_anyone():
    def scope(value):
        return {"client": ("198.51.100.9", 1), "headers": [(b"x-forwarded-for", value)]}

    assert naive_key(scope(b"203.0.113.1")) != naive_key(scope(b"203.0.113.2"))

Run everything with python -m pytest -q:

............................. [100%]
29 passed in 1.12s

That is 29 tests (13 about the limiters, 11 of them property-based when each limiter is counted separately, and 16 about the application) in about a second. Now the thorough profile, with 3,000 generated cases for every property:

$ python -m pytest -q --hypothesis-profile=thorough

............................. [100%]
29 passed in 32.96s

It took about half a minute and passed too. Use the quick profile on every commit and the thorough profile in a nightly job: rare coincidences, such as a request landing on the exact moment a token appears, are what the larger run is for.

Step 10: Break it on purpose

A test that cannot fail proves nothing. The way to find out whether yours can is mutation checking: change the code in one small, plausible, wrong way, run the tests, and see that something fails. mutations.py does this for ten bugs, each one a mistake a real person could make during a refactor. It copies the project into a temporary folder, applies one change, runs pytest with -x (stop at the first failure) and reports the failing test:

# mutations.py
"""Break the code on purpose, one bug at a time, and check that the tests notice."""
import os
import re
import shutil
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))

MUTANTS = [
    ("Retry-After rounded down", "limiters.py",
     "return str(max(1, math.ceil(seconds)))", "return str(max(1, int(seconds)))"),
    ("bucket never caps its tokens", "limiters.py",
     "tokens = min(self.capacity, tokens + (now - stamp) * self.rate)", "tokens = tokens + (now - stamp) * self.rate"),
    ("a refused request still spends a token", "limiters.py",
     "        self.state[key] = (tokens, now)\n        return Decision(False,", "        self.state[key] = (tokens - 1, now)\n        return Decision(False,"),
    ("GCRA forgets max(tat, now)", "limiters.py",
     "tat = max(self.state.get(key, now), now)", "tat = self.state.get(key, now)"),
    ("a refused request pushes GCRA's clock forward", "limiters.py",
     "        if now < tat - self.tolerance:\n            return Decision(False,", "        if now < tat - self.tolerance:\n            self.state[key] = tat + self.interval\n            return Decision(False,"),
    ("purge drops active clients", "limiters.py",
     "if self._idle(saved, now)]", "if not self._idle(saved, now)]"),
    ("the background purge does nothing", "app.py",
     "limiter.purge()  # forget clients whose buckets are full again", "pass"),
    ("IPv6 clients keyed by full address", "middleware.py",
     "return str(ipaddress.ip_network((address, 64), strict=False))", "return str(address)"),
    ("lifespan scope reaches the key function", "middleware.py",
     'if scope["type"] != "http" or scope["path"] in self.exempt:', 'if scope["path"] in self.exempt:'),
    ("CORS added inside the limiter", "app.py",
     "    app.add_middleware(RateLimitMiddleware, limiter=limiter, key_func=key_func)\n    if cors_origins:\n        app.add_middleware(\n            CORSMiddleware,\n            allow_origins=list(cors_origins),\n            expose_headers=[\"Retry-After\"],\n        )\n",
     "    if cors_origins:\n        app.add_middleware(\n            CORSMiddleware,\n            allow_origins=list(cors_origins),\n            expose_headers=[\"Retry-After\"],\n        )\n    app.add_middleware(RateLimitMiddleware, limiter=limiter, key_func=key_func)\n"),
]


def run_tests(folder):
    env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", COLUMNS="170")
    result = subprocess.run([sys.executable, "-m", "pytest", "-x", "-q", "-p", "no:cacheprovider", "test_properties.py", "test_app.py"],
                            cwd=folder, env=env, capture_output=True, text=True)
    return result.returncode, result.stdout


killed = 0
for label, filename, old, new in MUTANTS:
    folder = tempfile.mkdtemp(prefix="mutant_")
    try:
        for name in os.listdir(HERE):
            if name.endswith(".py"):
                shutil.copy(os.path.join(HERE, name), folder)
        path = os.path.join(folder, filename)
        source = open(path, encoding="utf-8").read()
        assert source.count(old) == 1, f"{label}: the text to replace occurs {source.count(old)} times"
        open(path, "w", encoding="utf-8", newline="\n").write(source.replace(old, new))
        code, output = run_tests(folder)
        failed = re.findall(r"^FAILED (\S+)", output, flags=re.M)
        verdict = "caught by " + failed[0].split("::", 1)[-1][:70] if code != 0 and failed else ("caught (error)" if code != 0 else "NOT CAUGHT")
        killed += code != 0
        print(f"{label:48} {verdict}")
    finally:
        shutil.rmtree(folder, ignore_errors=True)
print(f"\n{killed} of {len(MUTANTS)} deliberate bugs were caught")

Run python mutations.py:

Retry-After rounded down                         caught by test_retry_after_header_rounds_up_and_never_says_zero
bucket never caps its tokens                     caught by test_never_more_than_burst_plus_rate_times_time[bucket]
a refused request still spends a token           caught by test_retry_after_is_neither_too_short_nor_too_long[bucket]
GCRA forgets max(tat, now)                       caught by test_never_more_than_burst_plus_rate_times_time[gcra]
a refused request pushes GCRA's clock forward    caught by test_retry_after_is_neither_too_short_nor_too_long[gcra]
purge drops active clients                       caught by test_purging_never_changes_a_decision[bucket]
the background purge does nothing                caught by test_the_app_purges_idle_clients_in_the_background
IPv6 clients keyed by full address               caught by test_client_key[2001:db8:abcd:12::1-2001:db8:abcd:12::/64]
lifespan scope reaches the key function          caught by test_lifespan_events_pass_through_the_middleware
CORS added inside the limiter                    caught by test_a_refusal_keeps_its_cors_headers_when_cors_wraps_the_limiter

10 of 10 deliberate bugs were caught

All ten are caught, and the failing test names read like a bug report. Rounding Retry-After down is caught by the header test. A bucket that never caps its tokens is caught by the burst-plus-rate property. A bucket that spends a token on a refused request, and a GCRA that moves its schedule forward on a refusal, are both caught by the honest-wait property: the more a client retries, the longer its penalty, which is exactly what a rate limiter should not do. Forgetting the max in GCRA is caught by the burst property, and a purge that deletes busy clients by the purge-invariance property. The last four are about the plumbing rather than the algorithm: the background purge that does nothing, IPv6 addresses keyed individually, the middleware choking on a lifespan scope, and CORS placed inside the limiter. If one of them had survived, that would be a test to write.

How to confirm the whole thing works

From the project folder, with the virtual environment active and ports 8000 and 8121 to 8126 free, run each step and then the checks:

$ python step02_bucket.py
$ python step03_gcra.py
$ python step04_memory.py
$ python step05_retry_after.py
$ python step06_live_demo.py
$ python step07_proxy_matrix.py
$ python step08_ipv6.py
$ python -m pytest -q
$ python -m pytest -q --hypothesis-profile=thorough
$ python mutations.py

You should see, in order: the seven-request table from Step 2; 100,000 identical decisions under exact arithmetic; GCRA at 62 bytes per client and 200,000 flooded clients purged down to one; the rounded-down header costing 8, 16 and 6 requests against 2; the curl session from Step 6; the six proxy cases A to F (20 of 20, 20 of 20, 5 of 20, 20 of 20, 5 of 15, 15 of 15); 1,000 of 1,000 against 5 of 1,000 for IPv6; 29 passing tests twice; and ten of ten deliberate bugs caught. For your own deployment, repeat the six-case check from Step 7 through your real proxy, because the right answer depends on the proxy’s configuration more than on your code.

Common mistakes, with the numbers from this tutorial

• Believing X-Forwarded-For. 20 of 20 forged requests got through (case A). Let the server resolve the address and trust only your own proxies.
• Trusting every proxy address. A trust list of * also let 20 of 20 through (case D).
• Not configuring the app behind a proxy. Three honest users shared one budget and got 5 of 15 (case E).
• Keying IPv6 clients by address. 1,000 of 1,000 requests from one /64 passed.
• Rounding Retry-After down. The client sent 8, 16 and 6 requests instead of 2.
• Never forgetting clients. 200,000 one-request clients stayed in memory until a purge removed them.
• Testing with floats and a real clock. Slow, flaky, and unable to check exact boundaries.

Where to go next

The limiter is correct for one process. If you run several worker processes or several servers, each has its own dictionary, so a client’s effective budget is multiplied by the number of workers it reaches. A shared store (Redis is the usual choice) or limits at the edge fix that; keep the interface you built here, because check(key) hides where the state lives. I did not measure a multi-worker setup in this tutorial, so test yours.

From here you could give different routes different budgets, key authenticated requests by account instead of address, and advertise RateLimit-Policy and RateLimit once the IETF draft settles. Overload that comes from everyone at once, rather than from one client, needs a different tool, which our tutorial on backpressure and load shedding covers, and the earlier rate limiter tutorial shows the window-based algorithms and the thread-safety problem that this one skipped.