matrixlib.megolm module

Megolm-related functionality

class matrixlib.megolm.EncryptionAlgorithmChanged(room_id: str, original_algorithm: str, new_algorithm: str)

A message indicating that the encryption algorithm for a room changed

Create new instance of EncryptionAlgorithmChanged(room_id, original_algorithm, new_algorithm)

new_algorithm: str

Alias for field number 2

original_algorithm: str

Alias for field number 1

room_id: str

Alias for field number 0

class matrixlib.megolm.EncryptionDisabled(room_id: str, original_algorithm: str)

A message indicating that encryption was disabled in a room

Create new instance of EncryptionDisabled(room_id, original_algorithm)

original_algorithm: str

The encryption algorithm that was initially set for the room

room_id: str

Alias for field number 0

class matrixlib.megolm.EncryptionEnabled(room_id: str, algorithm: str)

A message indicating that encryption was enabled in a room

Create new instance of EncryptionEnabled(room_id, algorithm)

algorithm: str

Alias for field number 1

room_id: str

Alias for field number 0

class matrixlib.megolm.InboundMegolmSession

Manages a Megolm session for decrypting

Do not use initializer function. Use the from_* methods instead

property authenticated: bool

Whether we know that the Megolm session comes from the associated sender_key

client: Client

decrypt(event: RoomEvent) → dict

Decrypt an m.room.encrypted event encrypted with Megolm

Arguments:

event:

the encrypted event

Returns the decrypted event, which will be a dict that should have type (the decrypted event type), content (the event content), and room_id (the ID of the room the event was sent to) properties.

classmethod from_outbound_session(c: Client, outbound: OutboundMegolmSession, key: bytes | None = None) → InboundMegolmSession

Create an InboundMegolmSession from an OutboundMegolmSession

Arguments:

c:

the client object

outbound:

the OutboundMegolmSession to use

key:

a 32-byte binary used to encrypt the objects in storage. If not specified, uses the same key as used by outbound

classmethod from_room_key(c: Client, user_id: str, sender_key: Curve25519PublicKey, room_key_content: dict, key: bytes) → InboundMegolmSession

Create an InboundMegolmSession from an m.room_key event

Arguments:

c:

the client object

user_id:

the user ID of the sender of the m.room_key event

sender_key:

the identity key of the sender of the m.room_key event

room_key_content:

the content of the m.room_key event

key:

a 32-byte binary used to encrypt the objects in storage

classmethod from_storage(c: Client, room_id: str, user_id: str, session_id: str, key: bytes) → InboundMegolmSession | None

Load a session from storage

Arguments:

c:

the client object

room_id:

the ID of the room that the session belongs to

user_id:

the ID of the user that the session belongs to

session ID:

the ID of the Megolm session

key:

a 32-byte binary used to encrypt the objects in storage

key: bytes

room_id: str

property sender_key: Curve25519PublicKey

The identity key of the Megolm session’s sender

session: InboundGroupSession

session_data: dict[str, Any]

user_id: str

class matrixlib.megolm.OlmMegolmEncryptionManager(c: Client, key: bytes, device_keys_manager: DeviceKeysManager | None = None, device_tracker: DeviceTracker | None = None, room_state_tracker: RoomStateTracker | None = None)

Manages encryption and decryption of room events

Arguments:

c:

the client object

key:

a 32-byte binary used to encrypt the objects in storage

device_keys_manager:

a devices.DeviceKeysManager object to use. If none is provided, a new one will be created.

device_tracker:

a devices.DeviceTracker object to use. If none is provided, a new one will be created.

async decrypt_room_event(encrypted: RoomEvent) → RoomEvent

Decrypt an event

Returns the room event that would have been received if the event was not encrypted.

Arguments:

encrypted:

the encrypted room event

discard_session(room_id: str) → None

Discard the outbound Megolm session for a room

This forces a new session to be created the next time we send an event to the room.

Arguments:

room_id:

the ID of the room

async encrypt_and_send_room_event(room_id: str, event_type: str, event_content: dict[str, Any], retry_ms: int = 0, txn_id: str | None = None) → None

Send an event to a room

Arguments:

room_id:

the ID of the room to send to

event_type:

the type of event to send

event_content:

the content of the event

retry_ms:

how long to retry sending, in milliseconds

txn_id:

the transaction ID to use. If none is specified, one is generated.

is_room_encrypted(room_id: str) → bool

Determines whether events sent to the given room should be encrypted.

May raise

• EncryptionDisabledError if encryption was enabled in the room in the past, but is now disabled

• EncryptionAlgorithmChangedError if the encryption algorithm for the room changed from its initial algorithm

• UnknownRoomEncryptionAlgorithmError if the encryption algorithm for the room is not known (that is, it is not Megolm)

async pre_send_megolm_session(room_id: str) → None

Pre-send the Megolm session for a room.

If the room does not yet have an outbound Megolm session, or the Megolm session needs to be rotated, a new one will be created.

This can be used before a room event is ready to be sent, to reduce the number of devices that we need to send the Megolm session to

Arguments:

room_id:

the ID of the room

class matrixlib.megolm.OutboundMegolmSession(c: Client, room_id: str, room_state_tracker: RoomStateTracker, device_tracker: DeviceTracker, device_keys_manager: DeviceKeysManager, key: bytes | None = None, session_id: str | None = None)

Manages a Megolm session for decrypting

Arguments:

c:

the client object

room_id:

the ID of the room that the session belongs to

room_state_tracker:

a RoomStateTracker object

device_tracker:

a DeviceTracker object

device_keys_manager:

a DeviceKeysManager object

key:

a 32-byte binary used to encrypt the objects in storage

session_id:

if given, will load a session from storage. If omitted a new session will be created

encrypt(event_type: str, content: dict) → dict

Encrypt an event

Arguments:

event_type:

the type of the event (e.g. m.room.message)

content:

the event content

Returns the content of a m.room.encrypted event

async get_session_key_for_sending() → list[Tuple[dict, dict]]

Get the devices that we need to send the Megolm session, along with the data to send.

Returns a list of tuples. The first item in the tuple is the device key to send to, and the second item in the tuple is the contents of an m.room_key event to send to the device. The event must be sent encrypted.

mark_as_sent(device_keys: Iterable[dict]) → None

Indicate that the session has been sent to the given devices.

Arguments:

device_keys:

an interable of devicec keys

property session_id: str

The ID of the Megolm session

exception matrixlib.megolm.SessionExpiredException

Indicates that the session has expired and must be rotated

class matrixlib.megolm.UnknownRoomEncryptionAlgorithm(room_id: str, algorithm: str)

A message indicating that the encryption algorithm for a room was initially set to an unknown algorithm

Create new instance of UnknownRoomEncryptionAlgorithm(room_id, algorithm)

algorithm: str

Alias for field number 1

room_id: str

Alias for field number 0