feat(users): invite by email — a one-time link where the invitee makes their own account (milestone 496)
CI / lint (push) Successful in 3s
CI / integration (push) Failing after 1m17s
CI / unit (push) Successful in 1m43s
CI / publish (push) Skipped

The operator's ask: an invite link sent by mail, the key in the URL, and
the new user choosing their own username, password and email.

- user_invites (migration 0037): the key's sha256 only, the address, the
  role and days the admin chose, who sent it, used/revoked.
- core/invites: create (revokes an older open invite to the address),
  renew (new key, new clock), find_pending, the four states, signup checks.
- /invite/<key>, outside sign-in and throttled like login: the invitee's
  username, email (prefilled, theirs to change) and password; the account
  takes the admin's role, the key is used up in the same transaction and
  they are signed in. A bad key gets one plain message.
- Settings › Users: Invite by email (address, role, 1/7/30 days), sent from
  the admin in their zone; without a mail server, or if sending fails, the
  link is shown once to copy. Invites not yet used are listed with Resend
  and Withdraw.
- Audit: user.invited, user.invite_resent, user.invite_revoked, user.joined.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-08 16:54:44 -04:00
co-authored by Claude Opus 5.5
parent ac38a04642
commit 2b6e12859a
18 changed files with 874 additions and 4 deletions
+111
View File
@@ -1,4 +1,6 @@
from __future__ import annotations
from datetime import datetime, timezone
import bcrypt
from quart import Blueprint, render_template, request, redirect, url_for, session
from sqlalchemy import select, func
@@ -291,3 +293,112 @@ async def setup_post():
await log_audit(current_app, user.id, user.username, "auth.setup",
detail={"username": username})
return redirect(url_for("dashboard.index"))
# ── Invites (milestone 496) ──────────────────────────────────────────────────
_INVITE_GONE = ("This invite link has expired, been used or been withdrawn. "
"Ask whoever sent it for a new one.")
async def _invite_page(key: str, invite=None, inviter: str = "", error: str | None = None,
form=None, status: int = 200):
"""The invite page: the form while `invite` works, else one plain message
that doesn't say whether the key was unknown, used, revoked or expired."""
if invite is None and error is None:
error = _INVITE_GONE
return await render_template(
"auth/invite.html", key=key, invite=invite, inviter=inviter, error=error,
form=form or {}, min_password=_min_password()), status
def _min_password() -> int:
from steward.settings.users import MIN_PASSWORD
return MIN_PASSWORD
async def _inviter_name(db, invite) -> str:
if invite.invited_by is None:
return "An admin"
who = await db.get(User, invite.invited_by)
return who.username if who else "An admin"
def _invite_throttled() -> str | None:
"""Why this address may not try an invite key now (too many bad ones), or None."""
from quart import current_app
wait_s = login_throttle.retry_after(
client_ip(), current_app.config.get("LOGIN_MAX_FAILURES", 5),
current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15) * 60)
if not wait_s:
return None
minutes = max(1, round(wait_s / 60))
return f"Too many tries from this address. Try again in {minutes} minute{'s' if minutes != 1 else ''}."
@auth_bp.get("/invite/<key>")
@public_route("an invite link: the key is the credential")
async def invite(key: str):
from quart import current_app
from steward.core.invites import find_pending
if wait := _invite_throttled():
return await _invite_page(key, error=wait, status=429)
async with current_app.db_sessionmaker() as db:
found = await find_pending(db, key)
inviter = await _inviter_name(db, found) if found else ""
if found is None:
login_throttle.failed(client_ip())
return await _invite_page(key, status=404)
return await _invite_page(key, found, inviter, form={"email": found.email})
@auth_bp.post("/invite/<key>")
@public_route("an invite link: the key is the credential")
async def invite_accept(key: str):
"""Make the invitee's local account from their own username, email and
password, with the role the admin chose; use the key up and sign them in,
in one transaction, so a key makes one account."""
from quart import current_app
from sqlalchemy.exc import IntegrityError
from steward.core.audit import log_audit
from steward.core.invites import find_pending, signup_problem, taken
from steward.settings.users import _hash
if wait := _invite_throttled():
return await _invite_page(key, error=wait, status=429)
form = await request.form
username = form.get("username") or ""
email = (form.get("email") or "").strip()
password, confirm = form.get("password") or "", form.get("confirm") or ""
shown = {"username": username, "email": email}
try:
async with current_app.db_sessionmaker() as db:
async with db.begin():
found = await find_pending(db, key, lock=True)
if found is None:
login_throttle.failed(client_ip())
return await _invite_page(key, status=404)
inviter = await _inviter_name(db, found)
problem = signup_problem(username, email, password, confirm) or await taken(db, username, email)
if problem:
return await _invite_page(key, found, inviter, problem, shown, 400)
user = User(username=username, email=email, role=found.role, auth_source="local",
password_hash=_hash(password))
db.add(user)
await db.flush()
found.used_at = datetime.now(timezone.utc)
found.used_by = user.id
invite_id, role = found.id, found.role.value
db.expunge(user)
except IntegrityError:
async with current_app.db_sessionmaker() as db:
found = await find_pending(db, key)
inviter = await _inviter_name(db, found) if found else ""
if found is None:
return await _invite_page(key, status=404)
return await _invite_page(key, found, inviter,
"That username or email was just taken. Pick another.", shown, 400)
login_user(user)
await log_audit(current_app, user.id, user.username, "user.joined", "user", user.id,
detail={"username": user.username, "role": role, "invite": invite_id,
"invited_by": inviter})
return redirect(url_for("dashboard.index"))