minimost.ratelimit

minimost.ratelimit

Dependency-free, in-process abuse throttling for MiniMost.

This module provides the two primitives the application uses to blunt denial-of-service (DoS) attempts from authenticated and unauthenticated clients alike, using nothing beyond the Python standard library and Flask:

• RateLimiter — a thread-safe sliding-window counter. Each call to RateLimiter.hit() records one event against a key (an IP address or a username) and reports whether that key has exceeded its allowance over the trailing window. The rate_limit() decorator wraps a Flask view with this check so a flood of requests to expensive endpoints (login, signup, message send, uploads) is rejected with 429 Too Many Requests instead of tying up workers or CPU.

• ConcurrencyLimiter — a thread-safe count of currently held resources per key. It backs the cap on simultaneous Server-Sent Events streams per user (minimost.events), so one actor cannot pin every worker thread by opening connections that never close.

Scope and limitations

The counters live in the worker process, so with several Gunicorn workers the effective ceiling is limit × workers. That is deliberate: the goal is to stop pathological abuse cheaply with zero new dependencies and no shared store on the hot path, not to enforce a globally exact quota. For a hard, cross-worker ceiling, put a reverse proxy in front (e.g. Nginx limit_req / limit_conn); these in-process limits complement that layer rather than replace it.

Configuration

Limits are generous by default — tuned so ordinary interactive use never trips them — and can be overridden in settings.json (re-read, cached by mtime, so edits take effect without a restart):

• rate_limit_enabled — master on/off switch (default true). Read once in minimost.create_app() into app.config["RATELIMIT_ENABLED"].

• rate_limits — an object mapping an action name to a [max, window_seconds] pair, overriding the matching entry in DEFAULT_LIMITS.

• max_event_streams_per_user — the per-user concurrent /events cap (default DEFAULT_MAX_EVENT_STREAMS).

class minimost.ratelimit.RateLimiter

Bases: object

A thread-safe sliding-window rate limiter keyed by arbitrary strings.

Each key maps to a collections.deque of monotonic timestamps for the events seen within the trailing window; expired timestamps are evicted lazily on the next hit(). All state is in-process and guarded by a single lock, so the limiter is safe to share across Gunicorn gthread worker threads.

__init__()

hit(key, limit, window)

Record one event for key and report whether it is allowed.

Parameters:

• key – The identity being limited (e.g. "login:1.2.3.4").

• limit – Maximum number of events permitted within window.

• window – Trailing window length in seconds.

Returns:

(allowed, retry_after) — allowed is False when the key has already reached limit events in the window (the event is not recorded in that case), and retry_after is a whole-second hint for the Retry-After header (0 when allowed).

Return type:

tuple[bool, int]

_gc(now)

Drop keys idle longer than the GC horizon. Caller holds the lock.

reset()

Forget all recorded events (used between tests).

class minimost.ratelimit.ConcurrencyLimiter

Bases: object

A thread-safe count of currently-held resources per key.

Unlike RateLimiter, which counts events over time, this tracks how many resources (e.g. open SSE streams) a key holds right now: acquire() takes a slot if one is free and release() returns it.

__init__()

acquire(key, limit)

Take a slot for key if fewer than limit are held.

Returns:

True if a slot was acquired, False if key is already at limit.

Return type:

bool

release(key)

Return a previously acquired slot for key (never goes negative).

reset()

Forget all held slots (used between tests).

minimost.ratelimit.reset_all()

Reset both limiters. Intended for test isolation.

minimost.ratelimit._load_settings()

Return parsed settings.json, cached by file mtime.

Any read or parse error yields an empty dict so callers fall back to the built-in defaults.

minimost.ratelimit._is_number(value)

True for a real int/float (rejecting bool, which is an int).

minimost.ratelimit.limit_for(name)

Return the (max_events, window_seconds) limit for action name.

Falls back to DEFAULT_LIMITS when no valid override is present in the rate_limits object of settings.json.

minimost.ratelimit.max_event_streams()

Return the per-user concurrent /events cap from settings (or default).

minimost.ratelimit._enabled()

True unless the app has disabled limiting via RATELIMIT_ENABLED.

minimost.ratelimit.client_ip()

Return the remote address for the current request, or "unknown".

Behind a reverse proxy this is the proxy’s address unless the proxy is configured to forward the real client IP and the app is set up to trust it; see the deployment docs.

minimost.ratelimit.rate_limit_message(retry_after)

Return a friendly ‘slow down’ message naming the wait in whole seconds.

Shared by every throttled response (plain text, rendered HTML, and JSON) so the wording the user sees is identical however the route reports the limit.

minimost.ratelimit._too_many(retry_after)

Build the default plain-text 429 response with a Retry-After header.

minimost.ratelimit.rate_limit(name, by='ip', on_limit=None)

Decorate a Flask view so calls to it are throttled per name.

Parameters:

• name – Key into DEFAULT_LIMITS (and the rate_limits settings override) selecting this endpoint’s allowance.

• by – "ip" to key the limit on the client IP (for unauthenticated endpoints like login/signup), or "user" to key it on the logged-in username, falling back to the IP when there is no session.

• on_limit – Optional callable on_limit(retry_after) returning the Flask response to send when the limit is hit, instead of the default plain-text 429. Use it to surface the limit the way the caller expects — a re-rendered HTML form with an inline error, or a JSON error for a fetch endpoint. rate_limit_message() builds the standard wording, and callers should keep the 429 status and Retry-After header (e.g. (render_template(...), 429, {"Retry-After": str(r)})).

Returns:

The wrapped view, which returns 429 once the limit is hit and otherwise calls through unchanged. Limiting is skipped entirely when app.config["RATELIMIT_ENABLED"] is false.

Apply it below minimost.auth.login_required() so authenticated by="user" endpoints see the established session:

@chat_bp.route("/send/<channel>", methods=["POST"])
@auth.login_required
@ratelimit.rate_limit("send", by="user")
def send(channel):
    ...

minimost.ratelimit.acquire_stream(user)

Try to take an /events stream slot for user.

Returns:

True if the user is below the concurrent-stream cap (a slot is now held and must be released with release_stream()), False if they are already at the cap.

Return type:

bool

minimost.ratelimit.release_stream(user)

Release an /events stream slot previously taken for user.