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