# Copyright (c) 2020 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 Optional, Dict, Any, List, NewType from uuid import UUID from attr import dataclass import attr from mautrix.types import SerializableAttrs, SerializableEnum, ExtensibleEnum GroupID = NewType('GroupID', str) @dataclass(frozen=True, eq=False) class Address(SerializableAttrs): number: Optional[str] = None uuid: Optional[UUID] = None @property def is_valid(self) -> bool: return bool(self.number) or bool(self.uuid) @property def best_identifier(self) -> str: return str(self.uuid) if self.uuid else self.number def __eq__(self, other: 'Address') -> bool: if not isinstance(other, Address): return False if self.uuid and other.uuid: return self.uuid == other.uuid elif self.number and other.number: return self.number == other.number return False def __hash__(self) -> int: if self.uuid: return hash(self.uuid) return hash(self.number) @classmethod def parse(cls, value: str) -> 'Address': return Address(number=value) if value.startswith("+") else Address(uuid=UUID(value)) @dataclass class Account(SerializableAttrs): account_id: str device_id: int address: Address @dataclass class LinkSession(SerializableAttrs): uri: str session_id: str @dataclass class TrustLevel(SerializableEnum): TRUSTED_UNVERIFIED = "TRUSTED_UNVERIFIED" TRUSTED_VERIFIED = "TRUSTED_VERIFIED" UNTRUSTED = "UNTRUSTED" @dataclass class Identity(SerializableAttrs): trust_level: TrustLevel added: int safety_number: str qr_code_data: str @dataclass class GetIdentitiesResponse(SerializableAttrs): address: Address identities: List[Identity] @dataclass class Contact(SerializableAttrs): address: Address name: Optional[str] = None color: Optional[str] = None profile_key: Optional[str] = attr.ib(default=None, metadata={"json": "profileKey"}) message_expiration_time: int = attr.ib(default=0, metadata={"json": "messageExpirationTime"}) @dataclass class Capabilities(SerializableAttrs): gv2: bool = False storage: bool = False gv1_migration: bool = attr.ib(default=False, metadata={"json": "gv1-migration"}) @dataclass class Profile(SerializableAttrs): name: str = "" profile_name: str = "" avatar: str = "" identity_key: str = "" unidentified_access: str = "" unrestricted_unidentified_access: bool = False address: Optional[Address] = None expiration_time: int = 0 capabilities: Optional[Capabilities] = None @dataclass class Group(SerializableAttrs): group_id: GroupID = attr.ib(metadata={"json": "groupId"}) name: str = "Unknown group" # Sometimes "UPDATE" type: Optional[str] = None # Not always present members: List[Address] = attr.ib(factory=lambda: []) avatar_id: int = attr.ib(default=0, metadata={"json": "avatarId"}) @dataclass(kw_only=True) class GroupV2ID(SerializableAttrs): id: GroupID revision: Optional[int] = None class AccessControlMode(SerializableEnum): UNKNOWN = "UNKNOWN" ANY = "ANY" MEMBER = "MEMBER" ADMINISTRATOR = "ADMINISTRATOR" UNSATISFIABLE = "UNSATISFIABLE" UNRECOGNIZED = "UNRECOGNIZED" @dataclass class GroupAccessControl(SerializableAttrs): attributes: AccessControlMode = AccessControlMode.UNKNOWN link: AccessControlMode = AccessControlMode.UNKNOWN members: AccessControlMode = AccessControlMode.UNKNOWN class GroupMemberRole(SerializableEnum): UNKNOWN = "UNKNOWN" DEFAULT = "DEFAULT" ADMINISTRATOR = "ADMINISTRATOR" UNRECOGNIZED = "UNRECOGNIZED" @dataclass class GroupMember(SerializableAttrs): uuid: UUID joined_revision: int = 0 role: GroupMemberRole = GroupMemberRole.UNKNOWN @dataclass(kw_only=True) class GroupV2(GroupV2ID, SerializableAttrs): title: str avatar: Optional[str] = None timer: Optional[int] = None master_key: Optional[str] = attr.ib(default=None, metadata={"json": "masterKey"}) invite_link: Optional[str] = attr.ib(default=None, metadata={"json": "inviteLink"}) access_control: GroupAccessControl = attr.ib(factory=lambda: GroupAccessControl(), metadata={"json": "accessControl"}) members: List[Address] member_detail: List[GroupMember] = attr.ib(factory=lambda: [], metadata={"json": "memberDetail"}) pending_members: List[Address] = attr.ib(factory=lambda: [], metadata={"json": "pendingMembers"}) pending_member_detail: List[GroupMember] = attr.ib(factory=lambda: [], metadata={"json": "pendingMemberDetail"}) requesting_members: List[Address] = attr.ib(factory=lambda: [], metadata={"json": "requestingMembers"}) @dataclass class Attachment(SerializableAttrs): width: int = 0 height: int = 0 caption: Optional[str] = None preview: Optional[str] = None blurhash: Optional[str] = None voice_note: bool = attr.ib(default=False, metadata={"json": "voiceNote"}) content_type: Optional[str] = attr.ib(default=None, metadata={"json": "contentType"}) custom_filename: Optional[str] = attr.ib(default=None, metadata={"json": "customFilename"}) # Only for incoming id: Optional[str] = None incoming_filename: Optional[str] = attr.ib(default=None, metadata={"json": "storedFilename"}) digest: Optional[str] = None # Only for outgoing outgoing_filename: Optional[str] = attr.ib(default=None, metadata={"json": "filename"}) @dataclass class Quote(SerializableAttrs): id: int author: Address text: Optional[str] = None # TODO: attachments, mentions @dataclass(kw_only=True) class Reaction(SerializableAttrs): emoji: str remove: bool = False target_author: Address = attr.ib(metadata={"json": "targetAuthor"}) target_sent_timestamp: int = attr.ib(metadata={"json": "targetSentTimestamp"}) @dataclass class Sticker(SerializableAttrs): attachment: Attachment pack_id: str = attr.ib(metadata={"json": "packID"}) pack_key: str = attr.ib(metadata={"json": "packKey"}) sticker_id: int = attr.ib(metadata={"json": "stickerID"}) @dataclass class RemoteDelete(SerializableAttrs): target_sent_timestamp: int = attr.ib(metadata={"json": "targetSentTimestamp"}) @dataclass class Mention(SerializableAttrs): uuid: UUID length: int start: int = 0 @dataclass class MessageData(SerializableAttrs): timestamp: int body: Optional[str] = None quote: Optional[Quote] = None reaction: Optional[Reaction] = None attachments: List[Attachment] = attr.ib(factory=lambda: []) sticker: Optional[Sticker] = None mentions: List[Mention] = attr.ib(factory=lambda: []) group: Optional[Group] = None group_v2: Optional[GroupV2ID] = attr.ib(default=None, metadata={"json": "groupV2"}) end_session: bool = attr.ib(default=False, metadata={"json": "endSession"}) expires_in_seconds: int = attr.ib(default=0, metadata={"json": "expiresInSeconds"}) profile_key_update: bool = attr.ib(default=False, metadata={"json": "profileKeyUpdate"}) view_once: bool = attr.ib(default=False, metadata={"json": "viewOnce"}) remote_delete: Optional[RemoteDelete] = attr.ib(default=None, metadata={"json": "remoteDelete"}) @dataclass class SentSyncMessage(SerializableAttrs): message: MessageData timestamp: int expiration_start_timestamp: Optional[int] = attr.ib(default=None, metadata={ "json": "expirationStartTimestamp"}) is_recipient_update: bool = attr.ib(default=False, metadata={"json": "isRecipientUpdate"}) unidentified_status: Dict[str, bool] = attr.ib(factory=lambda: {}) destination: Optional[Address] = None class TypingAction(SerializableEnum): UNKNOWN = "UNKNOWN" STARTED = "STARTED" STOPPED = "STOPPED" @dataclass class TypingNotification(SerializableAttrs): action: TypingAction timestamp: int group_id: Optional[GroupID] = attr.ib(default=None, metadata={"json": "groupId"}) @dataclass class OwnReadReceipt(SerializableAttrs): sender: Address timestamp: int class ReceiptType(SerializableEnum): UNKNOWN = "UNKNOWN" DELIVERY = "DELIVERY" READ = "READ" VIEWED = "VIEWED" @dataclass class Receipt(SerializableAttrs): type: ReceiptType timestamps: List[int] when: int @dataclass class ContactSyncMeta(SerializableAttrs): id: Optional[str] = None @dataclass class ConfigItem(SerializableAttrs): present: bool = False @dataclass class ClientConfiguration(SerializableAttrs): read_receipts: Optional[ConfigItem] = attr.ib(factory=lambda: ConfigItem(), metadata={"json": "readReceipts"}) typing_indicators: Optional[ConfigItem] = attr.ib(factory=lambda: ConfigItem(), metadata={"json": "typingIndicators"}) link_previews: Optional[ConfigItem] = attr.ib(factory=lambda: ConfigItem(), metadata={"json": "linkPreviews"}) unidentified_delivery_indicators: Optional[ConfigItem] = attr.ib( factory=lambda: ConfigItem(), metadata={"json": "unidentifiedDeliveryIndicators"}) class StickerPackOperation(ExtensibleEnum): INSTALL = "INSTALL" # there are very likely others @dataclass class StickerPackOperations(SerializableAttrs): type: StickerPackOperation pack_id: str = attr.ib(metadata={"json": "packID"}) pack_key: str = attr.ib(metadata={"json": "packKey"}) @dataclass class SyncMessage(SerializableAttrs): sent: Optional[SentSyncMessage] = None typing: Optional[TypingNotification] = None read_messages: Optional[List[OwnReadReceipt]] = attr.ib(default=None, metadata={"json": "readMessages"}) contacts: Optional[ContactSyncMeta] = None groups: Optional[ContactSyncMeta] = None configuration: Optional[ClientConfiguration] = None # blocked_list: Optional[???] = attr.ib(default=None, metadata={"json": "blockedList"}) sticker_pack_operations: Optional[List[StickerPackOperations]] = attr.ib( default=None, metadata={"json": "stickerPackOperations"}) contacts_complete: bool = attr.ib(default=False, metadata={"json": "contactsComplete"}) class MessageType(SerializableEnum): CIPHERTEXT = "CIPHERTEXT" UNIDENTIFIED_SENDER = "UNIDENTIFIED_SENDER" RECEIPT = "RECEIPT" PREKEY_BUNDLE = "PREKEY_BUNDLE" KEY_EXCHANGE = "KEY_EXCHANGE" UNKNOWN = "UNKNOWN" @dataclass(kw_only=True) class Message(SerializableAttrs): username: str source: Address timestamp: int timestamp_iso: str = attr.ib(metadata={"json": "timestampISO"}) type: MessageType source_device: Optional[int] = attr.ib(metadata={"json": "sourceDevice"}, default=None) server_timestamp: Optional[int] = attr.ib(metadata={"json": "serverTimestamp"}, default=None) server_delivered_timestamp: int = attr.ib(metadata={"json": "serverDeliveredTimestamp"}) has_content: bool = attr.ib(metadata={"json": "hasContent"}, default=False) is_unidentified_sender: Optional[bool] = attr.ib(metadata={"json": "isUnidentifiedSender"}, default=None) has_legacy_message: bool = attr.ib(default=False, metadata={"json": "hasLegacyMessage"}) data_message: Optional[MessageData] = attr.ib(default=None, metadata={"json": "dataMessage"}) sync_message: Optional[SyncMessage] = attr.ib(default=None, metadata={"json": "syncMessage"}) typing: Optional[TypingNotification] = None receipt: Optional[Receipt] = None class ListenAction(SerializableEnum): STARTED = "started" STOPPED = "stopped" @dataclass class ListenEvent(SerializableAttrs): action: ListenAction username: str exception: Optional[str] = None