From 229db525379cb5ff5474357ca7ea74f59d15e0c0 Mon Sep 17 00:00:00 2001 From: HgO Date: Wed, 26 Oct 2022 15:53:45 +0200 Subject: [PATCH] add tests for key verification callbacks --- matrix_alertbot/callback.py | 4 +- matrix_alertbot/main.py | 2 +- tests/test_callback.py | 261 +++++++++++++++++++++++++++++++++++- tests/test_command.py | 2 +- tests/utils.py | 2 +- 5 files changed, 265 insertions(+), 6 deletions(-) diff --git a/matrix_alertbot/callback.py b/matrix_alertbot/callback.py index 5c1f8a2..e04456a 100644 --- a/matrix_alertbot/callback.py +++ b/matrix_alertbot/callback.py @@ -303,7 +303,7 @@ class Callbacks: f"Verification has been cancelled by {event.sender} for reason: {event.reason}." ) - async def key_verification_accept(self, event: KeyVerificationKey): + async def key_verification_confirm(self, event: KeyVerificationKey): sas = self.matrix_client.key_verifications[event.transaction_id] emoji_list = sas.get_emoji() emoji_str = " ".join(emoji for emoji, alt_text in emoji_list) @@ -343,7 +343,7 @@ class Callbacks: todevice_msg = sas.get_mac() except LocalProtocolError as e: # e.g. it might have been cancelled by ourselves - logger.warn(f"Unable to conclude verification with {event.sender}: {e}.") + logger.warning(f"Unable to conclude verification with {event.sender}: {e}.") return event_response = await self.matrix_client.to_device(todevice_msg) diff --git a/matrix_alertbot/main.py b/matrix_alertbot/main.py index bff3a36..88ae052 100644 --- a/matrix_alertbot/main.py +++ b/matrix_alertbot/main.py @@ -168,7 +168,7 @@ def main() -> None: callbacks.key_verification_cancel, (KeyVerificationCancel,) ) matrix_client.add_to_device_callback( - callbacks.key_verification_accept, (KeyVerificationKey,) + callbacks.key_verification_confirm, (KeyVerificationKey,) ) matrix_client.add_to_device_callback( callbacks.key_verification_end, (KeyVerificationMac,) diff --git a/tests/test_callback.py b/tests/test_callback.py index 3f67fe2..702767f 100644 --- a/tests/test_callback.py +++ b/tests/test_callback.py @@ -3,6 +3,7 @@ from typing import Dict from unittest.mock import MagicMock, Mock, patch import nio +import nio.crypto from diskcache import Cache import matrix_alertbot.callback @@ -14,6 +15,10 @@ from matrix_alertbot.command import BaseCommand from tests.utils import make_awaitable +def key_verification_get_mac_raise_protocol_error(): + raise nio.LocalProtocolError + + class CallbacksTestCase(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: # Create a Callbacks object and give it some Mock'd objects to use @@ -48,7 +53,7 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase): fake_invite_event.sender = "@some_other_fake_user:example.com" # Pretend that attempting to join a room is always successful - self.fake_matrix_client.join.return_value = make_awaitable(None) + self.fake_matrix_client.join.return_value = make_awaitable() # Pretend that we received an invite event await self.callbacks.invite(self.fake_room, fake_invite_event) @@ -540,6 +545,260 @@ class CallbacksTestCase(unittest.IsolatedAsyncioTestCase): fake_command.assert_not_called() self.fake_cache.__getitem__.assert_not_called() + async def test_key_verification_start(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.from_device = "ABCDEFGH" + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.short_authentication_string = ["emoji"] + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.accept_key_verification.return_value = make_awaitable() + self.fake_matrix_client.to_device.return_value = make_awaitable() + + fake_sas = Mock() + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_start(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.accept_key_verification.assert_called_once_with( + fake_transaction_id + ) + self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.share_key()) + + async def test_key_verification_start_with_emoji_not_supported(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.from_device = "ABCDEFGH" + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.short_authentication_string = [] + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.accept_key_verification.return_value = make_awaitable() + self.fake_matrix_client.to_device.return_value = make_awaitable() + + fake_sas = Mock() + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_start(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.accept_key_verification.assert_not_called() + self.fake_matrix_client.to_device.assert_not_called() + + async def test_key_verification_start_with_accept_key_verification_error( + self, + ) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.from_device = "ABCDEFGH" + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.short_authentication_string = ["emoji"] + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.accept_key_verification.return_value = make_awaitable( + Mock(spec=nio.ToDeviceError) + ) + self.fake_matrix_client.to_device.return_value = make_awaitable() + + fake_sas = Mock() + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_start(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.accept_key_verification.assert_called_once_with( + fake_transaction_id + ) + self.fake_matrix_client.to_device.assert_not_called() + + async def test_key_verification_start_with_to_device_error( + self, + ) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.from_device = "ABCDEFGH" + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.short_authentication_string = ["emoji"] + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.accept_key_verification.return_value = make_awaitable() + self.fake_matrix_client.to_device.return_value = make_awaitable( + Mock(spec=nio.ToDeviceError) + ) + + fake_sas = Mock() + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_start(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.accept_key_verification.assert_called_once_with( + fake_transaction_id + ) + self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.share_key()) + + async def test_key_verification_cancel(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_key_verification_event = Mock(spec=nio.KeyVerificationCancel) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.reason = "fake reason" + + # Pretend that we received a text message event + await self.callbacks.key_verification_cancel(fake_key_verification_event) + + # Check that we attempted to execute the command + + async def test_key_verification_confirm(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.confirm_short_auth_string.return_value = ( + make_awaitable() + ) + + fake_sas = Mock() + fake_sas.get_emoji.return_value = [ + ("emoji1", "alt text1"), + ("emoji2", "alt text2"), + ] + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_confirm(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.confirm_short_auth_string.assert_called_once_with( + fake_transaction_id + ) + + async def test_key_verification_confirm_with_error(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.confirm_short_auth_string.return_value = make_awaitable( + Mock(spec=nio.ToDeviceError) + ) + + fake_sas = Mock() + fake_sas.get_emoji.return_value = [ + ("emoji1", "alt text1"), + ("emoji2", "alt text2"), + ] + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_confirm(fake_key_verification_event) + + # Check that we attempted to execute the command + self.fake_matrix_client.confirm_short_auth_string.assert_called_once_with( + fake_transaction_id + ) + + async def test_key_verification_end(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.to_device.return_value = make_awaitable() + + fake_sas = Mock() + fake_sas.verified_devices = ["HGFEDCBA"] + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_end(fake_key_verification_event) + + # Check that we attempted to execute the command + fake_sas.get_mac.assert_called_once_with() + self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.get_mac()) + + async def test_key_verification_end_with_mac_error(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.to_device.return_value = make_awaitable() + + fake_sas = Mock() + fake_sas.get_mac.side_effect = key_verification_get_mac_raise_protocol_error + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_end(fake_key_verification_event) + + # Check that we attempted to execute the command + fake_sas.get_mac.assert_called_once_with() + self.fake_matrix_client.to_device.assert_not_called() + + async def test_key_verification_end_with_to_device_error(self) -> None: + """Tests the callback for RoomMessageText with the command prefix""" + # Tests that the bot process messages in the room that contain a command + fake_transaction_id = "fake transaction id" + + fake_key_verification_event = Mock(spec=nio.KeyVerificationStart) + fake_key_verification_event.sender = "@some_other_fake_user:example.com" + fake_key_verification_event.transaction_id = fake_transaction_id + + self.fake_matrix_client.to_device.return_value = make_awaitable( + Mock(spec=nio.ToDeviceError) + ) + + fake_sas = Mock() + fake_transactions_dict = {fake_transaction_id: fake_sas} + self.fake_matrix_client.key_verifications = fake_transactions_dict + + # Pretend that we received a text message event + await self.callbacks.key_verification_end(fake_key_verification_event) + + # Check that we attempted to execute the command + fake_sas.get_mac.assert_called_once_with() + self.fake_matrix_client.to_device.assert_called_once_with(fake_sas.get_mac()) + @patch.object(matrix_alertbot.callback.CommandFactory, "create", autospec=True) async def test_unknown(self, fake_command_create: Mock) -> None: """Tests the callback for RoomMessageText with the command prefix""" diff --git a/tests/test_command.py b/tests/test_command.py index b7c3477..a298ebe 100644 --- a/tests/test_command.py +++ b/tests/test_command.py @@ -85,7 +85,7 @@ class CommandTestCase(unittest.IsolatedAsyncioTestCase): self.fake_matrix_client = Mock(spec=nio.AsyncClient) self.fake_matrix_client.user = "@fake_user:example.com" # Pretend that attempting to send a message is always successful - self.fake_matrix_client.room_send.return_value = make_awaitable(None) + self.fake_matrix_client.room_send.return_value = make_awaitable() self.fake_cache = MagicMock(spec=Cache) self.fake_cache.__getitem__.side_effect = cache_get_item diff --git a/tests/utils.py b/tests/utils.py index 3fcf429..202d0e8 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -11,7 +11,7 @@ def run_coroutine(result: Awaitable[Any]) -> Any: return result -def make_awaitable(result: Any) -> Awaitable[Any]: +def make_awaitable(result: Any = None) -> Awaitable[Any]: """ Makes an awaitable, suitable for mocking an `async` function. This uses Futures as they can be awaited multiple times so can be returned