webmail-client/backend/app/identity/app_passwords.py

"""App passwords: one generated password per mail app (SPEC App. D §D.5).

A mail app (Thunderbird, a phone's Mail app) signs in to the hosted mailbox
over IMAP/SMTP, and a calendar or contacts app to the DAV server — never with
the account password (no 2FA code can be asked for there), always with one of
these. Each gets a password of its own, shown once, that can be
revoked without touching the main password or the other apps. The webmail
itself never accepts one.

With hosted mailboxes on, the mail server makes the password (``app_`` and 36
more letters and digits: its format names the credential) and checks it for
IMAP/SMTP itself; its argon2id hash is kept here too, for the DAV server, with
the server's id to revoke it there. Without, it is 16 random letters, for DAV
only. Either way the hash is of the bare password: apps send exactly that, so
it is copied and entered without the spaces it is shown with.
"""

from __future__ import annotations

import secrets
import string
import uuid

from sqlalchemy import delete, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app import hosted
from app.identity.passwords import hash_password, verify_password
from app.models import User, UserAppPassword

LENGTH = 16  # 26^16 ≈ 2^75
SERVER_LENGTH = 40  # the mail server's: "app_" + base32 of a 4-byte id and 18 random bytes
MAX_PER_USER = 25
NAME_MAX = 60
_ALPHABET = string.ascii_lowercase


def generation_key(user_id: uuid.UUID) -> str:
    """Redis counter bumped whenever the user's app passwords change: DAV
    sign-ins cached under the old value stop counting (app/dav/auth.py)."""
    return f"apppw:gen:{user_id}"


async def _changed(user_id: uuid.UUID) -> None:
    from app.redis_client import get_redis

    await get_redis().incr(generation_key(user_id))


class TooMany(Exception):
    """The user already has MAX_PER_USER app passwords."""


def generate() -> str:
    return "".join(secrets.choice(_ALPHABET) for _ in range(LENGTH))


async def list_for(db: AsyncSession, user_id: uuid.UUID) -> list[UserAppPassword]:
    rows = await db.scalars(
        select(UserAppPassword)
        .where(UserAppPassword.user_id == user_id)
        .order_by(UserAppPassword.created_at.desc())
    )
    return list(rows)


async def create(db: AsyncSession, user_id: uuid.UUID, name: str) -> tuple[UserAppPassword, str]:
    """A new app password named ``name``; returns the row and the password
    itself, which is never available again."""
    count = await db.scalar(
        select(func.count()).select_from(UserAppPassword).where(UserAppPassword.user_id == user_id)
    )
    if (count or 0) >= MAX_PER_USER:
        raise TooMany
    name = name.strip()[:NAME_MAX]
    server_id: str | None = None
    if hosted.enabled():
        from app.hosted import admin as mail_admin

        handle = await db.scalar(select(User.handle).where(User.id == user_id))
        server_id, password = await mail_admin.create_app_password(str(handle), name)
    else:
        password = generate()
    row = UserAppPassword(
        user_id=user_id, name=name, password_hash=hash_password(password), server_id=server_id
    )
    db.add(row)
    await db.commit()
    return row, password


async def _revoke_on_server(
    db: AsyncSession, user_id: uuid.UUID, rows: list[UserAppPassword]
) -> None:
    """Revoke on the mail server too (first: if it cannot be reached, the
    password keeps working there, so nothing is revoked here either)."""
    server_ids = [r.server_id for r in rows if r.server_id]
    if server_ids and hosted.enabled():
        from app.hosted import admin as mail_admin

        handle = await db.scalar(select(User.handle).where(User.id == user_id))
        await mail_admin.delete_app_passwords(str(handle), server_ids)


async def revoke(db: AsyncSession, user_id: uuid.UUID, app_password_id: uuid.UUID) -> bool:
    rows = list(
        await db.scalars(
            select(UserAppPassword).where(
                UserAppPassword.id == app_password_id, UserAppPassword.user_id == user_id
            )
        )
    )
    await _revoke_on_server(db, user_id, rows)
    result = await db.execute(
        delete(UserAppPassword).where(
            UserAppPassword.id == app_password_id, UserAppPassword.user_id == user_id
        )
    )
    await db.commit()
    await _changed(user_id)
    return bool(result.rowcount)  # type: ignore[attr-defined]


async def revoke_all(db: AsyncSession, user_id: uuid.UUID) -> None:
    """Every app signs in again (the main password was reset)."""
    await _revoke_on_server(db, user_id, await list_for(db, user_id))
    await db.execute(delete(UserAppPassword).where(UserAppPassword.user_id == user_id))
    await db.commit()
    await _changed(user_id)


async def verify(db: AsyncSession, user_id: uuid.UUID, password: str) -> UserAppPassword | None:
    """The app password ``password`` matches, if any — for the mail server's
    sign-in and for tests; never for the webmail's own sign-in."""
    if len(password) not in (LENGTH, SERVER_LENGTH):
        return None
    for row in await list_for(db, user_id):
        if verify_password(row.password_hash, password):
            return row
    return None