Admin user management, password reset by email, and the Ultimo Email name

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>

backend/app/api/admin.py

@@ -1,24 +1,40 @@
 """Instance administration (SPEC §6.1).
 
-The backend seam for the admin panel. Today it carries only the registration
-gate — the first account on a fresh install becomes the administrator and closes
-signup behind itself, and this is how that gets reopened. Further instance-wide
-options land here as the panel grows.
+The backend seam for the admin panel: the registration gate — the first account
+on a fresh install becomes the administrator and closes signup behind itself,
+and this is how that gets reopened — the default quota, and managing users
+(search, per-user quota, block, delete). An administrator cannot block or delete
+their own account, so the instance can never lock out its last admin that way.
 """
 
 from __future__ import annotations
 
 import datetime as dt
+import uuid
 from typing import Annotated
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.api.errors import raise_api_error
 from app.db import get_session
 from app.identity.deps import CurrentAdmin, require_csrf
-from app.identity.schemas import AdminSettingsIn, AdminSettingsOut
+from app.identity.mailer import MailerError, mailer_config, send_mail
+from app.identity.schemas import (
+    AdminMailIn,
+    AdminMailOut,
+    AdminMailTestIn,
+    AdminSettingsIn,
+    AdminSettingsOut,
+    AdminUserOut,
+    AdminUserPatch,
+    MessageOut,
+)
 from app.identity.service import app_settings
-from app.models import AppSettings
+from app.identity.user_admin import delete_user, search_users, set_blocked
+from app.models import Account, AppSettings, Message, User
+from app.security.crypto import encrypt_secret
 
 router = APIRouter(prefix="/admin", tags=["admin"])
 
@@ -54,3 +70,134 @@ async def write_settings(
     row.updated_at = dt.datetime.now(dt.UTC)
     await db.commit()
     return _out(row)
+
+
+# --- Outgoing system mail -------------------------------------------------------
+
+
+def _mail_out(row: AppSettings) -> AdminMailOut:
+    return AdminMailOut(
+        smtp_host=row.smtp_host,
+        smtp_port=row.smtp_port,
+        smtp_security=row.smtp_security,
+        smtp_username=row.smtp_username,
+        has_password=row.smtp_secret_ct is not None,
+        smtp_from=row.smtp_from,
+        public_url=row.public_url,
+        configured=bool(row.smtp_host and row.smtp_from and row.public_url),
+    )
+
+
+@router.get("/mail", response_model=AdminMailOut)
+async def read_mail(admin: CurrentAdmin, db: DbSession) -> AdminMailOut:
+    return _mail_out(await app_settings(db))
+
+
+@router.put("/mail", response_model=AdminMailOut, dependencies=[Depends(require_csrf)])
+async def write_mail(body: AdminMailIn, admin: CurrentAdmin, db: DbSession) -> AdminMailOut:
+    """Merge the supplied SMTP fields. The password is stored vault-encrypted."""
+    row = await app_settings(db, for_update=True)
+    for name in ("smtp_host", "smtp_username", "smtp_from"):
+        value = getattr(body, name)
+        if value is not None:
+            setattr(row, name, value.strip())
+    if body.smtp_port is not None:
+        row.smtp_port = body.smtp_port
+    if body.smtp_security is not None:
+        row.smtp_security = body.smtp_security
+    if body.smtp_password is not None:
+        row.smtp_secret_ct = (
+            encrypt_secret(body.smtp_password.encode()) if body.smtp_password else None
+        )
+    if body.public_url is not None:
+        url = body.public_url.strip().rstrip("/")
+        if url and not url.startswith(("http://", "https://")):
+            raise_api_error(400, "invalid_url", "the address must start with http:// or https://")
+        row.public_url = url
+    row.updated_at = dt.datetime.now(dt.UTC)
+    await db.commit()
+    return _mail_out(row)
+
+
+@router.post("/mail/test", response_model=MessageOut, dependencies=[Depends(require_csrf)])
+async def test_mail(body: AdminMailTestIn, admin: CurrentAdmin, db: DbSession) -> MessageOut:
+    """Send a test message with the saved settings, reporting the server's error."""
+    cfg = mailer_config(await app_settings(db))
+    if cfg is None:
+        raise_api_error(
+            409, "mail_not_configured", "set the SMTP server, From address and public address first"
+        )
+    text = (
+        "This is a test message from Ultimo Email.\n\n"
+        "Outgoing system mail works: password-reset links will be sent from this account.\n"
+    )
+    html = f"<p>{text.replace(chr(10) * 2, '</p><p>').strip()}</p>"
+    try:
+        await send_mail(cfg, body.to.strip(), "Ultimo Email test message", text, html)
+    except MailerError as exc:
+        raise_api_error(502, "mail_failed", f"sending failed: {exc}")
+    return MessageOut()
+
+
+# --- Users --------------------------------------------------------------------
+
+
+def _user_out(user: User, used_bytes: int, accounts: list[str]) -> AdminUserOut:
+    return AdminUserOut(
+        id=user.id,
+        handle=user.handle,
+        is_admin=user.is_admin,
+        blocked_at=user.blocked_at.isoformat() if user.blocked_at else None,
+        quota_bytes=user.quota_bytes,
+        used_bytes=used_bytes,
+        created_at=user.created_at.isoformat(),
+        accounts=accounts,
+    )
+
+
+async def _target(db: AsyncSession, admin: User, user_id: uuid.UUID) -> User:
+    """The user an admin action is aimed at; never the admin themselves."""
+    user = await db.scalar(select(User).where(User.id == user_id))
+    if user is None:
+        raise_api_error(404, "not_found", "user not found")
+    if user.id == admin.id:
+        raise_api_error(409, "cannot_modify_self", "you cannot do this to your own account")
+    return user
+
+
+@router.get("/users", response_model=list[AdminUserOut])
+async def list_users(
+    admin: CurrentAdmin, db: DbSession, q: str = Query("", max_length=200)
+) -> list[AdminUserOut]:
+    """Search users by login, or by a mail account's address or display name."""
+    return [_user_out(u, used, emails) for u, used, emails in await search_users(db, q)]
+
+
+@router.patch("/users/{user_id}", response_model=AdminUserOut, dependencies=[Depends(require_csrf)])
+async def patch_user(
+    user_id: uuid.UUID, body: AdminUserPatch, admin: CurrentAdmin, db: DbSession
+) -> AdminUserOut:
+    """Change a user's quota and/or block or unblock them."""
+    user = await _target(db, admin, user_id)
+    if body.quota_bytes is not None:
+        user.quota_bytes = body.quota_bytes
+    if body.blocked is not None:
+        await set_blocked(db, user, body.blocked)
+    await db.commit()
+    used = await db.scalar(
+        select(func.coalesce(func.sum(Message.size_bytes), 0)).where(Message.user_id == user.id)
+    )
+    emails = list(
+        await db.scalars(
+            select(Account.email).where(Account.user_id == user.id).order_by(Account.email)
+        )
+    )
+    return _user_out(user, int(used or 0), emails)
+
+
+@router.delete("/users/{user_id}", dependencies=[Depends(require_csrf)])
+async def remove_user(user_id: uuid.UUID, admin: CurrentAdmin, db: DbSession) -> dict[str, str]:
+    """Delete a user and all of their data. Cannot be undone."""
+    user = await _target(db, admin, user_id)
+    await delete_user(db, user)
+    return {"status": "deleted"}

backend/app/events.py

@@ -41,6 +41,9 @@ ALLOWED_EVENTS: frozenset[str] = frozenset(
         # Sync v2 §3: published on login and every WS connect; drives tier
         # promotion in the sync workers (§5). Payload: {user_id}.
         "user.active",
+        # The user's sessions were revoked by an administrator (blocked or
+        # deleted): open tabs drop to the sign-in screen. Payload: {user_id}.
+        "user.signed_out",
     }
 )
 

backend/app/identity/deps.py

@@ -30,8 +30,9 @@ async def current_user(
 ) -> User:
     """Load the authenticated :class:`User` (I-7: scoped by the session's user_id)."""
     user = await db.scalar(select(User).where(User.id == session.user_id))
-    if user is None:
-        # Session references a deleted user; treat as unauthenticated.
+    if user is None or user.blocked_at is not None:
+        # Session references a deleted or blocked user; treat as unauthenticated.
+        # (Blocking also revokes the sessions; this covers any request in flight.)
         raise_api_error(401, "not_authenticated", "authentication required")
     return user
 

backend/app/identity/mailer.py

@@ -0,0 +1,110 @@
+"""Outgoing system mail (SPEC §6.1): mail the server sends on its own behalf.
+
+Today that is only the password-reset link. It goes out through the SMTP
+account the administrator configured in the admin panel (``app_settings``),
+never through a user's own mail account — the whole point is that the user
+cannot sign in. The administrator's server is trusted, so unlike user mail
+accounts this connection is not held to the SSRF guard (it may well be a relay
+on the LAN).
+"""
+
+from __future__ import annotations
+
+import html as html_lib
+import logging
+from dataclasses import dataclass
+from email.message import EmailMessage
+from email.utils import formataddr, make_msgid
+
+import aiosmtplib
+
+from app.models import AppSettings
+from app.security.crypto import decrypt_secret
+
+logger = logging.getLogger(__name__)
+
+SENDER_NAME = "Ultimo Email"
+_TIMEOUT = 20.0
+
+
+class MailerError(Exception):
+    """System mail is not configured, or the SMTP server refused the message."""
+
+
+@dataclass(frozen=True)
+class MailerConfig:
+    host: str
+    port: int
+    security: str  # starttls | ssl | none
+    username: str
+    password: str
+    sender: str
+    public_url: str
+
+
+def mailer_config(row: AppSettings) -> MailerConfig | None:
+    """The usable configuration, or None while the admin has not set one up."""
+    if not (row.smtp_host and row.smtp_from and row.public_url):
+        return None
+    password = decrypt_secret(row.smtp_secret_ct).decode() if row.smtp_secret_ct else ""
+    return MailerConfig(
+        host=row.smtp_host,
+        port=row.smtp_port,
+        security=row.smtp_security,
+        username=row.smtp_username,
+        password=password,
+        sender=row.smtp_from,
+        public_url=row.public_url.rstrip("/"),
+    )
+
+
+async def send_mail(cfg: MailerConfig, to: str, subject: str, text: str, html: str) -> None:
+    msg = EmailMessage()
+    msg["From"] = formataddr((SENDER_NAME, cfg.sender))
+    msg["To"] = to
+    msg["Subject"] = subject
+    msg["Message-ID"] = make_msgid(domain=cfg.sender.rsplit("@", 1)[-1])
+    msg["Auto-Submitted"] = "auto-generated"  # RFC 3834: do not auto-reply to this
+    msg.set_content(text)
+    msg.add_alternative(html, subtype="html")
+    try:
+        await aiosmtplib.send(
+            msg,
+            hostname=cfg.host,
+            port=cfg.port,
+            username=cfg.username or None,
+            password=cfg.password or None,
+            use_tls=cfg.security == "ssl",
+            start_tls=cfg.security == "starttls",
+            timeout=_TIMEOUT,
+        )
+    except (aiosmtplib.SMTPException, OSError) as exc:
+        raise MailerError(str(exc) or type(exc).__name__) from exc
+
+
+def reset_mail(link: str, handle: str) -> tuple[str, str, str]:
+    """(subject, text, html) of the password-reset message."""
+    subject = "Reset your Ultimo Email password"
+    text = (
+        f"Someone asked to change the password of the Ultimo Email account {handle}.\n\n"
+        f"To choose a new password, open this link within 30 minutes:\n\n{link}\n\n"
+        "The link works once. If you did not ask for this, ignore this email — "
+        "your password stays as it is.\n"
+    )
+    h, u = html_lib.escape(handle), html_lib.escape(link, quote=True)
+    muted = "font-size:13px;color:#6b7280"
+    button = (
+        "display:inline-block;background:#2563eb;color:#ffffff;text-decoration:none;"
+        "padding:10px 18px;border-radius:8px;font-weight:bold"
+    )
+    body = "font-family:Arial,Helvetica,sans-serif;font-size:15px;color:#1f2937;line-height:1.5"
+    html = (
+        f'<!doctype html>\n<html><body style="{body}">\n'
+        f"<p>Someone asked to change the password of the Ultimo Email account <b>{h}</b>.</p>\n"
+        "<p>To choose a new password, open this link within 30 minutes:</p>\n"
+        f'<p><a href="{u}" style="{button}">Change password</a></p>\n'
+        f'<p style="{muted}">Or paste this address into your browser:<br>{u}</p>\n'
+        f'<p style="{muted}">The link works once. If you did not ask for this, ignore this '
+        "email — your password stays as it is.</p>\n</body></html>\n"
+    )
+    return subject, text, html

backend/app/identity/reset.py

@@ -1,43 +1,76 @@
 """Password reset tokens (SPEC §6.1).
 
-Single-use, 30-minute, Redis-backed tokens. Only issued when the account has a
-recovery email on file. Actual delivery of the token to the recovery address
-requires an app-level mailer (not part of M0/M1); until then the token is logged
-server-side. See docs/open-questions.md.
+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 logging
 import secrets
 import uuid
 
 from app.redis_client import aw, get_redis
 
-logger = logging.getLogger(__name__)
-
 _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, recovery_email: str) -> str:
+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)
-    # TODO(spec): deliver via the recovery address once an app mailer exists.
-    logger.info("password reset token issued for user=%s recovery=%s", user_id, recovery_email)
     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
-    if raw is None:
-        return None
-    try:
-        return uuid.UUID(raw)
-    except ValueError:
-        return None
+    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)

backend/app/identity/routes.py

@@ -3,9 +3,11 @@ TOTP setup/enable, session list/revoke, password reset."""
 
 from __future__ import annotations
 
+import logging
+import re
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Request, Response
+from fastapi import APIRouter, BackgroundTasks, Depends, Request, Response
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -20,13 +22,23 @@ from app.identity.cookies import (
     set_auth_cookies,
 )
 from app.identity.deps import CurrentSession, CurrentUser, require_csrf
+from app.identity.mailer import MailerConfig, MailerError, mailer_config, reset_mail, send_mail
 from app.identity.passwords import hash_password, verify_password
-from app.identity.reset import consume_token, issue_token
+from app.identity.reset import (
+    consume_token,
+    issue_token,
+    peek_token,
+    record_totp_failure,
+    request_allowed,
+)
 from app.identity.schemas import (
     LoginRequest,
     MessageOut,
+    PasswordResetCheckIn,
+    PasswordResetCheckOut,
     PasswordResetIn,
     PasswordResetRequestIn,
+    RecoveryEmailIn,
     RegisterRequest,
     RegistrationStatusOut,
     SessionOut,
@@ -36,10 +48,12 @@ from app.identity.schemas import (
     UserOut,
 )
 from app.identity.service import (
+    AccountBlocked,
     RateLimited,
     RegistrationError,
     TotpInvalid,
     TotpRequired,
+    app_settings,
     authenticate,
     register_user,
     registration_status,
@@ -56,8 +70,14 @@ from app.models import User
 from app.redis_client import get_redis
 from app.sync import leases
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/auth", tags=["auth"])
 
+# Deliberately loose: the recovery address only has to be something mail can be
+# sent to; the mail server is the real judge.
+_EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+")
+
 DbSession = Annotated[AsyncSession, Depends(get_session)]
 
 
@@ -128,6 +148,10 @@ async def login(body: LoginRequest, request: Request, response: Response, db: Db
         raise_api_error(401, "totp_required", "a 2FA code is required")
     except TotpInvalid:
         raise_api_error(401, "totp_invalid", "incorrect 2FA code")
+    except AccountBlocked:
+        raise_api_error(
+            403, "account_blocked", "this account has been blocked by the administrator"
+        )
     if user is None:
         raise_api_error(401, "invalid_credentials", "incorrect handle or password")
     await _establish_session(request, response, user)
@@ -209,25 +233,86 @@ async def revoke(sid: str, user: CurrentUser) -> MessageOut:
 # --- Password reset (SPEC §6.1) ---------------------------------------------
 
 
+async def _send_reset_mail(cfg: MailerConfig, to: str, handle: str, token: str) -> None:
+    """Runs after the response has gone out (see password_reset_request)."""
+    # The token rides in the fragment, which browsers never send to a server, so
+    # it stays out of access logs and Referer headers.
+    subject, text, html = reset_mail(f"{cfg.public_url}/reset-password#token={token}", handle)
+    try:
+        await send_mail(cfg, to, subject, text, html)
+    except MailerError as exc:
+        logger.warning("password reset mail for %s could not be sent: %s", handle, exc)
+
+
 @router.post("/password/reset-request", response_model=MessageOut)
-async def password_reset_request(body: PasswordResetRequestIn, db: DbSession) -> MessageOut:
-    # Always return ok (do not leak whether the handle exists / has recovery mail).
+async def password_reset_request(
+    body: PasswordResetRequestIn, background: BackgroundTasks, db: DbSession
+) -> MessageOut:
+    """Email a reset link to the account's recovery address.
+
+    The answer is the same whether or not the account exists, has a recovery
+    address, or hit the hourly cap, and the mail is sent after the response —
+    so neither the reply nor its timing tells a stranger anything.
+    """
     handle = resolve_handle(body.handle)
     user = await db.scalar(select(User).where(User.handle == handle))
-    if user is not None and user.recovery_email:
-        await issue_token(user.id, user.recovery_email)
+    if user is None or not user.recovery_email:
+        return MessageOut()
+    cfg = mailer_config(await app_settings(db))
+    if cfg is None:
+        logger.warning("password reset requested for %s, but system mail is not set up", handle)
+        return MessageOut()
+    if not await request_allowed(user.id):
+        return MessageOut()
+    token = await issue_token(user.id)
+    background.add_task(_send_reset_mail, cfg, user.recovery_email, user.handle, token)
     return MessageOut()
 
 
+@router.post("/password/reset/check", response_model=PasswordResetCheckOut)
+async def password_reset_check(body: PasswordResetCheckIn, db: DbSession) -> PasswordResetCheckOut:
+    """Whether a reset link still works, so the page can say so before asking
+    for a new password. Does not use the link up."""
+    user_id = await peek_token(body.token)
+    user = await db.get(User, user_id) if user_id else None
+    if user is None:
+        return PasswordResetCheckOut(valid=False)
+    return PasswordResetCheckOut(valid=True, totp_required=user.totp_secret is not None)
+
+
 @router.post("/password/reset", response_model=MessageOut)
 async def password_reset(body: PasswordResetIn, db: DbSession) -> MessageOut:
-    user_id = await consume_token(body.token)
-    if user_id is None:
-        raise_api_error(400, "invalid_token", "reset token is invalid or expired")
-    user = await db.get(User, user_id)
+    """Set a new password with a reset link. With 2FA on, the code is checked
+    before anything changes; a missing or wrong code leaves the link usable
+    (up to MAX_TOTP_FAILURES wrong codes)."""
+    user_id = await peek_token(body.token)
+    user = await db.get(User, user_id) if user_id else None
     if user is None:
-        raise_api_error(400, "invalid_token", "reset token is invalid or expired")
+        raise_api_error(400, "invalid_token", "this link is invalid or has expired")
+    if user.totp_secret is not None:
+        if not body.totp:
+            raise_api_error(401, "totp_required", "a 2FA code is required")
+        if not totp_mod.verify(totp_mod.decrypt_totp_secret(user.totp_secret), body.totp):
+            await record_totp_failure(body.token)
+            raise_api_error(401, "totp_invalid", "incorrect 2FA code")
+    # Only now is the link spent — and only if nobody spent it in the meantime.
+    if await consume_token(body.token) != user.id:
+        raise_api_error(400, "invalid_token", "this link is invalid or has expired")
     user.password_hash = hash_password(body.password)
     await db.commit()
-    await destroy_all_sessions(user_id)  # force re-login everywhere after reset
+    await destroy_all_sessions(user.id)  # force re-login everywhere after reset
     return MessageOut()
+
+
+@router.put("/recovery-email", response_model=UserOut, dependencies=[Depends(require_csrf)])
+async def set_recovery_email(body: RecoveryEmailIn, user: CurrentUser, db: DbSession) -> UserOut:
+    """Add, change or remove the recovery address. Takes the password: whoever
+    controls this address can take the account over with a reset link."""
+    if not verify_password(user.password_hash, body.password):
+        raise_api_error(403, "invalid_password", "incorrect password")
+    email = (body.recovery_email or "").strip().lower() or None
+    if email is not None and not _EMAIL_RE.fullmatch(email):
+        raise_api_error(400, "invalid_email", "enter a valid email address")
+    user.recovery_email = email
+    await db.commit()
+    return _user_out(user)

backend/app/identity/schemas.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import uuid
+from typing import Literal
 
 from pydantic import BaseModel, Field
 
@@ -47,6 +48,23 @@ class PasswordResetRequestIn(BaseModel):
 class PasswordResetIn(BaseModel):
     token: str = Field(min_length=1)
     password: str = Field(min_length=8, max_length=1024)
+    totp: str | None = None  # required when the account has 2FA enabled
+
+
+class PasswordResetCheckIn(BaseModel):
+    token: str = Field(min_length=1)
+
+
+class PasswordResetCheckOut(BaseModel):
+    valid: bool
+    # Tells the page to expect the 2FA step; only the link's holder learns it.
+    totp_required: bool = False
+
+
+class RecoveryEmailIn(BaseModel):
+    password: str = Field(min_length=1, max_length=1024)
+    # None (or blank) removes the recovery address.
+    recovery_email: str | None = Field(default=None, max_length=320)
 
 
 class UserOut(BaseModel):
@@ -79,5 +97,55 @@ class AdminSettingsIn(BaseModel):
     default_quota_bytes: int | None = Field(default=None, gt=0)
 
 
+class AdminMailOut(BaseModel):
+    smtp_host: str
+    smtp_port: int
+    smtp_security: str
+    smtp_username: str
+    # The password itself never leaves the server.
+    has_password: bool
+    smtp_from: str
+    public_url: str
+    # Host, From address and public address are all set: reset mail can go out.
+    configured: bool
+
+
+class AdminMailIn(BaseModel):
+    """Merge semantics: omitted fields are left as they are."""
+
+    smtp_host: str | None = Field(default=None, max_length=255)
+    smtp_port: int | None = Field(default=None, ge=1, le=65535)
+    smtp_security: Literal["starttls", "ssl", "none"] | None = None
+    smtp_username: str | None = Field(default=None, max_length=320)
+    # Omitted keeps the stored password; an empty string removes it.
+    smtp_password: str | None = Field(default=None, max_length=1024)
+    smtp_from: str | None = Field(default=None, max_length=320)
+    public_url: str | None = Field(default=None, max_length=500)
+
+
+class AdminMailTestIn(BaseModel):
+    to: str = Field(min_length=3, max_length=320)
+
+
+class AdminUserOut(BaseModel):
+    id: uuid.UUID
+    handle: str
+    is_admin: bool
+    # ISO timestamp while blocked; null otherwise.
+    blocked_at: str | None
+    quota_bytes: int
+    used_bytes: int
+    created_at: str
+    # The user's connected mail accounts, so the admin can tell who this is.
+    accounts: list[str]
+
+
+class AdminUserPatch(BaseModel):
+    """Merge semantics: omitted fields are left as they are."""
+
+    quota_bytes: int | None = Field(default=None, gt=0)
+    blocked: bool | None = None
+
+
 class MessageOut(BaseModel):
     status: str = "ok"

backend/app/identity/service.py

@@ -53,6 +53,10 @@ class TotpInvalid(Exception):
     """Password verified but the supplied TOTP code was wrong."""
 
 
+class AccountBlocked(Exception):
+    """Password verified but an administrator has blocked the account."""
+
+
 def resolve_handle(handle_input: str) -> str:
     """Normalize a login identifier to the full ``localpart@APP_DOMAIN`` form."""
     text = handle_input.strip().lower()
@@ -164,6 +168,9 @@ async def authenticate(
     ok = verify_password(stored_hash, password)
     if user is None or not ok:
         return None
+    # Only after the password: a stranger must not learn that a handle is blocked.
+    if user.blocked_at is not None:
+        raise AccountBlocked
 
     if user.totp_secret is not None:
         from app.identity.totp import decrypt_totp_secret

backend/app/identity/user_admin.py

@@ -0,0 +1,125 @@
+"""Administrator actions on user accounts (SPEC §6.1 admin panel).
+
+Search, per-user quota, blocking and deletion. Blocking keeps every byte of the
+user's data (and keeps their mail syncing) so unblocking is instant; it only
+refuses sign-in and revokes the sessions they hold. Deletion is total: mail
+accounts (through the same path the user's own "remove account" takes, so the
+sync runners stop), plugin rows, blobs on disk, the search index and the user
+row, whose foreign keys cascade to everything else core owns.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import shutil
+import uuid
+from pathlib import Path
+
+from sqlalchemy import func, or_, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.accounts.service import delete_account
+from app.config import get_settings
+from app.events import get_event_bus
+from app.identity.sessions import destroy_all_sessions
+from app.models import Account, Message, User
+from app.plugins.db import plugin_session
+from app.search.factory import get_search_backend
+
+SEARCH_LIMIT = 25
+
+
+async def search_users(db: AsyncSession, q: str) -> list[tuple[User, int, list[str]]]:
+    """Users whose login, or one of whose mail accounts' address or display name,
+    contains ``q`` (case-insensitive). An empty ``q`` lists the first users by
+    login. Each row comes with its storage used and its mail account addresses."""
+    stmt = select(User).order_by(User.handle).limit(SEARCH_LIMIT)
+    q = q.strip()
+    if q:
+        # Typed % and _ are literal characters, not wildcards.
+        escaped = q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        like = f"%{escaped}%"
+        by_account = select(Account.user_id).where(
+            or_(Account.email.ilike(like), Account.display_name.ilike(like))
+        )
+        stmt = stmt.where(or_(User.handle.ilike(like), User.id.in_(by_account)))
+    users = list(await db.scalars(stmt))
+    if not users:
+        return []
+    ids = [u.id for u in users]
+    used: dict[uuid.UUID, int] = {
+        user_id: int(total)
+        for user_id, total in await db.execute(
+            select(Message.user_id, func.coalesce(func.sum(Message.size_bytes), 0))
+            .where(Message.user_id.in_(ids))
+            .group_by(Message.user_id)
+        )
+    }
+    emails: dict[uuid.UUID, list[str]] = {}
+    for user_id, email in await db.execute(
+        select(Account.user_id, Account.email)
+        .where(Account.user_id.in_(ids))
+        .order_by(Account.email)
+    ):
+        emails.setdefault(user_id, []).append(email)
+    return [(u, int(used.get(u.id, 0)), emails.get(u.id, [])) for u in users]
+
+
+async def set_blocked(db: AsyncSession, user: User, blocked: bool) -> None:
+    """Block or unblock. Blocking also signs the user out everywhere."""
+    if blocked and user.blocked_at is None:
+        user.blocked_at = dt.datetime.now(dt.UTC)
+        await db.flush()
+        await _sign_out_everywhere(user.id)
+    elif not blocked:
+        user.blocked_at = None
+
+
+async def _sign_out_everywhere(user_id: uuid.UUID) -> None:
+    """Revoke every session, and tell the user's open tabs so they leave at once
+    instead of showing already-loaded mail until their next request."""
+    await destroy_all_sessions(user_id)
+    await get_event_bus().publish("user.signed_out", {"user_id": str(user_id)})
+
+
+async def _plugin_tables(db: AsyncSession) -> list[str]:
+    """Every plugin table holding per-user rows (``plg_*`` with a ``user_id``)."""
+    rows = await db.execute(
+        text(
+            "SELECT table_name FROM information_schema.columns "
+            "WHERE table_schema = current_schema() AND column_name = 'user_id' "
+            "AND table_name LIKE 'plg\\_%' ORDER BY table_name"
+        )
+    )
+    return [r[0] for r in rows]
+
+
+async def delete_user(db: AsyncSession, user: User) -> None:
+    """Delete ``user`` and everything they own. Commits."""
+    user_id = user.id
+
+    # Mail accounts first, one by one: this is what stops their sync runners and
+    # tidies threads, exactly as when the user removes an account themselves.
+    for account_id in list(await db.scalars(select(Account.id).where(Account.user_id == user_id))):
+        await delete_account(db, user_id=user_id, account_id=account_id)
+    await db.commit()
+
+    # Plugin tables carry no foreign key to users and sit behind row-level
+    # security, so they are cleared from inside the user's own RLS context.
+    tables = await _plugin_tables(db)
+    if tables:
+        async with plugin_session(user_id) as ps:
+            for table in tables:
+                await ps.execute(
+                    text(f'DELETE FROM "{table}" WHERE user_id = :uid'), {"uid": user_id}
+                )
+
+    await _sign_out_everywhere(user_id)
+    await get_search_backend().delete_user(user_id)
+
+    # The user row: its foreign keys cascade to the rest of core's tables.
+    await db.delete(user)
+    await db.commit()
+
+    # Whatever blobs are left (drafts' attachments and the like) go with the folder.
+    shutil.rmtree(Path(get_settings().blob_root) / str(user_id), ignore_errors=True)

backend/app/models.py

@@ -69,6 +69,19 @@ class AppSettings(Base):
     default_quota_bytes: Mapped[int] = mapped_column(
         BigInteger, nullable=False, server_default=text("5368709120")
     )
+    # Outgoing system mail (password-reset links): the SMTP account the server
+    # sends from. The password is vault-encrypted (like Account.secret_ct).
+    smtp_host: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
+    smtp_port: Mapped[int] = mapped_column(Integer, nullable=False, server_default=text("587"))
+    smtp_security: Mapped[str] = mapped_column(
+        Text, nullable=False, server_default=text("'starttls'")
+    )  # starttls | ssl | none
+    smtp_username: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
+    smtp_secret_ct: Mapped[bytes | None] = mapped_column(BYTEA)
+    smtp_from: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
+    # Where users reach the app, e.g. https://mail.example.com — reset links are
+    # built from this, never from the request's Host header.
+    public_url: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("''"))
     updated_at: Mapped[dt.datetime] = _now()
 
 
@@ -91,6 +104,9 @@ class User(Base):
     # The first account created on a fresh install bootstraps as administrator;
     # the admin panel can grant it to others later.
     is_admin: Mapped[bool] = mapped_column(nullable=False, server_default=text("false"))
+    # Set while an administrator has blocked the account (no sign-in, sessions
+    # refused); NULL when not blocked.
+    blocked_at: Mapped[dt.datetime | None] = mapped_column(TIMESTAMP(timezone=True))
     created_at: Mapped[dt.datetime] = _now()
 
 

backend/app/ws.py

@@ -55,6 +55,10 @@ async def ws_endpoint(websocket: WebSocket) -> None:
             payload = data.get("payload", {})
             if str(payload.get("user_id")) == user_id:
                 await websocket.send_json(data)
+                if data.get("event") == "user.signed_out":
+                    # The session this socket was opened with no longer exists.
+                    await websocket.close(code=1008)
+                    return
 
     async def receive() -> None:
         while True:

backend/migrations/versions/0019_user_blocked.py

@@ -0,0 +1,29 @@
+"""administrator can block a user
+
+``users.blocked_at`` is set while an administrator has blocked the account: it
+cannot sign in and its sessions stop working. NULL means not blocked. A
+timestamp rather than a flag so the admin panel can say since when.
+
+Revision ID: 0019
+Revises: 0018
+Create Date: 2026-09-30
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0019"
+down_revision: str | None = "0018"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.execute("ALTER TABLE users ADD COLUMN blocked_at timestamptz")
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE users DROP COLUMN IF EXISTS blocked_at")

backend/migrations/versions/0020_system_mail.py

@@ -0,0 +1,48 @@
+"""outgoing system mail (password reset) configured in the admin panel
+
+The server sends its own mail — today only password-reset links — through an
+SMTP account the administrator sets up in the admin panel. The password is
+envelope-encrypted with MASTER_KEY like every mail-account secret. ``public_url``
+is where users reach the app; reset links are built from it, never from the
+request's Host header, which an attacker controls.
+
+Revision ID: 0020
+Revises: 0019
+Create Date: 2026-09-30
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+revision: str = "0020"
+down_revision: str | None = "0019"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_COLUMNS = (
+    ("smtp_host", "text NOT NULL DEFAULT ''"),
+    ("smtp_port", "integer NOT NULL DEFAULT 587"),
+    ("smtp_security", "text NOT NULL DEFAULT 'starttls'"),
+    ("smtp_username", "text NOT NULL DEFAULT ''"),
+    ("smtp_secret_ct", "bytea"),
+    ("smtp_from", "text NOT NULL DEFAULT ''"),
+    ("public_url", "text NOT NULL DEFAULT ''"),
+)
+
+
+def upgrade() -> None:
+    for name, ddl in _COLUMNS:
+        op.execute(f"ALTER TABLE app_settings ADD COLUMN {name} {ddl}")
+    op.execute(
+        "ALTER TABLE app_settings ADD CONSTRAINT ck_app_settings_smtp_security "
+        "CHECK (smtp_security IN ('starttls', 'ssl', 'none'))"
+    )
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE app_settings DROP CONSTRAINT IF EXISTS ck_app_settings_smtp_security")
+    for name, _ in reversed(_COLUMNS):
+        op.execute(f"ALTER TABLE app_settings DROP COLUMN IF EXISTS {name}")

backend/tests/api/test_admin_users.py

@@ -0,0 +1,196 @@
+"""Admin user management (SPEC §6.1): search, quota, block, delete.
+
+Runs against the dev database with real sessions, so every user it creates is
+removed again on the way out.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+
+import httpx
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport
+from sqlalchemy import delete, select, text
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.config import get_settings
+from app.identity import user_admin
+from app.identity.passwords import hash_password
+from app.identity.sessions import create_session, resolve_session
+from app.main import create_app
+from app.models import Account, PluginKv, User
+from app.plugins.db import plugin_session
+
+CSRF = "csrf-token"
+PASSWORD = "hunter2pass"
+
+
+def _handle() -> str:
+    return f"adm{uuid.uuid4().hex[:8]}@test.local"
+
+
+class World:
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+        self.created: list[uuid.UUID] = []
+
+    async def user(self, *, admin: bool = False) -> User:
+        user = User(handle=_handle(), password_hash=hash_password(PASSWORD), is_admin=admin)
+        self.session.add(user)
+        await self.session.commit()
+        self.created.append(user.id)
+        return user
+
+    async def client(self, user: User) -> httpx.AsyncClient:
+        token = await create_session(user.id, "pytest")
+        return httpx.AsyncClient(
+            transport=ASGITransport(app=create_app()),
+            base_url="http://testserver",
+            cookies={"session": token, "csrf": CSRF},
+            headers={"X-CSRF": CSRF},
+        )
+
+
+@pytest_asyncio.fixture
+async def world() -> AsyncIterator[World]:
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    session = async_sessionmaker(engine, expire_on_commit=False)()
+    w = World(session)
+    try:
+        yield w
+    finally:
+        await session.rollback()
+        await session.execute(delete(User).where(User.id.in_(w.created)))
+        await session.commit()
+        await session.close()
+        await engine.dispose()
+
+
+def _account(user_id: uuid.UUID, email: str, name: str = "") -> Account:
+    return Account(
+        user_id=user_id,
+        email=email,
+        display_name=name,
+        imap_config={},
+        smtp_config={},
+        secret_ct=b"x",
+        discovery_src="manual",
+        state="live",
+    )
+
+
+async def test_only_an_admin_can_manage_users(world: World) -> None:
+    plain = await world.user()
+    async with await world.client(plain) as c:
+        assert (await c.get("/api/admin/users")).status_code == 403
+
+
+async def test_search_by_login_and_by_mail_account(world: World) -> None:
+    admin = await world.user(admin=True)
+    target = await world.user()
+    tag = uuid.uuid4().hex[:10]
+    world.session.add(_account(target.id, f"x{tag}@example.com", f"Zed {tag}"))
+    await world.session.commit()
+    async with await world.client(admin) as c:
+        by_login = (await c.get("/api/admin/users", params={"q": target.handle[:11]})).json()
+        assert [u["id"] for u in by_login] == [str(target.id)]
+        by_email = (await c.get("/api/admin/users", params={"q": f"x{tag}@"})).json()
+        assert [u["id"] for u in by_email] == [str(target.id)]
+        assert by_email[0]["accounts"] == [f"x{tag}@example.com"]
+        by_name = (await c.get("/api/admin/users", params={"q": f"zed {tag}"})).json()
+        assert [u["id"] for u in by_name] == [str(target.id)]
+        # A typed % is a character, not "match anything".
+        assert (await c.get("/api/admin/users", params={"q": f"{tag}%"})).json() == []
+
+
+async def test_change_quota(world: World) -> None:
+    admin = await world.user(admin=True)
+    target = await world.user()
+    async with await world.client(admin) as c:
+        resp = await c.patch(f"/api/admin/users/{target.id}", json={"quota_bytes": 123456789})
+        assert resp.status_code == 200
+        assert resp.json()["quota_bytes"] == 123456789
+        bad = await c.patch(f"/api/admin/users/{target.id}", json={"quota_bytes": 0})
+        assert bad.status_code == 422
+
+
+class _Bus:
+    def __init__(self) -> None:
+        self.published: list[tuple[str, dict[str, object]]] = []
+
+    async def publish(self, event: str, payload: dict[str, object]) -> None:
+        self.published.append((event, payload))
+
+
+@pytest.fixture
+def bus(monkeypatch: pytest.MonkeyPatch) -> _Bus:
+    fake = _Bus()
+    monkeypatch.setattr(user_admin, "get_event_bus", lambda: fake)
+    return fake
+
+
+async def test_block_refuses_login_and_ends_sessions_then_unblock(world: World, bus: _Bus) -> None:
+    admin = await world.user(admin=True)
+    target = await world.user()
+    token = await create_session(target.id, "pytest")
+    async with await world.client(admin) as c:
+        resp = await c.patch(f"/api/admin/users/{target.id}", json={"blocked": True})
+        assert resp.status_code == 200 and resp.json()["blocked_at"]
+    # The session the user held is gone, and their open tabs are told to leave.
+    assert await resolve_session(token) is None
+    assert ("user.signed_out", {"user_id": str(target.id)}) in bus.published
+    anon = httpx.AsyncClient(transport=ASGITransport(app=create_app()), base_url="http://t")
+    async with anon:
+        # ...and they cannot sign back in, told why only after a correct password.
+        login = {"handle": target.handle, "password": PASSWORD}
+        blocked = await anon.post("/api/auth/login", json=login)
+        assert blocked.status_code == 403
+        assert blocked.json()["error"]["code"] == "account_blocked"
+        wrong = await anon.post("/api/auth/login", json={**login, "password": "nope-nope"})
+        assert wrong.json()["error"]["code"] == "invalid_credentials"
+
+        async with await world.client(admin) as c:
+            resp = await c.patch(f"/api/admin/users/{target.id}", json={"blocked": False})
+            assert resp.json()["blocked_at"] is None
+        assert (await anon.post("/api/auth/login", json=login)).status_code == 200
+
+
+async def test_admin_cannot_block_or_delete_themselves(world: World) -> None:
+    admin = await world.user(admin=True)
+    async with await world.client(admin) as c:
+        assert (
+            await c.patch(f"/api/admin/users/{admin.id}", json={"blocked": True})
+        ).status_code == 409
+        assert (await c.delete(f"/api/admin/users/{admin.id}")).status_code == 409
+
+
+async def test_delete_removes_the_user_and_their_data(world: World, bus: _Bus) -> None:
+    admin = await world.user(admin=True)
+    target = await world.user()
+    world.session.add(_account(target.id, f"d{uuid.uuid4().hex[:8]}@example.com"))
+    world.session.add(PluginKv(plugin="labels", user_id=target.id, key="k", value={"a": 1}))
+    await world.session.commit()
+    # A plugin row: no foreign key to users, behind row-level security.
+    async with plugin_session(target.id) as ps:
+        await ps.execute(
+            text("INSERT INTO plg_contacts_books (user_id, name) VALUES (:u, 'Book')"),
+            {"u": target.id},
+        )
+    token = await create_session(target.id, "pytest")
+
+    async with await world.client(admin) as c:
+        assert (await c.delete(f"/api/admin/users/{target.id}")).status_code == 200
+
+    s = world.session
+    assert await s.scalar(select(User).where(User.id == target.id)) is None
+    assert await s.scalar(select(Account).where(Account.user_id == target.id)) is None
+    assert await s.scalar(select(PluginKv).where(PluginKv.user_id == target.id)) is None
+    async with plugin_session(target.id) as ps:
+        left = await ps.scalar(text("SELECT count(*) FROM plg_contacts_books"))
+    assert left == 0
+    assert await resolve_session(token) is None
+    assert ("user.signed_out", {"user_id": str(target.id)}) in bus.published

backend/tests/identity/test_identity_m1.py

@@ -1,4 +1,5 @@
-"""M1 identity: 2FA (TOTP), session list/revoke, password reset (SPEC §6.1/§9.1)."""
+"""M1 identity: 2FA (TOTP), session list/revoke (SPEC §6.1/§9.1). The password reset
+flow is in test_password_reset.py."""
 
 from __future__ import annotations
 
@@ -19,7 +20,6 @@ from app.identity.passwords import hash_password
 from app.identity.sessions import create_session
 from app.main import create_app
 from app.models import User
-from app.redis_client import get_redis
 
 CSRF = "csrf-token"
 PASSWORD = "hunter2pass"
@@ -145,43 +145,6 @@ async def test_sessions_list_and_revoke(ctx: Ctx) -> None:
         assert me.status_code == 401
 
 
-async def test_password_reset_flow(ctx: Ctx) -> None:
-    # request a reset; token is stored in redis (not returned)
-    req = await ctx.client.post("/api/auth/password/reset-request", json={"handle": ctx.handle})
-    assert req.status_code == 200
-
-    redis = get_redis()
-    keys = await redis.keys("pwreset:*")
-    # Pick the token bound to THIS user (redis DB may hold others from prior runs).
-    token = None
-    for key in keys:
-        if await redis.get(key) == str(ctx.user_id):
-            token = key.split(":", 1)[1]
-            break
-    assert token is not None, "a reset token should have been issued for this user"
-
-    new_password = "brandNewPass9"
-    reset = await ctx.client.post(
-        "/api/auth/password/reset", json={"token": token, "password": new_password}
-    )
-    assert reset.status_code == 200
-
-    # token is single-use
-    again = await ctx.client.post(
-        "/api/auth/password/reset", json={"token": token, "password": new_password}
-    )
-    assert again.status_code == 400
-
-    # new password works, old one does not
-    async with _fresh_client(ctx.app) as anon:
-        good = await anon.post(
-            "/api/auth/login", json={"handle": ctx.handle, "password": new_password}
-        )
-        assert good.status_code == 200
-        bad = await anon.post("/api/auth/login", json={"handle": ctx.handle, "password": PASSWORD})
-        assert bad.status_code == 401
-
-
 async def test_reset_request_unknown_handle_still_ok(ctx: Ctx) -> None:
     # must not leak whether a handle exists
     resp = await ctx.client.post("/api/auth/password/reset-request", json={"handle": "nobody-here"})

backend/tests/identity/test_password_reset.py

@@ -0,0 +1,309 @@
+"""Password reset by email, recovery address, and the system mailer (SPEC §6.1).
+
+The SMTP send is replaced with a recorder, so nothing leaves the machine; the
+reset link it would have sent is read back from what was recorded.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+
+import httpx
+import pyotp
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport
+from sqlalchemy import delete, update
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
+
+from app.api import admin as admin_api
+from app.config import get_settings
+from app.identity import routes
+from app.identity.mailer import MailerConfig
+from app.identity.passwords import hash_password
+from app.identity.reset import MAX_TOTP_FAILURES
+from app.identity.service import app_settings
+from app.identity.sessions import create_session
+from app.identity.totp import encrypt_totp_secret
+from app.main import create_app
+from app.models import AppSettings, User
+from app.redis_client import get_redis
+
+CSRF = "csrf-token"
+PASSWORD = "hunter2pass"
+NEW_PASSWORD = "brandNewPass9"
+CFG = MailerConfig(
+    host="smtp.example.com",
+    port=587,
+    security="starttls",
+    username="postmaster@example.com",
+    password="pw",
+    sender="postmaster@example.com",
+    public_url="https://mail.example.com",
+)
+
+
+@dataclass
+class Sent:
+    mails: list[tuple[str, str, str]] = field(default_factory=list)  # (to, subject, text)
+
+    def token(self) -> str:
+        text = self.mails[-1][2]
+        link = next(w for w in text.split() if w.startswith("https://mail.example.com/"))
+        assert link.startswith("https://mail.example.com/reset-password#token=")
+        return link.split("#token=", 1)[1]
+
+
+@dataclass
+class Ctx:
+    session: AsyncSession
+    anon: httpx.AsyncClient
+    sent: Sent
+    created: list[uuid.UUID] = field(default_factory=list)
+
+    async def user(
+        self, *, recovery: str | None = "me@elsewhere.example", admin: bool = False
+    ) -> User:
+        user = User(
+            handle=f"pr{uuid.uuid4().hex[:8]}@test.local",
+            password_hash=hash_password(PASSWORD),
+            recovery_email=recovery,
+            is_admin=admin,
+        )
+        self.session.add(user)
+        await self.session.commit()
+        self.created.append(user.id)
+        await get_redis().delete(f"pwreset-rate:{user.id}")
+        return user
+
+    async def signed_in(self, user: User) -> httpx.AsyncClient:
+        token = await create_session(user.id, "pytest")
+        return httpx.AsyncClient(
+            transport=ASGITransport(app=create_app()),
+            base_url="http://testserver",
+            cookies={"session": token, "csrf": CSRF},
+            headers={"X-CSRF": CSRF},
+        )
+
+
+@pytest_asyncio.fixture
+async def ctx(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[Ctx]:
+    sent = Sent()
+
+    async def fake_send(cfg: MailerConfig, to: str, subject: str, text: str, html: str) -> None:
+        sent.mails.append((to, subject, text))
+
+    monkeypatch.setattr(routes, "send_mail", fake_send)
+    monkeypatch.setattr(admin_api, "send_mail", fake_send)
+    monkeypatch.setattr(routes, "mailer_config", lambda row: CFG)
+
+    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
+    session = async_sessionmaker(engine, expire_on_commit=False)()
+    anon = httpx.AsyncClient(transport=ASGITransport(app=create_app()), base_url="http://t")
+    c = Ctx(session=session, anon=anon, sent=sent)
+    try:
+        yield c
+    finally:
+        await anon.aclose()
+        await session.execute(delete(User).where(User.id.in_(c.created)))
+        await session.commit()
+        await session.close()
+        await engine.dispose()
+
+
+async def _request(ctx: Ctx, handle: str) -> httpx.Response:
+    return await ctx.anon.post("/api/auth/password/reset-request", json={"handle": handle})
+
+
+async def _login(ctx: Ctx, handle: str, password: str) -> int:
+    resp = await ctx.anon.post("/api/auth/login", json={"handle": handle, "password": password})
+    return resp.status_code
+
+
+# --- requesting a link -----------------------------------------------------------
+
+
+async def test_request_mails_a_one_time_link_to_the_recovery_address(ctx: Ctx) -> None:
+    user = await ctx.user()
+    assert (await _request(ctx, user.handle)).status_code == 200
+    assert [m[0] for m in ctx.sent.mails] == ["me@elsewhere.example"]
+    token = ctx.sent.token()
+
+    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
+    assert check.json() == {"valid": True, "totp_required": False}
+
+    reset = await ctx.anon.post(
+        "/api/auth/password/reset", json={"token": token, "password": NEW_PASSWORD}
+    )
+    assert reset.status_code == 200
+    # Used up: a second go fails, and the page is told so up front.
+    again = await ctx.anon.post(
+        "/api/auth/password/reset", json={"token": token, "password": "anotherPass12"}
+    )
+    assert again.json()["error"]["code"] == "invalid_token"
+    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
+    assert check.json()["valid"] is False
+
+    assert await _login(ctx, user.handle, NEW_PASSWORD) == 200
+    assert await _login(ctx, user.handle, PASSWORD) == 401
+
+
+async def test_the_answer_never_says_whether_the_account_exists(ctx: Ctx) -> None:
+    no_recovery = await ctx.user(recovery=None)
+    for handle in ("nobody-here-at-all", no_recovery.handle):
+        resp = await _request(ctx, handle)
+        assert resp.status_code == 200 and resp.json() == {"status": "ok"}
+    assert ctx.sent.mails == []
+
+
+async def test_no_mail_while_system_mail_is_not_set_up(
+    ctx: Ctx, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(routes, "mailer_config", lambda row: None)
+    user = await ctx.user()
+    assert (await _request(ctx, user.handle)).status_code == 200
+    assert ctx.sent.mails == []
+
+
+async def test_requests_are_capped_per_hour(ctx: Ctx) -> None:
+    user = await ctx.user()
+    for _ in range(5):
+        assert (await _request(ctx, user.handle)).status_code == 200
+    assert len(ctx.sent.mails) == 3
+
+
+# --- two-step verification ---------------------------------------------------------
+
+
+async def test_2fa_code_is_required_before_the_password_changes(ctx: Ctx) -> None:
+    user = await ctx.user()
+    secret = pyotp.random_base32()
+    await ctx.session.execute(
+        update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
+    )
+    await ctx.session.commit()
+    await _request(ctx, user.handle)
+    token = ctx.sent.token()
+    body = {"token": token, "password": NEW_PASSWORD}
+
+    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
+    assert check.json()["totp_required"] is True
+    missing = await ctx.anon.post("/api/auth/password/reset", json=body)
+    assert missing.json()["error"]["code"] == "totp_required"
+    wrong = await ctx.anon.post("/api/auth/password/reset", json={**body, "totp": "000000"})
+    assert wrong.json()["error"]["code"] == "totp_invalid"
+    # Nothing changed yet, and the link still works.
+    assert await _login(ctx, user.handle, PASSWORD) == 401  # 2FA account: needs a code
+    good = await ctx.anon.post(
+        "/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
+    )
+    assert good.status_code == 200
+    login = await ctx.anon.post(
+        "/api/auth/login",
+        json={"handle": user.handle, "password": NEW_PASSWORD, "totp": pyotp.TOTP(secret).now()},
+    )
+    assert login.status_code == 200
+
+
+async def test_too_many_wrong_codes_burn_the_link(ctx: Ctx) -> None:
+    user = await ctx.user()
+    secret = pyotp.random_base32()
+    await ctx.session.execute(
+        update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
+    )
+    await ctx.session.commit()
+    await _request(ctx, user.handle)
+    token = ctx.sent.token()
+    body = {"token": token, "password": NEW_PASSWORD, "totp": "000000"}
+    for _ in range(MAX_TOTP_FAILURES):
+        await ctx.anon.post("/api/auth/password/reset", json=body)
+    late = await ctx.anon.post(
+        "/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
+    )
+    assert late.json()["error"]["code"] == "invalid_token"
+
+
+# --- the recovery address ----------------------------------------------------------
+
+
+async def test_changing_the_recovery_email_takes_the_password(ctx: Ctx) -> None:
+    user = await ctx.user(recovery=None)
+    async with await ctx.signed_in(user) as c:
+        wrong = await c.put(
+            "/api/auth/recovery-email",
+            json={"password": "not-it", "recovery_email": "new@example.com"},
+        )
+        assert wrong.status_code == 403
+        bad = await c.put(
+            "/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": "nope"}
+        )
+        assert bad.json()["error"]["code"] == "invalid_email"
+        ok = await c.put(
+            "/api/auth/recovery-email",
+            json={"password": PASSWORD, "recovery_email": " New@Example.com "},
+        )
+        assert ok.json()["recovery_email"] == "new@example.com"
+        removed = await c.put(
+            "/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": ""}
+        )
+        assert removed.json()["recovery_email"] is None
+
+
+# --- admin: system mail settings -----------------------------------------------------
+
+
+@pytest_asyncio.fixture
+async def restore_settings(ctx: Ctx) -> AsyncIterator[None]:
+    row = await app_settings(ctx.session)
+    saved = {
+        k: getattr(row, k)
+        for k in (
+            "smtp_host",
+            "smtp_port",
+            "smtp_security",
+            "smtp_username",
+            "smtp_secret_ct",
+            "smtp_from",
+            "public_url",
+        )
+    }
+    await ctx.session.commit()
+    yield
+    await ctx.session.execute(update(AppSettings).values(**saved))
+    await ctx.session.commit()
+
+
+async def test_admin_configures_system_mail_without_ever_seeing_the_password(
+    ctx: Ctx, restore_settings: None
+) -> None:
+    admin = await ctx.user(admin=True)
+    async with await ctx.signed_in(admin) as c:
+        resp = await c.put(
+            "/api/admin/mail",
+            json={
+                "smtp_host": "smtp.example.com",
+                "smtp_port": 465,
+                "smtp_security": "ssl",
+                "smtp_username": "postmaster@example.com",
+                "smtp_password": "s3cret",
+                "smtp_from": "postmaster@example.com",
+                "public_url": "https://mail.example.com/",
+            },
+        )
+        out = resp.json()
+        assert out["has_password"] is True and out["configured"] is True
+        assert out["public_url"] == "https://mail.example.com"
+        assert "s3cret" not in resp.text
+        # Omitting the password keeps it; an empty one removes it.
+        kept = (await c.put("/api/admin/mail", json={"smtp_port": 587})).json()
+        assert kept["has_password"] is True
+        gone = (await c.put("/api/admin/mail", json={"smtp_password": ""})).json()
+        assert gone["has_password"] is False
+        bad = await c.put("/api/admin/mail", json={"public_url": "mail.example.com"})
+        assert bad.json()["error"]["code"] == "invalid_url"
+        test = await c.post("/api/admin/mail/test", json={"to": "me@example.com"})
+        assert test.status_code == 200
+    assert ctx.sent.mails[-1][:2] == ("me@example.com", "Ultimo Email test message")