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

1""" 

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

3""" 

4 

5import abc 

6import asyncio 

7import contextlib 

8import logging 

9import re 

10import tempfile 

11from collections.abc import Callable, Sequence 

12from copy import copy 

13from datetime import datetime 

14from pathlib import Path 

15from typing import Any, ClassVar, Concatenate, Generic, ParamSpec, TypeVar, cast 

16 

17import aiohttp 

18import qrcode 

19from slixmpp import JID, ComponentXMPP, Iq 

20from slixmpp.exceptions import IqError, IqTimeout, XMPPError 

21from slixmpp.plugins.xep_0004.stanza.form import Form 

22from slixmpp.plugins.xep_0060.stanza import OwnerAffiliation 

23from slixmpp.plugins.xep_0356.privilege import PrivilegedIqError 

24from slixmpp.types import MessageTypes 

25from slixmpp.xmlstream.xmlstream import NotConnectedError 

26from sqlalchemy.orm import Session as OrmSession 

27 

28import slidge.command.categories 

29from slidge.command.adhoc import AdhocProvider 

30from slidge.command.admin import Exec 

31from slidge.command.base import Command, FormField 

32from slidge.command.chat_command import ChatCommandProvider 

33from slidge.command.register import RegistrationType 

34from slidge.contact import LegacyContact 

35from slidge.core import config 

36from slidge.core.dispatcher.session_dispatcher import SessionDispatcher 

37from slidge.core.mixins.avatar import convert_avatar 

38from slidge.core.mixins.message import MessageMixin 

39from slidge.core.pubsub import PubSubComponent 

40from slidge.db import GatewayUser, SlidgeStore 

41from slidge.db.avatar import CachedAvatar, avatar_cache 

42from slidge.db.meta import JSONSerializable 

43from slidge.slixfix.delivery_receipt import DeliveryReceipt 

44from slidge.slixfix.roster import RosterBackend 

45from slidge.util import SubclassableOnce 

46from slidge.util.types import AnyGateway, Avatar, MessageOrPresenceTypeVar, SessionType 

47 

48T = TypeVar("T") 

49P = ParamSpec("P") 

50 

51 

52class BaseGateway( 

53 ComponentXMPP, 

54 MessageMixin, 

55 SubclassableOnce, 

56 Generic[SessionType], 

57 abc.ABC, 

58): 

59 """ 

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

61 

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

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

64 ``.xmpp`` attribute. 

65 

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

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

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

69 

70 Abstract methods related to the registration process must be overriden 

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

72 

73 - :meth:`.validate` 

74 - :meth:`.validate_two_factor_code` 

75 - :meth:`.get_qr_text` 

76 - :meth:`.confirm_qr` 

77 

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

79 :attr:`REGISTRATION_TYPE`. 

80 

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

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

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

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

85 `mto` parameter. 

86 

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

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

89 

90 .. code-block:: python 

91 

92 self.send_presence( 

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

94 pto="someonwelse@anotherexample.com", 

95 ) 

96 

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

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

99 as sending messages, or displaying a custom status. 

100 

101 """ 

102 

103 COMPONENT_NAME: str = NotImplemented 

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

105 COMPONENT_TYPE: str = "" 

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

107 COMPONENT_AVATAR: Avatar | Path | str | None = None 

108 """ 

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

110 """ 

111 

112 REGISTRATION_FIELDS: Sequence[FormField] = [ 

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

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

115 ] 

116 """ 

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

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

119 """ 

120 REGISTRATION_INSTRUCTIONS: str = "Enter your credentials" 

121 """ 

122 The text presented to a user who wants to register (or modify) their 

123 :term:`legacy <Legacy>` account configuration. 

124 """ 

125 REGISTRATION_TYPE: RegistrationType = RegistrationType.SINGLE_STEP_FORM 

126 """ 

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

128 login to the :term:`legacy network <Legacy Network>`. 

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

130 once per user (unless they unregister). 

131 

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

133 presented to the user. 

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

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

136 """ 

137 

138 REGISTRATION_2FA_TITLE = "Enter your 2FA code" 

139 REGISTRATION_2FA_INSTRUCTIONS = ( 

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

141 ) 

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

143 

144 PREFERENCES: ClassVar[list[FormField]] = [ 

145 FormField( 

146 var="sync_presence", 

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

148 value="true", 

149 required=True, 

150 type="boolean", 

151 ), 

152 FormField( 

153 var="sync_avatar", 

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

155 value="true", 

156 required=True, 

157 type="boolean", 

158 ), 

159 FormField( 

160 var="always_invite_when_adding_bookmarks", 

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

162 value="true", 

163 required=True, 

164 type="boolean", 

165 ), 

166 FormField( 

167 var="last_seen_fallback", 

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

169 value="true", 

170 required=True, 

171 type="boolean", 

172 ), 

173 FormField( 

174 var="roster_push", 

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

176 value="true", 

177 required=True, 

178 type="boolean", 

179 ), 

180 FormField( 

181 var="reaction_fallback", 

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

183 value="false", 

184 required=True, 

185 type="boolean", 

186 ), 

187 ] 

188 

189 ROSTER_GROUP: str = "slidge" 

190 """ 

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

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

193 """ 

194 WELCOME_MESSAGE = ( 

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

196 "or just start messaging away!" 

197 ) 

198 """ 

199 A welcome message displayed to users on registration. 

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

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

202 incoming messages from components. 

203 """ 

204 

205 SEARCH_FIELDS: Sequence[FormField] = [ 

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

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

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

209 ] 

210 """ 

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

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

213 their usernames, eg their phone number. 

214 

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

216 (restricted to registered users). 

217 

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

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

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

221 """ 

222 SEARCH_TITLE: str = "Search for legacy contacts" 

223 """ 

224 Title of the search form. 

225 """ 

226 SEARCH_INSTRUCTIONS: str = "" 

227 """ 

228 Instructions of the search form. 

229 """ 

230 

231 MARK_ALL_MESSAGES = False 

232 """ 

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

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

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

236 """ 

237 

238 PROPER_RECEIPTS = False 

239 """ 

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

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

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

243 """ 

244 

245 GROUPS = False 

246 """ 

247 This must be set to True if this gateway supports groups. 

248 """ 

249 SPACES = False 

250 """ 

251 This must be set to True if this gateway supports spaces, cf :xep:`0503`. 

252 """ 

253 

254 mtype: MessageTypes = "chat" 

255 is_group = False 

256 _can_send_carbon = False 

257 store: SlidgeStore 

258 _session_cls: type[SessionType] 

259 

260 http: aiohttp.ClientSession 

261 avatar: CachedAvatar | None = None 

262 

263 def __init__(self) -> None: 

264 if config.COMPONENT_NAME: 

265 self.COMPONENT_NAME = config.COMPONENT_NAME 

266 if config.WELCOME_MESSAGE: 

267 self.WELCOME_MESSAGE = config.WELCOME_MESSAGE 

268 self.log = log 

269 self.datetime_started = datetime.now() 

270 # FIXME: ugly hack to work with the BaseSender mixin :/ 

271 self.xmpp = cast("AnyGateway", self) 

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

273 super().__init__( 

274 config.JID, 

275 config.SECRET, 

276 config.SERVER, 

277 config.PORT, 

278 plugin_whitelist=SLIXMPP_PLUGINS, 

279 plugin_config={ 

280 "xep_0077": { 

281 "form_fields": None, 

282 "form_instructions": self.REGISTRATION_INSTRUCTIONS, 

283 "enable_subscription": self.REGISTRATION_TYPE 

284 == RegistrationType.SINGLE_STEP_FORM, 

285 }, 

286 "xep_0100": { 

287 "component_name": self.COMPONENT_NAME, 

288 "type": self.COMPONENT_TYPE, 

289 }, 

290 "xep_0184": { 

291 "auto_ack": False, 

292 "auto_request": False, 

293 }, 

294 "xep_0363": { 

295 "upload_service": config.UPLOAD_SERVICE, 

296 }, 

297 }, 

298 fix_error_ns=True, 

299 ) 

300 self.loop.set_exception_handler(self.__exception_handler) 

301 self.loop.create_task(self.__set_http()) 

302 self.has_crashed: bool = False 

303 self.use_origin_id = False 

304 

305 if config.USER_JID_VALIDATOR is None: 

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

307 log.info( 

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

309 config.USER_JID_VALIDATOR, 

310 ) 

311 self.jid_validator: re.Pattern[str] = re.compile(config.USER_JID_VALIDATOR) 

312 self.qr_pending_registrations = dict[ 

313 str, asyncio.Future[JSONSerializable | None] 

314 ]() 

315 

316 self.register_plugins() 

317 self.__setup_legacy_module_subclasses() 

318 

319 self.get_session_from_stanza = self._session_cls.from_stanza 

320 self.get_session_from_user = self._session_cls.from_user 

321 

322 self.__register_slixmpp_events() 

323 self.__register_slixmpp_api() 

324 self.roster.set_backend(RosterBackend(self)) 

325 

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

327 self.pubsub: PubSubComponent = self.plugin["pubsub"] # type:ignore[typeddict-item] 

328 self.delivery_receipt = DeliveryReceipt(self) 

329 

330 # with this we receive user avatar updates 

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

332 

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

334 

335 if self.GROUPS: 

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

337 self.plugin["xep_0030"].add_feature(self.plugin["xep_0463"].stanza.NS) 

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

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

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

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

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

343 category="conference", 

344 name=self.COMPONENT_NAME, 

345 itype="text", 

346 jid=self.boundjid, 

347 ) 

348 if self.SPACES: 

349 self.plugin["xep_0030"].add_feature("urn:xmpp:spaces:0") 

350 

351 self.__adhoc_handler = AdhocProvider(self) 

352 self.__chat_commands_handler = ChatCommandProvider(self) 

353 

354 self.__dispatcher = SessionDispatcher(self) 

355 

356 self.__register_commands() 

357 

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

359 

360 def __setup_legacy_module_subclasses(self) -> None: 

361 from ..contact.roster import LegacyRoster 

362 from ..group.bookmarks import LegacyBookmarks 

363 from ..group.participant import LegacyParticipant 

364 from ..group.room import LegacyMUC 

365 from .session import BaseSession 

366 

367 self._session_cls = BaseSession.get_unique_subclass() # type:ignore 

368 contact_cls = LegacyContact.get_self_or_unique_subclass() 

369 muc_cls = LegacyMUC.get_self_or_unique_subclass() 

370 participant_cls = LegacyParticipant.get_self_or_unique_subclass() 

371 bookmarks_cls = LegacyBookmarks.get_self_or_unique_subclass() 

372 roster_cls = LegacyRoster.get_self_or_unique_subclass() 

373 

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

375 form = Form() 

376 form["type"] = "result" 

377 form.add_field( 

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

379 ) 

380 form.add_field("max_reactions_per_user", value="1", type="text-single") 

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

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

383 

384 self._session_cls.xmpp = self 

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

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

387 

388 self._session_cls._bookmarks_cls = bookmarks_cls # type:ignore[assignment] 

389 self._session_cls._roster_cls = roster_cls # type:ignore[assignment] 

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

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

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

393 

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

395 await self._session_cls.kill_by_jid(jid) 

396 

397 async def __set_http(self) -> None: 

398 self.http = aiohttp.ClientSession() 

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

400 return 

401 avatar_cache.http = self.http 

402 

403 def __register_commands(self) -> None: 

404 for cls in Command.subclasses: # type:ignore[misc] 

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

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

407 continue 

408 if cls is Exec: 

409 if config.DEV_MODE: 

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

411 else: 

412 continue 

413 if cls.CATEGORY == slidge.command.categories.GROUPS and not self.GROUPS: 

414 continue 

415 if cls.CATEGORY == slidge.command.categories.SPACES and not self.SPACES: 

416 continue 

417 c = cls(cast(AnyGateway, self)) 

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

419 self.__adhoc_handler.register(c) 

420 self.__chat_commands_handler.register(c) 

421 

422 def __exception_handler( 

423 self, loop: asyncio.AbstractEventLoop, context: dict[Any, Any] 

424 ) -> None: 

425 """ 

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

427 

428 :param loop: 

429 :param context: 

430 :return: 

431 """ 

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

433 exc = context.get("exception") 

434 if exc is None: 

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

436 elif isinstance(exc, SystemExit): 

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

438 else: 

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

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

441 self.has_crashed = True 

442 loop.stop() 

443 

444 def __register_slixmpp_events(self) -> None: 

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

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

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

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

449 self.del_event_handler( 

450 "roster_subscription_request", self._handle_new_subscription 

451 ) 

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

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

454 

455 def __register_slixmpp_api(self) -> None: 

456 def with_session( 

457 func: Callable[Concatenate[OrmSession, P], T], commit: bool = True 

458 ) -> Callable[P, T]: 

459 def wrapped(*a: P.args, **kw: P.kwargs) -> T: 

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

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

462 if commit: 

463 orm.commit() 

464 return res 

465 

466 return wrapped 

467 

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

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

470 ) 

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

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

473 ) 

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

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

476 ) 

477 

478 @property # type:ignore[override] 

479 def jid(self) -> JID: 

480 # Override to avoid slixmpp deprecation warnings. 

481 return self.boundjid 

482 

483 @jid.setter 

484 def jid(self, jid: JID) -> None: 

485 raise RuntimeError 

486 

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

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

489 

490 await self.__setup_attachments() 

491 

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

493 disco = self.plugin["xep_0030"] 

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

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

496 

497 if self.COMPONENT_AVATAR is not None: 

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

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

500 assert avatar is not None 

501 try: 

502 cached_avatar = await avatar_cache.convert_or_get(avatar) 

503 except Exception as e: 

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

505 cached_avatar = None 

506 else: 

507 assert cached_avatar is not None 

508 self.avatar = cached_avatar 

509 else: 

510 cached_avatar = None 

511 

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

513 users = orm.query(GatewayUser).all() 

514 for user in users: 

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

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

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

518 # as last resort. 

519 try: 

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

521 await self.__add_component_to_mds_whitelist(user.jid) 

522 except (IqError, IqTimeout) as e: 

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

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

525 log.warning( 

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

527 user, 

528 exc_info=e, 

529 ) 

530 continue 

531 session = self._session_cls.from_user(user) 

532 session.create_task(self.login_wrap(session)) 

533 if cached_avatar is not None: 

534 await self.pubsub.broadcast_avatar( 

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

536 ) 

537 

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

539 

540 async def __setup_attachments(self) -> None: 

541 if config.NO_UPLOAD_PATH: 

542 if config.NO_UPLOAD_URL_PREFIX is None: 

543 raise RuntimeError( 

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

545 ) 

546 elif not config.UPLOAD_SERVICE: 

547 try: 

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

549 self.infer_real_domain() 

550 ) 

551 except XMPPError: 

552 info_iq = None 

553 log.exception( 

554 "The upload service could not be automatically determine. " 

555 "Attachments to XMPP will not work. " 

556 "Either specify 'upload-service' or 'no-upload-path' to fix that." 

557 ) 

558 if info_iq is None: 

559 if self.REGISTRATION_TYPE == RegistrationType.QRCODE: 

560 log.warning( 

561 "No method was configured for attachment and slidge " 

562 "could not automatically determine the JID of a usable upload service. " 

563 "Users likely won't be able to register since this network uses a " 

564 "QR-code based registration flow." 

565 ) 

566 if not config.USE_ATTACHMENT_ORIGINAL_URLS: 

567 log.warning( 

568 "Setting USE_ATTACHMENT_ORIGINAL_URLS to True since no method was configured " 

569 "for attachments and no upload service was found. NB: this does not work for all " 

570 "networks, especially for the E2EE attachments." 

571 ) 

572 config.USE_ATTACHMENT_ORIGINAL_URLS = True 

573 else: 

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

575 config.UPLOAD_SERVICE = info_iq["from"] 

576 

577 def infer_real_domain(self) -> JID: 

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

579 

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

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

582 # MDS node so we receive MDS events 

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

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

585 

586 try: 

587 await self.plugin["xep_0356"].send_privileged_iq(iq_creation) 

588 except PermissionError: 

589 log.warning( 

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

591 "create the MDS node of %s", 

592 user_jid, 

593 ) 

594 except PrivilegedIqError as exc: 

595 nested = exc.nested_error() 

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

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

598 log.exception( 

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

600 ) 

601 except Exception as e: 

602 log.exception( 

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

604 user_jid, 

605 exc_info=e, 

606 ) 

607 

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

609 iq_affiliation["pubsub_owner"]["affiliations"]["node"] = self.plugin[ 

610 "xep_0490" 

611 ].stanza.NS 

612 

613 aff = OwnerAffiliation() 

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

615 aff["affiliation"] = "member" 

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

617 

618 try: 

619 await self.plugin["xep_0356"].send_privileged_iq(iq_affiliation) 

620 except PermissionError: 

621 log.warning( 

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

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

624 user_jid, 

625 ) 

626 except Exception as e: 

627 log.exception( 

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

629 user_jid, 

630 exc_info=e, 

631 ) 

632 

633 async def login_wrap(self, session: SessionType) -> str: 

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

635 session.is_logging_in = True 

636 try: 

637 status = await session.login() 

638 except Exception as e: 

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

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

641 msg = ( 

642 "You are not connected to this gateway! " 

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

644 ) 

645 session.send_gateway_message(msg) 

646 session.logged = False 

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

648 return msg 

649 

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

651 session.logged = True 

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

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

654 await session.contacts._fill(orm) 

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

656 r.set_result(True) 

657 if self.GROUPS: 

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

659 await session.bookmarks.fill() 

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

661 r.set_result(True) 

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

663 if status is None: 

664 status = "Logged in" 

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

666 

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

668 session.create_task(self.fetch_user_avatar(session)) 

669 else: 

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

671 session.user.avatar_hash = None 

672 orm.add(session.user) 

673 orm.commit() 

674 return status 

675 

676 async def fetch_user_avatar(self, session: SessionType) -> None: 

677 try: 

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

679 session.user_jid.bare, 

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

681 ifrom=self.boundjid.bare, 

682 ) 

683 except IqTimeout: 

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

685 return 

686 except IqError as e: 

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

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

689 try: 

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

691 except NotImplementedError: 

692 pass 

693 else: 

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

695 session.user.avatar_hash = None 

696 orm.add(session.user) 

697 orm.commit() 

698 return 

699 await self.__dispatcher.on_avatar_metadata_info( 

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

701 ) 

702 

703 def _send( 

704 self, 

705 stanza: MessageOrPresenceTypeVar, 

706 **send_kwargs: Any, # noqa:ANN401 

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) -> None: 

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 "The admin controls that with the USER_JID_VALIDATOR option.", 

720 ) 

721 

722 def send_raw(self, data: str | bytes) -> None: 

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

724 # to make them more readable. 

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

726 stripped = copy(data) if isinstance(data, str) else data.decode("utf-8") 

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

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

729 # does not matter much 

730 for el in LOG_STRIP_ELEMENTS: 

731 stripped = re.sub( 

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

733 "\1[STRIPPED]\3", 

734 stripped, 

735 flags=re.DOTALL | re.IGNORECASE, 

736 ) 

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

738 if not self.transport: 

739 raise NotConnectedError() 

740 if isinstance(data, str): 

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

742 self.transport.write(data) 

743 

744 def get_session_from_jid(self, j: JID) -> SessionType | None: 

745 try: 

746 return self._session_cls.from_jid(j) 

747 except XMPPError: 

748 return None 

749 

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

751 # """ 

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

753 # 

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

755 # 

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

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

758 # 

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

760 # """ 

761 if isinstance(exception, IqError): 

762 iq = exception.iq 

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

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

765 elif isinstance(exception, IqTimeout): 

766 iq = exception.iq 

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

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

769 elif isinstance(exception, SyntaxError): 

770 # Hide stream parsing errors that occur when the 

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

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

773 pass 

774 else: 

775 if exception: 

776 log.exception(exception) 

777 self.loop.stop() 

778 exit(1) 

779 

780 async def make_registration_form( 

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

782 ) -> Iq: 

783 self.raise_if_not_allowed_jid(iq.get_from()) 

784 reg = iq["register"] 

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

786 user = ( 

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

788 ) 

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

790 

791 form = reg["form"] 

792 form.add_field( 

793 "FORM_TYPE", 

794 ftype="hidden", 

795 value="jabber:iq:register", 

796 ) 

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

798 form["instructions"] = self.REGISTRATION_INSTRUCTIONS 

799 

800 if user is not None: 

801 reg["registered"] = False 

802 form.add_field( 

803 "remove", 

804 label="Remove my registration", 

805 required=True, 

806 ftype="boolean", 

807 value=False, 

808 ) 

809 

810 for field in self.REGISTRATION_FIELDS: 

811 if field.var in reg.interfaces: 

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

813 if val is None: 

814 reg.add_field(field.var) 

815 else: 

816 reg[field.var] = val 

817 

818 reg["instructions"] = self.REGISTRATION_INSTRUCTIONS 

819 

820 for field in self.REGISTRATION_FIELDS: 

821 form.add_field( 

822 field.var, 

823 label=field.label,