mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-04 22:36:34 +08:00
382 lines
12 KiB
Python
382 lines
12 KiB
Python
"""
|
|
验证通行密钥挑战事务与用户身份绑定。
|
|
"""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from starlette.requests import Request
|
|
from starlette.responses import Response
|
|
|
|
from app.api.endpoints import mfa as mfa_endpoint
|
|
from app.application.security.passkey import (
|
|
PASSKEY_CHALLENGE_TTL_SECONDS,
|
|
PasskeyChallengeStore,
|
|
configure_passkey_challenge_cache,
|
|
)
|
|
from app.runtime.cache import TTLCache
|
|
|
|
|
|
def _request() -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": "/api/v1/mfa/passkey/authenticate/finish",
|
|
"headers": [(b"host", b"testserver")],
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": ("testclient", 123),
|
|
}
|
|
)
|
|
|
|
|
|
def setup_function():
|
|
"""
|
|
为每个用例重建独立的通行密钥挑战缓存。
|
|
"""
|
|
configure_passkey_challenge_cache(TTLCache(
|
|
region="passkey_challenge",
|
|
maxsize=4096,
|
|
ttl=PASSKEY_CHALLENGE_TTL_SECONDS,
|
|
))
|
|
|
|
|
|
def test_registration_transaction_is_bound_to_current_user():
|
|
"""
|
|
注册事务绑定发起用户，其他用户不能使用该事务完成凭据注册。
|
|
"""
|
|
token = PasskeyChallengeStore.issue(
|
|
challenge="server-challenge",
|
|
purpose="registration",
|
|
user_id=1,
|
|
)
|
|
request = mfa_endpoint.PassKeyRegistrationFinish(
|
|
credential={"id": "credential-id"},
|
|
transaction_token=token,
|
|
name="test",
|
|
)
|
|
|
|
with patch.object(
|
|
mfa_endpoint.PassKeyHelper,
|
|
"verify_registration_response",
|
|
) as verify:
|
|
result = mfa_endpoint.passkey_register_finish(
|
|
passkey_req=request,
|
|
current_user=SimpleNamespace(id=2, name="other"),
|
|
)
|
|
|
|
assert not result.success
|
|
assert result.message == "注册请求已失效，请重新发起注册"
|
|
verify.assert_not_called()
|
|
|
|
|
|
def test_registration_uses_server_challenge():
|
|
"""
|
|
注册校验采用服务端保存的挑战值，忽略客户端自行提交的挑战。
|
|
"""
|
|
token = PasskeyChallengeStore.issue(
|
|
challenge="server-challenge",
|
|
purpose="registration",
|
|
user_id=1,
|
|
)
|
|
request = mfa_endpoint.PassKeyRegistrationFinish(
|
|
credential={"id": "credential-id", "challenge": "client-challenge"},
|
|
transaction_token=token,
|
|
name="test",
|
|
)
|
|
passkey = Mock()
|
|
service = SimpleNamespace(create=Mock(return_value=passkey))
|
|
|
|
with patch.object(
|
|
mfa_endpoint.PassKeyHelper,
|
|
"verify_registration_response",
|
|
return_value=("credential-id", b"public-key", 0, "aaguid"),
|
|
) as verify:
|
|
result = mfa_endpoint.passkey_register_finish(
|
|
passkey_req=request,
|
|
current_user=SimpleNamespace(id=1, name="user"),
|
|
service=service,
|
|
)
|
|
|
|
assert result.success
|
|
verify.assert_called_once_with(
|
|
credential=request.credential,
|
|
expected_challenge="server-challenge",
|
|
)
|
|
service.create.assert_called_once()
|
|
|
|
|
|
def test_authentication_transaction_rejects_other_user_credential():
|
|
"""
|
|
认证事务拒绝使用其他用户持有的凭据，避免跨账户完成认证。
|
|
"""
|
|
token = PasskeyChallengeStore.issue(
|
|
challenge="server-challenge",
|
|
purpose="authentication",
|
|
user_id=1,
|
|
)
|
|
request = mfa_endpoint.PassKeyAuthenticationFinish(
|
|
credential={"id": "credential-id"},
|
|
transaction_token=token,
|
|
)
|
|
passkey = SimpleNamespace(user_id=2)
|
|
user = SimpleNamespace(id=2, is_active=True)
|
|
|
|
service = SimpleNamespace(get_by_credential_id=Mock(return_value=passkey))
|
|
lookup = Mock(return_value=user)
|
|
with patch.object(
|
|
mfa_endpoint,
|
|
"_extract_and_standardize_credential_id",
|
|
return_value="credential-id",
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_user_name_lookup",
|
|
return_value=lookup,
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"_verify_passkey_and_update",
|
|
) as verify:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
mfa_endpoint.passkey_authenticate_finish(
|
|
request=_request(),
|
|
response=Response(),
|
|
passkey_req=request,
|
|
service=service,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
verify.assert_not_called()
|
|
|
|
|
|
def test_authentication_finish_token_cannot_be_replayed():
|
|
"""
|
|
认证完成后消费事务令牌，再次提交同一令牌必须失败。
|
|
"""
|
|
token = PasskeyChallengeStore.issue(
|
|
challenge="server-challenge",
|
|
purpose="authentication",
|
|
user_id=1,
|
|
)
|
|
request = mfa_endpoint.PassKeyAuthenticationFinish(
|
|
credential={"id": "credential-id"},
|
|
transaction_token=token,
|
|
)
|
|
passkey = SimpleNamespace(user_id=1)
|
|
user = SimpleNamespace(
|
|
id=1,
|
|
name="user",
|
|
is_active=True,
|
|
is_superuser=False,
|
|
avatar="",
|
|
permissions={},
|
|
)
|
|
|
|
service = SimpleNamespace(get_by_credential_id=Mock(return_value=passkey))
|
|
lookup = Mock(return_value=user)
|
|
token_response = SimpleNamespace(access_token="access-token", level=1)
|
|
with patch.object(
|
|
mfa_endpoint,
|
|
"_extract_and_standardize_credential_id",
|
|
return_value="credential-id",
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_user_id_lookup",
|
|
return_value=lookup,
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"_verify_passkey_and_update",
|
|
return_value=(True, 0),
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_auth_service",
|
|
return_value=SimpleNamespace(build_token_response=Mock(return_value=token_response)),
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"set_or_refresh_resource_token_cookie",
|
|
):
|
|
result = mfa_endpoint.passkey_authenticate_finish(
|
|
request=_request(),
|
|
response=Response(),
|
|
passkey_req=request,
|
|
service=service,
|
|
)
|
|
with pytest.raises(HTTPException) as replay_error:
|
|
mfa_endpoint.passkey_authenticate_finish(
|
|
request=_request(),
|
|
response=Response(),
|
|
passkey_req=request,
|
|
service=service,
|
|
)
|
|
|
|
assert result.access_token == "access-token"
|
|
assert replay_error.value.status_code == 401
|
|
assert replay_error.value.detail == "认证请求已失效"
|
|
|
|
|
|
def test_authentication_finish_fails_closed_when_challenge_backend_fails(
|
|
monkeypatch,
|
|
):
|
|
"""
|
|
challenge 后端故障不得继续查询凭证或签发 Token。
|
|
"""
|
|
class FailingCache:
|
|
"""
|
|
模拟 challenge 原子领取期间缓存不可用。
|
|
"""
|
|
|
|
def consume(self, _key):
|
|
"""
|
|
报告后端不可用。
|
|
"""
|
|
raise RuntimeError("cache unavailable")
|
|
|
|
monkeypatch.setattr(PasskeyChallengeStore, "_cache", FailingCache())
|
|
service = SimpleNamespace(get_by_credential_id=Mock())
|
|
auth_service = SimpleNamespace(build_token_response=Mock())
|
|
|
|
with patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_auth_service",
|
|
return_value=auth_service,
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"set_or_refresh_resource_token_cookie",
|
|
) as set_cookie, pytest.raises(HTTPException) as exc_info:
|
|
mfa_endpoint.passkey_authenticate_finish(
|
|
request=_request(),
|
|
response=Response(),
|
|
passkey_req=mfa_endpoint.PassKeyAuthenticationFinish(
|
|
credential={"id": "credential-id"},
|
|
transaction_token="transaction-token",
|
|
),
|
|
service=service,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
service.get_by_credential_id.assert_not_called()
|
|
auth_service.build_token_response.assert_not_called()
|
|
set_cookie.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize("cas_failure", [False, RuntimeError("write failed")])
|
|
def test_authentication_finish_does_not_issue_token_when_sign_count_write_fails(
|
|
cas_failure,
|
|
):
|
|
"""
|
|
签名计数 CAS 冲突或事务失败时，认证不得越过持久化门禁。
|
|
"""
|
|
token = PasskeyChallengeStore.issue(
|
|
challenge="server-challenge",
|
|
purpose="authentication",
|
|
user_id=1,
|
|
)
|
|
passkey_req = mfa_endpoint.PassKeyAuthenticationFinish(
|
|
credential={"id": "credential-id"},
|
|
transaction_token=token,
|
|
)
|
|
passkey = SimpleNamespace(
|
|
id=10,
|
|
user_id=1,
|
|
public_key="public-key",
|
|
sign_count=5,
|
|
)
|
|
user = SimpleNamespace(
|
|
id=1,
|
|
name="user",
|
|
is_active=True,
|
|
is_superuser=False,
|
|
)
|
|
compare_and_update = Mock()
|
|
if isinstance(cas_failure, Exception):
|
|
compare_and_update.side_effect = cas_failure
|
|
else:
|
|
compare_and_update.return_value = cas_failure
|
|
service = SimpleNamespace(
|
|
get_by_credential_id=Mock(return_value=passkey),
|
|
compare_and_update_sign_count=compare_and_update,
|
|
)
|
|
auth_service = SimpleNamespace(build_token_response=Mock())
|
|
|
|
with patch.object(
|
|
mfa_endpoint,
|
|
"_extract_and_standardize_credential_id",
|
|
return_value="credential-id",
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_user_id_lookup",
|
|
return_value=Mock(return_value=user),
|
|
), patch.object(
|
|
mfa_endpoint.PassKeyHelper,
|
|
"verify_authentication_response",
|
|
return_value=(True, 6),
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"get_configured_auth_service",
|
|
return_value=auth_service,
|
|
), patch.object(
|
|
mfa_endpoint,
|
|
"set_or_refresh_resource_token_cookie",
|
|
) as set_cookie:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
mfa_endpoint.passkey_authenticate_finish(
|
|
request=_request(),
|
|
response=Response(),
|
|
passkey_req=passkey_req,
|
|
service=service,
|
|
)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
compare_and_update.assert_called_once_with(
|
|
passkey_id=10,
|
|
expected_sign_count=5,
|
|
sign_count=6,
|
|
)
|
|
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()
|