Coverage for slidge / core / gateway.py: 64%

1""" 

2This module extends slixmpp.ComponentXMPP to make writing new LegacyClients easier 

3""" 

4 

5import asyncio 

6import logging 

7import re 

8import tempfile 

9from copy import copy 

10from datetime import datetime 

11from pathlib import Path 

12from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Type, Union, cast 

13 

14import aiohttp 

15import qrcode 

16from slixmpp import JID, ComponentXMPP, Iq, Message, Presence 

17from slixmpp.exceptions import IqError, IqTimeout, XMPPError 

18from slixmpp.plugins.xep_0004 import Form 

19from slixmpp.plugins.xep_0060.stanza import OwnerAffiliation 

20from slixmpp.plugins.xep_0356.permissions import IqPermission 

21from slixmpp.plugins.xep_0356.privilege import PrivilegedIqError 

22from slixmpp.types import MessageTypes 

23from slixmpp.xmlstream.xmlstream import NotConnectedError 

24 

25from slidge import LegacyContact, command # noqa: F401 

26from slidge.command.adhoc import AdhocProvider 

27from slidge.command.admin import Exec 

28from slidge.command.base import Command, FormField 

29from slidge.command.chat_command import ChatCommandProvider 

30from slidge.command.register import RegistrationType 

31from slidge.core import config 

32from slidge.core.dispatcher.session_dispatcher import SessionDispatcher 

33from slidge.core.mixins import MessageMixin 

34from slidge.core.mixins.attachment import AttachmentMixin 

35from slidge.core.mixins.avatar import convert_avatar 

36from slidge.core.pubsub import PubSubComponent 

37from slidge.core.session import BaseSession 

38from slidge.db import GatewayUser, SlidgeStore 

39from slidge.db.avatar import CachedAvatar, avatar_cache 

40from slidge.db.meta import JSONSerializable 

41from slidge.slixfix.delivery_receipt import DeliveryReceipt 

42from slidge.slixfix.roster import RosterBackend 

43from slidge.util import ABCSubclassableOnceAtMost 

44from slidge.util.types import Avatar, MessageOrPresenceTypeVar 

45from slidge.util.util import timeit 

46 

47if TYPE_CHECKING: 

48 pass 

49 

50 

51class BaseGateway( 

52 ComponentXMPP, 

53 MessageMixin, 

54 metaclass=ABCSubclassableOnceAtMost, 

55): 

56 """ 

57 The gateway component, handling registrations and un-registrations. 

58 

59 On slidge launch, a singleton is instantiated, and it will be made available 

60 to public classes such :class:`.LegacyContact` or :class:`.BaseSession` as the 

61 ``.xmpp`` attribute. 

62 

63 Must be subclassed by a legacy module to set up various aspects of the XMPP 

64 component behaviour, such as its display name or welcome message, via 

65 class attributes :attr:`.COMPONENT_NAME` :attr:`.WELCOME_MESSAGE`. 

66 

67 Abstract methods related to the registration process must be overriden 

68 for a functional :term:`Legacy Module`: 

69 

70 - :meth:`.validate` 

71 - :meth:`.validate_two_factor_code` 

72 - :meth:`.get_qr_text` 

73 - :meth:`.confirm_qr` 

74 

75 NB: Not all of these must be overridden, it depends on the 

76 :attr:`REGISTRATION_TYPE`. 

77 

78 The other methods, such as :meth:`.send_text` or :meth:`.react` are the same 

79 as those of :class:`.LegacyContact` and :class:`.LegacyParticipant`, because 

80 the component itself is also a "messaging actor", ie, an :term:`XMPP Entity`. 

81 For these methods, you need to specify the JID of the recipient with the 

82 `mto` parameter. 

83 

84 Since it inherits from :class:`slixmpp.componentxmpp.ComponentXMPP`,you also 

85 have a hand on low-level XMPP interactions via slixmpp methods, e.g.: 

86 

87 .. code-block:: python 

88 

89 self.send_presence( 

90 pfrom="somebody@component.example.com", 

91 pto="someonwelse@anotherexample.com", 

92 ) 

93 

94 However, you should not need to do so often since the classes of the plugin 

95 API provides higher level abstractions around most commonly needed use-cases, such 

96 as sending messages, or displaying a custom status. 

97 

98 """ 

99 

100 COMPONENT_NAME: str = NotImplemented 

101 """Name of the component, as seen in service discovery by XMPP clients""" 

102 COMPONENT_TYPE: str = "" 

103 """Type of the gateway, should follow https://xmpp.org/registrar/disco-categories.html""" 

104 COMPONENT_AVATAR: Optional[Avatar | Path | str] = None 

105 """ 

106 Path, bytes or URL used by the component as an avatar. 

107 """ 

108 

109 REGISTRATION_FIELDS: Sequence[FormField] = [ 

110 FormField(var="username", label="User name", required=True), 

111 FormField(var="password", label="Password", required=True, private=True), 

112 ] 

113 """ 

114 Iterable of fields presented to the gateway user when registering using :xep:`0077` 

115 `extended <https://xmpp.org/extensions/xep-0077.html#extensibility>`_ by :xep:`0004`. 

116 """ 

117 REGISTRATION_INSTRUCTIONS: str = "Enter your credentials" 

118 """ 

119 The text presented to a user that wants to register (or modify) their legacy account 

120 configuration. 

121 """ 

122 REGISTRATION_TYPE: RegistrationType = RegistrationType.SINGLE_STEP_FORM 

123 """ 

124 This attribute determines how users register to the gateway, ie, how they 

125 login to the :term:`legacy service <Legacy Service>`. 

126 The credentials are then stored persistently, so this process should happen 

127 once per user (unless they unregister). 

128 

129 The registration process always start with a basic data form (:xep:`0004`) 

130 presented to the user. 

131 But the legacy login flow might require something more sophisticated, see 

132 :class:`.RegistrationType` for more details. 

133 """ 

134 

135 REGISTRATION_2FA_TITLE = "Enter your 2FA code" 

136 REGISTRATION_2FA_INSTRUCTIONS = ( 

137 "You should have received something via email or SMS, or something" 

138 ) 

139 REGISTRATION_QR_INSTRUCTIONS = "Flash this code or follow this link" 

140 

141 PREFERENCES = [ 

142 FormField( 

143 var="sync_presence", 

144 label="Propagate your XMPP presence to the legacy network.", 

145 value="true", 

146 required=True, 

147 type="boolean", 

148 ), 

149 FormField( 

150 var="sync_avatar", 

151 label="Propagate your XMPP avatar to the legacy network.", 

152 value="true", 

153 required=True, 

154 type="boolean", 

155 ), 

156 FormField( 

157 var="always_invite_when_adding_bookmarks", 

158 label="Send an invitation to join MUCs after adding them to the bookmarks.", 

159 value="true", 

160 required=True, 

161 type="boolean", 

162 ), 

163 FormField( 

164 var="last_seen_fallback", 

165 label="Use contact presence status message to show when they were last seen.", 

166 value="true", 

167 required=True, 

168 type="boolean", 

169 ), 

170 FormField( 

171 var="roster_push", 

172 label="Add contacts to your roster.", 

173 value="true", 

174 required=True, 

175 type="boolean", 

176 ), 

177 FormField( 

178 var="reaction_fallback", 

179 label="Receive fallback messages for reactions (for legacy XMPP clients)", 

180 value="false", 

181 required=True, 

182 type="boolean", 

183 ), 

184 ] 

185 

186 ROSTER_GROUP: str = "slidge" 

187 """ 

188 Name of the group assigned to a :class:`.LegacyContact` automagically 

189 added to the :term:`User`'s roster with :meth:`.LegacyContact.add_to_roster`. 

190 """ 

191 WELCOME_MESSAGE = ( 

192 "Thank you for registering. Type 'help' to list the available commands, " 

193 "or just start messaging away!" 

194 ) 

195 """ 

196 A welcome message displayed to users on registration. 

197 This is useful notably for clients that don't consider component JIDs as a 

198 valid recipient in their UI, yet still open a functional chat window on 

199 incoming messages from components. 

200 """ 

201 

202 SEARCH_FIELDS: Sequence[FormField] = [ 

203 FormField(var="first", label="First name", required=True), 

204 FormField(var="last", label="Last name", required=True), 

205 FormField(var="phone", label="Phone number", required=False), 

206 ] 

207 """ 

208 Fields used for searching items via the component, through :xep:`0055` (jabber search). 

209 A common use case is to allow users to search for legacy contacts by something else than 

210 their usernames, eg their phone number. 

211 

212 Plugins should implement search by overriding :meth:`.BaseSession.search` 

213 (restricted to registered users). 

214 

215 If there is only one field, it can also be used via the ``jabber:iq:gateway`` protocol 

216 described in :xep:`0100`. Limitation: this only works if the search request returns 

217 one result item, and if this item has a 'jid' var. 

218 """ 

219 SEARCH_TITLE: str = "Search for legacy contacts" 

220 """ 

221 Title of the search form. 

222 """ 

223 SEARCH_INSTRUCTIONS: str = "" 

224 """ 

225 Instructions of the search form. 

226 """ 

227 

228 MARK_ALL_MESSAGES = False 

229 """ 

230 Set this to True for :term:`legacy networks <Legacy Network>` that expects 

231 read marks for *all* messages and not just the latest one that was read 

232 (as most XMPP clients will only send a read mark for the latest msg). 

233 """ 

234 

235 PROPER_RECEIPTS = False 

236 """ 

237 Set this to True if the legacy service provides a real equivalent of message delivery receipts 

238 (:xep:`0184`), meaning that there is an event thrown when the actual device of a contact receives 

239 a message. Make sure to call Contact.received() adequately if this is set to True. 

240 """ 

241 

242 GROUPS = False 

243 

244 mtype: MessageTypes = "chat" 

245 is_group = False 

246 _can_send_carbon = False 

247 store: SlidgeStore 

248 

249 AVATAR_ID_TYPE: Callable[[str], Any] = str 

250 """ 

251 Modify this if the legacy network uses unique avatar IDs that are not strings. 

252 

253 This is required because we store those IDs as TEXT in the persistent SQL DB. 

254 The callable specified here will receive is responsible for converting the 

255 serialised-as-text version of the avatar unique ID back to the proper type. 

256 Common example: ``int``. 

257 """ 

258 # FIXME: do we really need this since we have session.xmpp_to_legacy_msg_id? 

259 # (maybe we do) 

260 LEGACY_MSG_ID_TYPE: Callable[[str], Any] = str 

261 """ 

262 Modify this if the legacy network uses unique message IDs that are not strings. 

263 

264 This is required because we store those IDs as TEXT in the persistent SQL DB. 

265 The callable specified here will receive is responsible for converting the 

266 serialised-as-text version of the message unique ID back to the proper type. 

267 Common example: ``int``. 

268 """ 

269 LEGACY_CONTACT_ID_TYPE: Callable[[str], Any] = str 

270 """ 

271 Modify this if the legacy network uses unique contact IDs that are not strings. 

272 

273 This is required because we store those IDs as TEXT in the persistent SQL DB. 

274 The callable specified here is responsible for converting the 

275 serialised-as-text version of the contact unique ID back to the proper type. 

276 Common example: ``int``. 

277 """ 

278 LEGACY_ROOM_ID_TYPE: Callable[[str], Any] = str 

279 """ 

280 Modify this if the legacy network uses unique room IDs that are not strings. 

281 

282 This is required because we store those IDs as TEXT in the persistent SQL DB. 

283 The callable specified here is responsible for converting the 

284 serialised-as-text version of the room unique ID back to the proper type. 

285 Common example: ``int``. 

286 """ 

287 

288 DB_POOL_SIZE: int = 5 

289 """ 

290 Size of the queue pool for sqlalchemy engine. Typically, when using python async 

291 libraries, this does not need to be changed. 

292 Change that if your gateway use separate threads to call into slidge. The value of 

293 this parameter should be equal or greater than the potential number of threads. 

294 """ 

295 

296 http: aiohttp.ClientSession 

297 avatar: CachedAvatar | None = None 

298 

299 def __init__(self) -> None: 

300 if config.COMPONENT_NAME: 

301 self.COMPONENT_NAME = config.COMPONENT_NAME 

302 if config.WELCOME_MESSAGE: 

303 self.WELCOME_MESSAGE = config.WELCOME_MESSAGE 

304 self.log = log 

305 self.datetime_started = datetime.now() 

306 self.xmpp = self # ugly hack to work with the BaseSender mixin :/ 

307 self.default_ns = "jabber:component:accept" 

308 super().__init__( 

309 config.JID, 

310 config.SECRET, 

311 config.SERVER, 

312 config.PORT, 

313 plugin_whitelist=SLIXMPP_PLUGINS, 

314 plugin_config={ 

315 "xep_0077": { 

316 "form_fields": None, 

317 "form_instructions": self.REGISTRATION_INSTRUCTIONS, 

318 "enable_subscription": self.REGISTRATION_TYPE 

319 == RegistrationType.SINGLE_STEP_FORM, 

320 }, 

321 "xep_0100": { 

322 "component_name": self.COMPONENT_NAME, 

323 "type": self.COMPONENT_TYPE, 

324 }, 

325 "xep_0184": { 

326 "auto_ack": False, 

327 "auto_request": False, 

328 }, 

329 "xep_0363": { 

330 "upload_service": config.UPLOAD_SERVICE, 

331 }, 

332 }, 

333 fix_error_ns=True, 

334 ) 

335 self.loop.set_exception_handler(self.__exception_handler) 

336 self.loop.create_task(self.__set_http()) 

337 self.has_crashed: bool = False 

338 self.use_origin_id = False 

339 

340 if config.USER_JID_VALIDATOR is None: 

341 config.USER_JID_VALIDATOR = f".*@{self.infer_real_domain()}" 

342 log.info( 

343 "No USER_JID_VALIDATOR was set, using '%s'.", 

344 config.USER_JID_VALIDATOR, 

345 ) 

346 self.jid_validator: re.Pattern = re.compile(config.USER_JID_VALIDATOR) 

347 self.qr_pending_registrations = dict[str, asyncio.Future[Optional[dict]]]() 

348 

349 self.register_plugins() 

350 self.__setup_legacy_module_subclasses() 

351 

352 self.get_session_from_stanza: Callable[ 

353 [Union[Message, Presence, Iq]], BaseSession 

354 ] = self._session_cls.from_stanza 

355 self.get_session_from_user: Callable[[GatewayUser], BaseSession] = ( 

356 self._session_cls.from_user 

357 ) 

358 

359 self.__register_slixmpp_events() 

360 self.__register_slixmpp_api() 

361 self.roster.set_backend(RosterBackend(self)) 

362 

363 self.register_plugin("pubsub", {"component_name": self.COMPONENT_NAME}) 

364 self.pubsub: PubSubComponent = self["pubsub"] 

365 self.delivery_receipt: DeliveryReceipt = DeliveryReceipt(self) 

366 

367 # with this we receive user avatar updates 

368 self.plugin["xep_0030"].add_feature("urn:xmpp:avatar:metadata+notify") 

369 

370 self.plugin["xep_0030"].add_feature("urn:xmpp:chat-markers:0") 

371 

372 if self.GROUPS: 

373 self.plugin["xep_0030"].add_feature("http://jabber.org/protocol/muc") 

374 self.plugin["xep_0030"].add_feature("urn:xmpp:mam:2") 

375 self.plugin["xep_0030"].add_feature("urn:xmpp:mam:2#extended") 

376 self.plugin["xep_0030"].add_feature(self.plugin["xep_0421"].namespace) 

377 self.plugin["xep_0030"].add_feature(self["xep_0317"].stanza.NS) 

378 self.plugin["xep_0030"].add_identity( 

379 category="conference", 

380 name=self.COMPONENT_NAME, 

381 itype="text", 

382 jid=self.boundjid, 

383 ) 

384 

385 # why does mypy need these type annotations? no idea 

386 self.__adhoc_handler: AdhocProvider = AdhocProvider(self) 

387 self.__chat_commands_handler: ChatCommandProvider = ChatCommandProvider(self) 

388 

389 self.__dispatcher = SessionDispatcher(self) 

390 

391 self.__register_commands() 

392 

393 MessageMixin.__init__(self) # ComponentXMPP does not call super().__init__() 

394 

395 def __setup_legacy_module_subclasses(self): 

396 from ..contact.roster import LegacyRoster 

397 from ..group.bookmarks import LegacyBookmarks 

398 from ..group.room import LegacyMUC, LegacyParticipant 

399 

400 session_cls: Type[BaseSession] = cast( 

401 Type[BaseSession], BaseSession.get_unique_subclass() 

402 ) 

403 contact_cls = LegacyContact.get_self_or_unique_subclass() 

404 muc_cls = LegacyMUC.get_self_or_unique_subclass() 

405 participant_cls = LegacyParticipant.get_self_or_unique_subclass() 

406 bookmarks_cls = LegacyBookmarks.get_self_or_unique_subclass() 

407 roster_cls = LegacyRoster.get_self_or_unique_subclass() 

408 

409 if contact_cls.REACTIONS_SINGLE_EMOJI: # type:ignore[attr-defined] 

410 form = Form() 

411 form["type"] = "result" 

412 form.add_field( 

413 "FORM_TYPE", "hidden", value="urn:xmpp:reactions:0:restrictions" 

414 ) 

415 form.add_field("max_reactions_per_user", value="1", type="number") 

416 form.add_field("scope", value="domain") 

417 self.plugin["xep_0128"].add_extended_info(data=form) 

418 

419 session_cls.xmpp = self # type:ignore[attr-defined] 

420 contact_cls.xmpp = self # type:ignore[attr-defined] 

421 muc_cls.xmpp = self # type:ignore[attr-defined] 

422 

423 self._session_cls = session_cls 

424 session_cls._bookmarks_cls = bookmarks_cls # type:ignore[misc,assignment] 

425 session_cls._roster_cls = roster_cls # type:ignore[misc,assignment] 

426 LegacyRoster._contact_cls = contact_cls # type:ignore[misc] 

427 LegacyBookmarks._muc_cls = muc_cls # type:ignore[misc] 

428 LegacyMUC._participant_cls = participant_cls # type:ignore[misc] 

429 

430 async def kill_session(self, jid: JID) -> None: 

431 await self._session_cls.kill_by_jid(jid) 

432 

433 async def __set_http(self) -> None: 

434 self.http = aiohttp.ClientSession() 

435 if getattr(self, "_test_mode", False): 

436 return 

437 avatar_cache.http = self.http 

438 

439 def __register_commands(self) -> None: 

440 for cls in Command.subclasses: 

441 if any(x is NotImplemented for x in [cls.CHAT_COMMAND, cls.NODE, cls.NAME]): 

442 log.debug("Not adding command '%s' because it looks abstract", cls) 

443 continue 

444 if cls is Exec: 

445 if config.DEV_MODE: 

446 log.warning(r"/!\ DEV MODE ENABLED /!\\") 

447 else: 

448 continue 

449 c = cls(self) 

450 log.debug("Registering %s", cls) 

451 self.__adhoc_handler.register(c) 

452 self.__chat_commands_handler.register(c) 

453 

454 def __exception_handler(self, loop: asyncio.AbstractEventLoop, context) -> None: 

455 """ 

456 Called when a task created by loop.create_task() raises an Exception 

457 

458 :param loop: 

459 :param context: 

460 :return: 

461 """ 

462 log.debug("Context in the exception handler: %s", context) 

463 exc = context.get("exception") 

464 if exc is None: 

465 log.debug("No exception in this context: %s", context) 

466 elif isinstance(exc, SystemExit): 

467 log.debug("SystemExit called in an asyncio task") 

468 else: 

469 log.error("Crash in an asyncio task: %s", context) 

470 log.exception("Crash in task", exc_info=exc) 

471 self.has_crashed = True 

472 loop.stop() 

473 

474 def __register_slixmpp_events(self) -> None: 

475 self.del_event_handler("presence_subscribe", self._handle_subscribe) 

476 self.del_event_handler("presence_unsubscribe", self._handle_unsubscribe) 

477 self.del_event_handler("presence_subscribed", self._handle_subscribed) 

478 self.del_event_handler("presence_unsubscribed", self._handle_unsubscribed) 

479 self.del_event_handler( 

480 "roster_subscription_request", self._handle_new_subscription 

481 ) 

482 self.del_event_handler("presence_probe", self._handle_probe) 

483 self.add_event_handler("session_start", self.__on_session_start) 

484 self.add_event_handler("privileges_advertised", self.__on_privileges_advertised) 

485 self.__upload_service_found = asyncio.Event() 

486 

487 def __register_slixmpp_api(self) -> None: 

488 def with_session(func, commit: bool = True): 

489 def wrapped(*a, **kw): 

490 with self.store.session() as orm: 

491 res = func(orm, *a, **kw) 

492 if commit: 

493 orm.commit() 

494 return res 

495 

496 return wrapped 

497 

498 self.plugin["xep_0231"].api.register( 

499 with_session(self.store.bob.get_bob, False), "get_bob" 

500 ) 

501 self.plugin["xep_0231"].api.register( 

502 with_session(self.store.bob.set_bob), "set_bob" 

503 ) 

504 self.plugin["xep_0231"].api.register( 

505 with_session(self.store.bob.del_bob), "del_bob" 

506 ) 

507 

508 @property # type: ignore 

509 def jid(self): # type:ignore[override] 

510 # Override to avoid slixmpp deprecation warnings. 

511 return self.boundjid 

512 

513 async def __on_session_start(self, event) -> None: 

514 log.debug("Gateway session start: %s", event) 

515 

516 await self.__setup_attachments() 

517 

518 # prevents XMPP clients from considering the gateway as an HTTP upload 

519 disco = self.plugin["xep_0030"] 

520 await disco.del_feature(feature="urn:xmpp:http:upload:0", jid=self.boundjid) 

521 await self.plugin["xep_0115"].update_caps(jid=self.boundjid) 

522 

523 if self.COMPONENT_AVATAR is not None: 

524 log.debug("Setting gateway avatar…") 

525 avatar = convert_avatar(self.COMPONENT_AVATAR, "!!---slidge---special---") 

526 assert avatar is not None 

527 try: 

528 cached_avatar = await avatar_cache.convert_or_get(avatar) 

529 except Exception as e: 

530 log.exception("Could not set the component avatar.", exc_info=e) 

531 cached_avatar = None 

532 else: 

533 assert cached_avatar is not None 

534 self.avatar = cached_avatar 

535 else: 

536 cached_avatar = None 

537 

538 with self.store.session() as orm: 

539 for user in orm.query(GatewayUser).all(): 

540 # TODO: before this, we should check if the user has removed us from their roster 

541 # while we were offline and trigger unregister from there. Presence probe does not seem 

542 # to work in this case, there must be another way. privileged entity could be used 

543 # as last resort. 

544 try: 

545 await self["xep_0100"].add_component_to_roster(user.jid) 

546 await self.__add_component_to_mds_whitelist(user.jid) 

547 except (IqError, IqTimeout) as e: 

548 # TODO: remove the user when this happens? or at least 

549 # this can happen when the user has unsubscribed from the XMPP server 

550 log.warning( 

551 "Error with user %s, not logging them automatically", 

552 user, 

553 exc_info=e, 

554 ) 

555 continue 

556 session = self._session_cls.from_user(user) 

557 session.create_task(self.login_wrap(session)) 

558 if cached_avatar is not None: 

559 await self.pubsub.broadcast_avatar( 

560 self.boundjid.bare, session.user_jid, cached_avatar 

561 ) 

562 

563 log.info("Slidge has successfully started") 

564 

565 async def __on_privileges_advertised(self, _) -> None: 

566 await self.__upload_service_found.wait() 

567 if self.xmpp["xep_0356"].granted_privileges[config.SERVER].iq.get( 

568 self.plugin["xep_0363"].stanza.Request.namespace 

569 ) in (IqPermission.SET, IqPermission.BOTH, IqPermission.GET): 

570 AttachmentMixin.PRIVILEGED_UPLOAD = True 

571 

572 async def __setup_attachments(self) -> None: 

573 if config.NO_UPLOAD_PATH: 

574 if config.NO_UPLOAD_URL_PREFIX is None: 

575 raise RuntimeError( 

576 "If you set NO_UPLOAD_PATH you must set NO_UPLOAD_URL_PREFIX too." 

577 ) 

578 elif not config.UPLOAD_SERVICE: 

579 info_iq = await self.xmpp.plugin["xep_0363"].find_upload_service( 

580 self.infer_real_domain() 

581 ) 

582 if info_iq is None: 

583 raise RuntimeError("Could not find upload service") 

584 log.info("Auto-discovered upload service: %s", info_iq["from"]) 

585 config.UPLOAD_SERVICE = info_iq["from"] 

586 self.__upload_service_found.set() 

587 

588 def infer_real_domain(self) -> JID: 

589 return JID(re.sub(r"^.*?\.", "", self.xmpp.boundjid.bare)) 

590 

591 async def __add_component_to_mds_whitelist(self, user_jid: JID) -> None: 

592 # Uses privileged entity to add ourselves to the whitelist of the PEP 

593 # MDS node so we receive MDS events 

594 iq_creation = Iq(sto=user_jid.bare, sfrom=user_jid, stype="set") 

595 iq_creation["pubsub"]["create"]["node"] = self["xep_0490"].stanza.NS 

596 

597 try: 

598 await self["xep_0356"].send_privileged_iq(iq_creation) 

599 except PermissionError: 

600 log.warning( 

601 "IQ privileges not granted for pubsub namespace, we cannot " 

602 "create the MDS node of %s", 

603 user_jid, 

604 ) 

605 except PrivilegedIqError as exc: 

606 nested = exc.nested_error() 

607 # conflict this means the node already exists, we can ignore that 

608 if nested is not None and nested.condition != "conflict": 

609 log.exception( 

610 "Could not create the MDS node of %s", user_jid, exc_info=exc 

611 ) 

612 except Exception as e: 

613 log.exception( 

614 "Error while trying to create to the MDS node of %s", 

615 user_jid, 

616 exc_info=e, 

617 ) 

618 

619 iq_affiliation = Iq(sto=user_jid.bare, sfrom=user_jid, stype="set") 

620 iq_affiliation["pubsub_owner"]["affiliations"]["node"] = self[ 

621 "xep_0490" 

622 ].stanza.NS 

623 

624 aff = OwnerAffiliation() 

625 aff["jid"] = self.boundjid.bare 

626 aff["affiliation"] = "member" 

627 iq_affiliation["pubsub_owner"]["affiliations"].append(aff) 

628 

629 try: 

630 await self["xep_0356"].send_privileged_iq(iq_affiliation) 

631 except PermissionError: 

632 log.warning( 

633 "IQ privileges not granted for pubsub#owner namespace, we cannot " 

634 "listen to the MDS events of %s", 

635 user_jid, 

636 ) 

637 except Exception as e: 

638 log.exception( 

639 "Error while trying to subscribe to the MDS node of %s", 

640 user_jid, 

641 exc_info=e, 

642 ) 

643 

644 @timeit 

645 async def login_wrap(self, session: "BaseSession") -> str: 

646 session.send_gateway_status("Logging in…", show="dnd") 

647 session.is_logging_in = True 

648 try: 

649 status = await session.login() 

650 except Exception as e: 

651 log.warning("Login problem for %s", session.user_jid, exc_info=e) 

652 session.send_gateway_status(f"Could not login: {e}", show="busy") 

653 msg = ( 

654 "You are not connected to this gateway! " 

655 f"Maybe this message will tell you why: {e}" 

656 ) 

657 session.send_gateway_message(msg) 

658 session.logged = False 

659 session.send_gateway_status("Login failed", show="dnd") 

660 return msg 

661 

662 log.info("Login success for %s", session.user_jid) 

663 session.logged = True 

664 session.send_gateway_status("Syncing contacts…", show="dnd") 

665 with self.store.session() as orm: 

666 await session.contacts._fill(orm) 

667 if not (r := session.contacts.ready).done(): 

668 r.set_result(True) 

669 if self.GROUPS: 

670 session.send_gateway_status("Syncing groups…", show="dnd") 

671 await session.bookmarks.fill() 

672 if not (r := session.bookmarks.ready).done(): 

673 r.set_result(True) 

674 self.send_presence(pto=session.user.jid.bare, ptype="probe") 

675 if status is None: 

676 status = "Logged in" 

677 session.send_gateway_status(status, show="chat") 

678 

679 if session.user.preferences.get("sync_avatar", False): 

680 session.create_task(self.fetch_user_avatar(session)) 

681 else: 

682 with self.store.session(expire_on_commit=False) as orm: 

683 session.user.avatar_hash = None 

684 orm.add(session.user) 

685 orm.commit() 

686 return status 

687 

688 async def fetch_user_avatar(self, session: BaseSession) -> None: 

689 try: 

690 iq = await self.xmpp.plugin["xep_0060"].get_items( 

691 session.user_jid.bare, 

692 self.xmpp.plugin["xep_0084"].stanza.MetaData.namespace, 

693 ifrom=self.boundjid.bare, 

694 ) 

695 except IqTimeout: 

696 self.log.warning("Iq timeout trying to fetch user avatar") 

697 return 

698 except IqError as e: 

699 self.log.debug("Iq error when trying to fetch user avatar: %s", e) 

700 if e.condition == "item-not-found": 

701 try: 

702 await session.on_avatar(None, None, None, None, None) 

703 except NotImplementedError: 

704 pass 

705 else: 

706 with self.store.session(expire_on_commit=False) as orm: 

707 session.user.avatar_hash = None 

708 orm.add(session.user) 

709 orm.commit() 

710 return 

711 await self.__dispatcher.on_avatar_metadata_info( 

712 session, iq["pubsub"]["items"]["item"]["avatar_metadata"]["info"] 

713 ) 

714 

715 def _send( 

716 self, stanza: MessageOrPresenceTypeVar, **send_kwargs 

717 ) -> MessageOrPresenceTypeVar: 

718 stanza.set_from(self.boundjid.bare) 

719 if mto := send_kwargs.get("mto"): 

720 stanza.set_to(mto) 

721 stanza.send() 

722 return stanza 

723 

724 def raise_if_not_allowed_jid(self, jid: JID): 

725 if not self.jid_validator.match(jid.bare): 

726 raise XMPPError( 

727 condition="not-allowed", 

728 text="Your account is not allowed to use this gateway. " 

729 "The admin controls that with the USER_JID_VALIDATOR option.", 

730 ) 

731 

732 def send_raw(self, data: Union[str, bytes]): 

733 # overridden from XMLStream to strip base64-encoded data from the logs 

734 # to make them more readable. 

735 if log.isEnabledFor(level=logging.DEBUG): 

736 if isinstance(data, str): 

737 stripped = copy(data) 

738 else: 

739 stripped = data.decode("utf-8") 

740 # there is probably a way to do that in a single RE, 

741 # but since it's only for debugging, the perf penalty 

742 # does not matter much 

743 for el in LOG_STRIP_ELEMENTS: 

744 stripped = re.sub( 

745 f"(<{el}.*?>)(.*)(</{el}>)", 

746 "\1[STRIPPED]\3", 

747 stripped, 

748 flags=re.DOTALL | re.IGNORECASE, 

749 ) 

750 log.debug("SEND: %s", stripped) 

751 if not self.transport: 

752 raise NotConnectedError() 

753 if isinstance(data, str): 

754 data = data.encode("utf-8") 

755 self.transport.write(data) 

756 

757 def get_session_from_jid(self, j: JID): 

758 try: 

759 return self._session_cls.from_jid(j) 

760 except XMPPError: 

761 pass 

762 

763 def exception(self, exception: Exception) -> None: 

764 # """ 

765 # Called when a task created by slixmpp's internal (eg, on slix events) raises an Exception. 

766 # 

767 # Stop the event loop and exit on unhandled exception. 

768 # 

769 # The default :class:`slixmpp.basexmpp.BaseXMPP` behaviour is just to 

770 # log the exception, but we want to avoid undefined behaviour. 

771 # 

772 # :param exception: An unhandled :class:`Exception` object. 

773 # """ 

774 if isinstance(exception, IqError): 

775 iq = exception.iq 

776 log.error("%s: %s", iq["error"]["condition"], iq["error"]["text"]) 

777 log.warning("You should catch IqError exceptions") 

778 elif isinstance(exception, IqTimeout): 

779 iq = exception.iq 

780 log.error("Request timed out: %s", iq) 

781 log.warning("You should catch IqTimeout exceptions") 

782 elif isinstance(exception, SyntaxError): 

783 # Hide stream parsing errors that occur when the 

784 # stream is disconnected (they've been handled, we 

785 # don't need to make a mess in the logs). 

786 pass 

787 else: 

788 if exception: 

789 log.exception(exception) 

790 self.loop.stop() 

791 exit(1) 

792 

793 async def make_registration_form( 

794 self, _jid: JID, _node: str, _ifrom: JID, iq: Iq 

795 ) -> Form: 

796 self.raise_if_not_allowed_jid(iq.get_from()) 

797 reg = iq["register"] 

798 with self.store.session() as orm: 

799 user = ( 

800 orm.query(GatewayUser).filter_by(jid=iq.get_from().bare).one_or_none() 

801 ) 

802 log.debug("User found: %s", user) 

803 

804 form = reg["form"] 

805 form.add_field( 

806 "FORM_TYPE", 

807 ftype="hidden", 

808 value="jabber:iq:register", 

809 ) 

810 form["title"] = f"Registration to '{self.COMPONENT_NAME}'" 

811 form["instructions"] = self.REGISTRATION_INSTRUCTIONS 

812 

813 if user is not None: 

814 reg["registered"] = False 

815 form.add_field( 

816 "remove", 

817 label="Remove my registration", 

818 required=True, 

819 ftype="boolean", 

820 value=False, 

821 ) 

822 

823 for field in self.REGISTRATION_FIELDS: 

824 if field.var in reg.interfaces: 

825 val = None if user is None else user.get(field.var) 

826 if val is None: 

827 reg.add_field(field.var) 

828 else: 

829 reg[field.var] = val 

830 

831 reg["instructions"] = self.REGISTRATION_INSTRUCTIONS 

832