lukasandClaude Opus 5.5
0ddc0cb001
Branding: the page titles and app name are now "Ultimo Email". Admin panel (now two columns with a divider): - Users: search by login or by a mail account's address or name (results only, nothing listed until something is typed). A user opens a window to change their quota, block/unblock them, or delete them; block and delete ask for confirmation, and an admin cannot do either to their own account. - Blocking (users.blocked_at, migration 0019) refuses sign-in with a clear message and revokes every session; a new user.signed_out event sends the user's open tabs to the sign-in screen at once. Any "not signed in" answer now does the same, not only /auth/me. Deleting removes mail accounts (sync stops), plugin rows (cleared inside the user's RLS context, since plugin tables have no FK to users), blobs, the search index, sessions and the user. - Outgoing system mail (migration 0020): SMTP server, port, security, login, password (vault-encrypted, never sent back), From address and the public address reset links point to, plus a "Send test" that reports the server's error. Password reset: - "Forgot password?" on the sign-in page asks for the user name and always answers "If this account exists you will receive an email to your recovery email address"; the mail goes out after the response, so neither the reply nor its timing reveals an account, and it is capped at 3 per hour. - The emailed link (token in the URL fragment, single use, 30 minutes) opens a page to set a new password. With 2-step verification on, the code is checked before anything changes; five wrong codes burn the link. - Settings > Profile: the recovery email can be added, changed or removed, confirmed with the current password, which must be typed — autofill and paste are refused (TypedPasswordInput). Also adds frontend/icons/ultimo-logo-bimi.svg, an SVG Tiny PS version of the logo for a BIMI record. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""Password reset tokens (SPEC §6.1).
|
|
|
|
Single-use, 30-minute, Redis-backed tokens, emailed as a link to the account's
|
|
recovery address through the system mailer (:mod:`app.identity.mailer`).
|
|
|
|
A token survives a wrong 2FA code (the page asks again) but not five: after
|
|
that it is burned, so the link cannot be used to guess codes. Requests are
|
|
capped per account so the reset form cannot be used to flood someone's inbox.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
import uuid
|
|
|
|
from app.redis_client import aw, get_redis
|
|
|
|
_PREFIX = "pwreset:"
|
|
_FAILS = "pwreset-fail:"
|
|
_RATE = "pwreset-rate:"
|
|
_TTL = 30 * 60 # 30 minutes
|
|
MAX_TOTP_FAILURES = 5
|
|
MAX_REQUESTS_PER_HOUR = 3
|
|
|
|
|
|
async def issue_token(user_id: uuid.UUID) -> str:
|
|
"""Create a single-use reset token bound to ``user_id`` and return it."""
|
|
token = secrets.token_urlsafe(32)
|
|
await get_redis().set(_PREFIX + token, str(user_id), ex=_TTL)
|
|
return token
|
|
|
|
|
|
async def request_allowed(user_id: uuid.UUID) -> bool:
|
|
"""Count a reset request; False once the account has had its hourly share."""
|
|
redis = get_redis()
|
|
key = _RATE + str(user_id)
|
|
count = int(await redis.incr(key))
|
|
if count == 1:
|
|
await redis.expire(key, 3600)
|
|
return count <= MAX_REQUESTS_PER_HOUR
|
|
|
|
|
|
def _user_id(raw: str | None) -> uuid.UUID | None:
|
|
if raw is None:
|
|
return None
|
|
try:
|
|
return uuid.UUID(raw)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
async def peek_token(token: str) -> uuid.UUID | None:
|
|
"""The user a still-valid token belongs to, without using it up."""
|
|
if not token:
|
|
return None
|
|
return _user_id(await get_redis().get(_PREFIX + token))
|
|
|
|
|
|
async def consume_token(token: str) -> uuid.UUID | None:
|
|
"""Atomically consume a reset token, returning the user_id (or None)."""
|
|
if not token:
|
|
return None
|
|
redis = get_redis()
|
|
raw = await aw(redis.getdel(_PREFIX + token)) # single-use: fetch and delete atomically
|
|
await redis.delete(_FAILS + token)
|
|
return _user_id(raw)
|
|
|
|
|
|
async def record_totp_failure(token: str) -> None:
|
|
"""Count a wrong 2FA code against the token; burn it at the limit."""
|
|
redis = get_redis()
|
|
fails = await redis.incr(_FAILS + token)
|
|
if fails == 1:
|
|
await redis.expire(_FAILS + token, _TTL)
|
|
if fails >= MAX_TOTP_FAILURES:
|
|
await redis.delete(_PREFIX + token, _FAILS + token)
|