Bump version to 0.13.0

Fixes #809

.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

.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,31 @@
+# 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,4 +1,4 @@
-FROM dock.mau.dev/tulir/lottieconverter:alpine-3.16
+FROM dock.mau.dev/tulir/lottieconverter:alpine-3.17
 
 RUN apk add --no-cache \
       python3 py3-pip py3-setuptools py3-wheel \
@@ -13,13 +13,6 @@ 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 \

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.13.0"
 __author__ = "Tulir Asokan <tulir@maunium.net>"

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,
@@ -63,8 +63,8 @@ from telethon.tl.types import (
     UpdateShort,
     UpdateShortChatMessage,
     UpdateShortMessage,
+    UpdateUser,
     UpdateUserName,
-    UpdateUserPhoto,
     UpdateUserStatus,
     UpdateUserTyping,
     User,
@@ -75,6 +75,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
 
@@ -235,18 +236,23 @@ class AbstractUser(ABC):
         )
         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:
@@ -320,7 +326,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,
             (
@@ -351,7 +357,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)
@@ -500,18 +506,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)}")
 
@@ -615,24 +626,28 @@ 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))
+                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])
 
@@ -681,7 +696,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()

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
@@ -184,7 +185,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 +252,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/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

@@ -134,15 +134,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 "+()- "})

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")
@@ -150,6 +151,7 @@ class Config(BaseBridgeConfig):
         copy("bridge.private_chat_portal_meta")
         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")

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/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/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

@@ -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
@@ -314,6 +317,8 @@ bridge:
     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.

mautrix_telegram/portal.py

@@ -171,7 +171,7 @@ from mautrix.types import (
     UserID,
     VideoInfo,
 )
-from mautrix.util import magic, variation_selector
+from mautrix.util import background_task, magic, variation_selector
 from mautrix.util.message_send_checkpoint import MessageSendCheckpointStatus
 from mautrix.util.simple_lock import SimpleLock
 from mautrix.util.simple_template import SimpleTemplate
@@ -727,7 +727,7 @@ class Portal(DBPortal, BasePortal):
                         self.log.exception(f"Failed to get entity through {user.tgid} for update")
                         return self.mxid
                 update = self.update_matrix_room(user, entity)
-                asyncio.create_task(update)
+                background_task.create(update)
                 await self.invite_to_matrix(invites or [])
             return self.mxid
         async with self._room_create_lock:
@@ -805,6 +805,12 @@ class Portal(DBPortal, BasePortal):
         participants_count = 2
         if isinstance(entity, Chat):
             participants_count = entity.participants_count
+            if entity.deactivated or entity.migrated_to:
+                self.log.error(
+                    "Throwing error for attempted portal creation "
+                    f"({entity.deactivated=}, {entity.migrated_to=})"
+                )
+                raise RuntimeError("Tried to create portal for deactivated chat")
         elif isinstance(entity, Channel) and not entity.broadcast:
             participants_count = entity.participants_count
         if participants_count is None and self.config["bridge.max_member_count"] > 0:
@@ -1525,11 +1531,11 @@ class Portal(DBPortal, BasePortal):
         )
         if self.peer_type == "channel":
             if not self.megagroup:
-                asyncio.create_task(
+                background_task.create(
                     self._try_handle_read_for_sponsored_msg(user, event_id, timestamp)
                 )
             else:
-                asyncio.create_task(self._poll_telegram_reactions(user))
+                background_task.create(self._poll_telegram_reactions(user))
 
     async def _preproc_kick_ban(
         self, user: u.User | p.Puppet, source: u.User
@@ -1964,7 +1970,7 @@ class Portal(DBPortal, BasePortal):
             message_type=msgtype,
         )
         await self._send_delivery_receipt(event_id)
-        asyncio.create_task(self._send_message_status(event_id, err=None))
+        background_task.create(self._send_message_status(event_id, err=None))
         if response.ttl_period:
             await self._mark_disappearing(
                 event_id=event_id,
@@ -1993,7 +1999,6 @@ class Portal(DBPortal, BasePortal):
             status.status = MessageStatus.RETRIABLE
         else:
             status.status = MessageStatus.SUCCESS
-        status.fill_legacy_booleans()
 
         await intent.send_message_event(
             room_id=self.mxid,
@@ -2258,7 +2263,7 @@ class Portal(DBPortal, BasePortal):
                 EventType.ROOM_REDACTION,
             )
             await self._send_delivery_receipt(redaction_event_id)
-            asyncio.create_task(self._send_message_status(redaction_event_id, err=None))
+            background_task.create(self._send_message_status(redaction_event_id, err=None))
 
     async def _handle_matrix_reaction_deletion(
         self, deleter: u.User, event_id: EventID, tg_space: TelegramID
@@ -2371,7 +2376,7 @@ class Portal(DBPortal, BasePortal):
                 EventType.REACTION,
             )
             await self._send_delivery_receipt(reaction_event_id)
-            asyncio.create_task(self._send_message_status(reaction_event_id, err=None))
+            background_task.create(self._send_message_status(reaction_event_id, err=None))
 
     async def _handle_matrix_reaction(
         self,
@@ -2574,7 +2579,7 @@ class Portal(DBPortal, BasePortal):
             # Ignore typing notifications from double puppeted users to avoid echoing
             return
         is_typing = isinstance(update.action, SendMessageTypingAction)
-        await user.default_mxid_intent.set_typing(self.mxid, is_typing=is_typing)
+        await user.default_mxid_intent.set_typing(self.mxid, timeout=5000 if is_typing else 0)
 
     async def handle_telegram_edit(
         self, source: au.AbstractUser, sender: p.Puppet | None, evt: Message
@@ -2587,7 +2592,7 @@ class Portal(DBPortal, BasePortal):
             return
 
         if self.peer_type != "channel" and isinstance(evt, Message) and evt.reactions is not None:
-            asyncio.create_task(
+            background_task.create(
                 self.try_handle_telegram_reactions(source, TelegramID(evt.id), evt.reactions)
             )
         sender_id = sender.tgid if sender else self.tgid
@@ -2648,7 +2653,7 @@ class Portal(DBPortal, BasePortal):
             source, intent, is_bot, evt, no_reply_fallback=True
         )
         converted.content.set_edit(editing_msg.mxid)
-        await intent.set_typing(self.mxid, is_typing=False)
+        await intent.set_typing(self.mxid, timeout=0)
         timestamp = evt.edit_date if evt.edit_date != evt.date else None
         event_id = await self._send_message(
             intent, converted.content, timestamp=timestamp, event_type=converted.type
@@ -2805,8 +2810,11 @@ class Portal(DBPortal, BasePortal):
             elif not insertion_id:
                 insertion_id = self.base_insertion_id
         await self.save()
-        # TODO this should probably check actual event count instead of message count
-        if event_count > 0 and self.backfill_msc2716:
+        if (
+            event_count > 0
+            and self.backfill_msc2716
+            and (not forward or not self.bridge.homeserver_software.is_hungry)
+        ):
             await self.main_intent.send_state_event(
                 self.mxid,
                 StateMarker,
@@ -3264,9 +3272,29 @@ class Portal(DBPortal, BasePortal):
 
     async def handle_telegram_message(
         self, source: au.AbstractUser, sender: p.Puppet | None, evt: Message
+    ) -> None:
+        try:
+            await self._handle_telegram_message(source, sender, evt)
+        except Exception:
+            sender_id = sender.tgid if sender else None
+            self.log.exception(
+                f"Failed to handle Telegram message {evt.id} from {sender_id} via {source.tgid}"
+            )
+            if self.config["bridge.incoming_bridge_error_reports"]:
+                intent = sender.intent_for(self) if sender else self.main_intent
+                await self._send_message(
+                    intent,
+                    TextMessageEventContent(
+                        msgtype=MessageType.NOTICE,
+                        body="Error processing message from Telegram",
+                    ),
+                )
+
+    async def _handle_telegram_message(
+        self, source: au.AbstractUser, sender: p.Puppet | None, evt: Message
     ) -> None:
         if not self.mxid:
-            self.log.trace("Got telegram message %d, but no room exists, creating...", evt.id)
+            self.log.debug("Got telegram message %d, but no room exists, creating...", evt.id)
             await self.create_matrix_room(source, invites=[source.mxid], update_if_exists=False)
             if not self.mxid:
                 self.log.warning("Room doesn't exist even after creating, dropping %d", evt.id)
@@ -3333,12 +3361,17 @@ class Portal(DBPortal, BasePortal):
                 f"Telegram user {sender.tgid} sent a message, but doesn't have a displayname,"
                 " updating info..."
             )
-            entity = await source.client.get_entity(sender.peer)
-            await sender.update_info(source, entity)
-            if not sender.displayname:
-                self.log.debug(
-                    f"Telegram user {sender.tgid} doesn't have a displayname even after"
-                    f" updating with data {entity!s}"
+            try:
+                entity = await source.client.get_entity(sender.peer)
+                await sender.update_info(source, entity)
+                if not sender.displayname:
+                    self.log.debug(
+                        f"Telegram user {sender.tgid} doesn't have a displayname even after"
+                        f" updating with data {entity!s}"
+                    )
+            except ValueError as e:
+                self.log.warning(
+                    f"Couldn't find entity to update profile of {sender.tgid}", exc_info=True
                 )
 
         if sender:
@@ -3350,7 +3383,7 @@ class Portal(DBPortal, BasePortal):
         converted = await self._msg_conv.convert(source, intent, is_bot, evt)
         if not converted:
             return
-        await intent.set_typing(self.mxid, is_typing=False)
+        await intent.set_typing(self.mxid, timeout=0)
         event_id = await self._send_message(
             intent, converted.content, timestamp=evt.date, event_type=converted.type
         )
@@ -3397,7 +3430,7 @@ class Portal(DBPortal, BasePortal):
             await intent.redact(self.mxid, event_id)
             return
         if isinstance(evt, Message) and evt.reactions:
-            asyncio.create_task(
+            background_task.create(
                 self.try_handle_telegram_reactions(
                     source, dbm.tgid, evt.reactions, dbm=dbm, timestamp=evt.date
                 )
@@ -3418,7 +3451,7 @@ class Portal(DBPortal, BasePortal):
         dm = DisappearingMessage(self.mxid, event_id, seconds, expiration_ts=expires_at * 1000)
         await dm.insert()
         if expires_at:
-            asyncio.create_task(self._disappear_event(dm))
+            background_task.create(self._disappear_event(dm))
 
     async def _create_room_on_action(
         self, source: au.AbstractUser, action: TypeMessageAction
@@ -3432,6 +3465,10 @@ class Portal(DBPortal, BasePortal):
             MessageActionChatJoinedByRequest,
         )
         if isinstance(action, create_and_exit) or isinstance(action, create_and_continue):
+            self.log.debug(
+                f"Got telegram action of type {type(action).__name__},"
+                " but no room exists, creating..."
+            )
             await self.create_matrix_room(
                 source, invites=[source.mxid], update_if_exists=isinstance(action, create_and_exit)
             )

mautrix_telegram/portal_util/message_convert.py

@@ -409,6 +409,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(
@@ -498,8 +500,10 @@ class TelegramMessageConverter:
             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(

mautrix_telegram/puppet.py

@@ -334,13 +334,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 +350,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 +382,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/user.py

@@ -22,6 +22,7 @@ import time
 
 from telethon.errors import (
     AuthKeyDuplicatedError,
+    AuthKeyError,
     RPCError,
     TakeoutInitDelayError,
     UnauthorizedError,
@@ -58,6 +59,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
 
@@ -207,17 +209,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) -> 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:
+        except AuthKeyDuplicatedError as e:
             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
+            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 +252,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 +260,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 +294,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:
@@ -567,27 +577,46 @@ 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()
+
+        # TODO send a management room notice for non-manual logouts?
+        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
 

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

@@ -38,6 +38,7 @@ from telethon.errors import (
 )
 
 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 +200,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
 
@@ -341,8 +342,16 @@ class AuthAPI(abc.ABC):
                 errcode="password_invalid",
                 error="Incorrect password.",
             )
-        except Exception:
+        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",

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,7 +2,7 @@
 # Uncommented lines after the group definition insert things into that group.
 
 #/speedups
-cryptg>=0.1,<0.4
+cryptg>=0.1,<0.5
 aiodns
 brotli
 
@@ -10,14 +10,12 @@ brotli
 pillow>=4,<10
 qrcode>=6,<8
 
-#/hq_thumbnails
-moviepy>=1,<2
 
 #/formattednumbers
 phonenumbers>=8,<9
 
 #/metrics
-prometheus_client>=0.6,<0.16
+prometheus_client>=0.6,<0.17
 
 #/e2be
 python-olm>=3,<4
@@ -25,4 +23,4 @@ pycryptodome>=3,<4
 unpaddedbase64>=1,<3
 
 #/sqlite
-aiosqlite>=0.16,<0.18
+aiosqlite>=0.16,<0.19

requirements.txt

@@ -3,9 +3,9 @@ python-magic>=0.4,<0.5
 commonmark>=0.8,<0.10
 aiohttp>=3,<4
 yarl>=1,<2
-mautrix>=0.18.8,<0.19
+mautrix>=0.19.4,<0.20
 #telethon>=1.25.4,<1.27
-tulir-telethon==1.27.0a1
+tulir-telethon==1.28.0a3
 asyncpg>=0.20,<0.28
 mako>=1,<2
 setuptools