Fix password recovery and credential lock order during deletion

papyrus/services/auth/_core.py

@@ -146,15 +146,26 @@ async def _create_exchange_code(
     return plain_code
 
 
+async def _lock_credential_owner(session: AsyncSession, user_id: UUID | None) -> None:
+    """Match deletion's owner-before-credential lock order and refresh cached account state."""
+    if user_id is None:
+        return
+
+    await session.scalar(
+        select(User).where(User.user_id == user_id).with_for_update().execution_options(populate_existing=True)
+    )
+
+
 async def _consume_exchange_code(session: AsyncSession, code: str, purpose: str) -> AuthExchangeCode:
-    result = await session.execute(
-        select(AuthExchangeCode).where(
-            AuthExchangeCode.code_hash == hash_opaque_token(code),
-            AuthExchangeCode.purpose == purpose,
-            AuthExchangeCode.used_at.is_(None),
-        )
+    query = select(AuthExchangeCode).where(
+        AuthExchangeCode.code_hash == hash_opaque_token(code),
+        AuthExchangeCode.purpose == purpose,
+        AuthExchangeCode.used_at.is_(None),
     )
 
+    user_id = await session.scalar(query.with_only_columns(AuthExchangeCode.user_id))
+    await _lock_credential_owner(session, user_id)
+    result = await session.execute(query.with_for_update().execution_options(populate_existing=True))
     exchange_code = result.scalar_one_or_none()
 
     if exchange_code is None or exchange_code.expires_at <= _now():
@@ -187,14 +198,15 @@ async def _issue_email_action_token(
 
 
 async def _consume_email_action_token(session: AsyncSession, token: str, action_type: str) -> EmailActionToken:
-    result = await session.execute(
-        select(EmailActionToken).where(
-            EmailActionToken.token_hash == hash_opaque_token(token),
-            EmailActionToken.action_type == action_type,
-            EmailActionToken.used_at.is_(None),
-        )
+    query = select(EmailActionToken).where(
+        EmailActionToken.token_hash == hash_opaque_token(token),
+        EmailActionToken.action_type == action_type,
+        EmailActionToken.used_at.is_(None),
     )
 
+    user_id = await session.scalar(query.with_only_columns(EmailActionToken.user_id))
+    await _lock_credential_owner(session, user_id)
+    result = await session.execute(query.with_for_update().execution_options(populate_existing=True))
     email_token = result.scalar_one_or_none()
 
     if email_token is None or email_token.expires_at <= _now():

papyrus/services/auth/email_flows.py

@@ -90,8 +90,11 @@ async def begin_password_reset(session: AsyncSession, email: str) -> str:
 
 
 async def reset_password(session: AsyncSession, token: str, new_password: str) -> str:
+    """Restore password access during the grace period without cancelling deletion or signing in."""
+    from papyrus.services.account_deletion import _lock_user
+
     email_token = await _consume_email_action_token(session, token, PASSWORD_RESET_ACTION)
-    user = await _get_active_user(session, email_token.user_id)
+    user = await _lock_user(session, email_token.user_id)
 
     credential_result = await session.execute(
         select(PasswordCredential).where(PasswordCredential.user_id == user.user_id)

tests/api/routes/test_account_password_recovery.py

@@ -0,0 +1,131 @@
+"""Password recovery preserves deletion state until the owner explicitly cancels."""
+
+from datetime import UTC, datetime, timedelta
+from urllib.parse import parse_qs, urlsplit
+from uuid import UUID
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.config import get_settings
+from papyrus.core.security import generate_opaque_token, hash_opaque_token, verify_password
+from papyrus.models import AuthSession, EmailActionToken, PasswordCredential, User
+from papyrus.services import account_deletion, auth, email
+
+
+async def test_forgot_password_during_grace_requires_explicit_cancellation(
+    client: AsyncClient,
+    auth_user: dict[str, str],
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    messages: list[str] = []
+
+    def capture_email(recipient: str, subject: str, body: str, **kwargs: object) -> None:
+        messages.append(body)
+
+    monkeypatch.setattr(email, "is_email_delivery_configured", lambda: True)
+    monkeypatch.setattr(email, "send_email", capture_email)
+    monkeypatch.setattr(get_settings(), "app_public_base_url", "https://app.example.test")
+    scheduled = await client.post("/v1/account-deletion/current", headers=auth_headers)
+    assert scheduled.status_code == 200
+    due = scheduled.json()["deletion_due_at"]
+    user_id = UUID(auth_user["user_id"])
+    user = await db_session.get(User, user_id)
+    assert user is not None
+    requested_at = user.deletion_requested_at
+    disabled_at = user.disabled_at
+
+    forgotten = await client.post("/v1/auth/forgot-password", json={"email": "user@example.com"})
+
+    assert forgotten.status_code == 200
+    token = parse_qs(urlsplit(messages[-1].strip().splitlines()[-1]).query)["token"][0]
+    password = "RecoveredP@ssword123"
+    reset = await client.post("/v1/auth/reset-password", json={"token": token, "password": password})
+    assert reset.status_code == 200, reset.text
+    assert set(reset.json()) == {"message"}
+    await db_session.refresh(user)
+    assert user.disabled_at == disabled_at
+    assert user.deletion_requested_at == requested_at
+    assert user.deletion_due_at == datetime.fromisoformat(due)
+
+    active_sessions = await db_session.scalar(
+        select(func.count())
+        .select_from(AuthSession)
+        .where(AuthSession.user_id == user_id, AuthSession.revoked_at.is_(None))
+    )
+
+    assert active_sessions == 0
+    assert (await client.get("/v1/users/me", headers=auth_headers)).status_code == 401
+    assert (
+        await client.post("/v1/auth/refresh", json={"refresh_token": auth_user["refresh_token"]})
+    ).status_code == 401
+    assert (
+        await client.post("/v1/auth/reset-password", json={"token": token, "password": password})
+    ).status_code == 400
+
+    login = await client.post("/v1/auth/login", json={"email": "user@example.com", "password": password})
+
+    assert login.status_code == 403
+    assert login.json()["error"]["code"] == "ACCOUNT_DELETION_PENDING"
+    authorization = login.json()["error"]["details"]["authorization"]
+    assert authorization["deletion_due_at"] == due
+
+    cancelled = await client.post(
+        "/v1/account-deletion/cancel", json={"authorization_token": authorization["authorization_token"]}
+    )
+
+    assert cancelled.status_code == 200
+    restored_headers = {"Authorization": f"Bearer {cancelled.json()['access_token']}"}
+    assert (await client.get("/v1/users/me", headers=restored_headers)).status_code == 200
+    restored_login = await client.post("/v1/auth/login", json={"email": "user@example.com", "password": password})
+    assert restored_login.status_code == 200
+
+    old_login = await client.post(
+        "/v1/auth/login", json={"email": "user@example.com", "password": auth_user["password"]}
+    )
+
+    assert old_login.status_code == 401
+
+
+@pytest.mark.parametrize("seconds_until_deletion", [None, 0, -1])
+async def test_password_reset_rejects_disabled_and_expired_recovery_accounts(
+    client: AsyncClient,
+    auth_user: dict[str, str],
+    db_session: AsyncSession,
+    monkeypatch: pytest.MonkeyPatch,
+    seconds_until_deletion: int | None,
+) -> None:
+    now = datetime.now(UTC)
+    monkeypatch.setattr(account_deletion, "_now", lambda: now)
+    user_id = UUID(auth_user["user_id"])
+    user = await db_session.get(User, user_id)
+    assert user is not None
+    user.disabled_at = now
+
+    if seconds_until_deletion is not None:
+        user.deletion_requested_at = now - timedelta(days=30)
+        user.deletion_due_at = now + timedelta(seconds=seconds_until_deletion)
+
+    token = generate_opaque_token()
+
+    db_session.add(
+        EmailActionToken(
+            user_id=user_id,
+            action_type=auth.PASSWORD_RESET_ACTION,
+            token_hash=hash_opaque_token(token),
+            expires_at=now + timedelta(minutes=5),
+        )
+    )
+
+    await db_session.commit()
+    response = await client.post("/v1/auth/reset-password", json={"token": token, "password": "RejectedP@ssword123"})
+    assert response.status_code == 403
+    credential = await db_session.get(PasswordCredential, user_id)
+    assert credential is not None
+    assert verify_password(auth_user["password"], credential.password_hash)
+    await db_session.refresh(user)
+    assert user.disabled_at == now

tests/services/test_auth_recovery_concurrency.py

@@ -0,0 +1,112 @@
+"""Deletion cancellation and authentication always lock the owner before credentials."""
+
+import asyncio
+from uuid import UUID
+
+import pytest
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from papyrus.core.exceptions import UnauthorizedError, ValidationError
+from papyrus.models import AuthSession, User
+from papyrus.schemas.account_deletion import AccountPasswordRequest
+from papyrus.schemas.auth import OAuthExchangeRequest
+from papyrus.services import account_deletion, auth
+from papyrus.services.auth import _core
+
+
+@pytest.mark.parametrize("flow", ["google", "password_reset"])
+async def test_cancellation_revokes_waiting_authentication_without_deadlock(
+    test_session_maker: async_sessionmaker[AsyncSession],
+    auth_user: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+    flow: str,
+) -> None:
+    user_id = UUID(auth_user["user_id"])
+
+    async with test_session_maker() as setup:
+        await account_deletion.schedule_current_account(setup, user_id, UUID(auth_user["session_id"]))
+
+        recovery = await account_deletion.authenticate_password(
+            setup, AccountPasswordRequest(email="user@example.com", password=auth_user["password"])
+        )
+
+        if flow == "google":
+            token = await _core._create_exchange_code(setup, purpose="login", redirect_uri="", user_id=user_id)
+        else:
+            token = await _core._issue_email_action_token(setup, user_id, auth.PASSWORD_RESET_ACTION, 5)
+
+        await setup.commit()
+
+    owner_locked = asyncio.Event()
+    resume_cancellation = asyncio.Event()
+    revoke_credentials = account_deletion._revoke_credentials
+
+    async def pause_before_revocation(session: AsyncSession, owner_id: UUID) -> None:
+        owner_locked.set()
+        await resume_cancellation.wait()
+        await revoke_credentials(session, owner_id)
+
+    monkeypatch.setattr(account_deletion, "_revoke_credentials", pause_before_revocation)
+
+    async with (
+        test_session_maker() as cancellation,
+        test_session_maker() as authentication,
+        test_session_maker() as monitor,
+    ):
+        cancellation_pid = await cancellation.scalar(select(func.pg_backend_pid()))
+        authentication_pid = await authentication.scalar(select(func.pg_backend_pid()))
+        cached_user = await authentication.get(User, user_id)
+        assert cached_user is not None
+        assert cached_user.disabled_at is not None
+
+        async def authenticate() -> None:
+            if flow == "google":
+                with pytest.raises(UnauthorizedError):
+                    await auth.exchange_login_code(authentication, OAuthExchangeRequest(code=token), None)
+            else:
+                with pytest.raises(ValidationError):
+                    await auth.reset_password(authentication, token, "RacingP@ssword123")
+
+        cancellation_task = asyncio.create_task(
+            account_deletion.cancel_deletion(cancellation, recovery.authorization_token)
+        )
+
+        authentication_task = None
+
+        try:
+            await asyncio.wait_for(owner_locked.wait(), timeout=5)
+            authentication_task = asyncio.create_task(authenticate())
+
+            async with asyncio.timeout(5):
+                while True:
+                    blockers = await monitor.scalar(select(func.pg_blocking_pids(authentication_pid)))
+
+                    if cancellation_pid in blockers:
+                        break
+
+                    await asyncio.sleep(0.01)
+
+            resume_cancellation.set()
+            restored, _ = await asyncio.wait_for(asyncio.gather(cancellation_task, authentication_task), timeout=5)
+            assert restored.user.user_id == user_id
+            assert cached_user.disabled_at is None
+            assert cached_user.deletion_due_at is None
+        finally:
+            resume_cancellation.set()
+            tasks = [task for task in (cancellation_task, authentication_task) if task is not None]
+
+            for task in tasks:
+                if not task.done():
+                    task.cancel()
+
+            await asyncio.gather(*tasks, return_exceptions=True)
+
+    async with test_session_maker() as verification:
+        active_sessions = await verification.scalar(
+            select(func.count())
+            .select_from(AuthSession)
+            .where(AuthSession.user_id == user_id, AuthSession.revoked_at.is_(None))
+        )
+
+        assert active_sessions == 1