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

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 http: aiohttp.ClientSession 

289 avatar: CachedAvatar | None = None 

290 

291 def __init__(self) -> None: 

292 if config.COMPONENT_NAME: 

293 self.COMPONENT_NAME = config.COMPONENT_NAME 

294 if config.WELCOME_MESSAGE: 

295 self.WELCOME_MESSAGE = config.WELCOME_MESSAGE 

296 self.log = log 

297 self.datetime_started = datetime.now() 

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

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

300 super().__init__( 

301 config.JID, 

302 config.SECRET, 

303 config.SERVER, 

304 config.PORT, 

305 plugin_whitelist=SLIXMPP_PLUGINS, 

306 plugin_config={ 

307 "xep_0077": { 

308 "form_fields": None, 

309 "form_instructions": self.REGISTRATION_INSTRUCTIONS, 

310 "enable_subscription": self.REGISTRATION_TYPE 

311 == RegistrationType.SINGLE_STEP_FORM, 

312 }, 

313 "xep_0100": { 

314 "component_name": self.COMPONENT_NAME, 

315 "type": self.COMPONENT_TYPE, 

316 }, 

317 "xep_0184": { 

318 "auto_ack": False, 

319 "auto_request": False, 

320 }, 

321 "xep_0363": { 

322 "upload_service": config.UPLOAD_SERVICE, 

323 }, 

324 }, 

325 fix_error_ns=True, 

326 ) 

327 self.loop.set_exception_handler(self.__exception_handler) 

328 self.loop.create_task(self.__set_http()) 

329 self.has_crashed: bool = False 

330 self.use_origin_id = False 

331 

332 if config.USER_JID_VALIDATOR is None: 

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

334 log.info( 

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

336 config.USER_JID_VALIDATOR, 

337 ) 

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

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

340 

341 self.register_plugins() 

342 self.__setup_legacy_module_subclasses() 

343 

344 self.get_session_from_stanza: Callable[ 

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

346 ] = self._session_cls.from_stanza 

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

348 self._session_cls.from_user 

349 ) 

350 

351 self.__register_slixmpp_events() 

352 self.__register_slixmpp_api() 

353 self.roster.set_backend(RosterBackend(self)) 

354 

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

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

357 self.delivery_receipt: DeliveryReceipt = DeliveryReceipt(self) 

358 

359 # with this we receive user avatar updates 

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

361 

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

363 

364 if self.GROUPS: 

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

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

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

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

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

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

371 category="conference", 

372 name=self.COMPONENT_NAME, 

373 itype="text", 

374 jid=self.boundjid, 

375 ) 

376 

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

378 self.__adhoc_handler: AdhocProvider = AdhocProvider(self) 

379 self.__chat_commands_handler: ChatCommandProvider = ChatCommandProvider(self) 

380 

381 self.__dispatcher = SessionDispatcher(self) 

382 

383 self.__register_commands() 

384 

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

386 

387 def __setup_legacy_module_subclasses(self): 

388 from ..contact.roster import LegacyRoster 

389 from ..group.bookmarks import LegacyBookmarks 

390 from ..group.room import LegacyMUC, LegacyParticipant 

391 

392 session_cls: Type[BaseSession] = cast( 

393 Type[BaseSession], BaseSession.get_unique_subclass() 

394 ) 

395 contact_cls = LegacyContact.get_self_or_unique_subclass() 

396 muc_cls = LegacyMUC.get_self_or_unique_subclass() 

397 participant_cls = LegacyParticipant.get_self_or_unique_subclass() 

398 bookmarks_cls = LegacyBookmarks.get_self_or_unique_subclass() 

399 roster_cls = LegacyRoster.get_self_or_unique_subclass() 

400 

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

402 form = Form() 

403 form["type"] = "result" 

404 form.add_field( 

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

406 ) 

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

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

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

410 

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

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

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

414 

415 self._session_cls = session_cls 

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

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

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

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

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

421 

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

423 await self._session_cls.kill_by_jid(jid) 

424 

425 async def __set_http(self) -> None: 

426 self.http = aiohttp.ClientSession() 

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

428 return 

429 avatar_cache.http = self.http 

430 

431 def __register_commands(self) -> None: 

432 for cls in Command.subclasses: 

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

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

435 continue 

436 if cls is Exec: 

437 if config.DEV_MODE: 

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

439 else: 

440 continue 

441 c = cls(self) 

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

443 self.__adhoc_handler.register(c) 

444 self.__chat_commands_handler.register(c) 

445 

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

447 """ 

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

449 

450 :param loop: 

451 :param context: 

452 :return: 

453 """ 

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

455 exc = context.get("exception") 

456 if exc is None: 

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

458 elif isinstance(exc, SystemExit): 

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

460 else: 

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

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

463 self.has_crashed = True 

464 loop.stop() 

465 

466 def __register_slixmpp_events(self) -> None: 

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

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

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

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

471 self.del_event_handler( 

472 "roster_subscription_request", self._handle_new_subscription 

473 ) 

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

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

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

477 self.__upload_service_found = asyncio.Event() 

478 

479 def __register_slixmpp_api(self) -> None: 

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

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

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

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

484 if commit: 

485 orm.commit() 

486 return res 

487 

488 return wrapped 

489 

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

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

492 ) 

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

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

495 ) 

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

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

498 ) 

499 

500 @property # type: ignore 

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

502 # Override to avoid slixmpp deprecation warnings. 

503 return self.boundjid 

504 

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

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

507 

508 await self.__setup_attachments() 

509 

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

511 disco = self.plugin["xep_0030"] 

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

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

514 

515 if self.COMPONENT_AVATAR is not None: 

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

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

518 assert avatar is not None 

519 try: 

520 cached_avatar = await avatar_cache.convert_or_get(avatar) 

521 except Exception as e: 

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

523 cached_avatar = None 

524 else: 

525 assert cached_avatar is not None 

526 self.avatar = cached_avatar 

527 else: 

528 cached_avatar = None 

529 

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

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

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

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

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

535 # as last resort. 

536 try: 

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

538 await self.__add_component_to_mds_whitelist(user.jid) 

539 except (IqError, IqTimeout) as e: 

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

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

542 log.warning( 

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

544 user, 

545 exc_info=e, 

546 ) 

547 continue 

548 session = self._session_cls.from_user(user) 

549 session.create_task(self.login_wrap(session)) 

550 if cached_avatar is not None: 

551 await self.pubsub.broadcast_avatar( 

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

553 ) 

554 

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

556 

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

558 await self.__upload_service_found.wait() 

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

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

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

562 AttachmentMixin.PRIVILEGED_UPLOAD = True 

563 

564 async def __setup_attachments(self) -> None: 

565 if config.NO_UPLOAD_PATH: 

566 if config.NO_UPLOAD_URL_PREFIX is None: 

567 raise RuntimeError( 

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

569 ) 

570 elif not config.UPLOAD_SERVICE: 

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

572 self.infer_real_domain() 

573 ) 

574 if info_iq is None: 

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

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

577 config.UPLOAD_SERVICE = info_iq["from"] 

578 self.__upload_service_found.set() 

579 

580 def infer_real_domain(self) -> JID: 

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

582 

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

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

585 # MDS node so we receive MDS events 

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

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

588 

589 try: 

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

591 except PermissionError: 

592 log.warning( 

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

594 "create the MDS node of %s", 

595 user_jid, 

596 ) 

597 except PrivilegedIqError as exc: 

598 nested = exc.nested_error() 

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

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

601 log.exception( 

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

603 ) 

604 except Exception as e: 

605 log.exception( 

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

607 user_jid, 

608 exc_info=e, 

609 ) 

610 

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

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

613 "xep_0490" 

614 ].stanza.NS 

615 

616 aff = OwnerAffiliation() 

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

618 aff["affiliation"] = "member" 

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

620 

621 try: 

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

623 except PermissionError: 

624 log.warning( 

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

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

627 user_jid, 

628 ) 

629 except Exception as e: 

630 log.exception( 

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

632 user_jid, 

633 exc_info=e, 

634 ) 

635 

636 @timeit 

637 async def login_wrap(self, session: "BaseSession") -> None: 

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

639 session.is_logging_in = True 

640 try: 

641 status = await session.login() 

642 except Exception as e: 

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

644 log.exception(e) 

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

646 session.send_gateway_message( 

647 "You are not connected to this gateway! " 

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

649 ) 

650 session.logged = False 

651 return 

652 

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

654 session.logged = True 

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

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

657 await session.contacts._fill(orm) 

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

659 r.set_result(True) 

660 if self.GROUPS: 

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

662 await session.bookmarks.fill() 

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

664 r.set_result(True) 

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

666 if status is None: 

667 session.send_gateway_status("Logged in", show="chat") 

668 else: 

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

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

671 session.create_task(self.fetch_user_avatar(session)) 

672 else: 

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

674 session.user.avatar_hash = None 

675 orm.add(session.user) 

676 orm.commit() 

677 

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

679 try: 

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

681 session.user_jid.bare, 

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

683 ifrom=self.boundjid.bare, 

684 ) 

685 except IqTimeout: 

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

687 return 

688 except IqError as e: 

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

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

691 try: 

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

693 except NotImplementedError: 

694 pass 

695 else: 

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

697 session.user.avatar_hash = None 

698 orm.add(session.user) 

699 orm.commit() 

700 return 

701 await self.__dispatcher.on_avatar_metadata_info( 

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

703 ) 

704 

705 def _send( 

706 self, stanza: MessageOrPresenceTypeVar, **send_kwargs 

707 ) -> MessageOrPresenceTypeVar: 

708 stanza.set_from(self.boundjid.bare) 

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

710 stanza.set_to(mto) 

711 stanza.send() 

712 return stanza 

713 

714 def raise_if_not_allowed_jid(self, jid: JID): 

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

716 raise XMPPError( 

717 condition="not-allowed", 

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

719 ) 

720 

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

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

723 # to make them more readable. 

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

725 if isinstance(data, str): 

726 stripped = copy(data) 

727 else: 

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

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

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

731 # does not matter much 

732 for el in LOG_STRIP_ELEMENTS: 

733 stripped = re.sub( 

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

735 "\1[STRIPPED]\3", 

736 stripped, 

737 flags=re.DOTALL | re.IGNORECASE, 

738 ) 

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

740 if not self.transport: 

741 raise NotConnectedError() 

742 if isinstance(data, str): 

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

744 self.transport.write(data) 

745 

746 def get_session_from_jid(self, j: JID): 

747 try: 

748 return self._session_cls.from_jid(j) 

749 except XMPPError: 

750 pass 

751 

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

753 # """ 

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

755 # 

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

757 # 

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

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

760 # 

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

762 # """ 

763 if isinstance(exception, IqError): 

764 iq = exception.iq 

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

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

767 elif isinstance(exception, IqTimeout): 

768 iq = exception.iq 

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

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

771 elif isinstance(exception, SyntaxError): 

772 # Hide stream parsing errors that occur when the 

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

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

775 pass 

776 else: 

777 if exception: 

778 log.exception(exception) 

779 self.loop.stop() 

780 exit(1) 

781 

782 def re_login(self, session: "BaseSession") -> None: 

783 async def w() -> None: 

784 session.cancel_all_tasks() 

785 try: 

786 await session.logout() 

787 except NotImplementedError: 

788 pass 

789 await self.login_wrap(session) 

790 

791 session.create_task(w()) 

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: