Merge branch '1-impossible-de-dechiffrer-les-messages' into 'master'
Resolve "Impossible de déchiffrer les messages" Closes #1 See merge request Neutrinet/matrix-alertbot!1
This commit is contained in:
HgO
commit
49005c3468
11 changed files with 455 additions and 20 deletions
9
.gitignore
vendored
9
.gitignore
vendored
|
|
@ -12,6 +12,7 @@ env3/
|
|||
*.db
|
||||
store/
|
||||
cache/
|
||||
token.json
|
||||
|
||||
# Config file
|
||||
config.yaml
|
||||
|
|
@ -28,8 +29,10 @@ dist/
|
|||
.mypy_cache/
|
||||
_version.py
|
||||
|
||||
# Config file
|
||||
config.yaml
|
||||
|
||||
# Log files
|
||||
*.log
|
||||
|
||||
# Code coverage
|
||||
.coverage
|
||||
coverage.xml
|
||||
coverage.lcov
|
||||
|
|
|
|||
|
|
@ -9,22 +9,34 @@ command_prefix: "!alert"
|
|||
matrix:
|
||||
# The Matrix User ID of the bot account
|
||||
user_id: "@bot:matrix.example.com"
|
||||
|
||||
# Matrix account password (optional if access token used)
|
||||
user_password: "password"
|
||||
|
||||
# Matrix account access token (optional if password used)
|
||||
# If not set, the server will provide an access token after log in,
|
||||
# which will be stored in the user token file (see below)
|
||||
#user_token: ""
|
||||
|
||||
# Path to the file where to store the user access token
|
||||
user_token_file: "token.json"
|
||||
|
||||
# The URL of the homeserver to connect to
|
||||
url: https://matrix.example.com
|
||||
|
||||
# The device ID that is **non pre-existing** device
|
||||
# If this device ID already exists, messages will be dropped silently in encrypted rooms
|
||||
# If not set the server will provide a device ID after log in. Note that this ID
|
||||
# will change each time the bot reconnects.
|
||||
# device_id: ABCDEFGHIJ
|
||||
|
||||
# What to name the logged in device
|
||||
device_name: matrix-alertbot
|
||||
|
||||
# List of rooms where the bot can interact
|
||||
allowed_rooms:
|
||||
- "!abcdefgh:matrix.example.com"
|
||||
|
||||
# List of allowed reactions to create silences.
|
||||
# Default is listed here.
|
||||
allowed_reactions: [🤫, 😶, 🤐, 🙊, 🔇, 🔕]
|
||||
|
|
@ -52,8 +64,10 @@ storage:
|
|||
|
||||
template:
|
||||
# Path to directory that contains templates for rendering alerts.
|
||||
# The directory must contains the files "alert.html.j2" and "alert.txt.j2" and must respect Jinja2 templating format.
|
||||
# Default is to use templates provided by the matrix_alertbot package. These templates are available in "matrix_alertbot/resources/templates".
|
||||
# The directory must contains the files "alert.html.j2" and "alert.txt.j2"
|
||||
# and must respect Jinja2 templating format.
|
||||
# Default is to use templates provided by the matrix_alertbot package.
|
||||
# These templates are available in "matrix_alertbot/resources/templates".
|
||||
path: "data/templates"
|
||||
|
||||
# Logging setup
|
||||
|
|
|
|||
|
|
@ -5,6 +5,10 @@ from nio import (
|
|||
AsyncClient,
|
||||
InviteMemberEvent,
|
||||
JoinError,
|
||||
KeyVerificationCancel,
|
||||
KeyVerificationKey,
|
||||
KeyVerificationMac,
|
||||
KeyVerificationStart,
|
||||
LocalProtocolError,
|
||||
MatrixRoom,
|
||||
MegolmEvent,
|
||||
|
|
@ -12,6 +16,7 @@ from nio import (
|
|||
RoomGetEventError,
|
||||
RoomMessageText,
|
||||
SendRetryError,
|
||||
ToDeviceError,
|
||||
UnknownEvent,
|
||||
)
|
||||
|
||||
|
|
@ -262,6 +267,97 @@ class Callbacks:
|
|||
f"commands a second time)."
|
||||
)
|
||||
|
||||
async def key_verification_start(self, event: KeyVerificationStart):
|
||||
"""Callback for when somebody wants to verify our devices."""
|
||||
if "emoji" not in event.short_authentication_string:
|
||||
logger.error(
|
||||
f"Unable to use emoji verification with {event.sender} on device {event.from_device}."
|
||||
)
|
||||
return
|
||||
|
||||
event_response = await self.matrix_client.accept_key_verification(
|
||||
event.transaction_id
|
||||
)
|
||||
if isinstance(event_response, ToDeviceError):
|
||||
logger.error(
|
||||
f"Unable to start key verification with {event.sender} on device {event.from_device}, got error: {event_response}."
|
||||
)
|
||||
return
|
||||
|
||||
sas = self.matrix_client.key_verifications[event.transaction_id]
|
||||
|
||||
todevice_msg = sas.share_key()
|
||||
event_response = await self.matrix_client.to_device(todevice_msg)
|
||||
if isinstance(event_response, ToDeviceError):
|
||||
logger.error(
|
||||
f"Unable to share key with {event.sender} on device {event.from_device}, got error: {event_response}."
|
||||
)
|
||||
return
|
||||
|
||||
async def key_verification_cancel(self, event: KeyVerificationCancel):
|
||||
# There is no need to issue a
|
||||
# client.cancel_key_verification(tx_id, reject=False)
|
||||
# here. The SAS flow is already cancelled.
|
||||
# We only need to inform the user.
|
||||
logger.info(
|
||||
f"Verification has been cancelled by {event.sender} for reason: {event.reason}."
|
||||
)
|
||||
|
||||
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)
|
||||
|
||||
logger.info(
|
||||
f"Received request to verify emojis from {event.sender}: {emoji_str}"
|
||||
)
|
||||
|
||||
event_response = await self.matrix_client.confirm_short_auth_string(
|
||||
event.transaction_id
|
||||
)
|
||||
if isinstance(event_response, ToDeviceError):
|
||||
logger.error(
|
||||
f"Unable to confirm emoji verification with {event.sender}, got error: {event_response}."
|
||||
)
|
||||
|
||||
# FIXME: We should allow manual cancel or reject
|
||||
# event_response = await self.matrix_client.cancel_key_verification(
|
||||
# event.transaction_id, reject=True
|
||||
# )
|
||||
# if isinstance(event_response, ToDeviceError):
|
||||
# logger.error(
|
||||
# f"Unable to reject emoji verification with {event.sender}, got error: {event_response}."
|
||||
# )
|
||||
#
|
||||
# event_response = await self.matrix_client.cancel_key_verification(
|
||||
# event.transaction_id, reject=False
|
||||
# )
|
||||
# if isinstance(event_response, ToDeviceError):
|
||||
# logger.error(
|
||||
# f"Unable to cancel emoji verification with {event.sender}, got error: {event_response}."
|
||||
# )
|
||||
|
||||
async def key_verification_end(self, event: KeyVerificationMac):
|
||||
sas = self.matrix_client.key_verifications[event.transaction_id]
|
||||
try:
|
||||
todevice_msg = sas.get_mac()
|
||||
except LocalProtocolError as e:
|
||||
# e.g. it might have been cancelled by ourselves
|
||||
logger.warning(f"Unable to conclude verification with {event.sender}: {e}.")
|
||||
return
|
||||
|
||||
event_response = await self.matrix_client.to_device(todevice_msg)
|
||||
if isinstance(event_response, ToDeviceError):
|
||||
logger.error(
|
||||
f"Unable to conclude verification with {event.sender}, got error: {event_response}."
|
||||
)
|
||||
return
|
||||
|
||||
verified_devices = " ".join(sas.verified_devices)
|
||||
logger.info(
|
||||
f"Successfully verified devices from {event.sender}: {verified_devices}"
|
||||
)
|
||||
|
||||
async def unknown(self, room: MatrixRoom, event: UnknownEvent) -> None:
|
||||
"""Callback for when an event with a type that is unknown to matrix-nio is received.
|
||||
Currently this is used for reaction events, which are not yet part of a released
|
||||
|
|
|
|||
|
|
@ -107,6 +107,9 @@ class Config:
|
|||
self.device_name: str = self._get_cfg(
|
||||
["matrix", "device_name"], default="matrix-alertbot"
|
||||
)
|
||||
self.user_token_file: str = self._get_cfg(
|
||||
["matrix", "user_token_file"], default="token.json"
|
||||
)
|
||||
self.homeserver_url: str = self._get_cfg(["matrix", "url"], required=True)
|
||||
self.allowed_rooms: list = self._get_cfg(
|
||||
["matrix", "allowed_rooms"], required=True
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
#!/usr/bin/env python3
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from asyncio import TimeoutError
|
||||
|
||||
|
|
@ -10,6 +12,10 @@ from nio import (
|
|||
AsyncClient,
|
||||
AsyncClientConfig,
|
||||
InviteMemberEvent,
|
||||
KeyVerificationCancel,
|
||||
KeyVerificationKey,
|
||||
KeyVerificationMac,
|
||||
KeyVerificationStart,
|
||||
LocalProtocolError,
|
||||
LoginError,
|
||||
MegolmEvent,
|
||||
|
|
@ -44,7 +50,14 @@ def create_matrix_client(config: Config) -> AsyncClient:
|
|||
encryption_enabled=False,
|
||||
)
|
||||
|
||||
# Initialize the matrix client
|
||||
# Load credentials from a previous session
|
||||
if os.path.exists(config.user_token_file):
|
||||
with open(config.user_token_file, "r") as ifd:
|
||||
credentials = json.load(ifd)
|
||||
config.user_token = credentials["access_token"]
|
||||
config.device_id = credentials["device_id"]
|
||||
|
||||
# Initialize the matrix client based on stored credentials
|
||||
matrix_client = AsyncClient(
|
||||
config.homeserver_url,
|
||||
config.user_id,
|
||||
|
|
@ -53,10 +66,6 @@ def create_matrix_client(config: Config) -> AsyncClient:
|
|||
config=matrix_client_config,
|
||||
)
|
||||
|
||||
if config.user_token:
|
||||
matrix_client.access_token = config.user_token
|
||||
matrix_client.user_id = config.user_id
|
||||
|
||||
return matrix_client
|
||||
|
||||
|
||||
|
|
@ -66,9 +75,12 @@ async def start_matrix_client(
|
|||
# Keep trying to reconnect on failure (with some time in-between)
|
||||
while True:
|
||||
try:
|
||||
if config.user_token:
|
||||
# Use token to log in
|
||||
matrix_client.load_store()
|
||||
if config.device_id and config.user_token:
|
||||
matrix_client.restore_login(
|
||||
user_id=config.user_id,
|
||||
device_id=config.device_id,
|
||||
access_token=config.user_token,
|
||||
)
|
||||
|
||||
# Sync encryption keys with the server
|
||||
if matrix_client.should_upload_keys:
|
||||
|
|
@ -96,6 +108,16 @@ async def start_matrix_client(
|
|||
)
|
||||
return False
|
||||
|
||||
# Save user's access token and device ID
|
||||
with open(config.user_token_file, "w") as ofd:
|
||||
json.dump(
|
||||
{
|
||||
"device_id": login_response.device_id,
|
||||
"access_token": login_response.access_token,
|
||||
},
|
||||
ofd,
|
||||
)
|
||||
|
||||
# Login succeeded!
|
||||
|
||||
logger.info(f"Logged in as {config.user_id}")
|
||||
|
|
@ -139,7 +161,18 @@ def main() -> None:
|
|||
matrix_client.add_event_callback(callbacks.decryption_failure, (MegolmEvent,))
|
||||
matrix_client.add_event_callback(callbacks.unknown, (UnknownEvent,))
|
||||
matrix_client.add_event_callback(callbacks.redaction, (RedactionEvent,))
|
||||
|
||||
matrix_client.add_to_device_callback(
|
||||
callbacks.key_verification_start, (KeyVerificationStart,)
|
||||
)
|
||||
matrix_client.add_to_device_callback(
|
||||
callbacks.key_verification_cancel, (KeyVerificationCancel,)
|
||||
)
|
||||
matrix_client.add_to_device_callback(
|
||||
callbacks.key_verification_confirm, (KeyVerificationKey,)
|
||||
)
|
||||
matrix_client.add_to_device_callback(
|
||||
callbacks.key_verification_end, (KeyVerificationMac,)
|
||||
)
|
||||
# Configure webhook server
|
||||
webhook_server = Webhook(matrix_client, alertmanager_client, cache, config)
|
||||
|
||||
|
|
|
|||
|
|
@ -9,25 +9,44 @@ command_prefix: "!alert"
|
|||
matrix:
|
||||
# The Matrix User ID of the bot account
|
||||
user_id: "@fakes_user:matrix.example.com"
|
||||
|
||||
# Matrix account password (optional if access token used)
|
||||
user_password: "password"
|
||||
|
||||
# Matrix account access token (optional if password used)
|
||||
# If not set, the server will provide an access token after log in,
|
||||
# which will be stored in the user token file (see below)
|
||||
#user_token: ""
|
||||
|
||||
# Path to the file where to store the user access token
|
||||
user_token_file: "token.json"
|
||||
|
||||
# The URL of the homeserver to connect to
|
||||
url: https://matrix.example.com
|
||||
|
||||
# The device ID that is **non pre-existing** device
|
||||
# If this device ID already exists, messages will be dropped silently in encrypted rooms
|
||||
# If not set the server will provide a device ID after log in. Note that this ID
|
||||
# will change each time the bot reconnects.
|
||||
device_id: ABCDEFGHIJ
|
||||
|
||||
# What to name the logged in device
|
||||
device_name: fake_device_name
|
||||
|
||||
# List of rooms where the bot can interact
|
||||
allowed_rooms:
|
||||
- "!abcdefgh:matrix.example.com"
|
||||
|
||||
# List of allowed reactions to create silences.
|
||||
allowed_reactions: [🤫, 😶, 🤐]
|
||||
|
||||
webhook:
|
||||
# Path to the socket for which the bot should listen to.
|
||||
# This is mutually exclusive with webhook.address option.
|
||||
socket: matrix-alertbot.socket
|
||||
|
||||
alertmanager:
|
||||
# Url to Alertmanager server
|
||||
url: http://localhost:9093
|
||||
|
||||
cache:
|
||||
|
|
@ -40,6 +59,9 @@ storage:
|
|||
path: "data/store"
|
||||
|
||||
template:
|
||||
# Path to directory that contains templates for rendering alerts.
|
||||
# The directory must contains the files "alert.html.j2" and "alert.txt.j2"
|
||||
# and must respect Jinja2 templating format.
|
||||
path: "data/templates"
|
||||
|
||||
# Logging setup
|
||||
|
|
|
|||
|
|
@ -6,20 +6,24 @@
|
|||
matrix:
|
||||
# The Matrix User ID of the bot account
|
||||
user_id: "@fakes_user:matrix.example.com"
|
||||
|
||||
# Matrix account password (optional if access token used)
|
||||
user_password: "password"
|
||||
# Matrix account access token (optional if password used)
|
||||
#user_token: ""
|
||||
|
||||
# The URL of the homeserver to connect to
|
||||
url: https://matrix.example.com
|
||||
|
||||
# List of rooms where the bot can interact
|
||||
allowed_rooms:
|
||||
- "!abcdefgh:matrix.example.com"
|
||||
|
||||
webhook:
|
||||
# Address and port for which the bot should listen to
|
||||
address: 0.0.0.0
|
||||
port: 8080
|
||||
|
||||
alertmanager:
|
||||
# Url to Alertmanager server
|
||||
url: http://localhost:9093
|
||||
|
||||
cache:
|
||||
|
|
|
|||
|
|
@ -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"""
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -95,6 +95,7 @@ class ConfigTestCase(unittest.TestCase):
|
|||
self.assertEqual("@fakes_user:matrix.example.com", config.user_id)
|
||||
self.assertEqual("password", config.user_password)
|
||||
self.assertIsNone(config.user_token)
|
||||
self.assertEqual("token.json", config.user_token_file)
|
||||
self.assertEqual("ABCDEFGHIJ", config.device_id)
|
||||
self.assertEqual("fake_device_name", config.device_name)
|
||||
self.assertEqual("https://matrix.example.com", config.homeserver_url)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue