integrations: Enhanced matrix bridge.

requirements.txt

@@ -15,3 +15,4 @@ types-python-dateutil
 types-pytz
 types-requests
 gitlint>=0.13.0
+-r ./zulip/integrations/bridge_with_matrix/requirements.txt

zulip/integrations/bridge_with_matrix/README.md

@@ -1,20 +1,24 @@
 # Matrix <--> Zulip bridge
 
-This acts as a bridge between Matrix and Zulip. It also enables a
+This acts as a bridge between Matrix and Zulip.
-Zulip topic to be federated between two Zulip servers.
 
-## Usage
+### Enhanced Features
+- Supporting multiple (Zulip topic, Matrix channel)-pairs.
+- Handling files according to their mimetype.
 
-### For IRC bridges
 
-Matrix has been bridged to the listed
+## Installation
-[IRC networks](https://github.com/matrix-org/matrix-appservice-irc/wiki/Bridged-IRC-networks),
-where the 'Room alias format' refers to the `room_id` for the corresponding IRC channel.
 
-For example, for the freenode channel `#zulip-test`, the `room_id` would be
+Run `pip install -r requirements.txt` in order to install the requirements.
-`#freenode_#zulip-test:matrix.org`.
+
+In case you'd like encryption to work, you need pip to install the `matrix-nio`
+package with e2e support:
+- First, you need to make sure that the development files of the `libolm`
+  C-library are installed on your system! See [the corresponding documentation
+  of matrix-nio](https://github.com/poljar/matrix-nio#installation) for further
+  information on this point.
+- `pip install matrix-nio[e2e]`
 
-Hence, this can also be used as a IRC <--> Zulip bridge.
 
 ## Steps to configure the Matrix bridge
 
@@ -27,19 +31,43 @@ details mentioned below. For example:
 * If you are running from the Zulip GitHub repo: `python matrix_bridge.py --write-sample-config matrix_bridge.conf`
 
 ### 1. Zulip endpoint
-1. Create a generic Zulip bot, with a full name like `IRC Bot` or `Matrix Bot`.
+1. Create a generic Zulip bot, with a full name such as `Matrix Bot`.
-2. Subscribe the bot user to the stream you'd like to bridge your IRC or Matrix
+2. The bot is able to subscribe to the necessary streams itself if they are
-   channel into.
+   public. (Note that the bridge will not try to create streams in case they
+   do not already exist. In that case, the bridge will fail at startup.)
+   Otherwise, you need to add the bot manually.
 3. In the `zulip` section of the configuration file, enter the bot's `zuliprc`
    details (`email`, `api_key`, and `site`).
 4. In the same section, also enter the Zulip `stream` and `topic`.
 
 ### 2. Matrix endpoint
-1. Create a user on [matrix.org](https://matrix.org/), preferably with
+1. Create a user on the matrix server of your choice, e.g. [matrix.org](https://matrix.org/),
-   a formal name like to `zulip-bot`.
+   preferably with a descriptive name such as `zulip-bot`.
-2. In the `matrix` section of the configuration file, enter the user's username
+2. In the `matrix` section of the configuration file, enter the user's Matrix
-   and password.
+   user ID `mxid` and password. Please use the Matrix user ID ([MXID](https://matrix.org/faq/#what-is-a-mxid%3F))
-3. Also enter the `host` and `room_id` into the same section.
+   as format for the username!
+3. Create the Matrix room(s) to be bridged in case they do not exits yet.
+   Remember to invite the bot to private rooms! Otherwise, this error will be
+   thrown: `Matrix bridge error: JoinError: M_UNKNOWN No known servers`.
+4. Enter the `host` and `room_id` into the same section.
+   In case the room is private you need to use the `Internal room ID` which has
+   the format `!aBcDeFgHiJkLmNoPqR:example.org`.
+   In the official Matrix client [Element](https://github.com/vector-im), you
+   can find this `Internal room ID` in the `Room Settings` under `Advanced`.
+
+### Adding more (Zulip topic, Matrix channel)-pairs
+1. Create a new section with a name starting with `additional_bridge`.
+2. Add a `room_id` for the Matrix side and a `stream` and a `topic` for the
+   Zulip side.
+
+Example:
+```
+[additional_bridge1]
+room_id = #zulip:matrix.org
+stream = matrix test
+topic = matrix test topic
+```
+
 
 ## Running the bridge
 
@@ -50,9 +78,29 @@ in a file called `matrix_bridge.conf`:
 
 * If you are running from the Zulip GitHub repo: run `python matrix_bridge.py -c matrix_bridge.conf`
 
-## Caveats for IRC mirroring
+
+## Notes regarding IRC
+
+### Usage for IRC bridges
+
+This can also be used to indirectly bridge between IRC and Zulip.
+
+Matrix has been bridged to the listed
+[IRC networks](https://matrix-org.github.io/matrix-appservice-irc/latest/bridged_networks.html),
+where the 'Room alias format' refers to the `room_id` for the corresponding IRC channel.
+
+For example, for the Libera Chat channel `#zulip-test`, the `room_id` would be
+`#zulip-test:libera.chat`.
+
+### Caveats for IRC mirroring
 
 There are certain
 [IRC channels](https://github.com/matrix-org/matrix-appservice-irc/wiki/Channels-from-which-the-IRC-bridge-is-banned)
 where the Matrix.org IRC bridge has been banned for technical reasons.
 You can't mirror those IRC channels using this integration.
+
+
+## TODO
+
+- Adding support for editing and deleting messages?
+- Handling encryption on the Matrix side (may need further discussion).

zulip/integrations/bridge_with_matrix/matrix_bridge.conf

@@ -0,0 +1,18 @@
+[matrix]
+host = https://matrix.org
+mxid = @username:matrix.org
+password = password
+room_id = #zulip:matrix.org
+
+[zulip]
+email = glitch-bot@chat.zulip.org
+api_key = aPiKeY
+site = https://chat.zulip.org
+stream = test here
+topic = matrix
+
+[additional_bridge1]
+room_id = #example:matrix.org
+stream = new test
+topic = matrix
+

zulip/integrations/bridge_with_matrix/matrix_bridge.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 import argparse
+import asyncio
 import configparser
 import logging
 import os
@@ -7,22 +8,34 @@ import re
 import signal
 import sys
 import traceback
+import urllib.request
 from collections import OrderedDict
-from types import FrameType
+from concurrent.futures import ThreadPoolExecutor
-from typing import Any, Callable, Dict, Optional
+from io import BytesIO
+from typing import Any, Dict, List, Match, Optional, Set, Tuple, Type, Union
 
-from matrix_client.client import MatrixClient
+if os.name != "nt":
-from matrix_client.errors import MatrixRequestError
+    import magic
-from requests.exceptions import MissingSchema
+    import magic.compat
+import nio
+from nio.responses import (
+    DownloadError,
+    DownloadResponse,
+    ErrorResponse,
+    JoinError,
+    JoinResponse,
+    LoginError,
+    LoginResponse,
+    Response,
+    SyncError,
+    SyncResponse,
+)
 
 import zulip
 
-GENERAL_NETWORK_USERNAME_REGEX = "@_?[a-zA-Z0-9]+_([a-zA-Z0-9-_]+):[a-zA-Z0-9.]+"
-MATRIX_USERNAME_REGEX = "@([a-zA-Z0-9-_]+):matrix.org"
-
 # change these templates to change the format of displayed message
-ZULIP_MESSAGE_TEMPLATE = "**{username}**: {message}"
+ZULIP_MESSAGE_TEMPLATE: str = "**{username}** [{uid}]: {message}"
-MATRIX_MESSAGE_TEMPLATE = "<{username}> {message}"
+MATRIX_MESSAGE_TEMPLATE: str = "<{username} ({uid})> {message}"
 
 
 class Bridge_ConfigException(Exception):
@@ -33,161 +46,388 @@ class Bridge_FatalMatrixException(Exception):
     pass
 
 
-class Bridge_ZulipFatalException(Exception):
+class Bridge_FatalZulipException(Exception):
     pass
 
 
-def matrix_login(matrix_client: Any, matrix_config: Dict[str, Any]) -> None:
+class MatrixToZulip:
-    try:
+    """
-        matrix_client.login_with_password(matrix_config["mxid"], matrix_config["password"])
+    Matrix -> Zulip
-    except MatrixRequestError as exception:
+    """
-        if exception.code == 403:
+
-            raise Bridge_FatalMatrixException("Bad mxid or password.")
+    non_formatted_messages: Dict[Type[nio.Event], str] = {
+        nio.StickerEvent: "sticker",
+    }
+
+    def __init__(
+        self,
+        matrix_client: nio.AsyncClient,
+        matrix_config: Dict[str, Any],
+        zulip_client: zulip.Client,
+        no_noise: bool,
+    ) -> None:
+        self.matrix_client: nio.AsyncClient = matrix_client
+        self.matrix_config: Dict[str, Any] = matrix_config
+        self.zulip_client: zulip.Client = zulip_client
+        self.no_noise: bool = no_noise
+
+    @classmethod
+    async def create(
+        cls,
+        matrix_client: nio.AsyncClient,
+        matrix_config: Dict[str, Any],
+        zulip_client: zulip.Client,
+        no_noise: bool,
+    ) -> "MatrixToZulip":
+        matrix_to_zulip: "MatrixToZulip" = cls(matrix_client, matrix_config, zulip_client, no_noise)
+
+        # Login to Matrix
+        await matrix_to_zulip.matrix_login()
+        await matrix_to_zulip.matrix_join_rooms()
+
+        # Do an initial sync of the matrix client in order to continue with
+        # the new messages and not with all the old ones.
+        sync_response: Union[SyncResponse, SyncError] = await matrix_client.sync()
+        if isinstance(sync_response, nio.SyncError):
+            raise Bridge_FatalMatrixException(sync_response.message)
+
+        return matrix_to_zulip
+
+    async def _matrix_to_zulip(self, room: nio.MatrixRoom, event: nio.Event) -> None:
+        logging.debug("_matrix_to_zulip; room %s, event: %s" % (str(room.room_id), str(event)))
+
+        # We do this to identify the messages generated from Zulip -> Matrix
+        # and we make sure we don't forward it again to the Zulip stream.
+        if event.sender == self.matrix_config["mxid"]:
+            return
+
+        if room.room_id not in self.matrix_config["bridges"]:
+            return
+        stream, topic = self.matrix_config["bridges"][room.room_id]
+
+        content: Optional[str] = await self.get_message_content_from_event(event, room)
+        if not content:
+            return
+
+        try:
+            result: Dict[str, Any] = self.zulip_client.send_message(
+                {
+                    "type": "stream",
+                    "to": stream,
+                    "subject": topic,
+                    "content": content,
+                }
+            )
+        except Exception as exception:
+            # Generally raised when user is forbidden
+            raise Bridge_FatalZulipException(exception)
+        if result["result"] != "success":
+            # Generally raised when API key is invalid
+            raise Bridge_FatalZulipException(result["msg"])
+
+        # Update the bot's read marker in order to show the other users which
+        # messages are already processed by the bot.
+        await self.matrix_client.room_read_markers(
+            room.room_id, fully_read_event=event.event_id, read_event=event.event_id
+        )
+
+    async def get_message_content_from_event(
+        self,
+        event: nio.Event,
+        room: nio.MatrixRoom,
+    ) -> Optional[str]:
+        message: str
+        sender: Optional[str] = room.user_name(event.sender)
+
+        if isinstance(event, nio.RoomMemberEvent):
+            if self.no_noise:
+                return None
+            # Join and leave events can be noisy. They are ignored by default.
+            # To enable these events pass `no_noise` as `False` as the script argument
+            message = event.state_key + " " + event.membership
+        elif isinstance(event, nio.RoomMessageFormatted):
+            message = event.body
+        elif isinstance(event, nio.RoomMessageMedia):
+            message = await self.handle_media(event)
+        elif type(event) in self.non_formatted_messages:
+            message = "sends " + self.non_formatted_messages[type(event)]
+        elif isinstance(event, nio.MegolmEvent):
+            message = "sends an encrypted message"
+        elif isinstance(event, nio.UnknownEvent) and event.type == "m.reaction":
+            return None
         else:
-            raise Bridge_FatalMatrixException("Check if your server details are correct.")
+            message = "event: " + type(event).__name__
-    except MissingSchema:
-        raise Bridge_FatalMatrixException("Bad URL format.")
 
+        return ZULIP_MESSAGE_TEMPLATE.format(username=sender, uid=event.sender, message=message)
 
-def matrix_join_room(matrix_client: Any, matrix_config: Dict[str, Any]) -> Any:
+    async def handle_media(self, event: nio.RoomMessageMedia) -> str:
-    try:
+        """Parse a nio.RoomMessageMedia event.
-        room = matrix_client.join_room(matrix_config["room_id"])
+
-        return room
+        Upload the media to zulip and build an appropriate message.
-    except MatrixRequestError as exception:
+        """
-        if exception.code == 403:
+        # Split the mxc uri in "server_name" and "media_id".
-            raise Bridge_FatalMatrixException("Room ID/Alias in the wrong format")
+        mxc_match: Optional[Match[str]] = re.fullmatch("mxc://([^/]+)/([^/]+)", event.url)
+        if mxc_match is None or len(mxc_match.groups()) != 2:
+            return "[message from bridge: media could not be handled]"
+        server_name, media_id = mxc_match.groups()
+
+        download: Union[DownloadResponse, DownloadError] = await self.matrix_client.download(
+            server_name, media_id
+        )
+        if isinstance(download, nio.DownloadError):
+            return "[message from bridge: media could not be downloaded]"
+
+        file_fake: BytesIO = BytesIO(download.body)
+        # zulip.client.do_api_query() needs a name. TODO: hacky...
+        file_fake.name = download.filename
+
+        result: Dict[str, Any] = self.zulip_client.upload_file(file_fake)
+        if result["result"] != "success":
+            return "[message from bridge: media could not be uploaded]"
+
+        message: str
+        if download.filename:
+            message = "[{}]({})".format(download.filename, result["uri"])
         else:
-            raise Bridge_FatalMatrixException("Couldn't find room.")
+            message = result["uri"]
+
+        return message
+
+    async def matrix_join_rooms(self) -> None:
+        for room_id in self.matrix_config["bridges"]:
+            result: Union[JoinResponse, JoinError] = await self.matrix_client.join(room_id)
+            if isinstance(result, nio.JoinError):
+                raise Bridge_FatalMatrixException(str(result))
+
+    async def matrix_login(self) -> None:
+        result: Union[LoginResponse, LoginError] = await self.matrix_client.login(
+            self.matrix_config["password"]
+        )
+        if isinstance(result, nio.LoginError):
+            raise Bridge_FatalMatrixException(str(result))
+
+    async def run(self) -> None:
+        print("Starting message handler on Matrix client")
+
+        # Set up event callback.
+        self.matrix_client.add_event_callback(self._matrix_to_zulip, nio.Event)
+
+        await self.matrix_client.sync_forever()
 
 
-def die(signal: int, frame: FrameType) -> None:
+class ZulipToMatrix:
+    """
+    Zulip -> Matrix
+    """
+
+    def __init__(
+        self,
+        zulip_client: zulip.Client,
+        zulip_config: Dict[str, Any],
+        matrix_client: nio.AsyncClient,
+    ) -> None:
+        self.zulip_client: zulip.Client = zulip_client
+        self.zulip_config: Dict[str, Any] = zulip_config
+        self.matrix_client: nio.AsyncClient = matrix_client
+        self.loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+        # Precompute the url of the Zulip server, needed later.
+        result: Dict[str, Any] = self.zulip_client.get_server_settings()
+        if result["result"] != "success":
+            raise Bridge_FatalZulipException("cannot get server settings")
+        self.server_url: str = result["realm_uri"]
+
+    @classmethod
+    async def create(
+        cls,
+        zulip_client: zulip.Client,
+        zulip_config: Dict[str, Any],
+        matrix_client: nio.AsyncClient,
+    ) -> "ZulipToMatrix":
+        zulip_to_matrix: "ZulipToMatrix" = cls(zulip_client, zulip_config, matrix_client)
+        zulip_to_matrix.ensure_stream_membership()
+        return zulip_to_matrix
+
+    def _matrix_send(self, **kwargs: Any) -> None:
+        """Wrapper for sending messages to the matrix server."""
+        result: Union[Response, ErrorResponse] = asyncio.run_coroutine_threadsafe(
+            self.matrix_client.room_send(**kwargs), self.loop
+        ).result()
+        if isinstance(result, nio.RoomSendError):
+            raise Bridge_FatalMatrixException(str(result))
+
+    def _zulip_to_matrix(self, msg: Dict[str, Any]) -> None:
+        logging.debug("_zulip_to_matrix; msg: %s" % (str(msg),))
+
+        room_id: Optional[str] = self.get_matrix_room_for_zulip_message(msg)
+        if room_id is None:
+            return
+
+        sender: str = msg["sender_full_name"]
+        content: str = MATRIX_MESSAGE_TEMPLATE.format(
+            username=sender, uid=msg["sender_id"], message=msg["content"]
+        )
+
+        # Forward Zulip message to Matrix.
+        self._matrix_send(
+            room_id=room_id,
+            message_type="m.room.message",
+            content={"msgtype": "m.text", "body": content},
+        )
+
+        # Get embedded files.
+        files_to_send, media_success = asyncio.run_coroutine_threadsafe(
+            self.handle_media(msg["content"]), self.loop
+        ).result()
+
+        if files_to_send:
+            self._matrix_send(
+                room_id=room_id,
+                message_type="m.room.message",
+                content={"msgtype": "m.text", "body": "This message contains the following files:"},
+            )
+            for file in files_to_send:
+                self._matrix_send(room_id=room_id, message_type="m.room.message", content=file)
+        if not media_success:
+            self._matrix_send(
+                room_id=room_id,
+                message_type="m.room.message",
+                content={
+                    "msgtype": "m.text",
+                    "body": "This message contained some files which could not be forwarded.",
+                },
+            )
+
+    def ensure_stream_membership(self) -> None:
+        """Ensure that the client is member of all necessary streams.
+
+        Note that this may create streams if they do not exist and if
+        the bot has enough rights to do so.
+        """
+        for stream, _ in self.zulip_config["bridges"]:
+            result: Dict[str, Any] = self.zulip_client.get_stream_id(stream)
+            if result["result"] == "error":
+                raise Bridge_FatalZulipException(f"cannot access stream '{stream}': {result}")
+            if result["result"] != "success":
+                raise Bridge_FatalZulipException(f"cannot checkout stream id for stream '{stream}'")
+            result = self.zulip_client.add_subscriptions(streams=[{"name": stream}])
+            if result["result"] != "success":
+                raise Bridge_FatalZulipException(f"cannot subscribe to stream '{stream}': {result}")
+
+    def get_matrix_room_for_zulip_message(self, msg: Dict[str, Any]) -> Optional[str]:
+        """Check whether we want to process the given message.
+
+        Return the room to which the given message should be forwarded, or
+        None if we do not want to process the given message.
+        """
+        if msg["type"] != "stream":
+            return None
+
+        # We do this to identify the messages generated from Matrix -> Zulip
+        # and we make sure we don't forward it again to the Matrix.
+        if msg["sender_email"] == self.zulip_config["email"]:
+            return None
+
+        key: Tuple[str, str] = (msg["display_recipient"], msg["subject"])
+        if key not in self.zulip_config["bridges"]:
+            return None
+
+        return self.zulip_config["bridges"][key]
+
+    async def handle_media(self, msg: str) -> Tuple[Optional[List[Dict[str, Any]]], bool]:
+        """Handle embedded media in the Zulip message.
+
+        Download the linked files from the Zulip server and upload them
+        to mthe matrix server.
+        Return a tuple containing the list of the messages which need
+        to be sent to the matrix room and a boolean flag indicating
+        whether there have been files for which the download/upload part
+        failed.
+        """
+        msgtype: str
+        files_to_send: List[Dict[str, Any]] = []
+        success: bool = True
+
+        for file in re.findall(r"\[[^\[\]]*\]\((/user_uploads/[^\(\)]*)\)", msg):
+            result: Dict[str, Any] = self.zulip_client.call_endpoint(file, method="GET")
+            if result["result"] != "success":
+                success = False
+                continue
+            try:
+                file_content: bytes = urllib.request.urlopen(self.server_url + result["url"]).read()
+            except Exception:
+                success = False
+                continue
+
+            mimetype: str
+            if os.name == "nt":
+                mimetype = "m.file"
+            else:
+                mimetype = magic.from_buffer(file_content, mime=True)
+            filename: str = file.split("/")[-1]
+
+            response, _ = await self.matrix_client.upload(
+                data_provider=BytesIO(file_content), content_type=mimetype, filename=filename
+            )
+            if isinstance(response, nio.UploadError):
+                success = False
+                continue
+
+            if mimetype.startswith("audio/"):
+                msgtype = "m.audio"
+            elif mimetype.startswith("image/"):
+                msgtype = "m.image"
+            elif mimetype.startswith("video/"):
+                msgtype = "m.video"
+            else:
+                msgtype = "m.file"
+
+            files_to_send.append(
+                {
+                    "body": filename,
+                    "info": {"mimetype": mimetype},
+                    "msgtype": msgtype,
+                    "url": response.content_uri,
+                }
+            )
+
+        return (files_to_send, success)
+
+    async def run(self) -> None:
+        print("Starting message handler on Zulip client")
+
+        self.loop = asyncio.get_event_loop()
+
+        with ThreadPoolExecutor() as executor:
+            await asyncio.get_event_loop().run_in_executor(
+                executor, self.zulip_client.call_on_each_message, self._zulip_to_matrix
+            )
+
+
+def die(*_: Any) -> None:
     # We actually want to exit, so run os._exit (so as not to be caught and restarted)
     os._exit(1)
 
 
-def matrix_to_zulip(
+def exception_handler(loop: asyncio.AbstractEventLoop, context: Dict[str, Any]) -> None:
-    zulip_client: zulip.Client,
+    loop.default_exception_handler(context)
-    zulip_config: Dict[str, Any],
+    os._exit(1)
-    matrix_config: Dict[str, Any],
-    no_noise: bool,
-) -> Callable[[Any, Dict[str, Any]], None]:
-    def _matrix_to_zulip(room: Any, event: Dict[str, Any]) -> None:
-        """
-        Matrix -> Zulip
-        """
-        content = get_message_content_from_event(event, no_noise)
-
-        zulip_bot_user = matrix_config["mxid"]
-        # We do this to identify the messages generated from Zulip -> Matrix
-        # and we make sure we don't forward it again to the Zulip stream.
-        not_from_zulip_bot = event["sender"] != zulip_bot_user
-
-        if not_from_zulip_bot and content:
-            try:
-                result = zulip_client.send_message(
-                    {
-                        "type": "stream",
-                        "to": zulip_config["stream"],
-                        "subject": zulip_config["topic"],
-                        "content": content,
-                    }
-                )
-            except Exception as exception:  # XXX This should be more specific
-                # Generally raised when user is forbidden
-                raise Bridge_ZulipFatalException(exception)
-            if result["result"] != "success":
-                # Generally raised when API key is invalid
-                raise Bridge_ZulipFatalException(result["msg"])
-
-    return _matrix_to_zulip
-
-
-def get_message_content_from_event(event: Dict[str, Any], no_noise: bool) -> Optional[str]:
-    irc_nick = shorten_irc_nick(event["sender"])
-    if event["type"] == "m.room.member":
-        if no_noise:
-            return None
-        # Join and leave events can be noisy. They are ignored by default.
-        # To enable these events pass `no_noise` as `False` as the script argument
-        if event["membership"] == "join":
-            content = ZULIP_MESSAGE_TEMPLATE.format(username=irc_nick, message="joined")
-        elif event["membership"] == "leave":
-            content = ZULIP_MESSAGE_TEMPLATE.format(username=irc_nick, message="quit")
-    elif event["type"] == "m.room.message":
-        if event["content"]["msgtype"] == "m.text" or event["content"]["msgtype"] == "m.emote":
-            content = ZULIP_MESSAGE_TEMPLATE.format(
-                username=irc_nick, message=event["content"]["body"]
-            )
-    else:
-        content = event["type"]
-    return content
-
-
-def shorten_irc_nick(nick: str) -> str:
-    """
-    Add nick shortner functions for specific IRC networks
-    Eg: For freenode change '@freenode_user:matrix.org' to 'user'
-    Check the list of IRC networks here:
-    https://github.com/matrix-org/matrix-appservice-irc/wiki/Bridged-IRC-networks
-    """
-    match = re.match(GENERAL_NETWORK_USERNAME_REGEX, nick)
-    if match:
-        return match.group(1)
-    # For matrix users
-    match = re.match(MATRIX_USERNAME_REGEX, nick)
-    if match:
-        return match.group(1)
-    return nick
-
-
-def zulip_to_matrix(config: Dict[str, Any], room: Any) -> Callable[[Dict[str, Any]], None]:
-    def _zulip_to_matrix(msg: Dict[str, Any]) -> None:
-        """
-        Zulip -> Matrix
-        """
-        message_valid = check_zulip_message_validity(msg, config)
-        if message_valid:
-            matrix_username = msg["sender_full_name"].replace(" ", "")
-            matrix_text = MATRIX_MESSAGE_TEMPLATE.format(
-                username=matrix_username, message=msg["content"]
-            )
-            # Forward Zulip message to Matrix
-            room.send_text(matrix_text)
-
-    return _zulip_to_matrix
-
-
-def check_zulip_message_validity(msg: Dict[str, Any], config: Dict[str, Any]) -> bool:
-    is_a_stream = msg["type"] == "stream"
-    in_the_specified_stream = msg["display_recipient"] == config["stream"]
-    at_the_specified_subject = msg["subject"] == config["topic"]
-
-    # We do this to identify the messages generated from Matrix -> Zulip
-    # and we make sure we don't forward it again to the Matrix.
-    not_from_zulip_bot = msg["sender_email"] != config["email"]
-    if is_a_stream and not_from_zulip_bot and in_the_specified_stream and at_the_specified_subject:
-        return True
-    return False
 
 
 def generate_parser() -> argparse.ArgumentParser:
-    description = """
+    description: str = """
-    Script to bridge between a topic in a Zulip stream, and a Matrix channel.
+    Bridge between Zulip topics and Matrix channels.
-
-    Tested connections:
-        * Zulip <-> Matrix channel
-        * Zulip <-> IRC channel (bridged via Matrix)
 
     Example matrix 'room_id' options might be, if via matrix.org:
         * #zulip:matrix.org (zulip channel on Matrix)
         * #freenode_#zulip:matrix.org (zulip channel on irc.freenode.net)"""
 
-    parser = argparse.ArgumentParser(
+    parser: argparse.ArgumentParser = argparse.ArgumentParser(
         description=description, formatter_class=argparse.RawTextHelpFormatter
     )
     parser.add_argument(
         "-c", "--config", required=False, help="Path to the config file for the bridge."
     )
+    parser.add_argument("-d", "--debug", action="store_true", help="debugging mode switch")
     parser.add_argument(
         "--write-sample-config",
         metavar="PATH",
@@ -210,27 +450,136 @@ def generate_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def read_configuration(config_file: str) -> Dict[str, Dict[str, str]]:
+def read_configuration(config_file: str) -> Dict[str, Dict[str, Any]]:
-    config = configparser.ConfigParser()
+    matrix_key_set: Set[str] = {"host", "mxid", "password"}
+    matrix_bridge_key_set: Set[str] = {"room_id"}
+    matrix_full_key_set: Set[str] = matrix_key_set | matrix_bridge_key_set
+    zulip_key_set: Set[str] = {"email", "api_key", "site"}
+    zulip_bridge_key_set: Set[str] = {"stream", "topic"}
+    zulip_full_key_set: Set[str] = zulip_key_set | zulip_bridge_key_set
+    bridge_key_set: Set[str] = {"room_id", "stream", "topic"}
+
+    config: configparser.ConfigParser = configparser.ConfigParser()
 
     try:
         config.read(config_file)
     except configparser.Error as exception:
         raise Bridge_ConfigException(str(exception))
 
-    if set(config.sections()) != {"matrix", "zulip"}:
+    if set(config.sections()) < {"matrix", "zulip"}:
         raise Bridge_ConfigException("Please ensure the configuration has zulip & matrix sections.")
 
-    # TODO Could add more checks for configuration content here
+    result: Dict[str, Dict[str, Any]] = {"matrix": {}, "zulip": {}}
+    # For Matrix: create a mapping with the Matrix room_ids as keys and
+    # the corresponding (stream, topic) tuple as value.
+    result["matrix"]["bridges"] = {}
+    # For Zulip: create a mapping with the tuple (stream, topic) as keys
+    # and the corresponding Matrix room_id as value.
+    result["zulip"]["bridges"] = {}
+    # One (and maybe the only) bridge is configured in the matrix/zulip
+    # sections to keep backwards compatibility with older configuration
+    # files.
+    first_bridge: Dict[str, Any] = {}
+    # Represent a (stream,topic) tuple.
+    zulip_target: Tuple[str, str]
 
-    return {section: dict(config[section]) for section in config.sections()}
+    for section in config.sections():
+        section_config: Dict[str, str] = dict(config[section])
+        section_keys: Set[str] = set(section_config.keys())
+
+        if section.startswith("additional_bridge"):
+            if section_keys != bridge_key_set:
+                raise Bridge_ConfigException(
+                    "Please ensure the bridge configuration section %s contain the following keys: %s."
+                    % (section, str(bridge_key_set))
+                )
+
+            zulip_target = (section_config["stream"], section_config["topic"])
+            result["zulip"]["bridges"][zulip_target] = section_config["room_id"]
+            result["matrix"]["bridges"][section_config["room_id"]] = zulip_target
+        elif section == "matrix":
+            if section_keys != matrix_full_key_set:
+                raise Bridge_ConfigException(
+                    "Please ensure the matrix configuration section contains the following keys: %s."
+                    % str(matrix_full_key_set)
+                )
+
+            result["matrix"].update({key: section_config[key] for key in matrix_key_set})
+
+            for key in matrix_bridge_key_set:
+                first_bridge[key] = section_config[key]
+
+            # Verify the format of the Matrix user ID.
+            if re.fullmatch(r"@[^:]+:.+", result["matrix"]["mxid"]) is None:
+                raise Bridge_ConfigException("Malformatted mxid.")
+        elif section == "zulip":
+            if section_keys != zulip_full_key_set:
+                raise Bridge_ConfigException(
+                    "Please ensure the zulip configuration section contains the following keys: %s."
+                    % str(zulip_full_key_set)
+                )
+
+            result["zulip"].update({key: section_config[key] for key in zulip_key_set})
+
+            for key in zulip_bridge_key_set:
+                first_bridge[key] = section_config[key]
+        else:
+            logging.warning("Unknown section %s" % (section,))
+
+    # Add the "first_bridge" to the bridges.
+    zulip_target = (first_bridge["stream"], first_bridge["topic"])
+    result["zulip"]["bridges"][zulip_target] = first_bridge["room_id"]
+    result["matrix"]["bridges"][first_bridge["room_id"]] = zulip_target
+
+    return result
+
+
+async def run(zulip_config: Dict[str, Any], matrix_config: Dict[str, Any], no_noise: bool) -> None:
+    asyncio.get_event_loop().set_exception_handler(exception_handler)
+
+    matrix_client: Optional[nio.AsyncClient] = None
+
+    print("Starting Zulip <-> Matrix mirroring bot")
+
+    # Initiate clients and start the event listeners.
+    backoff = zulip.RandomExponentialBackoff(timeout_success_equivalent=300)
+    while backoff.keep_going():
+        try:
+            zulip_client = zulip.Client(
+                email=zulip_config["email"],
+                api_key=zulip_config["api_key"],
+                site=zulip_config["site"],
+            )
+            matrix_client = nio.AsyncClient(matrix_config["host"], matrix_config["mxid"])
+
+            matrix_to_zulip: MatrixToZulip = await MatrixToZulip.create(
+                matrix_client, matrix_config, zulip_client, no_noise
+            )
+            zulip_to_matrix: ZulipToMatrix = await ZulipToMatrix.create(
+                zulip_client, zulip_config, matrix_client
+            )
+
+            await asyncio.gather(matrix_to_zulip.run(), zulip_to_matrix.run())
+
+        except Bridge_FatalMatrixException as exception:
+            sys.exit(f"Matrix bridge error: {exception}")
+        except Bridge_FatalZulipException as exception:
+            sys.exit(f"Zulip bridge error: {exception}")
+        except zulip.ZulipError as exception:
+            sys.exit(f"Zulip error: {exception}")
+        except Exception:
+            traceback.print_exc()
+        finally:
+            if matrix_client:
+                await matrix_client.close()
+        backoff.fail()
 
 
 def write_sample_config(target_path: str, zuliprc: Optional[str]) -> None:
     if os.path.exists(target_path):
         raise Bridge_ConfigException(f"Path '{target_path}' exists; not overwriting existing file.")
 
-    sample_dict = OrderedDict(
+    sample_dict: OrderedDict[str, OrderedDict[str, str]] = OrderedDict(
         (
             (
                 "matrix",
@@ -255,6 +604,16 @@ def write_sample_config(target_path: str, zuliprc: Optional[str]) -> None:
                     )
                 ),
             ),
+            (
+                "additional_bridge1",
+                OrderedDict(
+                    (
+                        ("room_id", "#example:matrix.org"),
+                        ("stream", "new test"),
+                        ("topic", "matrix"),
+                    )
+                ),
+            ),
         )
     )
 
@@ -262,19 +621,20 @@ def write_sample_config(target_path: str, zuliprc: Optional[str]) -> None:
         if not os.path.exists(zuliprc):
             raise Bridge_ConfigException(f"Zuliprc file '{zuliprc}' does not exist.")
 
-        zuliprc_config = configparser.ConfigParser()
+        zuliprc_config: configparser.ConfigParser = configparser.ConfigParser()
         try:
             zuliprc_config.read(zuliprc)
         except configparser.Error as exception:
             raise Bridge_ConfigException(str(exception))
 
-        # Can add more checks for validity of zuliprc file here
+        try:
+            sample_dict["zulip"]["email"] = zuliprc_config["api"]["email"]
+            sample_dict["zulip"]["site"] = zuliprc_config["api"]["site"]
+            sample_dict["zulip"]["api_key"] = zuliprc_config["api"]["key"]
+        except KeyError as exception:
+            raise Bridge_ConfigException("You provided an invalid zuliprc file: " + str(exception))
 
-        sample_dict["zulip"]["email"] = zuliprc_config["api"]["email"]
+    sample: configparser.ConfigParser = configparser.ConfigParser()
-        sample_dict["zulip"]["site"] = zuliprc_config["api"]["site"]
-        sample_dict["zulip"]["api_key"] = zuliprc_config["api"]["key"]
-
-    sample = configparser.ConfigParser()
     sample.read_dict(sample_dict)
     with open(target_path, "w") as target:
         sample.write(target)
@@ -282,10 +642,14 @@ def write_sample_config(target_path: str, zuliprc: Optional[str]) -> None:
 
 def main() -> None:
     signal.signal(signal.SIGINT, die)
+    signal.signal(signal.SIGTERM, die)
     logging.basicConfig(level=logging.WARNING)
 
-    parser = generate_parser()
+    parser: argparse.ArgumentParser = generate_parser()
-    options = parser.parse_args()
+    options: argparse.Namespace = parser.parse_args()
+
+    if options.debug:
+        logging.getLogger().setLevel(logging.DEBUG)
 
     if options.sample_config:
         try:
@@ -308,56 +672,18 @@ def main() -> None:
         sys.exit(1)
 
     try:
-        config = read_configuration(options.config)
+        config: Dict[str, Dict[str, Any]] = read_configuration(options.config)
     except Bridge_ConfigException as exception:
         print(f"Could not parse config file: {exception}")
         sys.exit(1)
 
     # Get config for each client
-    zulip_config = config["zulip"]
+    zulip_config: Dict[str, Any] = config["zulip"]
-    matrix_config = config["matrix"]
+    matrix_config: Dict[str, Any] = config["matrix"]
 
-    print(
+    loop: asyncio.AbstractEventLoop = asyncio.new_event_loop()
-        "IMPORTANT: Make sure that the bot accounts have been"
+    loop.run_until_complete(run(zulip_config, matrix_config, options.no_noise))
-        " subscribed to the relevant Matrix room / Zulip stream"
+    loop.close()
-    )
-
-    # Initiate clients
-    backoff = zulip.RandomExponentialBackoff(timeout_success_equivalent=300)
-    while backoff.keep_going():
-        print("Starting matrix mirroring bot (this may take a minute)")
-        try:
-            zulip_client = zulip.Client(
-                email=zulip_config["email"],
-                api_key=zulip_config["api_key"],
-                site=zulip_config["site"],
-            )
-            matrix_client = MatrixClient(matrix_config["host"])
-
-            # Login to Matrix
-            matrix_login(matrix_client, matrix_config)
-            # Join a room in Matrix
-            room = matrix_join_room(matrix_client, matrix_config)
-
-            room.add_listener(
-                matrix_to_zulip(zulip_client, zulip_config, matrix_config, options.no_noise)
-            )
-
-            print("Starting listener thread on Matrix client")
-            matrix_client.start_listener_thread()
-
-            print("Starting message handler on Zulip client")
-            zulip_client.call_on_each_message(zulip_to_matrix(zulip_config, room))
-
-        except Bridge_FatalMatrixException as exception:
-            sys.exit(f"Matrix bridge error: {exception}")
-        except Bridge_ZulipFatalException as exception:
-            sys.exit(f"Zulip bridge error: {exception}")
-        except zulip.ZulipError as exception:
-            sys.exit(f"Zulip error: {exception}")
-        except Exception:
-            traceback.print_exc()
-        backoff.fail()
 
 
 if __name__ == "__main__":

zulip/integrations/bridge_with_matrix/requirements.txt

@@ -1 +1,3 @@
-matrix-client==0.4.0
+matrix-nio
+python-magic
+python-magic-bin; platform_system == "Windows"

zulip/integrations/bridge_with_matrix/test_matrix.py

@@ -1,19 +1,21 @@
+import asyncio
 import os
 import shutil
 import sys
 from contextlib import contextmanager
 from subprocess import PIPE, Popen
 from tempfile import mkdtemp
+from typing import Any, Awaitable, Callable, Iterator, List
 from unittest import TestCase, mock
 
-from .matrix_bridge import check_zulip_message_validity, zulip_to_matrix
+import nio
+
+from .matrix_bridge import MatrixToZulip, ZulipToMatrix, read_configuration
 
 script_file = "matrix_bridge.py"
 script_dir = os.path.dirname(__file__)
 script = os.path.join(script_dir, script_file)
 
-from typing import Iterator, List
-
 sample_config_path = "matrix_test.conf"
 
 sample_config_text = """[matrix]
@@ -29,8 +31,29 @@ site = https://chat.zulip.org
 stream = test here
 topic = matrix
 
+[additional_bridge1]
+room_id = #example:matrix.org
+stream = new test
+topic = matrix
+
 """
 
+ZULIP_MESSAGE_TEMPLATE: str = "**{username}** [{uid}]: {message}"
+
+
+# For Python 3.6 compatibility.
+# (Since 3.8, there is unittest.IsolatedAsyncioTestCase!)
+# source: https://stackoverflow.com/a/46324983
+def async_test(coro: Callable[..., Awaitable[Any]]) -> Callable[..., Any]:
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        loop = asyncio.new_event_loop()
+        try:
+            return loop.run_until_complete(coro(*args, **kwargs))
+        finally:
+            loop.close()
+
+    return wrapper
+
 
 @contextmanager
 def new_temp_dir() -> Iterator[str]:
@@ -58,7 +81,7 @@ class MatrixBridgeScriptTests(TestCase):
     def test_help_usage_and_description(self) -> None:
         output_lines = self.output_from_script(["-h"])
         usage = f"usage: {script_file} [-h]"
-        description = "Script to bridge"
+        description = "Bridge between Zulip topics and Matrix channels."
         self.assertIn(usage, output_lines[0])
         blank_lines = [num for num, line in enumerate(output_lines) if line == ""]
         # There should be blank lines in the output
@@ -126,55 +149,116 @@ class MatrixBridgeScriptTests(TestCase):
                 ],
             )
 
+    def test_parse_multiple_bridges(self) -> None:
+        with new_temp_dir() as tempdir:
+            path = os.path.join(tempdir, sample_config_path)
+            output_lines = self.output_from_script(["--write-sample-config", path])
+            self.assertEqual(output_lines, [f"Wrote sample configuration to '{path}'"])
+
+            config = read_configuration(path)
+
+            self.assertIn("zulip", config)
+            self.assertIn("matrix", config)
+            self.assertIn("bridges", config["zulip"])
+            self.assertIn("bridges", config["matrix"])
+            self.assertEqual(
+                {
+                    ("test here", "matrix"): "#zulip:matrix.org",
+                    ("new test", "matrix"): "#example:matrix.org",
+                },
+                config["zulip"]["bridges"],
+            )
+            self.assertEqual(
+                {
+                    "#zulip:matrix.org": ("test here", "matrix"),
+                    "#example:matrix.org": ("new test", "matrix"),
+                },
+                config["matrix"]["bridges"],
+            )
+
+
+class MatrixBridgeMatrixToZulipTests(TestCase):
+    user_name = "John Smith"
+    user_uid = "@johnsmith:matrix.org"
+    room = mock.MagicMock()
+    room.user_name = lambda _: "John Smith"
+
+    def setUp(self) -> None:
+        self.matrix_to_zulip = mock.MagicMock()
+        self.matrix_to_zulip.get_message_content_from_event = (
+            lambda event: MatrixToZulip.get_message_content_from_event(
+                self.matrix_to_zulip, event, self.room
+            )
+        )
+
+    @async_test
+    async def test_get_message_content_from_event(self) -> None:
+        class RoomMemberEvent(nio.RoomMemberEvent):
+            def __init__(self, sender: str = self.user_uid) -> None:
+                self.sender = sender
+
+        class RoomMessageFormatted(nio.RoomMessageFormatted):
+            def __init__(self, sender: str = self.user_uid) -> None:
+                self.sender = sender
+                self.body = "this is a message"
+
+        self.assertIsNone(
+            await self.matrix_to_zulip.get_message_content_from_event(RoomMemberEvent())
+        )
+        self.assertEqual(
+            await self.matrix_to_zulip.get_message_content_from_event(RoomMessageFormatted()),
+            ZULIP_MESSAGE_TEMPLATE.format(
+                username=self.user_name, uid=self.user_uid, message="this is a message"
+            ),
+        )
+
 
 class MatrixBridgeZulipToMatrixTests(TestCase):
-    valid_zulip_config = dict(stream="some stream", topic="some topic", email="some@email")
+    room = mock.MagicMock()
+    valid_zulip_config = dict(
+        stream="some stream",
+        topic="some topic",
+        email="some@email",
+        bridges={("some stream", "some topic"): room},
+    )
     valid_msg = dict(
         sender_email="John@Smith.smith",  # must not be equal to config:email
+        sender_id=42,
         type="stream",  # Can only mirror Zulip streams
         display_recipient=valid_zulip_config["stream"],
         subject=valid_zulip_config["topic"],
     )
 
-    def test_zulip_message_validity_success(self) -> None:
+    def setUp(self) -> None:
-        zulip_config = self.valid_zulip_config
+        self.zulip_to_matrix = mock.MagicMock()
-        msg = self.valid_msg
+        self.zulip_to_matrix.zulip_config = self.valid_zulip_config
-        # Ensure the test inputs are valid for success
+        self.zulip_to_matrix.get_matrix_room_for_zulip_message = (
-        assert msg["sender_email"] != zulip_config["email"]
+            lambda msg: ZulipToMatrix.get_matrix_room_for_zulip_message(self.zulip_to_matrix, msg)
+        )
 
-        self.assertTrue(check_zulip_message_validity(msg, zulip_config))
+    def test_get_matrix_room_for_zulip_message_success(self) -> None:
+        self.assertEqual(
+            self.zulip_to_matrix.get_matrix_room_for_zulip_message(self.valid_msg), self.room
+        )
 
-    def test_zulip_message_validity_failure(self) -> None:
+    def test_get_matrix_room_for_zulip_message_failure(self) -> None:
-        zulip_config = self.valid_zulip_config
+        self.assertIsNone(
-
+            self.zulip_to_matrix.get_matrix_room_for_zulip_message(
-        msg_wrong_stream = dict(self.valid_msg, display_recipient="foo")
+                dict(self.valid_msg, type="private")
-        self.assertFalse(check_zulip_message_validity(msg_wrong_stream, zulip_config))
+            )
-
+        )
-        msg_wrong_topic = dict(self.valid_msg, subject="foo")
+        self.assertIsNone(
-        self.assertFalse(check_zulip_message_validity(msg_wrong_topic, zulip_config))
+            self.zulip_to_matrix.get_matrix_room_for_zulip_message(
-
+                dict(self.valid_msg, sender_email="some@email")
-        msg_not_stream = dict(self.valid_msg, type="private")
+            )
-        self.assertFalse(check_zulip_message_validity(msg_not_stream, zulip_config))
+        )
-
+        self.assertIsNone(
-        msg_from_bot = dict(self.valid_msg, sender_email=zulip_config["email"])
+            self.zulip_to_matrix.get_matrix_room_for_zulip_message(
-        self.assertFalse(check_zulip_message_validity(msg_from_bot, zulip_config))
+                dict(self.valid_msg, display_recipient="other stream")
-
+            )
-    def test_zulip_to_matrix(self) -> None:
+        )
-        room = mock.MagicMock()
+        self.assertIsNone(
-        zulip_config = self.valid_zulip_config
+            self.zulip_to_matrix.get_matrix_room_for_zulip_message(
-        send_msg = zulip_to_matrix(zulip_config, room)
+                dict(self.valid_msg, subject="other topic")
-
+            )
-        msg = dict(self.valid_msg, sender_full_name="John Smith")
+        )
-
-        expected = {
-            "hi": "{} hi",
-            "*hi*": "{} *hi*",
-            "**hi**": "{} **hi**",
-        }
-
-        for content in expected:
-            send_msg(dict(msg, content=content))
-
-        for (method, params, _), expect in zip(room.method_calls, expected.values()):
-            self.assertEqual(method, "send_text")
-            self.assertEqual(params[0], expect.format("<JohnSmith>"))

zulip/integrations/bridge_with_matrix/todo.md

@@ -0,0 +1,6 @@
+- Replace `asyncio.get_event_loop()` by `asyncio.get_running_loop()` as soon
+  as support for Python 3.6 is not necessary any more.
+  Reason: `get_event_loop` is depcrecated since Python 3.10.
+  See: https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_event_loop
+- Use `asyncio.run()` to run the `run()`-method as soon as support for Python
+  3.6 has been dropped.

zulip/setup.py

@@ -65,7 +65,6 @@ setup(
     },
     install_requires=[
         "requests[security]>=0.12.1",
-        "matrix_client",
         "distro",
         "click",
         "typing_extensions>=3.7",