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
pantalaimon - e2ee-aware reverse proxy daemon
nio-template - template for creating bots
matrix-tweetalong-bot - Twitter-backed watchalongs
weechat-matrix - lets Weechat communicate over Matrix
hemppa - generic modular Matrix bot
devops-bot - sysadmin Swiss army knife bot for team efficiency
podbot - play podcasts sent from AntennaPod via Riot.im elsewhere
delator - bot with logs and modular commands
cody - REPL for your matrix chat rooms
matrix-nio-send - predecessor of matrix-commander
Mirage - feature-rich keyboard-operable chat client written in Qt/QML
matrix-commander - CLI-oriented integratable client
matrix-archive - creates a YAML log of all room messages and media
matrix-eno-bot - admin and personal assistence bot
infinigpt-matrix - OpenAI GPT chatbot with infinite personalities
ollamarama-matrix - AI chatbot with infinite personalities, using local LLMs via Ollama
Matrix Nio - HACS - A HACS Integration to replace the core matrix Home Assistant integration
niobot - An extensive framework for building powerful Matrix bots with ease
matrix-reminder-bot - A bot to remind you about stuff
bartleby - A LLM chatbot and writing assistant using HuggingFace
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
9import aiofiles
10
11from nio import AsyncClient, LoginResponse
12
13CONFIG_FILE = "credentials.json"
14
15# Check out main() below to see how it's done.
16
17
18def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
19 """Writes the required login details to disk so we can log in later without
20 using a password.
21
22 Arguments:
23 resp {LoginResponse} -- the successful client login response.
24 homeserver -- URL of homeserver, e.g. "https://matrix.example.org"
25 """
26 # open the config file in write-mode
27 with open(CONFIG_FILE, "w") as f:
28 # write the login details to disk
29 json.dump(
30 {
31 "homeserver": homeserver, # e.g. "https://matrix.example.org"
32 "user_id": resp.user_id, # e.g. "@user:example.org"
33 "device_id": resp.device_id, # device ID, 10 uppercase letters
34 "access_token": resp.access_token, # cryptogr. access token
35 },
36 f,
37 )
38
39
40async def main() -> None:
41 # If there are no previously-saved credentials, we'll use the password
42 if not os.path.exists(CONFIG_FILE):
43 print(
44 "First time use. Did not find credential file. Asking for "
45 "homeserver, user, and password to create credential file."
46 )
47 homeserver = "https://matrix.example.org"
48 homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
49
50 if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
51 homeserver = "https://" + homeserver
52
53 user_id = "@user:example.org"
54 user_id = input(f"Enter your full user ID: [{user_id}] ")
55
56 device_name = "matrix-nio"
57 device_name = input(f"Choose a name for this device: [{device_name}] ")
58
59 client = AsyncClient(homeserver, user_id)
60 pw = getpass.getpass()
61
62 resp = await client.login(pw, device_name=device_name)
63
64 # check that we logged in successfully
65 if isinstance(resp, LoginResponse):
66 write_details_to_disk(resp, homeserver)
67 else:
68 print(f'homeserver = "{homeserver}"; user = "{user_id}"')
69 print(f"Failed to log in: {resp}")
70 sys.exit(1)
71
72 print(
73 "Logged in using a password. Credentials were stored.",
74 "Try running the script again to login with credentials.",
75 )
76
77 # Otherwise the config file exists, so we'll use the stored credentials
78 else:
79 # open the file in read-only mode
80 async with aiofiles.open(CONFIG_FILE) as f:
81 contents = await f.read()
82 config = json.loads(contents)
83 client = AsyncClient(config["homeserver"])
84
85 client.access_token = config["access_token"]
86 client.user_id = config["user_id"]
87 client.device_id = config["device_id"]
88
89 # Now we can send messages as the user
90 room_id = "!myfavouriteroomid:example.org"
91 room_id = input(f"Enter room id for test message: [{room_id}] ")
92
93 await client.room_send(
94 room_id,
95 message_type="m.room.message",
96 content={"msgtype": "m.text", "body": "Hello world!"},
97 )
98 print("Logged in using stored credentials. Sent a test message.")
99
100 # Either way we're logged in here, too
101 await client.close()
102
103
104asyncio.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 async with aiofiles.open(CONFIG_FILE) as f:
156 contents = await f.read()
157 config = json.loads(contents)
158 client = AsyncClient(config["homeserver"])
159
160 client.access_token = config["access_token"]
161 client.user_id = config["user_id"]
162 client.device_id = config["device_id"]
163
164 # Now we can send messages as the user
165 room_id = "!myfavouriteroomid:example.org"
166 room_id = input(f"Enter room id for image message: [{room_id}] ")
167
168 image = "exampledir/samplephoto.jpg"
169 image = input(f"Enter file name of image to send: [{image}] ")
170
171 await send_image(client, room_id, image)
172 print("Logged in using stored credentials. Sent a test message.")
173
174 # Close the client connection after we are done with it.
175 await client.close()
176
177
178asyncio.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
7import aiofiles
8
9from nio import (
10 AsyncClient,
11 ClientConfig,
12 InviteEvent,
13 LoginResponse,
14 MatrixRoom,
15 RoomMessageText,
16 crypto,
17 exceptions,
18)
19
20# This is a fully-documented example of how to do manual verification with nio,
21# for when you already know the device IDs of the users you want to trust. If
22# you want live verification using emojis, the process is more complicated and
23# will be covered in another example.
24
25# We're building on the restore_login example here to preserve device IDs and
26# therefore preserve trust; if @bob trusts @alice's device ID ABC and @alice
27# restarts this program, loading the same keys, @bob will preserve trust. If
28# @alice logged in again @alice would have new keys and a device ID XYZ, and
29# @bob wouldn't trust it.
30
31# The store is where we want to place encryption details like our keys, trusted
32# devices and blacklisted devices. Here we place it in the working directory,
33# but if you deploy your program you might consider /var or /opt for storage
34STORE_FOLDER = "nio_store/"
35
36# This file is for restoring login details after closing the program, so you
37# can preserve your device ID. If @alice logged in every time instead, @bob
38# would have to re-verify. See the restoring login example for more into.
39SESSION_DETAILS_FILE = "credentials.json"
40
41# Only needed for this example, this is who @alice will securely
42# communicate with. We need all the device IDs of this user so we can consider
43# them "trusted". If an unknown device shows up (like @bob signs into their
44# account on another device), this program will refuse to send a message in the
45# room. Try it!
46BOB_ID = "@bob:example.org"
47BOB_DEVICE_IDS = [
48 # You can find these in Riot under Settings > Security & Privacy.
49 # They may also be called "session IDs". You'll want to add ALL of them here
50 # for the one other user in your encrypted room
51 "URDEVICEID",
52]
53
54# the ID of the room you want your bot to join and send commands in.
55# This can be a direct message or room; Matrix treats them the same
56ROOM_ID = "!myfavouriteroom:example.org"
57
58ALICE_USER_ID = "@alice:example.org"
59ALICE_HOMESERVER = "https://matrix.example.org"
60ALICE_PASSWORD = "hunter2"
61
62
63class CustomEncryptedClient(AsyncClient):
64 def __init__(
65 self,
66 homeserver,
67 user="",
68 device_id="",
69 store_path="",
70 config=None,
71 ssl=None,
72 proxy=None,
73 ):
74 # Calling super.__init__ means we're running the __init__ method
75 # defined in AsyncClient, which this class derives from. That does a
76 # bunch of setup for us automatically
77 super().__init__(
78 homeserver,
79 user=user,
80 device_id=device_id,
81 store_path=store_path,
82 config=config,
83 ssl=ssl,
84 proxy=proxy,
85 )
86
87 # if the store location doesn't exist, we'll make it
88 if store_path and not os.path.isdir(store_path):
89 os.mkdir(store_path)
90
91 # auto-join room invites
92 self.add_event_callback(self.cb_autojoin_room, InviteEvent)
93
94 # print all the messages we receive
95 self.add_event_callback(self.cb_print_messages, RoomMessageText)
96
97 async def login(self) -> None:
98 """Log in either using the global variables or (if possible) using the
99 session details file.
100
101 NOTE: This method kinda sucks. Don't use these kinds of global
102 variables in your program; it would be much better to pass them
103 around instead. They are only used here to minimise the size of the
104 example.
105 """
106 # Restore the previous session if we can
107 # See the "restore_login.py" example if you're not sure how this works
108 if os.path.exists(SESSION_DETAILS_FILE) and os.path.isfile(
109 SESSION_DETAILS_FILE
110 ):
111 try:
112 async with aiofiles.open(SESSION_DETAILS_FILE) as f:
113 contents = await f.read()
114 config = json.loads(contents)
115 self.access_token = config["access_token"]
116 self.user_id = config["user_id"]
117 self.device_id = config["device_id"]
118
119 # This loads our verified/blacklisted devices and our keys
120 self.load_store()
121 print(
122 f"Logged in using stored credentials: {self.user_id} on {self.device_id}"
123 )
124
125 except OSError as err:
126 print(f"Couldn't load session from file. Logging in. Error: {err}")
127 except json.JSONDecodeError:
128 print("Couldn't read JSON file; overwriting")
129
130 # We didn't restore a previous session, so we'll log in with a password
131 if not self.user_id or not self.access_token or not self.device_id:
132 # this calls the login method defined in AsyncClient from nio
133 resp = await super().login(ALICE_PASSWORD)
134
135 if isinstance(resp, LoginResponse):
136 print("Logged in using a password; saving details to disk")
137 self.__write_details_to_disk(resp)
138 else:
139 print(f"Failed to log in: {resp}")
140 sys.exit(1)
141
142 def trust_devices(self, user_id: str, device_list: Optional[str] = None) -> None:
143 """Trusts the devices of a user.
144
145 If no device_list is provided, all of the users devices are trusted. If
146 one is provided, only the devices with IDs in that list are trusted.
147
148 Arguments:
149 user_id {str} -- the user ID whose devices should be trusted.
150
151 Keyword Arguments:
152 device_list {Optional[str]} -- The full list of device IDs to trust
153 from that user (default: {None})
154 """
155
156 print(f"{user_id}'s device store: {self.device_store[user_id]}")
157
158 # The device store contains a dictionary of device IDs and known
159 # OlmDevices for all users that share a room with us, including us.
160
161 # We can only run this after a first sync. We have to populate our
162 # device store and that requires syncing with the server.
163 for device_id, olm_device in self.device_store[user_id].items():
164 if device_list and device_id not in device_list:
165 # a list of trusted devices was provided, but this ID is not in
166 # that list. That's an issue.
167 print(
168 f"Not trusting {device_id} as it's not in {user_id}'s pre-approved list."
169 )
170 continue
171
172 if user_id == self.user_id and device_id == self.device_id:
173 # We cannot explicitly trust the device @alice is using
174 continue
175
176 self.verify_device(olm_device)
177 print(f"Trusting {device_id} from user {user_id}")
178
179 def cb_autojoin_room(self, room: MatrixRoom, event: InviteEvent):
180 """Callback to automatically joins a Matrix room on invite.
181
182 Arguments:
183 room {MatrixRoom} -- Provided by nio
184 event {InviteEvent} -- Provided by nio
185 """
186 self.join(room.room_id)
187 room = self.rooms[ROOM_ID]
188 print(f"Room {room.name} is encrypted: {room.encrypted}")
189
190 async def cb_print_messages(self, room: MatrixRoom, event: RoomMessageText):
191 """Callback to print all received messages to stdout.
192
193 Arguments:
194 room {MatrixRoom} -- Provided by nio
195 event {RoomMessageText} -- Provided by nio
196 """
197 if event.decrypted:
198 encrypted_symbol = "🛡 "
199 else:
200 encrypted_symbol = "⚠️ "
201 print(
202 f"{room.display_name} |{encrypted_symbol}| {room.user_name(event.sender)}: {event.body}"
203 )
204
205 async def send_hello_world(self):
206 # Now we send an encrypted message that @bob can read, although it will
207 # appear to be "unverified" when they see it, because @bob has not verified
208 # the device @alice is sending from.
209 # We'll leave that as an exercise for the reader.
210 try:
211 await self.room_send(
212 room_id=ROOM_ID,
213 message_type="m.room.message",
214 content={
215 "msgtype": "m.text",
216 "body": "Hello, this message is encrypted",
217 },
218 )
219 except exceptions.OlmUnverifiedDeviceError:
220 print("These are all known devices:")
221 device_store: crypto.DeviceStore = device_store # noqa: F821
222 [
223 print(
224 f"\t{device.user_id}\t {device.device_id}\t {device.trust_state}\t {device.display_name}"
225 )
226 for device in device_store
227 ]
228 sys.exit(1)
229
230 @staticmethod
231 def __write_details_to_disk(resp: LoginResponse) -> None:
232 """Writes login details to disk so that we can restore our session later
233 without logging in again and creating a new device ID.
234
235 Arguments:
236 resp {LoginResponse} -- the successful client login response.
237 """
238 with open(SESSION_DETAILS_FILE, "w") as f:
239 json.dump(
240 {
241 "access_token": resp.access_token,
242 "device_id": resp.device_id,
243 "user_id": resp.user_id,
244 },
245 f,
246 )
247
248
249async def run_client(client: CustomEncryptedClient) -> None:
250 """A basic encrypted chat application using nio."""
251
252 # This is our own custom login function that looks for a pre-existing config
253 # file and, if it exists, logs in using those details. Otherwise it will log
254 # in using a password.
255 await client.login()
256
257 # Here we create a coroutine that we can call in asyncio.gather later,
258 # along with sync_forever and any other API-related coroutines you'd like
259 # to do.
260 async def after_first_sync():
261 # We'll wait for the first firing of 'synced' before trusting devices.
262 # client.synced is an asyncio event that fires any time nio syncs. This
263 # code doesn't run in a loop, so it only fires once
264 print("Awaiting sync")
265 await client.synced.wait()
266
267 # In practice, you want to have a list of previously-known device IDs
268 # for each user you want to trust. Here, we require that list as a
269 # global variable
270 client.trust_devices(BOB_ID, BOB_DEVICE_IDS)
271
272 # In this case, we'll trust _all_ of @alice's devices. NOTE that this
273 # is a SUPER BAD IDEA in practice, but for the purpose of this example
274 # it'll be easier, since you may end up creating lots of sessions for
275 # @alice as you play with the script
276 client.trust_devices(ALICE_USER_ID)
277
278 await client.send_hello_world()
279
280 # We're creating Tasks here so that you could potentially write other
281 # Python coroutines to do other work, like checking an API or using another
282 # library. All of these Tasks will be run concurrently.
283 # For more details, check out https://docs.python.org/3/library/asyncio-task.html
284
285 # ensure_future() is for Python 3.5 and 3.6 compatibility. For 3.7+, use
286 # asyncio.create_task()
287 after_first_sync_task = asyncio.ensure_future(after_first_sync())
288
289 # We use full_state=True here to pull any room invites that occurred or
290 # messages sent in rooms _before_ this program connected to the
291 # Matrix server
292 sync_forever_task = asyncio.ensure_future(
293 client.sync_forever(30000, full_state=True)
294 )
295
296 await asyncio.gather(
297 # The order here IS significant! You have to register the task to trust
298 # devices FIRST since it awaits the first sync
299 after_first_sync_task,
300 sync_forever_task,
301 )
302
303
304async def main():
305 # By setting `store_sync_tokens` to true, we'll save sync tokens to our
306 # store every time we sync, thereby preventing reading old, previously read
307 # events on each new sync.
308 # For more info, check out https://matrix-nio.readthedocs.io/en/latest/nio.html#asyncclient
309 config = ClientConfig(store_sync_tokens=True)
310 client = CustomEncryptedClient(
311 ALICE_HOMESERVER,
312 ALICE_USER_ID,
313 store_path=STORE_FOLDER,
314 config=config,
315 ssl=False,
316 proxy="http://localhost:8080",
317 )
318
319 try:
320 await run_client(client)
321 except (asyncio.CancelledError, KeyboardInterrupt):
322 await client.close()
323
324
325# Run the main coroutine, which instantiates our custom subclass, trusts all the
326# devices, and syncs forever (or until your press Ctrl+C)
327
328if __name__ == "__main__":
329 try:
330 asyncio.run(main())
331 except KeyboardInterrupt:
332 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
71import aiofiles
72
73from nio import (
74 AsyncClient,
75 AsyncClientConfig,
76 KeyVerificationCancel,
77 KeyVerificationEvent,
78 KeyVerificationKey,
79 KeyVerificationMac,
80 KeyVerificationStart,
81 LocalProtocolError,
82 LoginResponse,
83 ToDeviceError,
84)
85
86# file to store credentials in case you want to run program multiple times
87CONFIG_FILE = "credentials.json" # login credentials JSON file
88# directory to store persistent data for end-to-end encryption
89STORE_PATH = "./store/" # local directory
90
91
92class Callbacks:
93 """Class to pass client to callback methods."""
94
95 def __init__(self, client):
96 """Store AsyncClient."""
97 self.client = client
98
99 async def to_device_callback(self, event): # noqa
100 """Handle events sent to device."""
101 try:
102 client = self.client
103
104 if isinstance(event, KeyVerificationStart): # first step
105 """first step: receive KeyVerificationStart
106 KeyVerificationStart(
107 source={'content':
108 {'method': 'm.sas.v1',
109 'from_device': 'DEVICEIDXY',
110 'key_agreement_protocols':
111 ['curve25519-hkdf-sha256', 'curve25519'],
112 'hashes': ['sha256'],
113 'message_authentication_codes':
114 ['hkdf-hmac-sha256', 'hmac-sha256'],
115 'short_authentication_string':
116 ['decimal', 'emoji'],
117 'transaction_id': 'SomeTxId'
118 },
119 'type': 'm.key.verification.start',
120 'sender': '@user2:example.org'
121 },
122 sender='@user2:example.org',
123 transaction_id='SomeTxId',
124 from_device='DEVICEIDXY',
125 method='m.sas.v1',
126 key_agreement_protocols=[
127 'curve25519-hkdf-sha256', 'curve25519'],
128 hashes=['sha256'],
129 message_authentication_codes=[
130 'hkdf-hmac-sha256', 'hmac-sha256'],
131 short_authentication_string=['decimal', 'emoji'])
132 """
133
134 if "emoji" not in event.short_authentication_string:
135 print(
136 "Other device does not support emoji verification "
137 f"{event.short_authentication_string}."
138 )
139 return
140 resp = await client.accept_key_verification(event.transaction_id)
141 if isinstance(resp, ToDeviceError):
142 print(f"accept_key_verification failed with {resp}")
143
144 sas = client.key_verifications[event.transaction_id]
145
146 todevice_msg = sas.share_key()
147 resp = await client.to_device(todevice_msg)
148 if isinstance(resp, ToDeviceError):
149 print(f"to_device failed with {resp}")
150
151 elif isinstance(event, KeyVerificationCancel): # anytime
152 """at any time: receive KeyVerificationCancel
153 KeyVerificationCancel(source={
154 'content': {'code': 'm.mismatched_sas',
155 'reason': 'Mismatched authentication string',
156 'transaction_id': 'SomeTxId'},
157 'type': 'm.key.verification.cancel',
158 'sender': '@user2:example.org'},
159 sender='@user2:example.org',
160 transaction_id='SomeTxId',
161 code='m.mismatched_sas',
162 reason='Mismatched short authentication string')
163 """
164
165 # There is no need to issue a
166 # client.cancel_key_verification(tx_id, reject=False)
167 # here. The SAS flow is already cancelled.
168 # We only need to inform the user.
169 print(
170 f"Verification has been cancelled by {event.sender} "
171 f'for reason "{event.reason}".'
172 )
173
174 elif isinstance(event, KeyVerificationKey): # second step
175 """Second step is to receive KeyVerificationKey
176 KeyVerificationKey(
177 source={'content': {
178 'key': 'SomeCryptoKey',
179 'transaction_id': 'SomeTxId'},
180 'type': 'm.key.verification.key',
181 'sender': '@user2:example.org'
182 },
183 sender='@user2:example.org',
184 transaction_id='SomeTxId',
185 key='SomeCryptoKey')
186 """
187 sas = client.key_verifications[event.transaction_id]
188
189 print(f"{sas.get_emoji()}")
190
191 yn = input("Do the emojis match? (Y/N) (C for Cancel) ")
192 if yn.lower() == "y":
193 print(
194 "Match! The verification for this " "device will be accepted."
195 )
196 resp = await client.confirm_short_auth_string(event.transaction_id)
197 if isinstance(resp, ToDeviceError):
198 print(f"confirm_short_auth_string failed with {resp}")
199 elif yn.lower() == "n": # no, don't match, reject
200 print(
201 "No match! Device will NOT be verified "
202 "by rejecting verification."
203 )
204 resp = await client.cancel_key_verification(
205 event.transaction_id, reject=True
206 )
207 if isinstance(resp, ToDeviceError):
208 print(f"cancel_key_verification failed with {resp}")
209 else: # C or anything for cancel
210 print("Cancelled by user! Verification will be " "cancelled.")
211 resp = await client.cancel_key_verification(
212 event.transaction_id, reject=False
213 )
214 if isinstance(resp, ToDeviceError):
215 print(f"cancel_key_verification failed with {resp}")
216
217 elif isinstance(event, KeyVerificationMac): # third step
218 """Third step is to receive KeyVerificationMac
219 KeyVerificationMac(
220 source={'content': {
221 'mac': {'ed25519:DEVICEIDXY': 'SomeKey1',
222 'ed25519:SomeKey2': 'SomeKey3'},
223 'keys': 'SomeCryptoKey4',
224 'transaction_id': 'SomeTxId'},
225 'type': 'm.key.verification.mac',
226 'sender': '@user2:example.org'},
227 sender='@user2:example.org',
228 transaction_id='SomeTxId',
229 mac={'ed25519:DEVICEIDXY': 'SomeKey1',
230 'ed25519:SomeKey2': 'SomeKey3'},
231 keys='SomeCryptoKey4')
232 """
233 sas = client.key_verifications[event.transaction_id]
234 try:
235 todevice_msg = sas.get_mac()
236 except LocalProtocolError as e:
237 # e.g. it might have been cancelled by ourselves
238 print(
239 f"Cancelled or protocol error: Reason: {e}.\n"
240 f"Verification with {event.sender} not concluded. "
241 "Try again?"
242 )
243 else:
244 resp = await client.to_device(todevice_msg)
245 if isinstance(resp, ToDeviceError):
246 print(f"to_device failed with {resp}")
247 print(
248 f"sas.we_started_it = {sas.we_started_it}\n"
249 f"sas.sas_accepted = {sas.sas_accepted}\n"
250 f"sas.canceled = {sas.canceled}\n"
251 f"sas.timed_out = {sas.timed_out}\n"
252 f"sas.verified = {sas.verified}\n"
253 f"sas.verified_devices = {sas.verified_devices}\n"
254 )
255 print(
256 "Emoji verification was successful!\n"
257 "Hit Control-C to stop the program or "
258 "initiate another Emoji verification from "
259 "another device or room."
260 )
261 else:
262 print(
263 f"Received unexpected event type {type(event)}. "
264 f"Event is {event}. Event will be ignored."
265 )
266 except BaseException:
267 print(traceback.format_exc())
268
269
270def write_details_to_disk(resp: LoginResponse, homeserver) -> None:
271 """Write the required login details to disk.
272
273 It will allow following logins to be made without password.
274
275 Arguments:
276 ---------
277 resp : LoginResponse - successful client login response
278 homeserver : str - URL of homeserver, e.g. "https://matrix.example.org"
279
280 """
281 # open the config file in write-mode
282 with open(CONFIG_FILE, "w") as f:
283 # write the login details to disk
284 json.dump(
285 {
286 "homeserver": homeserver, # e.g. "https://matrix.example.org"
287 "user_id": resp.user_id, # e.g. "@user:example.org"
288 "device_id": resp.device_id, # device ID, 10 uppercase letters
289 "access_token": resp.access_token, # cryptogr. access token
290 },
291 f,
292 )
293
294
295async def login() -> AsyncClient:
296 """Handle login with or without stored credentials."""
297 # Configuration options for the AsyncClient
298 client_config = AsyncClientConfig(
299 max_limit_exceeded=0,
300 max_timeouts=0,
301 store_sync_tokens=True,
302 encryption_enabled=True,
303 )
304
305 # If there are no previously-saved credentials, we'll use the password
306 if not os.path.exists(CONFIG_FILE):
307 print(
308 "First time use. Did not find credential file. Asking for "
309 "homeserver, user, and password to create credential file."
310 )
311 homeserver = "https://matrix.example.org"
312 homeserver = input(f"Enter your homeserver URL: [{homeserver}] ")
313
314 if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
315 homeserver = "https://" + homeserver
316
317 user_id = "@user:example.org"
318 user_id = input(f"Enter your full user ID: [{user_id}] ")
319
320 device_name = "matrix-nio"
321 device_name = input(f"Choose a name for this device: [{device_name}] ")
322
323 if not os.path.exists(STORE_PATH):
324 os.makedirs(STORE_PATH)
325
326 # Initialize the matrix client
327 client = AsyncClient(
328 homeserver,
329 user_id,
330 store_path=STORE_PATH,
331 config=client_config,
332 )
333 pw = getpass.getpass()
334
335 resp = await client.login(password=pw, device_name=device_name)
336
337 # check that we logged in successfully
338 if isinstance(resp, LoginResponse):
339 write_details_to_disk(resp, homeserver)
340 else:
341 print(f'homeserver = "{homeserver}"; user = "{user_id}"')
342 print(f"Failed to log in: {resp}")
343 sys.exit(1)
344
345 print(
346 "Logged in using a password. Credentials were stored. "
347 "On next execution the stored login credentials will be used."
348 )
349
350 # Otherwise the config file exists, so we'll use the stored credentials
351 else:
352 # open the file in read-only mode
353 async with aiofiles.open(CONFIG_FILE) as f:
354 contents = await f.read()
355 config = json.loads(contents)
356 # Initialize the matrix client based on credentials from file
357 client = AsyncClient(
358 config["homeserver"],
359 config["user_id"],
360 device_id=config["device_id"],
361 store_path=STORE_PATH,
362 config=client_config,
363 )
364
365 client.restore_login(
366 user_id=config["user_id"],
367 device_id=config["device_id"],
368 access_token=config["access_token"],
369 )
370 print("Logged in using stored credentials.")
371
372 return client
373
374
375async def main() -> None:
376 """Login and wait for and perform emoji verify."""
377 client = await login()
378 # Set up event callbacks
379 callbacks = Callbacks(client)
380 client.add_to_device_callback(callbacks.to_device_callback, (KeyVerificationEvent,))
381 # Sync encryption keys with the server
382 # Required for participating in encrypted rooms
383 if client.should_upload_keys:
384 await client.keys_upload()
385 print(
386 "This program is ready and waiting for the other party to initiate "
387 'an emoji verification with us by selecting "Verify by Emoji" '
388 "in their Matrix client."
389 )
390 await client.sync_forever(timeout=30000, full_state=True)
391
392
393try:
394 asyncio.run(main())
395except Exception:
396 print(traceback.format_exc())
397 sys.exit(1)
398except KeyboardInterrupt:
399 print("Received keyboard interrupt.")
400 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.