1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
# Copyright (c) 2022 Tulir Asokan
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from typing import List, Optional, Union
from attr import dataclass
import attr
from ..primitive import JSON, DeviceID, IdentityKey, RoomID, SessionID, SigningKey, UserID
from ..util import ExtensibleEnum, Obj, SerializableAttrs, deserializer, field
from .base import BaseEvent, EventType
from .beeper import BeeperRoomKeyAckEventContent
from .encrypted import EncryptedOlmEventContent, EncryptionAlgorithm
class RoomKeyWithheldCode(ExtensibleEnum):
BLACKLISTED: "RoomKeyWithheldCode" = "m.blacklisted"
UNVERIFIED: "RoomKeyWithheldCode" = "m.unverified"
UNAUTHORIZED: "RoomKeyWithheldCode" = "m.unauthorised"
UNAVAILABLE: "RoomKeyWithheldCode" = "m.unavailable"
NO_OLM_SESSION: "RoomKeyWithheldCode" = "m.no_olm"
BEEPER_REDACTED: "RoomKeyWithheldCode" = "com.beeper.redacted"
@dataclass
class RoomKeyWithheldEventContent(SerializableAttrs):
algorithm: EncryptionAlgorithm
sender_key: IdentityKey
code: RoomKeyWithheldCode
reason: Optional[str] = None
room_id: Optional[RoomID] = None
session_id: Optional[SessionID] = None
@dataclass
class RoomKeyEventContent(SerializableAttrs):
algorithm: EncryptionAlgorithm
room_id: RoomID
session_id: SessionID
session_key: str
beeper_max_age_ms: Optional[int] = field(json="com.beeper.max_age_ms", default=None)
beeper_max_messages: Optional[int] = field(json="com.beeper.max_messages", default=None)
beeper_is_scheduled: Optional[bool] = field(json="com.beeper.is_scheduled", default=False)
class KeyRequestAction(ExtensibleEnum):
REQUEST: "KeyRequestAction" = "request"
CANCEL: "KeyRequestAction" = "request_cancellation"
@dataclass
class RequestedKeyInfo(SerializableAttrs):
algorithm: EncryptionAlgorithm
room_id: RoomID
sender_key: IdentityKey
session_id: SessionID
@dataclass
class RoomKeyRequestEventContent(SerializableAttrs):
action: KeyRequestAction
requesting_device_id: DeviceID
request_id: str
body: Optional[RequestedKeyInfo] = None
@dataclass(kw_only=True)
class ForwardedRoomKeyEventContent(RoomKeyEventContent, SerializableAttrs):
sender_key: IdentityKey
signing_key: SigningKey = attr.ib(metadata={"json": "sender_claimed_ed25519_key"})
forwarding_key_chain: List[str] = attr.ib(metadata={"json": "forwarding_curve25519_key_chain"})
ToDeviceEventContent = Union[
Obj,
EncryptedOlmEventContent,
RoomKeyWithheldEventContent,
RoomKeyEventContent,
RoomKeyRequestEventContent,
ForwardedRoomKeyEventContent,
BeeperRoomKeyAckEventContent,
]
to_device_event_content_map = {
EventType.TO_DEVICE_ENCRYPTED: EncryptedOlmEventContent,
EventType.ROOM_KEY_WITHHELD: RoomKeyWithheldEventContent,
EventType.ROOM_KEY_REQUEST: RoomKeyRequestEventContent,
EventType.ROOM_KEY: RoomKeyEventContent,
EventType.FORWARDED_ROOM_KEY: ForwardedRoomKeyEventContent,
EventType.BEEPER_ROOM_KEY_ACK: BeeperRoomKeyAckEventContent,
}
@dataclass
class ToDeviceEvent(BaseEvent, SerializableAttrs):
sender: UserID
content: ToDeviceEventContent
@classmethod
def deserialize(cls, data: JSON) -> "ToDeviceEvent":
try:
evt_type = EventType.find(data.get("type"), t_class=EventType.Class.TO_DEVICE)
data.setdefault("content", {})["__mautrix_event_type"] = evt_type
except ValueError:
return Obj(**data)
evt = super().deserialize(data)
evt.type = evt_type
return evt
@staticmethod
@deserializer(ToDeviceEventContent)
def deserialize_content(data: JSON) -> ToDeviceEventContent:
evt_type = data.pop("__mautrix_event_type", None)
content_type = to_device_event_content_map.get(evt_type, None)
if not content_type:
return Obj(**data)
return content_type.deserialize(data)
@dataclass
class ASToDeviceEvent(ToDeviceEvent, SerializableAttrs):
to_user_id: UserID
to_device_id: DeviceID
|
