feat(auth): reset a forgotten password from an emailed link (milestone 497)
Step #5429 "Reset by email: request, link and new-password page" of milestone 497 "Passwords: change your own, and reset it by email". - Sign-in page: "Forgot your password?" while a mail server is set. - /forgot takes a username or email and answers the same either way; only an active local account with a real address gets mail. Requests count per address on their own throttle, so they never lock out sign-in. - password_resets (migration 0039): a one-time key, stored as its sha256, working for an hour; asking again replaces it. - /reset/<key>: the new password, then signed straight in; every other browser is signed out (sessions_since). Bad keys count as invite keys do. - Email "[Steward] Reset your password", in the account's own zone. - Audited: user.password_reset_requested, user.password_reset_by_email. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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 <x>", 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 "<x>" not in html
|
||||
assert all(u.startswith("https://steward.example") for u in re.findall(r"""https?://[^\s"'<>)]+""", html))
|
||||
@@ -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())
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user