bvandeusenandClaude Opus 5.5
ea523b8c37
Step #5428 "Account page: change your own password" of milestone 497 "Passwords: change your own, and reset it by email". A Password section in the page's one form, for local accounts: current, new, again. Blank keeps the password; a new one signs out every other browser (sessions_since) and keeps this one. A wrong current password counts toward the login throttle. Audited as account.password_changed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
116 lines
5.0 KiB
Python
116 lines
5.0 KiB
Python
"""Integration: local users manage their own password (milestone 497): change
|
|
it on the Account page, or reset it from an emailed link. The routes run
|
|
for real against Postgres; only mail delivery is replaced."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import time
|
|
import uuid
|
|
|
|
import bcrypt
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
CSRF = {"X-CSRF-Token": "test-token"}
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
if not os.environ.get("STEWARD_DATABASE_URL"):
|
|
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
|
|
app = create_app(testing=False)
|
|
yield app
|
|
login_throttle._failures.clear()
|
|
|
|
|
|
async def _local(app, password="old password", source="local"):
|
|
"""A local account with a real password, and a browser signed in as it."""
|
|
from steward.models.users import User, UserRole
|
|
uid = str(uuid.uuid4())
|
|
name = f"pw-{uid[:8]}"
|
|
async with app.db_sessionmaker() as s:
|
|
async with s.begin():
|
|
s.add(User(id=uid, username=name, email=f"{name}@example.com", role=UserRole.viewer,
|
|
password_hash=bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode(),
|
|
is_active=True, auth_source=source, time_zone="Europe/London"))
|
|
return uid, name, await _signed_in(app, uid, name)
|
|
|
|
|
|
async def _signed_in(app, uid, name, ago=60):
|
|
client = app.test_client()
|
|
async with client.session_transaction() as sess:
|
|
sess.update(user_id=uid, user_role="viewer", username=name, _csrf="test-token",
|
|
login_at=time.time() - ago)
|
|
return client
|
|
|
|
|
|
async def _user(app, uid):
|
|
from steward.models.users import User
|
|
async with app.db_sessionmaker() as s:
|
|
return await s.get(User, uid)
|
|
|
|
|
|
async def _audits(app, uid):
|
|
from sqlalchemy import select
|
|
from steward.models.audit import AuditEvent
|
|
async with app.db_sessionmaker() as s:
|
|
rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid)
|
|
.order_by(AuditEvent.timestamp))).scalars().all()
|
|
return [(r.action, json.loads(r.detail_json or "{}")) for r in rows]
|
|
|
|
|
|
def _checks(user, password):
|
|
return bcrypt.checkpw(password.encode(), user.password_hash.encode())
|
|
|
|
|
|
def test_a_local_user_changes_their_password_and_their_other_browsers_sign_out(app):
|
|
async def _go():
|
|
uid, name, me = await _local(app)
|
|
elsewhere = await _signed_in(app, uid, name, ago=120)
|
|
page = await (await me.get("/account/")).get_data(as_text=True)
|
|
assert 'name="current_password"' in page and 'name="new_password"' in page
|
|
keep = {"time_zone": "Europe/London", "email": f"{name}@example.com"}
|
|
|
|
# Blank: the rest saves and the password stays.
|
|
resp = await me.post("/account/", headers=CSRF, form=keep | {"current_password": "old password"})
|
|
assert resp.headers["Location"] == "/account/?done=saved"
|
|
assert _checks(await _user(app, uid), "old password")
|
|
|
|
resp = await me.post("/account/", headers=CSRF, form=keep | {
|
|
"current_password": "wrong one", "new_password": "new password", "new_confirm": "new password"})
|
|
assert resp.status_code == 400 and "current password isn" in await resp.get_data(as_text=True)
|
|
resp = await me.post("/account/", headers=CSRF, form=keep | {
|
|
"current_password": "old password", "new_password": "new password", "new_confirm": "other"})
|
|
assert resp.status_code == 400 and "passwords don" in await resp.get_data(as_text=True)
|
|
assert _checks(await _user(app, uid), "old password")
|
|
|
|
resp = await me.post("/account/", headers=CSRF, form=keep | {
|
|
"current_password": "old password", "new_password": "new password", "new_confirm": "new password"})
|
|
assert resp.headers["Location"] == "/account/?done=password"
|
|
assert _checks(await _user(app, uid), "new password")
|
|
page = await (await me.get(resp.headers["Location"])).get_data(as_text=True)
|
|
assert "Your password is changed" in page # still signed in here
|
|
assert (await elsewhere.get("/account/")).status_code == 302 # signed out there
|
|
assert "account.password_changed" in [a for a, _ in await _audits(app, uid)]
|
|
|
|
asyncio.run(_go())
|
|
|
|
|
|
def test_an_sso_account_has_no_password_to_change(app):
|
|
async def _go():
|
|
uid, name, me = await _local(app, source="oidc")
|
|
page = await (await me.get("/account/")).get_data(as_text=True)
|
|
assert 'name="new_password"' not in page and "changed there, not in Steward" in page
|
|
resp = await me.post("/account/", headers=CSRF, form={
|
|
"time_zone": "Europe/London", "email": f"{name}@example.com",
|
|
"current_password": "old password", "new_password": "new password", "new_confirm": "new password"})
|
|
assert resp.status_code == 302 and _checks(await _user(app, uid), "old password")
|
|
|
|
asyncio.run(_go())
|