Merge pull request #71 from tulir/bots

Add option for message relay bot

README.md

@@ -1,5 +1,5 @@
 # mautrix-telegram
-A Matrix-Telegram puppeting bridge.
+A Matrix-Telegram hybrid puppeting/relaybot bridge.
 
 ### [Wiki](https://github.com/tulir/mautrix-telegram/wiki)
 

ROADMAP.md

@@ -25,7 +25,7 @@
       * [x] Matrix users who have logged into Telegram
     * [x] Kicking
     * [ ] Joining
-      * [ ] Chat name as alias
+      * [x] Chat name as alias
       * [ ] ‡ Chat invite link as alias
     * [x] Leaving
   * [x] Room metadata changes (name, topic, avatar)
@@ -74,7 +74,7 @@
     * [x] At startup
     * [x] When receiving invite or message
   * [x] Private chat creation by inviting Matrix puppet of Telegram user to new room
-  * [ ] Option to use bot to relay messages for unauthenticated Matrix users
+  * [x] Option to use bot to relay messages for unauthenticated Matrix users
   * [ ] Option to use own Matrix account for messages sent from other Telegram clients
 * [Commands](https://github.com/tulir/mautrix-telegram/wiki/Management-commands)
   * [x] Logging in and out (`login` + code entering)

example-config.yaml

@@ -67,6 +67,8 @@ bridge:
     # Show message editing as a reply to the original message.
     # If this is false, message edits are not shown at all, as Matrix does not support editing yet.
     edits_as_replies: False
+    # Whether or not Matrix bot messages (type m.notice) should be bridged.
+    bridge_notices: False
 
     # The prefix for commands. Only required in non-management rooms.
     command_prefix: "!tg"
@@ -87,3 +89,5 @@ telegram:
     # Get your own API keys at https://my.telegram.org/apps
     api_id: 12345
     api_hash: tjyd5yge35lbodk1xwzw2jstp90k55qz
+    # (Optional) Create your own bot at https://t.me/BotFather
+    #bot_token: 123456789:ABCD-QBPd3VrWRhg623xYh07WUWErYA9eMI

mautrix_appservice/errors.py

@@ -31,7 +31,7 @@ class MatrixRequestError(MatrixError):
     """ The home server returned an error response. """
 
     def __init__(self, code=0, text="", errcode=None, message=None):
-        super().__init__("%d: %s" % (code, text))
+        super().__init__(f"{code}: {text}")
         self.code = code
         self.text = text
         self.errcode = errcode

mautrix_appservice/intent_api.py

@@ -194,12 +194,16 @@ class IntentAPI:
         content = {"displayname": name}
         return await self.client.request("PUT", f"/profile/{self.mxid}/displayname", content)
 
-    async def set_presence(self, status="online"):
+    async def set_presence(self, status="online", ignore_cache=False):
         await self.ensure_registered()
+        if not ignore_cache and self.state_store.has_presence(self.mxid, status):
+            return
         content = {
             "presence": status
         }
-        return await self.client.request("PUT", f"/presence/{self.mxid}/status", content)
+        resp = await self.client.request("PUT", f"/presence/{self.mxid}/status", content)
+        self.state_store.set_presence(self.mxid, status)
+        return resp
 
     async def set_avatar(self, url):
         await self.ensure_registered()
@@ -223,11 +227,14 @@ class IntentAPI:
     # region Room actions
 
     async def create_room(self, alias=None, is_public=False, name=None, topic=None,
-                          is_direct=False, invitees=None, initial_state=None):
+                          is_direct=False, invitees=None, initial_state=None,
+                          guests_can_join=False):
         await self.ensure_registered()
         content = {
-            "visibility": "public" if is_public else "private",
+            "visibility": "private",
             "is_direct": is_direct,
+            "preset": "public_chat" if is_public else "private_chat",
+            "guests_can_join": guests_can_join,
         }
         if alias:
             content["room_alias_name"] = alias
@@ -326,18 +333,29 @@ class IntentAPI:
             events.remove(event_id)
             await self.set_pinned_messages(room_id, events)
 
+    async def set_join_rule(self, room_id, join_rule):
+        if join_rule not in ("public", "knock", "invite", "private"):
+            raise ValueError(f"Invalid join rule \"{join_rule}\"")
+        await self.send_state_event(room_id, "m.room.join_rules", {
+            "join_rule": join_rule,
+        })
+
     async def get_event(self, room_id, event_id):
         await self.ensure_joined(room_id)
         return await self.client.request("GET", f"/rooms/{room_id}/event/{event_id}")
 
-    async def set_typing(self, room_id, is_typing=True, timeout=5000):
+    async def set_typing(self, room_id, is_typing=True, timeout=5000, ignore_cache=False):
         await self.ensure_joined(room_id)
+        if not ignore_cache and is_typing == self.state_store.is_typing(room_id, self.mxid):
+            return
         content = {
             "typing": is_typing
         }
         if is_typing:
             content["timeout"] = timeout
-        return await self.client.request("PUT", f"/rooms/{room_id}/typing/{self.mxid}", content)
+        resp = await self.client.request("PUT", f"/rooms/{room_id}/typing/{self.mxid}", content)
+        self.state_store.set_typing(room_id, self.mxid, is_typing, timeout)
+        return resp
 
     async def mark_read(self, room_id, event_id):
         await self.ensure_joined(room_id)
@@ -551,7 +569,7 @@ class IntentAPI:
             self.log.warning(
                 f"Power level of {self.mxid} is not enough for {event_type} in {room_id}")
             # raise IntentError(f"Power level of {self.mxid} is not enough"
-            #                  + f"for {event_type} in {room_id}")
+            #                   f"for {event_type} in {room_id}")
             return
         # TODO implement
 

mautrix_appservice/state_store.py

@@ -15,14 +15,21 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import json
+import time
 
 
 class StateStore:
     def __init__(self, autosave_file=None):
+        self.autosave_file = autosave_file
+
+        # Persistent storage
         self.registrations = set()
         self.memberships = {}
         self.power_levels = {}
-        self.autosave_file = autosave_file
+
+        # Non-persistent storage
+        self.presence = {}
+        self.typing = {}
 
     def save(self, file):
         if isinstance(file, str):
@@ -63,6 +70,29 @@ class StateStore:
         if self.autosave_file:
             self.save(self.autosave_file)
 
+    def set_presence(self, user, presence):
+        self.presence[user] = presence
+
+    def has_presence(self, user, presence):
+        try:
+            return self.presence[user] == presence
+        except KeyError:
+            return False
+
+    def set_typing(self, room_id, user, is_typing, timeout=0):
+        if is_typing:
+            ts = int(round(time.time() * 1000))
+            self.typing[(room_id, user)] = ts + timeout
+        else:
+            del self.typing[(room_id, user)]
+
+    def is_typing(self, room_id, user):
+        ts = int(round(time.time() * 1000))
+        try:
+            return self.typing[(room_id, user)] > ts
+        except KeyError:
+            return False
+
     def is_registered(self, user):
         return user in self.registrations
 

mautrix_telegram/__main__.py

@@ -29,9 +29,12 @@ from .config import Config
 from .matrix import MatrixHandler
 
 from .db import init as init_db
+from .abstract_user import init as init_abstract_user
 from .user import init as init_user, User
+from .bot import init as init_bot
 from .portal import init as init_portal
 from .puppet import init as init_puppet
+from .context import Context
 
 log = logging.getLogger("mau")
 time_formatter = logging.Formatter("[%(asctime)s] [%(levelname)s@%(name)s] %(message)s")
@@ -76,14 +79,22 @@ loop = asyncio.get_event_loop()
 appserv = AppService(config["homeserver.address"], config["homeserver.domain"],
                      config["appservice.as_token"], config["appservice.hs_token"],
                      config["appservice.bot_username"], log="mau.as", loop=loop)
-context = (appserv, db_session, config, loop)
+
+
+context = Context(appserv, db_session, config, loop, None, None)
 
 with appserv.run(config["appservice.hostname"], config["appservice.port"]) as start:
-    MatrixHandler(context)
     init_db(db_session)
+    init_abstract_user(context)
+    context.bot = init_bot(context)
+    context.mx = MatrixHandler(context)
     init_portal(context)
     init_puppet(context)
-    startup_actions = init_user(context) + [start]
+    startup_actions = init_user(context) + [start, context.mx.init_as_bot()]
+
+    if context.bot:
+        startup_actions.append(context.bot.start())
+
     try:
         loop.run_until_complete(asyncio.gather(*startup_actions, loop=loop))
         loop.run_forever()

mautrix_telegram/abstract_user.py

@@ -0,0 +1,235 @@
+# -*- coding: future_fstrings -*-
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2018 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import platform
+import os
+
+from telethon.tl.types import *
+
+from .tgclient import MautrixTelegramClient
+from .db import Message as DBMessage
+from . import portal as po, puppet as pu, __version__
+
+config = None
+
+
+class AbstractUser:
+    loop = None
+    log = None
+    db = None
+    az = None
+
+    def __init__(self):
+        self.connected = False
+        self.whitelisted = False
+        self.client = None
+        self.tgid = None
+        self.mxid = None
+
+    def _init_client(self):
+        self.log.debug(f"Initializing client for {self.name}")
+        device = f"{platform.system()} {platform.release()}"
+        sysversion = MautrixTelegramClient.__version__
+        self.client = MautrixTelegramClient(self.name,
+                                            config["telegram.api_id"],
+                                            config["telegram.api_hash"],
+                                            loop=self.loop,
+                                            app_version=__version__,
+                                            system_version=sysversion,
+                                            device_model=device)
+        self.client.add_update_handler(self._update_catch)
+
+    async def update(self, update):
+        return False
+
+    async def post_login(self):
+        raise NotImplementedError()
+
+    async def _update_catch(self, update):
+        try:
+            if not await self.update(update):
+                await self._update(update)
+        except Exception:
+            self.log.exception("Failed to handle Telegram update")
+
+    async def _get_dialogs(self, limit=None):
+        dialogs = await self.client.get_dialogs(limit=limit)
+        return [dialog.entity for dialog in dialogs if (
+            not isinstance(dialog.entity, (User, ChatForbidden, ChannelForbidden))
+            and not (isinstance(dialog.entity, Chat)
+                     and (dialog.entity.deactivated or dialog.entity.left)))]
+
+    @property
+    def name(self):
+        raise NotImplementedError()
+
+    @property
+    def logged_in(self):
+        return self.client and self.client.is_user_authorized()
+
+    @property
+    def has_full_access(self):
+        return self.logged_in and self.whitelisted
+
+    async def start(self):
+        self.connected = await self.client.connect()
+
+    async def ensure_started(self, even_if_no_session=False):
+        if not self.whitelisted:
+            return self
+        elif not self.connected and (even_if_no_session or os.path.exists(f"{self.name}.session")):
+            return await self.start()
+        return self
+
+    def stop(self):
+        self.client.disconnect()
+        self.client = None
+        self.connected = False
+
+    # region Telegram update handling
+
+    async def _update(self, update):
+        if isinstance(update,
+                      (UpdateShortChatMessage, UpdateShortMessage, UpdateNewChannelMessage,
+                       UpdateNewMessage, UpdateEditMessage, UpdateEditChannelMessage)):
+            await self.update_message(update)
+        elif isinstance(update, (UpdateChatUserTyping, UpdateUserTyping)):
+            await self.update_typing(update)
+        elif isinstance(update, UpdateUserStatus):
+            await self.update_status(update)
+        elif isinstance(update, (UpdateChatAdmins, UpdateChatParticipantAdmin)):
+            await self.update_admin(update)
+        elif isinstance(update, UpdateChatParticipants):
+            portal = po.Portal.get_by_tgid(update.participants.chat_id)
+            if portal and portal.mxid:
+                await portal.update_telegram_participants(update.participants.participants)
+        elif isinstance(update, UpdateChannelPinnedMessage):
+            portal = po.Portal.get_by_tgid(update.channel_id)
+            if portal and portal.mxid:
+                await portal.update_telegram_pin(self, update.id)
+        elif isinstance(update, (UpdateUserName, UpdateUserPhoto)):
+            await self.update_others_info(update)
+        elif isinstance(update, UpdateReadHistoryOutbox):
+            await self.update_read_receipt(update)
+        else:
+            self.log.debug("Unhandled update: %s", update)
+
+    async def update_read_receipt(self, update):
+        if not isinstance(update.peer, PeerUser):
+            self.log.debug("Unexpected read receipt peer: %s", update.peer)
+            return
+
+        portal = po.Portal.get_by_tgid(update.peer.user_id, self.tgid)
+        if not portal or not portal.mxid:
+            return
+
+        # We check that these are user read receipts, so tg_space is always the user ID.
+        message = DBMessage.query.get((update.max_id, self.tgid))
+        if not message:
+            return
+
+        puppet = pu.Puppet.get(update.peer.user_id)
+        await puppet.intent.mark_read(portal.mxid, message.mxid)
+
+    async def update_admin(self, update):
+        # TODO duplication not checked
+        portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
+        if isinstance(update, UpdateChatAdmins):
+            await portal.set_telegram_admins_enabled(update.enabled)
+        elif isinstance(update, UpdateChatParticipantAdmin):
+            await portal.set_telegram_admin(update.user_id)
+        else:
+            self.log.warning("Unexpected admin status update: %s", update)
+
+    async def update_typing(self, update):
+        if isinstance(update, UpdateUserTyping):
+            portal = po.Portal.get_by_tgid(update.user_id, self.tgid, "user")
+        else:
+            portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
+        sender = pu.Puppet.get(update.user_id)
+        await portal.handle_telegram_typing(sender, update)
+
+    async def update_others_info(self, update):
+        # TODO duplication not checked
+        puppet = pu.Puppet.get(update.user_id)
+        if isinstance(update, UpdateUserName):
+            if await puppet.update_displayname(self, update):
+                puppet.save()
+        elif isinstance(update, UpdateUserPhoto):
+            if await puppet.update_avatar(self, update.photo.photo_big):
+                puppet.save()
+        else:
+            self.log.warning("Unexpected other user info update: %s", update)
+
+    async def update_status(self, update):
+        puppet = pu.Puppet.get(update.user_id)
+        if isinstance(update.status, UserStatusOnline):
+            await puppet.intent.set_presence("online")
+        elif isinstance(update.status, UserStatusOffline):
+            await puppet.intent.set_presence("offline")
+        else:
+            self.log.warning("Unexpected user status update: %s", update)
+        return
+
+    def get_message_details(self, update):
+        if isinstance(update, UpdateShortChatMessage):
+            portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
+            sender = pu.Puppet.get(update.from_id)
+        elif isinstance(update, UpdateShortMessage):
+            portal = po.Portal.get_by_tgid(update.user_id, self.tgid, "user")
+            sender = pu.Puppet.get(self.tgid if update.out else update.user_id)
+        elif isinstance(update, (UpdateNewMessage, UpdateNewChannelMessage,
+                                 UpdateEditMessage, UpdateEditChannelMessage)):
+            update = update.message
+            if isinstance(update.to_id, PeerUser) and not update.out:
+                portal = po.Portal.get_by_tgid(update.from_id, peer_type="user",
+                                               tg_receiver=self.tgid)
+            else:
+                portal = po.Portal.get_by_entity(update.to_id, receiver_id=self.tgid)
+            sender = pu.Puppet.get(update.from_id) if update.from_id else None
+        else:
+            self.log.warning(
+                f"Unexpected message type in User#get_message_details: {type(update)}")
+            return update, None, None
+        return update, sender, portal
+
+    def update_message(self, original_update):
+        update, sender, portal = self.get_message_details(original_update)
+
+        if isinstance(update, MessageService):
+            if isinstance(update.action, MessageActionChannelMigrateFrom):
+                self.log.debug(f"Ignoring action %s to %s by %d", update.action,
+                               portal.tgid_log,
+                               sender.id)
+                return
+            self.log.debug("Handling action %s to %s by %d", update.action, portal.tgid_log,
+                           sender.id)
+            return portal.handle_telegram_action(self, sender, update)
+
+        user = sender.tgid if sender else "admin"
+        if isinstance(original_update, (UpdateEditMessage, UpdateEditChannelMessage)):
+            self.log.debug("Handling edit %s to %s by %s", update, portal.tgid_log, user)
+            return portal.handle_telegram_edit(self, sender, update)
+
+        self.log.debug("Handling message %s to %s by %s", update, portal.tgid_log, user)
+        return portal.handle_telegram_message(self, sender, update)
+
+    # endregion
+
+
+def init(context):
+    global config
+    AbstractUser.az, AbstractUser.db, config, AbstractUser.loop, _ = context

mautrix_telegram/bot.py

@@ -0,0 +1,120 @@
+# -*- coding: future_fstrings -*-
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2018 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import logging
+
+from telethon.tl.types import *
+from telethon.errors import ChannelInvalidError, ChannelPrivateError
+from telethon.tl.functions.messages import GetChatsRequest
+from telethon.tl.functions.channels import GetChannelsRequest
+
+from .abstract_user import AbstractUser
+from .db import BotChat
+from . import puppet as pu
+
+config = None
+
+
+class Bot(AbstractUser):
+    log = logging.getLogger("mau.bot")
+
+    def __init__(self, token):
+        super().__init__()
+        self.token = token
+        self.whitelisted = True
+        self._init_client()
+        self.chats = {chat.id: chat.type for chat in BotChat.query.all()}
+
+    async def start(self):
+        await super().start()
+        if not self.logged_in:
+            await self.client.sign_in(bot_token=self.token)
+        await self.post_login()
+        return self
+
+    async def post_login(self):
+        info = await self.client.get_me()
+        self.tgid = info.id
+        self.mxid = pu.Puppet.get_mxid_from_id(self.tgid)
+
+        chat_ids = [id for id, type in self.chats.items() if type == "chat"]
+        response = await self.client(GetChatsRequest(chat_ids))
+        for chat in response.chats:
+            if isinstance(chat, ChatForbidden) or chat.left or chat.deactivated:
+                self.remove_chat(chat.id)
+
+        channel_ids = [InputChannel(id, 0)
+                       for id, type in self.chats.items()
+                       if type == "channel"]
+        for id in channel_ids:
+            try:
+                await self.client(GetChannelsRequest([id]))
+            except (ChannelPrivateError, ChannelInvalidError):
+                self.remove_chat(id.channel_id)
+
+    def add_chat(self, id, type):
+        if id not in self.chats:
+            self.chats[id] = type
+            self.db.add(BotChat(id=id, type=type))
+            self.db.commit()
+
+    def remove_chat(self, id):
+        try:
+            del self.chats[id]
+        except KeyError:
+            pass
+        self.db.delete(BotChat.query.get(id))
+        self.db.commit()
+
+    async def update(self, update):
+        if not isinstance(update, (UpdateNewMessage, UpdateNewChannelMessage)):
+            return
+        elif not isinstance(update.message, MessageService):
+            return
+
+        to_id = update.message.to_id
+        if isinstance(to_id, PeerChannel):
+            to_id = to_id.channel_id
+            type = "channel"
+        elif isinstance(to_id, PeerChat):
+            to_id = to_id.chat_id
+            type = "chat"
+        else:
+            return
+
+        action = update.message.action
+        if isinstance(action, MessageActionChatAddUser):
+            if self.tgid in action.users:
+                self.add_chat(to_id, type)
+        elif isinstance(action, MessageActionChatDeleteUser):
+            if action.user_id == self.tgid:
+                self.remove_chat(to_id)
+
+    def is_in_chat(self, peer_id):
+        return peer_id in self.chats
+
+    @property
+    def name(self):
+        return "bot"
+
+
+def init(context):
+    global config
+    config = context.config
+    token = config["telegram.bot_token"]
+    if token:
+        return Bot(token)
+    return None

mautrix_telegram/commands/auth.py

@@ -19,6 +19,7 @@ import asyncio
 from telethon.errors import *
 
 from . import command_handler
+from .. import puppet as pu
 
 
 @command_handler(needs_auth=False)
@@ -32,6 +33,18 @@ async def ping(evt):
         return await evt.reply("You're not logged in.")
 
 
+@command_handler()
+async def ping_bot(evt):
+    if not evt.tgbot:
+        return await evt.reply("Telegram message relay bot not configured.")
+    bot_info = await evt.tgbot.client.get_me()
+    mxid = pu.Puppet.get_mxid_from_id(bot_info.id)
+    displayname = bot_info.first_name
+    return await evt.reply("Telegram message relay bot is active: "
+                           f"[{displayname}](https://matrix.to/#/{mxid}) (ID {bot_info.id})\n\n"
+                           "To use the bot, simply invite it to a portal room.")
+
+
 @command_handler(needs_auth=False, management_only=True)
 def register(evt):
     return evt.reply("Not yet implemented.")
@@ -44,6 +57,7 @@ async def login(evt):
     elif len(evt.args) == 0:
         return await evt.reply("**Usage:** `$cmdprefix+sp login <phone number>`")
     phone_number = evt.args[0]
+    await evt.sender.ensure_started(even_if_no_session=True)
     await evt.sender.client.sign_in(phone_number)
     evt.sender.command_status = {
         "next": enter_code,
@@ -58,6 +72,7 @@ async def enter_code(evt):
         return await evt.reply("**Usage:** `$cmdprefix+sp enter-code <code>`")
 
     try:
+        await evt.sender.ensure_started(even_if_no_session=True)
         user = await evt.sender.client.sign_in(code=evt.args[0])
         asyncio.ensure_future(evt.sender.post_login(user), loop=evt.loop)
         evt.sender.command_status = None
@@ -98,6 +113,7 @@ async def enter_password(evt):
         return await evt.reply("**Usage:** `$cmdprefix+sp enter-password <password>`")
 
     try:
+        await evt.sender.ensure_started(even_if_no_session=True)
         user = await evt.sender.client.sign_in(password=evt.args[0])
         asyncio.ensure_future(evt.sender.post_login(user), loop=evt.loop)
         evt.sender.command_status = None

mautrix_telegram/commands/clean_rooms.py

@@ -52,7 +52,7 @@ async def _find_rooms(intent):
     return management_rooms, unidentified_rooms, portals, empty_portals
 
 
-@command_handler(needs_admin=True, name="clean-rooms")
+@command_handler(needs_admin=True, needs_auth=False, name="clean-rooms")
 async def clean_rooms(evt):
     if not evt.is_management:
         return await evt.reply("`clean-rooms` is a particularly spammy command. Please don't "
@@ -66,7 +66,7 @@ async def clean_rooms(evt):
               or ["No management rooms found."])
     reply.append("#### Active portal rooms (A)")
     reply += ([f"{n+1}. [P{n+1}](https://matrix.to/#/{portal.mxid}) "
-               + f"(to Telegram chat \"{portal.title}\")"
+               f"(to Telegram chat \"{portal.title}\")"
                for n, portal in enumerate(portals)]
               or ["No active portal rooms found."])
     reply.append("#### Unidentified rooms (U)")
@@ -75,7 +75,7 @@ async def clean_rooms(evt):
               or ["No unidentified rooms found."])
     reply.append("#### Inactive portal rooms (I)")
     reply += ([f"{n}. [E{n}](https://matrix.to/#/{portal.mxid}) "
-               + f"(to Telegram chat \"{portal.title}\")"
+               f"(to Telegram chat \"{portal.title}\")"
                for n, portal in enumerate(empty_portals)]
               or ["No inactive portal rooms found."])
 
@@ -141,21 +141,21 @@ async def set_rooms_to_clean(evt, management_rooms, unidentified_rooms, portals,
                 "**Usage:** `$cmdprefix+sp clean-groups <_M|A|U|I_><range>")
     else:
         return await evt.reply(f"Unknown room cleaning action `{command}`. "
-                               + "Use `$cmdprefix+sp cancel` to cancel room "
-                               + "cleaning.")
+                               "Use `$cmdprefix+sp cancel` to cancel room "
+                               "cleaning.")
 
     evt.sender.command_status = {
         "next": lambda confirm: execute_room_cleanup(confirm, rooms_to_clean),
         "action": "Room cleaning",
     }
     await evt.reply(f"To confirm cleaning up {len(rooms_to_clean)} rooms, type"
-                    + "`$cmdprefix+sp confirm-clean`.")
+                    "`$cmdprefix+sp confirm-clean`.")
 
 
 async def execute_room_cleanup(evt, rooms_to_clean):
     if len(evt.args) > 0 and evt.args[0] == "confirm-clean":
         await evt.reply(f"Cleaning {len(rooms_to_clean)} rooms. "
-                        + "This might take a while.")
+                        "This might take a while.")
         cleaned = 0
         for room in rooms_to_clean:
             if isinstance(room, po.Portal):

mautrix_telegram/commands/handler.py

@@ -27,7 +27,7 @@ def command_handler(needs_auth=True, management_only=False, needs_admin=False, n
         def wrapper(evt):
             if management_only and not evt.is_management:
                 return evt.reply(f"`{evt.command}` is a restricted command:"
-                                 + "you may only run it in management rooms.")
+                                 "you may only run it in management rooms.")
             elif needs_auth and not evt.sender.logged_in:
                 return evt.reply("This command requires you to be logged in.")
             elif needs_admin and not evt.sender.is_admin:
@@ -45,6 +45,8 @@ class CommandEvent:
         self.az = handler.az
         self.log = handler.log
         self.loop = handler.loop
+        self.tgbot = handler.tgbot
+        self.config = handler.config
         self.command_prefix = handler.command_prefix
         self.room_id = room
         self.sender = sender
@@ -87,7 +89,7 @@ class CommandHandler:
     log = logging.getLogger("mau.commands")
 
     def __init__(self, context):
-        self.az, self.db, self.config, self.loop = context
+        self.az, self.db, self.config, self.loop, self.tgbot = context
         self.command_prefix = self.config["bridge.command_prefix"]
 
     # region Utility functions for handling commands
@@ -110,6 +112,6 @@ class CommandHandler:
         except FloodWaitError as e:
             return evt.reply(f"Flood error: Please wait {format_duration(e.seconds)}")
         except Exception:
-            self.log.exception(f"Fatal error handling command "
-                               + f"{evt.command} {' '.join(args)} from {sender.mxid}")
+            self.log.exception("Fatal error handling command "
+                               f"{evt.command} {' '.join(args)} from {sender.mxid}")
             return evt.reply("Fatal error while handling command. Check logs for more details.")

mautrix_telegram/commands/meta.py

@@ -65,6 +65,7 @@ def help(evt):
                                        `channel` (defaults to `group`).
 
 #### Portal management  
+**ping-bot**                - Get info of the message relay Telegram bot.
 **upgrade**                 - Upgrade a normal Telegram group to a supergroup.  
 **invite-link**             - Get a Telegram invite link to the current chat.  
 **delete-portal**           - Forget the current portal room. Only works for group chats; to delete

mautrix_telegram/commands/telegram.py

@@ -51,7 +51,7 @@ async def search(evt):
     else:
         reply += ["**Results in contacts:**", ""]
     reply += [(f"* [{puppet.displayname}](https://matrix.to/#/{puppet.mxid}): "
-               + f"{puppet.id} ({similarity}% match)")
+               f"{puppet.id} ({similarity}% match)")
               for puppet, similarity in results]
 
     # TODO somehow show remote channel results when joining by alias is possible?
@@ -71,8 +71,8 @@ async def pm(evt):
         return await evt.reply("That doesn't seem to be a user.")
     portal = po.Portal.get_by_entity(user, evt.sender.tgid)
     await portal.create_matrix_room(evt.sender, user, [evt.sender.mxid])
-    return await evt.reply(
-        f"Created private chat room with {pu.Puppet.get_displayname(user, False)}")
+    return await evt.reply("Created private chat room with "
+                           f"{pu.Puppet.get_displayname(user, False)}")
 
 
 @command_handler()
@@ -116,9 +116,9 @@ async def delete_portal(evt):
         "action": "Portal deletion",
     }
     return await evt.reply("Please confirm deletion of portal "
-                           + f"[{room_id}](https://matrix.to/#/{room_id}) "
-                           + f"to Telegram chat \"{portal.title}\" "
-                           + "by typing `$cmdprefix+sp confirm-delete`")
+                           f"[{room_id}](https://matrix.to/#/{room_id}) "
+                           f"to Telegram chat \"{portal.title}\" "
+                           "by typing `$cmdprefix+sp confirm-delete`")
 
 
 @command_handler()
@@ -184,16 +184,16 @@ async def create(evt):
         return await evt.reply("Please set a title before creating a Telegram chat.")
     elif (not levels or not levels["users"] or evt.az.intent.mxid not in levels["users"] or
           levels["users"][evt.az.intent.mxid] < 100):
-        return await evt.reply(f"Please give "
-                               + f"[the bridge bot](https://matrix.to/#/{evt.az.intent.mxid})"
-                               + f" a power level of 100 before creating a Telegram chat.")
+        return await evt.reply("Please give "
+                               f"[the bridge bot](https://matrix.to/#/{evt.az.intent.mxid})"
+                               " a power level of 100 before creating a Telegram chat.")
     else:
         for user, level in levels["users"].items():
             if level >= 100 and user != evt.az.intent.mxid:
                 return await evt.reply(
                     f"Please make sure only the bridge bot has power level above"
-                    + f"99 before creating a Telegram chat.\n\n"
-                    + f"Use power level 95 instead of 100 for admins.")
+                    f"99 before creating a Telegram chat.\n\n"
+                    f"Use power level 95 instead of 100 for admins.")
 
     supergroup = type == "supergroup"
     type = {

mautrix_telegram/config.py

@@ -94,7 +94,7 @@ class Config(DictWithRecursion):
         self.set("appservice.hs_token", self._new_token())
 
         url = (f"{self['appservice.protocol']}://"
-               + f"{self['appservice.hostname']}:{self['appservice.port']}")
+               f"{self['appservice.hostname']}:{self['appservice.port']}")
         self._registration = {
             "id": self.get("appservice.id", "telegram"),
             "as_token": self["appservice.as_token"],

mautrix_telegram/context.py

@@ -0,0 +1,34 @@
+# -*- coding: future_fstrings -*-
+# mautrix-telegram - A Matrix-Telegram puppeting bridge
+# Copyright (C) 2018 Tulir Asokan
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+class Context:
+    def __init__(self, az, db, config, loop, bot, mx):
+        self.az = az
+        self.db = db
+        self.config = config
+        self.loop = loop
+        self.bot = bot
+        self.mx = mx
+
+    def __iter__(self):
+        yield self.az
+        yield self.db
+        yield self.config
+        yield self.loop
+        yield self.bot
+        # yield self.mx

mautrix_telegram/db.py

@@ -48,7 +48,7 @@ class Message(Base):
     tgid = Column(Integer, primary_key=True)
     tg_space = Column(Integer, primary_key=True)
 
-    __table_args__ = (UniqueConstraint('mxid', 'mx_room', 'tg_space', name='_mx_id_room'),)
+    __table_args__ = (UniqueConstraint("mxid", "mx_room", "tg_space", name="_mx_id_room"),)
 
 
 class UserPortal(Base):
@@ -95,9 +95,18 @@ class Puppet(Base):
     photo_id = Column(String, nullable=True)
 
 
+# Fucking Telegram not telling bots what chats they are in 3:<
+class BotChat(Base):
+    query = None
+    __tablename__ = "bot_chat"
+    id = Column(Integer, primary_key=True)
+    type = Column(String, nullable=False)
+
+
 def init(db_session):
     Portal.query = db_session.query_property()
     Message.query = db_session.query_property()
     UserPortal.query = db_session.query_property()
     User.query = db_session.query_property()
     Puppet.query = db_session.query_property()
+    BotChat.query = db_session.query_property()

mautrix_telegram/formatter.py

@@ -189,8 +189,8 @@ def matrix_reply_to_telegram(content, tg_space, room_id=None):
         reply = content["m.relates_to"]["m.in_reply_to"]
         room_id = room_id or reply["room_id"]
         event_id = reply["event_id"]
-        message = DBMessage.query.filter(DBMessage.mxid == event_id and
-                                         DBMessage.tg_space == tg_space and
+        message = DBMessage.query.filter(DBMessage.mxid == event_id,
+                                         DBMessage.tg_space == tg_space,
                                          DBMessage.mx_room == room_id).one_or_none()
         if message:
             return message.tgid
@@ -244,7 +244,7 @@ async def telegram_event_to_matrix(evt, source, native_replies=False, message_li
         if not fwd_from:
             fwd_from = "Unknown user"
         html = (f"Forwarded message from <b>{fwd_from}</b><br/>"
-                + f"<blockquote>{html}</blockquote>")
+                f"<blockquote>{html}</blockquote>")
 
     if evt.reply_to_msg_id:
         space = (evt.to_id.channel_id
@@ -271,7 +271,7 @@ async def telegram_event_to_matrix(evt, source, native_replies=False, message_li
                     displayname = puppet.displayname if puppet else sender
                     reply_to_user = f"<a href='https://matrix.to/#/{sender}'>{displayname}</a>"
                     reply_to_msg = (("<a href='https://matrix.to/#/"
-                                     + f"{msg.mx_room}/{msg.mxid}'>{reply_text}</a>")
+                                     f"{msg.mx_room}/{msg.mxid}'>{reply_text}</a>")
                                     if message_link_in_reply else "Reply")
                     quote = f"{reply_to_msg} to {reply_to_user}<blockquote>{body}</blockquote>"
                 except (ValueError, KeyError, MatrixRequestError):
@@ -331,8 +331,8 @@ def _telegram_to_matrix(text, entities):
         elif entity_type == MessageEntityPre:
             if entity.language:
                 html.append("<pre>"
-                            + f"<code class='language-{entity.language}'>{entity_text}</code>"
-                            + "</pre>")
+                            f"<code class='language-{entity.language}'>{entity_text}</code>"
+                            "</pre>")
             else:
                 html.append(f"<pre><code>{entity_text}</code></pre>")
         elif entity_type == MessageEntityMention:

mautrix_telegram/matrix.py

@@ -28,13 +28,13 @@ class MatrixHandler:
     log = logging.getLogger("mau.mx")
 
     def __init__(self, context):
-        self.az, self.db, self.config, _ = context
+        self.az, self.db, self.config, _, self.tgbot = context
         self.commands = CommandHandler(context)
 
         self.az.matrix_event_handler(self.handle_event)
 
     async def init_as_bot(self):
-        self.az.intent.set_display_name(
+        await self.az.intent.set_display_name(
             self.config.get("appservice.bot_displayname", "Telegram bridge bot"))
 
     async def handle_puppet_invite(self, room, puppet, inviter):
@@ -60,8 +60,8 @@ class MatrixHandler:
             if len(members) > 1:
                 await puppet.intent.error_and_leave(room, text=None, html=(
                     f"Please invite "
-                    + f"<a href='https://matrix.to/#/{self.az.intent.mxid}'>the bridge bot</a> "
-                    + f"first if you want to create a Telegram chat."))
+                    f"<a href='https://matrix.to/#/{self.az.intent.mxid}'>the bridge bot</a> "
+                    f"first if you want to create a Telegram chat."))
                 return
 
             await puppet.intent.join_room(room)
@@ -71,9 +71,9 @@ class MatrixHandler:
                     await puppet.intent.invite(portal.mxid, inviter.mxid)
                     await puppet.intent.send_notice(room, text=None, html=(
                         "You already have a private chat with me: "
-                        + f"<a href='https://matrix.to/#/{portal.mxid}'>"
-                        + "Link to room"
-                        + "</a>"))
+                        f"<a href='https://matrix.to/#/{portal.mxid}'>"
+                        "Link to room"
+                        "</a>"))
                     await puppet.intent.leave_room(room)
                     return
                 except MatrixRequestError:
@@ -88,7 +88,7 @@ class MatrixHandler:
                                                   "Telegram chat is created for this room.")
 
     async def handle_invite(self, room, user, inviter):
-        inviter = User.get_by_mxid(inviter)
+        inviter = await User.get_by_mxid(inviter).ensure_started()
         if not inviter.whitelisted:
             return
         elif user == self.az.bot_mxid:
@@ -101,6 +101,9 @@ class MatrixHandler:
             return
 
         user = User.get_by_mxid(user, create=False)
+        if not user:
+            return
+        await user.ensure_started()
         portal = Portal.get_by_mxid(room)
         if user and user.has_full_access and portal:
             await portal.invite_telegram(inviter, user)
@@ -110,7 +113,7 @@ class MatrixHandler:
         self.log.debug(f"{inviter} invited {user} to {room}")
 
     async def handle_join(self, room, user):
-        user = User.get_by_mxid(user)
+        user = await User.get_by_mxid(user).ensure_started()
 
         portal = Portal.get_by_mxid(room)
         if not portal:
@@ -120,19 +123,23 @@ class MatrixHandler:
             await portal.main_intent.kick(room, user.mxid,
                                           "You are not whitelisted on this Telegram bridge.")
             return
-        elif not user.logged_in:
-            # TODO[waiting-for-bots] once we have bot support, this won't be needed.
+        elif not user.logged_in and not portal.has_bot:
             await portal.main_intent.kick(room, user.mxid,
-                                          "You are not logged into this Telegram bridge.")
+                                          "This chat does not have a bot relaying "
+                                          "messages for unauthenticated users.")
             return
 
         self.log.debug(f"{user} joined {room}")
-        # TODO join Telegram chat if applicable
+        if user.logged_in:
+            await portal.join_matrix(user)
 
     async def handle_part(self, room, user, sender):
         self.log.debug(f"{user} left {room}")
 
         sender = User.get_by_mxid(sender, create=False)
+        if not sender:
+            return
+        await sender.ensure_started()
 
         portal = Portal.get_by_mxid(room)
         if not portal:
@@ -143,7 +150,10 @@ class MatrixHandler:
             await portal.leave_matrix(puppet, sender)
 
         user = User.get_by_mxid(user, create=False)
-        if user and user.logged_in:
+        if not user:
+            return
+        await user.ensure_started()
+        if user.logged_in:
             await portal.leave_matrix(user, sender)
 
     def is_command(self, message):
@@ -158,10 +168,12 @@ class MatrixHandler:
         self.log.debug(f"{sender} sent {message} to ${room}")
 
         is_command, text = self.is_command(message)
-        sender = User.get_by_mxid(sender)
+        sender = await User.get_by_mxid(sender).ensure_started()
+        if not sender.whitelisted:
+            return
 
         portal = Portal.get_by_mxid(room)
-        if sender.has_full_access and portal and not is_command:
+        if not is_command and portal and (sender.logged_in or portal.has_bot):
             await portal.handle_matrix_message(sender, message, event_id)
             return
 
@@ -187,19 +199,19 @@ class MatrixHandler:
 
     async def handle_redaction(self, room, sender, event_id):
         portal = Portal.get_by_mxid(room)
-        sender = User.get_by_mxid(sender)
+        sender = await User.get_by_mxid(sender).ensure_started()
         if sender.has_full_access and portal:
             await portal.handle_matrix_deletion(sender, event_id)
 
     async def handle_power_levels(self, room, sender, new, old):
         portal = Portal.get_by_mxid(room)
-        sender = User.get_by_mxid(sender)
+        sender = await User.get_by_mxid(sender).ensure_started()
         if sender.has_full_access and portal:
             await portal.handle_matrix_power_levels(sender, new["users"], old["users"])
 
     async def handle_room_meta(self, type, room, sender, content):
         portal = Portal.get_by_mxid(room)
-        sender = User.get_by_mxid(sender)
+        sender = await User.get_by_mxid(sender).ensure_started()
         if sender.has_full_access and portal:
             handler, content_key = {
                 "m.room.name": (portal.handle_matrix_title, "name"),
@@ -209,7 +221,7 @@ class MatrixHandler:
             if content_key not in content:
                 # FIXME handle
                 pass
-                await handler(sender, content[content_key])
+            await handler(sender, content[content_key])
 
     def filter_matrix_event(self, event):
         return (event["sender"] == self.az.bot_mxid
@@ -236,5 +248,5 @@ class MatrixHandler:
         elif type == "m.room.power_levels":
             await self.handle_power_levels(evt["room_id"], evt["sender"], evt["content"],
                                            evt["prev_content"])
-        elif type == "m.room.name" or type == "m.room.avatar" or type == "m.room.topic":
+        elif type in ("m.room.name", "m.room.avatar", "m.room.topic"):
             await self.handle_room_meta(type, evt["room_id"], evt["sender"], evt["content"])

mautrix_telegram/portal.py

@@ -44,6 +44,8 @@ class Portal:
     log = logging.getLogger("mau.portal")
     db = None
     az = None
+    bot = None
+    bridge_notices = False
     by_mxid = {}
     by_tgid = {}
 
@@ -62,6 +64,7 @@ class Portal:
 
         self._dedup = deque()
         self._dedup_mxid = {}
+        self._dedup_action = deque()
 
         if save_to_cache:
             if tgid:
@@ -88,36 +91,51 @@ class Portal:
         elif self.peer_type == "channel":
             return PeerChannel(channel_id=self.tgid)
 
-    def _hash_event(self, event):
-        if self.peer_type == "channel":
-            # Message IDs are unique per-channel
-            return event.id
+    @property
+    def has_bot(self):
+        return self.bot and self.bot.is_in_chat(self.tgid)
 
+    @staticmethod
+    def _hash_event(event):
         # Non-channel messages are unique per-user (wtf telegram), so we have no other choice than
         # to deduplicate based on a hash of the message content.
 
-        # The timestamp is only accurate to the second, so we can't rely on solely that either.
-        hash_content = [event.date.timestamp(), event.message]
-        if event.fwd_from:
-            hash_content += [event.fwd_from.from_id, event.fwd_from.channel_id]
-        elif isinstance(event, Message) and event.media:
-            try:
-                hash_content += {
-                    MessageMediaContact: lambda media: [media.user_id],
-                    MessageMediaDocument: lambda media: [media.document.id, media.caption],
-                    MessageMediaPhoto: lambda media: [media.photo.id, media.caption],
-                    MessageMediaGeo: lambda media: [media.geo.long, media.geo.lat],
-                }[type(event.media)](event.media)
-            except KeyError:
-                pass
+        # The timestamp is only accurate to the second, so we can't rely solely on that either.
+        if isinstance(event, MessageService):
+            hash_content = [event.date.timestamp(), event.from_id, event.action]
+        else:
+            hash_content = [event.date.timestamp(), event.message]
+            if event.fwd_from:
+                hash_content += [event.fwd_from.from_id, event.fwd_from.channel_id]
+            elif isinstance(event, Message) and event.media:
+                try:
+                    hash_content += {
+                        MessageMediaContact: lambda media: [media.user_id],
+                        MessageMediaDocument: lambda media: [media.document.id, media.caption],
+                        MessageMediaPhoto: lambda media: [media.photo.id, media.caption],
+                        MessageMediaGeo: lambda media: [media.geo.long, media.geo.lat],
+                    }[type(event.media)](event.media)
+                except KeyError:
+                    pass
 
         return hashlib.md5("-"
                            .join(str(a) for a in hash_content)
                            .encode("utf-8")
                            ).hexdigest()
 
+    def is_duplicate_action(self, event):
+        hash = self._hash_event(event) if self.peer_type != "channel" else event.id
+        if hash in self._dedup_action:
+            return True
+
+        self._dedup_action.append(hash)
+
+        if len(self._dedup_action) > 20:
+            self._dedup_action.popleft()
+        return False
+
     def is_duplicate(self, event, mxid=None):
-        hash = self._hash_event(event)
+        hash = self._hash_event(event) if self.peer_type != "channel" else event.id
         if hash in self._dedup:
             return self._dedup_mxid[hash]
 
@@ -196,8 +214,7 @@ class Portal:
         self._main_intent = puppet.intent if direct else self.az.intent
 
         if self.peer_type == "channel" and entity.username:
-            # TODO make public once safe
-            public = False
+            public = True
             alias = self._get_room_alias(entity.username)
             self.username = entity.username
         else:
@@ -206,7 +223,7 @@ class Portal:
             alias = None
 
         if alias:
-            # TODO properly handle existing room aliases
+            # TODO? properly handle existing room aliases
             await self.main_intent.remove_room_alias(alias)
 
         power_levels = self._get_base_power_levels({}, entity)
@@ -260,11 +277,32 @@ class Portal:
             groupname=username)
 
     async def sync_telegram_users(self, source, users):
+        allowed_tgids = set()
         for entity in users:
             puppet = p.Puppet.get(entity.id)
+            if self.bot and puppet.tgid == self.bot.tgid:
+                self.bot.add_chat(self.tgid, self.peer_type)
+            allowed_tgids.add(entity.id)
             await puppet.intent.ensure_joined(self.mxid)
             await puppet.update_info(source, entity)
 
+        joined_mxids = await self.main_intent.get_room_members(self.mxid)
+        for user in joined_mxids:
+            if user == self.az.intent.mxid:
+                continue
+            puppet_id = p.Puppet.get_id_from_mxid(user)
+            if puppet_id and puppet_id not in allowed_tgids:
+                if self.bot and puppet_id == self.bot.tgid:
+                    self.bot.remove_chat(self.tgid)
+                await self.main_intent.kick(self.mxid, user,
+                                            "User had left this Telegram chat.")
+                continue
+            mx_user = u.User.get_by_mxid(user, create=False)
+            if mx_user and not self.has_bot and mx_user.tgid not in allowed_tgids:
+                await self.main_intent.kick(self.mxid, mx_user.mxid,
+                                            "You had left this Telegram chat.")
+                continue
+
     async def add_telegram_user(self, user_id, source=None):
         puppet = p.Puppet.get(user_id)
         if source:
@@ -319,6 +357,9 @@ class Portal:
             self.username = username or None
             if self.username:
                 await self.main_intent.add_room_alias(self.mxid, self._get_room_alias())
+                await self.main_intent.set_join_rule(self.mxid, "public")
+            else:
+                await self.main_intent.set_join_rule(self.mxid, "invite")
             return True
         return False
 
@@ -404,7 +445,7 @@ class Portal:
         for member in members:
             if p.Puppet.get_id_from_mxid(member) or member == self.main_intent.mxid:
                 continue
-            user = u.User.get_by_mxid(member)
+            user = await u.User.get_by_mxid(member).ensure_started()
             if user.has_full_access:
                 authenticated.append(user)
         return authenticated
@@ -463,22 +504,42 @@ class Portal:
             channel = await self.get_input_entity(user)
             await user.client(LeaveChannelRequest(channel=channel))
 
+    async def join_matrix(self, user):
+        if self.peer_type == "channel":
+            await user.client(JoinChannelRequest(channel=await self.get_input_entity(user)))
+        else:
+            # We'll just assume the user is already in the chat.
+            pass
+
     async def handle_matrix_message(self, sender, message, event_id):
         type = message["msgtype"]
-        space = self.tgid if self.peer_type == "channel" else sender.tgid
+        if sender.logged_in:
+            client = sender.client
+            space = self.tgid if self.peer_type == "channel" else sender.tgid
+        else:
+            client = self.bot.client
+            space = self.tgid if self.peer_type == "channel" else self.bot.tgid
         reply_to = formatter.matrix_reply_to_telegram(message, space, room_id=self.mxid)
-        if type in {"m.text", "m.emote"}:
+
+        if type == "m.emote":
+            if "formatted_body" in message:
+                message["formatted_body"] = f"* {sender.displayname} {message['formatted_body']}"
+            message["body"] = f"* {sender.displayname} {message['body']}"
+            type = "m.text"
+        elif not sender.logged_in:
+            if "formatted_body" in message:
+                message["formatted_body"] = (f"<{sender.displayname}> "
+                                             f"{message['formatted_body']}")
+            message["body"] = f"<{sender.displayname}> {message['body']}"
+
+        if type == "m.text" or (self.bridge_notices and type == "m.notice"):
             if "format" in message and message["format"] == "org.matrix.custom.html":
                 message, entities = formatter.matrix_to_telegram(message["formatted_body"])
-                if type == "m.emote":
-                    message = "/me " + message
-                response = await sender.client.send_message(self.peer, message, entities=entities,
-                                                            reply_to=reply_to)
+                response = await client.send_message(self.peer, message, entities=entities,
+                                                     reply_to=reply_to)
             else:
-                if type == "m.emote":
-                    message["body"] = "/me " + message["body"]
-                response = await sender.client.send_message(self.peer, message["body"],
-                                                            reply_to=reply_to)
+                response = await client.send_message(self.peer, message["body"],
+                                                     reply_to=reply_to)
         elif type in {"m.image", "m.file", "m.audio", "m.video"}:
             file = await self.main_intent.download_file(message["url"])
 
@@ -491,8 +552,8 @@ class Portal:
             if "w" in info and "h" in info:
                 attributes.append(DocumentAttributeImageSize(w=info["w"], h=info["h"]))
 
-            response = await sender.client.send_file(self.peer, file, mime, caption, attributes,
-                                                     file_name, reply_to=reply_to)
+            response = await client.send_file(self.peer, file, mime, caption, attributes,
+                                              file_name, reply_to=reply_to)
         else:
             self.log.debug("Unhandled Matrix event: %s", message)
             return
@@ -506,8 +567,8 @@ class Portal:
 
     async def handle_matrix_deletion(self, deleter, event_id):
         space = self.tgid if self.peer_type == "channel" else deleter.tgid
-        message = DBMessage.query.filter(DBMessage.mxid == event_id and
-                                         DBMessage.tg_space == space and
+        message = DBMessage.query.filter(DBMessage.mxid == event_id,
+                                         DBMessage.tg_space == space,
                                          DBMessage.mx_room == self.mxid).one_or_none()
         if not message:
             return
@@ -553,10 +614,11 @@ class Portal:
             return
 
         if self.peer_type == "chat":
-            await sender.client(EditChatTitleRequest(chat_id=self.tgid, title=title))
+            response = await sender.client(EditChatTitleRequest(chat_id=self.tgid, title=title))
         else:
             channel = await self.get_input_entity(sender)
-            await sender.client(EditTitleRequest(channel=channel, title=title))
+            response = await sender.client(EditTitleRequest(channel=channel, title=title))
+        self._register_outgoing_actions_for_dedup(response)
         self.title = title
         self.save()
 
@@ -572,11 +634,12 @@ class Portal:
         photo = InputChatUploadedPhoto(file=uploaded)
 
         if self.peer_type == "chat":
-            updates = await sender.client(EditChatPhotoRequest(chat_id=self.tgid, photo=photo))
+            response = await sender.client(EditChatPhotoRequest(chat_id=self.tgid, photo=photo))
         else:
             channel = await self.get_input_entity(sender)
-            updates = await sender.client(EditPhotoRequest(channel=channel, photo=photo))
-        for update in updates.updates:
+            response = await sender.client(EditPhotoRequest(channel=channel, photo=photo))
+        self._register_outgoing_actions_for_dedup(response)
+        for update in response.updates:
             is_photo_update = (isinstance(update, UpdateNewMessage)
                                and isinstance(update.message, MessageService)
                                and isinstance(update.message.action, MessageActionChatEditPhoto))
@@ -586,6 +649,13 @@ class Portal:
                 self.save()
                 break
 
+    def _register_outgoing_actions_for_dedup(self, response):
+        for update in response.updates:
+            check_dedup = (isinstance(update, (UpdateNewMessage, UpdateNewChannelMessage))
+                           and isinstance(update.message, MessageService))
+            if check_dedup:
+                self.is_duplicate_action(update.message)
+
     # endregion
     # region Telegram chat info updating
 
@@ -635,7 +705,6 @@ class Portal:
 
         invites = await self._get_telegram_users_in_matrix_room()
         if len(invites) < 2:
-            # TODO[waiting-for-bots] This won't happen when the bot is enabled
             raise ValueError("Not enough Telegram users to create a chat")
 
         if self.peer_type == "chat":
@@ -658,6 +727,9 @@ class Portal:
         await self.update_info(source, entity)
         self.save()
 
+        if self.bot and self.bot.mxid in invites:
+            self.bot.add_chat(self.tgid, self.peer_type)
+
         levels = await self.main_intent.get_power_levels(self.mxid)
         levels = self._get_base_power_levels(levels, entity)
         already_saved = await self.handle_matrix_power_levels(source, levels["users"], {})
@@ -864,7 +936,8 @@ class Portal:
         self.db.add(DBMessage(tgid=evt.id, mx_room=self.mxid, mxid=mxid, tg_space=tg_space))
         self.db.commit()
 
-    async def handle_telegram_action(self, source, sender, action):
+    async def handle_telegram_action(self, source, sender, update):
+        action = update.action
         if not self.mxid:
             create_and_exit = (MessageActionChatCreate, MessageActionChannelCreate)
             create_and_continue = (MessageActionChatAddUser, MessageActionChatJoinedByLink)
@@ -874,6 +947,9 @@ class Portal:
             if not isinstance(action, create_and_continue):
                 return
 
+        if self.is_duplicate_action(update):
+            return
+
         # TODO figure out how to see changes to about text / channel username
         if isinstance(action, MessageActionChatEditTitle):
             if await self.update_title(action.title):
@@ -898,7 +974,10 @@ class Portal:
         else:
             self.log.debug("Unhandled Telegram action in %s: %s", self.title, action)
 
-    async def set_telegram_admin(self, puppet, user):
+    async def set_telegram_admin(self, user_id):
+        puppet = p.Puppet.get(user_id)
+        user = await u.User.get_by_tgid(user_id)
+
         levels = await self.main_intent.get_power_levels(self.mxid)
         if user:
             levels["users"][user.mxid] = 50
@@ -1078,4 +1157,5 @@ class Portal:
 
 def init(context):
     global config
-    Portal.az, Portal.db, config, _ = context
+    Portal.az, Portal.db, config, _, Portal.bot = context
+    Portal.bridge_notices = config["bridge.bridge_notices"]

mautrix_telegram/puppet.py

@@ -31,15 +31,14 @@ class Puppet:
     db = None
     az = None
     mxid_regex = None
+    username_template = None
+    hs_domain = None
     cache = {}
 
     def __init__(self, id=None, username=None, displayname=None, photo_id=None):
         self.id = id
+        self.mxid = self.get_mxid_from_id(self.id)
 
-        self.localpart = config.get("bridge.username_template", "telegram_{userid}").format(
-            userid=self.id)
-        hs = config["homeserver"]["domain"]
-        self.mxid = f"@{self.localpart}:{hs}"
         self.username = username
         self.displayname = displayname
         self.photo_id = photo_id
@@ -161,6 +160,10 @@ class Puppet:
             return int(match.group(1))
         return None
 
+    @classmethod
+    def get_mxid_from_id(cls, id):
+        return f"@{cls.username_template.format(userid=id)}:{cls.hs_domain}"
+
     @classmethod
     def find_by_username(cls, username):
         for _, puppet in cls.cache.items():
@@ -176,7 +179,8 @@ class Puppet:
 
 def init(context):
     global config
-    Puppet.az, Puppet.db, config, _ = context
-    localpart = config.get("bridge.username_template", "telegram_{userid}").format(userid="(.+)")
-    hs = config["homeserver"]["domain"]
-    Puppet.mxid_regex = re.compile(f"@{localpart}:{hs}")
+    Puppet.az, Puppet.db, config, _, _ = context
+    Puppet.username_template = config.get("bridge.username_template", "telegram_{userid}")
+    Puppet.hs_domain = config["homeserver"]["domain"]
+    localpart = Puppet.username_template.format(userid="(.+)")
+    Puppet.mxid_regex = re.compile(f"@{localpart}:{Puppet.hs_domain}")

mautrix_telegram/user.py

@@ -16,31 +16,28 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 import logging
 import asyncio
-import platform
+import re
 
 from telethon.tl.types import *
 from telethon.tl.types.contacts import ContactsNotModified
-from telethon.tl.types import User as TLUser
 from telethon.tl.functions.contacts import GetContactsRequest, SearchRequest
 from mautrix_appservice import MatrixRequestError
 
-from .db import User as DBUser, Message as DBMessage, Contact as DBContact
-from .tgclient import MautrixTelegramClient
-from . import portal as po, puppet as pu, __version__
+from .db import User as DBUser, Contact as DBContact
+from .abstract_user import AbstractUser
+from . import portal as po, puppet as pu
 
 config = None
 
 
-class User:
-    loop = None
+class User(AbstractUser):
     log = logging.getLogger("mau.user")
-    db = None
-    az = None
     by_mxid = {}
     by_tgid = {}
 
     def __init__(self, mxid, tgid=None, username=None, db_contacts=None, saved_contacts=0,
                  db_portals=None):
+        super().__init__()
         self.mxid = mxid
         self.tgid = tgid
         self.username = username
@@ -51,9 +48,6 @@ class User:
         self.db_portals = db_portals
 
         self.command_status = None
-        self.connected = False
-        self.client = None
-        self._init_client()
 
         self.is_admin = self.mxid in config.get("bridge.admins", [])
 
@@ -67,13 +61,17 @@ class User:
         if tgid:
             self.by_tgid[tgid] = self
 
-    @property
-    def logged_in(self):
-        return self.client.is_user_authorized()
+        self._init_client()
 
     @property
-    def has_full_access(self):
-        return self.logged_in and self.whitelisted
+    def name(self):
+        return self.mxid
+
+    @property
+    def displayname(self):
+        # TODO show better username
+        match = re.compile("@(.+):(.+)").match(self.mxid)
+        return match.group(1)
 
     @property
     def db_contacts(self):
@@ -129,23 +127,13 @@ class User:
     # endregion
     # region Telegram connection management
 
-    def _init_client(self):
-        device = f"{platform.system()} {platform.release()}"
-        sysversion = MautrixTelegramClient.__version__
-        self.client = MautrixTelegramClient(self.mxid,
-                                            config["telegram.api_id"],
-                                            config["telegram.api_hash"],
-                                            loop=self.loop,
-                                            app_version=__version__,
-                                            system_version=sysversion,
-                                            device_model=device)
-        self.client.add_update_handler(self.update_catch)
-
     async def start(self, delete_unless_authenticated=False):
-        self.connected = await self.client.connect()
+        await super().start()
         if self.logged_in:
+            self.log.debug(f"Ensuring post_login() for {self.name}")
             asyncio.ensure_future(self.post_login(), loop=self.loop)
         elif delete_unless_authenticated:
+            self.log.debug(f"Unauthenticated user {self.name} start()ed, deleting...")
             # User not logged in -> forget user
             self.client.disconnect()
             self.client.session.delete()
@@ -160,11 +148,6 @@ class User:
         except Exception:
             self.log.exception("Failed to run post-login functions")
 
-    def stop(self):
-        self.client.disconnect()
-        self.client = None
-        self.connected = False
-
     # endregion
     # region Telegram actions that need custom methods
 
@@ -234,14 +217,8 @@ class User:
         return await self._search_remote(query), True
 
     async def sync_dialogs(self):
-        dialogs = await self.client.get_dialogs(limit=30)
         creators = []
-        for dialog in dialogs:
-            entity = dialog.entity
-            invalid = (isinstance(entity, (TLUser, ChatForbidden, ChannelForbidden))
-                       or (isinstance(entity, Chat) and (entity.deactivated or entity.left)))
-            if invalid:
-                continue
+        for entity in await self._get_dialogs(limit=30):
             portal = po.Portal.get_by_entity(entity)
             self.portals[portal.tgid_full] = portal
             creators.append(portal.create_matrix_room(self, entity, invites=[self.mxid]))
@@ -283,135 +260,6 @@ class User:
             self.contacts.append(puppet)
         self.save()
 
-    # endregion
-    # region Telegram update handling
-
-    async def update_catch(self, update):
-        try:
-            await self.update(update)
-        except Exception:
-            self.log.exception("Failed to handle Telegram update")
-
-    async def update(self, update):
-        if isinstance(update, (UpdateShortChatMessage, UpdateShortMessage, UpdateNewChannelMessage,
-                               UpdateNewMessage, UpdateEditMessage, UpdateEditChannelMessage)):
-            await self.update_message(update)
-        elif isinstance(update, (UpdateChatUserTyping, UpdateUserTyping)):
-            await self.update_typing(update)
-        elif isinstance(update, UpdateUserStatus):
-            await self.update_status(update)
-        elif isinstance(update, (UpdateChatAdmins, UpdateChatParticipantAdmin)):
-            await self.update_admin(update)
-        elif isinstance(update, UpdateChatParticipants):
-            portal = po.Portal.get_by_tgid(update.participants.chat_id)
-            if portal and portal.mxid:
-                await portal.update_telegram_participants(update.participants.participants)
-        elif isinstance(update, UpdateChannelPinnedMessage):
-            portal = po.Portal.get_by_tgid(update.channel_id)
-            if portal and portal.mxid:
-                await portal.update_telegram_pin(self, update.id)
-        elif isinstance(update, (UpdateUserName, UpdateUserPhoto)):
-            await self.update_others_info(update)
-        elif isinstance(update, UpdateReadHistoryOutbox):
-            await self.update_read_receipt(update)
-        else:
-            self.log.debug("Unhandled update: %s", update)
-
-    async def update_read_receipt(self, update):
-        if not isinstance(update.peer, PeerUser):
-            self.log.debug("Unexpected read receipt peer: %s", update.peer)
-            return
-
-        portal = po.Portal.get_by_tgid(update.peer.user_id, self.tgid)
-        if not portal or not portal.mxid:
-            return
-
-        # We check that these are user read receipts, so tg_space is always the user ID.
-        message = DBMessage.query.get((update.max_id, self.tgid))
-        if not message:
-            return
-
-        puppet = pu.Puppet.get(update.peer.user_id)
-        await puppet.intent.mark_read(portal.mxid, message.mxid)
-
-    async def update_admin(self, update):
-        portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
-        if isinstance(update, UpdateChatAdmins):
-            await portal.set_telegram_admins_enabled(update.enabled)
-        elif isinstance(update, UpdateChatParticipantAdmin):
-            puppet = pu.Puppet.get(update.user_id)
-            user = User.get_by_tgid(update.user_id)
-            await portal.set_telegram_admin(puppet, user)
-
-    async def update_typing(self, update):
-        if isinstance(update, UpdateUserTyping):
-            portal = po.Portal.get_by_tgid(update.user_id, self.tgid, "user")
-        else:
-            portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
-        sender = pu.Puppet.get(update.user_id)
-        await portal.handle_telegram_typing(sender, update)
-
-    async def update_others_info(self, update):
-        puppet = pu.Puppet.get(update.user_id)
-        if isinstance(update, UpdateUserName):
-            if await puppet.update_displayname(self, update):
-                puppet.save()
-        elif isinstance(update, UpdateUserPhoto):
-            if await puppet.update_avatar(self, update.photo.photo_big):
-                puppet.save()
-
-    async def update_status(self, update):
-        puppet = pu.Puppet.get(update.user_id)
-        if isinstance(update.status, UserStatusOnline):
-            await puppet.intent.set_presence("online")
-        elif isinstance(update.status, UserStatusOffline):
-            await puppet.intent.set_presence("offline")
-        else:
-            self.log.warning("Unexpected user status update: %s", update)
-        return
-
-    def get_message_details(self, update):
-        if isinstance(update, UpdateShortChatMessage):
-            portal = po.Portal.get_by_tgid(update.chat_id, peer_type="chat")
-            sender = pu.Puppet.get(update.from_id)
-        elif isinstance(update, UpdateShortMessage):
-            portal = po.Portal.get_by_tgid(update.user_id, self.tgid, "user")
-            sender = pu.Puppet.get(self.tgid if update.out else update.user_id)
-        elif isinstance(update, (UpdateNewMessage, UpdateNewChannelMessage,
-                                 UpdateEditMessage, UpdateEditChannelMessage)):
-            update = update.message
-            if isinstance(update.to_id, PeerUser) and not update.out:
-                portal = po.Portal.get_by_tgid(update.from_id, peer_type="user",
-                                               tg_receiver=self.tgid)
-            else:
-                portal = po.Portal.get_by_entity(update.to_id, receiver_id=self.tgid)
-            sender = pu.Puppet.get(update.from_id) if update.from_id else None
-        else:
-            self.log.warning(
-                f"Unexpected message type in User#get_message_details: {type(update)}")
-            return update, None, None
-        return update, sender, portal
-
-    def update_message(self, original_update):
-        update, sender, portal = self.get_message_details(original_update)
-
-        if isinstance(update, MessageService):
-            if isinstance(update.action, MessageActionChannelMigrateFrom):
-                self.log.debug(f"Ignoring action %s to %s by %d", update.action, portal.tgid_log,
-                               sender.id)
-                return
-            self.log.debug("Handling action %s to %s by %d", update.action, portal.tgid_log,
-                           sender.id)
-            return portal.handle_telegram_action(self, sender, update.action)
-
-        user = sender.tgid if sender else "admin"
-        if isinstance(original_update, (UpdateEditMessage, UpdateEditChannelMessage)):
-            self.log.debug("Handling edit %s to %s by %s", update, portal.tgid_log, user)
-            return portal.handle_telegram_edit(self, sender, update)
-
-        self.log.debug("Handling message %s to %s by %s", update, portal.tgid_log, user)
-        return portal.handle_telegram_message(self, sender, update)
-
     # endregion
     # region Class instance lookup
 
@@ -425,14 +273,12 @@ class User:
         user = DBUser.query.get(mxid)
         if user:
             user = cls.from_db(user)
-            asyncio.ensure_future(user.start(), loop=cls.loop)
             return user
 
         if create:
             user = cls(mxid)
             cls.db.add(user.to_db())
             cls.db.commit()
-            asyncio.ensure_future(user.start(), loop=cls.loop)
             return user
 
         return None
@@ -447,7 +293,6 @@ class User:
         user = DBUser.query.filter(DBUser.tgid == tgid).one_or_none()
         if user:
             user = cls.from_db(user)
-            asyncio.ensure_future(user.start(), loop=cls.loop)
             return user
 
         return None
@@ -468,7 +313,7 @@ class User:
 
 def init(context):
     global config
-    User.az, User.db, config, User.loop = context
+    config = context.config
 
     users = [User.from_db(user) for user in DBUser.query.all()]
     return [user.start(delete_unless_authenticated=True) for user in users]

setup.py

@@ -17,12 +17,12 @@ setuptools.setup(
 
     install_requires=[
         "aiohttp>=2.3.10,<3",
-        "SQLAlchemy>=1.2.2,<2",
+        "SQLAlchemy>=1.2.3,<2",
         "alembic>=0.9.7",
         "Markdown>=2.6.11,<3",
         "ruamel.yaml>=0.15.35,<0.16",
         "Pillow>=5.0.0,<6",
-        "future-fstrings>=0.4.1",
+        "future-fstrings>=0.4.2",
         "python-magic>=0.4.15,<0.5",
     ],
     dependency_links=[