Bump version to 0.15.1

* Add quick 5 retries when connecting to Telegram

* Fix attempt initialisation

Co-authored-by: Tulir Asokan <tulir@maunium.net>

* Log only when retrying

Co-authored-by: Tulir Asokan <tulir@maunium.net>

---------

Co-authored-by: Tulir Asokan <tulir@maunium.net>

.github/workflows/python-lint.yml

@@ -9,14 +9,14 @@ jobs:
     - uses: actions/checkout@v2
     - uses: actions/setup-python@v2
       with:
-        python-version: "3.10"
+        python-version: "3.11"
     - uses: isort/isort-action@master
       with:
         sortPaths: "./mautrix_telegram"
     - uses: psf/black@stable
       with:
         src: "./mautrix_telegram"
-        version: "22.3.0"
+        version: "23.1.0"
     - name: pre-commit
       run: |
         pip install pre-commit

.gitignore

@@ -10,12 +10,19 @@ __pycache__
 /*.egg-info
 /.eggs
 
-/config.yaml
-/registration.yaml
+*.yaml
+!.pre-commit-config.yaml
+!example-config.yaml
+!/mautrix_telegram/web/provisioning/spec.yaml
+!/.github/workflows/*.yaml
+
+/start
+/mautrix
+/telethon
+
 *.log*
 *.db
+*.db-*
 /*.pickle
 *.bak
-/*.session
-/*.session-journal
 /*.json

.pre-commit-config.yaml

@@ -1,6 +1,6 @@
 repos:
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.1.0
+    rev: v4.4.0
     hooks:
       - id: trailing-whitespace
         exclude_types: [markdown]
@@ -8,13 +8,13 @@ repos:
       - id: check-yaml
       - id: check-added-large-files
   - repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 23.1.0
     hooks:
       - id: black
         language_version: python3
         files: ^mautrix_telegram/.*\.pyi?$
   - repo: https://github.com/PyCQA/isort
-    rev: 5.10.1
+    rev: 5.12.0
     hooks:
       - id: isort
         files: ^mautrix_telegram/.*\.pyi?$

CHANGELOG.md

@@ -1,3 +1,115 @@
+# v0.15.1 (2023-12-26)
+
+* Updated Telegram API to layer 169.
+* Updated Docker image to Alpine 3.19.
+* Fixed some potential cases where a portal room would be created for the
+  relaybot even if `ignore_unbridged_group_chat` was enabled.
+* Fixed member sync in groups with hidden members causing puppeted Matrix users
+  to be kicked even if they're still in the group.
+
+# v0.15.0 (2023-11-26)
+
+* Removed support for MSC2716 backfilling.
+* Added `add-contact` and `delete-contact` commands.
+* Updated Telegram API layer to 166.
+  * Includes receiving view-once media, blockquotes, quote replies and other
+    such things
+* Fixed AuthKeyNotFound errors not being handled and causing users to get stuck
+  in a non-logged-in state.
+
+# v0.14.2 (2023-09-19)
+
+* **Security:** Updated Pillow to 10.0.1.
+* Added support for double puppeting with arbitrary `as_token`s.
+  See [docs](https://docs.mau.fi/bridges/general/double-puppeting.html#appservice-method-new) for more info.
+* Added support for sending webm and tgs files as stickers.
+* Updated to Telegram API layer 161.
+* Fixed cached usernames for Telegram users being cleared incorrectly, leading
+  to mentions not being bridged as usernames.
+* Fixed reaction bridging failing if the server running the bridge was rebooted
+  less than 12 hours ago.
+
+# v0.14.1 (2023-06-26)
+
+### Added
+* Added option to delete megolm sessions that were received before the
+  automatic ratcheting options were introduced.
+* Added config option to use IPv6 for Telegram connection
+  (thanks to [@exciler] in [#920]).
+
+### Improved
+* Dropped support for Python 3.8.
+* Updated Docker image to Alpine 3.18.
+* Added timeout for forward backfills to prevent it from getting stuck
+  permanently.
+
+### Fixed
+* Fixed `bridge.filter.users` config option not being read correctly.
+* Fixed proxy support to use python-socks instead of pysocks.
+
+[@exciler]: https://github.com/exciler
+[#920]: https://github.com/mautrix/telegram/pull/920
+
+# v0.14.0 (2023-05-26)
+
+### Added
+* Added fallback messages for calls and premium gifts.
+* Added options to automatically ratchet/delete megolm sessions to minimize
+  access to old messages.
+* Added option to not set room name/avatar even in encrypted rooms.
+* Implemented appservice pinging using MSC2659.
+* Added option to disable or filter bridging direct chats
+  (thanks to [@Steffo99] in [#892]).
+* Added options to specify different limits for forward and catchup backfilling
+  depending on chat type.
+
+### Improved
+* Improved handling logouts and certain connection errors.
+* Changed reaction bridging to preserve timestamps.
+* Disabled creating portals for DMs that don't have any messages when
+  `sync_direct_chats` is enabled.
+
+### Fixed
+* Fixed syncing mute status when portal is created through incoming message
+  rather than in startup sync.
+* Fixed bridge incorrectly trusting member list and kicking users when
+  supergroup has member list hidden.
+* Fixed sending messages after creating groups from Matrix using relaybot
+  instead of puppet (thanks to [@maltee1] in [#902]).
+
+[@Steffo99]: https://github.com/Steffo99
+[@maltee1]: https://github.com/maltee1
+[#892]: https://github.com/mautrix/telegram/pull/892
+[#902]: https://github.com/mautrix/telegram/pull/902
+
+# v0.13.0 (2023-02-26)
+
+### Added
+* Added `allow_contact_info` config option to specify whether personal names
+  and avatars for other users should be bridged.
+  * The option is only safe to enable on single-user instances, using it
+    anywhere else will cause ghost user profiles to flip back and forth between
+    personal and default ones.
+* Added config option to notify Matrix room if bridging an incoming message
+  fails.
+
+### Improved
+* Updated Docker image to Alpine 3.17.
+* Updated to Telegram API layer 152.
+* Improved handling users getting logged out.
+* Removed support for creating accounts, as Telegram only allows requesting SMS
+  login codes on the official mobile clients now.
+* Replaced moviepy with calling ffmpeg directly for generating video thumbnails.
+
+### Fixed
+* Fixed handling Telegram chat upgrades when backfilling is enabled.
+* Fixed file transfers failing if transfering the thumbnail fails.
+* Fixed bridging unnamed files with unrecognized mime types.
+* Fixed enqueueing more backfill.
+* Fixed timestamps not being saved in `telegram_file` table.
+* Fixed issues with old events being replayed if the bridge was shut down
+  uncleanly.
+
 # v0.12.2 (2022-11-26)
 
 ### Added

Dockerfile

@@ -1,9 +1,11 @@
-FROM dock.mau.dev/tulir/lottieconverter:alpine-3.16
+FROM dock.mau.dev/tulir/lottieconverter:alpine-3.19
 
 RUN apk add --no-cache \
       python3 py3-pip py3-setuptools py3-wheel \
       py3-pillow \
       py3-aiohttp \
+      py3-asyncpg \
+      py3-aiosqlite \
       py3-magic \
       py3-ruamel.yaml \
       py3-commonmark \
@@ -13,17 +15,10 @@ RUN apk add --no-cache \
       # Indirect dependencies
       py3-idna \
       py3-rsa \
-      #moviepy
-        py3-decorator \
-        py3-tqdm \
-        py3-requests \
-        #py3-proglog \
-        #imageio
-          py3-numpy \
       #py3-telethon \ (outdated)
-        # Optional for socks proxies
-        py3-pysocks \
         py3-pyaes \
+        py3-aiodns \
+        py3-python-socks \
         # cryptg
           py3-cffi \
           py3-qrcode \
@@ -47,13 +42,13 @@ COPY requirements.txt /opt/mautrix-telegram/requirements.txt
 COPY optional-requirements.txt /opt/mautrix-telegram/optional-requirements.txt
 WORKDIR /opt/mautrix-telegram
 RUN apk add --virtual .build-deps python3-dev libffi-dev build-base \
- && pip3 install /cryptg-*.whl \
- && pip3 install --no-cache-dir -r requirements.txt -r optional-requirements.txt \
+ && pip3 install --break-system-packages /cryptg-*.whl \
+ && pip3 install --break-system-packages --no-cache-dir -r requirements.txt -r optional-requirements.txt \
  && apk del .build-deps \
  && rm -f /cryptg-*.whl
 
 COPY . /opt/mautrix-telegram
-RUN apk add git && pip3 install --no-cache-dir .[all] && apk del git \
+RUN apk add git && pip3 install --break-system-packages --no-cache-dir .[all] && apk del git \
   # This doesn't make the image smaller, but it's needed so that the `version` command works properly
   && cp mautrix_telegram/example-config.yaml . && rm -rf mautrix_telegram .git build
 

dev-requirements.txt

@@ -1,3 +1,3 @@
 pre-commit>=2.10.1,<3
 isort>=5.10.1,<6
-black>=22.3,<23
+black>=23,<24

mautrix_telegram/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "0.12.2"
+__version__ = "0.15.1"
 __author__ = "Tulir Asokan <tulir@maunium.net>"

mautrix_telegram/__main__.py

@@ -39,6 +39,8 @@ from .abstract_user import AbstractUser  # isort: skip
 class TelegramBridge(Bridge):
     module = "mautrix_telegram"
     name = "mautrix-telegram"
+    beeper_service_name = "telegram"
+    beeper_network_name = "telegram"
     command = "python -m mautrix-telegram"
     description = "A Matrix-Telegram puppeting bridge."
     repo_url = "https://github.com/mautrix/telegram"
@@ -50,6 +52,7 @@ class TelegramBridge(Bridge):
 
     config: Config
     bot: Bot | None
+    matrix: MatrixHandler
     public_website: PublicBridgeWebsite | None
     provisioning_api: ProvisioningAPI | None
 
@@ -101,8 +104,6 @@ class TelegramBridge(Bridge):
         self.log.info("Finished re-sending bridge info state events")
 
     def prepare_stop(self) -> None:
-        for puppet in Puppet.by_custom_mxid.values():
-            puppet.stop()
         self.add_shutdown_actions(user.stop() for user in User.by_tgid.values())
         if self.bot:
             self.add_shutdown_actions(self.bot.stop())

mautrix_telegram/abstract_user.py

@@ -22,7 +22,7 @@ import logging
 import platform
 import time
 
-from telethon.errors import UnauthorizedError
+from telethon.errors import AuthKeyError, UnauthorizedError
 from telethon.network import (
     Connection,
     ConnectionTcpFull,
@@ -38,6 +38,7 @@ from telethon.tl.types import (
     PeerChannel,
     PeerChat,
     PeerUser,
+    PhoneCallRequested,
     TypeUpdate,
     UpdateChannel,
     UpdateChannelUserTyping,
@@ -54,6 +55,7 @@ from telethon.tl.types import (
     UpdateNewChannelMessage,
     UpdateNewMessage,
     UpdateNotifySettings,
+    UpdatePhoneCall,
     UpdatePinnedChannelMessages,
     UpdatePinnedDialogs,
     UpdatePinnedMessages,
@@ -63,8 +65,8 @@ from telethon.tl.types import (
     UpdateShort,
     UpdateShortChatMessage,
     UpdateShortMessage,
+    UpdateUser,
     UpdateUserName,
-    UpdateUserPhoto,
     UpdateUserStatus,
     UpdateUserTyping,
     User,
@@ -75,6 +77,7 @@ from telethon.tl.types import (
 from mautrix.appservice import AppService
 from mautrix.errors import MatrixError
 from mautrix.types import PresenceState, UserID
+from mautrix.util import background_task
 from mautrix.util.logging import TraceLogger
 from mautrix.util.opt_prometheus import Counter, Histogram
 
@@ -205,6 +208,8 @@ class AbstractUser(ABC):
         sysversion = self.config["telegram.device_info.system_version"]
         appversion = self.config["telegram.device_info.app_version"]
         connection, proxy = self._proxy_settings
+        if proxy:
+            self.log.debug(f"Using proxy setting: {proxy}")
 
         assert isinstance(session, Session)
 
@@ -232,21 +237,27 @@ class AbstractUser(ABC):
             loop=self.loop,
             base_logger=base_logger,
             update_error_callback=self._telethon_update_error_callback,
+            use_ipv6=self.config["telegram.connection.use_ipv6"],
         )
         self.client.add_event_handler(self._update_catch)
 
+    @abstractmethod
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+        pass
+
     async def _telethon_update_error_callback(self, err: Exception) -> None:
+        if isinstance(err, (UnauthorizedError, AuthKeyError)):
+            background_task.create(self.on_signed_out(err))
+            return
         if self.config["telegram.exit_on_update_error"]:
             self.log.critical(f"Stopping due to update handling error {type(err).__name__}")
             self.bridge.manual_stop(50)
         else:
-            if isinstance(err, UnauthorizedError):
-                self.log.warning("Not recreating Telethon update loop")
-                return
-            self.log.info("Recreating Telethon update loop in 60 seconds")
+            self.log.info("Recreating Telethon connection in 60 seconds")
             await asyncio.sleep(60)
-            self.log.debug("Now recreating Telethon update loop")
-            self.client._updates_handle = self.loop.create_task(self.client._update_loop())
+            self.log.debug("Now recreating Telethon connection")
+            await self.stop()
+            await self.start()
 
     @abstractmethod
     async def update(self, update: TypeUpdate) -> bool:
@@ -295,7 +306,18 @@ class AbstractUser(ABC):
     async def start(self, delete_unless_authenticated: bool = False) -> AbstractUser:
         if not self.client:
             await self._init_client()
-        await self.client.connect()
+        attempts = 1
+        while True:
+            try:
+                await self.client.connect()
+            except Exception:
+                attempts += 1
+                if attempts > 10:
+                    raise
+                self.log.exception("Exception connecting to Telegram, retrying in 5s...")
+                await asyncio.sleep(5)
+            else:
+                break
         self.log.debug(f"{'Bot' if self.is_relaybot else self.mxid} connected: {self.connected}")
         return self
 
@@ -320,7 +342,7 @@ class AbstractUser(ABC):
     async def _update(self, update: TypeUpdate) -> None:
         if isinstance(update, UpdateShort):
             update = update.update
-        asyncio.create_task(self._handle_entity_updates(getattr(update, "_entities", {})))
+        background_task.create(self._handle_entity_updates(getattr(update, "_entities", {})))
         if isinstance(
             update,
             (
@@ -337,6 +359,8 @@ class AbstractUser(ABC):
             await self.delete_message(update)
         elif isinstance(update, UpdateDeleteChannelMessages):
             await self.delete_channel_message(update)
+        elif isinstance(update, UpdatePhoneCall):
+            await self.update_phone_call(update)
         elif isinstance(update, UpdateMessageReactions):
             await self.update_reactions(update)
         elif isinstance(update, (UpdateChatUserTyping, UpdateChannelUserTyping, UpdateUserTyping)):
@@ -351,7 +375,7 @@ class AbstractUser(ABC):
             await self.update_default_banned_rights(update)
         elif isinstance(update, (UpdatePinnedMessages, UpdatePinnedChannelMessages)):
             await self.update_pinned_messages(update)
-        elif isinstance(update, (UpdateUserName, UpdateUserPhoto)):
+        elif isinstance(update, (UpdateUserName, UpdateUser)):
             await self.update_others_info(update)
         elif isinstance(update, UpdateReadHistoryOutbox):
             await self.update_read_receipt(update)
@@ -443,6 +467,7 @@ class AbstractUser(ABC):
             return
 
         if not portal or not portal.mxid:
+            # TODO This explodes on channels because the field is channel_id
             self.log.debug(f"Dropping own read receipt in unknown chat ({update.peer})")
             return
 
@@ -500,18 +525,23 @@ class AbstractUser(ABC):
         except Exception:
             self.log.exception("Failed to handle entity updates")
 
-    async def update_others_info(self, update: UpdateUserName | UpdateUserPhoto) -> None:
+    async def update_others_info(self, update: UpdateUserName | UpdateUser) -> None:
         # TODO duplication not checked
         puppet = await pu.Puppet.get_by_tgid(TelegramID(update.user_id))
         if isinstance(update, UpdateUserName):
-            puppet.username = update.username
+            if len(update.usernames) > 1:
+                self.log.warning(
+                    "Got update with multiple usernames (%s) for %s, only saving first one",
+                    update.usernames,
+                    update.user_id,
+                )
+            puppet.username = update.usernames[0].username if update.usernames else None
             if await puppet.update_displayname(self, update):
                 await puppet.save()
                 await puppet.update_portals_meta()
-        elif isinstance(update, UpdateUserPhoto):
-            if await puppet.update_avatar(self, update.photo):
-                await puppet.save()
-                await puppet.update_portals_meta()
+        elif isinstance(update, UpdateUser):
+            info = await self.client.get_entity(puppet.peer)
+            await puppet.update_info(self, info)
         else:
             self.log.warning(f"Unexpected other user info update: {type(update)}")
 
@@ -606,6 +636,19 @@ class AbstractUser(ABC):
             return
         await portal.handle_telegram_reactions(self, TelegramID(update.msg_id), update.reactions)
 
+    async def update_phone_call(self, update: UpdatePhoneCall) -> None:
+        self.log.debug("Phone call update %s", update)
+        if not isinstance(update.phone_call, PhoneCallRequested):
+            return
+        tgid = TelegramID(update.phone_call.participant_id)
+        if tgid == self.tgid:
+            tgid = update.phone_call.admin_id
+        portal = await po.Portal.get_by_tgid(tgid, tg_receiver=self.tgid, peer_type="user")
+        if not portal or not portal.mxid or not portal.allow_bridging:
+            return
+        sender = await pu.Puppet.get_by_tgid(TelegramID(update.phone_call.admin_id))
+        await portal.handle_telegram_direct_call(self, sender, update)
+
     async def update_channel(self, update: UpdateChannel) -> None:
         portal = await po.Portal.get_by_tgid(TelegramID(update.channel_id))
         if not portal:
@@ -615,24 +658,32 @@ class AbstractUser(ABC):
             await portal.delete_telegram_user(self.tgid, sender=None)
         elif chan := getattr(update, "mau_channel", None):
             if not portal.mxid:
-                asyncio.create_task(self._delayed_create_channel(chan))
+                if (
+                    not self.is_relaybot
+                    or not self.config["bridge.relaybot.ignore_unbridged_group_chat"]
+                ):
+                    background_task.create(self._delayed_create_channel(chan))
             else:
                 self.log.debug("Updating channel info with data fetched by Telethon")
                 await portal.update_info(self, chan)
                 await portal.invite_to_matrix(self.mxid)
 
     async def _delayed_create_channel(self, chan: Channel) -> None:
-        self.log.debug("Waiting 5 seconds before handling UpdateChannel for non-existent portal")
+        self.log.debug(
+            f"Waiting 5 seconds before handling UpdateChannel for non-existent portal {chan.id}"
+        )
         await asyncio.sleep(5)
         portal = await po.Portal.get_by_tgid(TelegramID(chan.id))
         if portal.mxid:
             self.log.debug(
-                "Portal started existing after waiting 5 seconds, dropping UpdateChannel"
+                "Portal started existing after waiting 5 seconds, "
+                f"dropping UpdateChannel for {portal.tgid}"
             )
             return
         else:
             self.log.info(
-                "Creating Matrix room with data fetched by Telethon due to UpdateChannel"
+                f"Creating Matrix room for {portal.tgid}"
+                " with data fetched by Telethon due to UpdateChannel"
             )
             await portal.create_matrix_room(self, chan, invites=[self.mxid])
 
@@ -644,7 +695,15 @@ class AbstractUser(ABC):
         if not portal:
             return
         elif portal and not portal.allow_bridging:
-            self.log.debug(f"Ignoring message in portal {portal.tgid_log} (bridging disallowed)")
+            self.log.debug(
+                f"Ignoring message {update.id} in portal {portal.tgid_log} (bridging disallowed)"
+            )
+            return
+
+        if not portal.mxid and getattr(original_update, "mau_left_channel", False):
+            self.log.debug(
+                f"Ignoring message {update.id} in portal {portal.tgid_log} because user isn't in the chat"
+            )
             return
 
         if self.is_relaybot:
@@ -668,6 +727,22 @@ class AbstractUser(ABC):
             self.log.debug("Ignoring relaybot-sent message %s to %s", update.id, portal.tgid_log)
             return
 
+        task = self._call_portal_message_handler(update, original_update, portal, sender)
+        if portal.backfill_lock.locked:
+            self.log.debug(
+                f"{portal.tgid_log} is backfill locked, moving incoming message to async task"
+            )
+            background_task.create(task)
+        else:
+            await task
+
+    async def _call_portal_message_handler(
+        self,
+        update: UpdateMessageContent,
+        original_update: UpdateMessage,
+        portal: po.Portal,
+        sender: pu.Puppet,
+    ) -> None:
         await portal.backfill_lock.wait(f"update {update.id}")
 
         if isinstance(update, MessageService):
@@ -681,7 +756,7 @@ class AbstractUser(ABC):
                 await self.unregister_portal(update.action.chat_id, update.action.chat_id)
                 await self.register_portal(portal)
                 return
-            self.log.trace(
+            self.log.debug(
                 "Handling action %s to %s by %d",
                 update.action,
                 portal.tgid_log,

mautrix_telegram/bot.py

@@ -19,7 +19,12 @@ from typing import TYPE_CHECKING, Awaitable, Callable, Literal
 import logging
 import time
 
-from telethon.errors import ChannelInvalidError, ChannelPrivateError
+from telethon.errors import (
+    AuthKeyError,
+    ChannelInvalidError,
+    ChannelPrivateError,
+    UnauthorizedError,
+)
 from telethon.tl.functions.channels import GetChannelsRequest, GetParticipantRequest
 from telethon.tl.functions.messages import GetChatsRequest, GetFullChatRequest
 from telethon.tl.patched import Message, MessageService
@@ -145,6 +150,10 @@ class Bot(AbstractUser):
         await self.post_login()
         return self
 
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError) -> None:
+        self.log.fatal("Relay bot got signed out, crashing bridge", exc_info=err)
+        self.bridge.manual_stop(51)
+
     async def post_login(self) -> None:
         await self.init_permissions()
         info = await self.client.get_me()
@@ -386,7 +395,7 @@ class Bot(AbstractUser):
         def reply(reply_text: str) -> Awaitable[Message]:
             return self.client.send_message(message.chat_id, reply_text, reply_to=message.id)
 
-        if command == "start":
+        if command == "start" and message.is_private:
             pcm = self.config["bridge.relaybot.private_chat.message"]
             if pcm:
                 await reply(pcm)

mautrix_telegram/commands/handler.py

@@ -159,6 +159,7 @@ def command_handler(
     needs_admin: bool = False,
     management_only: bool = False,
     name: str | None = None,
+    aliases: list[str] | None = None,
     help_text: str = "",
     help_args: str = "",
     help_section: HelpSection = None,
@@ -167,6 +168,7 @@ def command_handler(
         _func,
         _handler_class=CommandHandler,
         name=name,
+        aliases=aliases,
         help_text=help_text,
         help_args=help_args,
         help_section=help_section,

mautrix_telegram/commands/portal/admin.py

@@ -39,8 +39,6 @@ async def clear_db_cache(evt: CommandEvent) -> EventID:
         await evt.reply("Cleared portal cache")
     elif section == "puppet":
         pu.Puppet.by_tgid = {}
-        for puppet in pu.Puppet.by_custom_mxid.values():
-            puppet.stop()
         pu.Puppet.by_custom_mxid = {}
         await asyncio.gather(
             *[puppet.try_start() async for puppet in pu.Puppet.all_with_custom_mxid()]
@@ -69,8 +67,6 @@ async def reload_user(evt: CommandEvent) -> EventID:
     if not user:
         return await evt.reply("User not found")
     puppet = await pu.Puppet.get_by_custom_mxid(mxid)
-    if puppet:
-        puppet.stop()
     await user.stop()
     del u.User.by_tgid[user.tgid]
     del u.User.by_mxid[user.mxid]

mautrix_telegram/commands/portal/bridge.py

@@ -21,6 +21,7 @@ import asyncio
 from telethon.tl.types import ChannelForbidden, ChatForbidden
 
 from mautrix.types import EventID, RoomID
+from mautrix.util import background_task
 
 from ... import portal as po
 from ...types import TelegramID
@@ -55,7 +56,9 @@ async def bridge(evt: CommandEvent) -> EventID:
         return await evt.reply(f"{that_this} room is already a portal room.")
 
     if not await user_has_power_level(room_id, evt.az.intent, evt.sender, "bridge"):
-        return await evt.reply(f"You do not have the permissions to bridge {that_this} room.")
+        return await evt.reply(
+            f"You do not have the permissions to bridge {that_this.lower()} room."
+        )
 
     # The /id bot command provides the prefixed ID, so we assume
     tgid_str = evt.args[0]
@@ -184,7 +187,7 @@ async def confirm_bridge(evt: CommandEvent) -> EventID | None:
         if not ok:
             return None
         elif coro:
-            asyncio.create_task(coro)
+            background_task.create(coro)
             await evt.reply("Cleaning up previous portal room...")
     elif portal.mxid:
         evt.sender.command_status = None
@@ -251,7 +254,7 @@ async def _locked_confirm_bridge(
     await portal.save()
     await portal.update_bridge_info()
 
-    asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
+    background_task.create(portal.update_matrix_room(user, entity, levels=levels))
 
     await warn_missing_power(levels, evt)
 

mautrix_telegram/commands/portal/create_chat.py

@@ -65,19 +65,11 @@ async def create(evt: CommandEvent) -> EventID:
         about=about,
         encrypted=encrypted,
     )
-    invites, errors = await portal.get_telegram_users_in_matrix_room(evt.sender, pre_create=True)
-    if len(errors) > 0:
-        error_list = "\n".join(f"* [{mxid}](https://matrix.to/#/{mxid})" for mxid in errors)
-        await evt.reply(
-            f"Failed to add the following users to the chat:\n\n{error_list}\n\n"
-            "You can try `$cmdprefix+sp search -r <username>` to help the bridge find "
-            "those users."
-        )
 
     await warn_missing_power(levels, evt)
 
     try:
-        await portal.create_telegram_chat(evt.sender, invites=invites, supergroup=supergroup)
+        await portal.create_telegram_chat(evt.sender, supergroup=supergroup)
     except ValueError as e:
         await portal.delete()
         return await evt.reply(e.args[0])

mautrix_telegram/commands/telegram/auth.py

@@ -22,7 +22,6 @@ import io
 from telethon.errors import (
     AccessTokenExpiredError,
     AccessTokenInvalidError,
-    FirstNameInvalidError,
     FloodWaitError,
     PasswordHashInvalidError,
     PhoneCodeExpiredError,
@@ -31,14 +30,12 @@ from telethon.errors import (
     PhoneNumberBannedError,
     PhoneNumberFloodError,
     PhoneNumberInvalidError,
-    PhoneNumberOccupiedError,
     PhoneNumberUnoccupiedError,
     SessionPasswordNeededError,
 )
 from telethon.tl.types import User
 
 from mautrix.client import Client
-from mautrix.errors import MForbidden
 from mautrix.types import (
     EventID,
     ImageInfo,
@@ -47,6 +44,7 @@ from mautrix.types import (
     TextMessageEventContent,
     UserID,
 )
+from mautrix.util import background_task
 from mautrix.util.format_duration import format_duration as fmt_duration
 
 from ... import user as u
@@ -94,70 +92,6 @@ async def ping_bot(evt: CommandEvent) -> EventID:
     )
 
 
-@command_handler(
-    needs_auth=False,
-    management_only=True,
-    help_section=SECTION_AUTH,
-    help_args="<_phone_> <_full name_>",
-    help_text="Register to Telegram",
-)
-async def register(evt: CommandEvent) -> EventID:
-    if await evt.sender.is_logged_in():
-        return await evt.reply("You are already logged in.")
-    elif len(evt.args) < 1:
-        return await evt.reply("**Usage:** `$cmdprefix+sp register <phone> <full name>`")
-
-    phone_number = evt.args[0]
-    if len(evt.args) == 2:
-        full_name = evt.args[1], ""
-    else:
-        full_name = " ".join(evt.args[1:-1]), evt.args[-1]
-
-    await _request_code(
-        evt,
-        phone_number,
-        {
-            "next": enter_code_register,
-            "action": "Register",
-            "full_name": full_name,
-        },
-    )
-    return await evt.reply(
-        "By signing up for Telegram, you agree to "
-        "the terms of service: https://telegram.org/tos"
-    )
-
-
-async def enter_code_register(evt: CommandEvent) -> EventID:
-    if len(evt.args) == 0:
-        return await evt.reply("**Usage:** `$cmdprefix+sp <code>`")
-    try:
-        await evt.sender.ensure_started(even_if_no_session=True)
-        first_name, last_name = evt.sender.command_status["full_name"]
-        user = await evt.sender.client.sign_up(evt.args[0], first_name, last_name)
-        asyncio.create_task(evt.sender.post_login(user, first_login=True))
-        evt.sender.command_status = None
-        return await evt.reply(f"Successfully registered to Telegram.")
-    except PhoneNumberOccupiedError:
-        return await evt.reply(
-            "That phone number has already been registered. "
-            "You can log in with `$cmdprefix+sp login`."
-        )
-    except FirstNameInvalidError:
-        return await evt.reply("Invalid name. Please set a Matrix displayname before registering.")
-    except PhoneCodeExpiredError:
-        return await evt.reply(
-            "Phone code expired. Try again with `$cmdprefix+sp register <phone>`."
-        )
-    except PhoneCodeInvalidError:
-        return await evt.reply("Invalid phone code.")
-    except Exception:
-        evt.log.exception("Error sending phone code")
-        return await evt.reply(
-            "Unhandled exception while sending code. Check console for more details."
-        )
-
-
 @command_handler(
     needs_auth=False,
     management_only=True,
@@ -317,7 +251,7 @@ async def _request_code(
     except PhoneNumberUnoccupiedError:
         return await evt.reply(
             "That phone number has not been registered. "
-            "Please register with `$cmdprefix+sp register <phone>`."
+            "Please sign up to Telegram using an official mobile client first."
         )
     except PhoneNumberInvalidError:
         return await evt.reply("That phone number is not valid.")
@@ -432,7 +366,7 @@ async def _finish_sign_in(evt: CommandEvent, user: User, login_as: u.User = None
             f"[{existing_user.displayname}] (https://matrix.to/#/{existing_user.mxid})"
             " was logged out from the account."
         )
-    asyncio.create_task(login_as.post_login(user, first_login=True))
+    background_task.create(login_as.post_login(user, first_login=True))
     evt.sender.command_status = None
     name = f"@{user.username}" if user.username else f"+{user.phone}"
     if login_as != evt.sender:

mautrix_telegram/commands/telegram/misc.py

@@ -18,8 +18,8 @@ from __future__ import annotations
 from typing import cast
 import base64
 import codecs
-import math
 import re
+import shlex
 
 from aiohttp import ClientSession, InvalidURL
 from telethon.errors import (
@@ -29,10 +29,10 @@ from telethon.errors import (
     InviteHashInvalidError,
     InviteRequestSentError,
     OptionsTooMuchError,
-    TakeoutInitDelayError,
     UserAlreadyParticipantError,
 )
 from telethon.tl.functions.channels import JoinChannelRequest
+from telethon.tl.functions.contacts import DeleteByPhonesRequest, ImportContactsRequest
 from telethon.tl.functions.messages import (
     CheckChatInviteRequest,
     GetBotCallbackAnswerRequest,
@@ -42,12 +42,14 @@ from telethon.tl.functions.messages import (
 from telethon.tl.patched import Message
 from telethon.tl.types import (
     InputMediaDice,
+    InputPhoneContact,
     MessageMediaGame,
     MessageMediaPoll,
     TypeInputPeer,
     TypeUpdates,
     User as TLUser,
 )
+from telethon.tl.types.contacts import ImportedContacts
 from telethon.tl.types.messages import BotCallbackAnswer
 
 from mautrix.types import EventID, Format
@@ -134,15 +136,16 @@ async def search(evt: CommandEvent) -> EventID:
 
 @command_handler(
     help_section=SECTION_CREATING_PORTALS,
-    help_args="<_identifier_>",
-    help_text="Open a private chat with the given Telegram user. The identifier is "
-    "either the internal user ID, the username or the phone number. "
-    "**N.B.** The phone numbers you start chats with must already be in "
-    "your contacts.",
+    help_args="<_username_>",
+    help_text=(
+        "Open a private chat with the given Telegram user. You can also use a "
+        "phone number instead of username, but you must have the number in "
+        "your Telegram contacts for that to work."
+    ),
 )
 async def pm(evt: CommandEvent) -> EventID:
     if len(evt.args) == 0:
-        return await evt.reply("**Usage:** `$cmdprefix+sp pm <user identifier>`")
+        return await evt.reply("**Usage:** `$cmdprefix+sp pm <username>`")
 
     try:
         id = "".join(evt.args).translate({ord(c): None for c in "+()- "})
@@ -160,6 +163,76 @@ async def pm(evt: CommandEvent) -> EventID:
     return await evt.reply(f"Created private chat room with {displayname}")
 
 
+async def _handle_contact(source: AbstractUser, user: TLUser) -> str:
+    puppet: pu.Puppet = await pu.Puppet.get_by_tgid(user.id)
+    await puppet.update_info(source, user)
+
+    params = []
+    if user.username:
+        params.append(f"[@{user.username}](https://t.me/{user.username})")
+    if user.phone:
+        params.append(f"+{user.phone}")
+    params.append(f"ID `{user.id}`")
+    params_str = " / ".join(params)
+    return f"[{puppet.displayname}](https://matrix.to/#/{puppet.mxid}): {params_str}"
+
+
+@command_handler(
+    help_section=SECTION_CREATING_PORTALS,
+    help_args="<_phone_> <_first name_> <_last name_>",
+    help_text="Add a phone number to your contacts on Telegram",
+)
+async def add_contact(evt: CommandEvent) -> EventID:
+    if len(evt.args) < 3:
+        return await evt.reply(
+            "**Usage:** `$cmdprefix+sp add-contact <phone> <first name> <last name>`"
+        )
+    try:
+        names = shlex.split(" ".join(evt.args[1:]))
+    except ValueError as e:
+        return await evt.reply(
+            f"Failed to parse names (use shell quoting for names with spaces): {e}"
+        )
+    if len(names) != 2:
+        return await evt.reply(
+            "Wrong number of names, must have first and last name "
+            "(use shell quoting for names with spaces)"
+        )
+    res: ImportedContacts = await evt.sender.client(
+        ImportContactsRequest(
+            contacts=[
+                InputPhoneContact(
+                    client_id=1, phone=evt.args[0], first_name=names[0], last_name=names[1]
+                )
+            ]
+        )
+    )
+    if res.retry_contacts:
+        return await evt.reply("Failed to import contacts")
+    elif not res.users:
+        return await evt.reply("Contact imported, but user not found on Telegram")
+    imported_str = "\n".join(
+        [f"* {await _handle_contact(evt.sender, user)}" for user in res.users]
+    )
+    return await evt.reply(f"Imported contacts:\n\n{imported_str}")
+
+
+@command_handler(
+    help_section=SECTION_CREATING_PORTALS,
+    help_args="<_phones..._>",
+    help_text="Remove phone numbers from your contacts on Telegram.",
+    aliases=["remove-contact", "delete-contacts", "remove-contacts"],
+)
+async def delete_contact(evt: CommandEvent) -> EventID:
+    if len(evt.args) == 0:
+        return await evt.reply("**Usage:** `$cmdprefix+sp delete-contact <phones...>`")
+    ok = await evt.sender.client(DeleteByPhonesRequest(phones=evt.args))
+    if ok:
+        return await evt.reply("Contacts deleted")
+    else:
+        return await evt.reply("Contacts not deleted?")
+
+
 async def _join(
     evt: CommandEvent, identifier: str, link_type: str
 ) -> tuple[TypeUpdates | None, EventID | None]:
@@ -439,14 +512,5 @@ async def backfill(evt: CommandEvent) -> None:
     if not evt.config["bridge.backfill.normal_groups"] and portal.peer_type == "chat":
         await evt.reply("Backfilling normal groups is disabled in the bridge config")
         return
-    if portal.backfill_msc2716:
-        messages_per_batch = evt.config["bridge.backfill.incremental.messages_per_batch"]
-        batches = math.ceil(limit / messages_per_batch)
-        rounded = ""
-        if batches * messages_per_batch != limit:
-            rounded = f" (rounded message limit to {batches}*{messages_per_batch})"
-        await portal.enqueue_backfill(evt.sender, priority=0, max_batches=batches)
-        await evt.reply(f"Backfill queued{rounded}")
-    else:
-        output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
-        await evt.reply(output)
+    output = await portal.forward_backfill(evt.sender, initial=False, override_limit=limit)
+    await evt.reply(output)

mautrix_telegram/config.py

@@ -101,6 +101,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.displayname_preference")
         copy("bridge.displayname_max_length")
         copy("bridge.allow_avatar_remove")
+        copy("bridge.allow_contact_info")
 
         copy("bridge.max_initial_member_sync")
         copy("bridge.max_member_count")
@@ -147,9 +148,19 @@ class Config(BaseBridgeConfig):
         copy("bridge.animated_emoji.args.width")
         copy("bridge.animated_emoji.args.height")
         copy("bridge.animated_emoji.args.fps")
-        copy("bridge.private_chat_portal_meta")
+        if isinstance(self.get("bridge.private_chat_portal_meta", "default"), bool):
+            base["bridge.private_chat_portal_meta"] = (
+                "always" if self["bridge.private_chat_portal_meta"] else "default"
+            )
+        else:
+            copy("bridge.private_chat_portal_meta")
+            if base["bridge.private_chat_portal_meta"] not in ("default", "always", "never"):
+                base["bridge.private_chat_portal_meta"] = "default"
+        copy("bridge.disable_reply_fallbacks")
+        copy("bridge.cross_room_replies")
         copy("bridge.delivery_receipts")
         copy("bridge.delivery_error_reports")
+        copy("bridge.incoming_bridge_error_reports")
         copy("bridge.message_status_events")
         copy("bridge.resend_bridge_info")
         copy("bridge.mute_bridging")
@@ -160,12 +171,29 @@ class Config(BaseBridgeConfig):
         copy("bridge.kick_on_logout")
         copy("bridge.always_read_joined_telegram_notice")
         copy("bridge.backfill.enable")
-        copy("bridge.backfill.msc2716")
-        copy("bridge.backfill.double_puppet_backfill")
         copy("bridge.backfill.normal_groups")
         copy("bridge.backfill.unread_hours_threshold")
-        copy("bridge.backfill.forward.initial_limit")
-        copy("bridge.backfill.forward.sync_limit")
+        if "bridge.backfill.forward" in self:
+            initial_limit = self.get("bridge.backfill.forward.initial_limit", 10)
+            sync_limit = self.get("bridge.backfill.forward.sync_limit", 100)
+            base["bridge.backfill.forward_limits.initial.user"] = initial_limit
+            base["bridge.backfill.forward_limits.initial.normal_group"] = initial_limit
+            base["bridge.backfill.forward_limits.initial.supergroup"] = initial_limit
+            base["bridge.backfill.forward_limits.initial.channel"] = initial_limit
+            base["bridge.backfill.forward_limits.sync.user"] = sync_limit
+            base["bridge.backfill.forward_limits.sync.normal_group"] = sync_limit
+            base["bridge.backfill.forward_limits.sync.supergroup"] = sync_limit
+            base["bridge.backfill.forward_limits.sync.channel"] = sync_limit
+        else:
+            copy("bridge.backfill.forward_limits.initial.user")
+            copy("bridge.backfill.forward_limits.initial.normal_group")
+            copy("bridge.backfill.forward_limits.initial.supergroup")
+            copy("bridge.backfill.forward_limits.initial.channel")
+            copy("bridge.backfill.forward_limits.sync.user")
+            copy("bridge.backfill.forward_limits.sync.normal_group")
+            copy("bridge.backfill.forward_limits.sync.supergroup")
+            copy("bridge.backfill.forward_limits.sync.channel")
+        copy("bridge.backfill.forward_timeout")
         copy("bridge.backfill.incremental.messages_per_batch")
         copy("bridge.backfill.incremental.post_batch_delay")
         copy("bridge.backfill.incremental.max_batches.user")
@@ -195,6 +223,7 @@ class Config(BaseBridgeConfig):
 
         copy("bridge.filter.mode")
         copy("bridge.filter.list")
+        copy("bridge.filter.users")
 
         copy("bridge.command_prefix")
 
@@ -239,6 +268,7 @@ class Config(BaseBridgeConfig):
         copy("telegram.connection.retry_delay")
         copy("telegram.connection.flood_sleep_threshold")
         copy("telegram.connection.request_retries")
+        copy("telegram.connection.use_ipv6")
 
         copy("telegram.device_info.device_model")
         copy("telegram.device_info.system_version")

mautrix_telegram/db/backfill_queue.py

@@ -24,7 +24,7 @@ from asyncpg import Record
 from attr import dataclass
 
 from mautrix.types import UserID
-from mautrix.util.async_db import Database
+from mautrix.util.async_db import Connection, Database
 
 from ..types import TelegramID
 
@@ -169,8 +169,8 @@ class Backfill:
         )
 
     @classmethod
-    async def delete_all(cls, user_mxid: UserID) -> None:
-        await cls.db.execute("DELETE FROM backfill_queue WHERE user_mxid=$1", user_mxid)
+    async def delete_all(cls, user_mxid: UserID, conn: Connection | None = None) -> None:
+        await (conn or cls.db).execute("DELETE FROM backfill_queue WHERE user_mxid=$1", user_mxid)
 
     @classmethod
     async def delete_for_portal(cls, tgid: int, tg_receiver: int) -> None:
@@ -186,7 +186,7 @@ class Backfill:
           AND type=$4
           AND dispatch_time IS NULL
           AND completed_at IS NULL
-        RETURNING {self.columns_str}
+        RETURNING queue_id, {self.columns_str}
         """
         q = f"""
         INSERT INTO backfill_queue ({self.columns_str})

mautrix_telegram/db/portal.py

@@ -167,7 +167,10 @@ class Portal:
             "UPDATE portal SET tgid=$1, tg_receiver=$1, peer_type=$2 "
             "WHERE tgid=$3 AND tg_receiver=$3"
         )
-        await self.db.execute(q, id, peer_type, self.tgid)
+        clear_queue = "DELETE FROM backfill_queue WHERE portal_tgid=$1 AND portal_tg_receiver=$2"
+        async with self.db.acquire() as conn, conn.transaction():
+            await conn.execute(clear_queue, self.tgid, self.tg_receiver)
+            await conn.execute(q, id, peer_type, self.tgid)
         self.tgid = id
         self.tg_receiver = id
         self.peer_type = peer_type

mautrix_telegram/db/puppet.py

@@ -48,6 +48,7 @@ class Puppet:
     avatar_url: ContentURI | None
     name_set: bool
     avatar_set: bool
+    contact_info_set: bool
     is_bot: bool | None
     is_channel: bool
     is_premium: bool
@@ -68,7 +69,7 @@ class Puppet:
     columns: ClassVar[str] = (
         "id, is_registered, displayname, displayname_source, displayname_contact, "
         "displayname_quality, disable_updates, username, phone, photo_id, avatar_url, "
-        "name_set, avatar_set, is_bot, is_channel, is_premium, "
+        "name_set, avatar_set, contact_info_set, is_bot, is_channel, is_premium, "
         "custom_mxid, access_token, next_batch, base_url"
     )
 
@@ -108,6 +109,7 @@ class Puppet:
             self.avatar_url,
             self.name_set,
             self.avatar_set,
+            self.contact_info_set,
             self.is_bot,
             self.is_channel,
             self.is_premium,
@@ -122,8 +124,9 @@ class Puppet:
         UPDATE puppet
         SET is_registered=$2, displayname=$3, displayname_source=$4, displayname_contact=$5,
             displayname_quality=$6, disable_updates=$7, username=$8, phone=$9, photo_id=$10,
-            avatar_url=$11, name_set=$12, avatar_set=$13, is_bot=$14, is_channel=$15,
-            is_premium=$16, custom_mxid=$17, access_token=$18, next_batch=$19, base_url=$20
+            avatar_url=$11, name_set=$12, avatar_set=$13, contact_info_set=$14, is_bot=$15,
+            is_channel=$16, is_premium=$17, custom_mxid=$18, access_token=$19, next_batch=$20,
+            base_url=$21
         WHERE id=$1
         """
         await self.db.execute(q, *self._values)
@@ -133,9 +136,9 @@ class Puppet:
         INSERT INTO puppet (
             id, is_registered, displayname, displayname_source, displayname_contact,
             displayname_quality, disable_updates, username, phone, photo_id, avatar_url, name_set,
-            avatar_set, is_bot, is_channel, is_premium, custom_mxid, access_token, next_batch,
-            base_url
+            avatar_set, contact_info_set, is_bot, is_channel, is_premium, custom_mxid,
+            access_token, next_batch, base_url
         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18,
-                  $19, $20)
+                  $19, $20, $21)
         """
         await self.db.execute(q, *self._values)

mautrix_telegram/db/telegram_file.py

@@ -92,9 +92,9 @@ class TelegramFile:
 
     async def insert(self) -> None:
         q = (
-            "INSERT INTO telegram_file (id, mxc, mime_type, was_converted, size, width, height, "
-            "                           thumbnail, decryption_info) "
-            "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
+            "INSERT INTO telegram_file (id, mxc, mime_type, was_converted, timestamp,"
+            "                           size, width, height, thumbnail, decryption_info) "
+            "VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)"
         )
         await self.db.execute(
             q,
@@ -102,6 +102,7 @@ class TelegramFile:
             self.mxc,
             self.mime_type,
             self.was_converted,
+            self.timestamp,
             self.size,
             self.width,
             self.height,

mautrix_telegram/db/telethon_session.py

@@ -123,19 +123,55 @@ class PgSession(MemorySession):
         date = datetime.datetime.utcfromtimestamp(row["date"])
         return updates.State(row["pts"], row["qts"], date, row["seq"], row["unread_count"])
 
+    _set_update_state_q = """
+    INSERT INTO telethon_update_state (session_id, entity_id, pts, qts, date, seq, unread_count)
+    VALUES ($1, $2, $3, $4, $5, $6, $7)
+    ON CONFLICT (session_id, entity_id) DO UPDATE SET
+        pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
+        unread_count=excluded.unread_count
+    """
+
     async def set_update_state(self, entity_id: int, row: updates.State) -> None:
-        q = """
-        INSERT INTO telethon_update_state(session_id, entity_id, pts, qts, date, seq, unread_count)
-        VALUES ($1, $2, $3, $4, $5, $6, $7)
-        ON CONFLICT (session_id, entity_id) DO UPDATE SET
-            pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
-            unread_count=excluded.unread_count
-        """
+        q = self._set_update_state_q
         ts = row.date.timestamp()
         await self.db.execute(
             q, self.session_id, entity_id, row.pts, row.qts, ts, row.seq, row.unread_count
         )
 
+    async def set_update_states(self, rows: list[tuple[int, updates.State]]) -> None:
+        rows = [
+            (
+                self.session_id,
+                entity_id,
+                row.pts,
+                row.qts,
+                row.date.timestamp(),
+                row.seq,
+                row.unread_count,
+            )
+            for entity_id, row in rows
+        ]
+        if self.db.scheme == Scheme.POSTGRES:
+            q = """
+            INSERT INTO telethon_update_state (
+                session_id, entity_id, pts, qts, date, seq, unread_count
+            )
+            VALUES (
+                $1,
+                unnest($2::bigint[]), unnest($3::bigint[]), unnest($4::bigint[]),
+                unnest($5::bigint[]), unnest($6::bigint[]), unnest($7::integer[])
+            )
+            ON CONFLICT (session_id, entity_id) DO UPDATE SET
+                pts=excluded.pts, qts=excluded.qts, date=excluded.date, seq=excluded.seq,
+                unread_count=excluded.unread_count
+            """
+            _, entity_ids, ptses, qtses, timestamps, seqs, unread_counts = zip(*rows)
+            await self.db.execute(
+                q, self.session_id, entity_ids, ptses, qtses, timestamps, seqs, unread_counts
+            )
+        else:
+            await self.db.executemany(self._set_update_state_q, rows)
+
     async def delete_update_state(self, entity_id: int) -> None:
         q = "DELETE FROM telethon_update_state WHERE session_id=$1 AND entity_id=$2"
         await self.db.execute(q, self.session_id, entity_id)

mautrix_telegram/db/upgrade/__init__.py

@@ -20,4 +20,5 @@ from . import (
     v15_backfill_anchor_id,
     v16_backfill_type,
     v17_message_find_recent,
+    v18_puppet_contact_info_set,
 )

mautrix_telegram/db/upgrade/v00_latest_revision.py

@@ -15,7 +15,7 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from mautrix.util.async_db import Connection, Scheme
 
-latest_version = 17
+latest_version = 18
 
 
 async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
@@ -113,6 +113,7 @@ async def create_latest_tables(conn: Connection, scheme: Scheme) -> int:
             avatar_url          TEXT,
             name_set            BOOLEAN NOT NULL DEFAULT false,
             avatar_set          BOOLEAN NOT NULL DEFAULT false,
+            contact_info_set    BOOLEAN NOT NULL DEFAULT false,
             is_bot              BOOLEAN,
             is_channel          BOOLEAN NOT NULL DEFAULT false,
             is_premium          BOOLEAN NOT NULL DEFAULT false,

mautrix_telegram/db/upgrade/v18_puppet_contact_info_set.py

@@ -0,0 +1,25 @@
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2022 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+from mautrix.util.async_db import Connection
+
+from . import upgrade_table
+
+
+@upgrade_table.register(description="Add contact_info_set column to puppet table")
+async def upgrade_v18(conn: Connection) -> None:
+    await conn.execute(
+        "ALTER TABLE puppet ADD COLUMN contact_info_set BOOLEAN NOT NULL DEFAULT false"
+    )

mautrix_telegram/db/user.py

@@ -21,9 +21,10 @@ from asyncpg import Record
 from attr import dataclass
 
 from mautrix.types import UserID
-from mautrix.util.async_db import Database, Scheme
+from mautrix.util.async_db import Connection, Database, Scheme
 
 from ..types import TelegramID
+from .backfill_queue import Backfill
 
 fake_db = Database.create("") if TYPE_CHECKING else None
 
@@ -73,6 +74,20 @@ class User:
     async def delete(self) -> None:
         await self.db.execute('DELETE FROM "user" WHERE mxid=$1', self.mxid)
 
+    async def remove_tgid(self) -> None:
+        async with self.db.acquire() as conn, conn.transaction():
+            if self.tgid:
+                await conn.execute('DELETE FROM contact WHERE "user"=$1', self.tgid)
+                await conn.execute('DELETE FROM user_portal WHERE "user"=$1', self.tgid)
+            await Backfill.delete_all(self.mxid, conn=conn)
+            self.tgid = None
+            self.tg_username = None
+            self.tg_phone = None
+            self.is_bot = False
+            self.is_premium = False
+            self.saved_contacts = 0
+            await self.save(conn=conn)
+
     @property
     def _values(self):
         return (
@@ -85,13 +100,13 @@ class User:
             self.saved_contacts,
         )
 
-    async def save(self) -> None:
+    async def save(self, conn: Connection | None = None) -> None:
         q = """
         UPDATE "user" SET tgid=$2, tg_username=$3, tg_phone=$4, is_bot=$5, is_premium=$6,
                           saved_contacts=$7
         WHERE mxid=$1
         """
-        await self.db.execute(q, *self._values)
+        await (conn or self.db).execute(q, *self._values)
 
     async def insert(self) -> None:
         q = """

mautrix_telegram/example-config.yaml

@@ -40,7 +40,7 @@ appservice:
 
     # The full URI to the database. SQLite and Postgres are supported.
     # Format examples:
-    #   SQLite:   sqlite:///filename.db
+    #   SQLite:   sqlite:filename.db
     #   Postgres: postgres://username:password@hostname/dbname
     database: postgres://username:password@hostname/dbname
     # Additional arguments for asyncpg.create_pool() or sqlite3.connect()
@@ -145,6 +145,9 @@ bridge:
     # as there's no way to determine whether an avatar is removed or just hidden from some users. If
     # you're on a single-user instance, this should be safe to enable.
     allow_avatar_remove: false
+    # Should contact names and profile pictures be allowed?
+    # This is only safe to enable on single-user instances.
+    allow_contact_info: false
 
     # Maximum number of members to sync per portal when starting up. Other members will be
     # synced when they send messages. The maximum is 10000, after which the Telegram server
@@ -271,6 +274,27 @@ bridge:
         # Enable key sharing? If enabled, key requests for rooms where users are in will be fulfilled.
         # You must use a client that supports requesting keys from other users to use this feature.
         allow_key_sharing: false
+        # Options for deleting megolm sessions from the bridge.
+        delete_keys:
+            # Beeper-specific: delete outbound sessions when hungryserv confirms
+            # that the user has uploaded the key to key backup.
+            delete_outbound_on_ack: false
+            # Don't store outbound sessions in the inbound table.
+            dont_store_outbound: false
+            # Ratchet megolm sessions forward after decrypting messages.
+            ratchet_on_decrypt: false
+            # Delete fully used keys (index >= max_messages) after decrypting messages.
+            delete_fully_used_on_decrypt: false
+            # Delete previous megolm sessions from same device when receiving a new one.
+            delete_prev_on_new_session: false
+            # Delete megolm sessions received from a device when the device is deleted.
+            delete_on_device_delete: false
+            # Periodically delete megolm sessions when 2x max_age has passed since receiving the session.
+            periodically_delete_expired: false
+            # Delete inbound megolm sessions that don't have the received_at field used for
+            # automatic ratcheting and expired session deletion. This is meant as a migration
+            # to delete old keys prior to the bridge update.
+            delete_outdated_inbound: false
         # What level of device verification should be required from users?
         #
         # Valid levels:
@@ -306,14 +330,27 @@ bridge:
             # default.
             messages: 100
 
-    # Whether or not to explicitly set the avatar and room name for private
-    # chat portal rooms. This will be implicitly enabled if encryption.default is true.
-    private_chat_portal_meta: false
+            # Disable rotating keys when a user's devices change?
+            # You should not enable this option unless you understand all the implications.
+            disable_device_change_key_rotation: false
+
+    # Whether to explicitly set the avatar and room name for private chat portal rooms.
+    # If set to `default`, this will be enabled in encrypted rooms and disabled in unencrypted rooms.
+    # If set to `always`, all DM rooms will have explicit names and avatars set.
+    # If set to `never`, DM rooms will never have names and avatars set.
+    private_chat_portal_meta: default
+    # Disable generating reply fallbacks? Some extremely bad clients still rely on them,
+    # but they're being phased out and will be completely removed in the future.
+    disable_reply_fallbacks: false
+    # Should cross-chat replies from Telegram be bridged? Most servers and clients don't support this.
+    cross_room_replies: false
     # Whether or not the bridge should send a read receipt from the bridge bot when a message has
     # been sent to Telegram.
     delivery_receipts: false
     # Whether or not delivery errors should be reported as messages in the Matrix room.
     delivery_error_reports: false
+    # Should errors in incoming message handling send a message to the Matrix room?
+    incoming_bridge_error_reports: false
     # Whether the bridge should send the message status as a custom com.beeper.message_send_status event.
     message_status_events: false
     # Set this to true to tell the bridge to re-send m.bridge events to all rooms on the next run.
@@ -342,20 +379,6 @@ bridge:
     backfill:
         # Allow backfilling at all?
         enable: true
-        # Use MSC2716 for backfilling?
-        #
-        # This requires a server with MSC2716 support, which is currently an experimental feature in Synapse.
-        # It can be enabled by setting experimental_features -> msc2716_enabled to true in homeserver.yaml.
-        msc2716: false
-        # Use double puppets for backfilling?
-        #
-        # If using MSC2716, the double puppets must be in the appservice's user ID namespace
-        # (because the bridge can't use the double puppet access token with batch sending).
-        #
-        # Even without MSC2716, bridging old messages with correct timestamps requires the double
-        # puppets to be in an appservice namespace, or the server to be modified to allow
-        # overriding timestamps anyway.
-        double_puppet_backfill: false
         # Whether or not to enable backfilling in normal groups.
         # Normal groups have numerous technical problems in Telegram, and backfilling normal groups
         # will likely cause problems if there are multiple Matrix users in the group.
@@ -365,17 +388,26 @@ bridge:
         # Set to -1 to let any chat be unread.
         unread_hours_threshold: 720
 
-        # Forward backfilling limits. These apply to both MSC2716 and legacy backfill.
+        # Forward backfilling limits.
         #
         # Using a negative initial limit is not recommended, as it would try to backfill everything in a single batch.
-        # MSC2716 and the incremental settings are meant for backfilling everything incrementally rather than at once.
-        forward:
+        forward_limits:
             # Number of messages to backfill immediately after creating a portal.
-            initial_limit: 10
+            initial:
+                user: 50
+                normal_group: 100
+                supergroup: 10
+                channel: 10
             # Number of messages to backfill when syncing chats.
-            sync_limit: 100
+            sync:
+                user: 100
+                normal_group: 100
+                supergroup: 100
+                channel: 100
+        # Timeout for forward backfills in seconds. If you have a high limit, you'll have to increase this too.
+        forward_timeout: 900
 
-        # Settings for incremental backfill of history. These only apply when using MSC2716.
+        # Settings for incremental backfill of history. These only apply to Beeper, as upstream abandoned MSC2716.
         incremental:
             # Maximum number of messages to backfill per batch.
             messages_per_batch: 100
@@ -453,7 +485,6 @@ bridge:
     # Filter rooms that can/can't be bridged. Can also be managed using the `filter` and
     # `filter-mode` management commands.
     #
-    # Filters do not affect direct chats.
     # An empty blacklist will essentially disable the filter.
     filter:
         # Filter mode to use. Either "blacklist" or "whitelist".
@@ -462,6 +493,11 @@ bridge:
         mode: blacklist
         # The list of group/channel IDs to filter.
         list: []
+        # How to handle direct chats:
+        # If users is "null", direct chats will follow the previous settings.
+        # If users is "true", direct chats will always be bridged.
+        # If users is "false", direct chats will never be bridged.
+        users: true
 
     # The prefix for commands. Only required in non-management rooms.
     command_prefix: "!tg"
@@ -562,6 +598,8 @@ telegram:
         # is not recommended, since some requests can always trigger a call fail (such as searching
         # for messages).
         request_retries: 5
+        # Use IPv6 for Telethon connection
+        use_ipv6: false
 
     # Device info sent to Telegram.
     device_info:

mautrix_telegram/formatter/__init__.py

@@ -1,2 +1,2 @@
 from .from_matrix import matrix_reply_to_telegram, matrix_to_telegram
-from .from_telegram import telegram_to_matrix
+from .from_telegram import telegram_text_to_matrix_html, telegram_to_matrix

mautrix_telegram/formatter/from_matrix/parser.py

@@ -82,14 +82,6 @@ class MatrixParser(BaseMatrixParser[TelegramMessage]):
         prefix = "#" * length + " "
         return TelegramMessage.join(children, "").prepend(prefix).format(TelegramEntityType.BOLD)
 
-    async def blockquote_to_fstring(
-        self, node: HTMLNode, ctx: RecursionContext
-    ) -> TelegramMessage:
-        msg = await self.tag_aware_parse_node(node, ctx)
-        children = msg.trim().split("\n")
-        children = [child.prepend("> ") for child in children]
-        return TelegramMessage.join(children, "\n")
-
     async def color_to_fstring(self, msg: TelegramMessage, color: str) -> TelegramMessage:
         return msg
 

mautrix_telegram/formatter/from_telegram.py

@@ -176,6 +176,21 @@ async def _convert_custom_emoji(
                 entities[i] = ReuploadedCustomEmoji(entity, custom_emojis[entity.document_id])
 
 
+async def telegram_text_to_matrix_html(
+    source: au.AbstractUser,
+    text: str,
+    entities: list[TypeMessageEntity],
+    client: MautrixTelegramClient | None = None,
+) -> str:
+    if not entities:
+        return escape(text).replace("\n", "<br/>")
+    await _convert_custom_emoji(source, entities, client=client)
+    text = add_surrogate(text)
+    html = await _telegram_entities_to_matrix_catch(text, entities)
+    html = del_surrogate(html)
+    return html
+
+
 async def telegram_to_matrix(
     evt: Message | SponsoredMessage,
     source: au.AbstractUser,
@@ -192,10 +207,10 @@ async def telegram_to_matrix(
     )
     entities = override_entities or evt.entities
     if entities:
-        await _convert_custom_emoji(source, entities, client=client)
         content.format = Format.HTML
-        html = await _telegram_entities_to_matrix_catch(add_surrogate(content.body), entities)
-        content.formatted_body = del_surrogate(html)
+        content.formatted_body = await telegram_text_to_matrix_html(
+            source, content.body, entities, client=client
+        )
 
     if require_html:
         content.ensure_has_html()
@@ -333,7 +348,11 @@ async def _telegram_entities_to_matrix(
         last_offset = relative_offset + (0 if skip_entity else entity.length)
     html.append(text_to_html(text[last_offset:]))
 
-    return "".join(html)
+    html_string = "".join(html)
+    # Remove redundant <br>'s after block tags
+    html_string = html_string.replace("</blockquote><br/>", "</blockquote>")
+    html_string = html_string.replace("</pre><br/>", "</pre>")
+    return html_string
 
 
 def _parse_pre(html: list[str], entity_text: str, language: str) -> bool:

mautrix_telegram/matrix.py

@@ -61,21 +61,6 @@ class MatrixHandler(BaseMatrixHandler):
 
         self._previously_typing = {}
 
-    async def check_versions(self) -> None:
-        await super().check_versions()
-        if self.config["bridge.backfill.msc2716"] and not (
-            support := self.versions.supports("org.matrix.msc2716")
-        ):
-            self.log.fatal(
-                "Backfilling with MSC2716 is enabled in bridge config, but "
-                + (
-                    "batch sending is not enabled on homeserver"
-                    if support is False
-                    else "homeserver does not support batch sending"
-                )
-            )
-            sys.exit(18)
-
     async def handle_puppet_group_invite(
         self,
         room_id: RoomID,
@@ -135,20 +120,8 @@ class MatrixHandler(BaseMatrixHandler):
             levels.users[self.az.bot_mxid] = 100 if invited_by_level >= 100 else invited_by_level
             await double_puppet.intent.set_power_levels(room_id, levels)
 
-        invites, errors = await portal.get_telegram_users_in_matrix_room(
-            invited_by, pre_create=True
-        )
-        if len(errors) > 0:
-            error_list = "\n".join(f"* [{mxid}](https://matrix.to/#/{mxid})" for mxid in errors)
-            await portal.az.intent.send_notice(
-                room_id,
-                f"Failed to add the following users to the chat:\n\n{error_list}\n\n"
-                "You can try `$cmdprefix+sp search -r <username>` to help the bridge find "
-                "those users.",
-            )
-
         try:
-            await portal.create_telegram_chat(invited_by, invites=invites, supergroup=True)
+            await portal.create_telegram_chat(invited_by, supergroup=True)
         except ValueError as e:
             await portal.delete()
             await portal.az.intent.send_notice(room_id, e.args[0])
@@ -186,7 +159,8 @@ class MatrixHandler(BaseMatrixHandler):
             await portal.main_intent.kick_user(
                 room_id,
                 user.mxid,
-                "This chat does not have a bot relaying messages for unauthenticated users.",
+                "This chat does not have a bot on the Telegram side for relaying messages sent by"
+                " unauthenticated Matrix users.",
             )
             return
 

mautrix_telegram/portal_util/message_convert.py

@@ -34,6 +34,8 @@ from telethon.tl.types import (
     DocumentAttributeVideo,
     Game,
     InputPhotoFileLocation,
+    InputStickerSetID,
+    InputStickerSetShortName,
     Message,
     MessageEntityPre,
     MessageMediaContact,
@@ -42,11 +44,14 @@ from telethon.tl.types import (
     MessageMediaGame,
     MessageMediaGeo,
     MessageMediaGeoLive,
+    MessageMediaInvoice,
     MessageMediaPhoto,
     MessageMediaPoll,
+    MessageMediaStory,
     MessageMediaUnsupported,
     MessageMediaVenue,
     MessageMediaWebPage,
+    MessageReplyStoryHeader,
     PeerChannel,
     PeerUser,
     Photo,
@@ -104,6 +109,7 @@ class DocAttrs(NamedTuple):
     mime_type: str | None
     is_sticker: bool
     sticker_alt: str | None
+    sticker_pack_ref: dict | None
     width: int
     height: int
     is_gif: bool
@@ -142,6 +148,8 @@ class TelegramMessageConverter:
             MessageMediaUnsupported: self._convert_unsupported,
             MessageMediaGame: self._convert_game,
             MessageMediaContact: self._convert_contact,
+            MessageMediaStory: self._convert_story,
+            MessageMediaInvoice: self._convert_invoice,
         }
         self._allowed_media = tuple(self._media_converters.keys())
 
@@ -252,20 +260,48 @@ class TelegramMessageConverter:
     ) -> None:
         if not evt.reply_to:
             return
+        elif isinstance(evt.reply_to, MessageReplyStoryHeader):
+            return
+
+        if evt.reply_to.quote and content.msgtype.is_text:
+            content.ensure_has_html()
+            quote_html = await formatter.telegram_text_to_matrix_html(
+                source, evt.reply_to.quote_text, evt.reply_to.quote_entities
+            )
+            content.formatted_body = (
+                f"<blockquote data-telegram-partial-reply>{quote_html}</blockquote>"
+                f"{content.formatted_body}"
+            )
+
         space = (
             evt.peer_id.channel_id
             if isinstance(evt, Message) and isinstance(evt.peer_id, PeerChannel)
             else source.tgid
         )
+        if evt.reply_to.reply_to_peer_id and evt.reply_to.reply_to_peer_id != evt.peer_id:
+            if not self.config["bridge.cross_room_replies"]:
+                return
+            space = (
+                evt.reply_to.reply_to_peer_id.channel_id
+                if isinstance(evt.reply_to.reply_to_peer_id, PeerChannel)
+                else source.tgid
+            )
+
         reply_to_id = TelegramID(evt.reply_to.reply_to_msg_id)
         msg = await DBMessage.get_one_by_tgid(reply_to_id, space)
-        if not msg or msg.mx_room != self.portal.mxid:
+        no_fallback = no_fallback or self.config["bridge.disable_reply_fallbacks"]
+        if not msg:
+            # TODO try to find room ID when generating deterministic ID for cross-room reply
             if deterministic_id:
                 content.set_reply(self.deterministic_event_id(space, reply_to_id))
             return
+        elif msg.mx_room != self.portal.mxid and not self.config["bridge.cross_room_replies"]:
+            return
         elif not isinstance(content, TextMessageEventContent) or no_fallback:
             # Not a text message, just set the reply metadata and return
             content.set_reply(msg.mxid)
+            if msg.mx_room != self.portal.mxid:
+                content.relates_to.in_reply_to["room_id"] = msg.mx_room
             return
 
         # Text message, try to fetch original message to generate reply fallback.
@@ -280,6 +316,8 @@ class TelegramMessageConverter:
         except Exception:
             self.log.exception("Failed to get event to add reply fallback")
             content.set_reply(msg.mxid)
+        if msg.mx_room != self.portal.mxid:
+            content.relates_to.in_reply_to["room_id"] = msg.mx_room
 
     @staticmethod
     def _photo_size_key(photo: TypePhotoSize) -> int:
@@ -409,6 +447,8 @@ class TelegramMessageConverter:
             mimetype=file.mime_type,
             size=self._photo_size_key(largest_size),
         )
+        if media.spoiler:
+            info["fi.mau.telegram.spoiler"] = True
         ext = sane_mimetypes.guess_extension(file.mime_type)
         name = f"disappearing_image{ext}" if media.ttl_seconds else f"image{ext}"
         content = MediaMessageEventContent(
@@ -426,9 +466,21 @@ class TelegramMessageConverter:
         return ConvertedMessage(
             content=content,
             caption=caption_content,
-            disappear_seconds=media.ttl_seconds,
+            disappear_seconds=self._adjust_ttl(media.ttl_seconds),
         )
 
+    @staticmethod
+    def _adjust_ttl(ttl: int | None) -> int | None:
+        if not ttl:
+            return None
+        elif ttl == 2147483647:
+            # View-once media, set low TTL
+            return 15
+        else:
+            # Increase media TTL because it's supposed to be counted from opening the media,
+            # but we can only count it from read receipt.
+            return ttl * 5
+
     async def _convert_document(
         self,
         source: au.AbstractUser,
@@ -494,12 +546,15 @@ class TelegramMessageConverter:
                 info["fi.mau.telegram.gif"] = True
             else:
                 info["fi.mau.telegram.animated_sticker"] = True
+            info["fi.mau.gif"] = True
             info["fi.mau.loop"] = True
             info["fi.mau.autoplay"] = True
             info["fi.mau.hide_controls"] = True
             info["fi.mau.no_audio"] = True
+        if evt.media.spoiler:
+            info["fi.mau.telegram.spoiler"] = True
         if not name:
-            ext = sane_mimetypes.guess_extension(file.mime_type)
+            ext = sane_mimetypes.guess_extension(file.mime_type) or ""
             name = "unnamed_file" + ext
 
         content = MediaMessageEventContent(
@@ -532,7 +587,7 @@ class TelegramMessageConverter:
             type=event_type,
             content=content,
             caption=caption_content,
-            disappear_seconds=evt.media.ttl_seconds,
+            disappear_seconds=self._adjust_ttl(evt.media.ttl_seconds),
         )
 
     @staticmethod
@@ -694,10 +749,33 @@ class TelegramMessageConverter:
             )
         return ConvertedMessage(content=content)
 
+    @staticmethod
+    async def _convert_story(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
+        content = await formatter.telegram_to_matrix(
+            evt, source, client, override_text="Stories are not yet supported"
+        )
+        content.msgtype = MessageType.NOTICE
+        content["fi.mau.telegram.unsupported"] = True
+        return ConvertedMessage(content=content)
+
+    @staticmethod
+    async def _convert_invoice(
+        source: au.AbstractUser, evt: Message, client: MautrixTelegramClient, **_
+    ) -> ConvertedMessage:
+        content = await formatter.telegram_to_matrix(
+            evt, source, client, override_text="Invoices are not yet supported"
+        )
+        content.msgtype = MessageType.NOTICE
+        content["fi.mau.telegram.unsupported"] = True
+        return ConvertedMessage(content=content)
+
 
 def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAttrs:
     name, mime_type, is_sticker, sticker_alt, width, height = None, None, False, None, 0, 0
     is_gif, is_audio, is_voice, duration, waveform = False, False, False, 0, bytes()
+    sticker_pack_ref = None
     for attr in attributes:
         if isinstance(attr, DocumentAttributeFilename):
             name = name or attr.file_name
@@ -705,6 +783,13 @@ def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAt
         elif isinstance(attr, DocumentAttributeSticker):
             is_sticker = True
             sticker_alt = attr.alt
+            if isinstance(attr.stickerset, InputStickerSetID):
+                sticker_pack_ref = {
+                    "id": str(attr.stickerset.id),
+                    "access_hash": str(attr.stickerset.access_hash),
+                }
+            elif isinstance(attr.stickerset, InputStickerSetShortName):
+                sticker_pack_ref = {"short_name": attr.stickerset.short_name}
         elif isinstance(attr, DocumentAttributeAnimated):
             is_gif = True
         elif isinstance(attr, DocumentAttributeVideo):
@@ -718,17 +803,18 @@ def _parse_document_attributes(attributes: list[TypeDocumentAttribute]) -> DocAt
             waveform = decode_waveform(attr.waveform) if attr.waveform else b""
 
     return DocAttrs(
-        name,
-        mime_type,
-        is_sticker,
-        sticker_alt,
-        width,
-        height,
-        is_gif,
-        is_audio,
-        is_voice,
-        duration,
-        waveform,
+        name=name,
+        mime_type=mime_type,
+        is_sticker=is_sticker,
+        sticker_alt=sticker_alt,
+        sticker_pack_ref=sticker_pack_ref,
+        width=width,
+        height=height,
+        is_gif=is_gif,
+        is_audio=is_audio,
+        is_voice=is_voice,
+        duration=duration,
+        waveform=waveform,
     )
 
 
@@ -754,6 +840,13 @@ def _parse_document_meta(
         mime_type = file.mime_type or document.mime_type
     info = ImageInfo(size=file.size, mimetype=mime_type)
 
+    if attrs.is_sticker:
+        info["fi.mau.telegram.sticker"] = {
+            "alt": attrs.sticker_alt,
+            "id": str(document.id),
+            "pack": attrs.sticker_pack_ref,
+        }
+
     if attrs.mime_type and not file.was_converted:
         file.mime_type = attrs.mime_type or file.mime_type
     if file.width and file.height:
@@ -773,6 +866,10 @@ def _parse_document_meta(
             size=file.thumbnail.size,
         )
     elif attrs.is_sticker:
+        if not info.width or not info.height:
+            info.width = 256
+            info.height = 256
+
         # This is a hack for bad clients like Element iOS that require a thumbnail
         info.thumbnail_info = ImageInfo.deserialize(info.serialize())
         if file.decryption_info:

mautrix_telegram/puppet.py

@@ -55,6 +55,7 @@ if TYPE_CHECKING:
 
 
 class Puppet(DBPuppet, BasePuppet):
+    bridge: TelegramBridge
     config: Config
     hs_domain: str
     mxid_template: SimpleTemplate[TelegramID]
@@ -78,6 +79,7 @@ class Puppet(DBPuppet, BasePuppet):
         avatar_url: ContentURI | None = None,
         name_set: bool = False,
         avatar_set: bool = False,
+        contact_info_set: bool = False,
         is_bot: bool = False,
         is_channel: bool = False,
         is_premium: bool = False,
@@ -100,6 +102,7 @@ class Puppet(DBPuppet, BasePuppet):
             avatar_url=avatar_url,
             name_set=name_set,
             avatar_set=avatar_set,
+            contact_info_set=contact_info_set,
             is_bot=is_bot,
             is_channel=is_channel,
             is_premium=is_premium,
@@ -154,6 +157,7 @@ class Puppet(DBPuppet, BasePuppet):
 
     @classmethod
     def init_cls(cls, bridge: "TelegramBridge") -> AsyncIterable[Awaitable[None]]:
+        cls.bridge = bridge
         cls.config = bridge.config
         cls.loop = bridge.loop
         cls.mx = bridge.matrix
@@ -265,11 +269,14 @@ class Puppet(DBPuppet, BasePuppet):
             is_bot != self.is_bot or is_channel != self.is_channel or is_premium != self.is_premium
         )
 
-        self.is_bot = is_bot
+        if is_bot is not None:
+            self.is_bot = is_bot
         self.is_channel = is_channel
-        self.is_premium = is_premium
+        if is_premium is not None:
+            self.is_premium = is_premium
 
-        if self.username != info.username:
+        if self.username != info.username and (info.username or not info.min):
+            self.log.debug(f"Updating username {self.username} -> {info.username}")
             self.username = info.username
             changed = True
 
@@ -279,6 +286,8 @@ class Puppet(DBPuppet, BasePuppet):
 
         if not self.disable_updates:
             try:
+                changed = await self._update_contact_info(force=changed) or changed
+
                 changed = (
                     await self.update_displayname(source, info, client_override=client_override)
                     or changed
@@ -296,8 +305,37 @@ class Puppet(DBPuppet, BasePuppet):
             await self.update_portals_meta()
             await self.save()
 
+    async def _update_contact_info(self, force: bool = False) -> bool:
+        if not self.bridge.homeserver_software.is_hungry:
+            return False
+
+        if self.contact_info_set and not force:
+            return False
+
+        try:
+            identifiers = []
+            if self.username:
+                identifiers.append(f"telegram:{self.username}")
+            if self.phone:
+                phone = "+" + self.phone.lstrip("+")
+                identifiers.append(f"tel:{phone}")
+            await self.default_mxid_intent.beeper_update_profile(
+                {
+                    "com.beeper.bridge.identifiers": identifiers,
+                    "com.beeper.bridge.remote_id": str(self.tgid),
+                    "com.beeper.bridge.service": "telegram",
+                    "com.beeper.bridge.network": "telegram",
+                    "com.beeper.bridge.is_network_bot": self.is_bot,
+                }
+            )
+            self.contact_info_set = True
+        except Exception:
+            self.log.exception("Error updating contact info")
+            self.contact_info_set = False
+        return True
+
     async def update_portals_meta(self) -> None:
-        if not p.Portal.private_chat_portal_meta and not self.mx.e2ee:
+        if p.Portal.private_chat_portal_meta != "always" and not self.mx.e2ee:
             return
         async for portal in p.Portal.find_private_chats_with(self.tgid):
             await portal.update_info_from_puppet(self)
@@ -334,13 +372,15 @@ class Puppet(DBPuppet, BasePuppet):
 
         if isinstance(info, UpdateUserName):
             info = await (client_override or source.client).get_entity(self.peer)
-        if isinstance(info, Channel) or not info.contact:
-            self.displayname_contact = False
-        elif not self.displayname_contact:
-            if not self.displayname:
-                self.displayname_contact = True
-            else:
-                return False
+        is_contact_name = not isinstance(info, Channel) and info.contact
+        # Reject name change if the contact status is moving in an unwanted direction,
+        # and we already have a name for the ghost.
+        if (
+            is_contact_name != self.displayname_contact
+            and is_contact_name != self.config["bridge.allow_contact_info"]
+            and self.displayname
+        ):
+            return False
 
         displayname, quality = self.get_displayname(info)
         needs_reset = displayname != self.displayname or not self.name_set
@@ -348,12 +388,14 @@ class Puppet(DBPuppet, BasePuppet):
         if needs_reset and is_high_quality:
             allow_because = f"{allow_because} and quality {quality} >= {self.displayname_quality}"
             self.log.debug(
-                f"Updating displayname of {self.id} (src: {source.tgid}, allowed "
-                f"because {allow_because}) from {self.displayname} to {displayname}"
+                f"Updating displayname of {self.id} (src: {source.tgid}, "
+                f"contact: {is_contact_name}, allowed because {allow_because}) "
+                f"from {self.displayname} to {displayname}"
             )
             self.log.trace("Displayname source data: %s", info)
             self.displayname = displayname
             self.displayname_source = source.tgid
+            self.displayname_contact = is_contact_name
             self.displayname_quality = quality
             try:
                 await self.default_mxid_intent.set_displayname(
@@ -378,6 +420,16 @@ class Puppet(DBPuppet, BasePuppet):
     ) -> bool:
         if self.disable_updates:
             return False
+        if (
+            isinstance(photo, UserProfilePhoto)
+            and photo.personal
+            and not self.config["bridge.allow_contact_info"]
+        ):
+            self.log.trace(
+                "Dropping user avatar as it's personal "
+                "and contact info is disabled in bridge config"
+            )
+            return False
 
         if photo is None or isinstance(photo, (UserProfilePhotoEmpty, ChatPhotoEmpty)):
             photo_id = ""

mautrix_telegram/tgclient.py

@@ -22,6 +22,7 @@ from telethon.tl.patched import Message
 from telethon.tl.types import (
     InputMediaUploadedDocument,
     InputMediaUploadedPhoto,
+    InputReplyToMessage,
     TypeDocumentAttribute,
     TypeInputMedia,
     TypeInputPeer,
@@ -67,6 +68,10 @@ class MautrixTelegramClient(TelegramClient):
         entity = await self.get_input_entity(entity)
         reply_to = utils.get_message_id(reply_to)
         request = SendMediaRequest(
-            entity, media, message=caption or "", entities=entities or [], reply_to_msg_id=reply_to
+            entity,
+            media,
+            message=caption or "",
+            entities=entities or [],
+            reply_to=InputReplyToMessage(reply_to_msg_id=reply_to) if reply_to else None,
         )
         return self._get_response_message(request, await self(request), entity)

mautrix_telegram/user.py

@@ -22,6 +22,8 @@ import time
 
 from telethon.errors import (
     AuthKeyDuplicatedError,
+    AuthKeyError,
+    AuthKeyNotFound,
     RPCError,
     TakeoutInitDelayError,
     UnauthorizedError,
@@ -38,6 +40,9 @@ from telethon.tl.types import (
     ChatForbidden,
     InputUserSelf,
     Message,
+    MessageActionContactSignUp,
+    MessageActionHistoryClear,
+    MessageService,
     NotifyPeer,
     PeerUser,
     TypeUpdate,
@@ -51,6 +56,7 @@ from telethon.tl.types import (
     User as TLUser,
 )
 from telethon.tl.types.contacts import ContactsNotModified
+from telethon.tl.types.help import AppConfig
 from telethon.tl.types.messages import AvailableReactions
 
 from mautrix.appservice import DOUBLE_PUPPET_SOURCE_KEY
@@ -58,6 +64,7 @@ from mautrix.bridge import BaseUser, async_getter_lock
 from mautrix.client import Client
 from mautrix.errors import MatrixRequestError, MNotFound
 from mautrix.types import PushActionType, PushRuleKind, PushRuleScope, RoomID, RoomTagInfo, UserID
+from mautrix.util import background_task
 from mautrix.util.bridge_state import BridgeState, BridgeStateEvent
 from mautrix.util.opt_prometheus import Gauge
 
@@ -104,6 +111,7 @@ class User(DBUser, AbstractUser, BaseUser):
     _available_emoji_reactions_fetched: float
     _available_emoji_reactions_lock: asyncio.Lock
     _app_config: dict[str, Any] | None
+    _app_config_hash: int
 
     def __init__(
         self,
@@ -141,6 +149,7 @@ class User(DBUser, AbstractUser, BaseUser):
         self._available_emoji_reactions_fetched = 0
         self._available_emoji_reactions_lock = asyncio.Lock()
         self._app_config = None
+        self._app_config_hash = 0
 
         (
             self.relaybot_whitelisted,
@@ -207,17 +216,30 @@ class User(DBUser, AbstractUser, BaseUser):
         async with self._ensure_started_lock:
             return cast(User, await super().ensure_started(even_if_no_session))
 
+    async def on_signed_out(self, err: UnauthorizedError | AuthKeyError | AuthKeyNotFound) -> None:
+        error_code = "tg-auth-error"
+        if isinstance(err, AuthKeyDuplicatedError):
+            error_code = "tg-auth-key-duplicated"
+            message = None
+        else:
+            message = str(err)
+        self.log.warning(f"User got signed out with {err}, deleting data...")
+        try:
+            await self.log_out(
+                state=BridgeStateEvent.BAD_CREDENTIALS,
+                error=error_code,
+                message=message,
+                delete=False,
+            )
+        except Exception:
+            self.log.exception("Error handling external logout")
+
     async def start(self, delete_unless_authenticated: bool = False) -> User:
         try:
             await super().start()
-        except AuthKeyDuplicatedError:
-            self.log.warning("Got AuthKeyDuplicatedError in start()")
-            await self.push_bridge_state(
-                BridgeStateEvent.BAD_CREDENTIALS, error="tg-auth-key-duplicated"
-            )
-            await self.client.disconnect()
-            await self.client.session.delete()
-            self.client = None
+        except (AuthKeyDuplicatedError, AuthKeyNotFound) as e:
+            self.log.warning(f"Got {type(e).__name__} in start()")
+            await self.on_signed_out(e)
             if not delete_unless_authenticated:
                 # The caller wants the client to be connected, so restart the connection.
                 await super().start()
@@ -237,12 +259,7 @@ class User(DBUser, AbstractUser, BaseUser):
             if delete_unless_authenticated or self.tgid:
                 self.log.error(f"Authorization error in start(): {type(e)}: {e}")
             if self.tgid:
-                await self.push_bridge_state(
-                    BridgeStateEvent.BAD_CREDENTIALS,
-                    error="tg-auth-error",
-                    message=str(e),
-                    ttl=3600,
-                )
+                await self.on_signed_out(e)
         except RPCError as e:
             self.log.error(f"Unknown RPC error in start(): {type(e)}: {e}")
             if self.tgid:
@@ -250,10 +267,10 @@ class User(DBUser, AbstractUser, BaseUser):
         else:
             # Authenticated, run post login
             self.log.debug(f"Ensuring post_login() for {self.name}")
-            asyncio.create_task(self.post_login())
+            background_task.create(self.post_login())
             return self
         # Not authenticated, delete data if necessary
-        if delete_unless_authenticated:
+        if delete_unless_authenticated and self.client is not None:
             self.log.debug(f"Unauthenticated user {self.name} start()ed, deleting session...")
             await self.client.disconnect()
             await self.client.session.delete()
@@ -284,18 +301,18 @@ class User(DBUser, AbstractUser, BaseUser):
                     BridgeStateEvent.BACKFILLING
                     if self._is_backfilling
                     else BridgeStateEvent.CONNECTED,
-                    ttl=3600,
                     info=self._bridge_state_info,
                 )
             else:
                 await self.push_bridge_state(
-                    BridgeStateEvent.TRANSIENT_DISCONNECT, ttl=240, error="tg-not-connected"
+                    BridgeStateEvent.TRANSIENT_DISCONNECT, error="tg-not-connected"
                 )
 
     async def fill_bridge_state(self, state: BridgeState) -> None:
         await super().fill_bridge_state(state)
-        state.remote_id = str(self.tgid)
-        state.remote_name = self.human_tg_id
+        if self.tgid:
+            state.remote_id = str(self.tgid)
+            state.remote_name = self.human_tg_id
 
     async def get_bridge_states(self) -> list[BridgeState]:
         if not self.tgid:
@@ -477,13 +494,16 @@ class User(DBUser, AbstractUser, BaseUser):
         self.log.info(f"Creating portal for {portal.tgid_log} as part of backfill loop")
         try:
             await portal.create_matrix_room(
-                self, client=client, update_if_exists=False, invites=[self.mxid]
+                self,
+                client=client,
+                update_if_exists=False,
+                invites=[self.mxid],
+                from_dialog_sync=True,
             )
         except Exception:
             self.log.exception(f"Error while creating {portal.tgid_log}")
         else:
-            puppet = await pu.Puppet.get_by_custom_mxid(self.mxid)
-            await self._post_sync_dialog(portal, puppet, was_created=True, **post_sync_args)
+            await self.post_sync_dialog(portal, puppet=None, was_created=True, **post_sync_args)
 
     async def update(self, update: TypeUpdate) -> bool:
         if not self.is_bot:
@@ -525,7 +545,7 @@ class User(DBUser, AbstractUser, BaseUser):
             await self.stop()
             return None
 
-    async def update_info(self, info: TLUser = None) -> None:
+    async def update_info(self, info: TLUser | None = None) -> None:
         if not info:
             info = await self.get_me()
             if not info:
@@ -567,27 +587,49 @@ class User(DBUser, AbstractUser, BaseUser):
                 except MatrixRequestError:
                     pass
 
-    async def log_out(self) -> bool:
+    async def log_out(
+        self,
+        delete: bool = True,
+        do_logout: bool = True,
+        state: BridgeStateEvent = BridgeStateEvent.LOGGED_OUT,
+        error: str | None = None,
+        message: str | None = None,
+    ) -> bool:
         puppet = await pu.Puppet.get_by_tgid(self.tgid)
-        if puppet.is_real_user:
+        if puppet is not None and puppet.is_real_user:
             await puppet.switch_mxid(None, None)
         try:
             await self.kick_from_portals()
         except Exception:
             self.log.exception("Failed to kick user from portals on logout")
-        await self.push_bridge_state(BridgeStateEvent.LOGGED_OUT)
         if self.tgid:
             try:
                 del self.by_tgid[self.tgid]
             except KeyError:
                 pass
-            self.tgid = None
-        ok = await self.client.log_out()
-        sess = self.client.session
-        await self.stop()
-        await sess.delete()
-        await self.delete()
-        self.by_mxid.pop(self.mxid, None)
+        ok = False
+        if self.client is not None:
+            sess = self.client.session
+            # Try to send a logout request. If it succeeds, this also disconnects the client and
+            # deletes the session, but we do those again later just to be safe.
+            if do_logout:
+                ok = await self.client.log_out()
+            # Force-disconnect the client and set it to None
+            await self.stop()
+            await sess.delete()
+
+        # Drop LOGGED_OUT states if the user was already logged out previously
+        # and doesn't have a remote ID anymore
+        # TODO send a management room notice for non-manual logouts?
+        if self.tgid or state != BridgeStateEvent.LOGGED_OUT:
+            await self.push_bridge_state(state, error=error, message=message)
+        if delete:
+            await self.delete()
+            self.by_mxid.pop(self.mxid, None)
+            self.log.info("User deleted")
+        else:
+            await self.remove_tgid()
+            self.log.info("User telegram ID cleared")
         self._track_metric(METRIC_LOGGED_IN, False)
         return ok
 
@@ -714,12 +756,12 @@ class User(DBUser, AbstractUser, BaseUser):
         )
         await self._mute_room(puppet, portal, update.notify_settings.mute_until.timestamp())
 
-    async def _sync_dialog(
-        self, portal: po.Portal, dialog: Dialog, should_create: bool, puppet: pu.Puppet | None
-    ) -> None:
-        was_created = False
-        post_sync_args = {
-            "last_message_ts": cast(datetime, dialog.date).timestamp(),
+    @staticmethod
+    def dialog_to_sync_args(dialog: Dialog) -> dict:
+        return {
+            "last_message_ts": (
+                cast(datetime, dialog.date).timestamp() if dialog.date else time.time()
+            ),
             "unread_count": dialog.unread_count,
             "max_read_id": dialog.dialog.read_inbox_max_id,
             "mute_until": (
@@ -730,6 +772,24 @@ class User(DBUser, AbstractUser, BaseUser):
             "pinned": dialog.pinned,
             "archived": dialog.archived,
         }
+
+    async def _sync_dialog(
+        self, portal: po.Portal, dialog: Dialog, should_create: bool, puppet: pu.Puppet | None
+    ) -> None:
+        if (
+            not portal.mxid
+            and isinstance(dialog.message, MessageService)
+            and isinstance(
+                dialog.message.action, (MessageActionContactSignUp, MessageActionHistoryClear)
+            )
+        ):
+            self.log.debug(
+                f"Not syncing {portal.tgid_log} "
+                f"(last message is a {type(dialog.message.action).__name__})"
+            )
+            return
+        was_created = False
+        post_sync_args = self.dialog_to_sync_args(dialog)
         if portal.mxid:
             self.log.debug(f"Backfilling and updating {portal.tgid_log} (dialog sync)")
             try:
@@ -743,7 +803,9 @@ class User(DBUser, AbstractUser, BaseUser):
         elif should_create:
             self.log.debug(f"Creating portal for {portal.tgid_log} immediately (dialog sync)")
             try:
-                await portal.create_matrix_room(self, dialog.entity, invites=[self.mxid])
+                await portal.create_matrix_room(
+                    self, dialog.entity, invites=[self.mxid], from_dialog_sync=True
+                )
                 was_created = True
             except Exception:
                 self.log.exception(f"Error while creating {portal.tgid_log}")
@@ -756,7 +818,7 @@ class User(DBUser, AbstractUser, BaseUser):
                 extra_data=post_sync_args,
             )
         if portal.mxid and puppet and puppet.is_real_user:
-            await self._post_sync_dialog(
+            await self.post_sync_dialog(
                 portal=portal,
                 puppet=puppet,
                 was_created=was_created,
@@ -764,10 +826,10 @@ class User(DBUser, AbstractUser, BaseUser):
             )
         self.log.debug(f"_sync_dialog finished for {portal.tgid_log}")
 
-    async def _post_sync_dialog(
+    async def post_sync_dialog(
         self,
         portal: po.Portal,
-        puppet: pu.Puppet,
+        puppet: pu.Puppet | None,
         was_created: bool,
         max_read_id: int,
         last_message_ts: float,
@@ -776,6 +838,10 @@ class User(DBUser, AbstractUser, BaseUser):
         pinned: bool,
         archived: bool,
     ) -> None:
+        if puppet is None:
+            puppet = await pu.Puppet.get_by_custom_mxid(self.mxid)
+            if not puppet or not puppet.is_real_user:
+                return
         self.log.debug(
             f"Running dialog post-sync for {portal.tgid_log} with args "
             f"{was_created=}, {max_read_id=}, {last_message_ts=}, {unread_count=}, "
@@ -912,11 +978,18 @@ class User(DBUser, AbstractUser, BaseUser):
         self.log.debug("Contact syncing complete")
         return contacts
 
+    @property
+    def _available_reactions_up_to_date(self) -> bool:
+        return (
+            bool(self._available_emoji_reactions)
+            and self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic()
+        )
+
     async def get_available_reactions(self) -> set[str]:
-        if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+        if self._available_reactions_up_to_date:
             return self._available_emoji_reactions
         async with self._available_emoji_reactions_lock:
-            if self._available_emoji_reactions_fetched + 12 * 60 * 60 > time.monotonic():
+            if self._available_reactions_up_to_date:
                 return self._available_emoji_reactions
             self.log.debug("Fetching available emoji reactions")
             available_reactions = await self.client(
@@ -926,13 +999,18 @@ class User(DBUser, AbstractUser, BaseUser):
                 self._available_emoji_reactions = {
                     react.reaction
                     for react in available_reactions.reactions
-                    if self.is_premium or not react.premium
+                    if not react.inactive and (self.is_premium or not react.premium)
                 }
                 self._available_emoji_reactions_hash = available_reactions.hash
                 self._available_emoji_reactions_fetched = time.monotonic()
                 self.log.debug(
                     "Got available emoji reactions: %s", self._available_emoji_reactions
                 )
+            elif self._available_emoji_reactions is None:
+                self.log.warning(
+                    f"Got {available_reactions} in response to available reactions request"
+                    " even though nothing is cached"
+                )
             return self._available_emoji_reactions
 
     def tl_to_json(self) -> Any:
@@ -940,8 +1018,9 @@ class User(DBUser, AbstractUser, BaseUser):
 
     async def get_app_config(self) -> dict[str, Any]:
         if not self._app_config:
-            cfg = await self.client(GetAppConfigRequest())
-            self._app_config = util.parse_tl_json(cfg)
+            cfg: AppConfig = await self.client(GetAppConfigRequest(hash=self._app_config_hash))
+            self._app_config = util.parse_tl_json(cfg.config)
+            self._app_config_hash = cfg.hash
         return self._app_config
 
     async def get_max_reactions(self, is_premium: bool | None = None) -> int:

mautrix_telegram/util/file_transfer.py

@@ -22,7 +22,6 @@ import asyncio
 import logging
 import pickle
 import pkgutil
-import tempfile
 import time
 
 from asyncpg import UniqueViolationError
@@ -46,7 +45,7 @@ from telethon.tl.types import (
 )
 
 from mautrix.appservice import IntentAPI
-from mautrix.util import magic, variation_selector
+from mautrix.util import ffmpeg, magic, variation_selector
 
 from .. import abstract_user as au
 from ..db import TelegramFile as DBTelegramFile
@@ -61,11 +60,6 @@ try:
 except ImportError:
     Image = None
 
-try:
-    from moviepy.editor import VideoFileClip
-except ImportError:
-    VideoFileClip = None
-
 try:
     from mautrix.crypto.attachments import encrypt_attachment
 except ImportError:
@@ -103,29 +97,16 @@ def convert_image(
         return source_mime, file, None, None
 
 
-def _read_video_thumbnail(
-    data: bytes,
-    video_ext: str = "mp4",
-    frame_ext: str = "png",
-    max_size: tuple[int, int] = (1024, 720),
-) -> tuple[bytes, int, int]:
-    with tempfile.NamedTemporaryFile(prefix="mxtg_video_", suffix=f".{video_ext}") as file:
-        # We don't have any way to read the video from memory, so save it to disk.
-        file.write(data)
-
-        # Read temp file and get frame
-        frame = VideoFileClip(file.name).get_frame(0)
-
-    # Convert to png and save to BytesIO
-    image = Image.fromarray(frame).convert("RGBA")
-
-    thumbnail_file = BytesIO()
-    if max_size:
-        image.thumbnail(max_size, Image.ANTIALIAS)
-    image.save(thumbnail_file, frame_ext)
-
-    w, h = image.size
-    return thumbnail_file.getvalue(), w, h
+async def _read_video_thumbnail(data: bytes, mime_type: str) -> tuple[bytes, int, int]:
+    first_frame = await ffmpeg.convert_bytes(
+        data,
+        output_extension=".png",
+        output_args=("-update", "1", "-frames:v", "1"),
+        input_mime=mime_type,
+        logger=log,
+    )
+    width, height = Image.open(BytesIO(first_frame)).size
+    return first_frame, width, height
 
 
 def _location_to_id(location: TypeLocation) -> str:
@@ -151,7 +132,7 @@ async def transfer_thumbnail_to_matrix(
     height: int | None = None,
     async_upload: bool = False,
 ) -> DBTelegramFile | None:
-    if not Image or not VideoFileClip:
+    if not Image or not ffmpeg.ffmpeg_path:
         return None
 
     loc_id = _location_to_id(thumbnail_loc)
@@ -170,10 +151,12 @@ async def transfer_thumbnail_to_matrix(
     video_ext = sane_mimetypes.guess_extension(mime_type)
     if custom_data:
         file = custom_data
-    elif VideoFileClip and video_ext and video:
+    elif video_ext and video:
+        log.debug(f"Generating thumbnail for video {loc_id} with ffmpeg")
         try:
-            file, width, height = _read_video_thumbnail(video, video_ext, frame_ext="png")
-        except OSError:
+            file, width, height = await _read_video_thumbnail(video, mime_type=mime_type)
+        except Exception:
+            log.warning(f"Failed to generate thumbnail for {loc_id}", exc_info=True)
             return None
         mime_type = "image/png"
     else:
@@ -350,10 +333,10 @@ async def _unlocked_transfer_file_to_matrix(
             client, intent, loc_id, location, filename, encrypt, parallel_id
         )
         mime_type = location.mime_type
-        file = None
+        unencrypted_file = None
     else:
         try:
-            file = await client.download_file(location)
+            unencrypted_file = file = await client.download_file(location)
         except (LocationInvalidError, FileIdInvalidError):
             return None
         except (AuthBytesInvalidError, AuthKeyInvalidError, SecurityError) as e:
@@ -401,34 +384,37 @@ async def _unlocked_transfer_file_to_matrix(
             width=width,
             height=height,
         )
-    if thumbnail and (mime_type.startswith("video/") or mime_type == "image/gif"):
-        if isinstance(thumbnail, (PhotoSize, PhotoCachedSize)):
-            thumbnail = thumbnail.location
-        try:
+    try:
+        if thumbnail and (mime_type.startswith("video/") or mime_type == "image/gif"):
+            if isinstance(thumbnail, (PhotoSize, PhotoCachedSize)):
+                thumbnail = thumbnail.location
+            try:
+                db_file.thumbnail = await transfer_thumbnail_to_matrix(
+                    client,
+                    intent,
+                    thumbnail,
+                    video=unencrypted_file,
+                    mime_type=mime_type,
+                    encrypt=encrypt,
+                    async_upload=async_upload,
+                )
+            except FileIdInvalidError:
+                log.warning(f"Failed to transfer thumbnail {thumbnail!s}", exc_info=True)
+        elif converted_anim and converted_anim.thumbnail_data:
             db_file.thumbnail = await transfer_thumbnail_to_matrix(
                 client,
                 intent,
-                thumbnail,
-                video=file,
-                mime_type=mime_type,
+                location,
+                video=None,
                 encrypt=encrypt,
+                custom_data=converted_anim.thumbnail_data,
+                mime_type=converted_anim.thumbnail_mime,
+                width=converted_anim.width,
+                height=converted_anim.height,
                 async_upload=async_upload,
             )
-        except FileIdInvalidError:
-            log.warning(f"Failed to transfer thumbnail for {thumbnail!s}", exc_info=True)
-    elif converted_anim and converted_anim.thumbnail_data:
-        db_file.thumbnail = await transfer_thumbnail_to_matrix(
-            client,
-            intent,
-            location,
-            video=None,
-            encrypt=encrypt,
-            custom_data=converted_anim.thumbnail_data,
-            mime_type=converted_anim.thumbnail_mime,
-            width=converted_anim.width,
-            height=converted_anim.height,
-            async_upload=async_upload,
-        )
+    except Exception:
+        log.exception(f"Failed to transfer thumbnail for {loc_id}")
 
     try:
         await db_file.insert()

mautrix_telegram/web/common/auth_api.py

@@ -35,9 +35,11 @@ from telethon.errors import (
     PhoneNumberInvalidError,
     PhoneNumberUnoccupiedError,
     SessionPasswordNeededError,
+    SessionRevokedError,
 )
 
 from mautrix.bridge import InvalidAccessToken, OnlyLoginSelf
+from mautrix.util import background_task
 from mautrix.util.format_duration import format_duration
 
 from ...commands.telegram.auth import enter_password
@@ -199,7 +201,7 @@ class AuthAPI(abc.ABC):
         existing_user = await User.get_by_tgid(user_info.id)
         if existing_user and existing_user != user:
             await existing_user.log_out()
-        asyncio.create_task(user.post_login(user_info, first_login=True))
+        background_task.create(user.post_login(user_info, first_login=True))
         if user.command_status and user.command_status["action"] == "Login":
             user.command_status = None
 
@@ -287,6 +289,17 @@ class AuthAPI(abc.ABC):
                 errcode="phone_number_unoccupied",
                 error="That phone number has not been registered.",
             )
+        except FloodWaitError as e:
+            return self.get_login_response(
+                mxid=user.mxid,
+                state="code",
+                status=429,
+                errcode="flood_wait",
+                error=(
+                    "You tried to enter your phone code too many times. "
+                    f"Please wait for {format_duration(e.seconds)} before trying again."
+                ),
+            )
         except SessionPasswordNeededError:
             if not password_in_data:
                 if user.command_status and user.command_status["action"] == "Login":
@@ -341,12 +354,42 @@ class AuthAPI(abc.ABC):
                 errcode="password_invalid",
                 error="Incorrect password.",
             )
-        except Exception:
+        except SessionRevokedError:
+            return self.get_login_response(
+                mxid=user.mxid,
+                state="request",
+                status=401,
+                errcode="session_revoked",
+                error=(
+                    "Please try again. Login cancelled because your other sessions were "
+                    "terminated via the Telegram app."
+                ),
+            )
+        except FloodWaitError as e:
+            return self.get_login_response(
+                mxid=user.mxid,
+                state="password",
+                status=429,
+                errcode="flood_wait",
+                error=(
+                    "You tried to enter your password too many times. "
+                    f"Please wait for {format_duration(e.seconds)} before trying again."
+                ),
+            )
+        except Exception as e:
             self.log.exception("Error sending password")
+            if isinstance(e, ValueError) and "You must provide a phone and a code" in str(e):
+                return self.get_login_response(
+                    mxid=user.mxid,
+                    state="request",
+                    status=400,
+                    errcode="phone_code_not_entered",
+                    error="Please request a new phone code and enter it first.",
+                )
             return self.get_login_response(
                 mxid=user.mxid,
                 state="password",
                 status=500,
                 errcode="unknown_error",
-                error="Internal server error while sending password.",
+                error=f"Internal server error while sending password. {e}",
             )

mautrix_telegram/web/provisioning/__init__.py

@@ -17,10 +17,14 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Awaitable, Callable
 import asyncio
+import datetime
 import json
 import logging
 
 from aiohttp import web
+from telethon.errors import SessionPasswordNeededError
+from telethon.tl.custom import QRLogin
+from telethon.tl.functions.messages import GetAllStickersRequest
 from telethon.tl.types import ChannelForbidden, ChatForbidden, TypeChat, User as TLUser
 from telethon.utils import get_peer_id, resolve_id
 
@@ -28,6 +32,7 @@ from mautrix.appservice import AppService
 from mautrix.client import Client
 from mautrix.errors import IntentError, MatrixRequestError
 from mautrix.types import UserID
+from mautrix.util import background_task
 
 from ...commands.portal.util import get_initial_state, user_has_power_level
 from ...portal import Portal
@@ -72,9 +77,12 @@ class ProvisioningAPI(AuthAPI):
         )
         self.app.router.add_route("POST", f"{user_prefix}/pm/{{identifier}}", self.start_dm)
 
+        self.app.router.add_route("GET", f"{user_prefix}/stickersets", self.get_stickersets)
+
         self.app.router.add_route("POST", f"{user_prefix}/retry_takeout", self.retry_takeout)
 
         self.app.router.add_route("POST", f"{user_prefix}/logout", self.logout)
+        self.app.router.add_route("GET", f"{user_prefix}/login/qr", self.login_qr)
         self.app.router.add_route("POST", f"{user_prefix}/login/bot_token", self.send_bot_token)
         self.app.router.add_route("POST", f"{user_prefix}/login/request_code", self.request_code)
         self.app.router.add_route("POST", f"{user_prefix}/login/send_code", self.send_code)
@@ -220,7 +228,7 @@ class ProvisioningAPI(AuthAPI):
             portal.photo_id = ""
             await portal.save()
 
-        asyncio.create_task(portal.update_matrix_room(user, entity, levels=levels))
+        background_task.create(portal.update_matrix_room(user, entity, levels=levels))
 
         return web.Response(status=202, body="{}")
 
@@ -341,7 +349,7 @@ class ProvisioningAPI(AuthAPI):
                 self.log.exception("Failed to disconnect chat")
                 return self.get_error_response(500, "exception", "Failed to disconnect chat")
         else:
-            asyncio.create_task(coro)
+            background_task.create(coro)
         return web.json_response({}, status=200 if sync else 202)
 
     async def get_user_info(self, request: web.Request) -> web.Response:
@@ -497,6 +505,18 @@ class ProvisioningAPI(AuthAPI):
             status=201 if just_created else 200,
         )
 
+    async def get_stickersets(self, request: web.Request) -> web.Response:
+        _, user, err = await self.get_user_request_info(
+            request, expect_logged_in=True, want_data=False
+        )
+        if err is not None:
+            return err
+        result = await user.client(GetAllStickersRequest(0))
+        resp = []
+        for stickerset in result.sets:
+            resp.append(stickerset.short_name)
+        return web.json_response(resp, status=200)
+
     async def retry_takeout(self, request: web.Request) -> web.Response:
         data, user, err = await self.get_user_request_info(
             request, expect_logged_in=True, want_data=False
@@ -513,6 +533,50 @@ class ProvisioningAPI(AuthAPI):
         user.takeout_retry_immediate.set()
         return web.json_response({}, status=200)
 
+    async def login_qr(self, request: web.Request) -> web.Response:
+        _, user, err = await self.get_user_request_info(request, websocket=True)
+        if err is not None:
+            return err
+
+        await user.ensure_started(even_if_no_session=True)
+        qr_login = QRLogin(user.client, ignored_ids=[])
+
+        ws = web.WebSocketResponse(protocols=["net.maunium.telegram.login"])
+        await ws.prepare(request)
+
+        retries = 0
+        user_info = None
+        while retries < 4:
+            try:
+                await qr_login.recreate()
+                await ws.send_json(
+                    {
+                        "code": qr_login.url,
+                        "timeout": int(
+                            (
+                                qr_login.expires - datetime.datetime.now(tz=datetime.timezone.utc)
+                            ).total_seconds()
+                        ),
+                    }
+                )
+                user_info = await qr_login.wait()
+                break
+            except asyncio.TimeoutError:
+                retries += 1
+            except SessionPasswordNeededError:
+                await ws.send_json({"success": False, "error": "password-needed"})
+                await ws.close()
+                return ws
+        else:
+            await ws.send_json({"success": False, "error": "timeout"})
+            await ws.close()
+            return ws
+
+        await self.postprocess_login(user, user_info)
+        await ws.send_json({"success": True})
+        await ws.close()
+        return ws
+
     async def send_bot_token(self, request: web.Request) -> web.Response:
         data, user, err = await self.get_user_request_info(request)
         if err is not None:
@@ -638,6 +702,15 @@ class ProvisioningAPI(AuthAPI):
             )
         return None
 
+    def check_websocket_authorization(self, request: web.Request) -> web.Response | None:
+        auth_parts = request.headers.get("Sec-WebSocket-Protocol").split(",")
+        for part in auth_parts:
+            if part.strip() == f"net.maunium.telegram.auth-{self.secret}":
+                return None
+        return self.get_error_response(
+            error="Shared secret is not valid.", errcode="shared_secret_invalid", status=401
+        )
+
     @staticmethod
     async def get_data(request: web.Request) -> dict | None:
         try:
@@ -692,8 +765,12 @@ class ProvisioningAPI(AuthAPI):
         expect_logged_in: bool | None = False,
         require_puppeting: bool = False,
         want_data: bool = True,
+        websocket: bool = False,
     ) -> tuple[dict | None, User | None, web.Response | None]:
-        err = self.check_authorization(request)
+        if not websocket:
+            err = self.check_authorization(request)
+        else:
+            err = self.check_websocket_authorization(request)
         if err is not None:
             return None, None, err
 

optional-requirements.txt

@@ -2,22 +2,19 @@
 # Uncommented lines after the group definition insert things into that group.
 
 #/speedups
-cryptg>=0.1,<0.4
+cryptg>=0.1,<0.5
 aiodns
 brotli
 
 #/qr_login
-pillow>=4,<10
+pillow>=10.0.1,<11
 qrcode>=6,<8
 
-#/hq_thumbnails
-moviepy>=1,<2
-
 #/formattednumbers
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.16
+prometheus_client>=0.6,<0.20
 
 #/e2be
 python-olm>=3,<4
@@ -25,4 +22,7 @@ pycryptodome>=3,<4
 unpaddedbase64>=1,<3
 
 #/sqlite
-aiosqlite>=0.16,<0.18
+aiosqlite>=0.16,<0.20
+
+#/proxy
+python-socks[asyncio]

requirements.txt

@@ -1,11 +1,10 @@
-ruamel.yaml>=0.15.35,<0.18
+ruamel.yaml>=0.15.35,<0.19
 python-magic>=0.4,<0.5
 commonmark>=0.8,<0.10
 aiohttp>=3,<4
 yarl>=1,<2
-mautrix>=0.18.8,<0.19
-#telethon>=1.25.4,<1.27
-tulir-telethon==1.27.0a1
-asyncpg>=0.20,<0.28
+mautrix>=0.20.3,<0.21
+tulir-telethon==1.34.0a2
+asyncpg>=0.20,<0.30
 mako>=1,<2
 setuptools

setup.py

@@ -51,7 +51,7 @@ setuptools.setup(
 
     install_requires=install_requires,
     extras_require=extras_require,
-    python_requires="~=3.8",
+    python_requires="~=3.9",
 
     classifiers=[
         "Development Status :: 4 - Beta",
@@ -60,9 +60,10 @@ setuptools.setup(
         "Framework :: AsyncIO",
         "Programming Language :: Python",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11",
+        "Programming Language :: Python :: 3.12",
     ],
     package_data={"mautrix_telegram": [
         "web/public/*.mako", "web/public/*.png", "web/public/*.css",