bot.py (12244B)
1 # SPDX-FileCopyrightText: 2023-present MTRNord <support@nordgedanken.dev> 2 # 3 # SPDX-License-Identifier: AGPL-3.0-or-later 4 5 import asyncio 6 import getpass 7 import json 8 import os 9 import sys 10 11 import logbook # type: ignore 12 from logbook import Logger, StreamHandler 13 from nio import ( 14 AsyncClient, 15 AsyncClientConfig, 16 CallCandidatesEvent, 17 CallHangupEvent, 18 CallInviteEvent, 19 CallMemberEvent, 20 InviteEvent, 21 LoginResponse, 22 MatrixRoom, 23 MSC3401CallEvent, 24 ProfileGetDisplayNameResponse, 25 RoomMessageText, 26 ToDeviceCallAnswerEvent, 27 ToDeviceCallCandidatesEvent, 28 ToDeviceCallHangupEvent, 29 ToDeviceCallInviteEvent, 30 ) 31 from nio.events import to_device 32 33 from matrix_call_multitrack_recorder.recorder import Recorder 34 35 StreamHandler(sys.stdout).push_application() 36 logger = Logger(__name__) 37 logger.level = logbook.INFO 38 # crypto.logger.level = logbook.DEBUG 39 to_device.logger.level = logbook.DEBUG 40 41 # file to store credentials in case you want to run program multiple times 42 CONFIG_FILE = "credentials.json" # login credentials JSON file 43 # directory to store persistent data for end-to-end encryption 44 STORE_PATH = "./store/" # local directory 45 46 47 class RecordingBot: 48 client: AsyncClient 49 loop: asyncio.AbstractEventLoop 50 recorder: Recorder 51 52 def __init__(self, client) -> None: 53 self.client = client 54 self.loop = asyncio.get_event_loop() 55 self.recorder = Recorder(self.client) 56 57 async def start(self) -> None: 58 logger.info("Starting client") 59 60 if self.client.should_upload_keys: 61 await self.client.keys_upload() 62 logger.debug("Uploaded keys") 63 64 self.client.add_event_callback(self.message_callback, RoomMessageText) # type: ignore 65 self.client.add_event_callback(self.call_invite, CallInviteEvent) # type: ignore 66 self.client.add_event_callback(self.call_candidates, CallCandidatesEvent) # type: ignore 67 self.client.add_event_callback(self.call_hangup, CallHangupEvent) # type: ignore 68 self.client.add_event_callback(self.msc3401_call, MSC3401CallEvent) # type: ignore 69 self.client.add_event_callback(self.msc3401_call_member, CallMemberEvent) # type: ignore 70 71 self.client.add_to_device_callback( 72 self.to_device_call_candidates, (ToDeviceCallCandidatesEvent,) # type: ignore 73 ) 74 self.client.add_to_device_callback( 75 self.to_device_call_invite, (ToDeviceCallInviteEvent,) # type: ignore 76 ) 77 self.client.add_to_device_callback( 78 self.to_device_call_hangup, (ToDeviceCallHangupEvent,) # type: ignore 79 ) 80 self.client.add_to_device_callback( 81 self.call_answer, (ToDeviceCallAnswerEvent,) # type: ignore 82 ) 83 84 self.client.add_event_callback(self.cb_autojoin_room, InviteEvent) # type: ignore 85 logger.info("Listening for calls") 86 87 await self.client.sync_forever( 88 timeout=30000, full_state=True, set_presence="online" 89 ) 90 91 async def stop(self) -> None: 92 logger.info("Stopping client") 93 await self.recorder.stop() 94 await self.client.close() 95 96 async def message_callback(self, room: MatrixRoom, event: RoomMessageText) -> None: 97 """Handles incoming messages.""" 98 99 async def command_handling() -> None: 100 if event.body.startswith("!help"): 101 await self.client.room_send( 102 room.room_id, 103 "m.room.message", 104 { 105 "msgtype": "m.notice", 106 "body": "Please note that commands only work with the new voip (Element-Call)\n\nCommands:\n!help - Shows this help\n!stop - Stops the recording\n!start - Starts the recording", 107 }, 108 ignore_unverified_devices=True, 109 ) 110 await self.client.update_receipt_marker(room.room_id, event.event_id) 111 elif event.body.startswith("!stop"): 112 await self.recorder.leave_call(room) 113 await self.client.update_receipt_marker(room.room_id, event.event_id) 114 115 elif event.body.startswith("!start"): 116 117 await self.recorder.join_call(room) 118 await self.client.update_receipt_marker(room.room_id, event.event_id) 119 120 asyncio.create_task(command_handling()) 121 122 # TODO: Negotiate new streams 123 # TODO: Handle m.call.negotiate to-device events 124 # TODO: Send m.call.negotiate events with type offer 125 async def msc3401_call_member( 126 self, room: MatrixRoom, event: CallMemberEvent 127 ) -> None: 128 """Handles incoming MSC3401 call members.""" 129 if event.sender == self.client.user_id: 130 return 131 132 # logger.info(f"MSC3401 call member event: {event}") 133 if not event.calls: 134 conf_id = await self.recorder.remove_connection(room) 135 if conf_id: 136 self.recorder.remove_other(conf_id, event.sender) 137 138 for call in event.calls: 139 for device in call["m.devices"]: 140 if device["device_id"] != self.client.device_id: 141 self.recorder.add_call(call["m.call_id"], room) 142 self.recorder.track_others( 143 call["m.call_id"], 144 device["device_id"], 145 event.sender, 146 device["session_id"], 147 ) 148 # TODO: Can I reuse the same connection? Do I have the info needed? 149 # Is it a new connection? How do I see if it changed? 150 151 # asyncio.create_task(self.handle_call_invite(event, room)) 152 153 async def msc3401_call(self, room: MatrixRoom, event: MSC3401CallEvent) -> None: 154 """Handles incoming MSC3401 calls.""" 155 self.recorder.add_call(event.state_key, room) 156 157 async def call_invite(self, room: MatrixRoom, event: CallInviteEvent) -> None: 158 """Handles incoming call invites.""" 159 160 asyncio.create_task(self.recorder.handle_call_invite(event, room)) 161 162 async def call_answer(self, event: ToDeviceCallAnswerEvent) -> None: 163 """Handles incoming call answers.""" 164 165 asyncio.create_task(self.recorder.handle_call_answer(event)) 166 167 async def to_device_call_invite(self, event: CallInviteEvent) -> None: 168 """Handles incoming call invites.""" 169 logger.info("Received to-device call invite") 170 171 asyncio.create_task(self.recorder.handle_call_invite(event, None)) 172 173 async def call_candidates( 174 self, room: MatrixRoom, event: CallCandidatesEvent 175 ) -> None: 176 """Handles incoming call candidates.""" 177 178 asyncio.create_task(self.recorder.handle_call_candidates(room, event)) 179 180 async def to_device_call_candidates(self, event: CallCandidatesEvent) -> None: 181 """Handles incoming call candidates.""" 182 logger.info("Received to-device call candidates") 183 184 asyncio.create_task(self.recorder.handle_call_candidates(None, event)) 185 186 async def call_hangup(self, room: MatrixRoom, event: CallHangupEvent) -> None: 187 """Handles call hangups.""" 188 asyncio.create_task(self.recorder.handle_call_hangup(room, event)) 189 190 async def to_device_call_hangup(self, event: CallHangupEvent) -> None: 191 """Handles call hangups.""" 192 asyncio.create_task(self.recorder.handle_call_hangup(None, event)) 193 194 async def cb_autojoin_room(self, room: MatrixRoom, event: InviteEvent) -> None: 195 """Callback to automatically joins a Matrix room on invite. 196 197 Arguments: 198 room {MatrixRoom} -- Provided by nio 199 event {InviteEvent} -- Provided by nio 200 """ 201 202 async def join_task() -> None: 203 await self.client.join(room.room_id) 204 sender_display_name = await self.client.get_displayname(event.sender) 205 if isinstance(sender_display_name, ProfileGetDisplayNameResponse): 206 response = f"""Hello, I am a bot that records calls. Use !help to see available commands. 207 I was invited by {sender_display_name.displayname}""" 208 else: 209 response = f"""Hello, I am a bot that records calls. Use !help to see available commands. 210 I was invited by {event.sender}""" 211 # FIXME: We should also check if the join worked or not. 212 while room.room_id not in self.client.rooms: 213 logger.debug("Waiting for room to be joined") 214 await asyncio.sleep(0.2) 215 await self.client.room_send( 216 room.room_id, 217 "m.room.message", 218 { 219 "msgtype": "m.text", 220 "body": response, 221 }, 222 ignore_unverified_devices=True, 223 ) 224 225 asyncio.create_task(join_task()) 226 227 logger.info(f"Joining room {room.room_id} on invite from {event.sender}") 228 229 230 def write_details_to_disk(resp: LoginResponse, homeserver) -> None: 231 """Writes the required login details to disk so we can log in later without 232 using a password. 233 234 Arguments: 235 resp {LoginResponse} -- the successful client login response. 236 homeserver -- URL of homeserver, e.g. "https://matrix.example.org" 237 """ 238 # open the config file in write-mode 239 with open(CONFIG_FILE, "w", encoding="utf8") as f: 240 # write the login details to disk 241 json.dump( 242 { 243 "homeserver": homeserver, # e.g. "https://matrix.example.org" 244 "user_id": resp.user_id, # e.g. "@user:example.org" 245 "device_id": resp.device_id, # device ID, 10 uppercase letters 246 "access_token": resp.access_token, # cryptogr. access token 247 }, 248 f, 249 ) 250 251 252 async def login() -> AsyncClient: 253 client_config = AsyncClientConfig( 254 store_sync_tokens=True, 255 encryption_enabled=True, 256 ) 257 # If there are no previously-saved credentials, we'll use the password 258 if not os.path.exists(CONFIG_FILE): 259 logger.info( 260 "First time use. Did not find credential file. Asking for " 261 "homeserver, user, and password to create credential file." 262 ) 263 homeserver = "https://matrix.midnightthoughts.space" 264 homeserver = input(f"Enter your homeserver URL: [{homeserver}] ") 265 266 if not (homeserver.startswith("https://") or homeserver.startswith("http://")): 267 homeserver = "https://" + homeserver 268 269 user_id = "@user:midnightthoughts.space" 270 user_id = input(f"Enter your full user ID: [{user_id}] ") 271 272 device_name = "matrix-call-multitrack-recorder" 273 # device_name = input(f"Choose a name for this device: [{device_name}] ") 274 275 if not os.path.exists(STORE_PATH): 276 os.makedirs(STORE_PATH) 277 278 client = AsyncClient( 279 homeserver, 280 user_id, 281 store_path=STORE_PATH, 282 config=client_config, 283 ) 284 pw = getpass.getpass() 285 286 resp = await client.login(pw, device_name=device_name) 287 288 # check that we logged in succesfully 289 if isinstance(resp, LoginResponse): 290 write_details_to_disk(resp, homeserver) 291 logger.info( 292 "Logged in using a password. Credentials were stored. " 293 "On next execution the stored login credentials will be used." 294 ) 295 else: 296 logger.info(f'homeserver = "{homeserver}"; user = "{user_id}"') 297 logger.error(f"Failed to log in: {resp}") 298 sys.exit(1) 299 300 # Otherwise the config file exists, so we'll use the stored credentials 301 else: 302 # open the file in read-only mode 303 with open(CONFIG_FILE, "r", encoding="utf8") as f: 304 config = json.load(f) 305 client = AsyncClient( 306 config["homeserver"], 307 config["user_id"], 308 device_id=config["device_id"], 309 store_path=STORE_PATH, 310 config=client_config, 311 ) 312 313 client.restore_login( 314 user_id=config["user_id"], 315 device_id=config["device_id"], 316 access_token=config["access_token"], 317 ) 318 return client