fix: detect atomic passkey counter anomalies (Closes #641)

src/main.py

@@ -1595,9 +1595,19 @@ async def verify_passkey_authorization(
             stored=stored,
         )
         updated = await asyncio.to_thread(
-            store.update_counter, stored.credential_id, verified.new_sign_count
+            store.advance_counter,
+            stored.credential_id,
+            expected=stored.sign_count,
+            new=verified.new_sign_count,
         )
         if not updated:
+            await asyncio.to_thread(
+                _security_event_store().record,
+                "passkey_counter_anomaly",
+                method="passkey",
+                device_label=stored.device_label,
+                target=f"{payload.action}:{payload.target}",
+            )
             raise ValueError("stale passkey counter")
         grant = await dashboard_auth.issue_step_up(
             request.state.dashboard_session,
@@ -1686,9 +1696,19 @@ async def verify_passkey_authentication(
             stored=stored,
         )
         updated = await asyncio.to_thread(
-            store.update_counter, stored.credential_id, verified.new_sign_count
+            store.advance_counter,
+            stored.credential_id,
+            expected=stored.sign_count,
+            new=verified.new_sign_count,
         )
         if not updated:
+            await asyncio.to_thread(
+                _security_event_store().record,
+                "passkey_counter_anomaly",
+                method="passkey",
+                device_label=stored.device_label,
+                target="sign_in:dashboard",
+            )
             raise ValueError("stale passkey counter")
         await dashboard_auth.revoke_managed_session(stored.management_id)
         signed, session = await asyncio.to_thread(

src/passkey_store.py

@@ -225,13 +225,26 @@ class PasskeyStore:
             raise SessionStoreError("Passkey registry is temporarily unavailable") from exc
         return StoredPasskey(*row) if row else None
 
-    def update_counter(self, credential_id: bytes, new_sign_count: int) -> bool:
+    def advance_counter(
+        self, credential_id: bytes, *, expected: int, new: int
+    ) -> bool:
+        if expected == 0 and new == 0:
+            comparison = "sign_count = 0"
+        elif new > expected:
+            comparison = "sign_count = ?"
+        else:
+            return False
         try:
             with self._connect() as connection:
+                parameters = (
+                    (new, credential_id)
+                    if expected == 0 and new == 0
+                    else (new, credential_id, expected)
+                )
                 cursor = connection.execute(
-                    "UPDATE passkey_credentials SET sign_count = ? "
-                    "WHERE credential_id = ? AND sign_count <= ?",
-                    (new_sign_count, credential_id, new_sign_count),
+                    f"UPDATE passkey_credentials SET sign_count = ? "
+                    f"WHERE credential_id = ? AND {comparison}",
+                    parameters,
                 )
                 return cursor.rowcount == 1
         except (OSError, sqlite3.Error) as exc: