# Copyright (C) 2020 by Animath
# SPDX-License-Identifier: GPL-3.0-or-later
from enum import Enum
import os
from asgiref.sync import async_to_sync
class Matrix:
"""
Utility class to manage interaction with the Matrix homeserver.
This log in the @tfjmbot account (must be created before).
The access token is then stored.
All is done with this bot account, that is a server administrator.
Tasks are normally asynchronous, but for compatibility we make
them synchronous.
"""
_token = None
_device_id = None
@classmethod
async def _get_client(cls): # pragma: no cover
"""
Retrieve the bot account.
If not logged, log in and store access token.
"""
if not os.getenv("SYNAPSE_PASSWORD"):
return FakeMatrixClient()
from nio import AsyncClient
client = AsyncClient("https://tfjm.org", "@tfjmbot:tfjm.org")
client.user_id = "@tfjmbot:tfjm.org"
if os.path.isfile(".matrix_token"):
with open(".matrix_device", "r") as f:
cls._device_id = f.read().rstrip(" \t\r\n")
client.device_id = cls._device_id
with open(".matrix_token", "r") as f:
cls._token = f.read().rstrip(" \t\r\n")
client.access_token = cls._token
return client
await client.login(password=os.getenv("SYNAPSE_PASSWORD"), device_name="Plateforme")
cls._token = client.access_token
cls._device_id = client.device_id
with open(".matrix_token", "w") as f:
f.write(cls._token)
with open(".matrix_device", "w") as f:
f.write(cls._device_id)
return client
@classmethod
@async_to_sync
async def set_display_name(cls, name: str):
"""
Set the display name of the bot account.
"""
client = await cls._get_client()
return await client.set_displayname(name)
@classmethod
@async_to_sync
async def set_avatar(cls, avatar_url: str): # pragma: no cover
"""
Set the display avatar of the bot account.
"""
client = await cls._get_client()
return await client.set_avatar(avatar_url)
@classmethod
@async_to_sync
async def get_avatar(cls): # pragma: no cover
"""
Set the display avatar of the bot account.
"""
client = await cls._get_client()
resp = await client.get_avatar()
return resp.avatar_url if hasattr(resp, "avatar_url") else resp
@classmethod
@async_to_sync
async def upload(
cls,
data_provider,
content_type: str = "application/octet-stream",
filename: str = None,
encrypt: bool = False,
monitor=None,
filesize: int = None,
): # pragma: no cover
"""
Upload a file to the content repository.
Returns a tuple containing:
- Either a `UploadResponse` if the request was successful, or a
`UploadError` if there was an error with the request
- A dict with file decryption info if encrypt is ``True``,
else ``None``.
Args:
data_provider (Callable, SynchronousFile, AsyncFile): A function
returning the data to upload or a file object. File objects
must be opened in binary mode (``mode="r+b"``). Callables
returning a path string, Path, async iterable or aiofiles
open binary file object allow the file data to be read in an
asynchronous and lazy way (without reading the entire file
into memory). Returning a synchronous iterable or standard
open binary file object will still allow the data to be read
lazily, but not asynchronously.
The function will be called again if the upload fails
due to a server timeout, in which case it must restart
from the beginning.
Callables receive two arguments: the total number of
429 "Too many request" errors that occured, and the total
number of server timeout exceptions that occured, thus
cleanup operations can be performed for retries if necessary.
content_type (str): The content MIME type of the file,
e.g. "image/png".
Defaults to "application/octet-stream", corresponding to a
generic binary file.
Custom values are ignored if encrypt is ``True``.
filename (str, optional): The file's original name.
encrypt (bool): If the file's content should be encrypted,
necessary for files that will be sent to encrypted rooms.
Defaults to ``False``.
monitor (TransferMonitor, optional): If a ``TransferMonitor``
object is passed, it will be updated by this function while
uploading.
From this object, statistics such as currently
transferred bytes or estimated remaining time can be gathered
while the upload is running as a task; it also allows
for pausing and cancelling.
filesize (int, optional): Size in bytes for the file to transfer.
If left as ``None``, some servers might refuse the upload.
"""
client = await cls._get_client()
return await client.upload(data_provider, content_type, filename, encrypt, monitor, filesize) \
if not isinstance(client, FakeMatrixClient) else None, None
@classmethod
@async_to_sync
async def create_room(
cls,
visibility=None,
alias=None,
name=None,
topic=None,
room_version=None,
federate=True,
is_direct=False,
preset=None,
invite=(),
initial_state=(),
power_level_override=None,
):
"""
Create a new room.
Returns either a `RoomCreateResponse` if the request was successful or
a `RoomCreateError` if there was an error with the request.
Args:
visibility (RoomVisibility): whether to have the room published in
the server's room directory or not.
Defaults to ``RoomVisibility.private``.
alias (str, optional): The desired canonical alias local part.
For example, if set to "foo" and the room is created on the
"example.com" server, the room alias will be
"#foo:example.com".
name (str, optional): A name to set for the room.
topic (str, optional): A topic to set for the room.
room_version (str, optional): The room version to set.
If not specified, the homeserver will use its default setting.
If a version not supported by the homeserver is specified,
a 400 ``M_UNSUPPORTED_ROOM_VERSION`` error will be returned.
federate (bool): Whether to allow users from other homeservers from
joining the room. Defaults to ``True``.
Cannot be changed later.
is_direct (bool): If this should be considered a
direct messaging room.
If ``True``, the server will set the ``is_direct`` flag on
``m.room.member events`` sent to the users in ``invite``.
Defaults to ``False``.
preset (RoomPreset, optional): The selected preset will set various
rules for the room.
If unspecified, the server will choose a preset from the
``visibility``: ``RoomVisibility.public`` equates to
``RoomPreset.public_chat``, and
``RoomVisibility.private`` equates to a
``RoomPreset.private_chat``.
invite (list): A list of user id to invite to the room.
initial_state (list): A list of state event dicts to send when
the room is created.
For example, a room could be made encrypted immediatly by
having a ``m.room.encryption`` event dict.
power_level_override (dict): A ``m.room.power_levels content`` dict
to override the default.
The dict will be applied on top of the generated
``m.room.power_levels`` event before it is sent to the room.
"""
client = await cls._get_client()
return await client.room_create(
visibility, alias, name, topic, room_version, federate, is_direct, preset, invite, initial_state,
power_level_override)
@classmethod
async def resolve_room_alias(cls, room_alias: str):
"""
Resolve a room alias to a room ID.
Return None if the alias does not exist.
"""
client = await cls._get_client()
resp = await client.room_resolve_alias(room_alias)
return resp.room_id if resp and hasattr(resp, "room_id") else None
@classmethod
@async_to_sync
async def invite(cls, room_id: str, user_id: str):
"""
Invite a user to a room.
Returns either a `RoomInviteResponse` if the request was successful or
a `RoomInviteError` if there was an error with the request.
Args:
room_id (str): The room id of the room that the user will be
invited to.
user_id (str): The user id of the user that should be invited.
"""
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
return await client.room_invite(room_id, user_id)
@classmethod
@async_to_sync
async def send_message(cls, room_id: str, body: str, formatted_body: str = None,
msgtype: str = "m.text", html: bool = True):
"""
Send a message to a room.
"""
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
content = {
"msgtype": msgtype,
"body": body,
"formatted_body": formatted_body or body,
}
if html:
content["format"] = "org.matrix.custom.html"
return await client.room_send(
room_id=room_id,
message_type="m.room.message",
content=content,
)
@classmethod
@async_to_sync
async def add_integration(cls, room_id: str, widget_url: str, state_key: str,
widget_type: str = "customwidget", widget_name: str = "Custom widget",
widget_title: str = ""):
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
content = {
"type": widget_type,
"url": widget_url,
"name": widget_name,
"data": {
"curl": widget_url,
"title": widget_title,
},
"creatorUserId": client.user,
"roomId": room_id,
"id": state_key,
}
return await client.room_put_state(
room_id=room_id,
event_type="im.vector.modular.widgets",
content=content,
state_key=state_key,
)
@classmethod
@async_to_sync
async def remove_integration(cls, room_id: str, state_key: str):
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
return await client.room_put_state(
room_id=room_id,
event_type="im.vector.modular.widgets",
content={},
state_key=state_key,
)
@classmethod
@async_to_sync
async def kick(cls, room_id: str, user_id: str, reason: str = None):
"""
Kick a user from a room, or withdraw their invitation.
Kicking a user adjusts their membership to "leave" with an optional
reason.
²
Returns either a `RoomKickResponse` if the request was successful or
a `RoomKickError` if there was an error with the request.
Args:
room_id (str): The room id of the room that the user will be
kicked from.
user_id (str): The user_id of the user that should be kicked.
reason (str, optional): A reason for which the user is kicked.
"""
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
return await client.room_kick(room_id, user_id, reason)
@classmethod
@async_to_sync
async def set_room_power_level(cls, room_id: str, user_id: str, power_level: int): # pragma: no cover
"""
Put a given power level to a user in a certain room.
Returns either a `RoomPutStateResponse` if the request was successful or
a `RoomPutStateError` if there was an error with the request.
Args:
room_id (str): The room id of the room where the power level
of the user should be updated.
user_id (str): The user_id of the user which power level should
be updated.
power_level (int): The target power level to give.
"""
client = await cls._get_client()
if isinstance(client, FakeMatrixClient):
return None
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
resp = await client.room_get_state_event(room_id, "m.room.power_levels")
content = resp.content
content["users"][user_id] = power_level
return await client.room_put_state(room_id, "m.room.power_levels", content=content, state_key=resp.state_key)
@classmethod
@async_to_sync
async def set_room_power_level_event(cls, room_id: str, event: str, power_level: int): # pragma: no cover
"""
Define the minimal power level to have to send a certain event type
in a given room.
Returns either a `RoomPutStateResponse` if the request was successful or
a `RoomPutStateError` if there was an error with the request.
Args:
room_id (str): The room id of the room where the power level
of the event should be updated.
event (str): The event name which minimal power level should
be updated.
power_level (int): The target power level to give.
"""
client = await cls._get_client()
if isinstance(client, FakeMatrixClient):
return None
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
resp = await client.room_get_state_event(room_id, "m.room.power_levels")
content = resp.content
if event.startswith("m."):
content["events"][event] = power_level
else:
content[event] = power_level
return await client.room_put_state(room_id, "m.room.power_levels", content=content, state_key=resp.state_key)
@classmethod
@async_to_sync
async def set_room_avatar(cls, room_id: str, avatar_uri: str):
"""
Define the avatar of a room.
Returns either a `RoomPutStateResponse` if the request was successful or
a `RoomPutStateError` if there was an error with the request.
Args:
room_id (str): The room id of the room where the avatar
should be changed.
avatar_uri (str): The internal avatar URI to apply.
"""
client = await cls._get_client()
if room_id.startswith("#"):
room_id = await cls.resolve_room_alias(room_id)
return await client.room_put_state(room_id, "m.room.avatar", content={
"url": avatar_uri
}, state_key="")
if os.getenv("SYNAPSE_PASSWORD"): # pragma: no cover
from nio import RoomVisibility, RoomPreset
RoomVisibility = RoomVisibility
RoomPreset = RoomPreset
else:
# When running tests, faking matrix-nio classes to don't include the module
class RoomVisibility(Enum):
private = 'private'
public = 'public'
class RoomPreset(Enum):
private_chat = "private_chat"
trusted_private_chat = "trusted_private_chat"
public_chat = "public_chat"
class FakeMatrixClient:
"""
Simulate a Matrix client to run tests, if no Matrix homeserver is connected.
"""
def __getattribute__(self, item):
async def func(*_, **_2):
return None
return func