Files
FabledSteward/steward/auth/throttle.py
T
bvandeusenandClaude Opus 5.5 593b583d23
CI / lint (push) Failing after 3s
CI / integration (push) Successful in 1m37s
CI / unit (push) Failing after 1m43s
CI / publish (push) Skipped
feat(auth): reset a forgotten password from an emailed link (milestone 497)
Step #5429 "Reset by email: request, link and new-password page" of
milestone 497 "Passwords: change your own, and reset it by email".

- Sign-in page: "Forgot your password?" while a mail server is set.
- /forgot takes a username or email and answers the same either way; only
  an active local account with a real address gets mail. Requests count
  per address on their own throttle, so they never lock out sign-in.
- password_resets (migration 0039): a one-time key, stored as its sha256,
  working for an hour; asking again replaces it.
- /reset/<key>: the new password, then signed straight in; every other
  browser is signed out (sessions_since). Bad keys count as invite keys do.
- Email "[Steward] Reset your password", in the account's own zone.
- Audited: user.password_reset_requested, user.password_reset_by_email.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-08 20:17:24 -04:00

53 lines
2.0 KiB
Python

"""Pause password guessing: after too many failed logins from one address,
that address's logins are refused for a while.
Keyed by client address (steward/core/proxy.py: behind a trusted proxy, the
address it reports), not by username, so nobody can lock a user out by
failing logins in their name. Held in memory: a restart forgets it, which
costs an attacker nothing they couldn't get by waiting, and keeps it free of
the database. The limits are Settings → General (login.max_failures,
login.lockout_minutes).
"""
from __future__ import annotations
import time
from collections import deque
class LoginThrottle:
def __init__(self) -> None:
self._failures: dict[str, deque[float]] = {}
def _recent(self, key: str, window_s: float, now: float) -> deque[float]:
times = self._failures.get(key)
if times is None:
return deque()
while times and times[0] <= now - window_s:
times.popleft()
if not times:
del self._failures[key]
return times
def retry_after(self, key: str, max_failures: int, window_s: float,
now: float | None = None) -> int:
"""Seconds until `key` may try again; 0 when it may try now."""
now = time.monotonic() if now is None else now
times = self._recent(key, window_s, now)
if len(times) < max_failures:
return 0
return max(1, int(times[-max_failures] + window_s - now) + 1)
def failed(self, key: str, now: float | None = None) -> None:
now = time.monotonic() if now is None else now
self._failures.setdefault(key, deque()).append(now)
def succeeded(self, key: str) -> None:
self._failures.pop(key, None)
login_throttle = LoginThrottle()
# Password-reset emails asked for from one address (milestone 497): every
# request counts, matched or not, so it can't be used to find accounts, and
# kept apart from login_throttle so asking for links never locks out sign-in.
reset_throttle = LoginThrottle()