src/main.py

@@ -1745,8 +1745,9 @@ async def revoke_active_device(
             headers={"Cache-Control": "no-store"},
         )
     try:
-        await asyncio.to_thread(_passkey_store().revoke_management_id, management_id)
+        revoked = await asyncio.to_thread(
-        revoked = await dashboard_auth.revoke_managed_session(management_id)
+            _passkey_store().revoke_device_access, management_id
+        )
     except dashboard_auth.SessionStoreError:
         return JSONResponse(
             {"detail": "Session registry is temporarily unavailable"},
@@ -1790,8 +1791,7 @@ async def sign_out_all_devices(
             headers={"Cache-Control": "no-store"},
         )
     try:
-        await asyncio.to_thread(_passkey_store().revoke_all)
+        await asyncio.to_thread(_passkey_store().revoke_all_access)
-        await dashboard_auth.revoke_all_sessions()
     except dashboard_auth.SessionStoreError:
         return JSONResponse(
             {"detail": "Session registry is temporarily unavailable"},

src/passkey_store.py

@@ -246,3 +246,37 @@ class PasskeyStore:
                 connection.execute("DELETE FROM passkey_challenges")
         except (OSError, sqlite3.Error) as exc:
             raise SessionStoreError("Passkey registry is temporarily unavailable") from exc
+
+    def revoke_device_access(self, management_id: str) -> bool:
+        """Atomically remove one active session, its grants, and its linked passkey."""
+        try:
+            with self._connect() as connection:
+                session = connection.execute(
+                    "SELECT session_hash FROM active_sessions WHERE management_id = ?",
+                    (management_id,),
+                ).fetchone()
+                if session is None:
+                    return False
+                connection.execute(
+                    "DELETE FROM step_up_grants WHERE session_hash = ?", (session[0],)
+                )
+                connection.execute(
+                    "DELETE FROM passkey_credentials WHERE management_id = ?", (management_id,)
+                )
+                cursor = connection.execute(
+                    "DELETE FROM active_sessions WHERE management_id = ?", (management_id,)
+                )
+                return cursor.rowcount == 1
+        except (OSError, sqlite3.Error) as exc:
+            raise SessionStoreError("Session registry is temporarily unavailable") from exc
+
+    def revoke_all_access(self) -> None:
+        """Atomically remove every passkey, challenge, grant, and active session."""
+        try:
+            with self._connect() as connection:
+                connection.execute("DELETE FROM passkey_credentials")
+                connection.execute("DELETE FROM passkey_challenges")
+                connection.execute("DELETE FROM step_up_grants")
+                connection.execute("DELETE FROM active_sessions")
+        except (OSError, sqlite3.Error) as exc:
+            raise SessionStoreError("Session registry is temporarily unavailable") from exc

tests/test_access_revocation.py

@@ -0,0 +1,97 @@
+import sqlite3
+
+import pytest
+
+from src.passkey_store import PasskeyStore
+from src.session_store import SessionStore, SessionStoreError
+
+
+def test_global_access_revocation_rolls_back_every_credential_and_session_change(tmp_path):
+    database = tmp_path / "sessions.sqlite3"
+    sessions = SessionStore(database, clock=lambda: 1_000.0)
+    sessions.activate(
+        "phone-session",
+        2_000,
+        management_id="phone-management-id",
+        device_label="Phone",
+    )
+    sessions.mint_step_up(
+        "phone-session",
+        action="revoke_all_sessions",
+        target="all",
+        ttl_seconds=90,
+    )
+    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
+    passkeys.register(
+        credential_id=b"phone-credential",
+        public_key=b"phone-public-key",
+        sign_count=0,
+        device_label="Phone",
+        management_id="phone-management-id",
+    )
+    passkeys.issue_challenge(
+        b"pending-challenge",
+        session_id="phone-session",
+        purpose="authentication",
+        action="sign_in",
+        target="dashboard",
+    )
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "CREATE TRIGGER block_session_delete BEFORE DELETE ON active_sessions "
+            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
+        )
+
+    with pytest.raises(SessionStoreError):
+        passkeys.revoke_all_access()
+
+    with sqlite3.connect(database) as connection:
+        assert connection.execute("SELECT COUNT(*) FROM passkey_credentials").fetchone() == (1,)
+        assert connection.execute("SELECT COUNT(*) FROM passkey_challenges").fetchone() == (1,)
+        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (1,)
+        assert connection.execute("SELECT COUNT(*) FROM active_sessions").fetchone() == (1,)
+
+
+def test_device_access_revocation_rolls_back_target_and_preserves_other_devices(tmp_path):
+    database = tmp_path / "sessions.sqlite3"
+    sessions = SessionStore(database, clock=lambda: 1_000.0)
+    passkeys = PasskeyStore(database, clock=lambda: 1_000.0)
+    for label in ("Phone", "Laptop"):
+        slug = label.lower()
+        sessions.activate(
+            f"{slug}-session",
+            2_000,
+            management_id=f"{slug}-management-id",
+            device_label=label,
+        )
+        sessions.mint_step_up(
+            f"{slug}-session",
+            action="close_issue",
+            target="stackchain/dashboard#1",
+            ttl_seconds=90,
+        )
+        passkeys.register(
+            credential_id=f"{slug}-credential".encode(),
+            public_key=f"{slug}-public-key".encode(),
+            sign_count=0,
+            device_label=label,
+            management_id=f"{slug}-management-id",
+        )
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "CREATE TRIGGER block_phone_delete BEFORE DELETE ON active_sessions "
+            "WHEN OLD.management_id = 'phone-management-id' "
+            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
+        )
+
+    with pytest.raises(SessionStoreError):
+        passkeys.revoke_device_access("phone-management-id")
+
+    with sqlite3.connect(database) as connection:
+        assert connection.execute(
+            "SELECT management_id FROM active_sessions ORDER BY management_id"
+        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
+        assert connection.execute(
+            "SELECT management_id FROM passkey_credentials ORDER BY management_id"
+        ).fetchall() == [("laptop-management-id",), ("phone-management-id",)]
+        assert connection.execute("SELECT COUNT(*) FROM step_up_grants").fetchone() == (2,)

tests/test_dashboard_auth.py

@@ -283,6 +283,64 @@ async def test_revoking_an_enrolled_device_also_revokes_its_passkey(
     assert passkey_options.status_code == 404
 
 
+@pytest.mark.anyio
+async def test_device_revocation_failure_preserves_its_passkey_and_session(
+    access_control,
+):
+    transport = httpx.ASGITransport(app=main.app)
+    async with (
+        httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
+        httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
+    ):
+        await phone.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        await laptop.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Laptop"},
+        )
+        phone_device = next(
+            item
+            for item in (await laptop.get("/api/v1/sessions")).json()["devices"]
+            if item["device_label"] == "Phone"
+        )
+        store = main._passkey_store()
+        await asyncio.to_thread(
+            store.register,
+            credential_id=b"phone-credential",
+            public_key=b"phone-public-key",
+            sign_count=0,
+            device_label="Phone",
+            management_id=phone_device["management_id"],
+        )
+        grant = await fresh_grant(
+            laptop, "revoke_device", phone_device["management_id"]
+        )
+        with sqlite3.connect(store.path) as connection:
+            connection.execute(
+                "CREATE TRIGGER block_phone_delete BEFORE DELETE ON active_sessions "
+                f"WHEN OLD.management_id = '{phone_device['management_id']}' "
+                "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
+            )
+
+        response = await laptop.delete(
+            f"/api/v1/sessions/{phone_device['management_id']}",
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": laptop.cookies["stackchain_csrf"],
+                "X-Step-Up-Grant": grant,
+            },
+        )
+        phone_session = await phone.get("/api/v1/session")
+        passkeys = await laptop.get("/api/v1/passkeys")
+
+    assert response.status_code == 503
+    assert phone_session.status_code == 200
+    assert phone_session.json()["authenticated"] is True
+    assert [item["device_label"] for item in passkeys.json()["passkeys"]] == ["Phone"]
+
+
 @pytest.mark.anyio
 async def test_orphaned_passkey_is_listed_safely_and_can_be_selectively_revoked(
     access_control, monkeypatch
@@ -1099,16 +1157,10 @@ async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_cont
         grant = await fresh_grant(client, "revoke_all_sessions", "all")
 
         class BrokenStore:
-            def status(self, session_id, expires_at, *, idle_timeout_seconds):
+            def revoke_all_access(self):
-                return "active"
-
-            def consume_step_up(self, grant, session_id, *, action, target):
-                return True
-
-            def revoke_all(self):
                 raise SessionStoreError("database path and secret details")
 
-        monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
+        monkeypatch.setattr(main, "_passkey_store", lambda: BrokenStore())
         response = await client.delete(
             "/api/v1/sessions",
             headers={
@@ -1125,6 +1177,49 @@ async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_cont
     assert "database path" not in response.text
 
 
+@pytest.mark.anyio
+async def test_sign_out_all_devices_failure_preserves_passkeys_and_sessions(access_control):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        device = (await client.get("/api/v1/sessions")).json()["devices"][0]
+        store = main._passkey_store()
+        await asyncio.to_thread(
+            store.register,
+            credential_id=b"phone-credential",
+            public_key=b"phone-public-key",
+            sign_count=0,
+            device_label="Phone",
+            management_id=device["management_id"],
+        )
+        grant = await fresh_grant(client, "revoke_all_sessions", "all")
+        with sqlite3.connect(store.path) as connection:
+            connection.execute(
+                "CREATE TRIGGER block_session_delete BEFORE DELETE ON active_sessions "
+                "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
+            )
+
+        response = await client.delete(
+            "/api/v1/sessions",
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": client.cookies["stackchain_csrf"],
+                "X-Step-Up-Grant": grant,
+            },
+        )
+        session = await client.get("/api/v1/session")
+        passkeys = await client.get("/api/v1/passkeys")
+
+    assert response.status_code == 503
+    assert session.status_code == 200
+    assert session.json()["authenticated"] is True
+    assert passkeys.status_code == 200
+    assert [item["device_label"] for item in passkeys.json()["passkeys"]] == ["Phone"]
+
+
 @pytest.mark.anyio
 async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
     calls = 0