From 593b583d231366b9e0a24d3b4f273b3320922bb9 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 8 Oct 2026 20:17:24 -0400 Subject: [PATCH] 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/: 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 --- steward/app.py | 2 + steward/auth/routes.py | 139 +++++++++++++++++- steward/auth/throttle.py | 4 + steward/core/mail.py | 9 ++ steward/core/password_resets.py | 76 ++++++++++ .../versions/0039_password_resets.py | 33 +++++ steward/models/__init__.py | 1 + steward/models/password_resets.py | 21 +++ steward/static/steward.css | 4 +- steward/templates/auth/forgot.html | 30 ++++ steward/templates/auth/login.html | 3 + steward/templates/auth/reset.html | 32 ++++ steward/templates/email/password_reset.html | 19 +++ steward/templates/email/password_reset.txt | 13 ++ tests/core/test_password_resets.py | 37 +++++ tests/integration/test_passwords.py | 110 +++++++++++++- tests/test_breadcrumbs.py | 2 + 17 files changed, 528 insertions(+), 7 deletions(-) create mode 100644 steward/core/password_resets.py create mode 100644 steward/migrations/versions/0039_password_resets.py create mode 100644 steward/models/password_resets.py create mode 100644 steward/templates/auth/forgot.html create mode 100644 steward/templates/auth/reset.html create mode 100644 steward/templates/email/password_reset.html create mode 100644 steward/templates/email/password_reset.txt create mode 100644 tests/core/test_password_resets.py diff --git a/steward/app.py b/steward/app.py index 93edcfd..9732f02 100644 --- a/steward/app.py +++ b/steward/app.py @@ -166,6 +166,8 @@ def create_app( from .auth import csrf as _csrf app.jinja_env.globals["csrf_token"] = _csrf.csrf_token app.jinja_env.globals["csrf_field"] = _csrf.csrf_field + # The sign-in page offers "Forgot your password?" only while mail can send (milestone 497). + app.jinja_env.globals["reset_by_email"] = lambda: bool(app.config.get("SMTP", {}).get("host")) app.before_request(_csrf.check_request) # The session cookie is Secure, and HSTS is sent, only on requests that # arrived over HTTPS — directly or as a trusted proxy reports it diff --git a/steward/auth/routes.py b/steward/auth/routes.py index 44ae5ed..8f2fd4a 100644 --- a/steward/auth/routes.py +++ b/steward/auth/routes.py @@ -324,8 +324,9 @@ async def _inviter_name(db, invite) -> str: 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.""" +def _too_many_tries() -> str | None: + """Why this address may not try an invite or reset 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), @@ -341,7 +342,7 @@ def _invite_throttled() -> str | None: async def invite(key: str): from quart import current_app from steward.core.invites import find_pending - if wait := _invite_throttled(): + if wait := _too_many_tries(): return await _invite_page(key, error=wait, status=429) async with current_app.db_sessionmaker() as db: found = await find_pending(db, key) @@ -363,7 +364,7 @@ async def invite_accept(key: str): 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(): + if wait := _too_many_tries(): return await _invite_page(key, error=wait, status=429) form = await request.form username = form.get("username") or "" @@ -402,3 +403,133 @@ async def invite_accept(key: str): detail={"username": user.username, "role": role, "invite": invite_id, "invited_by": inviter}) return redirect(url_for("dashboard.index")) + + +# ── Reset a password by email (milestone 497) ──────────────────────────────── + +_RESET_GONE = "This reset link has expired or been used. Ask for a new one." +_RESET_SENT = ("If that names a Steward account with a password and an email address, " + "a link to choose a new password is on its way to that address. It works for an hour.") + + +def _mail_ready() -> bool: + from quart import current_app + return bool(current_app.config.get("SMTP", {}).get("host")) + + +async def _forgot_page(sent: bool = False, error: str | None = None, status: int = 200): + return await render_template("auth/forgot.html", mail_ready=_mail_ready(), sent=sent, + error=error), status + + +@auth_bp.get("/forgot") +@public_route("asks for a password-reset email") +async def forgot(): + return await _forgot_page() + + +@auth_bp.post("/forgot") +@public_route("asks for a password-reset email") +async def forgot_post(): + """Email a reset link to the account the form names, if it may have one. + The answer is the same either way, so the page can't be used to find + accounts; every request from an address counts toward its limit.""" + from quart import current_app + from steward.auth.throttle import reset_throttle + from steward.core import mail + from steward.core.audit import log_audit + from steward.core.formatting import zone + from steward.core.password_resets import account_for, issue, link + from steward.core.settings import public_base_url + if not _mail_ready(): + return await _forgot_page(status=400) + ip = client_ip() + if reset_throttle.retry_after(ip, current_app.config.get("LOGIN_MAX_FAILURES", 5), + current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15) * 60): + return await _forgot_page(error="Too many requests from this address. Try again later.", status=429) + reset_throttle.failed(ip) + name = (await request.form).get("name") or "" + async with current_app.db_sessionmaker() as db: + async with db.begin(): + user = await account_for(db, name) + if user is not None: + reset, key = await issue(db, user) + to, username, user_id, tz = user.email, user.username, user.id, user.time_zone + if user is None: + return await _forgot_page(sent=True) + cfg = current_app.config.get("SMTP", {}) + subject, text, html = mail.reset_email(username, reset.expires_at, + link(public_base_url(request), key), zone(tz)) + try: + await mail.send(cfg, [mail.message(cfg, to, subject, text, html)]) + except Exception as exc: # the page still says the same; the log says why + current_app.logger.warning("Password reset email to %s failed: %s", to, exc) + await log_audit(current_app, user_id, username, "user.password_reset_requested", "user", user_id, + detail={"address": ip}) + return await _forgot_page(sent=True) + + +async def _reset_page(key: str, username: str | None = None, error: str | None = None, + status: int = 200): + """The new-password form while the link works, else one plain message.""" + if username is None and error is None: + error = _RESET_GONE + return await render_template("auth/reset.html", key=key, username=username, error=error, + min_password=_min_password()), status + + +async def _reset_target(db, key: str, lock: bool = False): + """(link, account) for a working key whose account may still be reset by + email; (None, None) otherwise.""" + from steward.core.password_resets import find_pending + found = await find_pending(db, key, lock=lock) + user = await db.get(User, found.user_id) if found else None + if user is None or not user.is_active or user.auth_source != "local": + return None, None + return found, user + + +@auth_bp.get("/reset/") +@public_route("a password-reset link: the key is the credential") +async def reset(key: str): + from quart import current_app + if wait := _too_many_tries(): + return await _reset_page(key, error=wait, status=429) + async with current_app.db_sessionmaker() as db: + _, user = await _reset_target(db, key) + if user is None: + login_throttle.failed(client_ip()) + return await _reset_page(key, status=404) + return await _reset_page(key, user.username) + + +@auth_bp.post("/reset/") +@public_route("a password-reset link: the key is the credential") +async def reset_post(key: str): + """Set the new password, use the link up, sign out every browser signed + in as the account and sign this one in, in one transaction.""" + from quart import current_app + from steward.core.audit import log_audit + from steward.settings.users import _hash, password_problem + if wait := _too_many_tries(): + return await _reset_page(key, error=wait, status=429) + form = await request.form + async with current_app.db_sessionmaker() as db: + async with db.begin(): + found, user = await _reset_target(db, key, lock=True) + if user is None: + login_throttle.failed(client_ip()) + return await _reset_page(key, status=404) + problem = password_problem(form.get("password") or "", form.get("confirm") or "") + if problem: + return await _reset_page(key, user.username, problem, 400) + now = datetime.now(timezone.utc) + user.password_hash = _hash(form.get("password")) + user.sessions_since = now + found.used_at = now + db.expunge(user) + login_throttle.succeeded(client_ip()) + login_user(user) + await log_audit(current_app, user.id, user.username, "user.password_reset_by_email", "user", user.id, + detail={"username": user.username}) + return redirect(url_for("dashboard.index")) diff --git a/steward/auth/throttle.py b/steward/auth/throttle.py index f06cf73..e65aee0 100644 --- a/steward/auth/throttle.py +++ b/steward/auth/throttle.py @@ -46,3 +46,7 @@ class LoginThrottle: 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() diff --git a/steward/core/mail.py b/steward/core/mail.py index 6fb5bbf..ee8169d 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -271,6 +271,15 @@ ROLE_WORDS = { } +def reset_email(username: str, expires_at: datetime, url: str, tz: tzinfo) -> tuple[str, str, str]: + """(subject, text, html) for a password-reset link (milestone 497): whose + account, when the link stops working (in the account's own zone) and the + link, with what to do if they didn't ask.""" + facts = {"username": username, "link": url, "expires": moment(expires_at, tz)} + text, html = _render("password_reset", r=facts) + return "[Steward] Reset your password", text, html + + def invite_email(inviter: str, role: str, expires_at: datetime, url: str, tz: tzinfo) -> tuple[str, str, str]: """(subject, text, html) for an invite (milestone 496): who sent it, the diff --git a/steward/core/password_resets.py b/steward/core/password_resets.py new file mode 100644 index 0000000..72335b7 --- /dev/null +++ b/steward/core/password_resets.py @@ -0,0 +1,76 @@ +"""Reset a local password from an emailed link (milestone 497). + +Anyone may ask from the sign-in page, by username or email, and the page +answers the same whether or not an account matched. Only an active local +account with an address mail can reach gets a link. The link carries a +one-time key (only its sha256 is kept) and works for an hour; asking again +replaces it. Using it sets the password, signs out every browser signed in +as the account (users.sessions_since) and signs the person in. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from sqlalchemy import delete, func, or_, select + +from steward.core.api_tokens import hash_token +from steward.core.invites import new_key +from steward.models.password_resets import PasswordReset +from steward.models.users import User + +LIFETIME = timedelta(hours=1) + + +def state(reset: PasswordReset, now: datetime | None = None) -> str: + """"used", "expired" or "pending".""" + now = now or datetime.now(timezone.utc) + if reset.used_at is not None: + return "used" + if reset.expires_at <= now: + return "expired" + return "pending" + + +def link(base: str, key: str) -> str: + return f"{base.rstrip('/')}/reset/{key}" + + +async def account_for(db, name: str) -> User | None: + """The account a "username or email" names, when it may be reset by + email: active, local, and with an address mail can reach.""" + from steward.core.mail import is_real_address + name = (name or "").strip() + if not name or len(name) > 255: + return None + user = (await db.execute(select(User).where( + or_(User.username == name, func.lower(User.email) == name.lower())) + .order_by((User.username == name).desc()).limit(1))).scalar_one_or_none() + if user is None or not user.is_active or user.auth_source != "local" or not is_real_address(user.email): + return None + return user + + +async def issue(db, user: User) -> tuple[PasswordReset, str]: + """A new link for `user`, flushed, and its raw key (sent once, never + stored). The account's earlier unused links stop working. Call in a + transaction.""" + await db.execute(delete(PasswordReset).where( + PasswordReset.user_id == user.id, PasswordReset.used_at.is_(None))) + now = datetime.now(timezone.utc) + key = new_key() + reset = PasswordReset(user_id=user.id, key_hash=hash_token(key), created_at=now, + expires_at=now + LIFETIME) + db.add(reset) + await db.flush() + return reset, key + + +async def find_pending(db, key: str, lock: bool = False) -> PasswordReset | None: + """The link `key` opens, while it still works; None for any other key.""" + if not key or len(key) > 128: + return None + query = select(PasswordReset).where(PasswordReset.key_hash == hash_token(key)) + if lock: + query = query.with_for_update() + reset = (await db.execute(query)).scalar_one_or_none() + return reset if reset is not None and state(reset) == "pending" else None diff --git a/steward/migrations/versions/0039_password_resets.py b/steward/migrations/versions/0039_password_resets.py new file mode 100644 index 0000000..ce7b00f --- /dev/null +++ b/steward/migrations/versions/0039_password_resets.py @@ -0,0 +1,33 @@ +"""Reset a local password from an emailed link (milestone 497): a one-time +key, stored as its sha256, that works for an hour. + +Revision ID: 0039_password_resets +Revises: 0038_user_sessions_since +Create Date: 2026-10-09 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0039_password_resets" +down_revision: Union[str, None] = "0038_user_sessions_since" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "password_resets", + sa.Column("id", sa.Integer, primary_key=True, autoincrement=True), + sa.Column("user_id", sa.String(36), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False), + sa.Column("key_hash", sa.String(64), nullable=False, unique=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("used_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_password_resets_user_id", "password_resets", ["user_id"]) + + +def downgrade() -> None: + op.drop_index("ix_password_resets_user_id", table_name="password_resets") + op.drop_table("password_resets") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index 65591b6..d282c86 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -2,6 +2,7 @@ from .base import Base from .users import User from .api_tokens import ApiToken from .invites import UserInvite +from .password_resets import PasswordReset from .hosts import Host from .monitors import Monitor, MonitorResult, MonitorType from .metrics import PluginMetric diff --git a/steward/models/password_resets.py b/steward/models/password_resets.py new file mode 100644 index 0000000..44dfe3c --- /dev/null +++ b/steward/models/password_resets.py @@ -0,0 +1,21 @@ +from __future__ import annotations +from datetime import datetime, timezone +from sqlalchemy import DateTime, ForeignKey, Integer, String +from sqlalchemy.orm import Mapped, mapped_column +from .base import Base + + +class PasswordReset(Base): + """An emailed link to set a new password for a local account (milestone + 497). The link carries a one-time key; only its sha256 is stored, as for + invites. It works for an hour, once; asking again replaces it.""" + __tablename__ = "password_resets" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + user_id: Mapped[str] = mapped_column( + String(36), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + key_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/steward/static/steward.css b/steward/static/steward.css index 4040fa4..7bb1fba 100644 --- a/steward/static/steward.css +++ b/steward/static/steward.css @@ -451,11 +451,13 @@ a.stat-tile:hover { border-color: var(--border-mid); background: var(--bg-elevat .form-section > .section-title { margin-bottom: 0.9rem; } .form-pair { display: grid; grid-template-columns: repeat(auto-fit, minmax(min(14rem, 100%), 1fr)); gap: 0 1rem; } .form-pair[hidden] { display: none; } -/* The invite page (milestone 496): the login page's centred card, with labelled fields. */ +/* The invite, forgot-password and reset pages (milestones 496, 497): the login + page's centred card, with labelled fields. */ .invite-page { max-width: 420px; margin: 4rem auto; } .invite-title { font-family: var(--font-serif); font-size: 1.5rem; font-weight: 500; margin: 0 0 0.5rem; } .invite-lede { color: var(--text-muted); margin: 0 0 1.25rem; } .invite-submit { width: 100%; margin-top: 0.25rem; } +.login-forgot { text-align: center; margin: 1rem 0 0; font-size: 0.85rem; } /* A link shown once to copy (an invite without mail): the value and its Copy button. */ .copy-once { display: flex; gap: 0.5rem; align-items: center; } .copy-once .input { flex: 1; font-family: ui-monospace, monospace; font-size: 0.8rem; } diff --git a/steward/templates/auth/forgot.html b/steward/templates/auth/forgot.html new file mode 100644 index 0000000..2460a2c --- /dev/null +++ b/steward/templates/auth/forgot.html @@ -0,0 +1,30 @@ +{# auth/forgot.html — ask for a password-reset email (milestone 497). It answers + the same whether or not an account matched. #} +{% extends "base.html" %} +{% block title %}Forgot your password — Steward{% endblock %} +{% block content %} +
+
+

Forgot your password?

+ {% if error %}{% endif %} + {% if sent %} +
If that names a Steward account with a password and an email address, a link to choose a new password is on its way to that address. It works for an hour.
+ Back to sign in + {% elif not mail_ready %} +

Steward has no mail server set, so it can't email you a link. Ask an admin to set a new password for you in Settings › Users.

+ Back to sign in + {% else %} +

Enter your username or email address. If your account has a Steward password, we'll email you a link to choose a new one.

+
{{ csrf_field() }} +
+ + + Accounts that sign in with LDAP or single sign-on change their password there. +
+ +
+

Back to sign in

+ {% endif %} +
+
+{% endblock %} diff --git a/steward/templates/auth/login.html b/steward/templates/auth/login.html index f33282f..098589c 100644 --- a/steward/templates/auth/login.html +++ b/steward/templates/auth/login.html @@ -32,6 +32,9 @@ Sign In + {% if reset_by_email() %} + + {% endif %} {% endblock %} diff --git a/steward/templates/auth/reset.html b/steward/templates/auth/reset.html new file mode 100644 index 0000000..ff4ce6b --- /dev/null +++ b/steward/templates/auth/reset.html @@ -0,0 +1,32 @@ +{# auth/reset.html — choose a new password from an emailed link (milestone 497). + `username` is set while the link works; otherwise only `error` shows. #} +{% extends "base.html" %} +{% block title %}Choose a new password — Steward{% endblock %} +{% block content %} +
+
+

Choose a new password

+ {% if username %} +

For {{ username }}. Once it's set you're signed in here, and signed out anywhere else.

+ {% endif %} + {% if error %}{% endif %} + {% if username %} +
{{ csrf_field() }} + +
+ + + At least {{ min_password }} characters. +
+
+ + +
+ +
+ {% else %} + Ask for a new link + {% endif %} +
+
+{% endblock %} diff --git a/steward/templates/email/password_reset.html b/steward/templates/email/password_reset.html new file mode 100644 index 0000000..502e38e --- /dev/null +++ b/steward/templates/email/password_reset.html @@ -0,0 +1,19 @@ +{# A password-reset email's HTML part (milestone 497): r = the facts mail.reset_email builds. #} +{% extends "_layout.html" %} +{% from "_parts.html" import button, fact with context %} +{% block title %}Reset your Steward password{% endblock %} +{% block body %} +Reset your Steward password +Someone asked to reset the password for {{ r.username }}. Open the link to choose a new one. Anywhere you're signed in now, you'll be signed out. + + +{{ fact("Account", r.username) }} +{{ fact("Link works until", r.expires) }} +
+ +{{ button(r.link, "Choose a new password") }} +Or paste this into your browser: {{ r.link }} +{% endblock %} +{% block footer %} +The link works once. If you didn't ask for this, ignore it: your password is unchanged. +{% endblock %} diff --git a/steward/templates/email/password_reset.txt b/steward/templates/email/password_reset.txt new file mode 100644 index 0000000..8d8e298 --- /dev/null +++ b/steward/templates/email/password_reset.txt @@ -0,0 +1,13 @@ +{# A password-reset email's plain-text part; the same facts as password_reset.html. #} +Someone asked to reset the Steward password for {{ r.username }}. + +Open the link to choose a new password. Anywhere you're signed in now, you'll +be signed out. + +Link works until: {{ r.expires }} + +Choose a new password: {{ r.link }} + +-- +The link works once. If you didn't ask for this, ignore it: your password is +unchanged. diff --git a/tests/core/test_password_resets.py b/tests/core/test_password_resets.py new file mode 100644 index 0000000..2ff2c3f --- /dev/null +++ b/tests/core/test_password_resets.py @@ -0,0 +1,37 @@ +"""Reset a password by email (milestone 497): a link's states, where it +points, and the email that carries it.""" +import re +from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo + +from steward.core import mail, password_resets +from steward.models.password_resets import PasswordReset + +NOW = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc) + + +def _reset(**kw): + return PasswordReset(**({"user_id": "u", "key_hash": "x", "created_at": NOW, + "expires_at": NOW + password_resets.LIFETIME} | kw)) + + +def test_a_link_works_for_an_hour_and_once(): + assert password_resets.LIFETIME == timedelta(hours=1) + assert password_resets.state(_reset(), NOW + timedelta(minutes=59)) == "pending" + assert password_resets.state(_reset(), NOW + timedelta(hours=1)) == "expired" + assert password_resets.state(_reset(used_at=NOW), NOW) == "used" + + +def test_the_link_carries_the_key_under_steward_s_address(): + assert password_resets.link("https://steward.example/", "abc") == "https://steward.example/reset/abc" + + +def test_the_reset_email_names_the_account_the_deadline_and_what_to_do_if_unasked(): + url = "https://steward.example/reset/KEY123" + subject, text, html = mail.reset_email("ann ", NOW + timedelta(hours=1), url, ZoneInfo("Europe/London")) + assert subject == "[Steward] Reset your password" + for part in (text, html): + assert url in part and "Fri 9 Oct 2026 14:00 BST" in part + assert "your password is unchanged" in part and "UTC" not in part + assert "ann <x>" in html and "" not in html + assert all(u.startswith("https://steward.example") for u in re.findall(r"""https?://[^\s"'<>)]+""", html)) diff --git a/tests/integration/test_passwords.py b/tests/integration/test_passwords.py index cc4e1ed..d85cfcd 100644 --- a/tests/integration/test_passwords.py +++ b/tests/integration/test_passwords.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio import json import os +import re import time import uuid @@ -23,10 +24,13 @@ def app(): pytest.skip("needs Postgres") from steward.app import create_app from steward.auth.throttle import login_throttle - login_throttle._failures.clear() # wrong passwords below count against this address + from steward.auth.throttle import reset_throttle + for t in (login_throttle, reset_throttle): + t._failures.clear() # wrong passwords and reset requests below count against this address app = create_app(testing=False) yield app - login_throttle._failures.clear() + for t in (login_throttle, reset_throttle): + t._failures.clear() async def _local(app, password="old password", source="local"): @@ -113,3 +117,105 @@ def test_an_sso_account_has_no_password_to_change(app): assert resp.status_code == 302 and _checks(await _user(app, uid), "old password") asyncio.run(_go()) + + +_KEY = re.compile(r"/reset/([A-Za-z0-9_-]{20,})") + + +def _mail(app, monkeypatch): + from steward.core import mail + sent = [] + + async def fake_send(cfg, messages): + sent.extend(messages) + + app.config["SMTP"] = {"host": "mx.example", "port": 587, "from": "steward@example.com"} + monkeypatch.setattr(mail, "send", fake_send) + return sent + + +async def _stranger(app): + client = app.test_client() + await client.get("/login") + async with client.session_transaction() as sess: + sess["_csrf"] = "test-token" + return client + + +def test_a_forgotten_password_is_reset_from_the_emailed_link(app, monkeypatch): + sent = _mail(app, monkeypatch) + + async def _go(): + uid, name, elsewhere = await _local(app) + guest = await _stranger(app) + page = await (await guest.get("/login")).get_data(as_text=True) + assert 'href="/forgot"' in page + + # The same answer for a stranger's name, an SSO account and a real one. + sso, sso_name, _ = await _local(app, source="oidc") + answers = [] + for who in ("nobody-" + uid[:8], sso_name, f"{name}@EXAMPLE.com"): + resp = await guest.post("/forgot", headers=CSRF, form={"name": who}) + answers.append((resp.status_code, "on its way" in await resp.get_data(as_text=True))) + assert answers == [(200, True)] * 3 + assert [m["To"] for m in sent] == [f"{name}@example.com"] + text = sent[0].get_body(("plain",)).get_content() + assert name in text and "BST" in text + key = _KEY.search(text).group(1) + + page = await (await guest.get(f"/reset/{key}")).get_data(as_text=True) + assert "Choose a new password" in page and name in page + resp = await guest.post(f"/reset/{key}", headers=CSRF, form={"password": "brand new pw", "confirm": "nope"}) + assert resp.status_code == 400 and "passwords don" in await resp.get_data(as_text=True) + resp = await guest.post(f"/reset/{key}", headers=CSRF, form={"password": "brand new pw", "confirm": "brand new pw"}) + assert resp.status_code == 302 + assert _checks(await _user(app, uid), "brand new pw") + async with guest.session_transaction() as sess: + assert sess["user_id"] == uid # signed straight in + assert (await guest.get("/account/")).status_code == 200 + assert (await elsewhere.get("/account/")).status_code == 302 # the old browser is out + assert [a for a, _ in await _audits(app, uid)] == [ + "user.password_reset_requested", "user.password_reset_by_email"] + + again = await _stranger(app) + resp = await again.get(f"/reset/{key}") + assert resp.status_code == 404 and "expired or been used" in await resp.get_data(as_text=True) + resp = await again.post(f"/reset/{key}", headers=CSRF, form={"password": "third one!", "confirm": "third one!"}) + assert resp.status_code == 404 and _checks(await _user(app, uid), "brand new pw") + + asyncio.run(_go()) + + +def test_asking_again_kills_the_first_link_and_an_old_link_expires(app, monkeypatch): + sent = _mail(app, monkeypatch) + + async def _go(): + from datetime import datetime, timedelta, timezone + from sqlalchemy import update + from steward.models.password_resets import PasswordReset + uid, name, _ = await _local(app) + guest = await _stranger(app) + for _ in range(2): + await guest.post("/forgot", headers=CSRF, form={"name": name}) + first, second = (_KEY.search(m.get_body(("plain",)).get_content()).group(1) for m in sent) + assert (await guest.get(f"/reset/{first}")).status_code == 404 + assert (await guest.get(f"/reset/{second}")).status_code == 200 + + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(PasswordReset).where(PasswordReset.user_id == uid) + .values(expires_at=datetime.now(timezone.utc) - timedelta(seconds=1))) + assert (await guest.get(f"/reset/{second}")).status_code == 404 + + asyncio.run(_go()) + + +def test_without_mail_the_forgot_page_says_to_ask_an_admin(app): + async def _go(): + app.config["SMTP"] = {} + guest = await _stranger(app) + assert 'href="/forgot"' not in await (await guest.get("/login")).get_data(as_text=True) + page = await (await guest.get("/forgot")).get_data(as_text=True) + assert "Ask an admin" in page and 'name="name"' not in page + + asyncio.run(_go()) diff --git a/tests/test_breadcrumbs.py b/tests/test_breadcrumbs.py index 42bdd56..2fb8d0e 100644 --- a/tests/test_breadcrumbs.py +++ b/tests/test_breadcrumbs.py @@ -27,6 +27,8 @@ TOP_LEVEL = { "steward/templates/auth/login.html", "steward/templates/auth/setup.html", "steward/templates/auth/invite.html", + "steward/templates/auth/forgot.html", + "steward/templates/auth/reset.html", "steward/templates/errors/404.html", "steward/templates/errors/500.html", "steward/templates/errors/database_unavailable.html",