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

"""App passwords (SPEC App. D §D.5): one generated password per mail app,
shown once, revocable on its own, never accepted by the webmail's sign-in."""

from __future__ import annotations

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

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

from app.config import get_settings
from app.identity import app_passwords
from app.identity.passwords import hash_password
from app.identity.reset import issue_token
from app.identity.sessions import create_session
from app.main import create_app
from app.models import User, UserAppPassword

CSRF = "csrf-token"
PASSWORD = "hunter2pass"


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

    async def user(self) -> User:
        user = User(
            handle=f"ap{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password(PASSWORD)
        )
        self.db.add(user)
        await self.db.commit()
        self.created.append(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() -> AsyncIterator[Ctx]:
    engine = create_async_engine(get_settings().database_url, poolclass=NullPool)
    db = async_sessionmaker(engine, expire_on_commit=False)()
    anon = httpx.AsyncClient(transport=ASGITransport(app=create_app()), base_url="http://t")
    c = Ctx(db=db, anon=anon)
    try:
        yield c
    finally:
        await anon.aclose()
        await db.execute(delete(User).where(User.id.in_(c.created)))
        await db.commit()
        await db.close()
        await engine.dispose()


async def _create(client: httpx.AsyncClient, name: str, password: str = PASSWORD) -> httpx.Response:
    return await client.post("/api/auth/app-passwords", json={"name": name, "password": password})


async def test_created_once_listed_without_secret_and_revoked(ctx: Ctx) -> None:
    user = await ctx.user()
    client = await ctx.signed_in(user)

    assert (await _create(client, "Thunderbird", "wrong")).status_code == 403
    made = await _create(client, "  Thunderbird on laptop ")
    assert made.status_code == 200
    body = made.json()
    assert body["name"] == "Thunderbird on laptop"
    assert re.fullmatch(r"[a-z]{16}", body["app_password"])

    listed = (await client.get("/api/auth/app-passwords")).json()
    assert [a["name"] for a in listed] == ["Thunderbird on laptop"]
    assert "app_password" not in listed[0] and "password_hash" not in listed[0]
    assert listed[0]["last_used_at"] is None

    # What the mail server will check: this one matches, nothing else does.
    row = await app_passwords.verify(ctx.db, user.id, body["app_password"])
    assert row is not None and str(row.id) == body["id"]
    assert await app_passwords.verify(ctx.db, user.id, PASSWORD) is None
    assert await app_passwords.verify(ctx.db, user.id, "a" * 16) is None
    stored = await ctx.db.scalar(
        select(UserAppPassword.password_hash).where(UserAppPassword.id == row.id)
    )
    assert stored and stored.startswith("$argon2id$") and body["app_password"] not in stored

    # The webmail's own sign-in never takes an app password.
    login = await ctx.anon.post(
        "/api/auth/login", json={"handle": user.handle, "password": body["app_password"]}
    )
    assert login.status_code == 401

    # Someone else cannot revoke it; its owner can, once.
    other = await ctx.signed_in(await ctx.user())
    assert (await other.delete(f"/api/auth/app-passwords/{body['id']}")).status_code == 404
    assert (await client.delete(f"/api/auth/app-passwords/{body['id']}")).status_code == 200
    assert (await client.delete(f"/api/auth/app-passwords/{body['id']}")).status_code == 404
    assert await app_passwords.verify(ctx.db, user.id, body["app_password"]) is None
    await client.aclose()
    await other.aclose()


async def test_at_most_a_few_per_user(ctx: Ctx, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(app_passwords, "MAX_PER_USER", 2)
    client = await ctx.signed_in(await ctx.user())
    assert (await _create(client, "one")).status_code == 200
    assert (await _create(client, "two")).status_code == 200
    third = await _create(client, "three")
    assert third.status_code == 409 and third.json()["error"]["code"] == "too_many"
    await client.aclose()


async def test_password_reset_revokes_them_all(ctx: Ctx) -> None:
    user = await ctx.user()
    client = await ctx.signed_in(user)
    secret = (await _create(client, "Phone")).json()["app_password"]
    token = await issue_token(user.id)
    reset = await ctx.anon.post(
        "/api/auth/password/reset", json={"token": token, "password": "brandNewPass9"}
    )
    assert reset.status_code == 200
    assert await app_passwords.verify(ctx.db, user.id, secret) is None
    await client.aclose()