Rate limits and forms

Introduction
Limiting a view
1. Answering a refusal your own way
2. Sending the RateLimit headers
Counting by hand
1. Choosing the key
Sharing the count across workers
Guarding a public form
1. Turning a post down from another feature
Reference
Advanced
See also

Introduction

A public form or an API is an easy target for a script. A throttle gives each caller a budget of events per window, such as five messages an hour from one address, and refuses the rest with the seconds to wait. Every worker of the product shares the same count.

from splent_framework.ratelimit import SharedThrottle, rate_limited

contact_throttle = SharedThrottle("contact.send", max_attempts=5, window_seconds=3600)


@contact_bp.route("/contact", methods=["POST"])
@rate_limited(contact_throttle)
def send():
    contact_service.save(request.form)
    return redirect(url_for("contact.thanks"))

The sixth post from one address within the hour gets a 429 with Retry-After.

Limiting a view

rate_limited counts every call that gets through against the throttle, per caller address. A call past the budget is refused before your view runs, and is not counted.

The namespace, "contact.send" above, keeps budgets apart. Two throttles with different namespaces never share a count, even for the same address.

A budget of 0 turns the throttle off.

Answering a refusal your own way

def wait_page(seconds):
    flash(_("Too many messages. Try again in %(n)s seconds.", n=seconds), "warning")
    return redirect(url_for("contact.form"))


@contact_bp.route("/contact", methods=["POST"])
@rate_limited(contact_throttle, answer=wait_page)
def send():
    ...

answer(seconds) replaces the 429 with whatever you return. Without it, a page gets the app’s 429 page and an API gets a problem+json 429.

Sending the RateLimit headers

@rate_limited(search_throttle, headers=True)

Every answer that gets through then says the budget, what is left of it and when the oldest event leaves the window.

RateLimit-Policy: "search";q=10;w=60
RateLimit: "search";r=8;t=42

To add them to an answer yourself, call ratelimit_headers(throttle, key), which returns a dict.

Counting by hand

Sometimes only some events should count, such as wrong codes and not right ones. Use the throttle’s methods directly.

from splent_framework.ratelimit import (
    SharedThrottle,
    bucket_key,
    client_ip,
    too_many_requests,
)

wrong_codes = SharedThrottle("codes.redeem", max_attempts=5, window_seconds=900)


@codes_bp.route("/redeem", methods=["POST"])
def redeem():
    key = bucket_key(client_ip(), request.form.get("email", "").lower())
    wait = wrong_codes.retry_after(key)
    if wait:
        raise too_many_requests(wait)
    if not codes_service.redeem(request.form.get("code", "")):
        wrong_codes.record(key)
        flash(_("That code is not valid."), "danger")
        return redirect(url_for("codes.form"))
    wrong_codes.clear(key)
    return redirect(url_for("codes.done"))

Method	Does
`retry_after(key)`	0 while the key is under budget, else the seconds to wait
`record(key)`	Counts one event. `record_failure` is the same method
`clear(key)`	Forgets the key’s events, after a success for example
`state(key)`	`(events that count, seconds until the oldest leaves)`

too_many_requests(seconds, detail="") gives you the 429 to raise.

Choosing the key

client_ip() is the address the request came from. Behind a reverse proxy, set SPLENT_TRUSTED_PROXIES to the number of proxies, or every caller counts as the proxy. The framework never reads X-Forwarded-For on its own, since any caller can write it.

bucket_key(*parts) turns several parts, such as an address and an email, into one key of fixed length. What a caller sends never decides how much a key weighs.

Without a table, each process counts on its own. That is enough with one worker. With several, a guesser’s attempts would spread over as many budgets, so give the throttles a table in one of your features.

from splent_framework.db import db
from splent_framework.ratelimit import use_table


class ThrottleEvent(db.Model):
    __tablename__ = "throttle_event"

    id = db.Column(db.Integer, primary_key=True)
    bucket = db.Column(db.String(64), nullable=False)
    expires_at = db.Column(db.Integer, nullable=False)

    __table_args__ = (db.Index("ix_throttle_event_bucket", "bucket", "expires_at"),)


def init_feature(app):
    use_table(app, ThrottleEvent.__table__)

One table serves every throttle of the product. A row holds an HMAC of the namespace and the key, never what the caller sent. Each write also removes a batch of expired rows, so you need no cron.

Until the table exists, or while the database does not answer, each process counts on its own again and the log says so once.

Guarding a public form

A public form usually needs two checks. May this address post again now, and does a captcha or a spam filter turn the post down. form_guard asks both at once.

from splent_framework.guard import NOT_VERIFIED, form_guard

feedback_guard = form_guard(
    "feedback",
    "splent_feature_feedback",
    per_address=5,
    window_seconds=3600,
    setting="FEEDBACK_MAX_PER_HOUR",
)


@feedback_bp.route("/feedback", methods=["POST"])
def send():
    refused = feedback_guard.check()
    if refused and refused.reason == NOT_VERIFIED:
        flash(_("Please confirm you are not a robot."), "danger")
        return redirect(url_for("feedback.form"))
    if refused:
        flash(_("Too many messages. Try again in %(n)s seconds.", n=refused.retry_after), "warning")
        return redirect(url_for("feedback.form"))
    feedback_service.save(request.form)
    return redirect(url_for("feedback.thanks"))

check() answers None when the post may go on. Otherwise it answers a Refused whose reason is RATE_LIMITED, with retry_after seconds, or NOT_VERIFIED.

setting names a config key that holds the budget, so an operator can change it without a release. A budget of 0 turns the limit off.

Every post that reaches the veto counts against the budget, refused or not. Guessing a captcha spends it too.

Turning a post down from another feature

The guard defines the signal <form>-submitting, here feedback-submitting. A feature that protects forms connects to it and answers False to refuse a post.

from splent_framework.signals.signal_utils import connect_signal


@connect_signal("feedback-submitting", "splent_feature_spam_filter")
def check_feedback(sender, form=None, remoteip=None, **kwargs):
    if looks_like_spam(form.get("message", "")):
        return False
    return None

None lets the post through. The form never learns who listens. When the form’s own signals.py already defines the signal, the guard uses that one and keeps its listeners.

To send the listeners other fields than request.form, pass posted=. Any other keyword you give check() reaches them too, such as check(post_id=post.id).

Reference

Call	Defaults
`SharedThrottle(namespace, max_attempts, window_seconds, max_keys)`	`5`, `300`, `10000`
`rate_limited(throttle, key=client_ip, answer=None, headers=False)`
`ratelimit_headers(throttle, key, policy=None)`	The policy name is the namespace
`too_many_requests(retry_after, detail="")`
`use_table(app, table)`	A table with `id`, `bucket` and `expires_at`
`form_guard(form, provider, per_address=0, window_seconds=3600, setting=None)`	`provider` is the package of the feature that owns the form
`Refused(reason, retry_after=0)`	`reason` is `RATE_LIMITED` or `NOT_VERIFIED`

Advanced

A throttle’s budget can change while the app runs. throttle.max_attempts = 10 applies at once, to both ways of counting.

The count each process keeps on its own holds at most max_keys callers, dropping the coldest first, so a flood cannot fill the memory.