Rate limiting & lockout¶

Per-endpoint throttling (auth.rate_limit(...)) and escalating login lockout, both over a pluggable backend. Use redis_rate_limiter(...) in production.

crudauth.ratelimit.RateLimit `dataclass` ¶

RateLimit(times: int, seconds: int)

A fixed-window allowance: times events per seconds.

times=0 disables the limit (an explicit, documented off switch, never the low-friction default).

Raises:

Type	Description
`ValueError`	If `times` is negative or `seconds` isn't positive.

Example

CRUDAuth(..., rate_limits={"password_reset_request": RateLimit(3, 1800)})

disabled `property` ¶

disabled: bool

True when times == 0 (the limit is turned off).

crudauth.ratelimit.RateLimitResolver `module-attribute` ¶

RateLimitResolver = Callable[
    [Request, Principal | None],
    RateLimit | None | Awaitable[RateLimit | None],
]

A per-request limit: a sync or async function of the request and the principal (None when anonymous) returning a RateLimit, or None for no limit on that request.

crudauth.ratelimit.KeyBy ¶

Bases: str, Enum

Which dimension a CRUDAuth.rate_limit dependency keys on.

crudauth.ratelimit.RateLimitHeadersMiddleware ¶

RateLimitHeadersMiddleware(app: ASGIApp)

Copy the recorded rate-limit headers onto every response that carries none.

Opt-in: add it to the app to have 401, 404 and 422 responses report the budget they spent. A response that already carries the headers - a route that returned normally, or the limiter's own 429 - is left as it is, and a request no limiter counted gets nothing.

Example

from crudauth.ratelimit import RateLimitHeadersMiddleware

app.add_middleware(RateLimitHeadersMiddleware)

crudauth.ratelimit.LockoutConfig `dataclass` ¶

LockoutConfig(
    max_attempts: int = DEFAULT_LOGIN_MAX_ATTEMPTS,
    attempt_window_seconds: int = DEFAULT_LOGIN_ATTEMPT_WINDOW_SECONDS,
    lockout_base_seconds: int = DEFAULT_LOGIN_LOCKOUT_BASE_SECONDS,
    lockout_max_seconds: int = DEFAULT_LOGIN_LOCKOUT_MAX_SECONDS,
    round_retention_seconds: int = DEFAULT_LOGIN_ROUND_RETENTION_SECONDS,
    on_login_success: LoginSuccessClears = "clear_all",
)

Tuning for the escalating login lockout shared by /login and /token.

Passed as CRUDAuth(lockout=...), so it applies whichever transports are configured. See LockoutPolicy for how each value is used.

Parameters:

Name	Type	Description	Default
`max_attempts`	`int`	Failures allowed within `attempt_window_seconds` before a lockout trips.	`DEFAULT_LOGIN_MAX_ATTEMPTS`
`attempt_window_seconds`	`int`	How long failures keep counting.	`DEFAULT_LOGIN_ATTEMPT_WINDOW_SECONDS`
`lockout_base_seconds`	`int`	First lockout duration; doubles each round.	`DEFAULT_LOGIN_LOCKOUT_BASE_SECONDS`
`lockout_max_seconds`	`int`	Cap on the exponential lockout duration.	`DEFAULT_LOGIN_LOCKOUT_MAX_SECONDS`
`round_retention_seconds`	`int`	How long the escalation round count survives without a new lockout.	`DEFAULT_LOGIN_ROUND_RETENTION_SECONDS`
`on_login_success`	`LoginSuccessClears`	What a successful login clears, `"clear_all"` (default) or `"clear_user_only"`.	`'clear_all'`

Raises:

Type	Description
`ValueError`	If `max_attempts` or a duration isn't positive, or `on_login_success` isn't one of the two modes.

Example

from crudauth.ratelimit import LockoutConfig

CRUDAuth(..., lockout=LockoutConfig(max_attempts=10, lockout_max_seconds=900))

crudauth.ratelimit.LockoutPolicy ¶

LockoutPolicy(
    backend: RateLimiterBackend,
    *,
    max_attempts: int = DEFAULT_LOGIN_MAX_ATTEMPTS,
    attempt_window_seconds: int = DEFAULT_LOGIN_ATTEMPT_WINDOW_SECONDS,
    lockout_base_seconds: int = DEFAULT_LOGIN_LOCKOUT_BASE_SECONDS,
    lockout_max_seconds: int = DEFAULT_LOGIN_LOCKOUT_MAX_SECONDS,
    round_retention_seconds: int = DEFAULT_LOGIN_ROUND_RETENTION_SECONDS,
    on_login_success: LoginSuccessClears = "clear_all",
    fail_open: bool = False,
)

Tracks failed logins and escalates lockout duration across rounds.

Fails closed by default: if the backend errors, a login attempt is blocked rather than silently allowed. Rationale: under fail-open an attacker who can DoS the limiter backend then brute-forces with lockout disabled.

Note

Fail-closed means a network-backend outage blocks logins; pair a redis backend with HA redis (the in-memory backend has no outage mode).

Parameters:

Name	Type	Description	Default
`backend`	`RateLimiterBackend`	The shared rate-limiter backend (counters live here).	required
`max_attempts`	`int`	Failures allowed within `attempt_window_seconds` before a lockout trips.	`DEFAULT_LOGIN_MAX_ATTEMPTS`
`attempt_window_seconds`	`int`	Sliding window for counting failures.	`DEFAULT_LOGIN_ATTEMPT_WINDOW_SECONDS`
`lockout_base_seconds`	`int`	First lockout duration; doubles each round.	`DEFAULT_LOGIN_LOCKOUT_BASE_SECONDS`
`lockout_max_seconds`	`int`	Cap on the exponential lockout duration.	`DEFAULT_LOGIN_LOCKOUT_MAX_SECONDS`
`round_retention_seconds`	`int`	How long a round counter persists, so repeat offenders resume escalating rather than resetting. The TTL is slid forward on every lockout (sliding window), so a slow, paced attack keeps climbing the escalation ladder instead of forgetting it.	`DEFAULT_LOGIN_ROUND_RETENTION_SECONDS`
`on_login_success`	`LoginSuccessClears`	What a successful login clears. Both modes clear the username's own counters, lock and rounds. `"clear_all"` (default) also takes back the per-IP pressure that this username's own failures from this IP added in the current attempt window - friendly to a legitimate user behind a shared egress (corporate NAT / mobile CGNAT) who mistyped before getting in, while failures against other usernames stay counted, so logging into your own account can't launder a spray. `"clear_user_only"` keeps all per-IP pressure - tighter against a single-source brute force, but only safe when your per-IP key reliably identifies an individual client (you terminate behind a proxy with `trusted_proxy_hops` set AND your users aren't predominantly behind shared egress); otherwise it can lock out innocent co-located users. A success never clears an IP lock or the IP's escalation rounds.	`'clear_all'`
`fail_open`	`bool`	On a backend error, allow (`True`) or block (`False`).	`False`

check_and_record `async` ¶

check_and_record(
    ip_address: str, username: str, success: bool = False
) -> tuple[bool, int | None, int]

Record an attempt and report whether it's allowed.

Parameters:

Name	Type	Description	Default
`ip_address`	`str`	Caller IP (one of the two keyed dimensions). Keyed through client_ip_key, so an IPv6 client shares one budget across its `/64`.	required
`username`	`str`	Submitted username/email (the other dimension). Keyed through canonical_identifier, so case variants share one budget.	required
`success`	`bool`	When `True`, clears the username's counters (and, under `"clear_all"`, the IP pressure its own failures added) and allows.	`False`

Returns:

Type	Description
`tuple[bool, int \| None, int]`	`(allowed, attempts_remaining, retry_after_seconds)`.

Note

The escalation branch issues several sequential backend ops (per-IP and per-username lock + round counters). It runs only on repeated failures (already past the attempt cap), so it's intentionally not pipelined.

forget_attempt `async` ¶

forget_attempt(ip_address: str, username: str) -> None

Take back the one attempt a correct password recorded, leaving earlier failures.

For a login that still needs a second factor: the password neither clears the counters nor uses up the budget the codes that follow are checked against.

crudauth.ratelimit.RateLimiterBackend ¶

Bases: ABC

Async counter store backing rate limiting and login lockout.

A backend implements the dumb primitives; the package layers the fixed-window check and the escalating LockoutPolicy on top. The primitives raise on a backend error - the policy layer catches and decides fail-open vs fail-closed, so one backend can serve both fail-open window limits and fail-closed lockout.

Example

from crudauth.ratelimit import MemoryRateLimiterBackend
backend = MemoryRateLimiterBackend()
count, limited, retry_after = await backend.increment_and_check("ip:1.2.3.4", 5, 60)

increment `abstractmethod` `async` ¶

increment(
    key: str, amount: int = 1, expiry: int | None = None
) -> int

Increment key by amount, return the new count.

Note

TTL is armed on first touch only. expiry must be applied when the key is created (the increment that brings it into existence) and NOT re-armed on subsequent increments - otherwise sustained load would push the TTL forward forever and a fixed window would never close. The increment and arming the TTL must be atomic, so a failure between them can't leave a counter that never expires. Both in-tree backends honor this (Redis: INCRBY + EXPIRE NX in one transaction; memory: key absent from the deadline map); a new backend must too.

increment_and_refresh_ttl `abstractmethod` `async` ¶

increment_and_refresh_ttl(
    key: str, amount: int = 1, expiry: int | None = None
) -> int

Increment key and (re-)arm its TTL on every call, atomically.

The counterpart to increment's first-touch-only TTL: this slides the expiry forward on each call, for a counter that must live as long as activity continues - the lockout escalation "rounds" counter, so a slow, paced attack can't let it expire and reset the backoff. The increment and the TTL re-arm MUST be atomic (one round trip on a networked backend) so a concurrent attempt can't interleave between them.

get_count `abstractmethod` `async` ¶

get_count(key: str) -> int | None

Current counter value, or None if the key is absent.

reset `abstractmethod` `async` ¶

reset(key: str) -> None

Reset key to absent (alias of delete that ignores the result).

delete `abstractmethod` `async` ¶

delete(key: str) -> bool

Delete key; return whether it existed.

ping `abstractmethod` `async` ¶

ping() -> bool

Liveness check for the backing store.

get_ttl `async` ¶

get_ttl(key: str) -> int

Remaining TTL in seconds for key (0 if unknown/absent).

Overridable; the escalating lockout policy uses it. The default returns 0 for backends that can't report TTL.

Note

LockoutPolicy reads this to detect an active lockout. A backend that leaves the default 0 still counts attempts but can never hold a lockout open - so any backend used with lockout MUST implement get_ttl (both in-tree backends do).

increment_and_check `async` ¶

increment_and_check(
    key: str,
    limit: int,
    period: int,
    *,
    fail_open: bool = True,
) -> tuple[int, bool, int]

Fixed-window limit check over the dumb counter.

Override for a single-call atomic version (Redis does). The window-stamped key is what makes re-expiring on every increment correct: the key rolls each window, so old windows expire on their own rather than the TTL being pushed forever under sustained load.

Parameters:

Name	Type	Description	Default
`key`	`str`	Logical counter key (an action+identity namespace).	required
`limit`	`int`	Max events allowed within `period`.	required
`period`	`int`	Window length in seconds.	required
`fail_open`	`bool`	On a backend error, allow (`True`) or block (`False`).	`True`

Returns:

Type	Description
`tuple[int, bool, int]`	`(count, limited, retry_after_seconds)`.

initialize `async` ¶

initialize() -> None

Open connections / warm up. Default no-op.

close `async` ¶

close() -> None

Release resources. Default no-op.

crudauth.ratelimit.MemoryRateLimiterBackend ¶

MemoryRateLimiterBackend()

Bases: RateLimiterBackend

Dict-backed counters with monotonic TTLs. Not shared across processes.

Note

Window-stamped keys ({key}:{window_start}) roll every window, so an abandoned past window's key is never accessed again. Eviction is therefore lazy-on-access plus an occasional full sweep on increment, so a high-cardinality keyspace (per-IP / per-email) can't grow unbounded. Still single-process - use the redis backend in production.

increment `async` ¶

increment(
    key: str, amount: int = 1, expiry: int | None = None
) -> int

Increment; arm the TTL only on first touch (key absent from the deadline map) - the contract from RateLimiterBackend.increment.

increment_and_refresh_ttl `async` ¶

increment_and_refresh_ttl(
    key: str, amount: int = 1, expiry: int | None = None
) -> int

Increment and slide the TTL forward on every call (see the base method).

crudauth.ratelimit.redis_rate_limiter ¶

redis_rate_limiter(
    redis_url: str | None = None, client: Any = None
) -> RateLimiterBackend

Construct a Redis rate-limiter backend, guarding the optional dependency.

Parameters:

Name	Type	Description	Default
`redis_url`	`str \| None`	Connection URL (defaults to localhost when omitted).	`None`
`client`	`Any`	A pre-built `redis.asyncio` client to reuse instead of a URL.	`None`

Returns:

Type	Description
`RateLimiterBackend`	A [RedisBackend][crudauth.ratelimit.backends.redis.RedisBackend].

Raises:

Type	Description
`ImportError`	If the `redis` extra isn't installed.

Example

CRUDAuth(..., rate_limiter=redis_rate_limiter(redis_url=settings.REDIS_URL))

Rate limiting & lockout¶

crudauth.ratelimit.RateLimit dataclass ¶

disabled property ¶

crudauth.ratelimit.RateLimitResolver module-attribute ¶

crudauth.ratelimit.KeyBy ¶

crudauth.ratelimit.RateLimitHeadersMiddleware ¶

crudauth.ratelimit.LockoutConfig dataclass ¶

crudauth.ratelimit.LockoutPolicy ¶

check_and_record async ¶

forget_attempt async ¶

crudauth.ratelimit.RateLimiterBackend ¶

increment abstractmethod async ¶

increment_and_refresh_ttl abstractmethod async ¶

get_count abstractmethod async ¶

reset abstractmethod async ¶

delete abstractmethod async ¶

ping abstractmethod async ¶

get_ttl async ¶

increment_and_check async ¶

initialize async ¶

close async ¶

crudauth.ratelimit.MemoryRateLimiterBackend ¶

increment async ¶

increment_and_refresh_ttl async ¶

crudauth.ratelimit.redis_rate_limiter ¶

crudauth.ratelimit.RateLimit `dataclass` ¶

disabled `property` ¶

crudauth.ratelimit.RateLimitResolver `module-attribute` ¶

crudauth.ratelimit.LockoutConfig `dataclass` ¶

check_and_record `async` ¶

forget_attempt `async` ¶

increment `abstractmethod` `async` ¶

increment_and_refresh_ttl `abstractmethod` `async` ¶

get_count `abstractmethod` `async` ¶

reset `abstractmethod` `async` ¶

delete `abstractmethod` `async` ¶

ping `abstractmethod` `async` ¶

get_ttl `async` ¶

increment_and_check `async` ¶

initialize `async` ¶

close `async` ¶

increment `async` ¶

increment_and_refresh_ttl `async` ¶