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 sys 

11import tempfile 

12from collections.abc import Callable, Sequence 

13from copy import copy 

14from datetime import UTC, datetime 

15from pathlib import Path 

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

17 

18import aiohttp 

19import qrcode 

20from slixmpp import JID, ComponentXMPP, Iq 

21from slixmpp.exceptions import IqError, IqTimeout, XMPPError 

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

23from slixmpp.plugins.xep_0060.stanza import OwnerAffiliation 

24from slixmpp.plugins.xep_0356.privilege import PrivilegedIqError 

25from slixmpp.types import MessageTypes 

26from slixmpp.xmlstream.xmlstream import NotConnectedError 

27from sqlalchemy.orm import Session as OrmSession 

28 

29import slidge.command.categories 

30from slidge.command.adhoc import AdhocProvider 

31from slidge.command.admin import Exec 

32from slidge.command.base import Command, FormField 

33from slidge.command.chat_command import ChatCommandProvider 

34from slidge.command.register import RegistrationType 

35from slidge.contact import LegacyContact 

36from slidge.core import config 

37from slidge.core.dispatcher.session_dispatcher import SessionDispatcher 

38from slidge.core.mixins.avatar import convert_avatar 

39from slidge.core.mixins.message import MessageMixin 

40from slidge.core.pubsub import PubSubComponent 

41from slidge.db import GatewayUser, SlidgeStore 

42from slidge.db.avatar import CachedAvatar, avatar_cache 

43from slidge.db.meta import JSONSerializable 

44from slidge.slixfix.delivery_receipt import DeliveryReceipt 

45from slidge.slixfix.roster import RosterBackend 

46from slidge.util import SubclassableOnce 

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

48 

49T = TypeVar("T") 

50P = ParamSpec("P") 

51 

52 

53class BaseGateway( 

54 ComponentXMPP, 

55 MessageMixin, 

56 SubclassableOnce, 

57 abc.ABC, 

58 Generic[SessionType], 

59): 

60 """ 

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

62 

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

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

65 ``.xmpp`` attribute. 

66 

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

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

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

70 

71 Abstract methods related to the registration process must be overriden 

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

73 

74 - :meth:`.validate` 

75 - :meth:`.validate_two_factor_code` 

76 - :meth:`.get_qr_text` 

77 - :meth:`.confirm_qr` 

78 

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

80 :attr:`REGISTRATION_TYPE`. 

81 

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

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

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

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

86 `mto` parameter. 

87 

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

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

90 

91 .. code-block:: python 

92 

93 self.send_presence( 

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

95 pto="someonwelse@anotherexample.com", 

96 ) 

97 

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

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

100 as sending messages, or displaying a custom status. 

101 

102 """ 

103 

104 COMPONENT_NAME: str = NotImplemented 

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

106 COMPONENT_TYPE: str = "" 

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

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

109 """ 

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

111 """ 

112 

113 REGISTRATION_FIELDS: ClassVar[Sequence[FormField]] = [ 

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

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

116 ] 

117 """ 

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

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

120 """ 

121 REGISTRATION_INSTRUCTIONS: str = "Enter your credentials" 

122 """ 

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

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

125 """ 

126 REGISTRATION_TYPE: RegistrationType = RegistrationType.SINGLE_STEP_FORM 

127 """ 

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

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

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

131 once per user (unless they unregister). 

132 

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

134 presented to the user. 

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

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

137 """ 

138 

139 REGISTRATION_2FA_TITLE = "Enter your 2FA code" 

140 REGISTRATION_2FA_INSTRUCTIONS = ( 

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

142 ) 

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

144 

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

146 FormField( 

147 var="sync_presence", 

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

149 value="true", 

150 required=True, 

151 type="boolean", 

152 ), 

153 FormField( 

154 var="sync_avatar", 

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

156 value="true", 

157 required=True, 

158 type="boolean", 

159 ), 

160 FormField( 

161 var="always_invite_when_adding_bookmarks", 

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

163 value="true", 

164 required=True, 

165 type="boolean", 

166 ), 

167 FormField( 

168 var="last_seen_fallback", 

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

170 value="true", 

171 required=True, 

172 type="boolean", 

173 ), 

174 FormField( 

175 var="roster_push", 

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

177 value="true", 

178 required=True, 

179 type="boolean", 

180 ), 

181 FormField( 

182 var="reaction_fallback", 

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

184 value="false", 

185 required=True, 

186 type="boolean", 

187 ), 

188 ] 

189 

190 ROSTER_GROUP: str = "slidge" 

191 """ 

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

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

194 """ 

195 WELCOME_MESSAGE = ( 

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

197 "or just start messaging away!" 

198 ) 

199 """ 

200 A welcome message displayed to users on registration. 

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

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

203 incoming messages from components. 

204 """ 

205 

206 SEARCH_FIELDS: ClassVar[Sequence[FormField]] = [ 

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

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

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

210 ] 

211 """ 

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

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

214 their usernames, eg their phone number. 

215 

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

217 (restricted to registered users). 

218 

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

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

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

222 """ 

223 SEARCH_TITLE: str = "Search for legacy contacts" 

224 """ 

225 Title of the search form. 

226 """ 

227 SEARCH_INSTRUCTIONS: str = "" 

228 """ 

229 Instructions of the search form. 

230 """ 

231 

232 MARK_ALL_MESSAGES = False 

233 """ 

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

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

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

237 """ 

238 

239 PROPER_RECEIPTS = False 

240 """ 

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

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

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

244 """ 

245 

246 GROUPS = False 

247 """ 

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

249 """ 

250 SPACES = False 

251 """ 

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

253 """ 

254 

255 mtype: MessageTypes = "chat" 

256 is_group = False 

257 _can_send_carbon = False 

258 store: SlidgeStore 

259 _session_cls: type[SessionType] 

260 

261 http: aiohttp.ClientSession 

262 avatar: CachedAvatar | None = None 

263 

264 def __init__(self) -> None: 

265 if config.COMPONENT_NAME: 

266 self.COMPONENT_NAME = config.COMPONENT_NAME 

267 if config.WELCOME_MESSAGE: 

268 self.WELCOME_MESSAGE = config.WELCOME_MESSAGE 

269 self.log = log 

270 self.datetime_started = datetime.now(tz=UTC) 

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

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

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

274 super().__init__( 

275 config.JID, 

276 config.SECRET, 

277 config.SERVER, 

278 config.PORT, 

279 plugin_whitelist=SLIXMPP_PLUGINS, 

280 plugin_config={ 

281 "xep_0077": { 

282 "form_fields": None, 

283 "form_instructions": self.REGISTRATION_INSTRUCTIONS, 

284 "enable_subscription": self.REGISTRATION_TYPE 

285 == RegistrationType.SINGLE_STEP_FORM, 

286 }, 

287 "xep_0100": { 

288 "component_name": self.COMPONENT_NAME, 

289 "type": self.COMPONENT_TYPE, 

290 }, 

291 "xep_0184": { 

292 "auto_ack": False, 

293 "auto_request": False, 

294 }, 

295 "xep_0363": { 

296 "upload_service": config.UPLOAD_SERVICE, 

297 }, 

298 }, 

299 fix_error_ns=True, 

300 ) 

301 self.loop.set_exception_handler(self.__exception_handler) 

302 self.loop.create_task(self.__set_http()) 

303 self.has_crashed: bool = False 

304 self.use_origin_id = False 

305 

306 if config.USER_JID_VALIDATOR is None: 

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

308 log.info( 

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

310 config.USER_JID_VALIDATOR, 

311 ) 

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

313 self.qr_pending_registrations = dict[ 

314 str, asyncio.Future[JSONSerializable | None] 

315 ]() 

316 

317 self.register_plugins() 

318 self.__setup_legacy_module_subclasses() 

319 

320 self.get_session_from_stanza = self._session_cls.from_stanza 

321 self.get_session_from_user = self._session_cls.from_user 

322 

323 self.__register_slixmpp_events() 

324 self.__register_slixmpp_api() 

325 self.roster.set_backend(RosterBackend(self)) 

326 

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

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

329 self.delivery_receipt = DeliveryReceipt(self) 

330 

331 # with this we receive user avatar updates 

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

333 

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

335 

336 if self.GROUPS: 

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

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

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

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

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

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

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

344 category="conference", 

345 name=self.COMPONENT_NAME, 

346 itype="text", 

347 jid=self.boundjid, 

348 ) 

349 if self.SPACES: 

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

351 

352 self.__adhoc_handler = AdhocProvider(self) 

353 self.__chat_commands_handler = ChatCommandProvider(self) 

354 

355 self.__dispatcher = SessionDispatcher(self) 

356 

357 self.__register_commands() 

358 

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

360 

361 def __setup_legacy_module_subclasses(self) -> None: 

362 from ..contact.roster import LegacyRoster 

363 from ..group.bookmarks import LegacyBookmarks 

364 from ..group.participant import LegacyParticipant 

365 from ..group.room import LegacyMUC 

366 from .session import BaseSession 

367 

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

369 contact_cls = LegacyContact.get_self_or_unique_subclass() 

370 muc_cls = LegacyMUC.get_self_or_unique_subclass() 

371 participant_cls = LegacyParticipant.get_self_or_unique_subclass() 

372 bookmarks_cls = LegacyBookmarks.get_self_or_unique_subclass() 

373 roster_cls = LegacyRoster.get_self_or_unique_subclass() 

374 

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

376 form = Form() 

377 form["type"] = "result" 

378 form.add_field( 

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

380 ) 

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

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

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

384 

385 self._session_cls.xmpp = self 

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

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

388 

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

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

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

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

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

394 

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

396 await self._session_cls.kill_by_jid(jid) 

397 

398 async def __set_http(self) -> None: 

399 self.http = aiohttp.ClientSession() 

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

401 return 

402 avatar_cache.http = self.http 

403 

404 def __register_commands(self) -> None: 

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

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

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

408 continue 

409 if cls is Exec: 

410 if config.DEV_MODE: 

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

412 else: 

413 continue 

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

415 continue 

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

417 continue 

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

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

420 self.__adhoc_handler.register(c) 

421 self.__chat_commands_handler.register(c) 

422 

423 def __exception_handler( 

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

425 ) -> None: 

426 """ 

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

428 

429 :param loop: 

430 :param context: 

431 :return: 

432 """ 

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

434 exc = context.get("exception") 

435 if exc is None: 

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

437 elif isinstance(exc, SystemExit): 

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

439 else: 

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

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

442 self.has_crashed = True 

443 loop.stop() 

444 

445 def __register_slixmpp_events(self) -> None: 

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

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

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

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

450 self.del_event_handler( 

451 "roster_subscription_request", self._handle_new_subscription 

452 ) 

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

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

455 

456 def __register_slixmpp_api(self) -> None: 

457 def with_session( 

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

459 ) -> Callable[P, T]: 

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

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

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

463 if commit: 

464 orm.commit() 

465 return res 

466 

467 return wrapped 

468 

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

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

471 ) 

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

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

474 ) 

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

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

477 ) 

478 

479 @property # type:ignore[override] 

480 def jid(self) -> JID: 

481 # Override to avoid slixmpp deprecation warnings. 

482 return self.boundjid 

483 

484 @jid.setter 

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

486 raise RuntimeError 

487 

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

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

490 

491 await self.__setup_attachments() 

492 

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

494 disco = self.plugin["xep_0030"] 

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

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

497 

498 if self.COMPONENT_AVATAR is not None: 

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

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

501 assert avatar is not None 

502 try: 

503 cached_avatar = await avatar_cache.convert_or_get(avatar) 

504 except Exception as e: 

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

506 cached_avatar = None 

507 else: 

508 assert cached_avatar is not None 

509 self.avatar = cached_avatar 

510 else: 

511 cached_avatar = None 

512 

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

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

515 for user in users: 

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

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

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

519 # as last resort. 

520 try: 

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

522 await self.__add_component_to_mds_whitelist(user.jid) 

523 except (IqError, IqTimeout) as e: 

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

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

526 log.warning( 

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

528 user, 

529 exc_info=e, 

530 ) 

531 continue 

532 session = self._session_cls.from_user(user) 

533 session.create_task(self.login_wrap(session)) 

534 if cached_avatar is not None: 

535 await self.pubsub.broadcast_avatar( 

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

537 ) 

538 

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

540 

541 async def __setup_attachments(self) -> None: 

542 if config.NO_UPLOAD_PATH: 

543 if config.NO_UPLOAD_URL_PREFIX is None: 

544 raise RuntimeError( 

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

546 ) 

547 elif not config.UPLOAD_SERVICE: 

548 try: 

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

550 self.infer_real_domain() 

551 ) 

552 except XMPPError: 

553 info_iq = None 

554 log.exception( 

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

556 "Attachments to XMPP will not work. " 

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

558 ) 

559 if info_iq is None: 

560 if self.REGISTRATION_TYPE == RegistrationType.QRCODE: 

561 log.warning( 

562 "No method was configured for attachment and slidge " 

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

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

565 "QR-code based registration flow." 

566 ) 

567 if not config.USE_ATTACHMENT_ORIGINAL_URLS: 

568 log.warning( 

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

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

571 "networks, especially for the E2EE attachments." 

572 ) 

573 config.USE_ATTACHMENT_ORIGINAL_URLS = True 

574 else: 

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

576 config.UPLOAD_SERVICE = info_iq["from"] 

577 

578 def infer_real_domain(self) -> JID: 

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

580 

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

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

583 # MDS node so we receive MDS events 

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

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

586 

587 try: 

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

589 except PermissionError: 

590 log.warning( 

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

592 "create the MDS node of %s", 

593 user_jid, 

594 ) 

595 except PrivilegedIqError as exc: 

596 nested = exc.nested_error() 

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

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

599 log.exception( 

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

601 ) 

602 except Exception as e: 

603 log.exception( 

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

605 user_jid, 

606 exc_info=e, 

607 ) 

608 

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

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

611 "xep_0490" 

612 ].stanza.NS 

613 

614 aff = OwnerAffiliation() 

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

616 aff["affiliation"] = "member" 

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

618 

619 try: 

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

621 except PermissionError: 

622 log.warning( 

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

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

625 user_jid, 

626 ) 

627 except Exception as e: 

628 log.exception( 

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

630 user_jid, 

631 exc_info=e, 

632 ) 

633 

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

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

636 session.is_logging_in = True 

637 try: 

638 status = await session.login() 

639 except Exception as e: 

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

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

642 msg = ( 

643 "You are not connected to this gateway! " 

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

645 ) 

646 session.send_gateway_message(msg) 

647 session.logged = False 

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

649 return msg 

650 

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

652 session.logged = True 

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

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

655 await session.contacts._fill(orm) 

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

657 r.set_result(True) 

658 if self.GROUPS: 

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

660 await session.bookmarks.fill() 

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

662 r.set_result(True) 

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

664 if status is None: 

665 status = "Logged in" 

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

667 

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

669 session.create_task(self.fetch_user_avatar(session)) 

670 else: 

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

672 session.user.avatar_hash = None 

673 orm.add(session.user) 

674 orm.commit() 

675 return status 

676 

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

678 try: 

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

680 session.user_jid.bare, 

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

682 ifrom=self.boundjid.bare, 

683 ) 

684 except IqTimeout: 

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

686 return 

687 except IqError as e: 

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

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

690 try: 

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

692 except NotImplementedError: 

693 pass 

694 else: 

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

696 session.user.avatar_hash = None 

697 orm.add(session.user) 

698 orm.commit() 

699 return 

700 await self.__dispatcher.on_avatar_metadata_info( 

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

702 ) 

703 

704 def _send( 

705 self, 

706 stanza: MessageOrPresenceTypeVar, 

707 **send_kwargs: Any, # noqa:ANN401 

708 ) -> MessageOrPresenceTypeVar: 

709 stanza.set_from(self.boundjid.bare) 

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

711 stanza.set_to(mto) 

712 stanza.send() 

713 return stanza 

714 

715 def raise_if_not_allowed_jid(self, jid: JID) -> None: 

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

717 raise XMPPError( 

718 condition="not-allowed", 

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

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

721 ) 

722 

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

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

725 # to make them more readable. 

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

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

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

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

730 # does not matter much 

731 for el in LOG_STRIP_ELEMENTS: 

732 stripped = re.sub( 

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

734 "\1[STRIPPED]\3", 

735 stripped, 

736 flags=re.DOTALL | re.IGNORECASE, 

737 ) 

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

739 if not self.transport: 

740 raise NotConnectedError() 

741 if isinstance(data, str): 

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

743 self.transport.write(data) 

744 

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

746 try: 

747 return self._session_cls.from_jid(j) 

748 except XMPPError: 

749 return None 

750 

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

752 # """ 

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

754 # 

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

756 # 

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

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

759 # 

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

761 # """ 

762 if isinstance(exception, IqError): 

763 iq = exception.iq 

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

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

766 elif isinstance(exception, IqTimeout): 

767 iq = exception.iq 

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

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

770 elif isinstance(exception, SyntaxError): 

771 # Hide stream parsing errors that occur when the 

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

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

774 pass 

775 else: 

776 if exception: 

777 log.exception(exception) 

778 self.loop.stop() 

779 sys.exit(1) 

780 

781 async def make_registration_form( 

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

783 ) -> Iq: 

784 self.raise_if_not_allowed_jid(iq.get_from()) 

785 reg = iq["register"] 

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

787 user = ( 

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

789 ) 

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

791 

792 form = reg["form"] 

793 form.add_field( 

794 "FORM_TYPE", 

795 ftype="hidden", 

796 value="jabber:iq:register", 

797 ) 

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

799 form["instructions"] = self.REGISTRATION_INSTRUCTIONS 

800 

801 if user is not None: 

802 reg["registered"] = False 

803 form.add_field( 

804 "remove", 

805 label="Remove my registration", 

806 required=True, 

807 ftype="boolean", 

808 value=False, 

809 ) 

810 

811 for field in self.REGISTRATION_FIELDS: 

812 if field.var in reg.interfaces: 

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

814 if val is None: 

815 reg.add_field(field.var) 

816 else: 

817 reg[field.var] = val 

818 

819 reg["instructions"] = self.REGISTRATION_INSTRUCTIONS 

820 

821 for field in self.REGISTRATION_FIELDS: 

822 form.add_field(