Examples

If you’ve built something with matrix-nio and want to support the project, add a shield!

[![Built with matrix-nio](https://img.shields.io/badge/built%20with-matrix--nio-brightgreen)](https://github.com/poljar/matrix-nio)

To start making a chat bot quickly, considering using nio-template.

Attention

For E2EE support, python-olm is needed, which requires the libolm C library (version 3.x). After libolm has been installed, the e2ee enabled version of nio can be installed using pip install "matrix-nio[e2e]".

Projects built with nio

Are we missing a project? Submit a pull request and we’ll get you added! Just edit doc/built-with-nio.rst

A basic client

A basic client requires a few things before you start:

  • nio is installed

  • a Matrix homeserver URL (probably “https://matrix.example.org”)

  • a username and password for an account on that homeserver

  • a room ID for a room on that homeserver. In Riot, this is found in the Room’s settings page under “Advanced”

By far the easiest way to use nio is using the asyncio layer, unless you have special restrictions that disallow the use of asyncio.

All examples require Python 3.5+ for the async / await syntax.

 1import asyncio
 2
 3from nio import AsyncClient, MatrixRoom, RoomMessageText
 4
 5
 6async def message_callback(room: MatrixRoom, event: RoomMessageText) -> None:
 7    print(
 8        f"Message received in room {room.display_name}\n"
 9        f"{room.user_name(event.sender)} | {event.body}"
10    )
11
12
13async def main() -> None:
14    client = AsyncClient("https://matrix.example.org", "@alice:example.org")
15    client.add_event_callback(message_callback, RoomMessageText)
16
17    print(await client.login("my-secret-password"))
18    # "Logged in as @alice:example.org device id: RANDOMDID"
19
20    # If you made a new room and haven't joined as that user, you can use
21    # await client.join("your-room-id")
22
23    await client.room_send(
24        # Watch out! If you join an old room you'll see lots of old messages
25        room_id="!my-fave-room:example.org",
26        message_type="m.room.message",
27        content={"msgtype": "m.text", "body": "Hello world!"},
28    )
29    await client.sync_forever(timeout=30000)  # milliseconds
30
31
32asyncio.run(main())

Log in using a stored access_token

Using access tokens requires that when you first log in you save a few values to use later. In this example, we’re going to write them to disk as a JSON object, but you could also store them in a database, print them out and post them up on the wall beside your desk, text them to your sister in law, or anything else that allows you access to the values at a later date.

We’ve tried to keep this example small enough that it’s just enough to work; once you start writing your own programs with nio you may want to clean things up a bit.

This example requires that the user running it has write permissions to the folder they’re in. If you copied this repo to your computer, you probably have write permissions. Now run the program restore_login.py twice. First time around it will ask you for credentials like homeserver and password. On the second run, the program will log in for you automatically and it will send a “Hello World” message to the room you specify.

  1#!/usr/bin/env python3
  2
  3import asyncio
  4import getpass
  5import json
  6import os
  7import sys
  8
  9from nio import AsyncClient, LoginResponse
 10
 11CONFIG_FILE = "credentials.json"
 12
 13# Check out main() below to see how it's done.
 14
 15
 16def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
 17    """Writes the required login details to disk so we can log in later without
 18    using a password.
 19
 20    Arguments:
 21        resp {LoginResponse} -- the successful client login response.
 22        homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
 23    """
 24    # open the config file in write-mode
 25    with open(CONFIG_FILE, "w") as f:
 26        # write the login details to disk
 27        json.dump(
 28            {
 29                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
 30                "user_id": resp.user_id,  # e.g. "@user:example.org"
 31                "device_id": resp.device_id,  # device ID, 10 uppercase letters
 32                "access_token": resp.access_token,  # cryptogr. access token
 33            },
 34            f,
 35        )
 36
 37
 38async def main() -> None:
 39    # If there are no previously-saved credentials, we'll use the password
 40    if not os.path.exists(CONFIG_FILE):
 41        print(
 42            "First time use. Did not find credential file. Asking for "
 43            "homeserver, user, and password to create credential file."
 44        )
 45        homeserver = "https://matrix.example.org"
 46        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
 47
 48        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
 49            homeserver = "https://" + homeserver
 50
 51        user_id = "@user:example.org"
 52        user_id = input(f"Enter your full user ID: [{user_id}] ")
 53
 54        device_name = "matrix-nio"
 55        device_name = input(f"Choose a name for this device: [{device_name}] ")
 56
 57        client = AsyncClient(homeserver, user_id)
 58        pw = getpass.getpass()
 59
 60        resp = await client.login(pw, device_name=device_name)
 61
 62        # check that we logged in successfully
 63        if isinstance(resp, LoginResponse):
 64            write_details_to_disk(resp, homeserver)
 65        else:
 66            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
 67            print(f"Failed to log in: {resp}")
 68            sys.exit(1)
 69
 70        print(
 71            "Logged in using a password. Credentials were stored.",
 72            "Try running the script again to login with credentials.",
 73        )
 74
 75    # Otherwise the config file exists, so we'll use the stored credentials
 76    else:
 77        # open the file in read-only mode
 78        with open(CONFIG_FILE, "r") as f:
 79            config = json.load(f)
 80            client = AsyncClient(config["homeserver"])
 81
 82            client.access_token = config["access_token"]
 83            client.user_id = config["user_id"]
 84            client.device_id = config["device_id"]
 85
 86        # Now we can send messages as the user
 87        room_id = "!myfavouriteroomid:example.org"
 88        room_id = input(f"Enter room id for test message: [{room_id}] ")
 89
 90        await client.room_send(
 91            room_id,
 92            message_type="m.room.message",
 93            content={"msgtype": "m.text", "body": "Hello world!"},
 94        )
 95        print("Logged in using stored credentials. Sent a test message.")
 96
 97    # Either way we're logged in here, too
 98    await client.close()
 99
100
101asyncio.run(main())

Sending an image

Now that you have sent a first “Hello World” text message, how about going one step further and sending an image, like a photo from your last vacation. Run the send_image.py program and provide a filename to the photo. Voila, you have just sent your first image!

  1#!/usr/bin/env python3
  2
  3import asyncio
  4import getpass
  5import json
  6import os
  7import sys
  8
  9import aiofiles.os
 10import magic
 11from PIL import Image
 12
 13from nio import AsyncClient, LoginResponse, UploadResponse
 14
 15CONFIG_FILE = "credentials.json"
 16
 17# Check out main() below to see how it's done.
 18
 19
 20def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
 21    """Writes the required login details to disk so we can log in later without
 22    using a password.
 23
 24    Arguments:
 25        resp {LoginResponse} -- the successful client login response.
 26        homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
 27    """
 28    # open the config file in write-mode
 29    with open(CONFIG_FILE, "w") as f:
 30        # write the login details to disk
 31        json.dump(
 32            {
 33                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
 34                "user_id": resp.user_id,  # e.g. "@user:example.org"
 35                "device_id": resp.device_id,  # device ID, 10 uppercase letters
 36                "access_token": resp.access_token,  # cryptogr. access token
 37            },
 38            f,
 39        )
 40
 41
 42async def send_image(client, room_id, image):
 43    """Send image to room.
 44
 45    Arguments:
 46    ---------
 47    client : Client
 48    room_id : str
 49    image : str, file name of image
 50
 51    This is a working example for a JPG image.
 52        "content": {
 53            "body": "someimage.jpg",
 54            "info": {
 55                "size": 5420,
 56                "mimetype": "image/jpeg",
 57                "thumbnail_info": {
 58                    "w": 100,
 59                    "h": 100,
 60                    "mimetype": "image/jpeg",
 61                    "size": 2106
 62                },
 63                "w": 100,
 64                "h": 100,
 65                "thumbnail_url": "mxc://example.com/SomeStrangeThumbnailUriKey"
 66            },
 67            "msgtype": "m.image",
 68            "url": "mxc://example.com/SomeStrangeUriKey"
 69        }
 70
 71    """
 72    mime_type = magic.from_file(image, mime=True)  # e.g. "image/jpeg"
 73    if not mime_type.startswith("image/"):
 74        print("Drop message because file does not have an image mime type.")
 75        return
 76
 77    im = Image.open(image)
 78    (width, height) = im.size  # im.size returns (width,height) tuple
 79
 80    # first do an upload of image, then send URI of upload to room
 81    file_stat = await aiofiles.os.stat(image)
 82    async with aiofiles.open(image, "r+b") as f:
 83        resp, maybe_keys = await client.upload(
 84            f,
 85            content_type=mime_type,  # image/jpeg
 86            filename=os.path.basename(image),
 87            filesize=file_stat.st_size,
 88        )
 89    if isinstance(resp, UploadResponse):
 90        print("Image was uploaded successfully to server. ")
 91    else:
 92        print(f"Failed to upload image. Failure response: {resp}")
 93
 94    content = {
 95        "body": os.path.basename(image),  # descriptive title
 96        "info": {
 97            "size": file_stat.st_size,
 98            "mimetype": mime_type,
 99            "thumbnail_info": None,  # TODO
100            "w": width,  # width in pixel
101            "h": height,  # height in pixel
102            "thumbnail_url": None,  # TODO
103        },
104        "msgtype": "m.image",
105        "url": resp.content_uri,
106    }
107
108    try:
109        await client.room_send(room_id, message_type="m.room.message", content=content)
110        print("Image was sent successfully")
111    except Exception:
112        print(f"Image send of file {image} failed.")
113
114
115async def main() -> None:
116    # If there are no previously-saved credentials, we'll use the password
117    if not os.path.exists(CONFIG_FILE):
118        print(
119            "First time use. Did not find credential file. Asking for "
120            "homeserver, user, and password to create credential file."
121        )
122        homeserver = "https://matrix.example.org"
123        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
124
125        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
126            homeserver = "https://" + homeserver
127
128        user_id = "@user:example.org"
129        user_id = input(f"Enter your full user ID: [{user_id}] ")
130
131        device_name = "matrix-nio"
132        device_name = input(f"Choose a name for this device: [{device_name}] ")
133
134        client = AsyncClient(homeserver, user_id)
135        pw = getpass.getpass()
136
137        resp = await client.login(pw, device_name=device_name)
138
139        # check that we logged in successfully
140        if isinstance(resp, LoginResponse):
141            write_details_to_disk(resp, homeserver)
142        else:
143            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
144            print(f"Failed to log in: {resp}")
145            sys.exit(1)
146
147        print(
148            "Logged in using a password. Credentials were stored.",
149            "Try running the script again to login with credentials.",
150        )
151
152    # Otherwise the config file exists, so we'll use the stored credentials
153    else:
154        # open the file in read-only mode
155        with open(CONFIG_FILE, "r") as f:
156            config = json.load(f)
157            client = AsyncClient(config["homeserver"])
158
159            client.access_token = config["access_token"]
160            client.user_id = config["user_id"]
161            client.device_id = config["device_id"]
162
163        # Now we can send messages as the user
164        room_id = "!myfavouriteroomid:example.org"
165        room_id = input(f"Enter room id for image message: [{room_id}] ")
166
167        image = "exampledir/samplephoto.jpg"
168        image = input(f"Enter file name of image to send: [{image}] ")
169
170        await send_image(client, room_id, image)
171        print("Logged in using stored credentials. Sent a test message.")
172
173    # Close the client connection after we are done with it.
174    await client.close()
175
176
177asyncio.run(main())

Manual encryption key verification

Below is a program that works through manual encryption of other users when you already know all of their device IDs. It’s a bit dense but provides a good example in terms of being pythonic and using nio’s design features purposefully. It is not designed to be a template that you can immediately extend to run your bot, it’s designed to be an example of how to use nio.

The overall structure is this: we subclass nio’s AsyncClient class and add in our own handlers for a few things, namely:

  • automatically restoring login details from disk instead of creating new sessions each time we restart the process

  • callback for printing out any message we receive to stdout

  • callback for automatically joining any room @alice is invited to

  • a method for trusting devices using a user ID and (optionally) their list of trusted device IDs

  • a sample “hello world” encrypted message method

In main, we make an instance of that subclass, attempt to login, then create an asyncio coroutine to run later that will trust the devices and send the hello world message. We then create `asyncio Tasks <>`_ to run that coroutine as well as the sync_forever() coroutine that nio provides, which does most of the handling of required work for communicating with Matrix: it uploads keys, checks for new messages, executes callbacks when events occur that trigger those callbacks, etc. Main executes the result of those Tasks.

You’ll need two accounts, which we’ll call @alice:example.org and @bob:example.org. @alice will be your nio application and @bob will be your second user account. Before the script runs, make a new room with the @bob account, enable encryption and invite @alice. Note the room ID as you’ll need it for this script. You’ll also need all of @bob’s device IDs, which you can get from within Riot under the profile settings > Advanced section. They may be called “session IDs”. These are the device IDs that your program will trust, and getting them into nio is the manual part here. In another example we’ll document automatic emoji verification.

It may look long at first but much of the program is actually documentation explaining how it works. If you have questions about the example, please don’t hesitate to ask them on #nio:matrix.org.

If you are stuck, it may be useful to read this primer from Matrix.org on implementing end-to-end encryption: https://matrix.org/docs/guides/end-to-end-encryption-implementation-guide

To delete the store, or clear the trusted devices, simply remove “nio_store” in the working directory as well as “manual_encrypted_verify.json”. Then the example script will log in (with a new session ID) and generate new keys.

  1import asyncio
  2import json
  3import os
  4import sys
  5from typing import Optional
  6
  7from nio import (
  8    AsyncClient,
  9    ClientConfig,
 10    DevicesError,
 11    Event,
 12    InviteEvent,
 13    LocalProtocolError,
 14    LoginResponse,
 15    MatrixRoom,
 16    MatrixUser,
 17    RoomMessageText,
 18    RoomSendResponse,
 19    crypto,
 20    exceptions,
 21)
 22
 23# This is a fully-documented example of how to do manual verification with nio,
 24# for when you already know the device IDs of the users you want to trust. If
 25# you want live verification using emojis, the process is more complicated and
 26# will be covered in another example.
 27
 28# We're building on the restore_login example here to preserve device IDs and
 29# therefore preserve trust; if @bob trusts @alice's device ID ABC and @alice
 30# restarts this program, loading the same keys, @bob will preserve trust. If
 31# @alice logged in again @alice would have new keys and a device ID XYZ, and
 32# @bob wouldn't trust it.
 33
 34# The store is where we want to place encryption details like our keys, trusted
 35# devices and blacklisted devices. Here we place it in the working directory,
 36# but if you deploy your program you might consider /var or /opt for storage
 37STORE_FOLDER = "nio_store/"
 38
 39# This file is for restoring login details after closing the program, so you
 40# can preserve your device ID. If @alice logged in every time instead, @bob
 41# would have to re-verify. See the restoring login example for more into.
 42SESSION_DETAILS_FILE = "credentials.json"
 43
 44# Only needed for this example, this is who @alice will securely
 45# communicate with. We need all the device IDs of this user so we can consider
 46# them "trusted". If an unknown device shows up (like @bob signs into their
 47# account on another device), this program will refuse to send a message in the
 48# room. Try it!
 49BOB_ID = "@bob:example.org"
 50BOB_DEVICE_IDS = [
 51    # You can find these in Riot under Settings > Security & Privacy.
 52    # They may also be called "session IDs". You'll want to add ALL of them here
 53    # for the one other user in your encrypted room
 54    "URDEVICEID",
 55]
 56
 57# the ID of the room you want your bot to join and send commands in.
 58# This can be a direct message or room; Matrix treats them the same
 59ROOM_ID = "!myfavouriteroom:example.org"
 60
 61ALICE_USER_ID = "@alice:example.org"
 62ALICE_HOMESERVER = "https://matrix.example.org"
 63ALICE_PASSWORD = "hunter2"
 64
 65
 66class CustomEncryptedClient(AsyncClient):
 67    def __init__(
 68        self,
 69        homeserver,
 70        user="",
 71        device_id="",
 72        store_path="",
 73        config=None,
 74        ssl=None,
 75        proxy=None,
 76    ):
 77        # Calling super.__init__ means we're running the __init__ method
 78        # defined in AsyncClient, which this class derives from. That does a
 79        # bunch of setup for us automatically
 80        super().__init__(
 81            homeserver,
 82            user=user,
 83            device_id=device_id,
 84            store_path=store_path,
 85            config=config,
 86            ssl=ssl,
 87            proxy=proxy,
 88        )
 89
 90        # if the store location doesn't exist, we'll make it
 91        if store_path and not os.path.isdir(store_path):
 92            os.mkdir(store_path)
 93
 94        # auto-join room invites
 95        self.add_event_callback(self.cb_autojoin_room, InviteEvent)
 96
 97        # print all the messages we receive
 98        self.add_event_callback(self.cb_print_messages, RoomMessageText)
 99
100    async def login(self) -> None:
101        """Log in either using the global variables or (if possible) using the
102        session details file.
103
104        NOTE: This method kinda sucks. Don't use these kinds of global
105        variables in your program; it would be much better to pass them
106        around instead. They are only used here to minimise the size of the
107        example.
108        """
109        # Restore the previous session if we can
110        # See the "restore_login.py" example if you're not sure how this works
111        if os.path.exists(SESSION_DETAILS_FILE) and os.path.isfile(
112            SESSION_DETAILS_FILE
113        ):
114            try:
115                with open(SESSION_DETAILS_FILE, "r") as f:
116                    config = json.load(f)
117                    self.access_token = config["access_token"]
118                    self.user_id = config["user_id"]
119                    self.device_id = config["device_id"]
120
121                    # This loads our verified/blacklisted devices and our keys
122                    self.load_store()
123                    print(
124                        f"Logged in using stored credentials: {self.user_id} on {self.device_id}"
125                    )
126
127            except IOError as err:
128                print(f"Couldn't load session from file. Logging in. Error: {err}")
129            except json.JSONDecodeError:
130                print("Couldn't read JSON file; overwriting")
131
132        # We didn't restore a previous session, so we'll log in with a password
133        if not self.user_id or not self.access_token or not self.device_id:
134            # this calls the login method defined in AsyncClient from nio
135            resp = await super().login(ALICE_PASSWORD)
136
137            if isinstance(resp, LoginResponse):
138                print("Logged in using a password; saving details to disk")
139                self.__write_details_to_disk(resp)
140            else:
141                print(f"Failed to log in: {resp}")
142                sys.exit(1)
143
144    def trust_devices(self, user_id: str, device_list: Optional[str] = None) -> None:
145        """Trusts the devices of a user.
146
147        If no device_list is provided, all of the users devices are trusted. If
148        one is provided, only the devices with IDs in that list are trusted.
149
150        Arguments:
151            user_id {str} -- the user ID whose devices should be trusted.
152
153        Keyword Arguments:
154            device_list {Optional[str]} -- The full list of device IDs to trust
155                from that user (default: {None})
156        """
157
158        print(f"{user_id}'s device store: {self.device_store[user_id]}")
159
160        # The device store contains a dictionary of device IDs and known
161        # OlmDevices for all users that share a room with us, including us.
162
163        # We can only run this after a first sync. We have to populate our
164        # device store and that requires syncing with the server.
165        for device_id, olm_device in self.device_store[user_id].items():
166            if device_list and device_id not in device_list:
167                # a list of trusted devices was provided, but this ID is not in
168                # that list. That's an issue.
169                print(
170                    f"Not trusting {device_id} as it's not in {user_id}'s pre-approved list."
171                )
172                continue
173
174            if user_id == self.user_id and device_id == self.device_id:
175                # We cannot explicitly trust the device @alice is using
176                continue
177
178            self.verify_device(olm_device)
179            print(f"Trusting {device_id} from user {user_id}")
180
181    def cb_autojoin_room(self, room: MatrixRoom, event: InviteEvent):
182        """Callback to automatically joins a Matrix room on invite.
183
184        Arguments:
185            room {MatrixRoom} -- Provided by nio
186            event {InviteEvent} -- Provided by nio
187        """
188        self.join(room.room_id)
189        room = self.rooms[ROOM_ID]
190        print(f"Room {room.name} is encrypted: {room.encrypted}")
191
192    async def cb_print_messages(self, room: MatrixRoom, event: RoomMessageText):
193        """Callback to print all received messages to stdout.
194
195        Arguments:
196            room {MatrixRoom} -- Provided by nio
197            event {RoomMessageText} -- Provided by nio
198        """
199        if event.decrypted:
200            encrypted_symbol = "🛡 "
201        else:
202            encrypted_symbol = "⚠️ "
203        print(
204            f"{room.display_name} |{encrypted_symbol}| {room.user_name(event.sender)}: {event.body}"
205        )
206
207    async def send_hello_world(self):
208        # Now we send an encrypted message that @bob can read, although it will
209        # appear to be "unverified" when they see it, because @bob has not verified
210        # the device @alice is sending from.
211        # We'll leave that as an exercise for the reader.
212        try:
213            await self.room_send(
214                room_id=ROOM_ID,
215                message_type="m.room.message",
216                content={
217                    "msgtype": "m.text",
218                    "body": "Hello, this message is encrypted",
219                },
220            )
221        except exceptions.OlmUnverifiedDeviceError as err:
222            print("These are all known devices:")
223            device_store: crypto.DeviceStore = device_store
224            [
225                print(
226                    f"\t{device.user_id}\t {device.device_id}\t {device.trust_state}\t  {device.display_name}"
227                )
228                for device in device_store
229            ]
230            sys.exit(1)
231
232    @staticmethod
233    def __write_details_to_disk(resp: LoginResponse) -> None:
234        """Writes login details to disk so that we can restore our session later
235        without logging in again and creating a new device ID.
236
237        Arguments:
238            resp {LoginResponse} -- the successful client login response.
239        """
240        with open(SESSION_DETAILS_FILE, "w") as f:
241            json.dump(
242                {
243                    "access_token": resp.access_token,
244                    "device_id": resp.device_id,
245                    "user_id": resp.user_id,
246                },
247                f,
248            )
249
250
251async def run_client(client: CustomEncryptedClient) -> None:
252    """A basic encrypted chat application using nio."""
253
254    # This is our own custom login function that looks for a pre-existing config
255    # file and, if it exists, logs in using those details. Otherwise it will log
256    # in using a password.
257    await client.login()
258
259    # Here we create a coroutine that we can call in asyncio.gather later,
260    # along with sync_forever and any other API-related coroutines you'd like
261    # to do.
262    async def after_first_sync():
263        # We'll wait for the first firing of 'synced' before trusting devices.
264        # client.synced is an asyncio event that fires any time nio syncs. This
265        # code doesn't run in a loop, so it only fires once
266        print("Awaiting sync")
267        await client.synced.wait()
268
269        # In practice, you want to have a list of previously-known device IDs
270        # for each user you want to trust. Here, we require that list as a
271        # global variable
272        client.trust_devices(BOB_ID, BOB_DEVICE_IDS)
273
274        # In this case, we'll trust _all_ of @alice's devices. NOTE that this
275        # is a SUPER BAD IDEA in practice, but for the purpose of this example
276        # it'll be easier, since you may end up creating lots of sessions for
277        # @alice as you play with the script
278        client.trust_devices(ALICE_USER_ID)
279
280        await client.send_hello_world()
281
282    # We're creating Tasks here so that you could potentially write other
283    # Python coroutines to do other work, like checking an API or using another
284    # library. All of these Tasks will be run concurrently.
285    # For more details, check out https://docs.python.org/3/library/asyncio-task.html
286
287    # ensure_future() is for Python 3.5 and 3.6 compatibility. For 3.7+, use
288    # asyncio.create_task()
289    after_first_sync_task = asyncio.ensure_future(after_first_sync())
290
291    # We use full_state=True here to pull any room invites that occurred or
292    # messages sent in rooms _before_ this program connected to the
293    # Matrix server
294    sync_forever_task = asyncio.ensure_future(
295        client.sync_forever(30000, full_state=True)
296    )
297
298    await asyncio.gather(
299        # The order here IS significant! You have to register the task to trust
300        # devices FIRST since it awaits the first sync
301        after_first_sync_task,
302        sync_forever_task,
303    )
304
305
306async def main():
307    # By setting `store_sync_tokens` to true, we'll save sync tokens to our
308    # store every time we sync, thereby preventing reading old, previously read
309    # events on each new sync.
310    # For more info, check out https://matrix-nio.readthedocs.io/en/latest/nio.html#asyncclient
311    config = ClientConfig(store_sync_tokens=True)
312    client = CustomEncryptedClient(
313        ALICE_HOMESERVER,
314        ALICE_USER_ID,
315        store_path=STORE_FOLDER,
316        config=config,
317        ssl=False,
318        proxy="http://localhost:8080",
319    )
320
321    try:
322        await run_client(client)
323    except (asyncio.CancelledError, KeyboardInterrupt):
324        await client.close()
325
326
327# Run the main coroutine, which instantiates our custom subclass, trusts all the
328# devices, and syncs forever (or until your press Ctrl+C)
329
330if __name__ == "__main__":
331    try:
332        asyncio.run(main())
333    except KeyboardInterrupt:
334        pass

Interactive encryption key verification

One way to interactively verify a device is via emojis. On popular Matrix clients you will find that devices are flagged as trusted or untrusted. If a device is untrusted you can verify to make it trusted. Most clients have a red symbol for untrusted and a green icon for trusted. One can select un untrusted device and initiate a verify by emoji action. How would that look like in code? How can you add that to your application? Next we present a simple application that showcases emoji verification. Note, the app only accepts emoji verification. So, you have to start it on the other client (e.g. Element). Initiating an emoji verification is similar in code, consider doing it as “homework” if you feel up to it. But for now, let’s have a look how emoji verification can be accepted and processed.

  1#!/usr/bin/env python3
  2
  3"""verify_with_emoji.py A sample program to demo Emoji verification.
  4
  5# Objectives:
  6- Showcase the emoji verification using matrix-nio SDK
  7- This sample program tries to show the key steps involved in performing
  8    an emoji verification.
  9- It does so only for incoming request, outgoing emoji verification request
 10    are similar but not shown in this sample program
 11
 12# Prerequisites:
 13- You must have matrix-nio and components for end-to-end encryption installed
 14    See: https://github.com/poljar/matrix-nio
 15- You must have created a Matrix account already,
 16    and have username and password ready
 17- You must have already joined a Matrix room with someone, e.g. yourself
 18- This other party initiates an emoji verification with you
 19- You are using this sample program to accept this incoming emoji verification
 20    and follow the protocol to successfully verify the other party's device
 21
 22# Use Cases:
 23- Apply similar code in your Matrix bot
 24- Apply similar code in your Matrix client
 25- Just to learn about Matrix and the matrix-nio SDK
 26
 27# Running the Program:
 28- Change permissions to allow execution
 29    `chmod 755 ./verify_with_emoji.py`
 30- Optionally create a store directory, if not it will be done for you
 31    `mkdir ./store/`
 32- Run the program as-is, no changes needed
 33    `./verify_with_emoji.py`
 34- Run it as often as you like
 35
 36# Sample Screen Output when Running Program:
 37$ ./verify_with_emoji.py
 38First time use. Did not find credential file. Asking for
 39homeserver, user, and password to create credential file.
 40Enter your homeserver URL: [https://matrix.example.org] matrix.example.org
 41Enter your full user ID: [@user:example.org] @user:example.org
 42Choose a name for this device: [matrix-nio] verify_with_emoji
 43Password:
 44Logged in using a password. Credentials were stored.
 45On next execution the stored login credentials will be used.
 46This program is ready and waiting for the other party to initiate an emoji
 47verification with us by selecting "Verify by Emoji" in their Matrix client.
 48[('⚓', 'Anchor'), ('☎️', 'Telephone'), ('😀', 'Smiley'), ('😀', 'Smiley'),
 49 ('☂️', 'Umbrella'), ('⚓', 'Anchor'), ('☎️', 'Telephone')]
 50Do the emojis match? (Y/N) y
 51Match! Device will be verified by accepting verification.
 52sas.we_started_it = False
 53sas.sas_accepted = True
 54sas.canceled = False
 55sas.timed_out = False
 56sas.verified = True
 57sas.verified_devices = ['DEVICEIDXY']
 58Emoji verification was successful.
 59Hit Control-C to stop the program or initiate another Emoji verification
 60from another device or room.
 61
 62"""
 63
 64import asyncio
 65import getpass
 66import json
 67import os
 68import sys
 69import traceback
 70
 71from nio import (
 72    AsyncClient,
 73    AsyncClientConfig,
 74    KeyVerificationCancel,
 75    KeyVerificationEvent,
 76    KeyVerificationKey,
 77    KeyVerificationMac,
 78    KeyVerificationStart,
 79    LocalProtocolError,
 80    LoginResponse,
 81    ToDeviceError,
 82)
 83
 84# file to store credentials in case you want to run program multiple times
 85CONFIG_FILE = "credentials.json"  # login credentials JSON file
 86# directory to store persistent data for end-to-end encryption
 87STORE_PATH = "./store/"  # local directory
 88
 89
 90class Callbacks(object):
 91    """Class to pass client to callback methods."""
 92
 93    def __init__(self, client):
 94        """Store AsyncClient."""
 95        self.client = client
 96
 97    async def to_device_callback(self, event):  # noqa
 98        """Handle events sent to device."""
 99        try:
100            client = self.client
101
102            if isinstance(event, KeyVerificationStart):  # first step
103                """first step: receive KeyVerificationStart
104                KeyVerificationStart(
105                    source={'content':
106                            {'method': 'm.sas.v1',
107                             'from_device': 'DEVICEIDXY',
108                             'key_agreement_protocols':
109                                ['curve25519-hkdf-sha256', 'curve25519'],
110                             'hashes': ['sha256'],
111                             'message_authentication_codes':
112                                ['hkdf-hmac-sha256', 'hmac-sha256'],
113                             'short_authentication_string':
114                                ['decimal', 'emoji'],
115                             'transaction_id': 'SomeTxId'
116                             },
117                            'type': 'm.key.verification.start',
118                            'sender': '@user2:example.org'
119                            },
120                    sender='@user2:example.org',
121                    transaction_id='SomeTxId',
122                    from_device='DEVICEIDXY',
123                    method='m.sas.v1',
124                    key_agreement_protocols=[
125                        'curve25519-hkdf-sha256', 'curve25519'],
126                    hashes=['sha256'],
127                    message_authentication_codes=[
128                        'hkdf-hmac-sha256', 'hmac-sha256'],
129                    short_authentication_string=['decimal', 'emoji'])
130                """
131
132                if "emoji" not in event.short_authentication_string:
133                    print(
134                        "Other device does not support emoji verification "
135                        f"{event.short_authentication_string}."
136                    )
137                    return
138                resp = await client.accept_key_verification(event.transaction_id)
139                if isinstance(resp, ToDeviceError):
140                    print(f"accept_key_verification failed with {resp}")
141
142                sas = client.key_verifications[event.transaction_id]
143
144                todevice_msg = sas.share_key()
145                resp = await client.to_device(todevice_msg)
146                if isinstance(resp, ToDeviceError):
147                    print(f"to_device failed with {resp}")
148
149            elif isinstance(event, KeyVerificationCancel):  # anytime
150                """at any time: receive KeyVerificationCancel
151                KeyVerificationCancel(source={
152                    'content': {'code': 'm.mismatched_sas',
153                                'reason': 'Mismatched authentication string',
154                                'transaction_id': 'SomeTxId'},
155                    'type': 'm.key.verification.cancel',
156                    'sender': '@user2:example.org'},
157                    sender='@user2:example.org',
158                    transaction_id='SomeTxId',
159                    code='m.mismatched_sas',
160                    reason='Mismatched short authentication string')
161                """
162
163                # There is no need to issue a
164                # client.cancel_key_verification(tx_id, reject=False)
165                # here. The SAS flow is already cancelled.
166                # We only need to inform the user.
167                print(
168                    f"Verification has been cancelled by {event.sender} "
169                    f'for reason "{event.reason}".'
170                )
171
172            elif isinstance(event, KeyVerificationKey):  # second step
173                """Second step is to receive KeyVerificationKey
174                KeyVerificationKey(
175                    source={'content': {
176                            'key': 'SomeCryptoKey',
177                            'transaction_id': 'SomeTxId'},
178                        'type': 'm.key.verification.key',
179                        'sender': '@user2:example.org'
180                    },
181                    sender='@user2:example.org',
182                    transaction_id='SomeTxId',
183                    key='SomeCryptoKey')
184                """
185                sas = client.key_verifications[event.transaction_id]
186
187                print(f"{sas.get_emoji()}")
188
189                yn = input("Do the emojis match? (Y/N) (C for Cancel) ")
190                if yn.lower() == "y":
191                    print(
192                        "Match! The verification for this " "device will be accepted."
193                    )
194                    resp = await client.confirm_short_auth_string(event.transaction_id)
195                    if isinstance(resp, ToDeviceError):
196                        print(f"confirm_short_auth_string failed with {resp}")
197                elif yn.lower() == "n":  # no, don't match, reject
198                    print(
199                        "No match! Device will NOT be verified "
200                        "by rejecting verification."
201                    )
202                    resp = await client.cancel_key_verification(
203                        event.transaction_id, reject=True
204                    )
205                    if isinstance(resp, ToDeviceError):
206                        print(f"cancel_key_verification failed with {resp}")
207                else:  # C or anything for cancel
208                    print("Cancelled by user! Verification will be " "cancelled.")
209                    resp = await client.cancel_key_verification(
210                        event.transaction_id, reject=False
211                    )
212                    if isinstance(resp, ToDeviceError):
213                        print(f"cancel_key_verification failed with {resp}")
214
215            elif isinstance(event, KeyVerificationMac):  # third step
216                """Third step is to receive KeyVerificationMac
217                KeyVerificationMac(
218                    source={'content': {
219                        'mac': {'ed25519:DEVICEIDXY': 'SomeKey1',
220                                'ed25519:SomeKey2': 'SomeKey3'},
221                        'keys': 'SomeCryptoKey4',
222                        'transaction_id': 'SomeTxId'},
223                        'type': 'm.key.verification.mac',
224                        'sender': '@user2:example.org'},
225                    sender='@user2:example.org',
226                    transaction_id='SomeTxId',
227                    mac={'ed25519:DEVICEIDXY': 'SomeKey1',
228                         'ed25519:SomeKey2': 'SomeKey3'},
229                    keys='SomeCryptoKey4')
230                """
231                sas = client.key_verifications[event.transaction_id]
232                try:
233                    todevice_msg = sas.get_mac()
234                except LocalProtocolError as e:
235                    # e.g. it might have been cancelled by ourselves
236                    print(
237                        f"Cancelled or protocol error: Reason: {e}.\n"
238                        f"Verification with {event.sender} not concluded. "
239                        "Try again?"
240                    )
241                else:
242                    resp = await client.to_device(todevice_msg)
243                    if isinstance(resp, ToDeviceError):
244                        print(f"to_device failed with {resp}")
245                    print(
246                        f"sas.we_started_it = {sas.we_started_it}\n"
247                        f"sas.sas_accepted = {sas.sas_accepted}\n"
248                        f"sas.canceled = {sas.canceled}\n"
249                        f"sas.timed_out = {sas.timed_out}\n"
250                        f"sas.verified = {sas.verified}\n"
251                        f"sas.verified_devices = {sas.verified_devices}\n"
252                    )
253                    print(
254                        "Emoji verification was successful!\n"
255                        "Hit Control-C to stop the program or "
256                        "initiate another Emoji verification from "
257                        "another device or room."
258                    )
259            else:
260                print(
261                    f"Received unexpected event type {type(event)}. "
262                    f"Event is {event}. Event will be ignored."
263                )
264        except BaseException:
265            print(traceback.format_exc())
266
267
268def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
269    """Write the required login details to disk.
270
271    It will allow following logins to be made without password.
272
273    Arguments:
274    ---------
275        resp : LoginResponse - successful client login response
276        homeserver : str - URL of homeserver, e.g. "https://matrix.example.org"
277
278    """
279    # open the config file in write-mode
280    with open(CONFIG_FILE, "w") as f:
281        # write the login details to disk
282        json.dump(
283            {
284                "homeserver": homeserver,  # e.g. "https://matrix.example.org"
285                "user_id": resp.user_id,  # e.g. "@user:example.org"
286                "device_id": resp.device_id,  # device ID, 10 uppercase letters
287                "access_token": resp.access_token,  # cryptogr. access token
288            },
289            f,
290        )
291
292
293async def login() -> AsyncClient:
294    """Handle login with or without stored credentials."""
295    # Configuration options for the AsyncClient
296    client_config = AsyncClientConfig(
297        max_limit_exceeded=0,
298        max_timeouts=0,
299        store_sync_tokens=True,
300        encryption_enabled=True,
301    )
302
303    # If there are no previously-saved credentials, we'll use the password
304    if not os.path.exists(CONFIG_FILE):
305        print(
306            "First time use. Did not find credential file. Asking for "
307            "homeserver, user, and password to create credential file."
308        )
309        homeserver = "https://matrix.example.org"
310        homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
311
312        if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
313            homeserver = "https://" + homeserver
314
315        user_id = "@user:example.org"
316        user_id = input(f"Enter your full user ID: [{user_id}] ")
317
318        device_name = "matrix-nio"
319        device_name = input(f"Choose a name for this device: [{device_name}] ")
320
321        if not os.path.exists(STORE_PATH):
322            os.makedirs(STORE_PATH)
323
324        # Initialize the matrix client
325        client = AsyncClient(
326            homeserver,
327            user_id,
328            store_path=STORE_PATH,
329            config=client_config,
330        )
331        pw = getpass.getpass()
332
333        resp = await client.login(password=pw, device_name=device_name)
334
335        # check that we logged in successfully
336        if isinstance(resp, LoginResponse):
337            write_details_to_disk(resp, homeserver)
338        else:
339            print(f'homeserver = "{homeserver}"; user = "{user_id}"')
340            print(f"Failed to log in: {resp}")
341            sys.exit(1)
342
343        print(
344            "Logged in using a password. Credentials were stored. "
345            "On next execution the stored login credentials will be used."
346        )
347
348    # Otherwise the config file exists, so we'll use the stored credentials
349    else:
350        # open the file in read-only mode
351        with open(CONFIG_FILE, "r") as f:
352            config = json.load(f)
353            # Initialize the matrix client based on credentials from file
354            client = AsyncClient(
355                config["homeserver"],
356                config["user_id"],
357                device_id=config["device_id"],
358                store_path=STORE_PATH,
359                config=client_config,
360            )
361
362            client.restore_login(
363                user_id=config["user_id"],
364                device_id=config["device_id"],
365                access_token=config["access_token"],
366            )
367        print("Logged in using stored credentials.")
368
369    return client
370
371
372async def main() -> None:
373    """Login and wait for and perform emoji verify."""
374    client = await login()
375    # Set up event callbacks
376    callbacks = Callbacks(client)
377    client.add_to_device_callback(callbacks.to_device_callback, (KeyVerificationEvent,))
378    # Sync encryption keys with the server
379    # Required for participating in encrypted rooms
380    if client.should_upload_keys:
381        await client.keys_upload()
382    print(
383        "This program is ready and waiting for the other party to initiate "
384        'an emoji verification with us by selecting "Verify by Emoji" '
385        "in their Matrix client."
386    )
387    await client.sync_forever(timeout=30000, full_state=True)
388
389
390try:
391    asyncio.run(main())
392except Exception:
393    print(traceback.format_exc())
394    sys.exit(1)
395except KeyboardInterrupt:
396    print("Received keyboard interrupt.")
397    sys.exit(0)

Further reading and exploration

In an external repo, not maintained by us, is a simple Matrix client that includes sending, receiving and verification. It gives an example of

  • how to send text, images, audio, video, other text files

  • listen to messages forever

  • get just the newest unread messages

  • get the last N messages

  • perform emoji verification

  • etc.

So, if you want more example code and want to explore further have a look at this external repo called matrix-commander. And of course, you should check out all the other projects built with matrix-nio. To do so, check out our built-with-marix-nio list.