FabledSteward/tests/integration/test_passwords.py

"""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 re
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
    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
    for t in (login_throttle, reset_throttle):
        t._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())


_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())