webmail-client/backend/tests/identity/test_password_reset.py

"""Password reset by email, recovery address, and the system mailer (SPEC §6.1).

The SMTP send is replaced with a recorder, so nothing leaves the machine; the
reset link it would have sent is read back from what was recorded.
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from dataclasses import dataclass, field

import httpx
import pyotp
import pytest
import pytest_asyncio
from httpx import ASGITransport
from sqlalchemy import delete, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.api import admin as admin_api
from app.config import get_settings
from app.identity import routes
from app.identity.mailer import MailerConfig
from app.identity.passwords import hash_password
from app.identity.reset import MAX_TOTP_FAILURES
from app.identity.service import app_settings
from app.identity.sessions import create_session
from app.identity.totp import encrypt_totp_secret
from app.main import create_app
from app.models import AppSettings, User
from app.redis_client import get_redis

CSRF = "csrf-token"
PASSWORD = "hunter2pass"
NEW_PASSWORD = "brandNewPass9"
CFG = MailerConfig(
    host="smtp.example.com",
    port=587,
    security="starttls",
    username="postmaster@example.com",
    password="pw",
    sender="postmaster@example.com",
    public_url="https://mail.example.com",
)


@dataclass
class Sent:
    mails: list[tuple[str, str, str]] = field(default_factory=list)  # (to, subject, text)

    def token(self) -> str:
        text = self.mails[-1][2]
        link = next(w for w in text.split() if w.startswith("https://mail.example.com/"))
        assert link.startswith("https://mail.example.com/reset-password#token=")
        return link.split("#token=", 1)[1]


@dataclass
class Ctx:
    session: AsyncSession
    anon: httpx.AsyncClient
    sent: Sent
    created: list[uuid.UUID] = field(default_factory=list)

    async def user(
        self, *, recovery: str | None = "me@elsewhere.example", admin: bool = False
    ) -> User:
        user = User(
            handle=f"pr{uuid.uuid4().hex[:8]}@test.local",
            password_hash=hash_password(PASSWORD),
            recovery_email=recovery,
            is_admin=admin,
        )
        self.session.add(user)
        await self.session.commit()
        self.created.append(user.id)
        await get_redis().delete(f"pwreset-rate:{user.id}")
        return user

    async def signed_in(self, user: User) -> httpx.AsyncClient:
        token = await create_session(user.id, "pytest")
        return httpx.AsyncClient(
            transport=ASGITransport(app=create_app()),
            base_url="http://testserver",
            cookies={"session": token, "csrf": CSRF},
            headers={"X-CSRF": CSRF},
        )


@pytest_asyncio.fixture
async def ctx(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[Ctx]:
    sent = Sent()

    async def fake_send(cfg: MailerConfig, to: str, subject: str, text: str, html: str) -> None:
        sent.mails.append((to, subject, text))

    monkeypatch.setattr(routes, "send_mail", fake_send)
    monkeypatch.setattr(admin_api, "send_mail", fake_send)
    monkeypatch.setattr(routes, "mailer_config", lambda row: CFG)

    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    session = async_sessionmaker(engine, expire_on_commit=False)()
    anon = httpx.AsyncClient(transport=ASGITransport(app=create_app()), base_url="http://t")
    c = Ctx(session=session, anon=anon, sent=sent)
    try:
        yield c
    finally:
        await anon.aclose()
        await session.execute(delete(User).where(User.id.in_(c.created)))
        await session.commit()
        await session.close()
        await engine.dispose()


async def _request(ctx: Ctx, handle: str) -> httpx.Response:
    return await ctx.anon.post("/api/auth/password/reset-request", json={"handle": handle})


async def _login(ctx: Ctx, handle: str, password: str) -> int:
    resp = await ctx.anon.post("/api/auth/login", json={"handle": handle, "password": password})
    return resp.status_code


# --- requesting a link -----------------------------------------------------------


async def test_request_mails_a_one_time_link_to_the_recovery_address(ctx: Ctx) -> None:
    user = await ctx.user()
    assert (await _request(ctx, user.handle)).status_code == 200
    assert [m[0] for m in ctx.sent.mails] == ["me@elsewhere.example"]
    token = ctx.sent.token()

    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
    assert check.json() == {"valid": True, "totp_required": False}

    reset = await ctx.anon.post(
        "/api/auth/password/reset", json={"token": token, "password": NEW_PASSWORD}
    )
    assert reset.status_code == 200
    # Used up: a second go fails, and the page is told so up front.
    again = await ctx.anon.post(
        "/api/auth/password/reset", json={"token": token, "password": "anotherPass12"}
    )
    assert again.json()["error"]["code"] == "invalid_token"
    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
    assert check.json()["valid"] is False

    assert await _login(ctx, user.handle, NEW_PASSWORD) == 200
    assert await _login(ctx, user.handle, PASSWORD) == 401


async def test_the_answer_never_says_whether_the_account_exists(ctx: Ctx) -> None:
    no_recovery = await ctx.user(recovery=None)
    for handle in ("nobody-here-at-all", no_recovery.handle):
        resp = await _request(ctx, handle)
        assert resp.status_code == 200 and resp.json() == {"status": "ok"}
    assert ctx.sent.mails == []


async def test_no_mail_while_system_mail_is_not_set_up(
    ctx: Ctx, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(routes, "mailer_config", lambda row: None)
    user = await ctx.user()
    assert (await _request(ctx, user.handle)).status_code == 200
    assert ctx.sent.mails == []


async def test_requests_are_capped_per_hour(ctx: Ctx) -> None:
    user = await ctx.user()
    for _ in range(5):
        assert (await _request(ctx, user.handle)).status_code == 200
    assert len(ctx.sent.mails) == 3


# --- two-step verification ---------------------------------------------------------


async def test_2fa_code_is_required_before_the_password_changes(ctx: Ctx) -> None:
    user = await ctx.user()
    secret = pyotp.random_base32()
    await ctx.session.execute(
        update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
    )
    await ctx.session.commit()
    await _request(ctx, user.handle)
    token = ctx.sent.token()
    body = {"token": token, "password": NEW_PASSWORD}

    check = await ctx.anon.post("/api/auth/password/reset/check", json={"token": token})
    assert check.json()["totp_required"] is True
    missing = await ctx.anon.post("/api/auth/password/reset", json=body)
    assert missing.json()["error"]["code"] == "totp_required"
    wrong = await ctx.anon.post("/api/auth/password/reset", json={**body, "totp": "000000"})
    assert wrong.json()["error"]["code"] == "totp_invalid"
    # Nothing changed yet, and the link still works.
    assert await _login(ctx, user.handle, PASSWORD) == 401  # 2FA account: needs a code
    good = await ctx.anon.post(
        "/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
    )
    assert good.status_code == 200
    login = await ctx.anon.post(
        "/api/auth/login",
        json={"handle": user.handle, "password": NEW_PASSWORD, "totp": pyotp.TOTP(secret).now()},
    )
    assert login.status_code == 200


async def test_too_many_wrong_codes_burn_the_link(ctx: Ctx) -> None:
    user = await ctx.user()
    secret = pyotp.random_base32()
    await ctx.session.execute(
        update(User).where(User.id == user.id).values(totp_secret=encrypt_totp_secret(secret))
    )
    await ctx.session.commit()
    await _request(ctx, user.handle)
    token = ctx.sent.token()
    body = {"token": token, "password": NEW_PASSWORD, "totp": "000000"}
    for _ in range(MAX_TOTP_FAILURES):
        await ctx.anon.post("/api/auth/password/reset", json=body)
    late = await ctx.anon.post(
        "/api/auth/password/reset", json={**body, "totp": pyotp.TOTP(secret).now()}
    )
    assert late.json()["error"]["code"] == "invalid_token"


# --- the recovery address ----------------------------------------------------------


async def test_changing_the_recovery_email_takes_the_password(ctx: Ctx) -> None:
    user = await ctx.user(recovery=None)
    async with await ctx.signed_in(user) as c:
        wrong = await c.put(
            "/api/auth/recovery-email",
            json={"password": "not-it", "recovery_email": "new@example.com"},
        )
        assert wrong.status_code == 403
        bad = await c.put(
            "/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": "nope"}
        )
        assert bad.json()["error"]["code"] == "invalid_email"
        ok = await c.put(
            "/api/auth/recovery-email",
            json={"password": PASSWORD, "recovery_email": " New@Example.com "},
        )
        assert ok.json()["recovery_email"] == "new@example.com"
        removed = await c.put(
            "/api/auth/recovery-email", json={"password": PASSWORD, "recovery_email": ""}
        )
        assert removed.json()["recovery_email"] is None


# --- admin: system mail settings -----------------------------------------------------


@pytest_asyncio.fixture
async def restore_settings(ctx: Ctx) -> AsyncIterator[None]:
    row = await app_settings(ctx.session)
    saved = {
        k: getattr(row, k)
        for k in (
            "smtp_host",
            "smtp_port",
            "smtp_security",
            "smtp_username",
            "smtp_secret_ct",
            "smtp_from",
            "public_url",
        )
    }
    await ctx.session.commit()
    yield
    await ctx.session.execute(update(AppSettings).values(**saved))
    await ctx.session.commit()


async def test_admin_configures_system_mail_without_ever_seeing_the_password(
    ctx: Ctx, restore_settings: None
) -> None:
    admin = await ctx.user(admin=True)
    async with await ctx.signed_in(admin) as c:
        resp = await c.put(
            "/api/admin/mail",
            json={
                "smtp_host": "smtp.example.com",
                "smtp_port": 465,
                "smtp_security": "ssl",
                "smtp_username": "postmaster@example.com",
                "smtp_password": "s3cret",
                "smtp_from": "postmaster@example.com",
                "public_url": "https://mail.example.com/",
            },
        )
        out = resp.json()
        assert out["has_password"] is True and out["configured"] is True
        assert out["public_url"] == "https://mail.example.com"
        assert "s3cret" not in resp.text
        # Omitting the password keeps it; an empty one removes it.
        kept = (await c.put("/api/admin/mail", json={"smtp_port": 587})).json()
        assert kept["has_password"] is True
        gone = (await c.put("/api/admin/mail", json={"smtp_password": ""})).json()
        assert gone["has_password"] is False
        bad = await c.put("/api/admin/mail", json={"public_url": "mail.example.com"})
        assert bad.json()["error"]["code"] == "invalid_url"
        test = await c.post("/api/admin/mail/test", json={"to": "me@example.com"})
        assert test.status_code == 200
    assert ctx.sent.mails[-1][:2] == ("me@example.com", "Ultimo Email test message")


async def test_reset_mail_carries_the_logo_inline(monkeypatch: pytest.MonkeyPatch) -> None:
    """The logo is a PNG inside the message (cid:), not a remote image or SVG."""
    import aiosmtplib

    from app.identity import mailer

    sent: list[object] = []

    async def fake_send(msg: object, **kwargs: object) -> None:
        sent.append(msg)

    monkeypatch.setattr(aiosmtplib, "send", fake_send)
    subject, text, html = mailer.reset_mail("https://mail.example.com/x", "a@test.local")
    await mailer.send_mail(CFG, "to@example.com", subject, text, html)
    msg = sent[0]
    parts = {p.get_content_type(): p for p in msg.walk()}  # type: ignore[attr-defined]
    assert {"multipart/alternative", "text/plain", "multipart/related", "text/html"} <= set(parts)
    logo = parts["image/png"]
    assert logo["Content-ID"] == "<ultimo-logo@ultimo.email>"
    assert logo.get_payload(decode=True).startswith(b"\x89PNG")
    assert 'src="cid:ultimo-logo@ultimo.email"' in parts["text/html"].get_content()