fix(auth): fix passkey deletion password verification

app/api/endpoints/mfa.py

@@ -34,7 +34,6 @@ from app.application.security.passkey import (
     PassKeyRegistrationVerificationError,
     PasskeyService,
 )
-from app.application.security.token import verify_password
 from app.application.security.user import (
     UserService,
     get_configured_user_id_lookup,
@@ -185,7 +184,7 @@ async def otp_disable(
 ) -> Any:
     """关闭当前用户的 OTP 验证功能"""
     # 验证密码
-    if not verify_password(data.password, str(current_user.hashed_password)):
+    if not await service.verify_password(current_user.id, data.password):
         return _SchemaResponse(success=False, message="密码错误")
     await service.update_otp(current_user.name, False, "")
     return _SchemaResponse(success=True)
@@ -508,13 +507,12 @@ async def passkey_delete(
     data: PassKeyDeleteRequest,
     current_user: ApiPrincipal = Depends(get_current_active_user_async),
     service: PasskeyService = Depends(get_passkey_service),
+    user_service: UserService = Depends(get_user_service),
 ) -> Any:
     """删除指定的 PassKey"""
     try:
         # 验证密码
-        if not verify_password(
+        if not await user_service.verify_password(current_user.id, data.password):
-            data.password, str(current_user.hashed_password)
-        ):
             return _SchemaResponse(success=False, message="密码错误")
 
         success = service.delete_by_id(data.passkey_id, current_user.id)

app/application/security/user.py

@@ -9,6 +9,8 @@ from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Any, Optional, Protocol, TypeAlias, TypeVar, Union, cast
 
+from app.application.security.token import verify_password as _verify_password
+
 FrozenJson: TypeAlias = Union[
     str,
     int,
@@ -195,6 +197,9 @@ class UserRepository(Protocol):
     async def async_get_by_id(self, user_id: int) -> Optional[UserSnapshot]:
         """按用户 ID 返回用户。"""
 
+    async def async_get_auth_by_id(self, user_id: int) -> Optional[UserAuthSnapshot]:
+        """按用户 ID 返回密码校验所需的认证快照。"""
+
     async def async_create(
         self,
         payload: dict[str, Any],
@@ -283,6 +288,16 @@ class UserService:
         """按用户 ID 查询用户。"""
         return await self._repository.async_get_by_id(user_id)
 
+    async def verify_password(self, user_id: int, password: str) -> bool:
+        """按用户 ID 读取认证快照并校验密码。"""
+        user = await self._repository.async_get_auth_by_id(user_id)
+        return bool(
+            user
+            and user.user.is_active
+            and user.hashed_password
+            and _verify_password(password, user.hashed_password)
+        )
+
     async def create(self, payload: dict[str, Any]) -> Optional[UserSnapshot]:
         """创建用户。"""
         return await self._write(lambda: self._repository.async_create(payload))

app/db/adapters/user.py

@@ -115,6 +115,14 @@ class SqlAlchemyUserRepository(UserRepository):
         model = await self._oper.async_get_by_id(user_id)
         return _to_snapshot(model) if model else None
 
+    async def async_get_auth_by_id(
+        self,
+        user_id: int,
+    ) -> Optional[UserAuthSnapshot]:
+        """在异步请求会话中按 ID 读取密码校验所需的认证快照。"""
+        model = await self._oper.async_get_by_id(user_id)
+        return _to_auth_snapshot(model) if model else None
+
     async def async_create(
         self,
         payload: dict[str, Any],

tests/test_mfa_passkey_transactions.py

@@ -1,5 +1,5 @@
 from types import SimpleNamespace
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from fastapi import HTTPException
@@ -309,3 +309,42 @@ def test_authentication_finish_does_not_issue_token_when_sign_count_write_fails(
     )
     auth_service.build_token_response.assert_not_called()
     set_cookie.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_passkey_delete_verifies_password_without_public_password_field():
+    """删除 PassKey 应通过认证服务校验密码，而不是读取公开用户快照字段。"""
+    data = mfa_endpoint.PassKeyDeleteRequest(passkey_id=10, password="password")
+    current_user = SimpleNamespace(id=7, name="user")
+    passkey_service = SimpleNamespace(delete_by_id=Mock(return_value=True))
+    user_service = SimpleNamespace(verify_password=AsyncMock(return_value=True))
+
+    result = await mfa_endpoint.passkey_delete(
+        data=data,
+        current_user=current_user,
+        service=passkey_service,
+        user_service=user_service,
+    )
+
+    assert result.success is True
+    user_service.verify_password.assert_awaited_once_with(7, "password")
+    passkey_service.delete_by_id.assert_called_once_with(10, 7)
+
+
+@pytest.mark.asyncio
+async def test_passkey_delete_rejects_invalid_password_before_deletion():
+    """密码错误时不得删除 PassKey。"""
+    data = mfa_endpoint.PassKeyDeleteRequest(passkey_id=10, password="wrong")
+    passkey_service = SimpleNamespace(delete_by_id=Mock())
+    user_service = SimpleNamespace(verify_password=AsyncMock(return_value=False))
+
+    result = await mfa_endpoint.passkey_delete(
+        data=data,
+        current_user=SimpleNamespace(id=7, name="user"),
+        service=passkey_service,
+        user_service=user_service,
+    )
+
+    assert result.success is False
+    assert result.message == "密码错误"
+    passkey_service.delete_by_id.assert_not_called()

tests/test_user_service.py

@@ -1,9 +1,11 @@
 """用户应用服务的请求级事务边界测试。"""
 
-from unittest.mock import AsyncMock, MagicMock
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from app.application.security import user as user_service_module
 from app.application.security.user import UserService, UserSnapshot, UserUpdateResult
 
 
@@ -139,3 +141,26 @@ async def test_user_service_does_not_rollback_committed_publish_failure() -> Non
 
     unit_of_work.commit.assert_awaited_once_with()
     unit_of_work.rollback.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_user_service_verifies_password_from_auth_snapshot() -> None:
+    """密码校验应读取认证快照，而不是依赖公开用户资料。"""
+    repository = MagicMock()
+    repository.async_get_auth_by_id = AsyncMock(
+        return_value=SimpleNamespace(
+            user=SimpleNamespace(is_active=True),
+            hashed_password="hashed",
+        )
+    )
+    service = UserService(repository, MagicMock(), MagicMock())
+
+    with patch.object(
+        user_service_module,
+        "_verify_password",
+        return_value=True,
+    ) as verify_password:
+        assert await service.verify_password(7, "password") is True
+
+    repository.async_get_auth_by_id.assert_awaited_once_with(7)
+    verify_password.assert_called_once_with("password", "hashed")