Tech-debt batch 21: password reset revokes all credentials; admin create/invite after commit (#686 #692)

CHANGELOG.md

@@ -14,6 +14,12 @@ All notable changes to this project are documented in this file. The format is b
 
 ## [Unreleased]
 
+### Fixed
+- **Behaviour change:** password resets revoke all browser sessions, access tokens
+  and refresh tokens, sharing revocation with password changes and sign-out-everywhere (#686).
+- Defer single admin create/invite events and invitation-related emails until the
+  request commits; failed commits deliver neither mail nor ordinary events (#692, #627).
+
 ## [0.0.41] — 2026-10-25
 
 ### Fixed

docs/modules/users.md

@@ -236,6 +236,18 @@ Everything else is DB-backed (initial values are pydantic defaults; edit under U
 | `Profile` | `/users/me` | `user` | `USER_DROPDOWN` | — | `990` | _logged-in_ |
 | `Logout` | `/users/logout` (POST) | `log-out` | `USER_DROPDOWN` | — | `999` | _logged-in_ |
 
+## Password reset and credential revocation
+
+A successful password reset, including one reached through an admin-generated
+reset link, signs the account out everywhere: it increments `session_version`,
+revokes all refresh tokens, and deletes all access-token rows (#686). Old browser
+sessions return to login; old access and refresh tokens receive `401`. Session
+counter caches are invalidated only after the revocation commits.
+
+Password changes and sign-out-everywhere use the same revocation. A password
+change reissues the calling browser's cookie so that browser remains signed in;
+a password reset and sign-out-everywhere preserve no existing credentials.
+
 ## Events
 
 | Event | Fields | Fired |
@@ -247,6 +259,14 @@ Everything else is DB-backed (initial values are pydantic defaults; edit under U
 | `UserDisabled` | `user_id` | on admin disable |
 | `RoleAssigned` | `user_id`, `role_name` | once per role on `PUT /admin/{user_id}/roles` |
 
+Single admin create/invite publishes `UserCreated` / `UserInvited` through the
+transactional admin-event helper: durable handlers use the request transaction
+(or outbox), and ordinary subscribers run only after commit. Invitation and
+request-less registration verification emails also wait for successful request
+finalization. A failed final commit discards those emails and ordinary event
+callbacks (#692, #627). Post-commit delivery failures are logged and do not
+undo the committed account; these callbacks do not retry mail delivery.
+
 ## CLI
 
 - `smpy users create-admin --email <e> --password <p> [--full-name <name>] [--force]` — creates (or, with `--force`, updates) an admin user. Idempotent: re-running with the same email is a no-op without `--force`.

modules/users/tests/test_revocation_broadcast.py

@@ -133,18 +133,20 @@ async def test_a_failure_after_the_bump_broadcasts_nothing(
     ):
         """Fail the endpoint *after* it has registered the on-commit callback.
 
-        ``delete`` is called for the access-token rows, which happens after the
-        bump and after ``db.on_commit(...)``. Making it raise reproduces "the
-        transaction did not commit" without needing to break the database.
+        Let the shared helper finish, then fail before the response commits.
+        This exercises rollback with an invalidation callback already queued.
         """
         import users.auth_local.self_account as self_account
 
         await _sign_in(anon_client)
 
-        def _explode(*args, **kwargs):
+        revoke = self_account.revoke_all_credentials
+
+        async def _explode(*args, **kwargs):
+            await revoke(*args, **kwargs)
             raise RuntimeError("simulated failure after the session_version bump")
 
-        monkeypatch.setattr(self_account, "delete", _explode)
+        monkeypatch.setattr(self_account, "revoke_all_credentials", _explode)
 
         with pytest.raises(RuntimeError, match="simulated failure"):
             await anon_client.post(_REVOKE)

modules/users/tests/test_users_admin_create_commit.py

@@ -0,0 +1,77 @@
+"""Single admin creation/invitation side effects follow the commit (#692, #627)."""
+
+import pytest
+from background_tasks.models import OutboxMessage
+from simple_module_db import transaction
+from sqlalchemy import event
+from sqlmodel import select
+from test_users_deletion_events import PausedUserTransport, record_user_event
+from users.contracts.events import UserCreated, UserInvited
+from users.models import User
+
+
+@pytest.mark.parametrize("operation,event_type", [("invite", UserInvited), ("create", UserCreated)])
+@pytest.mark.parametrize("durable", [False, True])
+@pytest.mark.parametrize("fail_commit", [True, False])
+async def test_admin_creation_side_effects_wait_for_commit(
+    admin_client, users_app, captured_mail, monkeypatch, operation, event_type, fail_commit, durable
+):
+    observed = []
+    before_commit = []
+
+    async def observe(message):
+        observed.append(message)
+
+    bus = users_app.state.sm.event_bus
+    bus.subscribe(event_type, observe)
+    if durable:
+        bus.durable_transport = PausedUserTransport()
+        bus.subscribe_durable(event_type, record_user_event)
+    email = "commit-boundary@example.com"
+
+    def fail(session):
+        raise RuntimeError("forced commit failure")
+
+    finalize = transaction.finalize_session
+
+    async def check_commit(session):
+        before_commit.append((list(captured_mail.messages), list(observed)))
+        if fail_commit:
+            event.listen(session.sync_session, "before_commit", fail, once=True)
+        await finalize(session)
+
+    monkeypatch.setattr(transaction, "finalize_session", check_commit)
+    db = users_app.state.sm.db
+    body = {"email": email, "role_names": []}
+    if operation == "create":
+        body["password"] = "CompliantPass1!"
+    response = await admin_client.post(
+        "/api/users/admin/invite" if operation == "invite" else "/api/users/admin", json=body
+    )
+
+    assert response.status_code == (500 if fail_commit else 201), response.text
+    assert before_commit == [([], [])]
+    async with db.session_factory() as session:
+        user = (await session.execute(select(User).where(User.email == email))).scalar_one_or_none()
+        outbox = (await session.execute(select(OutboxMessage))).scalars().all()
+    assert len(outbox) == (1 if durable and not fail_commit else 0)
+    if fail_commit:
+        assert captured_mail.messages == []
+        assert observed == []
+    else:
+        assert user is not None
+        assert len(observed) == 1
+        assert observed[0].user_id == user.id
+        assert observed[0].email == email
+        if operation == "invite":
+            assert any("/users/invite/accept?token=" in m.text for m in captured_mail.to(email))
+
+
+async def test_admin_invite_existing_email_is_a_conflict(admin_client, captured_mail):
+    body = {"email": "dup-invite@example.com", "role_names": []}
+    first = await admin_client.post("/api/users/admin/invite", json=body)
+    assert first.status_code == 201, first.text
+    sent = len(captured_mail.messages)
+    again = await admin_client.post("/api/users/admin/invite", json=body)
+    assert again.status_code == 409, again.text
+    assert len(captured_mail.messages) == sent

modules/users/tests/test_users_reset_credentials.py

@@ -0,0 +1,118 @@
+"""A mailbox password reset strands every previously issued credential (#686)."""
+
+from urllib.parse import parse_qs, urlsplit
+
+import httpx
+import pytest
+from sqlmodel import select
+from users.models import RefreshToken, UserAccessToken
+from users.session_revocation import CHANNEL
+
+
+@pytest.mark.parametrize("credential", ["refresh", "session", "access"])
+async def test_reset_revokes_old_credentials(admin_client, anon_client, users_app, credential):
+    login = await anon_client.post(
+        "/api/users/auth/login",
+        data={"username": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert login.status_code == 204, login.text
+    pair = await anon_client.post(
+        "/api/users/auth/token",
+        json={"email": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert pair.status_code == 200, pair.text
+    tokens = pair.json()
+    me = await anon_client.get("/api/users/me")
+    assert me.status_code == 200, me.text
+    # Warm the session-version cache before resetting in a separate client.
+    assert (await anon_client.get("/admin/users/")).status_code == 200
+    link = await admin_client.post(f"/api/users/admin/{me.json()['id']}/reset-password-link")
+    assert link.status_code == 200, link.text
+    token = parse_qs(urlsplit(link.json()["link"]).query)["token"][0]
+    invalidations = []
+    users_app.state.sm.invalidation.subscribe(CHANNEL, invalidations.append)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=users_app), base_url="http://testserver"
+    ) as fresh:
+        reset = await fresh.post(
+            "/api/users/auth/reset-password",
+            json={"token": token, "password": "NewCompliantPass1!"},
+        )
+        assert reset.status_code == 200, reset.text
+        assert [message.key for message in invalidations] == [me.json()["id"]]
+        async with users_app.state.sm.db.session_factory() as session:
+            assert (await session.execute(select(UserAccessToken))).scalars().all() == []
+            refresh_rows = (await session.execute(select(RefreshToken))).scalars().all()
+            assert refresh_rows and all(row.revoked_at is not None for row in refresh_rows)
+        if credential == "refresh":
+            response = await fresh.post(
+                "/api/users/auth/token/refresh", json={"refresh_token": tokens["refresh_token"]}
+            )
+            assert response.status_code == 401, response.text
+        elif credential == "access":
+            response = await fresh.get(
+                "/api/users/me", headers={"Authorization": f"Bearer {tokens['access_token']}"}
+            )
+            assert response.status_code == 401, response.text
+        else:
+            response = await anon_client.get("/admin/users/", follow_redirects=False)
+            assert response.status_code == 302, response.text
+            assert "/users/login" in response.headers["location"]
+
+
+async def test_reset_revokes_in_the_password_commit(
+    admin_client, anon_client, users_app, monkeypatch
+):
+    """A failure after the password is stored must not leave old credentials valid."""
+    from users.manager import UserManager
+
+    pair = await anon_client.post(
+        "/api/users/auth/token",
+        json={"email": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert pair.status_code == 200, pair.text
+    me = await anon_client.get(
+        "/api/users/me", headers={"Authorization": f"Bearer {pair.json()['access_token']}"}
+    )
+    link = await admin_client.post(f"/api/users/admin/{me.json()['id']}/reset-password-link")
+    token = parse_qs(urlsplit(link.json()["link"]).query)["token"][0]
+
+    async def explode(self, user, request=None):
+        raise RuntimeError("after-reset hook failed")
+
+    monkeypatch.setattr(UserManager, "on_after_reset_password", explode)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=users_app, raise_app_exceptions=False),
+        base_url="http://testserver",
+    ) as fresh:
+        reset = await fresh.post(
+            "/api/users/auth/reset-password",
+            json={"token": token, "password": "NewCompliantPass1!"},
+        )
+        assert reset.status_code == 500
+        refreshed = await fresh.post(
+            "/api/users/auth/token/refresh",
+            json={"refresh_token": pair.json()["refresh_token"]},
+        )
+        assert refreshed.status_code == 401, refreshed.text
+    async with users_app.state.sm.db.session_factory() as session:
+        assert (await session.execute(select(UserAccessToken))).scalars().all() == []
+
+
+async def test_rejected_reset_password_revokes_nothing(admin_client, anon_client, users_app):
+    login = await anon_client.post(
+        "/api/users/auth/login",
+        data={"username": "admin@example.com", "password": "AdminPass1!"},
+    )
+    assert login.status_code == 204, login.text
+    me = await anon_client.get("/api/users/me")
+    link = await admin_client.post(f"/api/users/admin/{me.json()['id']}/reset-password-link")
+    token = parse_qs(urlsplit(link.json()["link"]).query)["token"][0]
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=users_app), base_url="http://testserver"
+    ) as fresh:
+        reset = await fresh.post(
+            "/api/users/auth/reset-password", json={"token": token, "password": "12345678"}
+        )
+        assert reset.status_code == 400, reset.text
+    assert (await anon_client.get("/admin/users/", follow_redirects=False)).status_code == 200

modules/users/users/admin/api.py

@@ -8,6 +8,7 @@
 from fastapi import status as http_status
 from fastapi_users import exceptions as fa_exceptions
 from simple_module_core.events import EventBus
+from simple_module_db import RequestSession
 from simple_module_db.deps import get_db
 from simple_module_db.pagination import PageParams
 from simple_module_hosting.pagination import page_params
@@ -42,6 +43,8 @@
 from users.invites.bulk import bulk_router
 from users.invites.resend import resend_router
 
+_DUPLICATE_EMAIL = "A user with this email already exists."
+
 admin_router = APIRouter(
     prefix="/admin",
     dependencies=[Depends(RequiresPermission(PERM_USERS_MANAGE))],
@@ -91,20 +94,26 @@ async def admin_invite_user(
     bus: EventBus = Depends(get_event_bus),
     service: UserService = Depends(get_user_service),
     mailer=Depends(get_mailer),
+    db: RequestSession = Depends(get_db),
 ):
     """Invite a new user by email, optionally assigning roles."""
     invited_by = getattr(request.state, "user", None)
     invited_by_name = invited_by.name if invited_by else "Administrator"
-    user, token = await service.invite(
-        data.email, data.full_name, data.role_names, invited_by=invited_by
-    )
-    await mailer.send_invite(user.email, token, invited_by_name)
-    await bus.publish(
+    try:
+        user, token = await service.invite(
+            data.email, data.full_name, data.role_names, invited_by=invited_by
+        )
+    except fa_exceptions.UserAlreadyExists:
+        raise HTTPException(status_code=409, detail=_DUPLICATE_EMAIL) from None
+    db.on_commit(lambda: mailer.send_invite(user.email, token, invited_by_name))
+    await publish_admin_event(
+        bus,
         UserInvited(
             user_id=user.id,
             email=user.email,
             invited_by=(str(invited_by.id) if invited_by else None),
-        )
+        ),
+        db,
     )
     return service.to_list_item(user)
 
@@ -119,6 +128,7 @@ async def admin_create_user(
     request: Request,
     bus: EventBus = Depends(get_event_bus),
     service: UserService = Depends(get_user_service),
+    db: AsyncSession = Depends(get_db),
 ):
     """Create an active+verified user with an admin-set password."""
     creator = getattr(request.state, "user", None)
@@ -132,13 +142,12 @@ async def admin_create_user(
             created_by=created_by,
         )
     except fa_exceptions.UserAlreadyExists:
-        raise HTTPException(
-            status_code=409,
-            detail="A user with this email already exists.",
-        ) from None
+        raise HTTPException(status_code=409, detail=_DUPLICATE_EMAIL) from None
     except fa_exceptions.InvalidPasswordException as exc:
         raise HTTPException(status_code=400, detail=exc.reason) from None
-    await bus.publish(UserCreated(user_id=user.id, email=user.email, created_by=created_by))
+    await publish_admin_event(
+        bus, UserCreated(user_id=user.id, email=user.email, created_by=created_by), db
+    )
     return service.to_list_item(user)
 
 
@@ -154,10 +163,7 @@ async def admin_update_user(
     except UserNotFoundError:
         raise HTTPException(status_code=404, detail="User not found") from None
     except EmailAlreadyExistsError:
-        raise HTTPException(
-            status_code=409,
-            detail="A user with this email already exists.",
-        ) from None
+        raise HTTPException(status_code=409, detail=_DUPLICATE_EMAIL) from None
     return service.to_list_item(user)