mirror of https://github.com/poljar/matrix-nio.git
1543 lines
50 KiB
Python
1543 lines
50 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright © 2018-2019 Damir Jelić <poljar@termina.org.uk>
|
|
# Copyright © 2021 Famedly GmbH
|
|
#
|
|
# Permission to use, copy, modify, and/or distribute this software for
|
|
# any purpose with or without fee is hereby granted, provided that the
|
|
# above copyright notice and this permission notice appear in all copies.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
|
|
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
|
|
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
|
|
# SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
|
|
# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
|
|
# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
|
|
# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
|
|
|
|
from __future__ import unicode_literals
|
|
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from ..event_builders import RoomKeyRequestMessage
|
|
from ..schemas import Schemas
|
|
from .misc import BadEvent, BadEventType, UnknownBadEvent, validate_or_badevent, verify
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
"""Matrix Event class.
|
|
|
|
This is the base event class, most events inherit from this class.
|
|
|
|
Attributes:
|
|
source (dict): The source dictionary of the event. This allows access
|
|
to all the event fields in a non-secure way.
|
|
event_id (str): A globally unique event identifier.
|
|
sender (str): The fully-qualified ID of the user who sent this
|
|
event.
|
|
server_timestamp (int): Timestamp in milliseconds on originating
|
|
homeserver when this event was sent.
|
|
decrypted (bool): A flag signaling if the event was decrypted.
|
|
verified (bool): A flag signaling if the event is verified, is True if
|
|
the event was sent from a verified device.
|
|
sender_key (str, optional): The public key of the sender that was used
|
|
to establish the encrypted session. Is only set if decrypted is
|
|
True, otherwise None.
|
|
session_id (str, optional): The unique identifier of the session that
|
|
was used to decrypt the message. Is only set if decrypted is True,
|
|
otherwise None.
|
|
transaction_id (str, optional): The unique identifier that was used
|
|
when the message was sent. Is only set if the message was sent from
|
|
our own device, otherwise None.
|
|
|
|
"""
|
|
|
|
source: Dict[str, Any] = field()
|
|
|
|
event_id: str = field(init=False)
|
|
sender: str = field(init=False)
|
|
server_timestamp: int = field(init=False)
|
|
|
|
decrypted: bool = field(default=False, init=False)
|
|
verified: bool = field(default=False, init=False)
|
|
sender_key: Optional[str] = field(default=None, init=False)
|
|
session_id: Optional[str] = field(default=None, init=False)
|
|
transaction_id: Optional[str] = field(default=None, init=False)
|
|
|
|
def __post_init__(self):
|
|
self.event_id = self.source["event_id"]
|
|
self.sender = self.source["sender"]
|
|
self.server_timestamp = self.source["origin_server_ts"]
|
|
|
|
def flattened(
|
|
self,
|
|
_prefix: str = "",
|
|
_source: Optional[Dict[str, Any]] = None,
|
|
_flat: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Return a flattened version of the ``source`` dict with dotted keys.
|
|
|
|
Example:
|
|
>>> event.source
|
|
{"content": {"body": "foo"}, "m.test": {"key": "bar"}}
|
|
>>> event.source.flattened()
|
|
{"content.body": "foo", "m.test.key": "bar"}
|
|
|
|
"""
|
|
|
|
source = self.source if _source is None else _source
|
|
flat = {} if _flat is None else _flat
|
|
|
|
for key, value in source.items():
|
|
if isinstance(value, dict):
|
|
self.flattened(f"{_prefix}{key}.", value, flat)
|
|
else:
|
|
flat[f"{_prefix}{key}"] = value
|
|
|
|
return flat
|
|
|
|
@classmethod
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[Event, BadEventType]
|
|
"""Create an Event from a dictionary.
|
|
|
|
Args:
|
|
parsed_dict (dict): The dictionary representation of the event.
|
|
|
|
"""
|
|
return cls(parsed_dict)
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_event)
|
|
def parse_event(cls, event_dict):
|
|
# type: (Dict[Any, Any]) -> Union[Event, BadEventType]
|
|
"""Parse a Matrix event and create a higher level event object.
|
|
|
|
This function parses the type of the Matrix event and produces a higher
|
|
level event object representing the parsed event.
|
|
|
|
The event structure is checked for correctness and the event fields are
|
|
type-checked. If this validation process fails for an event an BadEvent
|
|
will be produced.
|
|
|
|
If the type of the event is now known an UnknownEvent will be produced.
|
|
|
|
Args:
|
|
event_dict (dict): The dictionary representation of the event.
|
|
|
|
"""
|
|
if "unsigned" in event_dict:
|
|
if "redacted_because" in event_dict["unsigned"]:
|
|
return RedactedEvent.from_dict(event_dict)
|
|
|
|
if event_dict["type"] == "m.room.message":
|
|
return RoomMessage.parse_event(event_dict)
|
|
elif event_dict["type"] == "m.room.create":
|
|
return RoomCreateEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.guest_access":
|
|
return RoomGuestAccessEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.join_rules":
|
|
return RoomJoinRulesEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.history_visibility":
|
|
return RoomHistoryVisibilityEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.member":
|
|
return RoomMemberEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.canonical_alias":
|
|
return RoomAliasEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.name":
|
|
return RoomNameEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.topic":
|
|
return RoomTopicEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.avatar":
|
|
return RoomAvatarEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.power_levels":
|
|
return PowerLevelsEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.encryption":
|
|
return RoomEncryptionEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.redaction":
|
|
return RedactionEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.tombstone":
|
|
return RoomUpgradeEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.room.encrypted":
|
|
return Event.parse_encrypted_event(event_dict)
|
|
elif event_dict["type"] == "m.sticker":
|
|
return StickerEvent.from_dict(event_dict)
|
|
elif event_dict["type"].startswith("m.call"):
|
|
return CallEvent.parse_event(event_dict)
|
|
|
|
return UnknownEvent.from_dict(event_dict)
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_encrypted)
|
|
def parse_encrypted_event(cls, event_dict):
|
|
"""Parse an encrypted event.
|
|
|
|
Encrypted events may have different fields depending on the algorithm
|
|
that was used to encrypt them.
|
|
|
|
This function checks the algorithm of the event and produces a higher
|
|
level event from the provided dictionary.
|
|
|
|
Args:
|
|
event_dict (dict): The dictionary representation of the encrypted
|
|
event.
|
|
|
|
Returns None if the algorithm of the event is unknown.
|
|
"""
|
|
content = event_dict["content"]
|
|
|
|
if content["algorithm"] == "m.megolm.v1.aes-sha2":
|
|
return MegolmEvent.from_dict(event_dict)
|
|
|
|
return UnknownEncryptedEvent.from_dict(event_dict)
|
|
|
|
@classmethod
|
|
def parse_decrypted_event(cls, event_dict):
|
|
# type: (Dict[Any, Any]) -> Union[Event, BadEventType]
|
|
"""Parse a decrypted event and create a higher level event object.
|
|
|
|
Args:
|
|
event_dict (dict): The dictionary representation of the event.
|
|
"""
|
|
if "unsigned" in event_dict:
|
|
if "redacted_because" in event_dict["unsigned"]:
|
|
return RedactedEvent.from_dict(event_dict)
|
|
|
|
# Events shouldn't be encrypted twice, this would lead to a loop in the
|
|
# parser path.
|
|
if event_dict["type"] == "m.room.encrypted":
|
|
try:
|
|
return BadEvent.from_dict(event_dict)
|
|
except KeyError:
|
|
return UnknownBadEvent(event_dict)
|
|
if event_dict["type"] == "m.room.message":
|
|
return RoomMessage.parse_decrypted_event(event_dict)
|
|
|
|
return Event.parse_event(event_dict)
|
|
|
|
|
|
@dataclass
|
|
class UnknownEvent(Event):
|
|
"""An Event which we do not understand.
|
|
|
|
This event is created every time nio tries to parse an event of an unknown
|
|
type. Since custom and extensible events are a feature of Matrix this
|
|
allows clients to use custom events but care should be taken that the
|
|
clients will be responsible to validate and type check the event.
|
|
|
|
Attributes:
|
|
type (str): The type of the event.
|
|
|
|
"""
|
|
|
|
type: str = field()
|
|
|
|
@classmethod
|
|
def from_dict(cls, event_dict):
|
|
return cls(
|
|
event_dict,
|
|
event_dict["type"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class UnknownEncryptedEvent(Event):
|
|
"""An encrypted event which we don't know how to decrypt.
|
|
|
|
This event is created every time nio tries to parse an event encrypted
|
|
event that was encrypted using an unknown algorithm.
|
|
|
|
Attributes:
|
|
type (str): The type of the event.
|
|
algorithm (str): The algorithm of the event.
|
|
|
|
"""
|
|
|
|
type: str = field()
|
|
algorithm: str = field()
|
|
|
|
@classmethod
|
|
def from_dict(cls, event_dict):
|
|
return cls(
|
|
event_dict,
|
|
event_dict["type"],
|
|
event_dict["content"]["algorithm"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MegolmEvent(Event):
|
|
"""An undecrypted Megolm event.
|
|
|
|
MegolmEvents are presented to library users only if the library fails
|
|
to decrypt the event because of a missing session key.
|
|
|
|
MegolmEvents can be stored for later use. If a RoomKeyEvent is later on
|
|
received with a session id that matches the session_id of this event
|
|
decryption can be retried.
|
|
|
|
Attributes:
|
|
event_id (str): A globally unique event identifier.
|
|
sender (str): The fully-qualified ID of the user who sent this
|
|
event.
|
|
server_timestamp (int): Timestamp in milliseconds on originating
|
|
homeserver when this event was sent.
|
|
sender_key (str): The public key of the sender that was used
|
|
to establish the encrypted session. Is only set if decrypted is
|
|
True, otherwise None.
|
|
device_id (str): The unique identifier of the device that was used to
|
|
encrypt the event.
|
|
session_id (str): The unique identifier of the session that
|
|
was used to encrypt the message.
|
|
ciphertext (str): The undecrypted ciphertext of the event.
|
|
algorithm (str): The encryption algorithm that was used to encrypt the
|
|
message.
|
|
room_id (str): The unique identifier of the room in which the message
|
|
was sent.
|
|
transaction_id (str, optional): The unique identifier that was used
|
|
when the message was sent. Is only set if the message was sent from
|
|
our own device, otherwise None.
|
|
|
|
"""
|
|
|
|
device_id: str = field()
|
|
ciphertext: str = field()
|
|
algorithm: str = field()
|
|
room_id: str = ""
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_megolm_encrypted)
|
|
def from_dict(cls, event_dict):
|
|
"""Create a MegolmEvent from a dictionary.
|
|
|
|
Args:
|
|
event_dict (Dict): Dictionary containing the event.
|
|
|
|
Returns a MegolmEvent if the event_dict contains a valid event or a
|
|
BadEvent if it's invalid.
|
|
"""
|
|
content = event_dict["content"]
|
|
|
|
ciphertext = content["ciphertext"]
|
|
sender_key = content["sender_key"]
|
|
session_id = content["session_id"]
|
|
device_id = content["device_id"]
|
|
algorithm = content["algorithm"]
|
|
|
|
room_id = event_dict.get("room_id", None)
|
|
tx_id = (
|
|
event_dict["unsigned"].get("transaction_id", None)
|
|
if "unsigned" in event_dict
|
|
else None
|
|
)
|
|
|
|
event = cls(
|
|
event_dict,
|
|
device_id,
|
|
ciphertext,
|
|
algorithm,
|
|
room_id,
|
|
)
|
|
|
|
event.sender_key = sender_key
|
|
event.session_id = session_id
|
|
event.transaction_id = tx_id
|
|
|
|
return event
|
|
|
|
def as_key_request(
|
|
self,
|
|
user_id: str,
|
|
requesting_device_id: str,
|
|
request_id: Optional[str] = None,
|
|
device_id: Optional[str] = None,
|
|
) -> RoomKeyRequestMessage:
|
|
"""Make a to-device message for a room key request.
|
|
|
|
MegolmEvents are presented to library users only if the library fails
|
|
to decrypt the event because of a missing session key.
|
|
|
|
A missing key can be requested later on by sending a key request, this
|
|
method creates a ToDeviceMessage that can be sent out if such a request
|
|
should be made.
|
|
|
|
Args:
|
|
user_id (str): The user id of the user that should receive the key
|
|
request.
|
|
requesting_device_id (str): The device id of the user that is
|
|
requesting the key.
|
|
request_id (str, optional): A unique string identifying the
|
|
request.
|
|
Defaults to the session id of the missing megolm session.
|
|
device_id (str, optional): The device id of the device that should
|
|
receive the request. Defaults to all the users devices.
|
|
"""
|
|
assert self.session_id
|
|
request_id = request_id or self.session_id
|
|
|
|
content = {
|
|
"action": "request",
|
|
"body": {
|
|
"algorithm": self.algorithm,
|
|
"session_id": self.session_id,
|
|
"room_id": self.room_id,
|
|
"sender_key": self.sender_key,
|
|
},
|
|
"request_id": request_id,
|
|
"requesting_device_id": requesting_device_id,
|
|
}
|
|
|
|
return RoomKeyRequestMessage(
|
|
"m.room_key_request",
|
|
user_id,
|
|
device_id or "*",
|
|
content,
|
|
request_id,
|
|
self.session_id,
|
|
self.room_id,
|
|
self.algorithm,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CallEvent(Event):
|
|
"""Base Class for Matrix call signalling events.
|
|
|
|
Attributes:
|
|
call_id (str): The unique identifier of the call.
|
|
version (int): The version of the VoIP specification this message
|
|
adheres to.
|
|
|
|
"""
|
|
|
|
call_id: str = field()
|
|
version: int = field()
|
|
|
|
@staticmethod
|
|
def parse_event(event_dict):
|
|
"""Parse a Matrix event and create a higher level event object.
|
|
|
|
This function parses the type of the Matrix event and produces a
|
|
higher level CallEvent object representing the parsed event.
|
|
|
|
The event structure is checked for correctness and the event fields are
|
|
type checked. If this validation process fails for an event an BadEvent
|
|
will be produced.
|
|
|
|
If the type of the event is now known an UnknownEvent will be produced.
|
|
|
|
Args:
|
|
event_dict (dict): The raw matrix event dictionary.
|
|
|
|
"""
|
|
if event_dict["type"] == "m.call.candidates":
|
|
event = CallCandidatesEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.call.invite":
|
|
event = CallInviteEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.call.answer":
|
|
event = CallAnswerEvent.from_dict(event_dict)
|
|
elif event_dict["type"] == "m.call.hangup":
|
|
event = CallHangupEvent.from_dict(event_dict)
|
|
else:
|
|
event = UnknownEvent.from_dict(event_dict)
|
|
|
|
return event
|
|
|
|
|
|
@dataclass
|
|
class CallCandidatesEvent(CallEvent):
|
|
"""Call event holding additional VoIP ICE candidates.
|
|
|
|
This event is sent by callers after sending an invite and by the callee
|
|
after answering. Its purpose is to give the other party additional ICE
|
|
candidates to try using to communicate.
|
|
|
|
Args:
|
|
candidates (list): A list of dictionaries describing the candidates.
|
|
"""
|
|
|
|
candidates: List[Dict[str, Any]] = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.call_candidates)
|
|
def from_dict(cls, event_dict):
|
|
content = event_dict.get("content", {})
|
|
return cls(
|
|
event_dict,
|
|
content["call_id"],
|
|
content["version"],
|
|
content["candidates"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CallInviteEvent(CallEvent):
|
|
"""Event representing an invitation to a VoIP call.
|
|
|
|
This event is sent by a caller when they wish to establish a call.
|
|
|
|
Attributes:
|
|
lifetime (integer): The time in milliseconds that the invite is valid
|
|
for.
|
|
offer (dict): The session description object. A dictionary containing
|
|
the keys "type" which must be "offer" for this event and "sdp"
|
|
which contains the SDP text of the session description.
|
|
|
|
"""
|
|
|
|
lifetime: int = field()
|
|
offer: Dict[str, Any] = field()
|
|
|
|
@property
|
|
def expired(self):
|
|
"""Property marking if the invite event expired."""
|
|
now = time.time()
|
|
return now - (self.server_timestamp / 1000) > (self.lifetime / 1000)
|
|
|
|
@classmethod
|
|
@verify(Schemas.call_invite)
|
|
def from_dict(cls, event_dict):
|
|
content = event_dict.get("content", {})
|
|
return cls(
|
|
event_dict,
|
|
content["call_id"],
|
|
content["version"],
|
|
content["lifetime"],
|
|
content["offer"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CallAnswerEvent(CallEvent):
|
|
"""Event representing the answer to a VoIP call.
|
|
|
|
This event is sent by the callee when they wish to answer the call.
|
|
|
|
Attributes:
|
|
answer (dict): The session description object. A dictionary containing
|
|
the keys "type" which must be "answer" for this event and "sdp"
|
|
which contains the SDP text of the session description.
|
|
|
|
"""
|
|
|
|
answer: Dict[str, Any] = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.call_answer)
|
|
def from_dict(cls, event_dict):
|
|
content = event_dict.get("content", {})
|
|
return cls(
|
|
event_dict,
|
|
content["call_id"],
|
|
content["version"],
|
|
content["answer"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CallHangupEvent(CallEvent):
|
|
"""An event representing the end of a VoIP call.
|
|
|
|
Sent by either party to signal their termination of the call. This can be
|
|
sent either once the call has has been established or before to abort the
|
|
call.
|
|
|
|
"""
|
|
|
|
@classmethod
|
|
@verify(Schemas.call_hangup)
|
|
def from_dict(cls, event_dict):
|
|
content = event_dict.get("content", {})
|
|
return cls(
|
|
event_dict,
|
|
content["call_id"],
|
|
content["version"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RedactedEvent(Event):
|
|
"""An event that has been redacted.
|
|
|
|
Attributes:
|
|
type (str): The type of the event that has been redacted.
|
|
redacter (str): The fully-qualified ID of the user who redacted the
|
|
event.
|
|
reason (str, optional): A string describing why the event was redacted,
|
|
can be None.
|
|
|
|
"""
|
|
|
|
type: str = field()
|
|
redacter: str = field()
|
|
reason: Optional[str] = field()
|
|
|
|
def __str__(self):
|
|
reason = f", reason: {self.reason}" if self.reason else ""
|
|
return f"Redacted event of type {self.type}, by {self.redacter}{reason}."
|
|
|
|
@property
|
|
def event_type(self):
|
|
"""Type of the event."""
|
|
return self.type
|
|
|
|
@classmethod
|
|
@verify(Schemas.redacted_event)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RedactedEvent, BadEventType]
|
|
redacter = parsed_dict["unsigned"]["redacted_because"]["sender"]
|
|
content_dict = parsed_dict["unsigned"]["redacted_because"]["content"]
|
|
reason = content_dict.get("reason", None)
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["type"],
|
|
redacter,
|
|
reason,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptionEvent(Event):
|
|
"""An event signaling that encryption has been enabled in a room."""
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_encryption)
|
|
def from_dict(cls, parsed_dict):
|
|
return cls(parsed_dict)
|
|
|
|
|
|
@dataclass
|
|
class RoomCreateEvent(Event):
|
|
"""The first event in a room, signaling that the room was created.
|
|
|
|
Attributes:
|
|
creator (str): The fully-qualified ID of the user who created the room.
|
|
federate (bool): A boolean flag telling us whether users on other
|
|
homeservers are able to join this room.
|
|
room_version (str): The version of the room. Different room versions
|
|
will have different event formats. Clients shouldn't worry about
|
|
this too much unless they want to perform room upgrades.
|
|
|
|
"""
|
|
|
|
creator: str = field()
|
|
federate: bool = True
|
|
room_version: str = "1"
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_create)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomCreateEvent, BadEventType]
|
|
creator = parsed_dict["content"]["creator"]
|
|
federate = parsed_dict["content"]["m.federate"]
|
|
version = parsed_dict["content"]["room_version"]
|
|
|
|
return cls(parsed_dict, creator, federate, version)
|
|
|
|
|
|
@dataclass
|
|
class RoomGuestAccessEvent(Event):
|
|
"""Event signaling whether guest users are allowed to join rooms.
|
|
|
|
Attributes:
|
|
guest_access (str): A string describing the guest access policy of the
|
|
room. Can be one of "can_join" or "forbidden".
|
|
|
|
"""
|
|
|
|
guest_access: str = "forbidden"
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_guest_access)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomGuestAccessEvent, BadEventType]
|
|
guest_access = parsed_dict["content"]["guest_access"]
|
|
|
|
return cls(parsed_dict, guest_access)
|
|
|
|
|
|
@dataclass
|
|
class RoomJoinRulesEvent(Event):
|
|
"""An event telling us how users can join the room.
|
|
|
|
Attributes:
|
|
join_rule (str): A string telling us how users may join the room, can
|
|
be one of "public" meaning anyone can join the room without any
|
|
restrictions or "invite" meaning users can only join if they have
|
|
been previously invited.
|
|
|
|
"""
|
|
|
|
join_rule: str = "invite"
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_join_rules)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomJoinRulesEvent, BadEventType]
|
|
join_rule = parsed_dict["content"]["join_rule"]
|
|
|
|
return cls(parsed_dict, join_rule)
|
|
|
|
|
|
@dataclass
|
|
class RoomHistoryVisibilityEvent(Event):
|
|
"""An event telling whether users can read the room history.
|
|
|
|
Room history visibility can be set up in multiple ways in Matrix:
|
|
|
|
* world_readable
|
|
All events value may be shared by any participating
|
|
homeserver with anyone, regardless of whether they have ever joined
|
|
the room.
|
|
* shared
|
|
Previous events are always accessible to newly joined
|
|
members. All events in the room are accessible, even those sent
|
|
when the member was not a part of the room.
|
|
* invited
|
|
Events are accessible to newly joined members from the
|
|
point they were invited onwards. Events stop being accessible when
|
|
the member's state changes to something other than invite or join.
|
|
* joined
|
|
Events are only accessible to members from the point on they
|
|
joined to the room and stop being accessible when they aren't
|
|
joined anymore.
|
|
|
|
Attributes:
|
|
history_visibility (str): A string describing who can read the room
|
|
history. One of "invited", "joined", "shared", "world_readable".
|
|
|
|
"""
|
|
|
|
history_visibility: str = "shared"
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_history_visibility)
|
|
def from_dict(
|
|
cls,
|
|
parsed_dict, # type: Dict[Any, Any]
|
|
):
|
|
# type: (...) -> Union[RoomHistoryVisibilityEvent, BadEventType]
|
|
history_visibility = parsed_dict["content"]["history_visibility"]
|
|
|
|
return cls(parsed_dict, history_visibility)
|
|
|
|
|
|
@dataclass
|
|
class RoomAliasEvent(Event):
|
|
"""An event informing us about which alias should be preferred.
|
|
|
|
Attributes:
|
|
canonical_alias (str): The alias that is considered canonical.
|
|
|
|
"""
|
|
|
|
canonical_alias: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_canonical_alias)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomAliasEvent, BadEventType]
|
|
canonical_alias = parsed_dict["content"].get("alias")
|
|
|
|
return cls(parsed_dict, canonical_alias)
|
|
|
|
|
|
@dataclass
|
|
class RoomNameEvent(Event):
|
|
"""Event holding the name of the room.
|
|
|
|
The room name is a human-friendly string designed to be displayed to the
|
|
end-user. The room name is not unique, as multiple rooms can have the same
|
|
room name set.
|
|
|
|
Attributes:
|
|
name (str): The name of the room.
|
|
|
|
"""
|
|
|
|
name: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_name)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomNameEvent, BadEventType]
|
|
room_name = parsed_dict["content"]["name"]
|
|
|
|
return cls(parsed_dict, room_name)
|
|
|
|
|
|
@dataclass
|
|
class RoomTopicEvent(Event):
|
|
"""Event holding the topic of a room.
|
|
|
|
A topic is a short message detailing what is currently being discussed in
|
|
the room. It can also be used as a way to display extra information about
|
|
the room, which may not be suitable for the room name.
|
|
|
|
Attributes:
|
|
topic (str): The topic of the room.
|
|
|
|
"""
|
|
|
|
topic: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_topic)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomTopicEvent, BadEventType]
|
|
canonical_alias = parsed_dict["content"]["topic"]
|
|
|
|
return cls(parsed_dict, canonical_alias)
|
|
|
|
|
|
@dataclass
|
|
class RoomAvatarEvent(Event):
|
|
"""Event holding a picture that is associated with the room.
|
|
|
|
Attributes:
|
|
avatar_url (str): The URL to the picture.
|
|
|
|
"""
|
|
|
|
avatar_url: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_avatar)
|
|
def from_dict(cls, parsed_dict):
|
|
# type (Dict[Any, Any]) -> Union[RoomAvatarEvent, BadEventType]
|
|
room_avatar_url = parsed_dict["content"]["url"]
|
|
|
|
return cls(parsed_dict, room_avatar_url)
|
|
|
|
|
|
@dataclass
|
|
class RoomMessage(Event):
|
|
"""Abstract room message class.
|
|
|
|
This class corespondents to a Matrix event of the m.room.message type. It
|
|
is used when messages are sent to the room.
|
|
|
|
The class has one child class per msgtype.
|
|
"""
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_message)
|
|
def parse_event(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomMessage, BadEventType]
|
|
content_dict = parsed_dict["content"]
|
|
|
|
if content_dict["msgtype"] == "m.text":
|
|
event = RoomMessageText.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.emote":
|
|
event = RoomMessageEmote.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.notice":
|
|
event = RoomMessageNotice.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.image":
|
|
event = RoomMessageImage.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.audio":
|
|
event = RoomMessageAudio.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.video":
|
|
event = RoomMessageVideo.from_dict(parsed_dict)
|
|
elif content_dict["msgtype"] == "m.file":
|
|
event = RoomMessageFile.from_dict(parsed_dict)
|
|
else:
|
|
event = RoomMessageUnknown.from_dict(parsed_dict)
|
|
|
|
if "unsigned" in parsed_dict:
|
|
txn_id = parsed_dict["unsigned"].get("transaction_id", None)
|
|
event.transaction_id = txn_id
|
|
|
|
return event
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_message)
|
|
def parse_decrypted_event(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomMessage, BadEventType]
|
|
msgtype = parsed_dict["content"]["msgtype"]
|
|
|
|
if msgtype == "m.image":
|
|
event = RoomEncryptedImage.from_dict(parsed_dict)
|
|
elif msgtype == "m.audio":
|
|
event = RoomEncryptedAudio.from_dict(parsed_dict)
|
|
elif msgtype == "m.video":
|
|
event = RoomEncryptedVideo.from_dict(parsed_dict)
|
|
elif msgtype == "m.file":
|
|
event = RoomEncryptedFile.from_dict(parsed_dict)
|
|
else:
|
|
event = RoomMessage.parse_event(parsed_dict)
|
|
|
|
if "unsigned" in parsed_dict:
|
|
txn_id = parsed_dict["unsigned"].get("transaction_id", None)
|
|
event.transaction_id = txn_id
|
|
|
|
return event
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageMedia(RoomMessage):
|
|
"""Base class for room messages containing a URI.
|
|
|
|
Attributes:
|
|
url (str): The URL of the file.
|
|
body (str): The description of the message.
|
|
|
|
"""
|
|
|
|
url: str = field()
|
|
body: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_message_media)
|
|
def from_dict(cls, parsed_dict):
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["content"]["url"],
|
|
parsed_dict["content"]["body"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptedMedia(RoomMessage):
|
|
"""Base class for encrypted room messages containing an URI.
|
|
|
|
Attributes:
|
|
url (str): The URL of the file.
|
|
body (str): The description of the message.
|
|
key (dict): The key that can be used to decrypt the file.
|
|
hashes (dict): A mapping from an algorithm name to a hash of the
|
|
ciphertext encoded as base64.
|
|
iv (str): The initialisation vector that was used to encrypt the file.
|
|
mimetype (str, optional): The mimetype of the message.
|
|
|
|
thumbnail_url (str, optional): The URL of the thumbnail file.
|
|
thumbnail_key (dict, optional): The key that can be used to decrypt the
|
|
thumbnail file.
|
|
thumbnail_hashes (dict, optional): A mapping from an algorithm name to
|
|
a hash of the thumbnail ciphertext encoded as base64.
|
|
thumbnail_iv (str, optional): The initialisation vector that was used
|
|
to encrypt the thumbnail file.
|
|
"""
|
|
|
|
url: str = field()
|
|
body: str = field()
|
|
key: Dict[str, Any] = field()
|
|
hashes: Dict[str, Any] = field()
|
|
iv: str = field()
|
|
mimetype: str = field()
|
|
|
|
thumbnail_url: Optional[str] = None
|
|
thumbnail_key: Optional[Dict] = None
|
|
thumbnail_hashes: Optional[Dict] = None
|
|
thumbnail_iv: Optional[str] = None
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_encrypted_media)
|
|
def from_dict(cls, parsed_dict):
|
|
info = parsed_dict["content"].get("info", {})
|
|
thumbnail_file = info.get("thumbnail_file", {})
|
|
|
|
thumbnail_url = thumbnail_file.get("url")
|
|
thumbnail_key = thumbnail_file.get("key")
|
|
thumbnail_hashes = thumbnail_file.get("hashes")
|
|
thumbnail_iv = thumbnail_file.get("iv")
|
|
|
|
mimetype = info.get("mimetype") or parsed_dict["content"]["file"].get(
|
|
"mimetype"
|
|
)
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["content"]["file"]["url"],
|
|
parsed_dict["content"]["body"],
|
|
parsed_dict["content"]["file"]["key"],
|
|
parsed_dict["content"]["file"]["hashes"],
|
|
parsed_dict["content"]["file"]["iv"],
|
|
mimetype,
|
|
thumbnail_url,
|
|
thumbnail_key,
|
|
thumbnail_hashes,
|
|
thumbnail_iv,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptedImage(RoomEncryptedMedia):
|
|
"""A room message containing an image where the file is encrypted."""
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptedAudio(RoomEncryptedMedia):
|
|
"""A room message containing an audio clip where the file is encrypted."""
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptedVideo(RoomEncryptedMedia):
|
|
"""A room message containing a video clip where the file is encrypted."""
|
|
|
|
|
|
@dataclass
|
|
class RoomEncryptedFile(RoomEncryptedMedia):
|
|
"""A room message containing a generic encrypted file."""
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageImage(RoomMessageMedia):
|
|
"""A room message containing an image."""
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageAudio(RoomMessageMedia):
|
|
"""A room message containing an audio clip."""
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageVideo(RoomMessageMedia):
|
|
"""A room message containing a video clip."""
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageFile(RoomMessageMedia):
|
|
"""A room message containing a generic file."""
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageUnknown(RoomMessage):
|
|
"""A m.room.message which we do not understand.
|
|
|
|
This event is created every time nio tries to parse a room message of an
|
|
unknown msgtype. Since custom and extensible events are a feature of Matrix
|
|
this allows clients to use custom messages but care should be taken that
|
|
the clients will be responsible to validate and type check the content of
|
|
the message.
|
|
|
|
Attributes:
|
|
msgtype (str): The msgtype of the room message.
|
|
content (dict): The dictionary holding the content of the room message.
|
|
The keys and values of this dictionary will differ depending on the
|
|
msgtype.
|
|
|
|
"""
|
|
|
|
msgtype: str = field()
|
|
content: Dict[str, Any] = field()
|
|
|
|
@classmethod
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> RoomMessage
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["content"]["msgtype"],
|
|
parsed_dict.get("content", {}),
|
|
)
|
|
|
|
@property
|
|
def type(self):
|
|
"""Get the msgtype of the room message."""
|
|
return self.msgtype
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageFormatted(RoomMessage):
|
|
"""Base abstract class for room messages that can have formatted bodies.
|
|
|
|
Attributes:
|
|
body (str): The textual body of the message.
|
|
formatted_body (str, optional): The formatted version of the body. Can
|
|
be None if the message doesn't contain a formatted version of the
|
|
body.
|
|
format (str, optional): The format used in the formatted_body. This
|
|
specifies how the formatted_body should be interpreted.
|
|
|
|
"""
|
|
|
|
body: str = field()
|
|
formatted_body: Optional[str] = field()
|
|
format: Optional[str] = field()
|
|
|
|
def __str__(self):
|
|
# type: () -> str
|
|
return f"{self.sender}: {self.body}"
|
|
|
|
@staticmethod
|
|
def _validate(parsed_dict):
|
|
raise NotImplementedError()
|
|
|
|
@classmethod
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomMessage, BadEventType]
|
|
bad = cls._validate(parsed_dict)
|
|
|
|
if bad:
|
|
return bad
|
|
|
|
body = parsed_dict["content"]["body"]
|
|
body_format = parsed_dict["content"].get("format")
|
|
|
|
# Only try to find the formatted body if the format is specified. It is
|
|
# required by the spec to have both or none specified.
|
|
if body_format:
|
|
formatted_body = parsed_dict["content"].get("formatted_body")
|
|
else:
|
|
formatted_body = None
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
body,
|
|
formatted_body,
|
|
body_format,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageText(RoomMessageFormatted):
|
|
"""A room message corresponding to the m.text msgtype.
|
|
|
|
This message is the most basic message and is used to represent text.
|
|
|
|
Attributes:
|
|
body (str): The textual body of the message.
|
|
formatted_body (str, optional): The formatted version of the body. Can
|
|
be None if the message doesn't contain a formatted version of the
|
|
body.
|
|
format (str, optional): The format used in the formatted_body. This
|
|
specifies how the formatted_body should be interpreted.
|
|
|
|
"""
|
|
|
|
@staticmethod
|
|
def _validate(parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Optional[BadEventType]
|
|
return validate_or_badevent(parsed_dict, Schemas.room_message_text)
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageEmote(RoomMessageFormatted):
|
|
"""A room message coresponding to the m.emote msgtype.
|
|
|
|
This message is similar to m.text except that the sender is 'performing'
|
|
the action contained in the body key, similar to /me in IRC.
|
|
|
|
Attributes:
|
|
body (str): The textual body of the message.
|
|
formatted_body (str, optional): The formatted version of the body. Can
|
|
be None if the message doesn't contain a formatted version of the
|
|
body.
|
|
format (str, optional): The format used in the formatted_body. This
|
|
specifies how the formatted_body should be interpreted.
|
|
|
|
"""
|
|
|
|
@staticmethod
|
|
def _validate(parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Optional[BadEventType]
|
|
return validate_or_badevent(parsed_dict, Schemas.room_message_emote)
|
|
|
|
|
|
@dataclass
|
|
class RoomMessageNotice(RoomMessageFormatted):
|
|
"""A room message corresponding to the m.notice msgtype.
|
|
|
|
Room notices are primarily intended for responses from automated
|
|
clients.
|
|
|
|
Attributes:
|
|
body (str): The textual body of the notice.
|
|
formatted_body (str, optional): The formatted version of the notice
|
|
body. Can be None if the message doesn't contain a formatted
|
|
version of the body.
|
|
format (str, optional): The format used in the formatted_body. This
|
|
specifies how the formatted_body should be interpreted.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _validate(parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Optional[BadEventType]
|
|
return validate_or_badevent(parsed_dict, Schemas.room_message_notice)
|
|
|
|
|
|
@dataclass
|
|
class DefaultLevels:
|
|
"""Class holding information about default power levels of a room.
|
|
|
|
Attributes:
|
|
ban (int): The level required to ban a user.
|
|
invite (int): The level required to invite a user.
|
|
kick (int): The level required to kick a user.
|
|
redact (int): The level required to redact events.
|
|
state_default (int): The level required to send state events. This can
|
|
be overridden by the events power level mapping.
|
|
events_default (int): The level required to send message events. This
|
|
can be overridden by the events power level mapping.
|
|
users_default (int): The default power level for every user in the
|
|
room. This can be overridden by the users power level mapping.
|
|
notifications (Dict[str, int]): The level required to send different
|
|
kinds of notifications. Used for ``sender_notification_permission``
|
|
conditions in push rules.
|
|
"""
|
|
|
|
ban: int = 50
|
|
invite: int = 50
|
|
kick: int = 50
|
|
redact: int = 50
|
|
state_default: int = 0
|
|
events_default: int = 0
|
|
users_default: int = 0
|
|
notifications: Dict[str, int] = field(default_factory=lambda: {"room": 50})
|
|
|
|
@classmethod
|
|
def from_dict(cls, parsed_dict):
|
|
"""Create a DefaultLevels object from a dictionary.
|
|
|
|
This creates the DefaultLevels object from a dictionary containing a
|
|
m.room.power_levels event. The event structure isn't checked in this
|
|
method.
|
|
|
|
This shouldn't be used directly, the `PowerLevelsEvent` method will
|
|
call this method to construct the DefaultLevels object.
|
|
"""
|
|
content = parsed_dict["content"]
|
|
return cls(
|
|
content["ban"],
|
|
content["invite"],
|
|
content["kick"],
|
|
content["redact"],
|
|
content["state_default"],
|
|
content["events_default"],
|
|
content["users_default"],
|
|
content["notifications"],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class PowerLevels:
|
|
"""Class holding information of room power levels.
|
|
|
|
Attributes:
|
|
defaults (DefaultLevels): The default power levels of the room.
|
|
users (dict): The power levels for specific users. This is a mapping
|
|
from user_id to power level for that user.
|
|
events (dict): The level required to send specific event types. This is
|
|
a mapping from event type to power level required.
|
|
|
|
"""
|
|
|
|
defaults: DefaultLevels = field(default_factory=DefaultLevels)
|
|
users: Dict[str, int] = field(default_factory=dict)
|
|
events: Dict[str, int] = field(default_factory=dict)
|
|
|
|
def get_state_event_required_level(self, event_type):
|
|
# type: (str) -> int
|
|
"""Get required power level to send a certain type of state event.
|
|
|
|
Returns an integer representing the required power level.
|
|
|
|
Args:
|
|
event_type (str): The type of matrix state event we want the
|
|
required level for, e.g. `m.room.name` or `m.room.topic`.
|
|
"""
|
|
return self.events.get(event_type, self.defaults.state_default)
|
|
|
|
def get_message_event_required_level(self, event_type):
|
|
# type: (str) -> int
|
|
"""Get required power level to send a certain type of message event.
|
|
|
|
Returns an integer representing the required power level.
|
|
|
|
Args:
|
|
event_type (str): The type of matrix message event we want the
|
|
required level for, e.g. `m.room.message`.
|
|
"""
|
|
return self.events.get(event_type, self.defaults.events_default)
|
|
|
|
def get_notification_required_level(self, notification_type: str) -> int:
|
|
"""Get required power level to send a certain type of notification.
|
|
|
|
Returns an integer representing the required power level.
|
|
|
|
Args:
|
|
notification_type (str): The type of notification to get the
|
|
required level for, e.g. ``"room"``.
|
|
"""
|
|
return self.defaults.notifications.get(notification_type, 50)
|
|
|
|
def get_user_level(self, user_id):
|
|
# type: (str) -> int
|
|
"""Get the power level of a user.
|
|
|
|
Returns an integer representing the user's power level.
|
|
|
|
Args:
|
|
user_id (str): The fully-qualified ID of the user for whom we would
|
|
like to get the power level.
|
|
"""
|
|
return self.users.get(user_id, self.defaults.users_default)
|
|
|
|
def can_user_send_state(self, user_id, event_type):
|
|
# type: (str, str) -> bool
|
|
"""Return whether a user has enough power to send certain state events.
|
|
|
|
Args:
|
|
user_id (str): The user to check the power of.
|
|
event_type (str): The type of matrix state event to check the
|
|
required power of, e.g. `m.room.encryption`.
|
|
"""
|
|
required_level = self.get_state_event_required_level(event_type)
|
|
return self.get_user_level(user_id) >= required_level
|
|
|
|
def can_user_send_message(self, user_id, event_type="m.room.message"):
|
|
# type: (str, str) -> bool
|
|
"""
|
|
Return whether a user has enough power to send certain message events.
|
|
|
|
Args:
|
|
user_id (str): The user to check the power of.
|
|
event_type (str): The type of matrix message event to check the
|
|
required power of, `m.room.message` by default.
|
|
"""
|
|
required_level = self.get_message_event_required_level(event_type)
|
|
return self.get_user_level(user_id) >= required_level
|
|
|
|
def can_user_invite(self, user_id):
|
|
# type: (str) -> bool
|
|
"""Return whether a user has enough power to invite others."""
|
|
return self.get_user_level(user_id) >= self.defaults.invite
|
|
|
|
def can_user_kick(
|
|
self,
|
|
user_id: str,
|
|
target_user_id: Optional[str] = None,
|
|
) -> bool:
|
|
"""Return whether a user has enough power to kick another.
|
|
|
|
If ``target_user_id`` is ``None``, returns whether ``user_id`` has
|
|
enough power to kick anyone with a lower power level than that user.
|
|
"""
|
|
level = self.get_user_level(user_id)
|
|
can_kick_lower = level >= self.defaults.kick
|
|
|
|
if target_user_id is None:
|
|
return can_kick_lower
|
|
|
|
return can_kick_lower and level > self.get_user_level(target_user_id)
|
|
|
|
def can_user_ban(
|
|
self,
|
|
user_id: str,
|
|
target_user_id: Optional[str] = None,
|
|
) -> bool:
|
|
"""Return whether a user has enough power to ban another.
|
|
|
|
If ``target_user_id`` is ``None``, returns whether ``user_id`` has
|
|
enough power to ban anyone with a lower power level than that user.
|
|
"""
|
|
level = self.get_user_level(user_id)
|
|
can_ban_lower = level >= self.defaults.ban
|
|
|
|
if target_user_id is None:
|
|
return can_ban_lower
|
|
|
|
return can_ban_lower and level > self.get_user_level(target_user_id)
|
|
|
|
def can_user_redact(self, user_id: str):
|
|
"""Return whether a user has enough power to redact other user's events."""
|
|
return self.get_user_level(user_id) >= self.defaults.redact
|
|
|
|
def can_user_notify(self, user_id: str, notification_type: str):
|
|
"""Return whether user has enough power to send a type of notification."""
|
|
required = self.get_notification_required_level(notification_type)
|
|
return self.get_user_level(user_id) >= required
|
|
|
|
def update(self, new_levels):
|
|
"""Update the power levels object with new levels.
|
|
|
|
Args:
|
|
new_levels (PowerLevels): A new PowerLevels object that we received
|
|
from a newer PowerLevelsEvent.
|
|
"""
|
|
if not isinstance(new_levels, PowerLevels):
|
|
return
|
|
|
|
self.defaults = new_levels.defaults
|
|
self.events.update(new_levels.events)
|
|
self.users.update(new_levels.users)
|
|
|
|
|
|
@dataclass
|
|
class PowerLevelsEvent(Event):
|
|
"""Class representing a m.room.power_levels event.
|
|
|
|
This event specifies the minimum level a user must have in order to perform
|
|
a certain action. It also specifies the levels of each user in the room.
|
|
|
|
Attributes:
|
|
power_levels (PowerLevels): The PowerLevels object holding information
|
|
of the power levels of the room.
|
|
|
|
"""
|
|
|
|
power_levels: PowerLevels = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_power_levels)
|
|
def from_dict(cls, parsed_dict):
|
|
default_levels = DefaultLevels.from_dict(parsed_dict)
|
|
|
|
users = parsed_dict["content"].get("users", {})
|
|
events = parsed_dict["content"].get("events", {})
|
|
|
|
levels = PowerLevels(default_levels, users, events)
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
levels,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RedactionEvent(Event):
|
|
"""An event signaling that another event has been redacted.
|
|
|
|
Events can be redacted by either room or server administrators. Redacting
|
|
an event means that all keys not required by the protocol are stripped off.
|
|
|
|
Attributes:
|
|
redacts (str): The event id of the event that has been redacted.
|
|
reason (str, optional): A string describing why the event was redacted,
|
|
can be None.
|
|
|
|
"""
|
|
|
|
redacts: str = field()
|
|
reason: Optional[str] = None
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_redaction)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RedactionEvent, BadEventType]
|
|
content = parsed_dict.get("content", {})
|
|
reason = content.get("reason", None)
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["redacts"],
|
|
reason,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomMemberEvent(Event):
|
|
"""Class representing to an m.room.member event.
|
|
|
|
Attributes:
|
|
state_key (str): The user_id this membership event relates to. In all
|
|
cases except for when membership is join, the user ID in the sender
|
|
attribute does not need to match the user ID in the state_key.
|
|
membership (str): The membership state of the user. One of "invite",
|
|
"join", "leave", "ban".
|
|
prev_membership (str, optional): The previous membership state that
|
|
this one is overwriting. Can be None in which case the membership
|
|
state is assumed to have been "leave".
|
|
content (dict): The content of the of the membership event.
|
|
prev_content(dict, optional): The content of a previous membership
|
|
event that this one is overwriting.
|
|
|
|
"""
|
|
|
|
state_key: str = field()
|
|
membership: str = field()
|
|
prev_membership: Optional[str] = field()
|
|
content: Dict[str, Any] = field()
|
|
prev_content: Optional[Dict[str, Any]] = None
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_membership)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[RoomMemberEvent, BadEventType]
|
|
content = parsed_dict.get("content", {})
|
|
unsigned = parsed_dict.get("unsigned", {})
|
|
prev_content = unsigned.get("prev_content", None)
|
|
|
|
membership = content["membership"]
|
|
prev_membership = prev_content.get("membership") if prev_content else None
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
parsed_dict["state_key"],
|
|
membership,
|
|
prev_membership,
|
|
content,
|
|
prev_content,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class StickerEvent(Event):
|
|
"""An event indicating the use of a sticker
|
|
|
|
Sticker messages are specialised image messages that are displayed
|
|
without controls. Sticker messages are intended to provide simple
|
|
"reaction" events in the message timeline.
|
|
|
|
Attributes:
|
|
body (str): A textual representation or associated description of
|
|
the sticker image. This could be the alt text of the original image,
|
|
or a message to accompany and further describe the sticker.
|
|
url (str): The URL to the sticker image.
|
|
content (dict): The content of the of the redaction event.
|
|
|
|
"""
|
|
|
|
body: str = field()
|
|
url: str = field()
|
|
content: Dict[str, Any] = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.sticker)
|
|
def from_dict(cls, parsed_dict):
|
|
# type: (Dict[Any, Any]) -> Union[StickerEvent, BadEventType]
|
|
content = parsed_dict.get("content", {})
|
|
|
|
body = content["body"]
|
|
url = content["url"]
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
body,
|
|
url,
|
|
content,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class RoomUpgradeEvent(Event):
|
|
"""Class representing to an m.room.tombstone event.
|
|
|
|
A state event signifying that a room has been upgraded to a
|
|
different room version, and that clients should go there.
|
|
|
|
Attributes:
|
|
body (str): A server-defined message.
|
|
replacement_room (str): The new room the client should be visiting.
|
|
content (dict): The content of the tombstone event.
|
|
"""
|
|
|
|
body: str = field()
|
|
replacement_room: str = field()
|
|
|
|
@classmethod
|
|
@verify(Schemas.room_tombstone)
|
|
def from_dict(cls, parsed_dict):
|
|
content = parsed_dict.get("content", {})
|
|
body = content.get("body", "")
|
|
replacement_room = content.get("replacement_room", "")
|
|
|
|
return cls(
|
|
parsed_dict,
|
|
body,
|
|
replacement_room,
|
|
)
|